Add schema support to MS SQL Server

src/Db/Adapter/PostgresAdapter.php

@@ -1506,8 +1506,9 @@
     protected function getGlobalSchemaName(): string
     {
         $options = $this->getOptions();
+        $config = $options['connection']->config() ?? [];
 
-        return empty($options['schema']) ? 'public' : $options['schema'];
+        return empty($config['schema']) ? 'public' : $config['schema'];
     }
 
     /**

src/Db/Adapter/SqlserverAdapter.php

@@ -98,12 +98,25 @@
         $this->getConnection()->rollback();
     }
 
+    /**
+     * Quotes a schema name for use in a query.
+     *
+     * @param string $schemaName Schema Name
+     * @return string
+     */
+    public function quoteSchemaName(string $schemaName): string
+    {
+        return $this->quoteColumnName($schemaName);
+    }
+
     /**
      * @inheritDoc
      */
     public function quoteTableName(string $tableName): string
     {
-        return str_replace('.', '].[', $this->quoteColumnName($tableName));
+        $parts = $this->getSchemaName($tableName);
+
+        return $this->quoteSchemaName($parts['schema']) . '.' . $this->quoteColumnName($parts['table']);
     }
 
     /**
@@ -123,8 +136,13 @@
             return true;
         }
 
+        $parts = $this->getSchemaName($tableName);
         /** @var array<string, mixed> $result */
-        $result = $this->fetchRow(sprintf("SELECT count(*) as [count] FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = '%s';", $tableName));
+        $result = $this->fetchRow(sprintf(
+            "SELECT count(*) as [count] FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = '%s' AND TABLE_NAME = '%s';",
+            $parts['schema'],
+            $parts['table'],
+        ));
 
         return $result['count'] > 0;
     }
@@ -135,6 +153,7 @@
     public function createTable(Table $table, array $columns = [], array $indexes = []): void
     {
         $options = $table->getOptions();
+        $parts = $this->getSchemaName($table->getName());
 
         // Add the default primary key
         if (!isset($options['id']) || (isset($options['id']) && $options['id'] === true)) {
@@ -170,7 +189,7 @@
 
         // set the primary key(s)
         if (isset($options['primary_key'])) {
-            $pkSql = sprintf('CONSTRAINT PK_%s PRIMARY KEY (', $table->getName());
+            $pkSql = sprintf('CONSTRAINT PK_%s PRIMARY KEY (', $parts['table']);
             /** @var string|array $primaryKey */
             $primaryKey = $options['primary_key'];
 
@@ -347,6 +366,7 @@
      */
     public function getColumns(string $tableName): array
     {
+        $parts = $this->getSchemaName($tableName);
         $columns = [];
         $sql = sprintf(
             "SELECT DISTINCT TABLE_SCHEMA AS [schema], TABLE_NAME as [table_name], COLUMN_NAME AS [name], DATA_TYPE AS [type],
@@ -356,9 +376,10 @@
             NUMERIC_SCALE AS [scale], ORDINAL_POSITION AS [ordinal_position],
             COLUMNPROPERTY(object_id(TABLE_NAME), COLUMN_NAME, 'IsIdentity') as [identity]
         FROM INFORMATION_SCHEMA.COLUMNS
-        WHERE TABLE_NAME = '%s'
+        WHERE TABLE_SCHEMA = '%s' AND TABLE_NAME = '%s'
         ORDER BY ordinal_position",
-            $tableName
+            $parts['schema'],
+            $parts['table'],
         );
         $rows = $this->fetchAll($sql);
         foreach ($rows as $columnInfo) {
@@ -415,11 +436,13 @@
      */
     public function hasColumn(string $tableName, string $columnName): bool
     {
+        $parts = $this->getSchemaName($tableName);
         $sql = sprintf(
             "SELECT count(*) as [count]
              FROM INFORMATION_SCHEMA.COLUMNS
-             WHERE TABLE_NAME = '%s' AND COLUMN_NAME = '%s'",
-            $tableName,
+             WHERE TABLE_SCHEMA = '%s' AND TABLE_NAME = '%s' AND COLUMN_NAME = '%s'",
+            $parts['schema'],
+            $parts['table'],
             $columnName
         );
         /** @var array<string, mixed> $result */
@@ -650,12 +673,19 @@
      */
     public function getIndexes(string $tableName): array
     {
+        $parts = $this->getSchemaName($tableName);
+
         $indexes = [];
-        $sql = "SELECT I.[name] AS [index_name], I.[index_id] as [index_id], T.[object_id] as [table_id]
+        $sql = sprintf(
+            "SELECT I.[name] AS [index_name], I.[index_id] as [index_id], T.[object_id] as [table_id]
 FROM sys.[tables] AS T
   INNER JOIN sys.[indexes] I ON T.[object_id] = I.[object_id]
-WHERE T.[is_ms_shipped] = 0 AND I.[type_desc] <> 'HEAP'  AND T.[name] = '{$tableName}'
-ORDER BY T.[name], I.[index_id];";
+  INNER JOIN sys.[schemas] S ON S.schema_id = T.schema_id
+WHERE T.[is_ms_shipped] = 0 AND I.[type_desc] <> 'HEAP' AND S.[name] = '%s' AND T.[name] = '%s'
+ORDER BY T.[name], I.[index_id];",
+            $parts['schema'],
+            $parts['table'],
+        );
 
         $rows = $this->fetchAll($sql);
         foreach ($rows as $row) {
@@ -803,6 +833,7 @@
      */
     public function getPrimaryKey(string $tableName): array
     {
+        $parts = $this->getSchemaName($tableName);
         $rows = $this->fetchAll(sprintf(
             "SELECT
                     tc.CONSTRAINT_NAME,
@@ -811,9 +842,11 @@
                 JOIN INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS kcu
                     ON tc.CONSTRAINT_NAME = kcu.CONSTRAINT_NAME
                 WHERE CONSTRAINT_TYPE = 'PRIMARY KEY'
+                    AND tc.CONSTRAINT_SCHEMA = '%s'
                     AND tc.TABLE_NAME = '%s'
                 ORDER BY kcu.ORDINAL_POSITION",
-            $tableName
+            $parts['schema'],
+            $parts['table'],
         ));
 
         $primaryKey = [
@@ -862,6 +895,7 @@
      */
     protected function getForeignKeys(string $tableName): array
     {
+        $parts = $this->getSchemaName($tableName);
         $foreignKeys = [];
         $rows = $this->fetchAll(sprintf(
             "SELECT
@@ -873,9 +907,10 @@
                     INFORMATION_SCHEMA.TABLE_CONSTRAINTS AS tc
                     JOIN INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS kcu ON tc.CONSTRAINT_NAME = kcu.CONSTRAINT_NAME
                     JOIN INFORMATION_SCHEMA.CONSTRAINT_COLUMN_USAGE AS ccu ON ccu.CONSTRAINT_NAME = tc.CONSTRAINT_NAME
-                WHERE CONSTRAINT_TYPE = 'FOREIGN KEY' AND tc.TABLE_NAME = '%s'
+                WHERE CONSTRAINT_TYPE = 'FOREIGN KEY' AND tc.TABLE_SCHEMA = '%s' AND tc.TABLE_NAME = '%s'
                 ORDER BY kcu.ORDINAL_POSITION",
-            $tableName
+            $parts['schema'],
+            $parts['table'],
         ));
         foreach ($rows as $row) {
             $foreignKeys[$row['CONSTRAINT_NAME']]['table'] = $row['TABLE_NAME'];
@@ -1171,15 +1206,17 @@
      * Gets the SqlServer Index Definition for an Index object.
      *
      * @param \Migrations\Db\Table\Index $index Index
-     * @param ?string $tableName Table name
+     * @param string $tableName Table name
      * @return string
      */
-    protected function getIndexSqlDefinition(Index $index, ?string $tableName): string
+    protected function getIndexSqlDefinition(Index $index, string $tableName): string
     {
+        $parts = $this->getSchemaName($tableName);
         $columnNames = (array)$index->getColumns();
+
         $indexName = $index->getName();
         if (!is_string($indexName)) {
-            $indexName = sprintf('%s_%s', (string)$tableName, implode('_', $columnNames));
+            $indexName = sprintf('%s_%s', $parts['table'], implode('_', $columnNames));
         }
         $order = $index->getOrder() ?? [];
         $columnNames = array_map(function ($columnName) use ($order) {
@@ -1198,7 +1235,7 @@
             'CREATE %s INDEX %s ON %s (%s) %s;',
             ($index->getType() === Index::UNIQUE ? 'UNIQUE' : ''),
             $indexName,
-            $this->quoteTableName((string)$tableName),
+            $this->quoteTableName($tableName),
             implode(',', $columnNames),
             $includedColumns
         );
@@ -1227,6 +1264,122 @@
         return $def;
     }
 
+    /**
+     * Creates the specified schema.
+     *
+     * @param string $schemaName Schema Name
+     * @return void
+     */
+    public function createSchema(string $schemaName = 'public'): void
+    {
+        if ($this->hasSchema($schemaName) === false) {
+            $sql = sprintf('CREATE SCHEMA %s', $this->quoteColumnName($schemaName));
+            $this->execute($sql);
+        }
+    }
+
+    /**
+     * Checks to see if a schema exists.
+     *
+     * @param string $schemaName Schema Name
+     * @return bool
+     */
+    public function hasSchema(string $schemaName): bool
+    {
+        $sql = sprintf(
+            'SELECT count(*) AS [count]
+             FROM sys.schemas
+             WHERE name = %s',
+            $this->quoteString($schemaName)
+        );
+        $result = $this->fetchRow($sql);
+        if (!$result) {
+            return false;
+        }
+
+        return $result['count'] > 0;
+    }
+
+    /**
+     * Drops the specified schema table.
+     *
+     * @param string $schemaName Schema name
+     * @return void
+     */
+    public function dropSchema(string $schemaName): void
+    {
+        $sql = sprintf('DROP SCHEMA IF EXISTS %s', $this->quoteSchemaName($schemaName));
+        $this->execute($sql);
+
+        foreach ($this->createdTables as $idx => $createdTable) {
+            if ($this->getSchemaName($createdTable)['schema'] === $this->quoteSchemaName($schemaName)) {
+                unset($this->createdTables[$idx]);
+            }
+        }
+    }
+
+    /**
+     * Drops all schemas.
+     *
+     * @return void
+     */
+    public function dropAllSchemas(): void
+    {
+        foreach ($this->getAllSchemas() as $schema) {
+            $this->dropSchema($schema);
+        }
+    }
+
+    /**
+     * Returns schemas.
+     *
+     * @return array
+     */
+    public function getAllSchemas(): array
+    {
+        $sql = "SELECT name
+                FROM sys.schemas
+                WHERE name not in ('information_schema', 'sys', 'guest', 'dbo') AND name not like 'db_%'";
+        $items = $this->fetchAll($sql);
+        $schemaNames = [];
+        foreach ($items as $item) {
+            $schemaNames[] = $item['name'];
+        }
+
+        return $schemaNames;
+    }
+
+    /**
+     * @param string $tableName Table name
+     * @return array
+     */
+    protected function getSchemaName(string $tableName): array
+    {
+        $schema = $this->getGlobalSchemaName();
+        $table = $tableName;
+        if (strpos($tableName, '.') !== false) {
+            [$schema, $table] = explode('.', $tableName);
+        }
+
+        return [
+            'schema' => $schema,
+            'table' => $table,
+        ];
+    }
+
+    /**
+     * Gets the schema name.
+     *
+     * @return string
+     */
+    protected function getGlobalSchemaName(): string
+    {
+        $options = $this->getOptions();
+        $config = $options['connection']->config() ?? [];
+
+        return empty($config['schema']) ? $this->schema : $config['schema'];
+    }
+
     /**
      * @inheritDoc
      */

tests/TestCase/Db/Adapter/PostgresAdapterTest.php

@@ -133,6 +133,30 @@ public function testQuoteSchemaName()
         $this->assertEquals('"schema.schema"', $this->adapter->quoteSchemaName('schema.schema'));
     }
 
+    public function testGetGlobalSchemaName()
+    {
+        $config = ConnectionManager::getConfig('test');
+        $config['schema'] = 'test_schema';
+        ConnectionManager::setConfig('test-schema', $config);
+        // Emulate the results of Util::parseDsn()
+        $this->config = [
+            'adapter' => 'postgres',
+            'connection' => ConnectionManager::get('test-schema'),
+            'database' => $config['database'],
+        ];
+
+        $this->adapter = new PostgresAdapter($this->config, $this->getConsoleIo());
+
+        $this->adapter->dropAllSchemas();
+        $this->adapter->createSchema('test_schema');
+
+        $this->adapter->disconnect();
+
+        $this->assertEquals('"test_schema"."table"', $this->adapter->quoteTableName('table'));
+
+        ConnectionManager::drop('test-schema');
+    }
+
     public function testQuoteTableName()
     {
         $this->assertEquals('"public"."table"', $this->adapter->quoteTableName('table'));