Keep supported indexes when converting table to columnar

src/backend/distributed/commands/alter_table.c

@@ -181,6 +181,11 @@ static TableConversionReturn * AlterTableSetAccessMethod(
 static TableConversionReturn * ConvertTable(TableConversionState *con);
 static bool SwitchToSequentialAndLocalExecutionIfShardNameTooLong(char *relationName,
 																  char *longestShardName);
+static void DropIndexesNotSupportedByColumnar(Oid relationId,
+											  bool suppressNoticeMessages);
+static char * GetIndexAccessMethodName(Oid indexId);
+static void DropConstraintRestrict(Oid relationId, Oid constraintId);
+static void DropIndexRestrict(Oid indexId);
 static void EnsureTableNotReferencing(Oid relationId, char conversionType);
 static void EnsureTableNotReferenced(Oid relationId, char conversionType);
 static void EnsureTableNotForeign(Oid relationId);
@@ -537,26 +542,17 @@ ConvertTable(TableConversionState *con)
 	List *preLoadCommands = GetPreLoadTableCreationCommands(con->relationId, true,
 															newAccessMethod);
 
-	bool includeIndexes = true;
 	if (con->accessMethod && strcmp(con->accessMethod, "columnar") == 0)
 	{
-		if (!con->suppressNoticeMessages)
+		DropIndexesNotSupportedByColumnar(con->relationId,
-		{
+										  con->suppressNoticeMessages);
-			List *explicitIndexesOnTable = GetExplicitIndexOidList(con->relationId);
-			Oid indexOid = InvalidOid;
-			foreach_oid(indexOid, explicitIndexesOnTable)
-			{
-				ereport(NOTICE, (errmsg("the index %s on table %s will be dropped, "
-										"because columnar tables cannot have indexes",
-										get_rel_name(indexOid),
-										quote_qualified_identifier(con->schemaName,
-																   con->relationName))));
-			}
-		}
-
-		includeIndexes = false;
 	}
 
+	/*
+	 * Since we already dropped unsupported indexes, we can safely pass
+	 * includeIndexes to be true.
+	 */
+	bool includeIndexes = true;
 	bool includeReplicaIdentity = true;
 	List *postLoadCommands = GetPostLoadTableCreationCommands(con->relationId,
 															  includeIndexes,
@@ -825,6 +821,117 @@ ConvertTable(TableConversionState *con)
 }
 
 
+/*
+ * DropIndexesNotSupportedByColumnar is a helper function used during accces
+ * method conversion to drop the indexes that are not supported by columnarAM.
+ */
+static void
+DropIndexesNotSupportedByColumnar(Oid relationId, bool suppressNoticeMessages)
+{
+	Relation columnarRelation = RelationIdGetRelation(relationId);
+	List *indexIdList = RelationGetIndexList(columnarRelation);
+
+	/*
+	 * Immediately close the relation since we might execute ALTER TABLE
+	 * for that relation.
+	 */
+	RelationClose(columnarRelation);
+
+	Oid indexId = InvalidOid;
+	foreach_oid(indexId, indexIdList)
+	{
+		char *indexAmName = GetIndexAccessMethodName(indexId);
+		if (ColumnarSupportsIndexAM(indexAmName))
+		{
+			continue;
+		}
+
+		if (!suppressNoticeMessages)
+		{
+			ereport(NOTICE, (errmsg("unsupported access method for index %s "
+									"on columnar table %s, given index and "
+									"the constraint depending on the index "
+									"(if any) will be dropped",
+									get_rel_name(indexId),
+									generate_qualified_relation_name(relationId))));
+		}
+
+		Oid constraintId = get_index_constraint(indexId);
+		if (OidIsValid(constraintId))
+		{
+			/* index is implied by a constraint, so drop the constraint itself */
+			DropConstraintRestrict(relationId, constraintId);
+		}
+		else
+		{
+			DropIndexRestrict(indexId);
+		}
+	}
+}
+
+
+/*
+ * GetIndexAccessMethodName returns access method name of index with indexId.
+ * If there is no such index, then errors out.
+ */
+static char *
+GetIndexAccessMethodName(Oid indexId)
+{
+	/* fetch pg_class tuple of the index relation */
+	HeapTuple indexTuple = SearchSysCache1(RELOID, ObjectIdGetDatum(indexId));
+	if (!HeapTupleIsValid(indexTuple))
+	{
+		ereport(ERROR, (errmsg("index with oid %u does not exist", indexId)));
+	}
+
+	Form_pg_class indexForm = (Form_pg_class) GETSTRUCT(indexTuple);
+	Oid indexAMId = indexForm->relam;
+	ReleaseSysCache(indexTuple);
+
+	/* fetch pg_am tuple of index' access method */
+	HeapTuple indexAMTuple = SearchSysCache1(AMOID, ObjectIdGetDatum(indexAMId));
+	if (!HeapTupleIsValid(indexAMTuple))
+	{
+		ereport(ERROR, (errmsg("access method with oid %u does not exist", indexAMId)));
+	}
+
+	Form_pg_am indexAMForm = (Form_pg_am) GETSTRUCT(indexAMTuple);
+	char *indexAmName = pstrdup(indexAMForm->amname.data);
+	ReleaseSysCache(indexAMTuple);
+
+	return indexAmName;
+}
+
+
+/*
+ * DropConstraintRestrict drops the constraint with constraintId by using spi.
+ */
+static void
+DropConstraintRestrict(Oid relationId, Oid constraintId)
+{
+	char *qualifiedRelationName = generate_qualified_relation_name(relationId);
+	char *constraintName = get_constraint_name(constraintId);
+	const char *quotedConstraintName = quote_identifier(constraintName);
+	StringInfo dropConstraintCommand = makeStringInfo();
+	appendStringInfo(dropConstraintCommand, "ALTER TABLE %s DROP CONSTRAINT %s RESTRICT;",
+					 qualifiedRelationName, quotedConstraintName);
+	ExecuteQueryViaSPI(dropConstraintCommand->data, SPI_OK_UTILITY);
+}
+
+
+/*
+ * DropIndexRestrict drops the index with indexId by using spi.
+ */
+static void
+DropIndexRestrict(Oid indexId)
+{
+	char *qualifiedIndexName = generate_qualified_relation_name(indexId);
+	StringInfo dropIndexCommand = makeStringInfo();
+	appendStringInfo(dropIndexCommand, "DROP INDEX %s RESTRICT;", qualifiedIndexName);
+	ExecuteQueryViaSPI(dropIndexCommand->data, SPI_OK_UTILITY);
+}
+
+
 /*
  * EnsureTableNotReferencing checks if the table has a reference to another
  * table and errors if it is.

src/test/regress/expected/alter_table_set_access_method.out

@@ -301,7 +301,6 @@ SELECT event FROM time_partitioned ORDER BY 1;
 \set VERBOSITY default
 DROP TABLE time_partitioned;
 -- test altering a table with index to columnar
--- the index will be dropped
 CREATE TABLE index_table (a INT) ;
 CREATE INDEX idx1 ON index_table (a);
 -- also create an index with statistics
@@ -321,8 +320,6 @@ SELECT a.amname FROM pg_class c, pg_am a where c.relname = 'index_table' AND c.r
 (1 row)
 
 SELECT alter_table_set_access_method('index_table', 'columnar');
-NOTICE:  the index idx1 on table alter_table_set_access_method.index_table will be dropped, because columnar tables cannot have indexes
-NOTICE:  the index idx2 on table alter_table_set_access_method.index_table will be dropped, because columnar tables cannot have indexes
 NOTICE:  creating a new table for alter_table_set_access_method.index_table
 NOTICE:  moving the data of alter_table_set_access_method.index_table
 NOTICE:  dropping the old alter_table_set_access_method.index_table
@@ -335,7 +332,9 @@ NOTICE:  renaming the new table to alter_table_set_access_method.index_table
 SELECT indexname FROM pg_indexes WHERE schemaname = 'alter_table_set_access_method' AND tablename = 'index_table';
  indexname
 ---------------------------------------------------------------------
-(0 rows)
+ idx1
+ idx2
+(2 rows)
 
 SELECT a.amname FROM pg_class c, pg_am a where c.relname = 'index_table' AND c.relnamespace = 'alter_table_set_access_method'::regnamespace AND c.relam = a.oid;
   amname
@@ -343,6 +342,91 @@ SELECT a.amname FROM pg_class c, pg_am a where c.relname = 'index_table' AND c.r
  columnar
 (1 row)
 
+CREATE TABLE "heap_\'tbl" (
+  c1 CIRCLE,
+  c2 TEXT,
+  i int4[],
+  p point,
+  a int,
+  EXCLUDE USING gist
+    (c1 WITH &&, (c2::circle) WITH &&)
+    WHERE (circle_center(c1) <> '(0,0)'),
+  EXCLUDE USING btree
+    (a WITH =)
+	INCLUDE(p)
+	WHERE (c2 < 'astring')
+);
+CREATE INDEX heap_tbl_gin ON "heap_\'tbl" USING gin (i);
+CREATE INDEX "heap_tbl_\'gist" ON "heap_\'tbl" USING gist(p);
+CREATE INDEX heap_tbl_brin ON "heap_\'tbl" USING brin (a) WITH (pages_per_range = 1);
+CREATE INDEX heap_tbl_hash ON "heap_\'tbl" USING hash (c2);
+ALTER TABLE "heap_\'tbl" ADD CONSTRAINT heap_tbl_unique UNIQUE (c2);
+CREATE UNIQUE INDEX "heap_tbl_\'btree" ON "heap_\'tbl" USING btree (a);
+ALTER TABLE "heap_\'tbl" ADD CONSTRAINT "heap_tbl_\'pkey" PRIMARY KEY USING INDEX "heap_tbl_\'btree";
+NOTICE:  ALTER TABLE / ADD CONSTRAINT USING INDEX will rename index "heap_tbl_\'btree" to "heap_tbl_\'pkey"
+SELECT indexname FROM pg_indexes
+WHERE schemaname = 'alter_table_set_access_method' AND
+      tablename = 'heap_\''tbl'
+ORDER BY indexname;
+       indexname
+---------------------------------------------------------------------
+ heap_\'tbl_a_p_excl
+ heap_\'tbl_c1_c2_excl
+ heap_tbl_\'gist
+ heap_tbl_\'pkey
+ heap_tbl_brin
+ heap_tbl_gin
+ heap_tbl_hash
+ heap_tbl_unique
+(8 rows)
+
+SELECT conname FROM pg_constraint
+WHERE conrelid = 'heap_\''tbl'::regclass
+ORDER BY conname;
+        conname
+---------------------------------------------------------------------
+ heap_\'tbl_a_p_excl
+ heap_\'tbl_c1_c2_excl
+ heap_tbl_\'pkey
+ heap_tbl_unique
+(4 rows)
+
+SELECT alter_table_set_access_method('heap_\''tbl', 'columnar');
+NOTICE:  unsupported access method for index heap_\'tbl_c1_c2_excl on columnar table alter_table_set_access_method."heap_\'tbl", given index and the constraint depending on the index (if any) will be dropped
+NOTICE:  unsupported access method for index heap_tbl_gin on columnar table alter_table_set_access_method."heap_\'tbl", given index and the constraint depending on the index (if any) will be dropped
+NOTICE:  unsupported access method for index heap_tbl_\'gist on columnar table alter_table_set_access_method."heap_\'tbl", given index and the constraint depending on the index (if any) will be dropped
+NOTICE:  unsupported access method for index heap_tbl_brin on columnar table alter_table_set_access_method."heap_\'tbl", given index and the constraint depending on the index (if any) will be dropped
+NOTICE:  creating a new table for alter_table_set_access_method."heap_\'tbl"
+NOTICE:  moving the data of alter_table_set_access_method."heap_\'tbl"
+NOTICE:  dropping the old alter_table_set_access_method."heap_\'tbl"
+NOTICE:  renaming the new table to alter_table_set_access_method."heap_\'tbl"
+ alter_table_set_access_method
+---------------------------------------------------------------------
+
+(1 row)
+
+SELECT indexname FROM pg_indexes
+WHERE schemaname = 'alter_table_set_access_method' AND
+      tablename = 'heap_\''tbl'
+ORDER BY indexname;
+      indexname
+---------------------------------------------------------------------
+ heap_\'tbl_a_p_excl
+ heap_tbl_\'pkey
+ heap_tbl_hash
+ heap_tbl_unique
+(4 rows)
+
+SELECT conname FROM pg_constraint
+WHERE conrelid = 'heap_\''tbl'::regclass
+ORDER BY conname;
+       conname
+---------------------------------------------------------------------
+ heap_\'tbl_a_p_excl
+ heap_tbl_\'pkey
+ heap_tbl_unique
+(3 rows)
+
 -- test different table types
 SET client_min_messages to WARNING;
 SELECT 1 FROM master_add_node('localhost', :master_port, groupId := 0);

src/test/regress/sql/alter_table_set_access_method.sql

@@ -102,7 +102,6 @@ SELECT event FROM time_partitioned ORDER BY 1;
 DROP TABLE time_partitioned;
 
 -- test altering a table with index to columnar
--- the index will be dropped
 CREATE TABLE index_table (a INT) USING heap;
 CREATE INDEX idx1 ON index_table (a);
 -- also create an index with statistics
@@ -114,6 +113,50 @@ SELECT alter_table_set_access_method('index_table', 'columnar');
 SELECT indexname FROM pg_indexes WHERE schemaname = 'alter_table_set_access_method' AND tablename = 'index_table';
 SELECT a.amname FROM pg_class c, pg_am a where c.relname = 'index_table' AND c.relnamespace = 'alter_table_set_access_method'::regnamespace AND c.relam = a.oid;
 
+CREATE TABLE "heap_\'tbl" (
+  c1 CIRCLE,
+  c2 TEXT,
+  i int4[],
+  p point,
+  a int,
+  EXCLUDE USING gist
+    (c1 WITH &&, (c2::circle) WITH &&)
+    WHERE (circle_center(c1) <> '(0,0)'),
+  EXCLUDE USING btree
+    (a WITH =)
+	INCLUDE(p)
+	WHERE (c2 < 'astring')
+);
+
+CREATE INDEX heap_tbl_gin ON "heap_\'tbl" USING gin (i);
+CREATE INDEX "heap_tbl_\'gist" ON "heap_\'tbl" USING gist(p);
+CREATE INDEX heap_tbl_brin ON "heap_\'tbl" USING brin (a) WITH (pages_per_range = 1);
+
+CREATE INDEX heap_tbl_hash ON "heap_\'tbl" USING hash (c2);
+ALTER TABLE "heap_\'tbl" ADD CONSTRAINT heap_tbl_unique UNIQUE (c2);
+
+CREATE UNIQUE INDEX "heap_tbl_\'btree" ON "heap_\'tbl" USING btree (a);
+ALTER TABLE "heap_\'tbl" ADD CONSTRAINT "heap_tbl_\'pkey" PRIMARY KEY USING INDEX "heap_tbl_\'btree";
+
+SELECT indexname FROM pg_indexes
+WHERE schemaname = 'alter_table_set_access_method' AND
+      tablename = 'heap_\''tbl'
+ORDER BY indexname;
+
+SELECT conname FROM pg_constraint
+WHERE conrelid = 'heap_\''tbl'::regclass
+ORDER BY conname;
+
+SELECT alter_table_set_access_method('heap_\''tbl', 'columnar');
+
+SELECT indexname FROM pg_indexes
+WHERE schemaname = 'alter_table_set_access_method' AND
+      tablename = 'heap_\''tbl'
+ORDER BY indexname;
+
+SELECT conname FROM pg_constraint
+WHERE conrelid = 'heap_\''tbl'::regclass
+ORDER BY conname;
 
 -- test different table types
 SET client_min_messages to WARNING;