Skip to content

Commit e431399

Browse files
Poroma-BanerjeePoroma-Banerjee
committed
Fix overlap detection with running max
Overlap detection compared only against the immediately previous batch's last value. A small intermediate batch (e.g. a 1-row boundary batch) could reset the comparison point and hide overlaps with later batches, requiring multiple compaction passes to converge. Track the running maximum of last values instead, so all overlaps within a segment are detected in a single pass. Uses full-tuple comparison to avoid mixing column values from different batches in multi-column orderby.
1 parent 77d9d32 commit e431399

4 files changed

Lines changed: 307 additions & 33 deletions

File tree

‎.unreleased/pr_10315‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
Fixes: #10315 Fix overlap detection with running max

‎tsl/src/compression/recompress.c‎

Lines changed: 58 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -77,10 +77,11 @@ typedef struct CompactChunkScanState
7777
Datum *curr_last;
7878
bool *curr_last_isnull;
7979

80-
/* Last-row orderby tuple of the batch processed just before the current
80+
/* Max last-row orderby tuple from the batches processed before the current
8181
* one. Holds copies so it survives advancing the index scan. */
82-
Datum *prev_last;
83-
bool *prev_last_isnull;
82+
Datum *max_last;
83+
bool *max_last_isnull;
84+
bool max_last_set; /* false until the first batch is recorded */
8485
} CompactChunkScanState;
8586

8687
static CompactChunkScanState *
@@ -95,12 +96,13 @@ compact_chunk_scan_state_init(RecompressContext *recompress_ctx)
9596
state->curr_first_isnull = palloc(sizeof(bool) * recompress_ctx->num_orderby);
9697
state->curr_last = palloc(sizeof(Datum) * recompress_ctx->num_orderby);
9798
state->curr_last_isnull = palloc(sizeof(bool) * recompress_ctx->num_orderby);
98-
state->prev_last = palloc0(sizeof(Datum) * recompress_ctx->num_orderby);
99-
state->prev_last_isnull = palloc(sizeof(bool) * recompress_ctx->num_orderby);
99+
state->max_last = palloc0(sizeof(Datum) * recompress_ctx->num_orderby);
100+
state->max_last_isnull = palloc(sizeof(bool) * recompress_ctx->num_orderby);
100101
for (int i = 0; i < recompress_ctx->num_orderby; i++)
101102
{
102-
state->prev_last_isnull[i] = true;
103+
state->max_last_isnull[i] = true;
103104
}
105+
state->max_last_set = false;
104106
return state;
105107
}
106108

@@ -112,14 +114,15 @@ compact_chunk_scan_state_reset(CompactChunkScanState *state, RecompressContext *
112114
for (int i = 0; i < recompress_ctx->num_orderby; i++)
113115
{
114116
int key = recompress_ctx->num_segmentby + i;
115-
if (!state->prev_last_isnull[i] && !recompress_ctx->key_byval[key] &&
116-
PointerIsValid(DatumGetPointer(state->prev_last[i])))
117+
if (!state->max_last_isnull[i] && !recompress_ctx->key_byval[key] &&
118+
PointerIsValid(DatumGetPointer(state->max_last[i])))
117119
{
118-
pfree(DatumGetPointer(state->prev_last[i]));
120+
pfree(DatumGetPointer(state->max_last[i]));
119121
}
120-
state->prev_last[i] = (Datum) 0;
121-
state->prev_last_isnull[i] = true;
122+
state->max_last[i] = (Datum) 0;
123+
state->max_last_isnull[i] = true;
122124
}
125+
state->max_last_set = false;
123126
}
124127

125128
static bool fetch_uncompressed_chunk_into_tuplesort(Tuplesortstate *tuplesortstate,
@@ -148,9 +151,9 @@ static IndexScanDesc compact_chunk_begin_index_scan(Relation compressed_chunk_re
148151
Relation index_rel, Snapshot snapshot);
149152
static void read_batch_firstlast(IndexScanDesc index_scan, RecompressContext *recompress_ctx,
150153
CompactChunkScanState *state);
151-
static void save_prev_last(CompactChunkScanState *state, RecompressContext *recompress_ctx);
154+
static void save_new_last(CompactChunkScanState *state, RecompressContext *recompress_ctx);
152155
static bool batches_overlap_firstlast(RecompressContext *recompress_ctx, Datum *prev_last,
153-
bool *prev_last_isnull, Datum *curr_first,
156+
bool *max_last_isnull, Datum *curr_first,
154157
bool *curr_first_isnull);
155158
static void decompress_batch_to_tuplesort(TupleTableSlot *slot, TupleDesc tupdesc,
156159
RowDecompressor *decompressor,
@@ -990,28 +993,41 @@ read_batch_firstlast(IndexScanDesc index_scan, RecompressContext *recompress_ctx
990993

991994
/*
992995
* Remember the current batch's last-row orderby tuple as the predecessor for
993-
* the next batch. The index tuple is only valid for the current scan position,
996+
* the next batch only if it is greater than the saved last-row orderby value.
997+
* The index tuple is only valid for the current scan position,
994998
* so pass-by-reference values are deep-copied to survive advancing the scan.
995999
*/
9961000
static void
997-
save_prev_last(CompactChunkScanState *state, RecompressContext *recompress_ctx)
1001+
save_new_last(CompactChunkScanState *state, RecompressContext *recompress_ctx)
9981002
{
1003+
/* Skip if max_last is already set and curr_last does not exceed its value. */
1004+
if (state->max_last_set && !batches_overlap_firstlast(recompress_ctx,
1005+
state->curr_last,
1006+
state->curr_last_isnull,
1007+
state->max_last,
1008+
state->max_last_isnull))
1009+
{
1010+
return;
1011+
}
1012+
1013+
state->max_last_set = true;
1014+
9991015
for (int i = 0; i < recompress_ctx->num_orderby; i++)
10001016
{
10011017
int key = recompress_ctx->num_segmentby + i;
10021018

1003-
if (!state->prev_last_isnull[i] && !recompress_ctx->key_byval[key] &&
1004-
PointerIsValid(DatumGetPointer(state->prev_last[i])))
1019+
if (!state->max_last_isnull[i] && !recompress_ctx->key_byval[key] &&
1020+
PointerIsValid(DatumGetPointer(state->max_last[i])))
10051021
{
1006-
pfree(DatumGetPointer(state->prev_last[i]));
1022+
pfree(DatumGetPointer(state->max_last[i]));
10071023
}
10081024

1009-
state->prev_last_isnull[i] = state->curr_last_isnull[i];
1010-
state->prev_last[i] = state->curr_last_isnull[i] ?
1011-
(Datum) 0 :
1012-
datumCopy(state->curr_last[i],
1013-
recompress_ctx->key_byval[key],
1014-
recompress_ctx->key_typlen[key]);
1025+
state->max_last_isnull[i] = state->curr_last_isnull[i];
1026+
state->max_last[i] = state->curr_last_isnull[i] ?
1027+
(Datum) 0 :
1028+
datumCopy(state->curr_last[i],
1029+
recompress_ctx->key_byval[key],
1030+
recompress_ctx->key_typlen[key]);
10151031
}
10161032
}
10171033

@@ -1137,13 +1153,15 @@ compact_chunk_find_overlapping_batches(Relation compressed_chunk_rel, IndexScanD
11371153
state->seg_values,
11381154
state->seg_isnull,
11391155
recompress_ctx->num_segmentby);
1140-
save_prev_last(state, recompress_ctx);
1156+
/* Reset running max for the new segment group. */
1157+
state->max_last_set = false;
1158+
save_new_last(state, recompress_ctx);
11411159
continue;
11421160
}
11431161

11441162
if (batches_overlap_firstlast(recompress_ctx,
1145-
state->prev_last,
1146-
state->prev_last_isnull,
1163+
state->max_last,
1164+
state->max_last_isnull,
11471165
state->curr_first,
11481166
state->curr_first_isnull))
11491167
{
@@ -1154,7 +1172,7 @@ compact_chunk_find_overlapping_batches(Relation compressed_chunk_rel, IndexScanD
11541172

11551173
/* No overlap: this batch becomes the predecessor for the next one. */
11561174
ItemPointerCopy(&index_scan->xs_heaptid, &state->previous_tid);
1157-
save_prev_last(state, recompress_ctx);
1175+
save_new_last(state, recompress_ctx);
11581176
}
11591177

11601178
ExecDropSingleTupleTableSlot(compressed_slot);
@@ -1242,7 +1260,7 @@ compact_chunk_recompress_overlapping_batches(
12421260
state->seg_values,
12431261
state->seg_isnull,
12441262
recompress_ctx->num_segmentby);
1245-
save_prev_last(state, recompress_ctx);
1263+
save_new_last(state, recompress_ctx);
12461264
}
12471265

12481266
while (index_getnext_slot(index_scan, ForwardScanDirection, compressed_slot))
@@ -1277,15 +1295,17 @@ compact_chunk_recompress_overlapping_batches(
12771295
state->seg_values,
12781296
state->seg_isnull,
12791297
recompress_ctx->num_segmentby);
1280-
save_prev_last(state, recompress_ctx);
1298+
/* Reset running max for the new segment group. */
1299+
state->max_last_set = false;
1300+
save_new_last(state, recompress_ctx);
12811301
continue;
12821302
}
12831303

12841304
/* A batch joins the current group when it overlaps its predecessor; the
12851305
* first batch that no longer overlaps closes the group. */
12861306
bool batch_overlaps = batches_overlap_firstlast(recompress_ctx,
1287-
state->prev_last,
1288-
state->prev_last_isnull,
1307+
state->max_last,
1308+
state->max_last_isnull,
12891309
state->curr_first,
12901310
state->curr_first_isnull);
12911311

@@ -1340,7 +1360,7 @@ compact_chunk_recompress_overlapping_batches(
13401360
}
13411361

13421362
ItemPointerCopy(&index_scan->xs_heaptid, &state->previous_tid);
1343-
save_prev_last(state, recompress_ctx);
1363+
save_new_last(state, recompress_ctx);
13441364
}
13451365

13461366
if (overlapping)
@@ -1351,6 +1371,11 @@ compact_chunk_recompress_overlapping_batches(
13511371
ExecDropSingleTupleTableSlot(previous_compressed_slot);
13521372
ExecDropSingleTupleTableSlot(compressed_slot);
13531373

1374+
ereport(DEBUG1,
1375+
(errmsg("compaction processed %d batches (max_batches %d)",
1376+
processed_batches,
1377+
max_batches)));
1378+
13541379
return found_overlaps;
13551380
}
13561381

‎tsl/test/expected/compact_chunk.out‎

Lines changed: 148 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1621,3 +1621,151 @@ SELECT chunk, _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('
16211621
_timescaledb_internal._hyper_14_16_chunk | {COMPRESSED}
16221622

16231623
DROP TABLE metrics_max_batches;
1624+
-- Test Small intermediate batches must not hide overlaps with later batches.
1625+
CREATE TABLE metrics_running_max (time TIMESTAMPTZ NOT NULL, value float)
1626+
WITH (tsdb.hypertable, tsdb.orderby='time');
1627+
NOTICE: using column "time" as partitioning column
1628+
INSERT INTO metrics_running_max SELECT '2025-07-07'::timestamptz + (i || ' minute')::interval, i FROM generate_series(1,1000) i;
1629+
INSERT INTO metrics_running_max SELECT '2025-07-07'::timestamptz + (i || ' minute')::interval, i FROM generate_series(500,1500) i;
1630+
INSERT INTO metrics_running_max SELECT '2025-07-07'::timestamptz + (i || ' minute')::interval, i FROM generate_series(1000,2000) i;
1631+
INSERT INTO metrics_running_max SELECT '2025-07-07'::timestamptz + (i || ' minute')::interval, i FROM generate_series(1500,2500) i;
1632+
-- Should resolve all overlaps in one pass.
1633+
SELECT _timescaledb_functions.compact_chunk(chunk) FROM show_chunks('metrics_running_max') chunk;
1634+
compact_chunk
1635+
------------------------------------------
1636+
_timescaledb_internal._hyper_15_17_chunk
1637+
1638+
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics_running_max') chunk;
1639+
chunk_status_text
1640+
-------------------
1641+
{COMPRESSED}
1642+
1643+
DROP TABLE metrics_running_max;
1644+
-- Test NULL last with NULLS LAST must stay as the running max.
1645+
CREATE TABLE metrics_null_boundary (time TIMESTAMPTZ NOT NULL, value float)
1646+
WITH (tsdb.hypertable, tsdb.orderby='value NULLS LAST');
1647+
NOTICE: using column "time" as partitioning column
1648+
-- insert in a pattern to trigger NULL value overlap
1649+
INSERT INTO metrics_null_boundary
1650+
SELECT '2025-01-02'::timestamptz + (i || ' minute')::interval,
1651+
CASE WHEN i = 1000 THEN NULL ELSE i::float END
1652+
FROM generate_series(1,1000) i;
1653+
INSERT INTO metrics_null_boundary
1654+
SELECT '2025-01-02'::timestamptz + ((999 + i) || ' minute')::interval, (499 + i)::float
1655+
FROM generate_series(1,11) i;
1656+
INSERT INTO metrics_null_boundary
1657+
SELECT '2025-01-02'::timestamptz + ((1010 + i) || ' minute')::interval, (799 + i)::float
1658+
FROM generate_series(1,101) i;
1659+
INSERT INTO metrics_null_boundary
1660+
SELECT '2025-01-02'::timestamptz + ((1200 + i) || ' minute')::interval, (950 + i)::float
1661+
FROM generate_series(1,50) i;
1662+
SELECT cs.compress_relid::regclass::text AS "MIXED_NULLS_CHUNK"
1663+
FROM _timescaledb_catalog.chunk ch
1664+
JOIN _timescaledb_catalog.compression_settings cs
1665+
ON cs.relid = ch.relid
1666+
JOIN _timescaledb_catalog.hypertable ht ON ch.hypertable_id = ht.id
1667+
WHERE ht.table_name = 'metrics_null_boundary'
1668+
ORDER BY ch.id LIMIT 1 \gset
1669+
SELECT ctid, _ts_meta_count, _ts_meta_min_1, _ts_meta_max_1, _ts_meta_v2_first_value, _ts_meta_v2_last_value
1670+
FROM :MIXED_NULLS_CHUNK
1671+
ORDER BY _ts_meta_min_1;
1672+
ctid | _ts_meta_count | _ts_meta_min_1 | _ts_meta_max_1 | _ts_meta_v2_first_value | _ts_meta_v2_last_value
1673+
-------+----------------+----------------+----------------+-------------------------+------------------------
1674+
(0,1) | 1000 | 1 | 999 | 1 |
1675+
(0,2) | 11 | 500 | 510 | 500 | 510
1676+
(0,3) | 101 | 800 | 900 | 800 | 900
1677+
(0,4) | 50 | 951 | 1000 | 951 | 1000
1678+
1679+
-- Should resolve all overlaps in one pass.
1680+
SELECT _timescaledb_functions.compact_chunk(chunk) FROM show_chunks('metrics_null_boundary') chunk;
1681+
compact_chunk
1682+
------------------------------------------
1683+
_timescaledb_internal._hyper_16_18_chunk
1684+
1685+
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics_null_boundary') chunk;
1686+
chunk_status_text
1687+
-------------------
1688+
{COMPRESSED}
1689+
1690+
SELECT ctid, _ts_meta_count, _ts_meta_min_1, _ts_meta_max_1, _ts_meta_v2_first_value, _ts_meta_v2_last_value
1691+
FROM :MIXED_NULLS_CHUNK
1692+
ORDER BY _ts_meta_min_1;
1693+
ctid | _ts_meta_count | _ts_meta_min_1 | _ts_meta_max_1 | _ts_meta_v2_first_value | _ts_meta_v2_last_value
1694+
-------+----------------+----------------+----------------+-------------------------+------------------------
1695+
(0,5) | 1000 | 1 | 894 | 1 | 894
1696+
(0,6) | 162 | 895 | 1000 | 895 |
1697+
1698+
DROP TABLE metrics_null_boundary;
1699+
-- Test with orderby DESC
1700+
CREATE TABLE metrics_rm_desc (time TIMESTAMPTZ NOT NULL, value float)
1701+
WITH (tsdb.hypertable, tsdb.orderby='time DESC');
1702+
NOTICE: using column "time" as partitioning column
1703+
INSERT INTO metrics_rm_desc SELECT '2025-07-07'::timestamptz + (i || ' minute')::interval, i FROM generate_series(1,1000) i;
1704+
INSERT INTO metrics_rm_desc SELECT '2025-07-07'::timestamptz + (i || ' minute')::interval, i FROM generate_series(500,1500) i;
1705+
INSERT INTO metrics_rm_desc SELECT '2025-07-07'::timestamptz + (i || ' minute')::interval, i FROM generate_series(1000,2000) i;
1706+
INSERT INTO metrics_rm_desc SELECT '2025-07-07'::timestamptz + (i || ' minute')::interval, i FROM generate_series(1500,2500) i;
1707+
SELECT _timescaledb_functions.compact_chunk(chunk) FROM show_chunks('metrics_rm_desc') chunk;
1708+
compact_chunk
1709+
------------------------------------------
1710+
_timescaledb_internal._hyper_17_19_chunk
1711+
1712+
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics_rm_desc') chunk;
1713+
chunk_status_text
1714+
-------------------
1715+
{COMPRESSED}
1716+
1717+
SELECT count(*) FROM metrics_rm_desc;
1718+
count
1719+
-------
1720+
4003
1721+
1722+
DROP TABLE metrics_rm_desc;
1723+
-- Running max with multi-column orderby and mixed directions.
1724+
CREATE TABLE metrics_rm_multi (time TIMESTAMPTZ NOT NULL, device TEXT, value float)
1725+
WITH (tsdb.hypertable, tsdb.orderby='time ASC, value DESC');
1726+
NOTICE: using column "time" as partitioning column
1727+
INSERT INTO metrics_rm_multi SELECT '2025-07-07'::timestamptz + (i || ' minute')::interval, 'd1', i FROM generate_series(1,1000) i;
1728+
INSERT INTO metrics_rm_multi SELECT '2025-07-07'::timestamptz + (i || ' minute')::interval, 'd1', i FROM generate_series(500,1500) i;
1729+
INSERT INTO metrics_rm_multi SELECT '2025-07-07'::timestamptz + (i || ' minute')::interval, 'd1', i FROM generate_series(1000,2000) i;
1730+
INSERT INTO metrics_rm_multi SELECT '2025-07-07'::timestamptz + (i || ' minute')::interval, 'd1', i FROM generate_series(1500,2500) i;
1731+
SELECT _timescaledb_functions.compact_chunk(chunk) FROM show_chunks('metrics_rm_multi') chunk;
1732+
compact_chunk
1733+
------------------------------------------
1734+
_timescaledb_internal._hyper_18_20_chunk
1735+
1736+
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics_rm_multi') chunk;
1737+
chunk_status_text
1738+
-------------------
1739+
{COMPRESSED}
1740+
1741+
SELECT count(*) FROM metrics_rm_multi;
1742+
count
1743+
-------
1744+
4003
1745+
1746+
DROP TABLE metrics_rm_multi;
1747+
-- Running max with segmentby: each segment tracked independently.
1748+
CREATE TABLE metrics_rm_seg (time TIMESTAMPTZ NOT NULL, device TEXT, value float)
1749+
WITH (tsdb.hypertable, tsdb.orderby='time', tsdb.segmentby='device');
1750+
NOTICE: using column "time" as partitioning column
1751+
-- Same overlapping pattern for two devices.
1752+
INSERT INTO metrics_rm_seg SELECT '2025-07-07'::timestamptz + (i || ' minute')::interval, d, i FROM generate_series(1,1000) i, (VALUES ('d1'), ('d2')) AS v(d);
1753+
INSERT INTO metrics_rm_seg SELECT '2025-07-07'::timestamptz + (i || ' minute')::interval, d, i FROM generate_series(500,1500) i, (VALUES ('d1'), ('d2')) AS v(d);
1754+
INSERT INTO metrics_rm_seg SELECT '2025-07-07'::timestamptz + (i || ' minute')::interval, d, i FROM generate_series(1000,2000) i, (VALUES ('d1'), ('d2')) AS v(d);
1755+
INSERT INTO metrics_rm_seg SELECT '2025-07-07'::timestamptz + (i || ' minute')::interval, d, i FROM generate_series(1500,2500) i, (VALUES ('d1'), ('d2')) AS v(d);
1756+
SELECT _timescaledb_functions.compact_chunk(chunk) FROM show_chunks('metrics_rm_seg') chunk;
1757+
compact_chunk
1758+
------------------------------------------
1759+
_timescaledb_internal._hyper_19_21_chunk
1760+
1761+
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics_rm_seg') chunk;
1762+
chunk_status_text
1763+
-------------------
1764+
{COMPRESSED}
1765+
1766+
SELECT count(*) FROM metrics_rm_seg;
1767+
count
1768+
-------
1769+
8006
1770+
1771+
DROP TABLE metrics_rm_seg;

0 commit comments

Comments
 (0)