Create a catalog entry for constraint-backed indexes

This change makes our catalog more in-line with the pg catalog.
It also will simplify a lot of other code.

Author: cevian

sql/chunk.sql

@@ -72,12 +72,5 @@ BEGIN
 
     SELECT * INTO STRICT hypertable_row FROM _timescaledb_catalog.hypertable WHERE id = chunk_row.hypertable_id;
     main_table_oid := format('%I.%I', hypertable_row.schema_name, hypertable_row.table_name)::regclass;
-
-    --create the hypertable-constraints copy
-    PERFORM _timescaledb_internal.create_chunk_constraint(chunk_row.id, oid)
-    FROM pg_constraint
-    WHERE conrelid = main_table_oid
-    AND _timescaledb_internal.need_chunk_constraint(oid);
-
 END
 $BODY$;

sql/chunk_constraint.sql

@@ -57,13 +57,17 @@ CREATE OR REPLACE FUNCTION _timescaledb_internal.create_chunk_constraint(
     chunk_id INTEGER,
     constraint_oid OID
 )
-    RETURNS VOID LANGUAGE PLPGSQL AS
+    RETURNS OID LANGUAGE PLPGSQL AS
 $BODY$
 DECLARE
     chunk_constraint_row _timescaledb_catalog.chunk_constraint;
+    chunk_row _timescaledb_catalog.chunk;
     constraint_row pg_constraint;
+    hypertable_index_class_row pg_class;
+    chunk_index_class_row pg_class;
     constraint_name TEXT;
     hypertable_constraint_name TEXT = NULL;
+    chunk_constraint_oid OID;
 BEGIN
     SELECT * INTO STRICT constraint_row FROM pg_constraint WHERE OID = constraint_oid;
     hypertable_constraint_name := constraint_row.conname;
@@ -72,7 +76,17 @@ BEGIN
     INSERT INTO _timescaledb_catalog.chunk_constraint (chunk_id, constraint_name, dimension_slice_id, hypertable_constraint_name)
     VALUES (chunk_id, constraint_name, NULL, hypertable_constraint_name) RETURNING * INTO STRICT chunk_constraint_row;
 
+    --create actual constraint
     PERFORM _timescaledb_internal.chunk_constraint_add_table_constraint(chunk_constraint_row);
+
+    SELECT * INTO STRICT chunk_row FROM _timescaledb_catalog.chunk chunk WHERE chunk.id = chunk_id;
+
+    SELECT oid INTO STRICT chunk_constraint_oid
+    FROM pg_constraint con
+    WHERE con.conrelid = format('%I.%I', chunk_row.schema_name, chunk_row.table_name)::regclass
+    AND con.conname = constraint_name;
+
+    RETURN chunk_constraint_oid;
 END
 $BODY$;
 
@@ -88,6 +102,7 @@ $BODY$
 DECLARE
     chunk_row _timescaledb_catalog.chunk;
     chunk_constraint_row _timescaledb_catalog.chunk_constraint;
+    constraint_row pg_constraint;
 BEGIN
     SELECT * INTO STRICT chunk_row FROM _timescaledb_catalog.chunk c WHERE c.id = chunk_id;
 
@@ -96,6 +111,11 @@ BEGIN
     AND cc.chunk_id = drop_chunk_constraint.chunk_id
     RETURNING * INTO STRICT chunk_constraint_row;
 
+    SELECT * INTO STRICT constraint_row
+    FROM pg_constraint con
+    WHERE con.conrelid = format('%I.%I', chunk_row.schema_name, chunk_row.table_name)::regclass
+    AND con.conname = constraint_name;
+
     IF alter_table THEN
         EXECUTE format(
             $$  ALTER TABLE %I.%I DROP CONSTRAINT %I $$,
@@ -106,71 +126,6 @@ BEGIN
 END
 $BODY$;
 
--- do I need to add a hypertable constraint to the chunks?;
-CREATE OR REPLACE FUNCTION _timescaledb_internal.need_chunk_constraint(
-    constraint_oid OID
-)
-    RETURNS BOOLEAN LANGUAGE PLPGSQL VOLATILE AS
-$BODY$
-DECLARE
-    constraint_row record;
-BEGIN
-    SELECT * INTO STRICT constraint_row FROM pg_constraint WHERE OID = constraint_oid;
-
-    IF constraint_row.contype IN ('c') THEN
-        -- check and not null constraints handled by regular inheritance (from docs):
-        --    All check constraints and not-null constraints on a parent table are automatically inherited by its children,
-        --    unless explicitly specified otherwise with NO INHERIT clauses. Other types of constraints
-        --    (unique, primary key, and foreign key constraints) are not inherited."
-
-        IF constraint_row.connoinherit THEN
-            RAISE 'NO INHERIT option not supported on hypertables: %', constraint_row.conname
-            USING ERRCODE = 'IO101';
-        END IF;
-
-        RETURN FALSE;
-    END IF;
-    RETURN TRUE;
-END
-$BODY$;
-
--- Creates a constraint on all chunks for a hypertable.
-CREATE OR REPLACE FUNCTION _timescaledb_internal.add_constraint(
-    hypertable_id INTEGER,
-    constraint_oid OID
-)
-    RETURNS VOID LANGUAGE PLPGSQL VOLATILE AS
-$BODY$
-DECLARE
-    constraint_row pg_constraint;
-    hypertable_row _timescaledb_catalog.hypertable;
-BEGIN
-    IF _timescaledb_internal.need_chunk_constraint(constraint_oid) THEN
-        SELECT * INTO STRICT constraint_row FROM pg_constraint WHERE OID = constraint_oid;
-
-        PERFORM _timescaledb_internal.create_chunk_constraint(c.id, constraint_oid)
-        FROM _timescaledb_catalog.chunk c
-        WHERE c.hypertable_id = add_constraint.hypertable_id;
-    END IF;
-END
-$BODY$;
-
-CREATE OR REPLACE FUNCTION _timescaledb_internal.add_constraint_by_name(
-    hypertable_id INTEGER,
-    constraint_name name
-)
-    RETURNS VOID LANGUAGE PLPGSQL VOLATILE AS
-$BODY$
-DECLARE
-    constraint_oid OID;
-BEGIN
-    SELECT oid INTO STRICT constraint_oid FROM pg_constraint WHERE conname = constraint_name
-    AND conrelid = _timescaledb_internal.main_table_from_hypertable(hypertable_id);
-
-    PERFORM _timescaledb_internal.add_constraint(hypertable_id, constraint_oid);
-END
-$BODY$;
-
 -- Drops constraint on all chunks for a hypertable.
 CREATE OR REPLACE FUNCTION _timescaledb_internal.drop_constraint(
     hypertable_id INTEGER,

sql/ddl_api.sql

@@ -118,10 +118,6 @@ BEGIN
             USING ERRCODE = 'IO101';
     END;
 
-    PERFORM _timescaledb_internal.add_constraint(hypertable_row.id, oid)
-    FROM pg_constraint
-    WHERE conrelid = main_table;
-
    IF create_default_indexes THEN
         PERFORM _timescaledb_internal.create_default_indexes(hypertable_row, main_table, partitioning_column);
     END IF;

sql/updates/pre-0.6.0--0.7.0-dev.sql

@@ -1 +1,26 @@
 DROP FUNCTION IF EXISTS drop_chunks(INTEGER, NAME, NAME, BOOLEAN);
+
+DROP FUNCTION _timescaledb_internal.create_chunk_constraint(integer,oid);
+DROP FUNCTION _timescaledb_internal.add_constraint(integer,oid);
+DROP FUNCTION _timescaledb_internal.add_constraint_by_name(integer,name);
+DROP FUNCTION _timescaledb_internal.need_chunk_constraint(oid);
+
+INSERT INTO _timescaledb_catalog.chunk_index (chunk_id, index_name, hypertable_id, hypertable_index_name)
+SELECT chunk_con.chunk_id, pg_chunk_index_class.relname, chunk.hypertable_id, pg_hypertable_index_class.relname
+FROM _timescaledb_catalog.chunk_constraint chunk_con
+INNER JOIN _timescaledb_catalog.chunk chunk ON (chunk_con.chunk_id = chunk.id)
+INNER JOIN _timescaledb_catalog.hypertable hypertable ON (chunk.hypertable_id = hypertable.id)
+INNER JOIN pg_constraint pg_chunk_con ON (
+        pg_chunk_con.conrelid = format('%I.%I', chunk.schema_name, chunk.table_name)::regclass
+        AND pg_chunk_con.conname = chunk_con.constraint_name
+)
+INNER JOIN pg_class pg_chunk_index_class ON (
+    pg_chunk_con.conindid = pg_chunk_index_class.oid
+)
+INNER JOIN pg_constraint pg_hypertable_con ON (
+        pg_hypertable_con.conrelid = format('%I.%I', hypertable.schema_name, hypertable.table_name)::regclass
+        AND pg_hypertable_con.conname = chunk_con.hypertable_constraint_name
+)
+INNER JOIN pg_class pg_hypertable_index_class ON (
+    pg_hypertable_con.conindid = pg_hypertable_index_class.oid
+);

src/catalog.c

@@ -94,8 +94,8 @@ const static InternalFunctionDef internal_function_definitions[_MAX_INTERNAL_FUN
 		.name = "ddl_change_owner",
 		.args = 2,
 	},
-	[DDL_ADD_CONSTRAINT] = {
-		.name = "add_constraint_by_name",
+	[DDL_CREATE_CHUNK_CONSTRAINT] = {
+		.name = "create_chunk_constraint",
 		.args = 2,
 	},
 	[DDL_DROP_CONSTRAINT] = {

src/catalog.h

@@ -32,18 +32,18 @@ enum CatalogTable
 };
 
 #define CatalogInternalCall1(func, datum1) \
-	OidFunctionCall1(catalog_get_internal_function_id(catalog_get(), func), datum1);
+	OidFunctionCall1(catalog_get_internal_function_id(catalog_get(), func), datum1)
 #define CatalogInternalCall2(func, datum1, datum2) \
-	OidFunctionCall2(catalog_get_internal_function_id(catalog_get(), func), datum1, datum2);
+	OidFunctionCall2(catalog_get_internal_function_id(catalog_get(), func), datum1, datum2)
 #define CatalogInternalCall3(func, datum1, datum2, datum3) \
-	OidFunctionCall3(catalog_get_internal_function_id(catalog_get(), func), datum1, datum2, datum3);
+	OidFunctionCall3(catalog_get_internal_function_id(catalog_get(), func), datum1, datum2, datum3)
 #define CatalogInternalCall4(func, datum1, datum2, datum3, datum4) \
-	OidFunctionCall4(catalog_get_internal_function_id(catalog_get(), func), datum1, datum2, datum3, datum4);
+	OidFunctionCall4(catalog_get_internal_function_id(catalog_get(), func), datum1, datum2, datum3, datum4)
 
 typedef enum InternalFunction
 {
 	DDL_CHANGE_OWNER = 0,
-	DDL_ADD_CONSTRAINT,
+	DDL_CREATE_CHUNK_CONSTRAINT,
 	DDL_DROP_CONSTRAINT,
 	DDL_DROP_HYPERTABLE,
 	DDL_RENAME_HYPERTABLE,

src/chunk.c

@@ -320,6 +320,8 @@ chunk_create_after_lock(Hypertable *ht, Point *p, const char *schema, const char
 	/* Create all indexes on the chunk */
 	chunk_index_create_all(ht->fd.id, ht->main_table_relid, chunk->fd.id, chunk->table_id);
 
+	chunk_constraint_create_all_on_chunk(ht, chunk);
+
 	catalog_restore_user(&sec_ctx);
 
 	return chunk;

src/chunk_constraint.c

@@ -4,12 +4,23 @@
 #include <access/heapam.h>
 #include <access/xact.h>
 #include <catalog/indexing.h>
+#include <catalog/pg_constraint.h>
+#include <catalog/objectaddress.h>
 
 #include "scanner.h"
 #include "chunk_constraint.h"
+#include "chunk_index.h"
 #include "dimension_slice.h"
 #include "hypercube.h"
 #include "chunk.h"
+#include "hypertable.h"
+#include "errors.h"
+#include "process_utility.h"
+
+#define create_chunk_constraint(chunk, constraint_oid)		\
+	DatumGetObjectId(CatalogInternalCall2(DDL_CREATE_CHUNK_CONSTRAINT, \
+						 Int32GetDatum(chunk->fd.id),	\
+						 ObjectIdGetDatum(constraint_oid)))
 
 static inline ChunkConstraint *
 chunk_constraint_fill(ChunkConstraint *cc, HeapTuple tuple)
@@ -190,3 +201,90 @@ chunk_constraint_insert_multi(ChunkConstraint *constraints, Size num_constraints
 
 	heap_close(rel, RowExclusiveLock);
 }
+
+static bool
+chunk_constraint_need_on_chunk(Form_pg_constraint conform)
+{
+	if (conform->contype == CONSTRAINT_CHECK)
+	{
+		/*
+		 * check and not null constraints handled by regular inheritance (from
+		 * docs): All check constraints and not-null constraints on a parent
+		 * table are automatically inherited by its children, unless
+		 * explicitly specified otherwise with NO INHERIT clauses. Other types
+		 * of constraints (unique, primary key, and foreign key constraints)
+		 * are not inherited."
+		 */
+		if (conform->connoinherit)
+		{
+			ereport(ERROR,
+					(errcode(ERRCODE_IO_OPERATION_NOT_SUPPORTED),
+					 errmsg("NO INHERIT option not supported on hypertables: %s", conform->conname.data)
+					 ));
+		}
+		return false;
+	}
+	return true;
+}
+
+static void
+chunk_constraint_create_on_chunk_impl(Hypertable *ht, Chunk *chunk, HeapTuple constraint_htup)
+{
+	Form_pg_constraint pg_constraint = (Form_pg_constraint) GETSTRUCT(constraint_htup);
+
+	if (chunk_constraint_need_on_chunk(pg_constraint))
+	{
+		Oid			hypertable_constraint_oid = HeapTupleGetOid(constraint_htup);
+		Oid			chunk_constraint_oid;
+		CatalogSecurityContext sec_ctx;
+
+		catalog_become_owner(catalog_get(), &sec_ctx);
+		process_utility_set_expect_chunk_modification(true);
+		chunk_constraint_oid = create_chunk_constraint(chunk, hypertable_constraint_oid);
+		process_utility_set_expect_chunk_modification(false);
+		catalog_restore_user(&sec_ctx);
+
+		Assert(OidIsValid(chunk_constraint_oid));
+
+		if (OidIsValid(pg_constraint->conindid) && pg_constraint->contype != CONSTRAINT_FOREIGN)
+		{
+			chunk_index_create_from_constraint(ht->fd.id, hypertable_constraint_oid, chunk->fd.id, chunk_constraint_oid);
+		}
+	}
+}
+
+void
+chunk_constraint_create_all_on_chunk(Hypertable *ht, Chunk *chunk)
+{
+	ScanKeyData skey;
+	Relation	rel;
+	SysScanDesc scan;
+	HeapTuple	htup;
+
+	ScanKeyInit(&skey,
+				Anum_pg_constraint_conrelid,
+				BTEqualStrategyNumber, F_OIDEQ, ht->main_table_relid);
+
+	rel = heap_open(ConstraintRelationId, AccessShareLock);
+	scan = systable_beginscan(rel, ConstraintRelidIndexId, true,
+							  NULL, 1, &skey);
+
+	while (HeapTupleIsValid(htup = systable_getnext(scan)))
+	{
+		chunk_constraint_create_on_chunk_impl(ht, chunk, htup);
+	}
+	systable_endscan(scan);
+	heap_close(rel, AccessShareLock);
+}
+
+void
+chunk_constraint_create_on_chunk(Hypertable *ht, Chunk *chunk, Oid constraintOid)
+{
+	Relation	rel;
+	HeapTuple	htup;
+
+	rel = heap_open(ConstraintRelationId, AccessShareLock);
+	htup = get_catalog_object_by_oid(rel, constraintOid);
+	chunk_constraint_create_on_chunk_impl(ht, chunk, htup);
+	heap_close(rel, AccessShareLock);
+}

src/chunk_constraint.h

@@ -6,6 +6,7 @@
 #include <nodes/pg_list.h>
 
 #include "catalog.h"
+#include "hypertable.h"
 
 typedef struct ChunkConstraint
 {
@@ -26,5 +27,6 @@ typedef struct ChunkScanCtx ChunkScanCtx;
 extern Chunk *chunk_constraint_scan_by_chunk_id(Chunk *chunk);
 extern int	chunk_constraint_scan_by_dimension_slice_id(DimensionSlice *slice, ChunkScanCtx *ctx);
 extern void chunk_constraint_insert_multi(ChunkConstraint *constraints, Size num_constraints);
-
+extern void chunk_constraint_create_all_on_chunk(Hypertable *ht, Chunk *chunk);
+extern void chunk_constraint_create_on_chunk(Hypertable *ht, Chunk *chunk, Oid constraintOid);
 #endif   /* TIMESCALEDB_CHUNK_CONSTRAINT_H */