PostgreSQLSchemaManagerTest.php

<?php

declare(strict_types=1);

namespace Doctrine\DBAL\Tests\Functional\Schema;

use Doctrine\DBAL\Platforms\AbstractPlatform;
use Doctrine\DBAL\Platforms\PostgreSQL120Platform;
use Doctrine\DBAL\Platforms\PostgreSQLPlatform;
use Doctrine\DBAL\Schema\Exception\TableDoesNotExist;
use Doctrine\DBAL\Schema\ForeignKeyConstraint;
use Doctrine\DBAL\Schema\Schema;
use Doctrine\DBAL\Schema\Table;
use Doctrine\DBAL\Schema\View;
use Doctrine\DBAL\Types\BlobType;
use Doctrine\DBAL\Types\DecimalType;
use Doctrine\DBAL\Types\IntegerType;
use Doctrine\DBAL\Types\JsonType;
use Doctrine\DBAL\Types\TextType;
use Doctrine\DBAL\Types\Type;
use Doctrine\DBAL\Types\Types;
use PHPUnit\Framework\Attributes\DataProvider;

use function array_map;
use function count;
use function sprintf;
use function strtolower;
use function version_compare;

class PostgreSQLSchemaManagerTest extends SchemaManagerFunctionalTestCase
{
    protected function supportsPlatform(AbstractPlatform $platform): bool
    {
        return $platform instanceof PostgreSQLPlatform;
    }

    public function testGetSchemaNames(): void
    {
        $names = $this->schemaManager->listSchemaNames();

        self::assertContains('public', $names, 'The public schema should be found.');
    }

    public function testSupportDomainTypeFallback(): void
    {
        $createDomainTypeSQL = 'CREATE DOMAIN MyMoney AS DECIMAL(18,2)';
        $this->connection->executeStatement($createDomainTypeSQL);

        $createTableSQL = 'CREATE TABLE domain_type_test (id INT PRIMARY KEY, value MyMoney)';
        $this->connection->executeStatement($createTableSQL);

        $table = $this->connection->createSchemaManager()->introspectTable('domain_type_test');
        self::assertInstanceOf(DecimalType::class, $table->getColumn('value')->getType());

        Type::addType('MyMoney', MoneyType::class);
        $this->connection->getDatabasePlatform()->registerDoctrineTypeMapping('MyMoney', 'MyMoney');

        $table = $this->connection->createSchemaManager()->introspectTable('domain_type_test');
        self::assertInstanceOf(MoneyType::class, $table->getColumn('value')->getType());
    }

    public function testAlterTableAutoIncrementAdd(): void
    {
        $tableFrom = new Table('autoinc_table_add');
        $tableFrom->addColumn('id', Types::INTEGER);
        $this->dropAndCreateTable($tableFrom);
        $tableFrom = $this->schemaManager->introspectTable('autoinc_table_add');
        self::assertFalse($tableFrom->getColumn('id')->getAutoincrement());

        $tableTo = new Table('autoinc_table_add');
        $column  = $tableTo->addColumn('id', Types::INTEGER);
        $column->setAutoincrement(true);

        $platform = $this->connection->getDatabasePlatform();
        $diff     = $this->schemaManager->createComparator()
            ->compareTables($tableFrom, $tableTo);

        $sql = $platform->getAlterTableSQL($diff);
        self::assertEquals(['ALTER TABLE autoinc_table_add ALTER id ADD GENERATED BY DEFAULT AS IDENTITY'], $sql);

        $this->schemaManager->alterTable($diff);
        $tableFinal = $this->schemaManager->introspectTable('autoinc_table_add');
        self::assertTrue($tableFinal->getColumn('id')->getAutoincrement());
    }

    public function testAlterTableAutoIncrementDrop(): void
    {
        $tableFrom = new Table('autoinc_table_drop');
        $column    = $tableFrom->addColumn('id', Types::INTEGER);
        $column->setAutoincrement(true);
        $this->dropAndCreateTable($tableFrom);
        $tableFrom = $this->schemaManager->introspectTable('autoinc_table_drop');
        self::assertTrue($tableFrom->getColumn('id')->getAutoincrement());

        $tableTo = new Table('autoinc_table_drop');
        $tableTo->addColumn('id', Types::INTEGER);

        $platform = $this->connection->getDatabasePlatform();
        $diff     = $this->schemaManager->createComparator()
            ->compareTables($tableFrom, $tableTo);

        self::assertEquals(
            ['ALTER TABLE autoinc_table_drop ALTER id DROP IDENTITY'],
            $platform->getAlterTableSQL($diff),
        );

        $this->schemaManager->alterTable($diff);
        $tableFinal = $this->schemaManager->introspectTable('autoinc_table_drop');
        self::assertFalse($tableFinal->getColumn('id')->getAutoincrement());
    }

    public function testListSameTableNameColumnsWithDifferentSchema(): void
    {
        $this->connection->executeStatement('CREATE SCHEMA another');
        $table = new Table('table');
        $table->addColumn('id', Types::INTEGER);
        $table->addColumn('name', Types::TEXT);
        $this->schemaManager->createTable($table);

        $anotherSchemaTable = new Table('another.table');
        $anotherSchemaTable->addColumn('id', Types::TEXT);
        $anotherSchemaTable->addColumn('email', Types::TEXT);
        $this->schemaManager->createTable($anotherSchemaTable);

        $table = $this->schemaManager->introspectTable('table');
        self::assertCount(2, $table->getColumns());
        self::assertTrue($table->hasColumn('id'));
        self::assertInstanceOf(IntegerType::class, $table->getColumn('id')->getType());
        self::assertTrue($table->hasColumn('name'));

        $anotherSchemaTable = $this->schemaManager->introspectTable('another.table');
        self::assertCount(2, $anotherSchemaTable->getColumns());
        self::assertTrue($anotherSchemaTable->hasColumn('id'));
        self::assertInstanceOf(TextType::class, $anotherSchemaTable->getColumn('id')->getType());
        self::assertTrue($anotherSchemaTable->hasColumn('email'));
    }

    public function testReturnQuotedAssets(): void
    {
        $this->connection->executeStatement('DROP TABLE IF EXISTS dbal91_something');

        $sql = 'create table dbal91_something'
            . ' (id integer CONSTRAINT id_something PRIMARY KEY NOT NULL, "table" integer)';
        $this->connection->executeStatement($sql);

        $sql = 'ALTER TABLE dbal91_something ADD CONSTRAINT something_input'
            . ' FOREIGN KEY( "table" ) REFERENCES dbal91_something ON UPDATE CASCADE;';
        $this->connection->executeStatement($sql);

        $table = $this->schemaManager->introspectTable('dbal91_something');

        self::assertEquals(
            [
                'CREATE TABLE dbal91_something (id INT NOT NULL, "table" INT DEFAULT NULL, PRIMARY KEY(id))',
                'CREATE INDEX IDX_A9401304ECA7352B ON dbal91_something ("table")',
                'ALTER TABLE dbal91_something ADD CONSTRAINT something_input FOREIGN KEY ("table")'
                    . ' REFERENCES dbal91_something (id) ON UPDATE CASCADE NOT DEFERRABLE INITIALLY IMMEDIATE',
            ],
            $this->connection->getDatabasePlatform()->getCreateTableSQL($table),
        );
    }

    public function testListForeignKeys(): void
    {
        $fkOptions   = ['SET NULL', 'SET DEFAULT', 'NO ACTION', 'CASCADE', 'RESTRICT'];
        $foreignKeys = [];
        $fkTable     = $this->getTestTable('test_create_fk1');
        foreach ($fkOptions as $i => $fkOption) {
            $fkTable->addColumn('foreign_key_test' . $i, Types::INTEGER);
            $foreignKeys[] = new ForeignKeyConstraint(
                ['foreign_key_test' . $i],
                'test_create_fk2',
                ['id'],
                'foreign_key_test' . $i . '_fk',
                ['onDelete' => $fkOption],
            );
        }

        $this->dropAndCreateTable($fkTable);
        $this->createTestTable('test_create_fk2');

        foreach ($foreignKeys as $foreignKey) {
            $this->schemaManager->createForeignKey($foreignKey, 'test_create_fk1');
        }

        $fkeys = $this->schemaManager->listTableForeignKeys('test_create_fk1');
        self::assertEquals(count($foreignKeys), count($fkeys));

        for ($i = 0; $i < count($fkeys); $i++) {
            self::assertEquals(['foreign_key_test' . $i], array_map('strtolower', $fkeys[$i]->getLocalColumns()));
            self::assertEquals(['id'], array_map('strtolower', $fkeys[$i]->getForeignColumns()));
            self::assertEquals('test_create_fk2', strtolower($fkeys[0]->getForeignTableName()));
            if ($foreignKeys[$i]->getOption('onDelete') === 'NO ACTION') {
                self::assertFalse($fkeys[$i]->hasOption('onDelete'));
            } else {
                self::assertEquals($foreignKeys[$i]->getOption('onDelete'), $fkeys[$i]->getOption('onDelete'));
            }
        }
    }

    public function testDefaultValueCharacterVarying(): void
    {
        $testTable = new Table('dbal511_default');
        $testTable->addColumn('id', Types::INTEGER);
        $testTable->addColumn('def', Types::STRING, ['default' => 'foo']);
        $testTable->setPrimaryKey(['id']);
        $this->dropAndCreateTable($testTable);

        $databaseTable = $this->schemaManager->introspectTable($testTable->getName());

        self::assertEquals('foo', $databaseTable->getColumn('def')->getDefault());
    }

    public function testJsonDefaultValue(): void
    {
        $testTable = new Table('test_json');
        $testTable
            ->addColumn('foo', Types::JSON)
            ->setDefault('{"key": "value with a single quote \' in string value"}');
        $this->dropAndCreateTable($testTable);

        $columns = $this->schemaManager->listTableColumns('test_json');

        self::assertSame(Type::getType(Types::JSON), $columns['foo']->getType());
        self::assertSame('{"key": "value with a single quote \' in string value"}', $columns['foo']->getDefault());
    }

    public function testBooleanDefault(): void
    {
        $table = new Table('ddc2843_bools');
        $table->addColumn('id', Types::INTEGER);
        $table->addColumn('checked', Types::BOOLEAN, ['default' => false]);

        $this->dropAndCreateTable($table);

        $databaseTable = $this->schemaManager->introspectTable($table->getName());

        self::assertTrue(
            $this->schemaManager->createComparator()
                ->compareTables($table, $databaseTable)
                ->isEmpty(),
        );
    }

    public function testGeneratedColumn(): void
    {
        if (! $this->connection->getDatabasePlatform() instanceof PostgreSQL120Platform) {
             self::markTestSkipped('Generated columns are not supported in Postgres 11 and earlier');
        }

        $table = new Table('ddc6198_generated_always_as');
        $table->addColumn('id', Types::INTEGER);
        $table->addColumn(
            'idIsOdd',
            Types::BOOLEAN,
            ['columnDefinition' => 'boolean GENERATED ALWAYS AS (id % 2 = 1) STORED', 'notNull' => false],
        );

        $this->dropAndCreateTable($table);

        $databaseTable = $this->schemaManager->introspectTable($table->getName());

        self::assertTrue(
            $this->schemaManager->createComparator()
                ->compareTables($table, $databaseTable)
                ->isEmpty(),
        );
    }

    /**
     * PostgreSQL stores BINARY columns as BLOB
     */
    protected function assertBinaryColumnIsValid(Table $table, string $columnName, int $expectedLength): void
    {
        self::assertInstanceOf(BlobType::class, $table->getColumn($columnName)->getType());
    }

    /**
     * PostgreSQL stores VARBINARY columns as BLOB
     */
    protected function assertVarBinaryColumnIsValid(Table $table, string $columnName, int $expectedLength): void
    {
        self::assertInstanceOf(BlobType::class, $table->getColumn($columnName)->getType());
    }

    /**
     * Although this test would pass in isolation on any platform, we keep it here for the following reasons:
     *
     * 1. The DBAL currently doesn't properly drop tables in the namespaces that need to be quoted
     *    (@see testListTableDetailsWhenCurrentSchemaNameQuoted()).
     * 2. The schema returned by {@see AbstractSchemaManager::introspectSchema()} doesn't contain views, so
     *    {@see AbstractSchemaManager::dropSchemaObjects()} cannot drop the tables that have dependent views
     *    (@see testListTablesExcludesViews()).
     * 3. In the case of inheritance, PHPUnit runs the tests declared immediately in the test class
     *    and then runs the tests declared in the parent.
     *
     * This test needs to be executed before the ones it conflicts with, so it has to be declared in the same class.
     */
    public function testDropWithAutoincrement(): void
    {
        $this->dropTableIfExists('test_autoincrement');

        $schema = new Schema();
        $table  = $schema->createTable('test_autoincrement');
        $table->addColumn('id', Types::INTEGER, [
            'notnull' => true,
            'autoincrement' => true,
        ]);
        $table->setPrimaryKey(['id']);

        $schemaManager = $this->connection->createSchemaManager();
        $schemaManager->createSchemaObjects($schema);

        $schema = $schemaManager->introspectSchema();
        $schemaManager->dropSchemaObjects($schema);

        self::assertFalse($schemaManager->tablesExist(['test_autoincrement']));
    }

    public function testListTableDetailsWhenCurrentSchemaNameQuoted(): void
    {
        $this->connection->executeStatement('CREATE SCHEMA "001_test"');
        $this->connection->executeStatement('SET search_path TO "001_test"');
        $this->markConnectionNotReusable();

        $this->testIntrospectReservedKeywordTableViaListTableDetails();
    }

    public function testListTablesExcludesViews(): void
    {
        $this->createTestTable('list_tables_excludes_views');

        $name = 'list_tables_excludes_views_test_view';
        $sql  = 'SELECT * from list_tables_excludes_views';

        $view = new View($name, $sql);

        $this->schemaManager->createView($view);

        $tables = $this->schemaManager->listTables();

        $foundTable = false;
        foreach ($tables as $table) {
            if (strtolower($table->getName()) !== 'list_tables_excludes_views_test_view') {
                continue;
            }

            $foundTable = true;
        }

        self::assertFalse($foundTable, 'View "list_tables_excludes_views_test_view" must not be found in table list');
    }

    public function testPartialIndexes(): void
    {
        $offlineTable = new Table('person');
        $offlineTable->addColumn('id', Types::INTEGER);
        $offlineTable->addColumn('name', Types::STRING);
        $offlineTable->addColumn('email', Types::STRING);
        $offlineTable->addUniqueIndex(['id', 'name'], 'simple_partial_index', ['where' => '(id IS NULL)']);

        $this->dropAndCreateTable($offlineTable);

        $onlineTable = $this->schemaManager->introspectTable('person');

        self::assertTrue(
            $this->schemaManager->createComparator()
                ->compareTables($offlineTable, $onlineTable)
                ->isEmpty(),
        );
        self::assertTrue($onlineTable->hasIndex('simple_partial_index'));
        self::assertTrue($onlineTable->getIndex('simple_partial_index')->hasOption('where'));
        self::assertSame('(id IS NULL)', $onlineTable->getIndex('simple_partial_index')->getOption('where'));
    }

    public function testJsonbColumn(): void
    {
        $table = new Table('test_jsonb');
        $table->addColumn('foo', Types::JSON)->setPlatformOption('jsonb', true);
        $this->dropAndCreateTable($table);

        $columns = $this->schemaManager->listTableColumns('test_jsonb');

        self::assertInstanceOf(JsonType::class, $columns['foo']->getType());
        self::assertTrue($columns['foo']->getPlatformOption('jsonb'));
    }

    public function testListNegativeColumnDefaultValue(): void
    {
        $table = new Table('test_default_negative');
        $table->addColumn('col_smallint', Types::SMALLINT, ['default' => -1]);
        $table->addColumn('col_integer', Types::INTEGER, ['default' => -1]);
        $table->addColumn('col_bigint', Types::BIGINT, ['default' => -1]);
        $table->addColumn('col_float', Types::FLOAT, ['default' => -1.1]);
        $table->addColumn('col_smallfloat', Types::SMALLFLOAT, ['default' => -1.1]);
        $table->addColumn('col_decimal', Types::DECIMAL, [
            'precision' => 2,
            'scale' => 1,
            'default' => -1.1,
        ]);
        $table->addColumn('col_string', Types::STRING, ['default' => '(-1)']);

        $this->dropAndCreateTable($table);

        $columns = $this->schemaManager->listTableColumns('test_default_negative');

        self::assertEquals(-1, $columns['col_smallint']->getDefault());
        self::assertEquals(-1, $columns['col_integer']->getDefault());
        self::assertEquals(-1, $columns['col_bigint']->getDefault());
        self::assertEquals(-1.1, $columns['col_float']->getDefault());
        self::assertEquals(-1.1, $columns['col_smallfloat']->getDefault());
        self::assertEquals(-1.1, $columns['col_decimal']->getDefault());
        self::assertEquals('(-1)', $columns['col_string']->getDefault());
    }

    /** @return mixed[][] */
    public static function serialTypes(): iterable
    {
        return [
            [Types::INTEGER],
            [Types::BIGINT],
        ];
    }

    #[DataProvider('serialTypes')]
    public function testAutoIncrementCreatesSerialDataTypesWithoutADefaultValue(string $type): void
    {
        $tableName = 'test_serial_type_' . $type;

        $table = new Table($tableName);
        $table->addColumn('id', $type, ['autoincrement' => true, 'notnull' => false]);

        $this->dropAndCreateTable($table);

        $columns = $this->schemaManager->listTableColumns($tableName);

        self::assertNull($columns['id']->getDefault());
    }

    #[DataProvider('serialTypes')]
    public function testAutoIncrementCreatesSerialDataTypesWithoutADefaultValueEvenWhenDefaultIsSet(string $type): void
    {
        $tableName = 'test_serial_type_with_default_' . $type;

        $table = new Table($tableName);
        $table->addColumn('id', $type, ['autoincrement' => true, 'notnull' => false, 'default' => 1]);

        $this->dropAndCreateTable($table);

        $columns = $this->schemaManager->listTableColumns($tableName);

        self::assertNull($columns['id']->getDefault());
    }

    #[DataProvider('autoIncrementTypeMigrations')]
    public function testAlterTableAutoIncrementIntToBigInt(string $from, string $to, string $expected): void
    {
        $table  = new Table('autoinc_type_modification');
        $column = $table->addColumn('id', $from);
        $column->setAutoincrement(true);
        $this->dropAndCreateTable($table);

        $oldTable = $this->schemaManager->introspectTable('autoinc_type_modification');
        self::assertTrue($oldTable->getColumn('id')->getAutoincrement());

        $newTable = new Table('autoinc_type_modification');
        $column   = $newTable->addColumn('id', $to);
        $column->setAutoincrement(true);

        $diff = $this->schemaManager->createComparator()
            ->compareTables($oldTable, $newTable);

        self::assertSame(
            ['ALTER TABLE autoinc_type_modification ALTER id TYPE ' . $expected],
            $this->connection->getDatabasePlatform()->getAlterTableSQL($diff),
        );

        $this->schemaManager->alterTable($diff);
        $tableFinal = $this->schemaManager->introspectTable('autoinc_type_modification');
        self::assertTrue($tableFinal->getColumn('id')->getAutoincrement());
    }

    public function testListTableColumnsOidConflictWithNonTableObject(): void
    {
        if (version_compare($this->connection->getServerVersion(), '12.0', '<')) {
            self::markTestSkipped('Manually setting the Oid is not supported in Postgres 11 and earlier');
        }

        $table = 'test_list_table_columns_oid_conflicts';
        $this->connection->executeStatement(sprintf('CREATE TABLE IF NOT EXISTS %s(id INT NOT NULL)', $table));
        $beforeColumns = $this->schemaManager->listTableColumns($table);
        self::assertArrayHasKey('id', $beforeColumns);

        $this->connection->executeStatement('CREATE EXTENSION IF NOT EXISTS pg_prewarm');
        $originalTableOid = $this->connection->fetchOne(
            'SELECT oid FROM pg_class WHERE pg_class.relname = ?',
            [$table],
        );

        $getConflictingOidSql = <<<'SQL'
SELECT objid
FROM pg_depend
JOIN pg_extension as ex on ex.oid = pg_depend.refobjid
WHERE ex.extname = 'pg_prewarm'
ORDER BY objid
LIMIT 1
SQL;
        $conflictingOid       = $this->connection->fetchOne($getConflictingOidSql);

        $this->connection->executeStatement(
            'UPDATE pg_attribute SET attrelid = ? WHERE attrelid = ?',
            [$conflictingOid, $originalTableOid],
        );
        $this->connection->executeStatement(
            'UPDATE pg_description SET objoid = ? WHERE objoid = ?',
            [$conflictingOid, $originalTableOid],
        );
        $this->connection->executeStatement(
            'UPDATE pg_class SET oid = ? WHERE oid = ?',
            [$conflictingOid, $originalTableOid],
        );

        $afterColumns = $this->schemaManager->listTableColumns($table);

        // revert to the database to original state prior to asserting result
        $this->connection->executeStatement(
            'UPDATE pg_attribute SET attrelid = ? WHERE attrelid = ?',
            [$originalTableOid, $conflictingOid],
        );
        $this->connection->executeStatement(
            'UPDATE pg_description SET objoid = ? WHERE objoid = ?',
            [$originalTableOid, $conflictingOid],
        );
        $this->connection->executeStatement(
            'UPDATE pg_class SET oid = ? WHERE oid = ?',
            [$originalTableOid, $conflictingOid],
        );
        $this->connection->executeStatement(sprintf('DROP TABLE IF EXISTS %s', $table));
        $this->connection->executeStatement('DROP EXTENSION IF EXISTS pg_prewarm');

        self::assertArrayHasKey('id', $afterColumns);
    }

    /** @return iterable<mixed[]> */
    public static function autoIncrementTypeMigrations(): iterable
    {
        return [
            'int->bigint' => ['integer', 'bigint', 'BIGINT'],
            'bigint->int' => ['bigint', 'integer', 'INT'],
        ];
    }

    public function testPartitionTable(): void
    {
        $this->connection->executeStatement('DROP TABLE IF EXISTS partitioned_table');
        $this->connection->executeStatement(
            'CREATE TABLE partitioned_table (id INT) PARTITION BY LIST (id);',
        );
        $this->connection->executeStatement('CREATE TABLE partition PARTITION OF partitioned_table FOR VALUES IN (1);');
        try {
            $this->schemaManager->introspectTable('partition');
        } catch (TableDoesNotExist $e) {
        }

        self::assertNotNull($e ?? null, 'Partition table should not be introspected');

        $tableFrom = $this->schemaManager->introspectTable('partitioned_table');

        $tableTo = $this->schemaManager->introspectTable('partitioned_table');
        $tableTo->addColumn('foo', Types::INTEGER);

        $platform = $this->connection->getDatabasePlatform();
        $diff     = $this->schemaManager->createComparator()->compareTables($tableFrom, $tableTo);

        $sql = $platform->getAlterTableSQL($diff);
        self::assertSame(['ALTER TABLE partitioned_table ADD foo INT NOT NULL'], $sql);

        $this->schemaManager->alterTable($diff);

        $tableFinal = $this->schemaManager->introspectTable('partitioned_table');
        self::assertTrue($tableFinal->hasColumn('id'));
        self::assertTrue($tableFinal->hasColumn('foo'));

        $partitionedTableCount = (int) ($this->connection->fetchOne(
            "select count(*) as count from pg_class where relname = 'partitioned_table' and relkind = 'p'",
        ));
        self::assertSame(1, $partitionedTableCount);

        $partitionsCount = (int) ($this->connection->fetchOne(
            <<<'SQL'
            select count(*) as count
            from pg_class parent
            inner join pg_inherits on pg_inherits.inhparent = parent.oid
            inner join pg_class child on pg_inherits.inhrelid = child.oid 
                and child.relkind = 'r'
                and child.relname = 'partition'
            where parent.relname = 'partitioned_table' and parent.relkind = 'p';
            SQL,
        ));
        self::assertSame(1, $partitionsCount);
    }
}

class MoneyType extends Type
{
    /**
     * {@inheritDoc}
     */
    public function getSQLDeclaration(array $column, AbstractPlatform $platform): string
    {
        return 'MyMoney';
    }
}