Skip to content

Commit 0bbecd2

Browse files
antekresicantekresic
committed
Fix segmentby crash in cagg invalidation tracking
When direct-compress writes (INSERT/COPY/cagg refresh) and direct batch delete record continuous aggregate invalidations, they look up _ts_meta_min/max metadata columns for the cagg time dimension to derive the invalidation range. Segmentby columns don't have those metadata columns (they are stored as scalars, not as compressed arrays with sparse-index columns), so when the time column is configured as segmentby the lookup yields NULL/ InvalidAttrNumber and the code crashes: * row_compressor_flush dereferences a NULL minmax_builder. * The direct batch delete consumer reads compressed_datums[-1] (AttrNumberGetAttrOffset(0) = -1). Detect the segmentby case in both paths and read the value from the segmentby column directly: every row in the batch shares the same segmentby value, so it serves as both bounds of the invalidation range. The remaining "no minmax and not segmentby" case shouldn't occur in a well-formed configuration; Ensure is used in row_compressor_flush so the invariant produces a clean ERROR rather than a crash if it ever does.
1 parent cf36bbc commit 0bbecd2

9 files changed

Lines changed: 381 additions & 42 deletions

‎tsl/src/compression/compression.c‎

Lines changed: 28 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1567,7 +1567,8 @@ row_compressor_flush(RowCompressor *row_compressor, BulkWriter *writer, bool cha
15671567
if (row_compressor->invalidation)
15681568
{
15691569
InvalidationSettings *settings = row_compressor->invalidation;
1570-
AttrNumber dim_attnum = AttrOffsetGetAttrNumber(settings->invalidation_column_offset);
1570+
int16 dim_offset = settings->invalidation_column_offset;
1571+
AttrNumber dim_attnum = AttrOffsetGetAttrNumber(dim_offset);
15711572

15721573
BatchMetadataBuilderMinMax *minmax_builder = NULL;
15731574
ListCell *lc;
@@ -1585,11 +1586,32 @@ row_compressor_flush(RowCompressor *row_compressor, BulkWriter *writer, bool cha
15851586
}
15861587
}
15871588

1588-
Assert(minmax_builder != NULL);
1589-
Datum min = row_compressor->compressed_values[minmax_builder->min_metadata_attr_offset];
1590-
Datum max = row_compressor->compressed_values[minmax_builder->max_metadata_attr_offset];
1591-
int64 start = ts_time_value_to_internal(min, minmax_builder->type_oid);
1592-
int64 end = ts_time_value_to_internal(max, minmax_builder->type_oid);
1589+
Datum min;
1590+
Datum max;
1591+
Oid type_oid;
1592+
if (minmax_builder != NULL)
1593+
{
1594+
min = row_compressor->compressed_values[minmax_builder->min_metadata_attr_offset];
1595+
max = row_compressor->compressed_values[minmax_builder->max_metadata_attr_offset];
1596+
type_oid = minmax_builder->type_oid;
1597+
}
1598+
else
1599+
{
1600+
/*
1601+
* No minmax sparse index for the cagg invalidation column means it
1602+
* is configured as segmentby. Every row in the batch shares the
1603+
* same segmentby value, so use it as both bounds.
1604+
*/
1605+
PerColumn *dim_column = &row_compressor->per_column[dim_offset];
1606+
Ensure(dim_column->compressor == NULL && dim_column->segment_info != NULL,
1607+
"cagg invalidation column has no minmax builder and is not segmentby");
1608+
Ensure(!dim_column->segment_info->is_null, "cagg invalidation column is NULL");
1609+
int16 compressed_col = row_compressor->uncompressed_col_to_compressed_col[dim_offset];
1610+
min = max = row_compressor->compressed_values[compressed_col];
1611+
type_oid = TupleDescAttr(row_compressor->in_desc, dim_offset)->atttypid;
1612+
}
1613+
int64 start = ts_time_value_to_internal(min, type_oid);
1614+
int64 end = ts_time_value_to_internal(max, type_oid);
15931615
continuous_agg_invalidate_range(settings->hypertable_id, settings->chunk_relid, start, end);
15941616
}
15951617

‎tsl/src/compression/compression_dml.c‎

Lines changed: 30 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -568,24 +568,40 @@ decompress_batches_for_update_delete(ModifyHypertableState *ht_state, Chunk *chu
568568
if (delete_only && ht_state->has_continuous_aggregate)
569569
{
570570
const Dimension *time_dim = hyperspace_get_open_dimension(ht_state->ht->space, 0);
571-
AttrNumber chunk_time_attno =
572-
get_attnum(chunk->table_id, NameStr(time_dim->fd.column_name));
571+
const char *time_col_name = NameStr(time_dim->fd.column_name);
572+
AttrNumber chunk_time_attno = get_attnum(chunk->table_id, time_col_name);
573573

574574
invalidation_ctx.hypertable_id = ht_state->ht->fd.id;
575575
invalidation_ctx.chunk_relid = chunk->table_id;
576576
invalidation_ctx.time_type_oid = time_dim->fd.column_type;
577-
invalidation_ctx.min_time_attno =
578-
compressed_column_metadata_attno(settings,
579-
chunk->table_id,
580-
chunk_time_attno,
581-
settings->fd.compress_relid,
582-
"min");
583-
invalidation_ctx.max_time_attno =
584-
compressed_column_metadata_attno(settings,
585-
chunk->table_id,
586-
chunk_time_attno,
587-
settings->fd.compress_relid,
588-
"max");
577+
578+
if (ts_array_is_member(settings->fd.segmentby, time_col_name))
579+
{
580+
/*
581+
* Time column is segmentby: every row in the batch shares the same
582+
* value, so use the segmentby column's compressed-tuple attno for
583+
* both bounds. Segmentby columns don't have _ts_meta_min/max
584+
* sparse-index columns to look up.
585+
*/
586+
AttrNumber compressed_attno = get_attnum(settings->fd.compress_relid, time_col_name);
587+
invalidation_ctx.min_time_attno = compressed_attno;
588+
invalidation_ctx.max_time_attno = compressed_attno;
589+
}
590+
else
591+
{
592+
invalidation_ctx.min_time_attno =
593+
compressed_column_metadata_attno(settings,
594+
chunk->table_id,
595+
chunk_time_attno,
596+
settings->fd.compress_relid,
597+
"min");
598+
invalidation_ctx.max_time_attno =
599+
compressed_column_metadata_attno(settings,
600+
chunk->table_id,
601+
chunk_time_attno,
602+
settings->fd.compress_relid,
603+
"max");
604+
}
589605
}
590606

591607
process_predicates(chunk,

‎tsl/test/expected/compression_update_delete-15.out‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3259,3 +3259,55 @@ SELECT * FROM _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log O
32593259
45 | 1577865600000000 | 1577865600000000
32603260
45 | 1609488000000000 | 1609488000000000
32613261

3262+
-- direct batch delete with time column as segmentby. Segmentby columns don't
3263+
-- have _ts_meta_min/max, so the invalidation context must read the segmentby
3264+
-- value directly. The invalidation log entry's bounds must equal the deleted
3265+
-- segmentby value; without the fix the consumer reads compressed_datums[-1]
3266+
-- and produces garbage bounds.
3267+
SET TIME ZONE 'UTC';
3268+
SET
3269+
CREATE TABLE cagg_inval_segmentby_time(time timestamptz NOT NULL, device text, value float)
3270+
WITH (tsdb.hypertable);
3271+
NOTICE: using column "time" as partitioning column
3272+
HINT: Use "timescaledb.partition_column" to specify a different column to use as partitioning column.
3273+
CREATE TABLE
3274+
INSERT INTO cagg_inval_segmentby_time
3275+
SELECT '2025-01-01 00:00:00+00'::timestamptz + (i * interval '5 minutes'), 'd1', random()
3276+
FROM generate_series(1, 50) i;
3277+
INSERT 0 50
3278+
-- Refresh on creation sets the cagg invalidation threshold so subsequent
3279+
-- DELETEs below it write to the hypertable invalidation log.
3280+
CREATE MATERIALIZED VIEW cagg_inval_segmentby_time_cagg
3281+
WITH (tsdb.continuous) AS
3282+
SELECT time_bucket('1 hour', time) FROM cagg_inval_segmentby_time GROUP BY 1;
3283+
NOTICE: refreshing continuous aggregate "cagg_inval_segmentby_time_cagg"
3284+
HINT: Use WITH NO DATA if you do not want to refresh the continuous aggregate on creation.
3285+
CREATE MATERIALIZED VIEW
3286+
ALTER TABLE cagg_inval_segmentby_time SET (
3287+
timescaledb.compress,
3288+
timescaledb.compress_segmentby = 'time'
3289+
);
3290+
NOTICE: updated compression settings will only apply to future compressions
3291+
DETAIL: Existing compressed chunks will not be recompressed.
3292+
HINT: Use compress_chunk(chunk, recompress => true) to recompress.
3293+
ALTER TABLE
3294+
SELECT count(compress_chunk(c)) FROM show_chunks('cagg_inval_segmentby_time') c;
3295+
count
3296+
-------
3297+
1
3298+
3299+
DELETE FROM cagg_inval_segmentby_time WHERE time = '2025-01-01 00:05:00+00';
3300+
DELETE 1
3301+
-- The single deleted segmentby batch covers exactly time = 2025-01-01 00:05:00 UTC.
3302+
-- Convert internal microseconds back to a timestamptz for a stable comparison.
3303+
SELECT to_timestamp(lowest_modified_value / 1000000.0) AS lowest,
3304+
to_timestamp(greatest_modified_value / 1000000.0) AS greatest
3305+
FROM _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log
3306+
WHERE hypertable_id = (SELECT id FROM _timescaledb_catalog.hypertable
3307+
WHERE table_name = 'cagg_inval_segmentby_time');
3308+
lowest | greatest
3309+
------------------------------+------------------------------
3310+
Wed Jan 01 00:05:00 2025 UTC | Wed Jan 01 00:05:00 2025 UTC
3311+
3312+
RESET TIME ZONE;
3313+
RESET

‎tsl/test/expected/compression_update_delete-16.out‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3259,3 +3259,55 @@ SELECT * FROM _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log O
32593259
45 | 1577865600000000 | 1577865600000000
32603260
45 | 1609488000000000 | 1609488000000000
32613261

3262+
-- direct batch delete with time column as segmentby. Segmentby columns don't
3263+
-- have _ts_meta_min/max, so the invalidation context must read the segmentby
3264+
-- value directly. The invalidation log entry's bounds must equal the deleted
3265+
-- segmentby value; without the fix the consumer reads compressed_datums[-1]
3266+
-- and produces garbage bounds.
3267+
SET TIME ZONE 'UTC';
3268+
SET
3269+
CREATE TABLE cagg_inval_segmentby_time(time timestamptz NOT NULL, device text, value float)
3270+
WITH (tsdb.hypertable);
3271+
NOTICE: using column "time" as partitioning column
3272+
HINT: Use "timescaledb.partition_column" to specify a different column to use as partitioning column.
3273+
CREATE TABLE
3274+
INSERT INTO cagg_inval_segmentby_time
3275+
SELECT '2025-01-01 00:00:00+00'::timestamptz + (i * interval '5 minutes'), 'd1', random()
3276+
FROM generate_series(1, 50) i;
3277+
INSERT 0 50
3278+
-- Refresh on creation sets the cagg invalidation threshold so subsequent
3279+
-- DELETEs below it write to the hypertable invalidation log.
3280+
CREATE MATERIALIZED VIEW cagg_inval_segmentby_time_cagg
3281+
WITH (tsdb.continuous) AS
3282+
SELECT time_bucket('1 hour', time) FROM cagg_inval_segmentby_time GROUP BY 1;
3283+
NOTICE: refreshing continuous aggregate "cagg_inval_segmentby_time_cagg"
3284+
HINT: Use WITH NO DATA if you do not want to refresh the continuous aggregate on creation.
3285+
CREATE MATERIALIZED VIEW
3286+
ALTER TABLE cagg_inval_segmentby_time SET (
3287+
timescaledb.compress,
3288+
timescaledb.compress_segmentby = 'time'
3289+
);
3290+
NOTICE: updated compression settings will only apply to future compressions
3291+
DETAIL: Existing compressed chunks will not be recompressed.
3292+
HINT: Use compress_chunk(chunk, recompress => true) to recompress.
3293+
ALTER TABLE
3294+
SELECT count(compress_chunk(c)) FROM show_chunks('cagg_inval_segmentby_time') c;
3295+
count
3296+
-------
3297+
1
3298+
3299+
DELETE FROM cagg_inval_segmentby_time WHERE time = '2025-01-01 00:05:00+00';
3300+
DELETE 1
3301+
-- The single deleted segmentby batch covers exactly time = 2025-01-01 00:05:00 UTC.
3302+
-- Convert internal microseconds back to a timestamptz for a stable comparison.
3303+
SELECT to_timestamp(lowest_modified_value / 1000000.0) AS lowest,
3304+
to_timestamp(greatest_modified_value / 1000000.0) AS greatest
3305+
FROM _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log
3306+
WHERE hypertable_id = (SELECT id FROM _timescaledb_catalog.hypertable
3307+
WHERE table_name = 'cagg_inval_segmentby_time');
3308+
lowest | greatest
3309+
------------------------------+------------------------------
3310+
Wed Jan 01 00:05:00 2025 UTC | Wed Jan 01 00:05:00 2025 UTC
3311+
3312+
RESET TIME ZONE;
3313+
RESET

‎tsl/test/expected/compression_update_delete-17.out‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3265,3 +3265,55 @@ SELECT * FROM _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log O
32653265
45 | 1577865600000000 | 1577865600000000
32663266
45 | 1609488000000000 | 1609488000000000
32673267

3268+
-- direct batch delete with time column as segmentby. Segmentby columns don't
3269+
-- have _ts_meta_min/max, so the invalidation context must read the segmentby
3270+
-- value directly. The invalidation log entry's bounds must equal the deleted
3271+
-- segmentby value; without the fix the consumer reads compressed_datums[-1]
3272+
-- and produces garbage bounds.
3273+
SET TIME ZONE 'UTC';
3274+
SET
3275+
CREATE TABLE cagg_inval_segmentby_time(time timestamptz NOT NULL, device text, value float)
3276+
WITH (tsdb.hypertable);
3277+
NOTICE: using column "time" as partitioning column
3278+
HINT: Use "timescaledb.partition_column" to specify a different column to use as partitioning column.
3279+
CREATE TABLE
3280+
INSERT INTO cagg_inval_segmentby_time
3281+
SELECT '2025-01-01 00:00:00+00'::timestamptz + (i * interval '5 minutes'), 'd1', random()
3282+
FROM generate_series(1, 50) i;
3283+
INSERT 0 50
3284+
-- Refresh on creation sets the cagg invalidation threshold so subsequent
3285+
-- DELETEs below it write to the hypertable invalidation log.
3286+
CREATE MATERIALIZED VIEW cagg_inval_segmentby_time_cagg
3287+
WITH (tsdb.continuous) AS
3288+
SELECT time_bucket('1 hour', time) FROM cagg_inval_segmentby_time GROUP BY 1;
3289+
NOTICE: refreshing continuous aggregate "cagg_inval_segmentby_time_cagg"
3290+
HINT: Use WITH NO DATA if you do not want to refresh the continuous aggregate on creation.
3291+
CREATE MATERIALIZED VIEW
3292+
ALTER TABLE cagg_inval_segmentby_time SET (
3293+
timescaledb.compress,
3294+
timescaledb.compress_segmentby = 'time'
3295+
);
3296+
NOTICE: updated compression settings will only apply to future compressions
3297+
DETAIL: Existing compressed chunks will not be recompressed.
3298+
HINT: Use compress_chunk(chunk, recompress => true) to recompress.
3299+
ALTER TABLE
3300+
SELECT count(compress_chunk(c)) FROM show_chunks('cagg_inval_segmentby_time') c;
3301+
count
3302+
-------
3303+
1
3304+
3305+
DELETE FROM cagg_inval_segmentby_time WHERE time = '2025-01-01 00:05:00+00';
3306+
DELETE 1
3307+
-- The single deleted segmentby batch covers exactly time = 2025-01-01 00:05:00 UTC.
3308+
-- Convert internal microseconds back to a timestamptz for a stable comparison.
3309+
SELECT to_timestamp(lowest_modified_value / 1000000.0) AS lowest,
3310+
to_timestamp(greatest_modified_value / 1000000.0) AS greatest
3311+
FROM _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log
3312+
WHERE hypertable_id = (SELECT id FROM _timescaledb_catalog.hypertable
3313+
WHERE table_name = 'cagg_inval_segmentby_time');
3314+
lowest | greatest
3315+
------------------------------+------------------------------
3316+
Wed Jan 01 00:05:00 2025 UTC | Wed Jan 01 00:05:00 2025 UTC
3317+
3318+
RESET TIME ZONE;
3319+
RESET

‎tsl/test/expected/compression_update_delete-18.out‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3265,3 +3265,55 @@ SELECT * FROM _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log O
32653265
45 | 1577865600000000 | 1577865600000000
32663266
45 | 1609488000000000 | 1609488000000000
32673267

3268+
-- direct batch delete with time column as segmentby. Segmentby columns don't
3269+
-- have _ts_meta_min/max, so the invalidation context must read the segmentby
3270+
-- value directly. The invalidation log entry's bounds must equal the deleted
3271+
-- segmentby value; without the fix the consumer reads compressed_datums[-1]
3272+
-- and produces garbage bounds.
3273+
SET TIME ZONE 'UTC';
3274+
SET
3275+
CREATE TABLE cagg_inval_segmentby_time(time timestamptz NOT NULL, device text, value float)
3276+
WITH (tsdb.hypertable);
3277+
NOTICE: using column "time" as partitioning column
3278+
HINT: Use "timescaledb.partition_column" to specify a different column to use as partitioning column.
3279+
CREATE TABLE
3280+
INSERT INTO cagg_inval_segmentby_time
3281+
SELECT '2025-01-01 00:00:00+00'::timestamptz + (i * interval '5 minutes'), 'd1', random()
3282+
FROM generate_series(1, 50) i;
3283+
INSERT 0 50
3284+
-- Refresh on creation sets the cagg invalidation threshold so subsequent
3285+
-- DELETEs below it write to the hypertable invalidation log.
3286+
CREATE MATERIALIZED VIEW cagg_inval_segmentby_time_cagg
3287+
WITH (tsdb.continuous) AS
3288+
SELECT time_bucket('1 hour', time) FROM cagg_inval_segmentby_time GROUP BY 1;
3289+
NOTICE: refreshing continuous aggregate "cagg_inval_segmentby_time_cagg"
3290+
HINT: Use WITH NO DATA if you do not want to refresh the continuous aggregate on creation.
3291+
CREATE MATERIALIZED VIEW
3292+
ALTER TABLE cagg_inval_segmentby_time SET (
3293+
timescaledb.compress,
3294+
timescaledb.compress_segmentby = 'time'
3295+
);
3296+
NOTICE: updated compression settings will only apply to future compressions
3297+
DETAIL: Existing compressed chunks will not be recompressed.
3298+
HINT: Use compress_chunk(chunk, recompress => true) to recompress.
3299+
ALTER TABLE
3300+
SELECT count(compress_chunk(c)) FROM show_chunks('cagg_inval_segmentby_time') c;
3301+
count
3302+
-------
3303+
1
3304+
3305+
DELETE FROM cagg_inval_segmentby_time WHERE time = '2025-01-01 00:05:00+00';
3306+
DELETE 1
3307+
-- The single deleted segmentby batch covers exactly time = 2025-01-01 00:05:00 UTC.
3308+
-- Convert internal microseconds back to a timestamptz for a stable comparison.
3309+
SELECT to_timestamp(lowest_modified_value / 1000000.0) AS lowest,
3310+
to_timestamp(greatest_modified_value / 1000000.0) AS greatest
3311+
FROM _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log
3312+
WHERE hypertable_id = (SELECT id FROM _timescaledb_catalog.hypertable
3313+
WHERE table_name = 'cagg_inval_segmentby_time');
3314+
lowest | greatest
3315+
------------------------------+------------------------------
3316+
Wed Jan 01 00:05:00 2025 UTC | Wed Jan 01 00:05:00 2025 UTC
3317+
3318+
RESET TIME ZONE;
3319+
RESET

0 commit comments

Comments
 (0)