Account for uncompressed rows in 'create_compressed_chunk'

`_timescaledb_internal.create_compressed_chunk` can be used to create a
compressed chunk with existing compressed data. It did not account for
the fact that the chunk can contain uncompressed data, in which case the
chunk status must be set to partial.

Fixes #5946

.unreleased/bugfix_5951

@@ -0,0 +1,4 @@
+Implements: #5951 _timescaledb_internal.create_compressed_chunk doesn't account for existing uncompressed rows
+
+Fixes: #5946
+

tsl/src/compression/api.c

@@ -770,6 +770,7 @@ tsl_create_compressed_chunk(PG_FUNCTION_ARGS)
 	Chunk *compress_ht_chunk;
 	Cache *hcache;
 	CompressChunkCxt cxt;
+	bool chunk_was_compressed;
 
 	Assert(!PG_ARGISNULL(0));
 	Assert(!PG_ARGISNULL(1));
@@ -812,7 +813,16 @@ tsl_create_compressed_chunk(PG_FUNCTION_ARGS)
 										  numrows_pre_compression,
 										  numrows_post_compression);
 
+	chunk_was_compressed = ts_chunk_is_compressed(cxt.srcht_chunk);
 	ts_chunk_set_compressed_chunk(cxt.srcht_chunk, compress_ht_chunk->fd.id);
+	if (!chunk_was_compressed)
+	{
+		/* The chunk was not compressed, before it had the compressed chunk
+		 * attached to it, so it must set it to be partial.
+		 * TODO: actually, we should check if the chunk contains uncompressed rows
+		 */
+		ts_chunk_set_partial(cxt.srcht_chunk);
+	}
 	ts_cache_release(hcache);
 
 	PG_RETURN_OID(chunk_relid);

tsl/test/expected/compression_create_compressed_table.out

@@ -0,0 +1,70 @@
+-- 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.
+
+\c :TEST_DBNAME :ROLE_SUPERUSER
+-- create compressed hypertable
+CREATE TABLE "public"."metrics" (
+    "time" timestamp with time zone NOT NULL,
+    "device_id" "text",
+    "val" double precision
+);
+SELECT create_hypertable('public.metrics', 'time');
+  create_hypertable   
+----------------------
+ (1,public,metrics,t)
+(1 row)
+
+ALTER TABLE metrics SET (timescaledb.compress, timescaledb.compress_orderby = 'time', timescaledb.compress_segmentby = 'device_id');
+-- insert uncompressed row into hypertable
+INSERT INTO metrics (time, device_id, val)
+VALUES('2023-05-01T00:00:00Z'::timestamptz, 1, 1.0);
+SELECT count(*) FROM _timescaledb_internal._hyper_1_1_chunk;
+ count 
+-------
+     1
+(1 row)
+
+-- compress these rows, we do this to get compressed row data for the test
+SELECT compress_chunk('_timescaledb_internal._hyper_1_1_chunk');
+             compress_chunk             
+----------------------------------------
+ _timescaledb_internal._hyper_1_1_chunk
+(1 row)
+
+-- create custom compressed chunk table
+CREATE TABLE "_timescaledb_internal"."custom_compressed_chunk"() INHERITS ("_timescaledb_internal"."_compressed_hypertable_2");
+-- copy compressed row from compressed table into custom compressed chunk table
+INSERT INTO "_timescaledb_internal"."custom_compressed_chunk" SELECT * FROM "_timescaledb_internal"."compress_hyper_2_2_chunk";
+-- decompress the rows, moving them back to uncompressed space
+SELECT decompress_chunk('"_timescaledb_internal"."_hyper_1_1_chunk"');
+            decompress_chunk            
+----------------------------------------
+ _timescaledb_internal._hyper_1_1_chunk
+(1 row)
+
+-- attach compressed chunk to parent chunk
+SELECT _timescaledb_internal.create_compressed_chunk(
+    '"_timescaledb_internal"."_hyper_1_1_chunk"'::TEXT::REGCLASS,
+    '"_timescaledb_internal"."custom_compressed_chunk"'::TEXT::REGCLASS,
+    8192,
+    8192,
+    16384,
+    8192,
+    8192,
+    16384,
+    1,
+    1
+);
+        create_compressed_chunk         
+----------------------------------------
+ _timescaledb_internal._hyper_1_1_chunk
+(1 row)
+
+-- select total rows in chunk (should be 2)
+SELECT count(*) FROM "_timescaledb_internal"."_hyper_1_1_chunk";
+ count 
+-------
+     2
+(1 row)
+

tsl/test/sql/CMakeLists.txt

@@ -13,6 +13,7 @@ set(TEST_FILES
     cagg_refresh.sql
     cagg_watermark.sql
     compressed_collation.sql
+    compression_create_compressed_table.sql
     compression_bgw.sql
     compression_conflicts.sql
     compression_insert.sql

tsl/test/sql/compression_create_compressed_table.sql

@@ -0,0 +1,50 @@
+-- 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.
+
+\c :TEST_DBNAME :ROLE_SUPERUSER
+
+-- create compressed hypertable
+CREATE TABLE "public"."metrics" (
+    "time" timestamp with time zone NOT NULL,
+    "device_id" "text",
+    "val" double precision
+);
+SELECT create_hypertable('public.metrics', 'time');
+ALTER TABLE metrics SET (timescaledb.compress, timescaledb.compress_orderby = 'time', timescaledb.compress_segmentby = 'device_id');
+
+-- insert uncompressed row into hypertable
+INSERT INTO metrics (time, device_id, val)
+VALUES('2023-05-01T00:00:00Z'::timestamptz, 1, 1.0);
+
+SELECT count(*) FROM _timescaledb_internal._hyper_1_1_chunk;
+
+-- compress these rows, we do this to get compressed row data for the test
+SELECT compress_chunk('_timescaledb_internal._hyper_1_1_chunk');
+
+-- create custom compressed chunk table
+CREATE TABLE "_timescaledb_internal"."custom_compressed_chunk"() INHERITS ("_timescaledb_internal"."_compressed_hypertable_2");
+
+-- copy compressed row from compressed table into custom compressed chunk table
+INSERT INTO "_timescaledb_internal"."custom_compressed_chunk" SELECT * FROM "_timescaledb_internal"."compress_hyper_2_2_chunk";
+
+-- decompress the rows, moving them back to uncompressed space
+SELECT decompress_chunk('"_timescaledb_internal"."_hyper_1_1_chunk"');
+
+-- attach compressed chunk to parent chunk
+SELECT _timescaledb_internal.create_compressed_chunk(
+    '"_timescaledb_internal"."_hyper_1_1_chunk"'::TEXT::REGCLASS,
+    '"_timescaledb_internal"."custom_compressed_chunk"'::TEXT::REGCLASS,
+    8192,
+    8192,
+    16384,
+    8192,
+    8192,
+    16384,
+    1,
+    1
+);
+
+-- select total rows in chunk (should be 2)
+SELECT count(*) FROM "_timescaledb_internal"."_hyper_1_1_chunk";
+