Support RETURNING clause with direct compress

The direct compress insert path added each tuple to the compressor and
moved on to the next row without ever evaluating the RETURNING clause,
so an INSERT with RETURNING returned no rows even though the data was
inserted. Project and return the RETURNING result for each inserted row
in the direct compress path, matching the normal insert path.

Fixes #10243

Author: svenklemm

.unreleased/fix_direct_compress_returning

@@ -0,0 +1 @@
+Fixes: #10280 RETURNING clause returned no rows for INSERT using Direct Compress

src/nodes/modify_hypertable_exec.c

@@ -2523,6 +2523,20 @@ ExecModifyTable(CustomScanState *cs_node, PlanState *pstate)
 				ts_cm_functions->compressor_add_slot(ht_state->compressor, ht_state->bulk_writer, chunk_slot);
 				if (node->canSetTag)
 					estate->es_processed++;
+
+				/*
+				 * Project the RETURNING result. Direct compress skips
+				 * ExecInsert, which normally handles this.
+				 */
+				if (chunk_rri->ri_projectReturning)
+				{
+					chunk_slot->tts_tableOid = RelationGetRelid(ctr->cis->rel);
+					return ExecProcessReturning(chunk_rri,
+												CMD_INSERT,
+												NULL,
+												chunk_slot,
+												context.planSlot);
+				}
 				continue;
 			}
 

tsl/test/expected/direct_compress_insert.out

@@ -961,3 +961,47 @@ SELECT count(*) FROM dc_excl;
 
 RESET timescaledb.enable_direct_compress_insert;
 DROP TABLE dc_excl;
+-- Direct compress must return rows for a RETURNING clause
+CREATE TABLE dc_ret(time timestamptz NOT NULL, device text NOT NULL, val int NOT NULL)
+    WITH (tsdb.hypertable, tsdb.partition_column='time');
+ALTER TABLE dc_ret SET (timescaledb.compress, timescaledb.compress_segmentby='device');
+NOTICE:  updated compression settings will only apply to future compressions
+SET timescaledb.enable_direct_compress_insert = true;
+-- confirm the direct compress path is used for this insert
+EXPLAIN (COSTS OFF, SUMMARY OFF, TIMING OFF)
+INSERT INTO dc_ret SELECT '2025-01-01'::timestamptz + (i * INTERVAL '1 minute'), 'd' || (i % 2)::text, i
+    FROM generate_series(1, 20) i RETURNING val;
+--- QUERY PLAN ---
+ Custom Scan (ModifyHypertable)
+   Direct Compress: true
+   ->  Insert on dc_ret
+         ->  Function Scan on generate_series i
+
+-- RETURNING should return one row per inserted tuple
+WITH inserted AS (
+    INSERT INTO dc_ret SELECT '2025-01-01'::timestamptz + (i * INTERVAL '1 minute'), 'd' || (i % 2)::text, i
+        FROM generate_series(1, 20) i RETURNING val, device
+)
+SELECT count(*), min(val), max(val), sum(val), count(DISTINCT device) FROM inserted;
+ count | min | max | sum | count 
+-------+-----+-----+-----+-------
+    20 |   1 |  20 | 210 |     2
+
+-- RETURNING with an expression referencing multiple columns
+WITH inserted AS (
+    INSERT INTO dc_ret SELECT '2025-01-01'::timestamptz + (i * INTERVAL '1 hour'), 'd' || (i % 2)::text, i
+        FROM generate_series(1, 20) i RETURNING device || ':' || val AS label
+)
+SELECT count(*), min(label), max(label) FROM inserted;
+ count |  min  | max  
+-------+-------+------
+    20 | d0:10 | d1:9
+
+-- rows were actually inserted
+SELECT count(*) FROM dc_ret;
+ count 
+-------
+    40
+
+RESET timescaledb.enable_direct_compress_insert;
+DROP TABLE dc_ret;

tsl/test/sql/direct_compress_insert.sql

@@ -569,3 +569,29 @@ SELECT count(*) FROM dc_excl;
 RESET timescaledb.enable_direct_compress_insert;
 DROP TABLE dc_excl;
 
+-- Direct compress must return rows for a RETURNING clause
+CREATE TABLE dc_ret(time timestamptz NOT NULL, device text NOT NULL, val int NOT NULL)
+    WITH (tsdb.hypertable, tsdb.partition_column='time');
+ALTER TABLE dc_ret SET (timescaledb.compress, timescaledb.compress_segmentby='device');
+SET timescaledb.enable_direct_compress_insert = true;
+-- confirm the direct compress path is used for this insert
+EXPLAIN (COSTS OFF, SUMMARY OFF, TIMING OFF)
+INSERT INTO dc_ret SELECT '2025-01-01'::timestamptz + (i * INTERVAL '1 minute'), 'd' || (i % 2)::text, i
+    FROM generate_series(1, 20) i RETURNING val;
+-- RETURNING should return one row per inserted tuple
+WITH inserted AS (
+    INSERT INTO dc_ret SELECT '2025-01-01'::timestamptz + (i * INTERVAL '1 minute'), 'd' || (i % 2)::text, i
+        FROM generate_series(1, 20) i RETURNING val, device
+)
+SELECT count(*), min(val), max(val), sum(val), count(DISTINCT device) FROM inserted;
+-- RETURNING with an expression referencing multiple columns
+WITH inserted AS (
+    INSERT INTO dc_ret SELECT '2025-01-01'::timestamptz + (i * INTERVAL '1 hour'), 'd' || (i % 2)::text, i
+        FROM generate_series(1, 20) i RETURNING device || ':' || val AS label
+)
+SELECT count(*), min(label), max(label) FROM inserted;
+-- rows were actually inserted
+SELECT count(*) FROM dc_ret;
+RESET timescaledb.enable_direct_compress_insert;
+DROP TABLE dc_ret;
+