Add decompress_batch SQL function

Introduce _timescaledb_functions.decompress_batch(record), which
expands a single compressed-chunk row back into the individual rows
it represents. The function is intended for use from a custom
logical decoding plugin that needs to materialize the original
tuples from a compressed batch.

The input descriptor is recovered from the record's own type info
(typeId/typmod in the HeapTupleHeader) via lookup_rowtype_tupdesc,
so callers can pass a row from a compressed chunk directly. The
output descriptor is taken from the call site's column definition
list through get_call_result_type, letting the caller specify the
shape of the decompressed rows in the AS t(...) clause.

Internally the function is a value-per-call SRF that calls
build_decompressor, deforms the input record into the decompressor's
compressed_datums/_is_nulls arrays, runs decompress_batch once, and
yields each tuple from decompressed_slots. row_decompressor_close
runs on SRF_RETURN_DONE.

Example:

  SELECT x.*
  FROM _timescaledb_internal.compress_hyper_2_2_chunk t
  LIMIT 1
  CROSS JOIN LATERAL _timescaledb_functions.decompress_batch(t)
      AS x(time timestamptz, device_id int, value float);

Author: zilder

.unreleased/pr_9684

@@ -0,0 +1 @@
+Implements: #9684 Add `decompress_batch` SQL function

sql/compression.sql

@@ -12,3 +12,8 @@ CREATE OR REPLACE FUNCTION _timescaledb_functions.compressed_data_column_size(_t
    AS '@MODULE_PATHNAME@', 'ts_compressed_data_column_size'
    LANGUAGE C IMMUTABLE PARALLEL SAFE;
 
+CREATE OR REPLACE FUNCTION _timescaledb_functions.decompress_batch(record)
+   RETURNS SETOF record
+   AS '@MODULE_PATHNAME@', 'ts_decompress_batch'
+   LANGUAGE C STRICT;
+

sql/updates/reverse-dev.sql

@@ -279,3 +279,4 @@ ALTER TABLE _timescaledb_catalog.continuous_aggs_jobs_refresh_ranges
 ANALYZE _timescaledb_catalog.continuous_agg;
 -- end rebuild _timescaledb_catalog.continuous_agg --
 
+DROP FUNCTION IF EXISTS _timescaledb_functions.decompress_batch(record);

src/cross_module_fn.c

@@ -60,6 +60,7 @@ CROSSMODULE_WRAPPER(compressed_data_decompress_forward);
 CROSSMODULE_WRAPPER(compressed_data_decompress_reverse);
 CROSSMODULE_WRAPPER(compressed_data_column_size);
 CROSSMODULE_WRAPPER(compressed_data_to_array);
+CROSSMODULE_WRAPPER(decompress_batch);
 CROSSMODULE_WRAPPER(compressed_data_send);
 CROSSMODULE_WRAPPER(compressed_data_recv);
 CROSSMODULE_WRAPPER(compressed_data_in);
@@ -367,6 +368,7 @@ TSDLLEXPORT CrossModuleFunctions ts_cm_functions_default = {
 	.compressed_data_decompress_reverse = error_no_default_fn_pg_community,
 	.compressed_data_column_size = error_no_default_fn_pg_community,
 	.compressed_data_to_array = error_no_default_fn_pg_community,
+	.decompress_batch = error_no_default_fn_pg_community,
 	.deltadelta_compressor_append = error_no_default_fn_pg_community,
 	.deltadelta_compressor_finish = error_no_default_fn_pg_community,
 	.gorilla_compressor_append = error_no_default_fn_pg_community,

src/cross_module_fn.h

@@ -149,6 +149,7 @@ typedef struct CrossModuleFunctions
 	PGFunction compressed_data_decompress_reverse;
 	PGFunction compressed_data_column_size;
 	PGFunction compressed_data_to_array;
+	PGFunction decompress_batch;
 	PGFunction deltadelta_compressor_append;
 	PGFunction deltadelta_compressor_finish;
 	PGFunction gorilla_compressor_append;

tsl/src/compression/compression.c

@@ -7,6 +7,7 @@
 #include <access/attmap.h>
 #include <access/attnum.h>
 #include <access/detoast.h>
+#include <access/htup_details.h>
 #include <access/skey.h>
 #include <access/tupdesc.h>
 #include <catalog/heap.h>
@@ -146,7 +147,7 @@ static void row_compressor_process_ordered_slot(RowCompressor *row_compressor, T
 static void row_compressor_update_group(RowCompressor *row_compressor, TupleTableSlot *row);
 static bool row_compressor_new_row_is_in_new_group(RowCompressor *row_compressor,
 												   TupleTableSlot *row);
-static void create_per_compressed_column(RowDecompressor *decompressor);
+static void create_per_compressed_column(RowDecompressor *decompressor, bool internal_error);
 static void row_compressor_append_row(RowCompressor *row_compressor, TupleTableSlot *row);
 static void row_compressor_flush(RowCompressor *row_compressor, BulkWriter *writer,
 								 bool changed_groups);
@@ -2024,8 +2025,9 @@ bulk_writer_close(BulkWriter *writer)
  ** decompress_chunk **
  **********************/
 
-RowDecompressor
-build_decompressor(const TupleDesc in_desc, const TupleDesc out_desc, Oid in_oid, Oid out_oid)
+static inline RowDecompressor
+build_decompressor_common(const TupleDesc in_desc, const TupleDesc out_desc, Oid in_oid,
+						  Oid out_oid, bool internal_error)
 {
 	AttrNumber count_meta_attnum = InvalidAttrNumber;
 	AttrMap *attrmap = build_decompress_attrmap(out_desc, in_desc, &count_meta_attnum);
@@ -2056,7 +2058,7 @@ build_decompressor(const TupleDesc in_desc, const TupleDesc out_desc, Oid in_oid
 		.attrmap = attrmap,
 	};
 
-	create_per_compressed_column(&decompressor);
+	create_per_compressed_column(&decompressor, internal_error);
 
 	/*
 	 * We need to make sure decompressed_is_nulls is in a defined state. While this
@@ -2073,6 +2075,13 @@ build_decompressor(const TupleDesc in_desc, const TupleDesc out_desc, Oid in_oid
 	return decompressor;
 }
 
+RowDecompressor
+build_decompressor(const TupleDesc in_desc, const TupleDesc out_desc, Oid in_oid, Oid out_oid)
+{
+	return build_decompressor_common(in_desc, out_desc, in_oid, out_oid, true);
+}
+
+
 void
 row_decompressor_init_stats(RowDecompressor *decompressor, Oid compressed_relid,
 							Oid uncompressed_relid, CmdType cmd_type)
@@ -2192,7 +2201,7 @@ decompress_chunk(Oid in_table, Oid out_table)
 }
 
 static void
-create_per_compressed_column(RowDecompressor *decompressor)
+create_per_compressed_column(RowDecompressor *decompressor, bool internal_error)
 {
 	Oid compressed_data_type_oid = ts_custom_type_cache_get(CUSTOM_TYPE_COMPRESSED_DATA)->type_oid;
 	Assert(OidIsValid(compressed_data_type_oid));
@@ -2236,12 +2245,15 @@ create_per_compressed_column(RowDecompressor *decompressor)
 		is_compressed = compressed_attr->atttypid == compressed_data_type_oid;
 		if (!is_compressed && compressed_attr->atttypid != decompressed_type)
 		{
-			elog(ERROR,
-				 "compressed table type '%s' does not match decompressed table type '%s' for "
-				 "segment-by column \"%s\"",
-				 format_type_be(compressed_attr->atttypid),
-				 format_type_be(decompressed_type),
-				 col_name);
+
+			ereport(ERROR,
+					(errcode(internal_error ? ERRCODE_INTERNAL_ERROR :
+											  ERRCODE_INVALID_PARAMETER_VALUE),
+					 errmsg("compressed table type '%s' does not match decompressed "
+							"table type '%s' for segment-by column \"%s\"",
+							format_type_be(compressed_attr->atttypid),
+							format_type_be(decompressed_type),
+							col_name)));
 		}
 
 		*per_compressed_col = (PerCompressedColumn){
@@ -2769,6 +2781,133 @@ tsl_compressed_data_decompress_reverse(PG_FUNCTION_ARGS)
 	;
 }
 
+/*
+ * decompress_batch(compressed_tuple record) RETURNS SETOF record
+ *
+ * Decompresses a single compressed batch (one row of a compressed chunk) into
+ * the individual rows it represents. The shape of the input compressed tuple
+ * is taken from the record's own type info (typeId/typmod in the header).
+ * The shape of the output rows is taken from the call site's column
+ * definition list (the AS t(...) clause).
+ */
+typedef struct DecompressBatchSRFContext
+{
+	RowDecompressor decompressor;
+	int next_row;
+	int total_rows;
+} DecompressBatchSRFContext;
+
+Datum
+tsl_decompress_batch(PG_FUNCTION_ARGS)
+{
+	FuncCallContext *funcctx;
+	DecompressBatchSRFContext *decompress_ctx;
+
+	if (SRF_IS_FIRSTCALL())
+	{
+		funcctx = SRF_FIRSTCALL_INIT();
+		MemoryContext oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
+
+		TupleDesc out_desc;
+		if (get_call_result_type(fcinfo, NULL, &out_desc) != TYPEFUNC_COMPOSITE)
+		{
+			ereport(ERROR,
+					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+					 errmsg("function returning record called in context "
+							"that cannot accept type record")));
+		}
+		BlessTupleDesc(out_desc);
+
+		HeapTupleHeader td = PG_GETARG_HEAPTUPLEHEADER(0);
+		TupleDesc in_desc =
+			lookup_rowtype_tupdesc(HeapTupleHeaderGetTypeId(td), HeapTupleHeaderGetTypMod(td));
+
+		/*
+		 * Verify that the input record actually looks like a compressed-chunk
+		 * row before handing it to the decompressor. The decompressor identifies
+		 * the batch row count through the "_ts_meta_count" metadata column and
+		 * indexes into the compressed datums using its position; a record that
+		 * lacks this column would otherwise lead to an out-of-bounds access. The
+		 * record is fully caller-controlled, so this must be a runtime check
+		 * rather than an assertion.
+		 */
+		bool has_count_column = false;
+		for (int i = 0; i < in_desc->natts; i++)
+		{
+			Form_pg_attribute attr = TupleDescAttr(in_desc, i);
+
+			if (attr->attisdropped)
+			{
+				continue;
+			}
+
+			if (strcmp(NameStr(attr->attname), COMPRESSION_COLUMN_METADATA_COUNT_NAME) == 0)
+			{
+				if (attr->atttypid != INT4OID)
+				{
+					ereport(ERROR,
+							(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+							 errmsg("input record is not a compressed batch"),
+							 errdetail("Column \"%s\" must have type integer.",
+									   COMPRESSION_COLUMN_METADATA_COUNT_NAME)));
+				}
+
+				has_count_column = true;
+				break;
+			}
+		}
+
+		if (!has_count_column)
+		{
+			ReleaseTupleDesc(in_desc);
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+					 errmsg("input record is not a compressed batch"),
+					 errdetail("A compressed batch must have a \"%s\" metadata column.",
+							   COMPRESSION_COLUMN_METADATA_COUNT_NAME)));
+		}
+
+		decompress_ctx = palloc0(sizeof(DecompressBatchSRFContext));
+		decompress_ctx->decompressor =
+			build_decompressor_common(in_desc, out_desc, InvalidOid, InvalidOid, false);
+
+		HeapTupleData tmp;
+		tmp.t_len = HeapTupleHeaderGetDatumLength(td);
+		ItemPointerSetInvalid(&tmp.t_self);
+		tmp.t_tableOid = InvalidOid;
+		tmp.t_data = td;
+
+		heap_deform_tuple(&tmp,
+						  in_desc,
+						  decompress_ctx->decompressor.compressed_datums,
+						  decompress_ctx->decompressor.compressed_is_nulls);
+
+		ReleaseTupleDesc(in_desc);
+
+		decompress_ctx->total_rows = decompress_batch(&decompress_ctx->decompressor);
+		decompress_ctx->next_row = 0;
+
+		funcctx->user_fctx = decompress_ctx;
+		funcctx->tuple_desc = decompress_ctx->decompressor.out_desc;
+		MemoryContextSwitchTo(oldcontext);
+	}
+
+	funcctx = SRF_PERCALL_SETUP();
+	decompress_ctx = (DecompressBatchSRFContext *) funcctx->user_fctx;
+
+	if (decompress_ctx->next_row >= decompress_ctx->total_rows)
+	{
+		row_decompressor_close(&decompress_ctx->decompressor);
+		SRF_RETURN_DONE(funcctx);
+	}
+
+	TupleTableSlot *slot =
+		decompress_ctx->decompressor.decompressed_slots[decompress_ctx->next_row++];
+	bool should_free;
+	HeapTuple tuple = ExecFetchSlotHeapTuple(slot, false, &should_free);
+	SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(tuple));
+}
+
 /*
  * compressed_data_to_array(compressed_data, element_type) -> anyarray
  */

tsl/src/compression/compression.h

@@ -351,6 +351,7 @@ extern Datum tsl_compressed_data_info(PG_FUNCTION_ARGS);
 extern Datum tsl_compressed_data_has_nulls(PG_FUNCTION_ARGS);
 extern Datum tsl_compressed_data_column_size(PG_FUNCTION_ARGS);
 extern Datum tsl_compressed_data_to_array(PG_FUNCTION_ARGS);
+extern Datum tsl_decompress_batch(PG_FUNCTION_ARGS);
 
 static void
 pg_attribute_unused() assert_num_compression_algorithms_sane(void)

tsl/src/init.c

@@ -142,6 +142,7 @@ CrossModuleFunctions tsl_cm_functions = {
 	.compressed_data_decompress_reverse = tsl_compressed_data_decompress_reverse,
 	.compressed_data_column_size = tsl_compressed_data_column_size,
 	.compressed_data_to_array = tsl_compressed_data_to_array,
+	.decompress_batch = tsl_decompress_batch,
 	.compressed_data_send = tsl_compressed_data_send,
 	.compressed_data_recv = tsl_compressed_data_recv,
 	.compressed_data_in = tsl_compressed_data_in,