Skip to content

Commit eeb5a66

Browse files
timescale-automationtimescale-automation
committed
Merge remote-tracking branch 'origin/2.29.x' into HEAD
2 parents 1a93608 + dec1bc3 commit eeb5a66

4 files changed

Lines changed: 306 additions & 32 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: 57 additions & 32 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,7 +151,7 @@ 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,
153156
bool *prev_last_isnull, Datum *curr_first,
154157
bool *curr_first_isnull);
@@ -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)