Delete orphaned dimension slices

When chunks are deleted, dimension slices can be orphaned, i.e., there
are no chunks or chunk constraints that reference such slices. This
change ensures that, when chunks are deleted, orphaned slices are also
deleted.

Author: erimatnor

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

@@ -43,3 +43,10 @@ DROP FUNCTION _timescaledb_internal.chunk_create_table(int, name);
 DROP INDEX _timescaledb_catalog.dimension_slice_dimension_id_range_start_range_end_idx;
 
 DROP FUNCTION _timescaledb_internal.drop_hypertable(int,boolean);
+
+-- Delete orphaned slices
+DELETE FROM _timescaledb_catalog.dimension_slice WHERE id IN
+(SELECT dimension_id FROM _timescaledb_catalog.chunk_constraint cc
+ FULL OUTER JOIN _timescaledb_catalog.dimension_slice ds
+ ON (ds.id = cc.dimension_slice_id)
+ WHERE dimension_slice_id IS NULL);

src/chunk.c

@@ -647,7 +647,7 @@ dimension_slice_and_chunk_constraint_join(ChunkScanCtx *scanctx, DimensionVec *v
 		 * For each dimension slice, find matching constraints. These will be
 		 * saved in the scan context
 		 */
-		chunk_constraint_scan_by_dimension_slice_id(vec->slices[i], scanctx);
+		chunk_constraint_scan_by_dimension_slice(vec->slices[i], scanctx);
 	}
 }
 
@@ -990,10 +990,26 @@ chunk_tuple_delete(TupleInfo *ti, void *data)
 {
 	FormData_chunk *form = (FormData_chunk *) GETSTRUCT(ti->tuple);
 	CatalogSecurityContext sec_ctx;
+	ChunkConstraints *ccs = chunk_constraints_alloc(2);
+	int			i;
 
-	chunk_constraint_delete_by_chunk_id(form->id);
+	chunk_constraint_delete_by_chunk_id(form->id, ccs);
 	chunk_index_delete_by_chunk_id(form->id, true);
 
+	/* Check for dimension slices that are orphaned by the chunk deletion */
+	for (i = 0; i < ccs->num_constraints; i++)
+	{
+		ChunkConstraint *cc = &ccs->constraints[i];
+
+		/*
+		 * Delete the dimension slice if there are no remaining constraints
+		 * referencing it
+		 */
+		if (is_dimension_constraint(cc) &&
+			chunk_constraint_scan_by_dimension_slice_id(cc->fd.dimension_slice_id, NULL) == 0)
+			dimension_slice_delete_by_id(cc->fd.dimension_slice_id, false);
+	}
+
 	catalog_become_owner(catalog_get(), &sec_ctx);
 	catalog_delete(ti->scanrel, ti->tuple);
 	catalog_restore_user(&sec_ctx);
@@ -1066,7 +1082,7 @@ chunk_recreate_all_constraints_for_dimension(Hyperspace *hs, int32 dimension_id)
 	chunk_scan_ctx_init(&chunkctx, hs, NULL);
 
 	for (i = 0; i < slices->num_slices; i++)
-		chunk_constraint_scan_by_dimension_slice_id(slices->slices[i], &chunkctx);
+		chunk_constraint_scan_by_dimension_slice(slices->slices[i], &chunkctx);
 
 	chunk_scan_ctx_foreach_chunk(&chunkctx, chunk_recreate_constraint, 0);
 	chunk_scan_ctx_destroy(&chunkctx);

src/chunk_constraint.c

@@ -67,10 +67,13 @@ chunk_constraints_expand(ChunkConstraints *ccs, int16 new_capacity)
 	ccs->constraints = repalloc(ccs->constraints, CHUNK_CONSTRAINTS_SIZE(new_capacity));
 }
 
-
 static
 void
-chunk_constraint_choose_name(Name dst, bool is_dimension, int32 dimension_slice_id, const char *hypertable_constraint_name, int32 chunk_id)
+chunk_constraint_choose_name(Name dst,
+							 bool is_dimension,
+							 int32 dimension_slice_id,
+							 const char *hypertable_constraint_name,
+							 int32 chunk_id)
 {
 	if (is_dimension)
 	{
@@ -216,7 +219,8 @@ chunk_constraints_add_from_tuple(ChunkConstraints *ccs, TupleInfo *ti)
 	heap_deform_tuple(ti->tuple, ti->desc, values, nulls);
 
 	constraint_name = DatumGetName(values[Anum_chunk_constraint_constraint_name - 1]);
-	if (heap_attisnull(ti->tuple, Anum_chunk_constraint_dimension_slice_id))
+
+	if (nulls[Anum_chunk_constraint_dimension_slice_id - 1])
 	{
 		dimension_slice_id = 0;
 		hypertable_constraint_name = DatumGetName(values[Anum_chunk_constraint_hypertable_constraint_name - 1]);
@@ -332,6 +336,9 @@ chunk_constraints_create(ChunkConstraints *ccs,
 								hypertable_id);
 }
 
+/*
+ * Scan filter function for only getting dimension constraints.
+ */
 static bool
 chunk_constraint_for_dimension_slice(TupleInfo *ti, void *data)
 {
@@ -343,7 +350,8 @@ chunk_constraint_tuple_found(TupleInfo *ti, void *data)
 {
 	ChunkConstraints *ccs = data;
 
-	chunk_constraints_add_from_tuple(ccs, ti);
+	if (NULL != ccs)
+		chunk_constraints_add_from_tuple(ccs, ti);
 
 	return true;
 }
@@ -469,34 +477,58 @@ chunk_constraint_dimension_slice_id_tuple_found(TupleInfo *ti, void *data)
 	return true;
 }
 
-/*
- * Scan for all chunk constraints that match the given slice ID. The chunk
- * constraints are saved in the chunk scan context.
- */
-int
-chunk_constraint_scan_by_dimension_slice_id(DimensionSlice *slice, ChunkScanCtx *ctx)
+static int
+scan_by_dimension_slice_id(int32 dimension_slice_id,
+						   tuple_found_func tuple_found,
+						   void *data)
 {
 	ScanKeyData scankey[1];
-	ChunkConstraintScanData data = {
-		.scanctx = ctx,
-		.slice = slice,
-	};
 
 	ScanKeyInit(&scankey[0],
 				Anum_chunk_constraint_chunk_id_dimension_slice_id_idx_dimension_slice_id,
 				BTEqualStrategyNumber,
 				F_INT4EQ,
-				Int32GetDatum(slice->fd.id));
+				Int32GetDatum(dimension_slice_id));
 
 	return chunk_constraint_scan_internal(CHUNK_CONSTRAINT_CHUNK_ID_DIMENSION_SLICE_ID_IDX,
 										  scankey,
 										  1,
-										  chunk_constraint_dimension_slice_id_tuple_found,
+										  tuple_found,
 										  chunk_constraint_for_dimension_slice,
-										  &data,
+										  data,
 										  AccessShareLock);
 }
 
+/*
+ * Scan for all chunk constraints that match the given slice ID. The chunk
+ * constraints are saved in the chunk scan context.
+ */
+int
+chunk_constraint_scan_by_dimension_slice(DimensionSlice *slice, ChunkScanCtx *ctx)
+{
+	ChunkConstraintScanData data = {
+		.scanctx = ctx,
+		.slice = slice,
+	};
+
+	return scan_by_dimension_slice_id(slice->fd.id,
+									  chunk_constraint_dimension_slice_id_tuple_found,
+									  &data);
+}
+
+/*
+ * Scan for chunk constraints given a dimension slice ID.
+ *
+ * Optionally, collect all chunk constraints if ChunkConstraints is non-NULL.
+ */
+int
+chunk_constraint_scan_by_dimension_slice_id(int32 dimension_slice_id, ChunkConstraints *ccs)
+{
+	return scan_by_dimension_slice_id(dimension_slice_id,
+									  chunk_constraint_tuple_found,
+									  ccs);
+}
+
 static bool
 chunk_constraint_need_on_chunk(Form_pg_constraint conform)
 {
@@ -592,6 +624,7 @@ chunk_constraint_create_on_chunk(Chunk *chunk, Oid constraint_oid)
 typedef struct ConstraintInfo
 {
 	const char *hypertable_constraint_name;
+	ChunkConstraints *ccs;
 } ConstraintInfo;
 
 typedef struct RenameHypertableConstraintInfo
@@ -601,9 +634,15 @@ typedef struct RenameHypertableConstraintInfo
 } RenameHypertableConstraintInfo;
 
 
+/*
+ * Delete a chunk constraint tuple.
+ *
+ * Optionally, the data argument is a ConstraintInfo.
+ */
 static bool
 chunk_constraint_delete_tuple(TupleInfo *ti, void *data)
 {
+	ConstraintInfo *info = data;
 	bool		isnull;
 	Datum		constrname = heap_getattr(ti->tuple, Anum_chunk_constraint_constraint_name,
 										  ti->desc, &isnull);
@@ -617,6 +656,10 @@ chunk_constraint_delete_tuple(TupleInfo *ti, void *data)
 	};
 	Oid			index_relid = get_constraint_index(constrobj.objectId);
 
+	/* Collect the deleted constraints */
+	if (NULL != info && 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
@@ -665,13 +708,20 @@ chunk_constraint_delete_by_hypertable_constraint_name(int32 chunk_id,
 													  RowExclusiveLock);
 }
 
+/*
+ * Delete all constraints for a chunk. Optionally, collect the deleted constraints.
+ */
 int
-chunk_constraint_delete_by_chunk_id(int32 chunk_id)
+chunk_constraint_delete_by_chunk_id(int32 chunk_id, ChunkConstraints *ccs)
 {
+	ConstraintInfo info = {
+		.ccs = ccs,
+	};
+
 	return chunk_constraint_scan_by_chunk_id_internal(chunk_id,
 													  chunk_constraint_delete_tuple,
 													  NULL,
-													  NULL,
+													  &info,
 													  RowExclusiveLock);
 }
 

src/chunk_constraint.h

@@ -36,14 +36,14 @@ typedef struct ChunkScanCtx ChunkScanCtx;
 extern ChunkConstraints *chunk_constraints_alloc(int size_hint);
 extern ChunkConstraints *chunk_constraint_scan_by_chunk_id(int32 chunk_id, Size count_hint);
 extern ChunkConstraints *chunk_constraints_copy(ChunkConstraints *constraints);
+extern int	chunk_constraint_scan_by_dimension_slice(DimensionSlice *slice, ChunkScanCtx *ctx);
+extern int	chunk_constraint_scan_by_dimension_slice_id(int32 dimension_slice_id, ChunkConstraints *ccs);
 extern int	chunk_constraints_add_dimension_constraints(ChunkConstraints *ccs, int32 chunk_id, Hypercube *cube);
 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 int	chunk_constraint_scan_by_dimension_slice_id(DimensionSlice *slice, ChunkScanCtx *ctx);
 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_chunk_id(int32 chunk_i);
+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 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);