Allow hypertable constraint creation USING INDEX

Prior to this commit, we disallow
```sql
ALTER TABLE foo ADD CONSTRAINT foo_pkey
PRIMARY KEY USING INDEX foo_pkey
```

Signed-off-by: Arunprasad Rajkumar <[email protected]>

sql/chunk_constraint.sql

@@ -4,7 +4,8 @@
 
 -- create constraint on newly created chunk based on hypertable constraint
 CREATE OR REPLACE FUNCTION _timescaledb_functions.chunk_constraint_add_table_constraint(
-    chunk_constraint_row  _timescaledb_catalog.chunk_constraint
+    chunk_constraint_row  _timescaledb_catalog.chunk_constraint,
+    using_index BOOLEAN
 )
     RETURNS VOID LANGUAGE PLPGSQL AS
 $BODY$
@@ -17,6 +18,8 @@ DECLARE
     def TEXT;
     indx_tablespace NAME;
     tablespace_def TEXT;
+    chunk_constraint_index_name NAME;
+    row_record RECORD;
 BEGIN
     SELECT * INTO STRICT chunk_row FROM _timescaledb_catalog.chunk c WHERE c.id = chunk_constraint_row.chunk_id;
     SELECT * INTO STRICT hypertable_row FROM _timescaledb_catalog.hypertable h WHERE h.id = chunk_row.hypertable_id;
@@ -30,15 +33,37 @@ BEGIN
               conrelid = format('%I.%I', hypertable_row.schema_name, hypertable_row.table_name)::regclass::oid;
 
         IF constraint_type IN ('p','u') THEN
-          -- since primary keys and unique constraints are backed by an index
-          -- they might have an index tablespace assigned
-          -- the tablspace is not part of the constraint definition so
-          -- we have to append it explicitly to preserve it
-          SELECT T.spcname INTO indx_tablespace
-          FROM pg_constraint C, pg_class I, pg_tablespace T
-          WHERE C.oid = constraint_oid AND C.contype IN ('p', 'u') AND I.oid = C.conindid AND I.reltablespace = T.oid;
+          IF using_index THEN
+            -- indexes created for constraints are named after the constraint
+            -- so we can find the index name by looking up the constraint name.
+            -- Drop the entry in chunk_index and it will be added back later
+            -- with the correct index name by the existing flow.
+            DELETE FROM _timescaledb_catalog.chunk_index
+            WHERE chunk_id = chunk_row.id AND hypertable_index_name = chunk_constraint_row.hypertable_constraint_name RETURNING index_name INTO STRICT chunk_constraint_index_name;
 
-          def := pg_get_constraintdef(constraint_oid);
+            CASE constraint_type
+              WHEN 'p' THEN
+                def := pg_catalog.format(
+                    $$ PRIMARY KEY USING INDEX %I $$,
+                    chunk_constraint_index_name
+                );
+              WHEN 'u' THEN
+                def := pg_catalog.format(
+                    $$ UNIQUE USING INDEX %I $$,
+                    chunk_constraint_index_name
+                );
+            END CASE;
+          ELSE
+            -- since primary keys and unique constraints are backed by an index
+            -- they might have an index tablespace assigned
+            -- the tablspace is not part of the constraint definition so
+            -- we have to append it explicitly to preserve it
+            SELECT T.spcname INTO indx_tablespace
+            FROM pg_constraint C, pg_class I, pg_tablespace T
+            WHERE C.oid = constraint_oid AND C.contype IN ('p', 'u') AND I.oid = C.conindid AND I.reltablespace = T.oid;
+
+            def := pg_get_constraintdef(constraint_oid);
+          END IF;
 
         ELSIF constraint_type = 't' THEN
           -- constraint triggers are copied separately with normal triggers

src/chunk_constraint.c

@@ -373,7 +373,7 @@ create_dimension_check_constraint(const Dimension *dim, const DimensionSlice *sl
  * Add a constraint to a chunk table.
  */
 static Oid
-chunk_constraint_create_on_table(const ChunkConstraint *cc, Oid chunk_oid)
+chunk_constraint_create_on_table(const ChunkConstraint *cc, Oid chunk_oid, bool using_index)
 {
 	HeapTuple tuple;
 	Datum values[Natts_chunk_constraint];
@@ -388,7 +388,9 @@ chunk_constraint_create_on_table(const ChunkConstraint *cc, Oid chunk_oid)
 	RelationClose(rel);
 
 	ts_catalog_database_info_become_owner(ts_catalog_database_info_get(), &sec_ctx);
-	CatalogInternalCall1(DDL_ADD_CHUNK_CONSTRAINT, HeapTupleGetDatum(tuple));
+	CatalogInternalCall2(DDL_ADD_CHUNK_CONSTRAINT,
+						 HeapTupleGetDatum(tuple),
+						 BoolGetDatum(using_index));
 	ts_catalog_restore_user(&sec_ctx);
 	heap_freetuple(tuple);
 
@@ -401,14 +403,14 @@ chunk_constraint_create_on_table(const ChunkConstraint *cc, Oid chunk_oid)
  */
 static Oid
 create_non_dimensional_constraint(const ChunkConstraint *cc, Oid chunk_oid, int32 chunk_id,
-								  Oid hypertable_oid, int32 hypertable_id)
+								  Oid hypertable_oid, int32 hypertable_id, bool using_index)
 {
 	Oid chunk_constraint_oid;
 
 	Assert(!is_dimension_constraint(cc));
 
 	ts_process_utility_set_expect_chunk_modification(true);
-	chunk_constraint_oid = chunk_constraint_create_on_table(cc, chunk_oid);
+	chunk_constraint_oid = chunk_constraint_create_on_table(cc, chunk_oid, using_index);
 	ts_process_utility_set_expect_chunk_modification(false);
 
 	/*
@@ -493,7 +495,8 @@ ts_chunk_constraints_create(const Hypertable *ht, const Chunk *chunk)
 											  chunk->table_id,
 											  chunk->fd.id,
 											  ht->main_table_relid,
-											  ht->fd.id);
+											  ht->fd.id,
+											  false);
 		}
 	}
 
@@ -851,7 +854,8 @@ ts_chunk_constraints_add_inheritable_check_constraints(ChunkConstraints *ccs, in
 }
 
 void
-ts_chunk_constraint_create_on_chunk(const Hypertable *ht, const Chunk *chunk, Oid constraint_oid)
+ts_chunk_constraint_create_on_chunk(const Hypertable *ht, const Chunk *chunk, Oid constraint_oid,
+									bool using_index)
 {
 	HeapTuple tuple;
 	Form_pg_constraint con;
@@ -876,7 +880,8 @@ ts_chunk_constraint_create_on_chunk(const Hypertable *ht, const Chunk *chunk, Oi
 										  chunk->table_id,
 										  chunk->fd.id,
 										  ht->main_table_relid,
-										  ht->fd.id);
+										  ht->fd.id,
+										  using_index);
 	}
 
 	ReleaseSysCache(tuple);

src/chunk_constraint.h

@@ -60,7 +60,7 @@ extern TSDLLEXPORT int ts_chunk_constraints_add_inheritable_check_constraints(
 extern TSDLLEXPORT void ts_chunk_constraints_insert_metadata(const ChunkConstraints *ccs);
 extern TSDLLEXPORT void ts_chunk_constraints_create(const Hypertable *ht, const Chunk *chunk);
 extern void ts_chunk_constraint_create_on_chunk(const Hypertable *ht, const Chunk *chunk,
-												Oid constraint_oid);
+												Oid constraint_oid, bool using_index);
 extern int ts_chunk_constraint_delete_by_hypertable_constraint_name(
 	int32 chunk_id, const char *hypertable_constraint_name, bool delete_metadata,
 	bool drop_constraint);

src/chunk_index.c

@@ -1016,12 +1016,11 @@ ts_chunk_index_rename(Chunk *chunk, Oid chunk_indexrelid, const char *new_name)
 }
 
 int
-ts_chunk_index_rename_parent(Hypertable *ht, Oid hypertable_indexrelid, const char *new_name)
+ts_chunk_index_rename_parent(Hypertable *ht, const char *old_name, const char *new_name)
 {
 	ScanKeyData scankey[2];
-	const char *indexname = get_rel_name(hypertable_indexrelid);
 	ChunkIndexRenameInfo renameinfo = {
-		.oldname = indexname,
+		.oldname = old_name,
 		.newname = new_name,
 		.isparent = true,
 	};
@@ -1035,7 +1034,7 @@ ts_chunk_index_rename_parent(Hypertable *ht, Oid hypertable_indexrelid, const ch
 				Anum_chunk_index_hypertable_id_hypertable_index_name_idx_hypertable_index_name,
 				BTEqualStrategyNumber,
 				F_NAMEEQ,
-				CStringGetDatum(indexname));
+				CStringGetDatum(old_name));
 
 	return chunk_index_scan_update(CHUNK_INDEX_HYPERTABLE_ID_HYPERTABLE_INDEX_NAME_IDX,
 								   scankey,

src/chunk_index.h

@@ -41,8 +41,7 @@ extern int ts_chunk_index_delete_by_chunk_id(int32 chunk_id, bool drop_index);
 extern void ts_chunk_index_delete_by_name(const char *schema, const char *index_name,
 										  bool drop_index);
 extern int ts_chunk_index_rename(Chunk *chunk, Oid chunk_indexrelid, const char *new_name);
-extern int ts_chunk_index_rename_parent(Hypertable *ht, Oid hypertable_indexrelid,
-										const char *new_name);
+extern int ts_chunk_index_rename_parent(Hypertable *ht, const char *oldname, const char *new_name);
 extern int ts_chunk_index_adjust_meta(int32 chunk_id, const char *ht_index_name,
 									  const char *old_name, const char *new_name);
 extern int ts_chunk_index_set_tablespace(Hypertable *ht, Oid hypertable_indexrelid,

src/process_utility.c

@@ -1969,7 +1969,8 @@ process_rename_index(ProcessUtilityArgs *args, Cache *hcache, Oid relid, RenameS
 
 	if (NULL != ht)
 	{
-		ts_chunk_index_rename_parent(ht, relid, stmt->newname);
+		const char *indexname = get_rel_name(relid);
+		ts_chunk_index_rename_parent(ht, indexname, stmt->newname);
 
 		add_hypertable_to_process_args(args, ht);
 	}
@@ -2180,13 +2181,33 @@ process_altertable_change_owner(Hypertable *ht, AlterTableCmd *cmd)
 	}
 }
 
+static void
+process_add_constraint_chunk_using_index(Hypertable *ht, Oid chunk_relid, void *arg)
+{
+	Oid hypertable_constraint_oid = *((Oid *) arg);
+	Chunk *chunk = ts_chunk_get_by_relid(chunk_relid, true);
+
+	ts_chunk_constraint_create_on_chunk(ht, chunk, hypertable_constraint_oid, true);
+}
+
+static void
+process_altertable_add_constraint_using_index(Hypertable *ht, const char *constraint_name)
+{
+	Oid hypertable_constraint_oid =
+		get_relation_constraint_oid(ht->main_table_relid, constraint_name, false);
+
+	Assert(constraint_name != NULL);
+
+	foreach_chunk(ht, process_add_constraint_chunk_using_index, &hypertable_constraint_oid);
+}
+
 static void
 process_add_constraint_chunk(Hypertable *ht, Oid chunk_relid, void *arg)
 {
 	Oid hypertable_constraint_oid = *((Oid *) arg);
 	Chunk *chunk = ts_chunk_get_by_relid(chunk_relid, true);
 
-	ts_chunk_constraint_create_on_chunk(ht, chunk, hypertable_constraint_oid);
+	ts_chunk_constraint_create_on_chunk(ht, chunk, hypertable_constraint_oid, false);
 }
 
 static void
@@ -2276,6 +2297,25 @@ verify_constraint_plaintable(RangeVar *relation, Constraint *constr)
 	ts_cache_release(hcache);
 }
 
+/*
+ * Postgres does not allow having different names for the constraint and the
+ * index it is using, so we need to rename the index to match the constraint
+ * name.
+ */
+static void
+rename_constraint_index(Hypertable *ht, Constraint *constr)
+{
+	ConstrType contype = constr->contype;
+	const char *conname = constr->conname;
+	const char *indexname = constr->indexname;
+
+	if (contype == CONSTR_UNIQUE || contype == CONSTR_PRIMARY)
+	{
+		if (indexname != NULL)
+			ts_chunk_index_rename_parent(ht, indexname, conname);
+	}
+}
+
 /*
  * Verify that a constraint is supported on a hypertable.
  */
@@ -3390,9 +3430,14 @@ process_altertable_start_table(ProcessUtilityArgs *args)
 				Assert(IsA(cmd->def, Constraint));
 
 				if (NULL == ht)
+				{
 					verify_constraint_plaintable(stmt->relation, (Constraint *) cmd->def);
+				}
 				else
+				{
 					verify_constraint_hypertable(ht, cmd->def);
+					rename_constraint_index(ht, (Constraint *) cmd->def);
+				}
 				break;
 			case AT_AlterColumnType:
 				Assert(IsA(cmd->def, ColumnDef));
@@ -3647,11 +3692,21 @@ process_altertable_end_subcmd(Hypertable *ht, Node *parsetree, ObjectAddress *ob
 			process_altertable_change_owner(ht, cmd);
 			break;
 		case AT_AddIndexConstraint:
-			ereport(ERROR,
-					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
-					 errmsg("hypertables do not support adding a constraint "
-							"using an existing index")));
-			break;
+		{
+			IndexStmt *stmt = (IndexStmt *) cmd->def;
+
+			Assert(IsA(cmd->def, IndexStmt));
+
+			const char *idxname = stmt->idxname;
+
+			if (idxname == NULL)
+			{
+				idxname = get_rel_name(obj->objectId);
+			}
+
+			process_altertable_add_constraint_using_index(ht, idxname);
+		}
+		break;
 		case AT_AddIndex:
 		{
 			IndexStmt *stmt = (IndexStmt *) cmd->def;

src/ts_catalog/catalog.c

@@ -280,7 +280,7 @@ typedef struct InternalFunctionDef
 static const InternalFunctionDef internal_function_definitions[_MAX_INTERNAL_FUNCTIONS] = {
 	[DDL_ADD_CHUNK_CONSTRAINT] = {
 		.name = "chunk_constraint_add_table_constraint",
-		.args = 1,
+		.args = 2,
 	},
 	[DDL_CONSTRAINT_CLONE] = {
 		.name = "constraint_clone",