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(*form));
+		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

@@ -66,6 +66,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
@@ -216,6 +218,7 @@ typedef struct RelationMergeInfo
 {
 	Oid relid;
 	struct VacuumCutoffs cutoffs;
+	FormData_compression_chunk_size ccs;
 	Chunk *chunk;
 	Relation rel;
 	char relpersistence;
@@ -827,6 +830,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);
 
@@ -843,6 +849,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);
@@ -853,6 +879,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;
 }
 
@@ -1048,6 +1090,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(WARNING,
+					 "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,7 +640,65 @@ 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)
+
+--
+-- Check that compression_chunk_size stats are also merged when we
+-- merge compressed chunks.
+--
+-- Use a view to compare merged stats against the total sum of that
+-- stats for all chunks. There are only two compressed chunks, 1 and
+-- 2. Show each chunks stats as the fraction of the total size. This
+-- is to make the test work across different architectures that show
+-- slightly different absolute disk sizes.
+---
+select
+    sum(ccs.uncompressed_heap_size) as total_uncompressed_heap_size,
+    sum(ccs.uncompressed_toast_size) as total_uncompressed_toast_size,
+    sum(ccs.uncompressed_index_size) as total_uncompressed_index_size,
+    sum(ccs.compressed_heap_size) as total_compressed_heap_size,
+    sum(ccs.compressed_toast_size) as total_compressed_toast_size,
+    sum(ccs.compressed_index_size) as total_compressed_index_size,
+    sum(ccs.numrows_pre_compression) as total_numrows_pre_compression,
+    sum(ccs.numrows_post_compression) as total_numrows_post_compression,
+    sum(ccs.numrows_frozen_immediately) as total_numrows_frozen_immediately
+from _timescaledb_catalog.compression_chunk_size ccs \gset
+-- View to show current chunk compression size stats as a fraction of
+-- the totals.
+create view compression_size_fraction as
+select
+    ccs.chunk_id,
+    ccs.compressed_chunk_id,
+    round(ccs.uncompressed_heap_size::numeric / :total_uncompressed_heap_size, 1) as uncompressed_heap_size_fraction,
+    ccs.uncompressed_toast_size::numeric as uncompressed_toast_size_fraction,
+    round(ccs.uncompressed_index_size::numeric / :total_uncompressed_index_size, 1) as uncompressed_index_size_fraction,
+    round(ccs.compressed_heap_size::numeric / :total_compressed_heap_size, 1) as compressed_heap_size_fraction,
+    round(ccs.compressed_toast_size::numeric / :total_compressed_toast_size, 1) as compressed_toast_size_fraction,
+    round(ccs.compressed_index_size::numeric / :total_compressed_index_size, 1) as compressed_index_size_fraction,
+    round(ccs.numrows_pre_compression ::numeric/ :total_numrows_pre_compression, 1) as numrows_pre_compression_fraction,
+    round(ccs.numrows_post_compression::numeric / :total_numrows_post_compression, 1) as numrows_post_compression_fraction,
+    round(ccs.numrows_frozen_immediately::numeric / :total_numrows_frozen_immediately, 1) as numrows_frozen_immediately_fraction
+from _timescaledb_catalog.compression_chunk_size ccs
+order by chunk_id;
+--
+-- Merge all chunks until only 1 remains.  Also check that metadata is
+-- merged.
+---
+select * from compression_size_fraction;
+ chunk_id | compressed_chunk_id | uncompressed_heap_size_fraction | uncompressed_toast_size_fraction | uncompressed_index_size_fraction | compressed_heap_size_fraction | compressed_toast_size_fraction | compressed_index_size_fraction | numrows_pre_compression_fraction | numrows_post_compression_fraction | numrows_frozen_immediately_fraction 
+----------+---------------------+---------------------------------+----------------------------------+----------------------------------+-------------------------------+--------------------------------+--------------------------------+----------------------------------+-----------------------------------+-------------------------------------
+        1 |                  17 |                             0.5 |                                0 |                              0.5 |                           0.5 |                            0.5 |                            0.5 |                              0.5 |                               0.5 |                                 0.5
+        2 |                  18 |                             0.5 |                                0 |                              0.5 |                           0.5 |                            0.5 |                            0.5 |                              0.5 |                               0.5 |                                 0.5
+(2 rows)
+
 select count(*), sum(device), round(sum(temp)::numeric, 4) from mergeme;
  count  |   sum   |     round     
 --------+---------+---------------
@@ -689,6 +760,15 @@ 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']);
+-- Final merge, involving the two compressed chunks 1 and 2. The stats
+-- should also be merged.
+select * from compression_size_fraction;
+ chunk_id | compressed_chunk_id | uncompressed_heap_size_fraction | uncompressed_toast_size_fraction | uncompressed_index_size_fraction | compressed_heap_size_fraction | compressed_toast_size_fraction | compressed_index_size_fraction | numrows_pre_compression_fraction | numrows_post_compression_fraction | numrows_frozen_immediately_fraction 
+----------+---------------------+---------------------------------+----------------------------------+----------------------------------+-------------------------------+--------------------------------+--------------------------------+----------------------------------+-----------------------------------+-------------------------------------
+        1 |                  17 |                             0.5 |                                0 |                              0.5 |                           0.5 |                            0.5 |                            0.5 |                              0.5 |                               0.5 |                                 0.5
+        2 |                  18 |                             0.5 |                                0 |                              0.5 |                           0.5 |                            0.5 |                            0.5 |                              0.5 |                               0.5 |                                 0.5
+(2 rows)
+
 select count(*), sum(device), round(sum(temp)::numeric, 4) from mergeme;
  count  |   sum   |     round     
 --------+---------+---------------
@@ -707,6 +787,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 compression_size_fraction;
+ chunk_id | compressed_chunk_id | uncompressed_heap_size_fraction | uncompressed_toast_size_fraction | uncompressed_index_size_fraction | compressed_heap_size_fraction | compressed_toast_size_fraction | compressed_index_size_fraction | numrows_pre_compression_fraction | numrows_post_compression_fraction | numrows_frozen_immediately_fraction 
+----------+---------------------+---------------------------------+----------------------------------+----------------------------------+-------------------------------+--------------------------------+--------------------------------+----------------------------------+-----------------------------------+-------------------------------------
+        1 |                  17 |                             1.0 |                                0 |                              1.0 |                           1.0 |                            1.0 |                            1.0 |                              1.0 |                               1.0 |                                 1.0
+(1 row)
+
 select count(*), sum(device), round(sum(temp)::numeric, 4) from mergeme;
  count  |   sum   |     round     
 --------+---------+---------------