Optimize direct compress status handling

svenklemm · svenklemm · commit 7247bce56e34 · 2025-08-25T16:15:50.000+02:00
Only mark the status as partial if we insert into an existing
uncompressed chunk but skip marking as partial when it is a
newly created chunk so the resulting status is fully compressed
instead.
diff --git a/.unreleased/pr_8529 b/.unreleased/pr_8529
@@ -0,0 +1 @@
+Implements: #8529 Optimize direct compress status handling
diff --git a/src/chunk_tuple_routing.c b/src/chunk_tuple_routing.c
@@ -66,7 +66,6 @@ ts_chunk_tuple_routing_find_chunk(ChunkTupleRouting *ctr, Point *point)
 {
 	Chunk *chunk = NULL;
 	ChunkInsertState *cis = NULL;
-
 	cis = ts_subspace_store_get(ctr->subspace, point);
 
 	/*
@@ -77,6 +76,9 @@ ts_chunk_tuple_routing_find_chunk(ChunkTupleRouting *ctr, Point *point)
 
 	if (!cis)
 	{
+		bool chunk_created = false;
+		bool needs_partial = false;
+
 		/*
 		 * Normally, for every row of the chunk except the first one, we expect
 		 * the chunk to exist already. The "create" function would take a lock
@@ -116,7 +118,10 @@ ts_chunk_tuple_routing_find_chunk(ChunkTupleRouting *ctr, Point *point)
 		}
 
 		if (!chunk)
+		{
 			chunk = ts_hypertable_create_chunk_for_point(ctr->hypertable, point);
+			chunk_created = true;
+		}
 
 		Ensure(chunk, "no chunk found or created");
 
@@ -137,10 +142,15 @@ ts_chunk_tuple_routing_find_chunk(ChunkTupleRouting *ctr, Point *point)
 				Chunk *compressed_chunk =
 					ts_cm_functions->compression_chunk_create(compressed_ht, chunk);
 				ts_chunk_set_compressed_chunk(chunk, compressed_chunk->fd.id);
+
+				/* mark chunk as partial unless completely new chunk */
+				if (!chunk_created)
+					needs_partial = true;
 			}
 		}
 
 		cis = chunk_insert_state_create(chunk->table_id, ctr);
+		cis->needs_partial = needs_partial;
 		ts_subspace_store_add(ctr->subspace, chunk->cube, cis, destroy_chunk_insert_state);
 	}
 
diff --git a/src/copy.c b/src/copy.c
@@ -289,6 +289,7 @@ TSCopyMultiInsertBufferInit(TSCopyMultiInsertInfo *miinfo, ChunkInsertState *cis
 				if (!ts_chunk_is_unordered(chunk))
 					ts_chunk_set_unordered(chunk);
 			}
+			cis->columnstore_insert = true;
 			break;
 		}
 	}
diff --git a/src/nodes/chunk_dispatch/chunk_insert_state.c b/src/nodes/chunk_dispatch/chunk_insert_state.c
@@ -582,7 +582,15 @@ ts_chunk_insert_state_destroy(ChunkInsertState *state)
 {
 	ResultRelInfo *rri = state->result_relation_info;
 
-	if (state->chunk_compressed && !state->chunk_partial)
+	/*
+	 * Check if we need to mark the chunk as partial.
+	 * We need to change chunk status to partial in the following cases:
+	 * - rowstore insert into compressed chunk
+	 * - columnstore insert into uncompressed chunk that is not a new chunk (flagged as
+	 * needs_partial in chunk_tuple_routing.c)
+	 */
+	if (state->chunk_compressed && !state->chunk_partial &&
+		(!state->columnstore_insert || state->needs_partial))
 	{
 		Oid chunk_relid = RelationGetRelid(state->result_relation_info->ri_RelationDesc);
 		Chunk *chunk = ts_chunk_get_by_relid(chunk_relid, true);
diff --git a/src/nodes/chunk_dispatch/chunk_insert_state.h b/src/nodes/chunk_dispatch/chunk_insert_state.h
@@ -98,6 +98,8 @@ typedef struct ChunkInsertState
 	/* for tracking compressed chunks */
 	bool chunk_compressed;
 	bool chunk_partial;
+	bool columnstore_insert;
+	bool needs_partial;
 
 	/* To speedup repeated calls of `decompress_batches_for_insert` */
 	CachedDecompressionState *cached_decompression_state;
diff --git a/tsl/test/expected/compressed_copy.out b/tsl/test/expected/compressed_copy.out
@@ -28,11 +28,9 @@ EXPLAIN (ANALYZE, COSTS OFF, SUMMARY OFF, TIMING OFF) SELECT * FROM metrics;
  Append (actual rows=3000 loops=1)
    ->  Custom Scan (ColumnarScan) on _hyper_1_3_chunk (actual rows=959 loops=1)
          ->  Seq Scan on compress_hyper_2_4_chunk (actual rows=1 loops=1)
-   ->  Seq Scan on _hyper_1_3_chunk (actual rows=0 loops=1)
    ->  Custom Scan (ColumnarScan) on _hyper_1_5_chunk (actual rows=2041 loops=1)
          ->  Seq Scan on compress_hyper_2_6_chunk (actual rows=3 loops=1)
-   ->  Seq Scan on _hyper_1_5_chunk (actual rows=0 loops=1)
-(7 rows)
+(5 rows)
 
 SELECT first(time,rn), last(time,rn) FROM (SELECT ROW_NUMBER() OVER () as rn, time FROM metrics) sub;
             first             |             last             
@@ -45,13 +43,11 @@ ROLLBACK;
 BEGIN;
 COPY metrics FROM PROGRAM 'seq 3000 | xargs -II date -d "2025-01-01 - I minute" +"%Y-%m-%d %H:%M:%S,d1,0.I"' WITH (FORMAT CSV);
 EXPLAIN (ANALYZE, COSTS OFF, SUMMARY OFF, TIMING OFF) SELECT * FROM metrics;
-                                   QUERY PLAN                                    
----------------------------------------------------------------------------------
- Append (actual rows=3000 loops=1)
-   ->  Custom Scan (ColumnarScan) on _hyper_1_7_chunk (actual rows=3000 loops=1)
-         ->  Seq Scan on compress_hyper_2_8_chunk (actual rows=3 loops=1)
-   ->  Seq Scan on _hyper_1_7_chunk (actual rows=0 loops=1)
-(4 rows)
+                                QUERY PLAN                                 
+---------------------------------------------------------------------------
+ Custom Scan (ColumnarScan) on _hyper_1_7_chunk (actual rows=3000 loops=1)
+   ->  Seq Scan on compress_hyper_2_8_chunk (actual rows=3 loops=1)
+(2 rows)
 
 SELECT first(time,rn), last(time,rn) FROM (SELECT ROW_NUMBER() OVER () as rn, time FROM metrics) sub;
             first             |             last             
@@ -70,11 +66,9 @@ EXPLAIN (ANALYZE, COSTS OFF, SUMMARY OFF, TIMING OFF) SELECT * FROM metrics;
  Append (actual rows=3000 loops=1)
    ->  Custom Scan (ColumnarScan) on _hyper_1_9_chunk (actual rows=959 loops=1)
          ->  Seq Scan on compress_hyper_2_10_chunk (actual rows=1 loops=1)
-   ->  Seq Scan on _hyper_1_9_chunk (actual rows=0 loops=1)
    ->  Custom Scan (ColumnarScan) on _hyper_1_11_chunk (actual rows=2041 loops=1)
          ->  Seq Scan on compress_hyper_2_12_chunk (actual rows=3 loops=1)
-   ->  Seq Scan on _hyper_1_11_chunk (actual rows=0 loops=1)
-(7 rows)
+(5 rows)
 
 SELECT first(time,rn), last(time,rn) FROM (SELECT ROW_NUMBER() OVER () as rn, time FROM metrics) sub;
             first             |             last             
@@ -87,13 +81,11 @@ ROLLBACK;
 BEGIN;
 COPY metrics FROM PROGRAM 'seq 3000 | xargs -II date -d "2025-01-01 - I minute" +"%Y-%m-%d %H:%M:%S,d1,0.I"' WITH (FORMAT CSV);
 EXPLAIN (ANALYZE, COSTS OFF, SUMMARY OFF, TIMING OFF) SELECT * FROM metrics;
-                                    QUERY PLAN                                    
-----------------------------------------------------------------------------------
- Append (actual rows=3000 loops=1)
-   ->  Custom Scan (ColumnarScan) on _hyper_1_13_chunk (actual rows=3000 loops=1)
-         ->  Seq Scan on compress_hyper_2_14_chunk (actual rows=3 loops=1)
-   ->  Seq Scan on _hyper_1_13_chunk (actual rows=0 loops=1)
-(4 rows)
+                                 QUERY PLAN                                 
+----------------------------------------------------------------------------
+ Custom Scan (ColumnarScan) on _hyper_1_13_chunk (actual rows=3000 loops=1)
+   ->  Seq Scan on compress_hyper_2_14_chunk (actual rows=3 loops=1)
+(2 rows)
 
 SELECT first(time,rn), last(time,rn) FROM (SELECT ROW_NUMBER() OVER () as rn, time FROM metrics) sub;
             first             |             last             
@@ -136,19 +128,17 @@ SET timescaledb.enable_direct_compress_copy = true;
 SET timescaledb.enable_direct_compress_copy_client_sorted = true;
 COPY metrics FROM PROGRAM 'seq 3000 | xargs -II date -d "2025-01-01 - I minute" +"%Y-%m-%d %H:%M:%S,d1,0.I"' WITH (FORMAT CSV);
 EXPLAIN (ANALYZE, COSTS OFF, SUMMARY OFF, TIMING OFF) SELECT * FROM metrics;
-                                    QUERY PLAN                                    
-----------------------------------------------------------------------------------
- Append (actual rows=3000 loops=1)
-   ->  Custom Scan (ColumnarScan) on _hyper_1_19_chunk (actual rows=3000 loops=1)
-         ->  Seq Scan on compress_hyper_2_20_chunk (actual rows=3 loops=1)
-   ->  Seq Scan on _hyper_1_19_chunk (actual rows=0 loops=1)
-(4 rows)
+                                 QUERY PLAN                                 
+----------------------------------------------------------------------------
+ Custom Scan (ColumnarScan) on _hyper_1_19_chunk (actual rows=3000 loops=1)
+   ->  Seq Scan on compress_hyper_2_20_chunk (actual rows=3 loops=1)
+(2 rows)
 
 -- status should be 9
 SELECT DISTINCT status FROM _timescaledb_catalog.chunk WHERE compressed_chunk_id IS NOT NULL;
  status 
 --------
-      9
+      1
 (1 row)
 
 ROLLBACK;
@@ -158,19 +148,17 @@ SET timescaledb.enable_direct_compress_copy = true;
 SET timescaledb.enable_direct_compress_copy_client_sorted = false;
 COPY metrics FROM PROGRAM 'seq 3000 | xargs -II date -d "2025-01-01 - I minute" +"%Y-%m-%d %H:%M:%S,d1,0.I"' WITH (FORMAT CSV);
 EXPLAIN (ANALYZE, COSTS OFF, SUMMARY OFF, TIMING OFF) SELECT * FROM metrics;
-                                    QUERY PLAN                                    
-----------------------------------------------------------------------------------
- Append (actual rows=3000 loops=1)
-   ->  Custom Scan (ColumnarScan) on _hyper_1_21_chunk (actual rows=3000 loops=1)
-         ->  Seq Scan on compress_hyper_2_22_chunk (actual rows=3 loops=1)
-   ->  Seq Scan on _hyper_1_21_chunk (actual rows=0 loops=1)
-(4 rows)
+                                 QUERY PLAN                                 
+----------------------------------------------------------------------------
+ Custom Scan (ColumnarScan) on _hyper_1_21_chunk (actual rows=3000 loops=1)
+   ->  Seq Scan on compress_hyper_2_22_chunk (actual rows=3 loops=1)
+(2 rows)
 
 -- status should be 11
 SELECT DISTINCT status FROM _timescaledb_catalog.chunk WHERE compressed_chunk_id IS NOT NULL;
  status 
 --------
-     11
+      3
 (1 row)
 
 ROLLBACK;
@@ -181,13 +169,11 @@ SET timescaledb.enable_direct_compress_copy = true;
 SET timescaledb.enable_direct_compress_copy_client_sorted = true;
 COPY metrics FROM PROGRAM 'seq 0 0.2 9.8 | sed -e ''s!.[0-9]$!!'' | xargs -II date -d "2025-01-01 - I minute" +"%Y-%m-%d %H:%M:%S,dI,0.I"' WITH (FORMAT CSV);
 EXPLAIN (ANALYZE, COSTS OFF, SUMMARY OFF, TIMING OFF) SELECT * FROM metrics;
-                                   QUERY PLAN                                   
---------------------------------------------------------------------------------
- Append (actual rows=50 loops=1)
-   ->  Custom Scan (ColumnarScan) on _hyper_1_23_chunk (actual rows=50 loops=1)
-         ->  Seq Scan on compress_hyper_2_24_chunk (actual rows=10 loops=1)
-   ->  Seq Scan on _hyper_1_23_chunk (actual rows=0 loops=1)
-(4 rows)
+                                QUERY PLAN                                
+--------------------------------------------------------------------------
+ Custom Scan (ColumnarScan) on _hyper_1_23_chunk (actual rows=50 loops=1)
+   ->  Seq Scan on compress_hyper_2_24_chunk (actual rows=10 loops=1)
+(2 rows)
 
 SELECT format('%I.%I',schema_name,table_name) AS "COMPRESSED_CHUNK" FROM _timescaledb_catalog.chunk where compressed_chunk_id IS NULL \gset
 -- should have 10 batches
@@ -215,3 +201,107 @@ SELECT DISTINCT status FROM _timescaledb_catalog.chunk WHERE compressed_chunk_id
 (0 rows)
 
 ROLLBACK;
+-- test chunk status handling
+CREATE TABLE metrics_status(time timestamptz) WITH (tsdb.hypertable,tsdb.partition_column='time');
+-- normal insert should result in chunk status 0
+INSERT INTO metrics_status SELECT '2025-01-01';
+SELECT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics_status') chunk;
+ chunk_status_text 
+-------------------
+ {}
+(1 row)
+
+BEGIN;
+-- compressed copy into uncompressed chunk should result in chunk status 11 (compressed,partial,unordered)
+SET timescaledb.enable_direct_compress_copy = true;
+SET timescaledb.enable_direct_compress_copy_client_sorted = false;
+COPY metrics_status FROM STDIN;
+SELECT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics_status') chunk;
+       chunk_status_text        
+--------------------------------
+ {COMPRESSED,UNORDERED,PARTIAL}
+(1 row)
+
+ROLLBACK;
+BEGIN;
+-- compressed sorted copy into uncompressed chunk should result in chunk status 9 (compressed,partial)
+SET timescaledb.enable_direct_compress_copy = true;
+SET timescaledb.enable_direct_compress_copy_client_sorted = true;
+COPY metrics_status FROM STDIN;
+SELECT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics_status') chunk;
+  chunk_status_text   
+----------------------
+ {COMPRESSED,PARTIAL}
+(1 row)
+
+ROLLBACK;
+TRUNCATE metrics_status;
+BEGIN;
+-- compressed copy into new chunk should result in chunk status 3 (compressed,unordered)
+SET timescaledb.enable_direct_compress_copy = true;
+SET timescaledb.enable_direct_compress_copy_client_sorted = false;
+COPY metrics_status FROM STDIN;
+SELECT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics_status') chunk;
+   chunk_status_text    
+------------------------
+ {COMPRESSED,UNORDERED}
+(1 row)
+
+ROLLBACK;
+BEGIN;
+-- compressed sorted copy into new chunk should result in chunk status 1 (compressed)
+SET timescaledb.enable_direct_compress_copy = true;
+SET timescaledb.enable_direct_compress_copy_client_sorted = true;
+COPY metrics_status FROM STDIN;
+SELECT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics_status') chunk;
+ chunk_status_text 
+-------------------
+ {COMPRESSED}
+(1 row)
+
+ROLLBACK;
+SET timescaledb.enable_direct_compress_copy = false;
+SET timescaledb.enable_direct_compress_copy_client_sorted = false;
+INSERT INTO metrics_status SELECT '2025-01-01';
+SELECT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics_status') chunk;
+ chunk_status_text 
+-------------------
+ {}
+(1 row)
+
+SELECT compress_chunk(show_chunks('metrics_status'));
+             compress_chunk              
+-----------------------------------------
+ _timescaledb_internal._hyper_3_33_chunk
+(1 row)
+
+SELECT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics_status') chunk;
+ chunk_status_text 
+-------------------
+ {COMPRESSED}
+(1 row)
+
+BEGIN;
+-- compressed copy into fully compressed chunk should result in chunk status 3 (compressed,unordered)
+SET timescaledb.enable_direct_compress_copy = true;
+SET timescaledb.enable_direct_compress_copy_client_sorted = false;
+COPY metrics_status FROM STDIN;
+SELECT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics_status') chunk;
+   chunk_status_text    
+------------------------
+ {COMPRESSED,UNORDERED}
+(1 row)
+
+ROLLBACK;
+BEGIN;
+-- compressed copy new chunk should result in chunk status 1 (compressed)
+SET timescaledb.enable_direct_compress_copy = true;
+SET timescaledb.enable_direct_compress_copy_client_sorted = true;
+COPY metrics_status FROM STDIN;
+SELECT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics_status') chunk;
+ chunk_status_text 
+-------------------
+ {COMPRESSED}
+(1 row)
+
+ROLLBACK;
diff --git a/tsl/test/sql/compressed_copy.sql b/tsl/test/sql/compressed_copy.sql

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1 @@`
	`1`	`+Implements: #8529 Optimize direct compress status handling`
Original file line number	Diff line number	Diff line change
`@@ -289,6 +289,7 @@ TSCopyMultiInsertBufferInit(TSCopyMultiInsertInfo miinfo, ChunkInsertState cis`
`289`	`289`	`if (!ts_chunk_is_unordered(chunk))`
`290`	`290`	`ts_chunk_set_unordered(chunk);`
`291`	`291`	`}`
	`292`	`+ cis->columnstore_insert = true;`
`292`	`293`	`break;`
`293`	`294`	`}`
`294`	`295`	`}`