Add support for included columns for indexes starting from PostgreSQL 12

ChangeLog/7.2.0-RC-dev.txt

@@ -0,0 +1 @@
+[postgresql] Support for included columns for indexes starting from PostgreSQL 12

Orm/Xtensive.Orm.PostgreSql/Sql.Drivers.PostgreSql/v12_0/Extractor.cs

@@ -2,6 +2,10 @@
 // This code is distributed under MIT license terms.
 // See the License.txt file in the project root for more information.
 
+using System;
+using System.Data.Common;
+using System.Linq;
+using System.Text.RegularExpressions;
 using Xtensive.Sql.Model;
 
 namespace Xtensive.Sql.Drivers.PostgreSql.v12_0
@@ -13,7 +17,10 @@ protected override void BuildPgCatalogSchema(Schema schema)
     {
       base.BuildPgCatalogSchema(schema);
       var defaultValuesTable = schema.Tables["pg_attrdef"];
-      defaultValuesTable.CreateColumn("adbin", new SqlValueType(SqlType.Binary));
+      _ = defaultValuesTable.CreateColumn("adbin", new SqlValueType(SqlType.Binary));
+
+      var indexesTable = schema.Tables["pg_index"];
+      CreateInt2Column(indexesTable, "indnkeyatts");
     }
 
     /// <inheritdoc/>
@@ -77,6 +84,126 @@ protected override ISqlCompileUnit BuildExtractTableAndDomainConstraintsQuery(Ex
       return select;
     }
 
+    /// <inheritdoc />
+    protected override ISqlCompileUnit BuildExtractTableIndexesQuery(ExtractionContext context)
+    {
+      var tableMap = context.TableMap;
+      var tableSpacesTable = PgTablespace;
+      var relationsTable = PgClass;
+      var indexTable = PgIndex;
+      var dependencyTable = PgDepend;
+
+      //subselect that index was not created automatically
+      var subSelect = SqlDml.Select(dependencyTable);
+      subSelect.Where = dependencyTable["classid"] == PgClassOid &&
+        dependencyTable["objid"] == indexTable["indexrelid"] &&
+        dependencyTable["deptype"] == 'i';
+      subSelect.Columns.Add(dependencyTable[0]);
+
+      //not automatically created indexes of our tables
+      var select = SqlDml.Select(indexTable
+        .InnerJoin(relationsTable, relationsTable["oid"] == indexTable["indexrelid"])
+        .LeftOuterJoin(tableSpacesTable, tableSpacesTable["oid"] == relationsTable["reltablespace"]));
+      select.Where = SqlDml.In(indexTable["indrelid"], CreateOidRow(tableMap.Keys)) && !SqlDml.Exists(subSelect);
+      select.Columns.Add(indexTable["indrelid"]);
+      select.Columns.Add(indexTable["indexrelid"]);
+      select.Columns.Add(relationsTable["relname"]);
+      select.Columns.Add(indexTable["indisunique"]);
+      select.Columns.Add(indexTable["indisclustered"]);
+      select.Columns.Add(indexTable["indkey"]);
+      select.Columns.Add(tableSpacesTable["spcname"]);
+      select.Columns.Add(indexTable["indnatts"]);
+      select.Columns.Add(indexTable["indnkeyatts"]);
+      select.Columns.Add(SqlDml.FunctionCall("pg_get_expr", indexTable["indexprs"], indexTable["indrelid"], true),
+        "indexprstext");
+      select.Columns.Add(SqlDml.FunctionCall("pg_get_expr", indexTable["indpred"], indexTable["indrelid"], true),
+        "indpredtext");
+      select.Columns.Add(SqlDml.FunctionCall("pg_get_indexdef", indexTable["indexrelid"]), "inddef");
+      AddSpecialIndexQueryColumns(select, tableSpacesTable, relationsTable, indexTable, dependencyTable);
+      return select;
+    }
+
+    /// <inheritdoc />
+    protected override int ReadTableIndexData(DbDataReader dataReader, ExtractionContext context)
+    {
+      var tableMap = context.TableMap;
+      var tableColumns = context.TableColumnMap;
+
+      var maxColumnNumber = 0;
+      var tableIdentifier = Convert.ToInt64(dataReader["indrelid"]);
+      var indexIdentifier = Convert.ToInt64(dataReader["indexrelid"]);
+      var indexName = dataReader["relname"].ToString();
+      var isUnique = dataReader.GetBoolean(dataReader.GetOrdinal("indisunique"));
+      var indexKey = (short[]) dataReader["indkey"];
+
+      var tablespaceName = (dataReader["spcname"] != DBNull.Value) ? dataReader["spcname"].ToString() : null;
+      var filterExpression = (dataReader["indpredtext"] != DBNull.Value)
+        ? dataReader["indpredtext"].ToString()
+        : string.Empty;
+
+      var table = tableMap[tableIdentifier];
+
+      var fullTextRegex =
+        @"(?<=CREATE INDEX \S+ ON \S+ USING (?:gist|gin)(?:\s|\S)*)to_tsvector\('(\w+)'::regconfig, \(*(?:(?:\s|\)|\(|\|)*(?:\(""(\S+)""\)|'\s')::text)+\)";
+      var indexScript = dataReader["inddef"].ToString();
+      var matches = Regex.Matches(indexScript, fullTextRegex, RegexOptions.Compiled);
+      if (matches.Count > 0) {
+        // Fulltext index
+        var fullTextIndex = table.CreateFullTextIndex(indexName);
+        foreach (Match match in matches) {
+          var columnConfigurationName = match.Groups[1].Value;
+          foreach (Capture capture in match.Groups[2].Captures) {
+            var columnName = capture.Value;
+            var fullTextColumn = fullTextIndex.Columns[columnName]
+              ?? fullTextIndex.CreateIndexColumn(table.Columns.Single(column => column.Name == columnName));
+            if (fullTextColumn.Languages[columnConfigurationName] == null)
+              fullTextColumn.Languages.Add(new Language(columnConfigurationName));
+          }
+        }
+      }
+      else {
+        //Regular index
+        var index = table.CreateIndex(indexName);
+        index.IsBitmap = false;
+        index.IsUnique = isUnique;
+        index.Filegroup = tablespaceName;
+        if (!string.IsNullOrEmpty(filterExpression))
+          index.Where = SqlDml.Native(filterExpression);
+
+        // Expression-based index
+        var some = dataReader["indexprstext"];
+        if (some != DBNull.Value) {
+          context.ExpressionIndexMap[indexIdentifier] = new ExpressionIndexInfo(index, indexKey);
+          maxColumnNumber = dataReader.GetInt16(dataReader.GetOrdinal("indnatts"));
+        }
+        else {
+          var keyColumnNumber = dataReader.GetInt16(dataReader.GetOrdinal("indnkeyatts"));
+          for (int j = 0; j < indexKey.Length; j++) {
+            if (j < keyColumnNumber) {
+              int colIndex = indexKey[j];
+              if (colIndex > 0) {
+                _ = index.CreateIndexColumn(tableColumns[tableIdentifier][colIndex], true);
+              }
+              else {
+                //column index is 0
+                //this means that this index column is an expression
+                //which is not possible with SqlDom tables
+              }
+            }
+            else {
+              int colIndex = indexKey[j];
+              index.NonkeyColumns.Add(tableColumns[tableIdentifier][colIndex]);
+            }
+          }
+        }
+
+        ReadSpecialIndexProperties(dataReader, index);
+      }
+
+      return maxColumnNumber;
+    }
+
+
     // Constructors
 
     public Extractor(SqlDriver driver)

Orm/Xtensive.Orm.PostgreSql/Sql.Drivers.PostgreSql/v12_0/ServerInfoProvider.cs

@@ -2,10 +2,14 @@
 // This code is distributed under MIT license terms.
 // See the License.txt file in the project root for more information.
 
+using Xtensive.Sql.Info;
+
 namespace Xtensive.Sql.Drivers.PostgreSql.v12_0
 {
   internal class ServerInfoProvider : v10_0.ServerInfoProvider
   {
+    protected override IndexFeatures GetIndexFeatures() => base.GetIndexFeatures() | IndexFeatures.NonKeyColumns;
+
     // Constructors
 
     public ServerInfoProvider(SqlDriver driver)

Orm/Xtensive.Orm.PostgreSql/Sql.Drivers.PostgreSql/v12_0/Translator.cs

@@ -2,10 +2,35 @@
 // This code is distributed under MIT license terms.
 // See the License.txt file in the project root for more information.
 
+using Xtensive.Sql.Compiler;
+using Xtensive.Sql.Ddl;
+
 namespace Xtensive.Sql.Drivers.PostgreSql.v12_0
 {
   internal class Translator : v10_0.Translator
   {
+    public override void Translate(SqlCompilerContext context, SqlCreateIndex node, CreateIndexSection section)
+    {
+      var index = node.Index;
+      if (!index.IsFullText) {
+        var output = context.Output;
+        switch (section) {
+          case CreateIndexSection.NonkeyColumnsEnter:
+            _ = output.AppendOpeningPunctuation("INCLUDE (");
+            break;
+          case CreateIndexSection.NonkeyColumnsExit:
+            _ = output.AppendClosingPunctuation(")");
+            break;
+          default:
+            base.Translate(context, node, section);
+            break;
+        }
+      }
+      else {
+        base.Translate(context, node, section);
+      }
+    }
+
     // Constructors
 
     public Translator(SqlDriver driver)

Orm/Xtensive.Orm.PostgreSql/Sql.Drivers.PostgreSql/v8_0/Translator.cs

@@ -294,22 +294,30 @@ public override void Translate(SqlCompilerContext context, SqlCreateIndex node,
           if (index.IsSpatial) {
             _ = output.Append(" USING GIST");
           }
+          break;
+        case CreateIndexSection.ColumnsEnter:
           _ = output.Append("(");
           break;
-        case CreateIndexSection.StorageOptions:
+        case CreateIndexSection.ColumnsExit:
           _ = output.Append(")");
+          break;
+        case CreateIndexSection.NonkeyColumnsEnter:
+          break;
+        case CreateIndexSection.NonkeyColumnsExit:
+          break;
+        case CreateIndexSection.StorageOptions:
           AppendIndexStorageParameters(output, index);
           if (!string.IsNullOrEmpty(index.Filegroup)) {
             _ = output.Append(" TABLESPACE ");
             TranslateIdentifier(output, index.Filegroup);
           }
 
-          break;
-        case CreateIndexSection.Exit:
           break;
         case CreateIndexSection.Where:
           _ = output.Append(" WHERE ");
           break;
+        case CreateIndexSection.Exit:
+          break;
         default:
           break;
           ;

Orm/Xtensive.Orm.PostgreSql/Sql.Drivers.PostgreSql/v8_3/Translator.cs

@@ -31,7 +31,10 @@ public override void Translate(SqlCompilerContext context, SqlCreateIndex node,
           TranslateIdentifier(output, index.Name);
           _ = output.Append(" ON ");
           Translate(context, index.DataTable);
-          _ = output.Append(" USING gin (");
+          _ = output.Append(" USING gin ");
+          break;
+        case CreateIndexSection.ColumnsEnter:
+          _ = output.Append("(");
           break;
         case CreateIndexSection.ColumnsExit:
           // Add actual columns list

Orm/Xtensive.Orm.Tests.Sql/PostgreSql/ExtractorTest.cs

@@ -92,10 +92,23 @@ protected override string GetForeignKeyExtractionCleanUpScript() =>
 
     protected override string GetIndexExtractionPrepareScript(string tableName)
     {
-      return
-        $"CREATE TABLE \"{tableName}\" (\"column1\" int, \"column2\" int);" +
-        $"\n CREATE INDEX \"{tableName}_index1_desc_asc\" on \"{tableName}\" (\"column1\" desc, \"column2\" asc);" +
-        $"\n CREATE UNIQUE INDEX \"{tableName}_index1_u_asc_desc\" on \"{tableName}\" (\"column1\" asc, \"column2\" desc);";
+      // CREATE TABLE table1 (column1 int,  column2 int);
+      // CREATE INDEX table1_index1_desc_asc on table1 (column1 desc, column2 asc);
+      // CREATE UNIQUE INDEX table1_index1_u_asc_desc on table1 (column1 asc, column2 desc);
+      // CREATE UNIQUE INDEX table1_index_with_included_columns on table1 (column1 asc) include (column2);
+      if (NonKeyColumnsSupported) {
+        return
+          $"CREATE TABLE \"{tableName}\" (\"column1\" int, \"column2\" int);" +
+          $"\n CREATE INDEX \"{tableName}_index1_desc_asc\" on \"{tableName}\" (\"column1\" desc, \"column2\" asc);" +
+          $"\n CREATE UNIQUE INDEX \"{tableName}_index1_u_asc_desc\" on \"{tableName}\" (\"column1\" asc, \"column2\" desc);" +
+          $"\n CREATE UNIQUE INDEX \"{tableName}_index_with_included_columns\" on \"{tableName}\" (\"column1\" asc) include (\"column2\");";
+      }
+      else {
+        return
+          $"CREATE TABLE \"{tableName}\" (\"column1\" int, \"column2\" int);" +
+          $"\n CREATE INDEX \"{tableName}_index1_desc_asc\" on \"{tableName}\" (\"column1\" desc, \"column2\" asc);" +
+          $"\n CREATE UNIQUE INDEX \"{tableName}_index1_u_asc_desc\" on \"{tableName}\" (\"column1\" asc, \"column2\" desc);";
+      }
     }
     protected override string GetIndexExtractionCleanUpScript(string tableName) => $"drop table \"{tableName}\";";
 

Orm/Xtensive.Orm.Tests/Storage/IgnoreRulesValidateTest.cs

@@ -344,6 +344,7 @@ public class IgnoreRulesValidateTest
 
     private readonly bool createConstraintsWithTable = StorageProviderInfo.Instance.Provider == StorageProvider.Sqlite;
     private readonly bool noExceptionOnIndexKeyColumnDrop = StorageProviderInfo.Instance.Provider.In(StorageProvider.PostgreSql, StorageProvider.MySql);
+    private readonly bool noExceptionOnIndexIncludedColumnDrop = StorageProviderInfo.Instance.Provider.In(StorageProvider.PostgreSql);
     private readonly SqlDriver sqlDriver = TestSqlDriver.Create(GetConnectionInfo());
 
     private Key changedOrderKey;
@@ -646,8 +647,13 @@ public void DropIncludedColumnOfIgnoredIndexTest()
       var ignoreRuleCollection = new IgnoreRuleCollection();
       _ = ignoreRuleCollection.IgnoreIndex("IX_Ignored_Index").WhenTable("MyEntity2");
 
-      _ = Assert.Throws<StorageException>(
-        () => BuildDomain(DomainUpgradeMode.Perform, ignoreRuleCollection, model5Types).Dispose());
+      if (noExceptionOnIndexIncludedColumnDrop) {
+        BuildDomain(DomainUpgradeMode.Perform, ignoreRuleCollection, model6Types).Dispose();
+      }
+      else {
+        _ = Assert.Throws<StorageException>(
+          () => BuildDomain(DomainUpgradeMode.Perform, ignoreRuleCollection, model5Types).Dispose());
+      }
     }
 
     [Test]