Update compression stats when merging chunks

Merge compression chunk size stats when merging chunks. The merged
chunk's stats is simply the sum of the stats of all compressed chunks
that are merged. This ensures that the aggregate stats do not change
due to the merge. Still, the stats might not accurately represent the
merged chunk since merging both compressed and non-compressed chunks
will leave some data uncompressed and this data won't be reflected in
the stats of the merged chunk.

Author: erimatnor

.unreleased/pg_7909

@@ -0,0 +1 @@
+Fixes: #7909 Update compression stats when merging chunks

src/ts_catalog/compression_chunk_size.c

@@ -4,7 +4,10 @@
  * LICENSE-APACHE for a copy of the license.
  */
 #include <postgres.h>
+#include <access/htup_details.h>
+#include <executor/tuptable.h>
 
+#include "export.h"
 #include "scan_iterator.h"
 #include "scanner.h"
 #include "ts_catalog/catalog.h"
@@ -43,3 +46,75 @@ ts_compression_chunk_size_delete(int32 uncompressed_chunk_id)
 
 	return count;
 }
+
+TSDLLEXPORT bool
+ts_compression_chunk_size_get(int32 chunk_id, Form_compression_chunk_size form)
+{
+	ScanIterator iterator =
+		ts_scan_iterator_create(COMPRESSION_CHUNK_SIZE, AccessExclusiveLock, CurrentMemoryContext);
+	bool found = false;
+
+	Assert(form != NULL);
+
+	init_scan_by_uncompressed_chunk_id(&iterator, chunk_id);
+	ts_scanner_foreach(&iterator)
+	{
+		TupleInfo *ti = ts_scan_iterator_tuple_info(&iterator);
+		bool should_free;
+		HeapTuple tuple = ts_scanner_fetch_heap_tuple(ti, false, &should_free);
+		memcpy(form, GETSTRUCT(tuple), sizeof(FormData_compression_chunk_size));
+		found = true;
+		Assert(form->chunk_id == chunk_id);
+
+		if (should_free)
+			heap_freetuple(tuple);
+
+		break;
+	}
+
+	ts_scan_iterator_close(&iterator);
+
+	return found;
+}
+
+TSDLLEXPORT bool
+ts_compression_chunk_size_update(int32 chunk_id, Form_compression_chunk_size form)
+{
+	ScanIterator iterator =
+		ts_scan_iterator_create(COMPRESSION_CHUNK_SIZE, RowExclusiveLock, CurrentMemoryContext);
+	bool found = false;
+	CatalogSecurityContext sec_ctx;
+
+	Assert(form != NULL);
+
+	init_scan_by_uncompressed_chunk_id(&iterator, chunk_id);
+	ts_scanner_foreach(&iterator)
+	{
+		TupleInfo *ti = ts_scan_iterator_tuple_info(&iterator);
+		bool should_free;
+		HeapTuple tuple = ts_scanner_fetch_heap_tuple(ti, false, &should_free);
+		HeapTuple copy = heap_copytuple(tuple);
+		Form_compression_chunk_size tupform = (Form_compression_chunk_size) GETSTRUCT(copy);
+
+		/* Don't update chunk IDs so copy from existing tuple */
+		form->chunk_id = tupform->chunk_id;
+		form->compressed_chunk_id = tupform->compressed_chunk_id;
+
+		memcpy(tupform, form, sizeof(FormData_compression_chunk_size));
+		ts_catalog_database_info_become_owner(ts_catalog_database_info_get(), &sec_ctx);
+		ts_catalog_update_tid_only(ti->scanrel, ts_scanner_get_tuple_tid(ti), copy);
+		ts_catalog_restore_user(&sec_ctx);
+		found = true;
+
+		heap_freetuple(copy);
+
+		if (should_free)
+			heap_freetuple(tuple);
+
+		break;
+	}
+
+	ts_scan_iterator_close(&iterator);
+
+	return found;
+}

src/ts_catalog/compression_chunk_size.h

@@ -8,4 +8,10 @@
 #include <compat/compat.h>
 #include <postgres.h>
 
+#include <ts_catalog/catalog.h>
+
 extern TSDLLEXPORT int ts_compression_chunk_size_delete(int32 uncompressed_chunk_id);
+extern TSDLLEXPORT bool ts_compression_chunk_size_get(int32 chunk_id,
+													  Form_compression_chunk_size form);
+extern TSDLLEXPORT bool ts_compression_chunk_size_update(int32 chunk_id,
+														 Form_compression_chunk_size form);

tsl/src/chunk.c

@@ -62,6 +62,8 @@
 #include "hypercube.h"
 #include "hypertable.h"
 #include "hypertable_cache.h"
+#include "ts_catalog/catalog.h"
+#include "ts_catalog/compression_chunk_size.h"
 #include "utils.h"
 
 /* Data in a frozen chunk cannot be modified. So any operation
@@ -212,6 +214,7 @@ typedef struct RelationMergeInfo
 {
 	Oid relid;
 	struct VacuumCutoffs cutoffs;
+	FormData_compression_chunk_size ccs;
 	Chunk *chunk;
 	Relation rel;
 	char relpersistence;
@@ -823,6 +826,9 @@ merge_relinfos(RelationMergeInfo *relinfos, int nrelids, int mergeindex)
 								  ExclusiveLock);
 	Relation new_rel = table_open(new_relid, AccessExclusiveLock);
 	double total_num_tuples = 0.0;
+	FormData_compression_chunk_size merged_ccs;
+
+	memset(&merged_ccs, 0, sizeof(FormData_compression_chunk_size));
 
 	pg17_workaround_init(new_rel, relinfos, nrelids);
 
@@ -839,6 +845,26 @@ merge_relinfos(RelationMergeInfo *relinfos, int nrelids, int mergeindex)
 			total_num_tuples += num_tuples;
 			relinfo->rel = NULL;
 		}
+
+		/*
+		 * Merge compression chunk size stats.
+		 *
+		 * Simply sum up the stats for all compressed relations that are
+		 * merged. Note that we don't add anything for non-compressed
+		 * relations that are merged because they don't have stats. This is a
+		 * bit weird because the data from uncompressed relations will not be
+		 * reflected in the stats of the merged chunk although the data is
+		 * part of the chunk.
+		 */
+		merged_ccs.compressed_heap_size += relinfo->ccs.compressed_heap_size;
+		merged_ccs.compressed_toast_size += relinfo->ccs.compressed_toast_size;
+		merged_ccs.compressed_index_size += relinfo->ccs.compressed_index_size;
+		merged_ccs.uncompressed_heap_size += relinfo->ccs.uncompressed_heap_size;
+		merged_ccs.uncompressed_toast_size += relinfo->ccs.uncompressed_toast_size;
+		merged_ccs.uncompressed_index_size += relinfo->ccs.uncompressed_index_size;
+		merged_ccs.numrows_post_compression += relinfo->ccs.numrows_post_compression;
+		merged_ccs.numrows_pre_compression += relinfo->ccs.numrows_pre_compression;
+		merged_ccs.numrows_frozen_immediately += relinfo->ccs.numrows_frozen_immediately;
 	}
 
 	pg17_workaround_cleanup(new_rel);
@@ -849,6 +875,22 @@ merge_relinfos(RelationMergeInfo *relinfos, int nrelids, int mergeindex)
 	table_close(new_rel, NoLock);
 	table_close(relRelation, RowExclusiveLock);
 
+	/*
+	 * Update compression chunk size stats, but only if at least one of the
+	 * merged chunks was compressed. In that case the merged metadata should
+	 * be non-zero.
+	 */
+	if (merged_ccs.compressed_heap_size > 0)
+	{
+		/*
+		 * The result relation should always be compressed because we pick the
+		 * first compressed one, if one exists.
+		 */
+
+		Assert(result_minfo->ccs.compressed_heap_size > 0);
+		ts_compression_chunk_size_update(result_minfo->chunk->fd.id, &merged_ccs);
+	}
+
 	return new_relid;
 }
 
@@ -1044,6 +1086,14 @@ chunk_merge_chunks(PG_FUNCTION_ARGS)
 
 			if (mergeindex == -1)
 				mergeindex = i;
+
+			/* Read compression chunk size stats */
+			bool found = ts_compression_chunk_size_get(chunk->fd.id, &relinfo->ccs);
+
+			if (!found)
+				elog(NOTICE,
+					 "missing compression chunk size stats for compressed chunk \"%s\"",
+					 NameStr(chunk->fd.table_name));
 		}
 
 		if (ts_chunk_is_frozen(chunk))

tsl/test/expected/merge_chunks.out

@@ -283,7 +283,20 @@ select * from chunk_info;
  _hyper_1_5_chunk | heap | (("time" >= 'Wed Jan 03 16:00:00 2024 PST'::timestamp with time zone) AND ("time" < 'Thu Jan 04 16:00:00 2024 PST'::timestamp with time zone))
 (10 rows)
 
+select * from _timescaledb_catalog.compression_chunk_size order by chunk_id;
+ chunk_id | compressed_chunk_id | uncompressed_heap_size | uncompressed_toast_size | uncompressed_index_size | compressed_heap_size | compressed_toast_size | compressed_index_size | numrows_pre_compression | numrows_post_compression | numrows_frozen_immediately 
+----------+---------------------+------------------------+-------------------------+-------------------------+----------------------+-----------------------+-----------------------+-------------------------+--------------------------+----------------------------
+        1 |                   6 |                   8192 |                       0 |                   32768 |                16384 |                  8192 |                 16384 |                       1 |                        1 |                          1
+        3 |                   7 |                   8192 |                       0 |                   32768 |                16384 |                  8192 |                 16384 |                       1 |                        1 |                          1
+(2 rows)
+
 call merge_chunks('{_timescaledb_internal._hyper_1_1_chunk, _timescaledb_internal._hyper_1_2_chunk, _timescaledb_internal._hyper_1_3_chunk}');
+select * from _timescaledb_catalog.compression_chunk_size order by chunk_id;
+ chunk_id | compressed_chunk_id | uncompressed_heap_size | uncompressed_toast_size | uncompressed_index_size | compressed_heap_size | compressed_toast_size | compressed_index_size | numrows_pre_compression | numrows_post_compression | numrows_frozen_immediately 
+----------+---------------------+------------------------+-------------------------+-------------------------+----------------------+-----------------------+-----------------------+-------------------------+--------------------------+----------------------------
+        1 |                   6 |                  16384 |                       0 |                   65536 |                32768 |                 16384 |                 32768 |                       2 |                        2 |                          2
+(1 row)
+
 select * from chunk_info;
       chunk       | tam  |                                                                checkconstraint                                                                 
 ------------------+------+------------------------------------------------------------------------------------------------------------------------------------------------
@@ -627,14 +640,41 @@ select * from partitions;
  _hyper_1_14_chunk | device      |           1431655764 | 9223372036854775807
 (24 rows)
 
--- Merge all chunks until only 1 remains
+-- Show which chunks are compressed. Their compression_chunk_size
+-- metadata should be merged.
+select chunk_name from timescaledb_information.chunks
+where is_compressed=true order by chunk_name;
+    chunk_name    
+------------------
+ _hyper_1_1_chunk
+ _hyper_1_2_chunk
+(2 rows)
+
+--
+-- Merge all chunks until only 1 remains.  Also check that metadata is
+-- merged.
+---
+select * from _timescaledb_catalog.compression_chunk_size order by chunk_id;
+ chunk_id | compressed_chunk_id | uncompressed_heap_size | uncompressed_toast_size | uncompressed_index_size | compressed_heap_size | compressed_toast_size | compressed_index_size | numrows_pre_compression | numrows_post_compression | numrows_frozen_immediately 
+----------+---------------------+------------------------+-------------------------+-------------------------+----------------------+-----------------------+-----------------------+-------------------------+--------------------------+----------------------------
+        1 |                  17 |                2441216 |                       0 |                 4489216 |                16384 |                688128 |                 16384 |                   46245 |                       48 |                         48
+        2 |                  18 |                2416640 |                       0 |                 4456448 |                16384 |                671744 |                 16384 |                   45832 |                       48 |                         48
+(2 rows)
+
 select count(*), sum(device), round(sum(temp)::numeric, 4) from mergeme;
  count  |   sum   |     round     
 --------+---------+---------------
  518406 | 2854401 | 10373952.7510
 (1 row)
 
 call merge_chunks(ARRAY['_timescaledb_internal._hyper_1_1_chunk', '_timescaledb_internal._hyper_1_4_chunk','_timescaledb_internal._hyper_1_5_chunk', '_timescaledb_internal._hyper_1_12_chunk']);
+select * from _timescaledb_catalog.compression_chunk_size order by chunk_id;
+ chunk_id | compressed_chunk_id | uncompressed_heap_size | uncompressed_toast_size | uncompressed_index_size | compressed_heap_size | compressed_toast_size | compressed_index_size | numrows_pre_compression | numrows_post_compression | numrows_frozen_immediately 
+----------+---------------------+------------------------+-------------------------+-------------------------+----------------------+-----------------------+-----------------------+-------------------------+--------------------------+----------------------------
+        1 |                  17 |                2441216 |                       0 |                 4489216 |                16384 |                688128 |                 16384 |                   46245 |                       48 |                         48
+        2 |                  18 |                2416640 |                       0 |                 4456448 |                16384 |                671744 |                 16384 |                   45832 |                       48 |                         48
+(2 rows)
+
 select count(*), sum(device), round(sum(temp)::numeric, 4) from mergeme;
  count  |   sum   |     round     
 --------+---------+---------------
@@ -665,6 +705,13 @@ select * from partitions;
 (18 rows)
 
 call merge_chunks(ARRAY['_timescaledb_internal._hyper_1_2_chunk', '_timescaledb_internal._hyper_1_10_chunk','_timescaledb_internal._hyper_1_13_chunk', '_timescaledb_internal._hyper_1_15_chunk']);
+select * from _timescaledb_catalog.compression_chunk_size order by chunk_id;
+ chunk_id | compressed_chunk_id | uncompressed_heap_size | uncompressed_toast_size | uncompressed_index_size | compressed_heap_size | compressed_toast_size | compressed_index_size | numrows_pre_compression | numrows_post_compression | numrows_frozen_immediately 
+----------+---------------------+------------------------+-------------------------+-------------------------+----------------------+-----------------------+-----------------------+-------------------------+--------------------------+----------------------------
+        1 |                  17 |                2441216 |                       0 |                 4489216 |                16384 |                688128 |                 16384 |                   46245 |                       48 |                         48
+        2 |                  18 |                2416640 |                       0 |                 4456448 |                16384 |                671744 |                 16384 |                   45832 |                       48 |                         48
+(2 rows)
+
 select count(*), sum(device), round(sum(temp)::numeric, 4) from mergeme;
  count  |   sum   |     round     
 --------+---------+---------------
@@ -689,6 +736,13 @@ select * from partitions;
 (12 rows)
 
 call merge_chunks(ARRAY['_timescaledb_internal._hyper_1_3_chunk', '_timescaledb_internal._hyper_1_11_chunk','_timescaledb_internal._hyper_1_14_chunk', '_timescaledb_internal._hyper_1_16_chunk']);
+select * from _timescaledb_catalog.compression_chunk_size order by chunk_id;
+ chunk_id | compressed_chunk_id | uncompressed_heap_size | uncompressed_toast_size | uncompressed_index_size | compressed_heap_size | compressed_toast_size | compressed_index_size | numrows_pre_compression | numrows_post_compression | numrows_frozen_immediately 
+----------+---------------------+------------------------+-------------------------+-------------------------+----------------------+-----------------------+-----------------------+-------------------------+--------------------------+----------------------------
+        1 |                  17 |                2441216 |                       0 |                 4489216 |                16384 |                688128 |                 16384 |                   46245 |                       48 |                         48
+        2 |                  18 |                2416640 |                       0 |                 4456448 |                16384 |                671744 |                 16384 |                   45832 |                       48 |                         48
+(2 rows)
+
 select count(*), sum(device), round(sum(temp)::numeric, 4) from mergeme;
  count  |   sum   |     round     
 --------+---------+---------------
@@ -707,6 +761,12 @@ select * from partitions;
 (6 rows)
 
 call merge_chunks(ARRAY['_timescaledb_internal._hyper_1_3_chunk', '_timescaledb_internal._hyper_1_1_chunk','_timescaledb_internal._hyper_1_2_chunk']);
+select * from _timescaledb_catalog.compression_chunk_size order by chunk_id;
+ chunk_id | compressed_chunk_id | uncompressed_heap_size | uncompressed_toast_size | uncompressed_index_size | compressed_heap_size | compressed_toast_size | compressed_index_size | numrows_pre_compression | numrows_post_compression | numrows_frozen_immediately 
+----------+---------------------+------------------------+-------------------------+-------------------------+----------------------+-----------------------+-----------------------+-------------------------+--------------------------+----------------------------
+        1 |                  17 |                4857856 |                       0 |                 8945664 |                32768 |               1359872 |                 32768 |                   92077 |                       96 |                         96
+(1 row)
+
 select count(*), sum(device), round(sum(temp)::numeric, 4) from mergeme;
  count  |   sum   |     round     
 --------+---------+---------------

tsl/test/sql/merge_chunks.sql

@@ -167,7 +167,11 @@ select compress_chunk('_timescaledb_internal._hyper_1_3_chunk');
 -- Test merging compressed chunks
 begin;
 select * from chunk_info;
+
+select * from _timescaledb_catalog.compression_chunk_size order by chunk_id;
+
 call merge_chunks('{_timescaledb_internal._hyper_1_1_chunk, _timescaledb_internal._hyper_1_2_chunk, _timescaledb_internal._hyper_1_3_chunk}');
+select * from _timescaledb_catalog.compression_chunk_size order by chunk_id;
 select * from chunk_info;
 select count(*) as num_orphaned_slices from orphaned_slices;
 select * from mergeme;
@@ -291,18 +295,30 @@ alter table _timescaledb_internal._hyper_1_2_chunk set access method hypercore;
 -- Show partitions before merge
 select * from partitions;
 
--- Merge all chunks until only 1 remains
+-- Show which chunks are compressed. Their compression_chunk_size
+-- metadata should be merged.
+select chunk_name from timescaledb_information.chunks
+where is_compressed=true order by chunk_name;
+--
+-- Merge all chunks until only 1 remains.  Also check that metadata is
+-- merged.
+---
+select * from _timescaledb_catalog.compression_chunk_size order by chunk_id;
 select count(*), sum(device), round(sum(temp)::numeric, 4) from mergeme;
 call merge_chunks(ARRAY['_timescaledb_internal._hyper_1_1_chunk', '_timescaledb_internal._hyper_1_4_chunk','_timescaledb_internal._hyper_1_5_chunk', '_timescaledb_internal._hyper_1_12_chunk']);
+select * from _timescaledb_catalog.compression_chunk_size order by chunk_id;
 select count(*), sum(device), round(sum(temp)::numeric, 4) from mergeme;
 select * from partitions;
 call merge_chunks(ARRAY['_timescaledb_internal._hyper_1_2_chunk', '_timescaledb_internal._hyper_1_10_chunk','_timescaledb_internal._hyper_1_13_chunk', '_timescaledb_internal._hyper_1_15_chunk']);
+select * from _timescaledb_catalog.compression_chunk_size order by chunk_id;
 select count(*), sum(device), round(sum(temp)::numeric, 4) from mergeme;
 select * from partitions;
 call merge_chunks(ARRAY['_timescaledb_internal._hyper_1_3_chunk', '_timescaledb_internal._hyper_1_11_chunk','_timescaledb_internal._hyper_1_14_chunk', '_timescaledb_internal._hyper_1_16_chunk']);
+select * from _timescaledb_catalog.compression_chunk_size order by chunk_id;
 select count(*), sum(device), round(sum(temp)::numeric, 4) from mergeme;
 select * from partitions;
 call merge_chunks(ARRAY['_timescaledb_internal._hyper_1_3_chunk', '_timescaledb_internal._hyper_1_1_chunk','_timescaledb_internal._hyper_1_2_chunk']);
+select * from _timescaledb_catalog.compression_chunk_size order by chunk_id;
 select count(*), sum(device), round(sum(temp)::numeric, 4) from mergeme;
 select * from partitions;
 select * from chunk_info;