Skip to content

Commit 742fabd

Browse files
zilderzilder
committed
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);
1 parent 0273d49 commit 742fabd

13 files changed

Lines changed: 322 additions & 0 deletions

File tree

‎.unreleased/pr_9684‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
Implements: #9684 Add `decompress_batch` SQL function

‎sql/compression.sql‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,3 +12,8 @@ CREATE OR REPLACE FUNCTION _timescaledb_functions.compressed_data_column_size(_t
1212
AS '@MODULE_PATHNAME@', 'ts_compressed_data_column_size'
1313
LANGUAGE C IMMUTABLE PARALLEL SAFE;
1414

15+
CREATE OR REPLACE FUNCTION _timescaledb_functions.decompress_batch(record)
16+
RETURNS SETOF record
17+
AS '@MODULE_PATHNAME@', 'ts_decompress_batch'
18+
LANGUAGE C STRICT;
19+

‎sql/updates/latest-dev.sql‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,3 +50,8 @@ WHERE dimension_slice_id IS NULL
5050
ALTER TABLE _timescaledb_catalog.hypertable SET (user_catalog_table = true);
5151
ALTER TABLE _timescaledb_catalog.chunk SET (user_catalog_table = true);
5252

53+
CREATE OR REPLACE FUNCTION _timescaledb_functions.decompress_batch(record)
54+
RETURNS SETOF record
55+
AS '@MODULE_PATHNAME@', 'ts_update_placeholder'
56+
LANGUAGE C STRICT;
57+

‎sql/updates/reverse-dev.sql‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,3 +49,4 @@ ON CONFLICT DO NOTHING;
4949
ALTER TABLE _timescaledb_catalog.hypertable RESET (user_catalog_table);
5050
ALTER TABLE _timescaledb_catalog.chunk RESET (user_catalog_table);
5151

52+
DROP FUNCTION IF EXISTS _timescaledb_functions.decompress_batch(record);

‎src/cross_module_fn.c‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -60,6 +60,7 @@ CROSSMODULE_WRAPPER(compressed_data_decompress_forward);
6060
CROSSMODULE_WRAPPER(compressed_data_decompress_reverse);
6161
CROSSMODULE_WRAPPER(compressed_data_column_size);
6262
CROSSMODULE_WRAPPER(compressed_data_to_array);
63+
CROSSMODULE_WRAPPER(decompress_batch);
6364
CROSSMODULE_WRAPPER(compressed_data_send);
6465
CROSSMODULE_WRAPPER(compressed_data_recv);
6566
CROSSMODULE_WRAPPER(compressed_data_in);
@@ -357,6 +358,7 @@ TSDLLEXPORT CrossModuleFunctions ts_cm_functions_default = {
357358
.compressed_data_decompress_reverse = error_no_default_fn_pg_community,
358359
.compressed_data_column_size = error_no_default_fn_pg_community,
359360
.compressed_data_to_array = error_no_default_fn_pg_community,
361+
.decompress_batch = error_no_default_fn_pg_community,
360362
.deltadelta_compressor_append = error_no_default_fn_pg_community,
361363
.deltadelta_compressor_finish = error_no_default_fn_pg_community,
362364
.gorilla_compressor_append = error_no_default_fn_pg_community,

‎src/cross_module_fn.h‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -147,6 +147,7 @@ typedef struct CrossModuleFunctions
147147
PGFunction compressed_data_decompress_reverse;
148148
PGFunction compressed_data_column_size;
149149
PGFunction compressed_data_to_array;
150+
PGFunction decompress_batch;
150151
PGFunction deltadelta_compressor_append;
151152
PGFunction deltadelta_compressor_finish;
152153
PGFunction gorilla_compressor_append;

‎tsl/src/compression/compression.c‎

Lines changed: 128 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
#include <access/attmap.h>
88
#include <access/attnum.h>
99
#include <access/detoast.h>
10+
#include <access/htup_details.h>
1011
#include <access/skey.h>
1112
#include <access/tupdesc.h>
1213
#include <catalog/heap.h>
@@ -2765,6 +2766,133 @@ tsl_compressed_data_decompress_reverse(PG_FUNCTION_ARGS)
27652766
;
27662767
}
27672768

2769+
/*
2770+
* decompress_batch(compressed_tuple record) RETURNS SETOF record
2771+
*
2772+
* Decompresses a single compressed batch (one row of a compressed chunk) into
2773+
* the individual rows it represents. The shape of the input compressed tuple
2774+
* is taken from the record's own type info (typeId/typmod in the header).
2775+
* The shape of the output rows is taken from the call site's column
2776+
* definition list (the AS t(...) clause).
2777+
*/
2778+
typedef struct DecompressBatchSRFContext
2779+
{
2780+
RowDecompressor decompressor;
2781+
int next_row;
2782+
int total_rows;
2783+
} DecompressBatchSRFContext;
2784+
2785+
Datum
2786+
tsl_decompress_batch(PG_FUNCTION_ARGS)
2787+
{
2788+
FuncCallContext *funcctx;
2789+
DecompressBatchSRFContext *decompress_ctx;
2790+
2791+
if (SRF_IS_FIRSTCALL())
2792+
{
2793+
funcctx = SRF_FIRSTCALL_INIT();
2794+
MemoryContext oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
2795+
2796+
TupleDesc out_desc;
2797+
if (get_call_result_type(fcinfo, NULL, &out_desc) != TYPEFUNC_COMPOSITE)
2798+
{
2799+
ereport(ERROR,
2800+
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
2801+
errmsg("function returning record called in context "
2802+
"that cannot accept type record")));
2803+
}
2804+
BlessTupleDesc(out_desc);
2805+
2806+
HeapTupleHeader td = PG_GETARG_HEAPTUPLEHEADER(0);
2807+
TupleDesc in_desc =
2808+
lookup_rowtype_tupdesc(HeapTupleHeaderGetTypeId(td), HeapTupleHeaderGetTypMod(td));
2809+
2810+
/*
2811+
* Verify that the input record actually looks like a compressed-chunk
2812+
* row before handing it to the decompressor. The decompressor identifies
2813+
* the batch row count through the "_ts_meta_count" metadata column and
2814+
* indexes into the compressed datums using its position; a record that
2815+
* lacks this column would otherwise lead to an out-of-bounds access. The
2816+
* record is fully caller-controlled, so this must be a runtime check
2817+
* rather than an assertion.
2818+
*/
2819+
bool has_count_column = false;
2820+
for (int i = 0; i < in_desc->natts; i++)
2821+
{
2822+
Form_pg_attribute attr = TupleDescAttr(in_desc, i);
2823+
2824+
if (attr->attisdropped)
2825+
{
2826+
continue;
2827+
}
2828+
2829+
if (strcmp(NameStr(attr->attname), COMPRESSION_COLUMN_METADATA_COUNT_NAME) == 0)
2830+
{
2831+
if (attr->atttypid != INT4OID)
2832+
{
2833+
ereport(ERROR,
2834+
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
2835+
errmsg("input record is not a compressed batch"),
2836+
errdetail("Column \"%s\" must have type integer.",
2837+
COMPRESSION_COLUMN_METADATA_COUNT_NAME)));
2838+
}
2839+
2840+
has_count_column = true;
2841+
break;
2842+
}
2843+
}
2844+
2845+
if (!has_count_column)
2846+
{
2847+
ReleaseTupleDesc(in_desc);
2848+
ereport(ERROR,
2849+
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
2850+
errmsg("input record is not a compressed batch"),
2851+
errdetail("A compressed batch must have a \"%s\" metadata column.",
2852+
COMPRESSION_COLUMN_METADATA_COUNT_NAME)));
2853+
}
2854+
2855+
decompress_ctx = palloc0(sizeof(DecompressBatchSRFContext));
2856+
decompress_ctx->decompressor =
2857+
build_decompressor(in_desc, out_desc, InvalidOid, InvalidOid);
2858+
2859+
HeapTupleData tmp;
2860+
tmp.t_len = HeapTupleHeaderGetDatumLength(td);
2861+
ItemPointerSetInvalid(&tmp.t_self);
2862+
tmp.t_tableOid = InvalidOid;
2863+
tmp.t_data = td;
2864+
2865+
heap_deform_tuple(&tmp,
2866+
in_desc,
2867+
decompress_ctx->decompressor.compressed_datums,
2868+
decompress_ctx->decompressor.compressed_is_nulls);
2869+
2870+
ReleaseTupleDesc(in_desc);
2871+
2872+
decompress_ctx->total_rows = decompress_batch(&decompress_ctx->decompressor);
2873+
decompress_ctx->next_row = 0;
2874+
2875+
funcctx->user_fctx = decompress_ctx;
2876+
funcctx->tuple_desc = decompress_ctx->decompressor.out_desc;
2877+
MemoryContextSwitchTo(oldcontext);
2878+
}
2879+
2880+
funcctx = SRF_PERCALL_SETUP();
2881+
decompress_ctx = (DecompressBatchSRFContext *) funcctx->user_fctx;
2882+
2883+
if (decompress_ctx->next_row >= decompress_ctx->total_rows)
2884+
{
2885+
row_decompressor_close(&decompress_ctx->decompressor);
2886+
SRF_RETURN_DONE(funcctx);
2887+
}
2888+
2889+
TupleTableSlot *slot =
2890+
decompress_ctx->decompressor.decompressed_slots[decompress_ctx->next_row++];
2891+
bool should_free;
2892+
HeapTuple tuple = ExecFetchSlotHeapTuple(slot, false, &should_free);
2893+
SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(tuple));
2894+
}
2895+
27682896
/*
27692897
* compressed_data_to_array(compressed_data, element_type) -> anyarray
27702898
*/

‎tsl/src/compression/compression.h‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -351,6 +351,7 @@ extern Datum tsl_compressed_data_info(PG_FUNCTION_ARGS);
351351
extern Datum tsl_compressed_data_has_nulls(PG_FUNCTION_ARGS);
352352
extern Datum tsl_compressed_data_column_size(PG_FUNCTION_ARGS);
353353
extern Datum tsl_compressed_data_to_array(PG_FUNCTION_ARGS);
354+
extern Datum tsl_decompress_batch(PG_FUNCTION_ARGS);
354355

355356
static void
356357
pg_attribute_unused() assert_num_compression_algorithms_sane(void)

‎tsl/src/init.c‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -140,6 +140,7 @@ CrossModuleFunctions tsl_cm_functions = {
140140
.compressed_data_decompress_reverse = tsl_compressed_data_decompress_reverse,
141141
.compressed_data_column_size = tsl_compressed_data_column_size,
142142
.compressed_data_to_array = tsl_compressed_data_to_array,
143+
.decompress_batch = tsl_decompress_batch,
143144
.compressed_data_send = tsl_compressed_data_send,
144145
.compressed_data_recv = tsl_compressed_data_recv,
145146
.compressed_data_in = tsl_compressed_data_in,

‎tsl/test/expected/decompress_batch.out‎

Lines changed: 96 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,96 @@
1+
-- This file and its contents are licensed under the Timescale License.
2+
-- Please see the included NOTICE for copyright information and
3+
-- LICENSE-TIMESCALE for a copy of the license.
4+
-- _timescaledb_functions.decompress_batch(record) RETURNS SETOF record
5+
-- expands a single row of a compressed chunk into the user-visible rows it
6+
-- was compressed from.
7+
SET datestyle TO ISO;
8+
CREATE TABLE metrics(time timestamptz NOT NULL, device_id int, value float)
9+
WITH (tsdb.hypertable, tsdb.orderby = 'time', tsdb.segmentby = 'device_id');
10+
NOTICE: using column "time" as partitioning column
11+
-- Populate with test data
12+
INSERT INTO metrics
13+
SELECT '2025-01-01'::timestamptz + (g || ' minute')::interval, g % 3, g::float
14+
FROM generate_series(1, 30) g;
15+
INSERT INTO metrics VALUES ('2025-01-01 02:00', NULL, NULL);
16+
SELECT compress_chunk(ch) FROM show_chunks('metrics') ch;
17+
compress_chunk
18+
----------------------------------------
19+
_timescaledb_internal._hyper_1_1_chunk
20+
21+
-- Capture the compressed chunk relation name
22+
SELECT format('%I.%I', cc.schema_name, cc.table_name) AS compressed_chunk
23+
FROM _timescaledb_catalog.chunk c
24+
JOIN _timescaledb_catalog.chunk cc ON c.compressed_chunk_id = cc.id
25+
JOIN _timescaledb_catalog.hypertable h ON c.hypertable_id = h.id
26+
WHERE h.table_name = 'metrics' \gset
27+
-- Verify set equality: no source row missing, no extra row introduced
28+
SELECT count(*) AS missing FROM (
29+
TABLE metrics
30+
EXCEPT ALL
31+
SELECT decomp.time, decomp.device_id, decomp.value
32+
FROM :compressed_chunk t,
33+
LATERAL _timescaledb_functions.decompress_batch(t)
34+
AS decomp(time timestamptz, device_id int, value float)
35+
) m;
36+
missing
37+
---------
38+
0
39+
40+
SELECT count(*) AS extras FROM (
41+
SELECT decomp.time, decomp.device_id, decomp.value
42+
FROM :compressed_chunk t,
43+
LATERAL _timescaledb_functions.decompress_batch(t)
44+
AS decomp(time timestamptz, device_id int, value float)
45+
EXCEPT ALL
46+
TABLE metrics
47+
) e;
48+
extras
49+
--------
50+
0
51+
52+
-- Decompressing a single batch yields exactly that batch.
53+
SELECT decomp.time, decomp.device_id, decomp.value
54+
FROM (SELECT t FROM :compressed_chunk t WHERE t.device_id = 1) comp,
55+
LATERAL _timescaledb_functions.decompress_batch(comp.t)
56+
AS decomp(time timestamptz, device_id int, value float)
57+
ORDER BY decomp.time;
58+
time | device_id | value
59+
------------------------+-----------+-------
60+
2025-01-01 00:01:00-08 | 1 | 1
61+
2025-01-01 00:04:00-08 | 1 | 4
62+
2025-01-01 00:07:00-08 | 1 | 7
63+
2025-01-01 00:10:00-08 | 1 | 10
64+
2025-01-01 00:13:00-08 | 1 | 13
65+
2025-01-01 00:16:00-08 | 1 | 16
66+
2025-01-01 00:19:00-08 | 1 | 19
67+
2025-01-01 00:22:00-08 | 1 | 22
68+
2025-01-01 00:25:00-08 | 1 | 25
69+
2025-01-01 00:28:00-08 | 1 | 28
70+
71+
DROP TABLE metrics CASCADE;
72+
\set ON_ERROR_STOP 0
73+
-- Calling `decompress_batch()` on a non-compressed record
74+
SELECT decomp.*
75+
FROM (VALUES (1, 2)) AS r(a, b),
76+
LATERAL _timescaledb_functions.decompress_batch(r)
77+
AS decomp(a int, b int);
78+
ERROR: input record is not a compressed batch
79+
-- Calling `decompress_batch()` on a "fake" compressed record with the
80+
-- `_ts_meta_count` column of incorrect type
81+
SELECT decomp.*
82+
FROM (VALUES (1::INT8, 2, 3)) AS r(_ts_meta_count, a, b),
83+
LATERAL _timescaledb_functions.decompress_batch(r)
84+
AS decomp(a int, b int);
85+
ERROR: input record is not a compressed batch
86+
\set ON_ERROR_STOP 1
87+
-- Calling `decompress_batch()` on a "fake" compressed record with zero
88+
-- compressed columns
89+
SELECT decomp.*
90+
FROM (VALUES (1,2,3)) AS r(_ts_meta_count, a, b),
91+
LATERAL _timescaledb_functions.decompress_batch(r)
92+
AS decomp(a int, b int);
93+
a | b
94+
---+---
95+
2 | 3
96+

0 commit comments

Comments
 (0)