Move index and constraints drop handling to event trigger

This Fixes at least two bugs:

1) A drop of a referenced table used to drop the associated
FK constraint but not the metadata associated with the constraint.
Fixes #43.

2) A drop of a column removed any indexes associated with the column
but not the metadata associated with the index.

Author: cevian

sql/ddl_triggers.sql

@@ -1,8 +1,12 @@
-CREATE OR REPLACE FUNCTION _timescaledb_internal.ddl_command_end() RETURNS event_trigger
-AS '@MODULE_PATHNAME@', 'timescaledb_ddl_command_end' LANGUAGE C;
+CREATE OR REPLACE FUNCTION _timescaledb_internal.process_ddl_event() RETURNS event_trigger
+AS '@MODULE_PATHNAME@', 'timescaledb_process_ddl_event' LANGUAGE C;
 
 DROP EVENT TRIGGER IF EXISTS timescaledb_ddl_command_end;
 --EVENT TRIGGER MUST exclude the ALTER EXTENSION tag.
 CREATE EVENT TRIGGER timescaledb_ddl_command_end ON ddl_command_end
 WHEN TAG IN ('ALTER TABLE','CREATE TRIGGER','CREATE TABLE','CREATE INDEX','ALTER INDEX')
-EXECUTE PROCEDURE _timescaledb_internal.ddl_command_end();
+EXECUTE PROCEDURE _timescaledb_internal.process_ddl_event();
+
+DROP EVENT TRIGGER IF EXISTS timescaledb_ddl_sql_drop;
+CREATE EVENT TRIGGER timescaledb_ddl_sql_drop ON sql_drop
+EXECUTE PROCEDURE _timescaledb_internal.process_ddl_event();

sql/updates/post-0.8.0--0.9.0-dev.sql

@@ -0,0 +1 @@
+DROP FUNCTION _timescaledb_internal.ddl_command_end();

src/catalog.c

@@ -71,6 +71,7 @@ static const TableIndexDef catalog_table_index_definitions[_MAX_CATALOG_TABLES]
 	[CHUNK_CONSTRAINT] = {
 		.length = _MAX_CHUNK_CONSTRAINT_INDEX,
 		.names = (char *[]) {
+			[CHUNK_CONSTRAINT_CHUNK_ID_CONSTRAINT_NAME_IDX] = "chunk_constraint_chunk_id_constraint_name_key",
 			[CHUNK_CONSTRAINT_CHUNK_ID_DIMENSION_SLICE_ID_IDX] = "chunk_constraint_chunk_id_dimension_slice_id_idx",
 		}
 	},

src/catalog.h

@@ -34,6 +34,10 @@ typedef enum CatalogTable
 } CatalogTable;
 
 #define INVALID_CATALOG_TABLE _MAX_CATALOG_TABLES
+#define INVALID_INDEXID -1
+
+#define CATALOG_INDEX(catalog, tableid, indexid) \
+    (indexid == INVALID_INDEXID ? InvalidOid : (catalog)->tables[tableid].index_ids[indexid])
 
 #define CatalogInternalCall1(func, datum1) \
 	OidFunctionCall1(catalog_get_internal_function_id(catalog_get(), func), datum1)
@@ -329,7 +333,8 @@ typedef FormData_chunk_constraint *Form_chunk_constraint;
 
 enum
 {
-	CHUNK_CONSTRAINT_CHUNK_ID_DIMENSION_SLICE_ID_IDX = 0,
+	CHUNK_CONSTRAINT_CHUNK_ID_CONSTRAINT_NAME_IDX = 0,
+	CHUNK_CONSTRAINT_CHUNK_ID_DIMENSION_SLICE_ID_IDX,
 	_MAX_CHUNK_CONSTRAINT_INDEX,
 };
 
@@ -340,6 +345,13 @@ enum Anum_chunk_constraint_chunk_id_dimension_slice_id_idx
 	_Anum_chunk_constraint_chunk_id_dimension_slice_id_idx_max,
 };
 
+enum Anum_chunk_constraint_chunk_id_constraint_name_idx
+{
+	Anum_chunk_constraint_chunk_id_constraint_name_idx_chunk_id = 1,
+	Anum_chunk_constraint_chunk_id_constraint_name_idx_constraint_name,
+	_Anum_chunk_constraint_chunk_id_constraint_name_idx_max,
+};
+
 /************************************
  *
  * Chunk index table definitions

src/chunk.c

@@ -566,7 +566,6 @@ chunk_fill_stub(Chunk *chunk_stub, bool tuplock)
 	ScannerCtx	ctx = {
 		.table = catalog->tables[CHUNK].id,
 		.index = catalog->tables[CHUNK].index_ids[CHUNK_ID_INDEX],
-		.scantype = ScannerTypeIndex,
 		.nkeys = 1,
 		.scankey = scankey,
 		.data = chunk_stub,
@@ -868,7 +867,6 @@ chunk_scan_internal(int indexid,
 	ScannerCtx	ctx = {
 		.table = catalog->tables[CHUNK].id,
 		.index = catalog->tables[CHUNK].index_ids[indexid],
-		.scantype = ScannerTypeIndex,
 		.nkeys = nkeys,
 		.data = data,
 		.scankey = scankey,

src/chunk_constraint.c

@@ -356,8 +356,6 @@ chunk_constraint_tuple_found(TupleInfo *ti, void *data)
 	return true;
 }
 
-#define NO_INDEXSCAN -1
-
 static int
 chunk_constraint_scan_internal(int indexid,
 							   ScanKeyData *scankey,
@@ -370,7 +368,7 @@ chunk_constraint_scan_internal(int indexid,
 	Catalog    *catalog = catalog_get();
 	ScannerCtx	scanctx = {
 		.table = catalog->tables[CHUNK_CONSTRAINT].id,
-		.scantype = (indexid == NO_INDEXSCAN) ? ScannerTypeHeap : ScannerTypeIndex,
+		.index = CATALOG_INDEX(catalog, CHUNK_CONSTRAINT, indexid),
 		.nkeys = nkeys,
 		.scankey = scankey,
 		.data = data,
@@ -380,9 +378,6 @@ chunk_constraint_scan_internal(int indexid,
 		.scandirection = ForwardScanDirection,
 	};
 
-	if (indexid != NO_INDEXSCAN)
-		scanctx.index = catalog->tables[CHUNK_CONSTRAINT].index_ids[indexid];
-
 	return scanner_scan(&scanctx);
 }
 
@@ -413,6 +408,38 @@ chunk_constraint_scan_by_chunk_id_internal(int32 chunk_id,
 										  lockmode);
 }
 
+static int
+chunk_constraint_scan_by_chunk_id_constraint_name_internal(int32 chunk_id,
+														   const char *constraint_name,
+														   tuple_found_func tuple_found,
+														   tuple_found_func tuple_filter,
+														   void *data,
+														   LOCKMODE lockmode)
+{
+	ScanKeyData scankey[2];
+
+
+	ScanKeyInit(&scankey[0],
+				Anum_chunk_constraint_chunk_id_constraint_name_idx_chunk_id,
+				BTEqualStrategyNumber,
+				F_INT4EQ,
+				Int32GetDatum(chunk_id));
+
+	ScanKeyInit(&scankey[1],
+				Anum_chunk_constraint_chunk_id_constraint_name_idx_constraint_name,
+				BTEqualStrategyNumber,
+				F_NAMEEQ,
+				DirectFunctionCall1(namein, CStringGetDatum(constraint_name)));
+
+	return chunk_constraint_scan_internal(CHUNK_CONSTRAINT_CHUNK_ID_CONSTRAINT_NAME_IDX,
+										  scankey,
+										  2,
+										  tuple_found,
+										  tuple_filter,
+										  data,
+										  lockmode);
+}
+
 /*
  * Scan all the chunk's constraints given its chunk ID.
  *
@@ -625,6 +652,8 @@ typedef struct ConstraintInfo
 {
 	const char *hypertable_constraint_name;
 	ChunkConstraints *ccs;
+	bool		delete_metadata;
+	bool		drop_constraint;
 } ConstraintInfo;
 
 typedef struct RenameHypertableConstraintInfo
@@ -652,24 +681,29 @@ chunk_constraint_delete_tuple(TupleInfo *ti, void *data)
 	ObjectAddress constrobj = {
 		.classId = ConstraintRelationId,
 		.objectId = get_relation_constraint_oid(chunk->table_id,
-												NameStr(*DatumGetName(constrname)), false),
+												NameStr(*DatumGetName(constrname)), true),
 	};
 	Oid			index_relid = get_constraint_index(constrobj.objectId);
 
 	/* Collect the deleted constraints */
-	if (NULL != info && NULL != info->ccs)
+	if (NULL != info->ccs)
 		chunk_constraint_tuple_found(ti, info->ccs);
 
-	/*
-	 * If this is an index constraint, we need to cleanup the index metadata.
-	 * Don't drop the index though, since that will happend when the
-	 * constraint is dropped.
-	 */
-	if (OidIsValid(index_relid))
-		chunk_index_delete(chunk, index_relid, false);
+	if (info->delete_metadata)
+	{
+		/*
+		 * If this is an index constraint, we need to cleanup the index
+		 * metadata. Don't drop the index though, since that will happend when
+		 * the constraint is dropped.
+		 */
+		if (OidIsValid(index_relid))
+			chunk_index_delete(chunk, index_relid, false);
 
-	catalog_delete(ti->scanrel, ti->tuple);
-	performDeletion(&constrobj, DROP_RESTRICT, 0);
+		catalog_delete(ti->scanrel, ti->tuple);
+	}
+
+	if (info->drop_constraint && OidIsValid(constrobj.objectId))
+		performDeletion(&constrobj, DROP_RESTRICT, 0);
 
 	return true;
 }
@@ -695,10 +729,13 @@ hypertable_constraint_tuple_filter(TupleInfo *ti, void *data)
 
 int
 chunk_constraint_delete_by_hypertable_constraint_name(int32 chunk_id,
-													  char *hypertable_constraint_name)
+													  char *hypertable_constraint_name,
+													  bool delete_metadata, bool drop_constraint)
 {
 	ConstraintInfo info = {
 		.hypertable_constraint_name = hypertable_constraint_name,
+		.delete_metadata = delete_metadata,
+		.drop_constraint = drop_constraint
 	};
 
 	return chunk_constraint_scan_by_chunk_id_internal(chunk_id,
@@ -708,6 +745,23 @@ chunk_constraint_delete_by_hypertable_constraint_name(int32 chunk_id,
 													  RowExclusiveLock);
 }
 
+int
+chunk_constraint_delete_by_constraint_name(int32 chunk_id, const char *constraint_name,
+										   bool delete_metadata, bool drop_constraint)
+{
+	ConstraintInfo info = {
+		.delete_metadata = delete_metadata,
+		.drop_constraint = drop_constraint
+	};
+
+	return chunk_constraint_scan_by_chunk_id_constraint_name_internal(chunk_id,
+																	  constraint_name,
+																	  chunk_constraint_delete_tuple,
+																	  NULL,
+																	  &info,
+																	  RowExclusiveLock);
+}
+
 /*
  * Delete all constraints for a chunk. Optionally, collect the deleted constraints.
  */
@@ -716,6 +770,8 @@ chunk_constraint_delete_by_chunk_id(int32 chunk_id, ChunkConstraints *ccs)
 {
 	ConstraintInfo info = {
 		.ccs = ccs,
+		.delete_metadata = true,
+		.drop_constraint = true,
 	};
 
 	return chunk_constraint_scan_by_chunk_id_internal(chunk_id,
@@ -728,6 +784,11 @@ chunk_constraint_delete_by_chunk_id(int32 chunk_id, ChunkConstraints *ccs)
 int
 chunk_constraint_delete_by_dimension_slice_id(int32 dimension_slice_id)
 {
+	ConstraintInfo info = {
+		.delete_metadata = true,
+		.drop_constraint = true,
+	};
+
 	ScanKeyData scankey[1];
 
 	ScanKeyInit(&scankey[0],
@@ -736,12 +797,12 @@ chunk_constraint_delete_by_dimension_slice_id(int32 dimension_slice_id)
 				F_INT4EQ,
 				Int32GetDatum(dimension_slice_id));
 
-	return chunk_constraint_scan_internal(NO_INDEXSCAN,
+	return chunk_constraint_scan_internal(INVALID_INDEXID,
 										  scankey,
 										  1,
 										  chunk_constraint_delete_tuple,
 										  NULL,
-										  NULL,
+										  &info,
 										  RowExclusiveLock);
 }
 

src/chunk_constraint.h

@@ -42,9 +42,10 @@ extern int	chunk_constraints_add_dimension_constraints(ChunkConstraints *ccs, in
 extern int	chunk_constraints_add_inheritable_constraints(ChunkConstraints *ccs, int32 chunk_id, Oid hypertable_oid);
 extern void chunk_constraints_create(ChunkConstraints *ccs, Oid chunk_oid, int32 chunk_id, Oid hypertable_oid, int32 hypertable_id);
 extern void chunk_constraint_create_on_chunk(Chunk *chunk, Oid constraint_oid);
-extern int	chunk_constraint_delete_by_hypertable_constraint_name(int32 chunk_id, char *hypertable_constraint_name);
+extern int	chunk_constraint_delete_by_hypertable_constraint_name(int32 chunk_id, char *hypertable_constraint_name, bool delete_metadata, bool drop_constraint);
 extern int	chunk_constraint_delete_by_chunk_id(int32 chunk_id, ChunkConstraints *ccs);
 extern int	chunk_constraint_delete_by_dimension_slice_id(int32 dimension_slice_id);
+extern int	chunk_constraint_delete_by_constraint_name(int32 chunk_id, const char *constraint_name, bool delete_metadata, bool drop_constraint);
 extern void chunk_constraint_recreate(ChunkConstraint *cc, Oid chunk_oid);
 extern int	chunk_constraint_rename_hypertable_constraint(int32 chunk_id, const char *oldname, const char *newname);