Fix uncompressed size estimate for text segmentby columns

The size estimate built a query that called compressed_data_column_size
on every variable length column. Variable length segmentby columns
were handled incorrectly and need to use pg_column_size instead of
compressed_data_column_size.

(cherry picked from commit ddae6bf)

Author: svenklemm

.unreleased/pr_10073

@@ -0,0 +1 @@
+Fixes: #10073 Fix uncompressed size estimate for varlen

sql/size_utils.sql

@@ -597,6 +597,7 @@ AS $$
 DECLARE
   v_compressed_chunk regclass;
   v_uncompressed_chunk regclass;
+  v_segmentby text[];
   v_index regclass;
   v_fixed_column_size integer;
   v_num_varlen_columns integer;
@@ -612,7 +613,7 @@ BEGIN
 
   v_compressed_chunk := $1;
 
-  SELECT relid INTO v_uncompressed_chunk FROM _timescaledb_catalog.compression_settings WHERE compress_relid = v_compressed_chunk;
+  SELECT relid, segmentby INTO v_uncompressed_chunk, v_segmentby FROM _timescaledb_catalog.compression_settings WHERE compress_relid = v_compressed_chunk;
   IF NOT FOUND THEN
     RETURN;
   END IF;
@@ -632,7 +633,11 @@ BEGIN
   v_tuple_data := v_tuple_data + 7 & ~7; -- align to 8 bytes
 
   IF v_num_varlen_columns > 0 THEN
-	  SELECT ' + (' || string_agg(format('sum(_timescaledb_functions.compressed_data_column_size(%I,NULL::%s))', attname, pg_catalog.format_type(atttypid, atttypmod)), ' + ') || ')' FROM pg_attribute INTO v_varlen_query WHERE attrelid = v_uncompressed_chunk AND attnum > 0 AND NOT attisdropped AND attlen = -1;
+	  SELECT ' + (' || string_agg(
+	    CASE WHEN attname = ANY(v_segmentby)
+	      THEN format('sum(pg_catalog.pg_column_size(%I) * _ts_meta_count)', attname)
+	      ELSE format('sum(_timescaledb_functions.compressed_data_column_size(%I,NULL::%s))', attname, pg_catalog.format_type(atttypid, atttypmod))
+	    END, ' + ') || ')' FROM pg_attribute INTO v_varlen_query WHERE attrelid = v_uncompressed_chunk AND attnum > 0 AND NOT attisdropped AND attlen = -1;
   END IF;
 
   EXECUTE format('SELECT sum(_ts_meta_count) FROM %s', v_compressed_chunk) INTO tuples;
@@ -643,7 +648,11 @@ BEGIN
   FOR v_index, v_varlen_query, v_columns IN
     SELECT
       i.indexrelid::regclass,
-      (SELECT ' + (' || string_agg(format('sum(_timescaledb_functions.compressed_data_column_size(%I,NULL::%s))', attname, pg_catalog.format_type(atttypid, atttypmod)), ' + ' ORDER BY attnum) || ')' FROM pg_attribute att WHERE att.attrelid=i.indrelid AND attnum =ANY(i.indkey)),
+      (SELECT ' + (' || string_agg(
+         CASE WHEN attname = ANY(v_segmentby)
+           THEN format('sum(pg_catalog.pg_column_size(%I) * _ts_meta_count)', attname)
+           ELSE format('sum(_timescaledb_functions.compressed_data_column_size(%I,NULL::%s))', attname, pg_catalog.format_type(atttypid, atttypmod))
+         END, ' + ' ORDER BY attnum) || ')' FROM pg_attribute att WHERE att.attrelid=i.indrelid AND attnum =ANY(i.indkey)),
       array_length(i.indkey,1) FROM pg_index i
     WHERE i.indrelid = v_uncompressed_chunk
   LOOP

tsl/test/expected/uncompressed_size.out

@@ -46,3 +46,24 @@ SELECT * FROM _timescaledb_functions.estimate_uncompressed_size('_timescaledb_in
 --------+---------------+------------+------------
    3000 |        138000 |      60000 |     198000
 
+-- test varlen segmentby column keeps
+CREATE TABLE t3 (time timestamptz, device text, value float) WITH (tsdb.hypertable, tsdb.segmentby = 'device');
+NOTICE:  using column "time" as partitioning column
+INSERT INTO t3 SELECT '2025-01-01'::timestamptz + format('%s ms', i)::interval, format('Device %s', i%10), 1 + 1/i FROM generate_series(1, 100000) i;
+CREATE INDEX t3_device_idx ON t3 (device);
+VACUUM FULL t3;
+SELECT compress_chunk(chunk) FROM show_chunks('t3') AS chunk;
+             compress_chunk             
+----------------------------------------
+ _timescaledb_internal._hyper_5_7_chunk
+
+SELECT l.relation_size > 0 AS relation_positive,
+       l.index_size > 0 AS index_positive,
+       l.total_size = l.relation_size + l.index_size AS total_consistent
+FROM show_chunks('t3') AS chunk
+JOIN _timescaledb_catalog.compression_settings cs ON cs.relid = chunk
+JOIN LATERAL _timescaledb_functions.estimate_uncompressed_size(cs.compress_relid) l ON true;
+ relation_positive | index_positive | total_consistent 
+-------------------+----------------+------------------
+ t                 | t              | t
+

tsl/test/sql/uncompressed_size.sql

@@ -22,3 +22,17 @@ INSERT INTO t1 SELECT '2026-01-01'::timestamptz + format('%s ms', i)::interval,
 SELECT compress_chunk(chunk) FROM show_chunks('t1') AS chunk;
 SELECT _timescaledb_functions.compressed_data_info(time), _timescaledb_functions.compressed_data_info(device), _timescaledb_functions.compressed_data_info(value) FROM _timescaledb_internal.compress_hyper_2_6_chunk;
 SELECT * FROM _timescaledb_functions.estimate_uncompressed_size('_timescaledb_internal.compress_hyper_2_6_chunk');
+
+-- test varlen segmentby column keeps
+CREATE TABLE t3 (time timestamptz, device text, value float) WITH (tsdb.hypertable, tsdb.segmentby = 'device');
+INSERT INTO t3 SELECT '2025-01-01'::timestamptz + format('%s ms', i)::interval, format('Device %s', i%10), 1 + 1/i FROM generate_series(1, 100000) i;
+CREATE INDEX t3_device_idx ON t3 (device);
+VACUUM FULL t3;
+SELECT compress_chunk(chunk) FROM show_chunks('t3') AS chunk;
+
+SELECT l.relation_size > 0 AS relation_positive,
+       l.index_size > 0 AS index_positive,
+       l.total_size = l.relation_size + l.index_size AS total_consistent
+FROM show_chunks('t3') AS chunk
+JOIN _timescaledb_catalog.compression_settings cs ON cs.relid = chunk
+JOIN LATERAL _timescaledb_functions.estimate_uncompressed_size(cs.compress_relid) l ON true;