Allow hypertable constraint creation USING INDEX

.unreleased/fix_7040

@@ -0,0 +1 @@
+Fixes: #7040 Allow hypertable constraint creation USING INDEX

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,39 @@ 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.
+            SELECT index_name INTO STRICT chunk_constraint_index_name FROM _timescaledb_catalog.chunk_index
+            WHERE chunk_id = chunk_row.id AND hypertable_index_name = chunk_constraint_row.hypertable_constraint_name;
 
-          def := pg_get_constraintdef(constraint_oid);
+            IF chunk_constraint_index_name IS NULL THEN
+              RAISE 'index not found for constraint %', chunk_constraint_row;
+            END IF;
+
+            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

sql/compat.sql

@@ -79,12 +79,12 @@ END$$
 SET search_path TO pg_catalog,pg_temp;
 
 
-CREATE OR REPLACE FUNCTION _timescaledb_internal.chunk_constraint_add_table_constraint(chunk_constraint_row _timescaledb_catalog.chunk_constraint) RETURNS void LANGUAGE PLPGSQL AS $$
+CREATE OR REPLACE FUNCTION _timescaledb_internal.chunk_constraint_add_table_constraint(chunk_constraint_row _timescaledb_catalog.chunk_constraint,using_index BOOLEAN) RETURNS void LANGUAGE PLPGSQL AS $$
 BEGIN
   IF current_setting('timescaledb.enable_deprecation_warnings', true)::bool THEN
     RAISE WARNING 'function _timescaledb_internal.chunk_constraint_add_table_constraint(_timescaledb_catalog.chunk_constraint) is deprecated and has been moved to _timescaledb_functions schema. this compatibility function will be removed in a future version.';
   END IF;
-  PERFORM _timescaledb_functions.chunk_constraint_add_table_constraint($1);
+  PERFORM _timescaledb_functions.chunk_constraint_add_table_constraint($1,$2);
 END$$
 SET search_path TO pg_catalog,pg_temp;
 

sql/updates/latest-dev.sql

@@ -2,3 +2,10 @@ CREATE FUNCTION _timescaledb_functions.compressed_data_info(_timescaledb_interna
     RETURNS TABLE (algorithm name, has_nulls bool)
     AS '@MODULE_PATHNAME@', 'ts_update_placeholder'
     LANGUAGE C STRICT IMMUTABLE SET search_path = pg_catalog, pg_temp;
+
+DROP FUNCTION IF EXISTS _timescaledb_functions.chunk_constraint_add_table_constraint(
+    chunk_constraint_row  _timescaledb_catalog.chunk_constraint
+);
+DROP FUNCTION IF EXISTS _timescaledb_internal.chunk_constraint_add_table_constraint(
+    chunk_constraint_row  _timescaledb_catalog.chunk_constraint
+);

sql/updates/reverse-dev.sql

@@ -1 +1,82 @@
 DROP FUNCTION _timescaledb_functions.compressed_data_info(_timescaledb_internal.compressed_data);
+
+DROP FUNCTION IF EXISTS _timescaledb_functions.chunk_constraint_add_table_constraint(
+    chunk_constraint_row  _timescaledb_catalog.chunk_constraint,
+    using_index BOOLEAN
+);
+DROP FUNCTION IF EXISTS _timescaledb_internal.chunk_constraint_add_table_constraint(
+    chunk_constraint_row  _timescaledb_catalog.chunk_constraint,
+    using_index BOOLEAN
+);
+
+CREATE FUNCTION _timescaledb_functions.chunk_constraint_add_table_constraint(
+    chunk_constraint_row  _timescaledb_catalog.chunk_constraint
+)
+    RETURNS VOID LANGUAGE PLPGSQL AS
+$BODY$
+DECLARE
+    chunk_row _timescaledb_catalog.chunk;
+    hypertable_row _timescaledb_catalog.hypertable;
+    constraint_oid OID;
+    constraint_type CHAR;
+    check_sql TEXT;
+    def TEXT;
+    indx_tablespace NAME;
+    tablespace_def TEXT;
+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;
+
+    IF chunk_constraint_row.dimension_slice_id IS NOT NULL THEN
+	    RAISE 'cannot create dimension constraint %', chunk_constraint_row;
+    ELSIF chunk_constraint_row.hypertable_constraint_name IS NOT NULL THEN
+
+        SELECT oid, contype INTO STRICT constraint_oid, constraint_type FROM pg_constraint
+        WHERE conname=chunk_constraint_row.hypertable_constraint_name AND
+              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;
+
+          def := pg_get_constraintdef(constraint_oid);
+
+        ELSIF constraint_type = 't' THEN
+          -- constraint triggers are copied separately with normal triggers
+          def := NULL;
+        ELSE
+          def := pg_get_constraintdef(constraint_oid);
+        END IF;
+
+    ELSE
+        RAISE 'unknown constraint type';
+    END IF;
+
+    IF def IS NOT NULL THEN
+        -- to allow for custom types with operators outside of pg_catalog
+        -- we set search_path to @extschema@
+        SET LOCAL search_path TO @extschema@, pg_temp;
+        EXECUTE pg_catalog.format(
+            $$ ALTER TABLE %I.%I ADD CONSTRAINT %I %s $$,
+            chunk_row.schema_name, chunk_row.table_name, chunk_constraint_row.constraint_name, def
+        );
+
+        -- if constraint (primary or unique) needs a tablespace then add it
+        -- via a separate ALTER INDEX SET TABLESPACE command. We cannot append it
+        -- to the "def" string above since it leads to a SYNTAX error when
+        -- "DEFERRABLE" or "INITIALLY DEFERRED" are used in the constraint
+        IF indx_tablespace IS NOT NULL THEN
+            EXECUTE pg_catalog.format(
+                $$ ALTER INDEX %I.%I SET TABLESPACE %I $$,
+                chunk_row.schema_name, chunk_constraint_row.constraint_name, indx_tablespace
+            );
+        END IF;
+
+    END IF;
+END
+$BODY$ SET search_path TO pg_catalog, pg_temp;

src/chunk_constraint.c

@@ -374,7 +374,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];
@@ -389,7 +389,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);
 
@@ -402,14 +404,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);
 
 	/*
@@ -431,7 +433,7 @@ create_non_dimensional_constraint(const ChunkConstraint *cc, Oid chunk_oid, int3
 	{
 		FormData_pg_constraint *constr = (FormData_pg_constraint *) GETSTRUCT(tuple);
 
-		if (OidIsValid(constr->conindid) && constr->contype != CONSTRAINT_FOREIGN)
+		if (OidIsValid(constr->conindid) && constr->contype != CONSTRAINT_FOREIGN && !using_index)
 			ts_chunk_index_create_from_constraint(hypertable_id,
 												  hypertable_constraint_oid,
 												  chunk_id,
@@ -494,7 +496,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);
 		}
 	}
 
@@ -838,7 +841,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;
@@ -863,7 +867,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

@@ -989,12 +989,11 @@ ts_chunk_index_adjust_meta(int32 chunk_id, const char *ht_index_name, const char
 }
 
 int
-ts_chunk_index_rename(Chunk *chunk, Oid chunk_indexrelid, const char *new_name)
+ts_chunk_index_rename(Chunk *chunk, const char *old_name, const char *new_name)
 {
 	ScanKeyData scankey[2];
-	const char *indexname = get_rel_name(chunk_indexrelid);
 	ChunkIndexRenameInfo renameinfo = {
-		.oldname = indexname,
+		.oldname = old_name,
 		.newname = new_name,
 	};
 
@@ -1007,7 +1006,7 @@ ts_chunk_index_rename(Chunk *chunk, Oid chunk_indexrelid, const char *new_name)
 				Anum_chunk_index_chunk_id_index_name_idx_index_name,
 				BTEqualStrategyNumber,
 				F_NAMEEQ,
-				CStringGetDatum(indexname));
+				CStringGetDatum(old_name));
 
 	return chunk_index_scan_update(CHUNK_INDEX_CHUNK_ID_INDEX_NAME_IDX,
 								   scankey,
@@ -1018,12 +1017,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,
 	};
@@ -1037,7 +1035,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

@@ -40,9 +40,8 @@ extern int ts_chunk_index_delete(int32 chunk_id, const char *indexname, bool dro
 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(Chunk *chunk, const char *old_name, const char *new_name);
+extern int ts_chunk_index_rename_parent(Hypertable *ht, const char *old_name, 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,