Simplify retention policy chunk dropping

Call function to drop chunks directly instead of going through
function manager.

Author: svenklemm

src/chunk.c

@@ -4001,6 +4001,84 @@ ts_chunk_drop_single_chunk(PG_FUNCTION_ARGS)
 	PG_RETURN_BOOL(true);
 }
 
+/*
+ * Drop chunks from a hypertable, dropping everything older than the given boundary.
+ *
+ * The boundary is either an "older_than" value (matched against the chunk
+ * ranges) or, when use_creation_time is set, a "drop_created_before" value
+ * (matched against the chunk creation time).
+ *
+ * Returns the number of dropped chunks.
+ */
+int
+ts_chunk_drop_chunks_by_boundary(Oid relid, Datum older_than, Oid older_than_type,
+								 bool use_creation_time)
+{
+	Cache *hcache = ts_hypertable_cache_pin();
+	Hypertable *ht = ts_resolve_hypertable_from_table_or_cagg(hcache, relid, false);
+	const Dimension *time_dim = hyperspace_get_open_dimension(ht->space, 0);
+	Oid time_type = ts_dimension_get_partition_type(time_dim);
+	int64 older_than_internal;
+	bool older_newer;
+	List *dropped_chunks;
+	MemoryContext oldcontext = CurrentMemoryContext;
+
+	/* UUID (v7) partitioning is treated as TIMESTAMPTZ, matching the boundary. */
+	if (IS_UUID_TYPE(time_type))
+	{
+		time_type = TIMESTAMPTZOID;
+	}
+
+	if (use_creation_time)
+	{
+		/* drop_created_before compares against the chunk creation time. */
+		int64 created_before =
+			ts_time_value_from_arg(older_than, older_than_type, TIMESTAMPTZOID, false);
+		older_than_internal = ts_internal_to_time_int64(created_before, TIMESTAMPTZOID);
+		older_newer = false;
+	}
+	else
+	{
+		older_than_internal = ts_time_value_from_arg(older_than, older_than_type, time_type, true);
+		older_newer = true;
+	}
+
+	PG_TRY();
+	{
+		dropped_chunks = ts_chunk_do_drop_chunks(ht,
+												 older_than_internal,
+												 PG_INT64_MIN,
+												 DEBUG2,
+												 time_type,
+												 older_than_type,
+												 older_newer);
+	}
+	PG_CATCH();
+	{
+		/* Replace the generic dependent objects hint with an accurate one since
+		 * we don't support CASCADE here. */
+		ErrorData *edata;
+
+		/* CopyErrorData requires we leave the error context first. */
+		MemoryContextSwitchTo(oldcontext);
+		edata = CopyErrorData();
+		FlushErrorState();
+
+		if (edata->sqlerrcode == ERRCODE_DEPENDENT_OBJECTS_STILL_EXIST)
+		{
+			edata->hint = pstrdup("Use DROP ... to drop the dependent objects.");
+		}
+
+		ts_cache_release(&hcache);
+		ReThrowError(edata);
+	}
+	PG_END_TRY();
+
+	ts_cache_release(&hcache);
+
+	return list_length(dropped_chunks);
+}
+
 Datum
 ts_chunk_drop_chunks(PG_FUNCTION_ARGS)
 {

src/chunk.h

@@ -213,6 +213,9 @@ extern TSDLLEXPORT void ts_chunk_drop_by_relid(Oid relid, DropBehavior behavior,
 extern TSDLLEXPORT List *ts_chunk_do_drop_chunks(Hypertable *ht, int64 older_than, int64 newer_than,
 												 int32 log_level, Oid time_type, Oid arg_type,
 												 bool older_newer);
+extern TSDLLEXPORT int ts_chunk_drop_chunks_by_boundary(Oid relid, Datum older_than,
+														Oid older_than_type,
+														bool use_creation_time);
 extern TSDLLEXPORT Chunk *
 ts_chunk_find_or_create_without_cuts(const Hypertable *ht, Hypercube *hc, const char *schema_name,
 									 const char *table_name, Oid chunk_table_relid, bool *created);

tsl/src/bgw_policy/job.c

@@ -300,10 +300,10 @@ policy_retention_execute(int32 job_id, Jsonb *config)
 		log_retention_boundary(LOG, &policy_data, "applying retention policy to hypertable");
 	}
 
-	chunk_invoke_drop_chunks(policy_data.object_relid,
-							 policy_data.boundary,
-							 policy_data.boundary_type,
-							 policy_data.use_creation_time);
+	ts_chunk_drop_chunks_by_boundary(policy_data.object_relid,
+									 policy_data.boundary,
+									 policy_data.boundary_type,
+									 policy_data.use_creation_time);
 
 	return true;
 }

tsl/src/chunk.c

@@ -34,8 +34,6 @@
 #include "chunk.h"
 #include "compression/compression.h"
 #include "debug_point.h"
-#include "extension.h"
-#include "hypertable.h"
 #include "utils.h"
 
 /* Data in a frozen chunk cannot be modified. So any operation
@@ -94,104 +92,3 @@ chunk_unfreeze_chunk(PG_FUNCTION_ARGS)
 	bool ret = ts_chunk_unset_frozen(chunk);
 	PG_RETURN_BOOL(ret);
 }
-
-/*
- * Invoke drop_chunks via fmgr so that the call can be deparsed and sent to
- * remote data nodes.
- *
- * Given that drop_chunks is an SRF, and has pseudo parameter types, we need
- * to provide a FuncExpr with type information for the deparser.
- *
- * Returns the number of dropped chunks.
- */
-int
-chunk_invoke_drop_chunks(Oid relid, Datum older_than, Datum older_than_type, bool use_creation_time)
-{
-	EState *estate;
-	ExprContext *econtext;
-	FuncExpr *fexpr;
-	List *args = NIL;
-	int num_results = 0;
-	SetExprState *state;
-	Oid restype;
-	Oid func_oid;
-	Const *TypeNullCons = makeNullConst(older_than_type, -1, InvalidOid);
-	Const *IntervalVal = makeConst(older_than_type,
-								   -1,
-								   InvalidOid,
-								   get_typlen(older_than_type),
-								   older_than,
-								   false,
-								   get_typbyval(older_than_type));
-	Const *argarr[DROP_CHUNKS_NARGS] = { makeConst(REGCLASSOID,
-												   -1,
-												   InvalidOid,
-												   sizeof(relid),
-												   ObjectIdGetDatum(relid),
-												   false,
-												   false),
-										 TypeNullCons,
-										 TypeNullCons,
-										 castNode(Const, makeBoolConst(false, true)),
-										 TypeNullCons,
-										 TypeNullCons };
-	Oid type_id[DROP_CHUNKS_NARGS] = { REGCLASSOID, ANYOID, ANYOID, BOOLOID, ANYOID, ANYOID };
-	char *const schema_name = ts_extension_schema_name();
-	List *const fqn = list_make2(makeString(schema_name), makeString(DROP_CHUNKS_FUNCNAME));
-
-	StaticAssertStmt(lengthof(type_id) == lengthof(argarr),
-					 "argarr and type_id should have matching lengths");
-
-	func_oid = LookupFuncName(fqn, lengthof(type_id), type_id, false);
-	Assert(func_oid); /* LookupFuncName should not return an invalid OID */
-
-	/* decide whether to use "older_than" or "drop_created_before" */
-	if (use_creation_time)
-	{
-		argarr[4] = IntervalVal;
-	}
-	else
-	{
-		argarr[1] = IntervalVal;
-	}
-
-	/* Prepare the function expr with argument list */
-	get_func_result_type(func_oid, &restype, NULL);
-
-	for (size_t i = 0; i < lengthof(argarr); i++)
-	{
-		args = lappend(args, argarr[i]);
-	}
-
-	fexpr = makeFuncExpr(func_oid, restype, args, InvalidOid, InvalidOid, COERCE_EXPLICIT_CALL);
-	fexpr->funcretset = true;
-
-	/* Execute the SRF */
-	estate = CreateExecutorState();
-	econtext = CreateExprContext(estate);
-	state = ExecInitFunctionResultSet(&fexpr->xpr, econtext, NULL);
-
-	while (true)
-	{
-		ExprDoneCond isdone;
-		bool isnull;
-
-		ExecMakeFunctionResultSet(state, econtext, estate->es_query_cxt, &isnull, &isdone);
-
-		if (isdone == ExprEndResult)
-		{
-			break;
-		}
-
-		if (!isnull)
-		{
-			num_results++;
-		}
-	}
-
-	/* Cleanup */
-	FreeExprContext(econtext, false);
-	FreeExecutorState(estate);
-
-	return num_results;
-}

tsl/src/chunk.h

@@ -11,8 +11,6 @@
 
 extern Datum chunk_freeze_chunk(PG_FUNCTION_ARGS);
 extern Datum chunk_unfreeze_chunk(PG_FUNCTION_ARGS);
-extern int chunk_invoke_drop_chunks(Oid relid, Datum older_than, Datum older_than_type,
-									bool use_creation_time);
 extern Datum chunk_merge_chunks(PG_FUNCTION_ARGS);
 extern Datum chunk_split_chunk(PG_FUNCTION_ARGS);
 extern void update_relstats(Relation catrel, Oid relid, BlockNumber num_pages, double ntuples);