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/latest-dev.sql

@@ -188,3 +188,8 @@ SET index = COALESCE(index, '[]'::jsonb) ||
             )
 WHERE cs.orderby IS NOT NULL;
 
+CREATE OR REPLACE FUNCTION _timescaledb_functions.decompress_batch(record)
+   RETURNS SETOF record
+   AS '@MODULE_PATHNAME@', 'ts_update_placeholder'
+   LANGUAGE C STRICT;
+

sql/updates/reverse-dev.sql

@@ -9,3 +9,5 @@ DROP FUNCTION IF EXISTS _timescaledb_functions.bloom1_hash(anyelement);
 -- Drop BIGINT-returning version so the downgrade script can recreate the
 -- INTEGER-returning version of compressed_data_column_size.
 DROP FUNCTION IF EXISTS _timescaledb_functions.compressed_data_column_size(_timescaledb_internal.compressed_data, ANYELEMENT);
+
+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);
@@ -358,6 +359,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>
@@ -2680,6 +2681,87 @@ 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));
+
+		decompress_ctx = palloc0(sizeof(DecompressBatchSRFContext));
+		decompress_ctx->decompressor = build_decompressor(in_desc, out_desc);
+
+		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

@@ -338,6 +338,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

@@ -141,6 +141,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,

tsl/test/expected/decompress_batch.out

@@ -0,0 +1,99 @@
+-- This file and its contents are licensed under the Timescale License.
+-- Please see the included NOTICE for copyright information and
+-- LICENSE-TIMESCALE for a copy of the license.
+-- _timescaledb_functions.decompress_batch(record) RETURNS SETOF record
+-- expands a single row of a compressed chunk into the user-visible rows it
+-- was compressed from.
+SET datestyle TO ISO;
+CREATE TABLE metrics(time timestamptz NOT NULL, device_id int, value float)
+    WITH (tsdb.hypertable, tsdb.orderby = 'time', tsdb.segmentby = 'device_id');
+NOTICE:  using column "time" as partitioning column
+INSERT INTO metrics
+SELECT '2025-01-01'::timestamptz + (g || ' minute')::interval, g % 3, g::float
+FROM generate_series(1, 30) g;
+-- A row with NULLs in user columns to verify round-trip of NULLs.
+INSERT INTO metrics VALUES ('2025-01-01 02:00', NULL, NULL);
+SELECT count(*) AS source_rows FROM metrics;
+ source_rows 
+-------------
+          31
+
+SELECT count(compress_chunk(ch)) FROM show_chunks('metrics') ch;
+ count 
+-------
+     1
+
+-- Capture the compressed chunk relation name.
+SELECT format('%I.%I', cc.schema_name, cc.table_name) AS compressed_chunk
+FROM _timescaledb_catalog.chunk c
+JOIN _timescaledb_catalog.chunk cc ON c.compressed_chunk_id = cc.id
+JOIN _timescaledb_catalog.hypertable h ON c.hypertable_id = h.id
+WHERE h.table_name = 'metrics' \gset
+-- Expanding every compressed row reproduces the source rowcount.
+SELECT count(*) AS round_trip_count
+FROM :compressed_chunk t,
+     LATERAL _timescaledb_functions.decompress_batch(t)
+         AS r(time timestamptz, device_id int, value float);
+ round_trip_count 
+------------------
+               31
+
+-- Set equality: no source row missing, no extra row introduced.
+SELECT count(*) AS missing FROM (
+    TABLE metrics
+    EXCEPT ALL
+    SELECT r.time, r.device_id, r.value
+    FROM :compressed_chunk t,
+         LATERAL _timescaledb_functions.decompress_batch(t)
+             AS r(time timestamptz, device_id int, value float)
+) m;
+ missing 
+---------
+       0
+
+SELECT count(*) AS extras FROM (
+    SELECT r.time, r.device_id, r.value
+    FROM :compressed_chunk t,
+         LATERAL _timescaledb_functions.decompress_batch(t)
+             AS r(time timestamptz, device_id int, value float)
+    EXCEPT ALL
+    TABLE metrics
+) e;
+ extras 
+--------
+      0
+
+-- Per-batch counts: each compressed row is one segmentby batch.
+SELECT t.device_id AS segment, count(*) AS batch_rows
+FROM :compressed_chunk t,
+     LATERAL _timescaledb_functions.decompress_batch(t)
+         AS r(time timestamptz, device_id int, value float)
+GROUP BY t.device_id
+ORDER BY t.device_id NULLS LAST;
+ segment | batch_rows 
+---------+------------
+       0 |         10
+       1 |         10
+       2 |         10
+         |          1
+
+-- Decompressing a single batch yields exactly that batch.
+SELECT r.time, r.device_id, r.value
+FROM (SELECT t FROM :compressed_chunk t WHERE t.device_id = 1) one,
+     LATERAL _timescaledb_functions.decompress_batch(one.t)
+         AS r(time timestamptz, device_id int, value float)
+ORDER BY r.time;
+          time          | device_id | value 
+------------------------+-----------+-------
+ 2025-01-01 00:01:00-08 |         1 |     1
+ 2025-01-01 00:04:00-08 |         1 |     4
+ 2025-01-01 00:07:00-08 |         1 |     7
+ 2025-01-01 00:10:00-08 |         1 |    10
+ 2025-01-01 00:13:00-08 |         1 |    13
+ 2025-01-01 00:16:00-08 |         1 |    16
+ 2025-01-01 00:19:00-08 |         1 |    19
+ 2025-01-01 00:22:00-08 |         1 |    22
+ 2025-01-01 00:25:00-08 |         1 |    25
+ 2025-01-01 00:28:00-08 |         1 |    28
+
+DROP TABLE metrics CASCADE;