From 96be56c17b8b399e4b0d2c2a47e18438ff0ab24e Mon Sep 17 00:00:00 2001 From: Alex Luong Date: Fri, 14 Aug 2026 15:20:07 +0700 Subject: [PATCH 1/3] fix(test): pin ClickHouse merges in the chlogstore dedup tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TestEventDedup injects six "legacy duplicate" event rows that reuse the originals' (event_time, event_id) and then asserts a raw count of 9. events is a ReplacingMergeTree keyed on exactly that pair, so a background merge collapses the nine rows to three whenever the server decides to run one. Nothing between the last batch.Send() and the SELECT holds it off, which makes the assertion a race against the merge scheduler: Send() -> SELECT -> merge => 9 pass Send() -> merge -> SELECT => 3 fail Observed as `expected 0x9, got 0x3` during a full-suite run; passes in isolation, because every test shares one ClickHouse server and only under load does the SELECT sit long enough for a merge to land first. The unmerged state is the one worth asserting on: production never reads with FINAL (chlogstore/README.md), which is why the read path dedups client-side at all. So the fix is to stop merges rather than force them — OPTIMIZE FINAL would delete the very condition under test. stopMerges() qualifies the table with currentDatabase(). The bare form applies server-wide and would break the conformance harness, whose FlushWrites calls OPTIMIZE TABLE ... FINAL; each test already gets its own test_ database, dropped on cleanup, so a qualified stop is scoped to the calling test. SYSTEM STOP MERGES accepts a nonexistent table without error (verified on 24.10.4.191), so the helper checks system.tables first — otherwise a wrong name silently restores the flake. TestFetchAndDedupTruncation has the same dependency and is pinned too. It inserts evt-trunc-a twice at one timestamp to force a short first batch; a merge there does not fail the test, it makes the LessOrEqual assertion vacuous and the overshoot path goes unexercised. Verified: package green including conformance, dedup tests 10x clean, and with merges stopped an explicit OPTIMIZE returns `code: 236, Cancelled merging parts` with the duplicate rows still present. Co-Authored-By: Claude Opus 5 (1M context) --- .../logstore/chlogstore/chlogstore_test.go | 42 +++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/internal/logstore/chlogstore/chlogstore_test.go b/internal/logstore/chlogstore/chlogstore_test.go index 914a9c914..c3d56137e 100644 --- a/internal/logstore/chlogstore/chlogstore_test.go +++ b/internal/logstore/chlogstore/chlogstore_test.go @@ -61,6 +61,35 @@ func setupClickHouseConnection(t *testing.T) clickhouse.DB { return chDB } +// stopMerges disables background merges for one table in the calling test's +// database. +// +// ReplacingMergeTree collapses rows sharing the ORDER BY key whenever the +// server decides to merge parts. Tests that assert on raw duplicate rows are +// asserting on the pre-merge state, which the engine is otherwise free to +// change at any moment. Qualifying the table name is what keeps this scoped: +// bare `SYSTEM STOP MERGES` applies server-wide, and every test shares one +// ClickHouse server. Each test has its own database (testinfra.NewClickHouseConfig), +// which is dropped on cleanup, so the setting goes with it. +func stopMerges(t *testing.T, chDB clickhouse.DB, table string) { + t.Helper() + + ctx := context.Background() + var database string + require.NoError(t, chDB.QueryRow(ctx, "SELECT currentDatabase()").Scan(&database)) + + // SYSTEM STOP MERGES accepts a table that does not exist without error, so + // a wrong name here would silently leave merges running and hand back the + // flake it is meant to remove. + var exists uint64 + require.NoError(t, chDB.QueryRow(ctx, + "SELECT count() FROM system.tables WHERE database = ? AND name = ?", + database, table).Scan(&exists)) + require.Equal(t, uint64(1), exists, "table %s.%s must exist before stopping merges", database, table) + + require.NoError(t, chDB.Exec(ctx, "SYSTEM STOP MERGES "+database+"."+table)) +} + func newHarness(_ context.Context, t *testing.T) (drivertest.Harness, error) { t.Helper() @@ -156,6 +185,13 @@ func TestEventDedup(t *testing.T) { chDB := setupClickHouseConnection(t) defer chDB.Close() + // The injected legacy duplicates below reuse the originals' (event_time, + // event_id), so a background merge would collapse them and the raw row + // count would drop from 9 to 3. Production reads unmerged parts — that is + // why the read path dedups at all — so this test asserts on the unmerged + // state and holds merges off to keep it. + stopMerges(t, chDB, "events") + logStore := NewLogStore(chDB, "") tenantID := "dedup-tenant" @@ -285,6 +321,12 @@ func TestFetchAndDedupTruncation(t *testing.T) { chDB := setupClickHouseConnection(t) defer chDB.Close() + // evt-trunc-a is inserted twice at the same event_time, so a background + // merge would collapse it and the first batch would come back already + // deduplicated — the overshoot this test exists to cover would never + // happen and the assertion would pass without exercising anything. + stopMerges(t, chDB, "events") + tenantID := "dedup-truncation" baseTime := time.Now().Truncate(time.Second) From aa4be73eefa38a54f0ff909ee409279eefe953fb Mon Sep 17 00:00:00 2001 From: Alex Luong Date: Fri, 14 Aug 2026 15:23:02 +0700 Subject: [PATCH 2/3] test(chlogstore): derive the stopped table from the log store itself MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to the merge-race fix. stopMerges was called with a "events" string literal, so the name could drift from the one the code under test queries and SYSTEM STOP MERGES would report success while doing nothing. NewLogStore already derives the name as prefix + "events" and keeps it on logStoreImpl.eventsTable. The tests are in-package, so they now assert the concrete type and pass that field — the same value the queries use, prefix handling included. TestFetchAndDedupTruncation's buildEventQuery call had the same literal and now uses the field too. The system.tables check stays as a backstop, since a silently-accepted unknown table would restore the race without failing anything. Co-Authored-By: Claude Opus 5 (1M context) --- .../logstore/chlogstore/chlogstore_test.go | 25 +++++++++++-------- 1 file changed, 15 insertions(+), 10 deletions(-) diff --git a/internal/logstore/chlogstore/chlogstore_test.go b/internal/logstore/chlogstore/chlogstore_test.go index c3d56137e..d76c22069 100644 --- a/internal/logstore/chlogstore/chlogstore_test.go +++ b/internal/logstore/chlogstore/chlogstore_test.go @@ -62,7 +62,8 @@ func setupClickHouseConnection(t *testing.T) clickhouse.DB { } // stopMerges disables background merges for one table in the calling test's -// database. +// database. Pass the logStore's own table field so the name cannot drift from +// the one the code under test queries. // // ReplacingMergeTree collapses rows sharing the ORDER BY key whenever the // server decides to merge parts. Tests that assert on raw duplicate rows are @@ -78,14 +79,14 @@ func stopMerges(t *testing.T, chDB clickhouse.DB, table string) { var database string require.NoError(t, chDB.QueryRow(ctx, "SELECT currentDatabase()").Scan(&database)) - // SYSTEM STOP MERGES accepts a table that does not exist without error, so - // a wrong name here would silently leave merges running and hand back the - // flake it is meant to remove. + // SYSTEM STOP MERGES reports success for a table that does not exist, so an + // unmatched name would leave merges running and silently restore the race + // this call exists to remove. var exists uint64 require.NoError(t, chDB.QueryRow(ctx, "SELECT count() FROM system.tables WHERE database = ? AND name = ?", database, table).Scan(&exists)) - require.Equal(t, uint64(1), exists, "table %s.%s must exist before stopping merges", database, table) + require.Equal(t, uint64(1), exists, "no table %s.%s to stop merges on", database, table) require.NoError(t, chDB.Exec(ctx, "SYSTEM STOP MERGES "+database+"."+table)) } @@ -185,14 +186,15 @@ func TestEventDedup(t *testing.T) { chDB := setupClickHouseConnection(t) defer chDB.Close() + // Concrete type so the test uses the same table names the code under test does. + logStore := NewLogStore(chDB, "").(*logStoreImpl) + // The injected legacy duplicates below reuse the originals' (event_time, // event_id), so a background merge would collapse them and the raw row // count would drop from 9 to 3. Production reads unmerged parts — that is // why the read path dedups at all — so this test asserts on the unmerged // state and holds merges off to keep it. - stopMerges(t, chDB, "events") - - logStore := NewLogStore(chDB, "") + stopMerges(t, chDB, logStore.eventsTable) tenantID := "dedup-tenant" baseTime := time.Now().Truncate(time.Second) @@ -321,11 +323,14 @@ func TestFetchAndDedupTruncation(t *testing.T) { chDB := setupClickHouseConnection(t) defer chDB.Close() + // Concrete type so the test uses the same table names the code under test does. + logStore := NewLogStore(chDB, "").(*logStoreImpl) + // evt-trunc-a is inserted twice at the same event_time, so a background // merge would collapse it and the first batch would come back already // deduplicated — the overshoot this test exists to cover would never // happen and the assertion would pass without exercising anything. - stopMerges(t, chDB, "events") + stopMerges(t, chDB, logStore.eventsTable) tenantID := "dedup-truncation" baseTime := time.Now().Truncate(time.Second) @@ -352,7 +357,7 @@ func TestFetchAndDedupTruncation(t *testing.T) { Compare: "<", SortDir: "desc", }, func(qi pagination.QueryInput) (string, []any) { - return buildEventQuery("events", driver.ListEventRequest{ + return buildEventQuery(logStore.eventsTable, driver.ListEventRequest{ TenantIDs: []string{tenantID}, TimeFilter: driver.TimeFilter{GTE: &startTime}, }, qi) From 9ff659c8bfac4cd17ab3052feb889477397233f0 Mon Sep 17 00:00:00 2001 From: Alex Luong Date: Fri, 14 Aug 2026 15:30:42 +0700 Subject: [PATCH 3/3] test(chlogstore): trim the stopMerges comments MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Keep what the code cannot say — why the table is qualified, and that STOP MERGES is silent on an unknown table — and drop the rest. Co-Authored-By: Claude Opus 5 (1M context) --- .../logstore/chlogstore/chlogstore_test.go | 35 ++++++------------- 1 file changed, 11 insertions(+), 24 deletions(-) diff --git a/internal/logstore/chlogstore/chlogstore_test.go b/internal/logstore/chlogstore/chlogstore_test.go index d76c22069..b58e3744e 100644 --- a/internal/logstore/chlogstore/chlogstore_test.go +++ b/internal/logstore/chlogstore/chlogstore_test.go @@ -61,17 +61,11 @@ func setupClickHouseConnection(t *testing.T) clickhouse.DB { return chDB } -// stopMerges disables background merges for one table in the calling test's -// database. Pass the logStore's own table field so the name cannot drift from -// the one the code under test queries. +// stopMerges holds duplicate rows in place for tests that assert on them — +// ReplacingMergeTree collapses rows sharing the ORDER BY key on merge. // -// ReplacingMergeTree collapses rows sharing the ORDER BY key whenever the -// server decides to merge parts. Tests that assert on raw duplicate rows are -// asserting on the pre-merge state, which the engine is otherwise free to -// change at any moment. Qualifying the table name is what keeps this scoped: -// bare `SYSTEM STOP MERGES` applies server-wide, and every test shares one -// ClickHouse server. Each test has its own database (testinfra.NewClickHouseConfig), -// which is dropped on cleanup, so the setting goes with it. +// The table must be qualified: bare SYSTEM STOP MERGES is server-wide, and every +// test shares one ClickHouse server. func stopMerges(t *testing.T, chDB clickhouse.DB, table string) { t.Helper() @@ -79,9 +73,7 @@ func stopMerges(t *testing.T, chDB clickhouse.DB, table string) { var database string require.NoError(t, chDB.QueryRow(ctx, "SELECT currentDatabase()").Scan(&database)) - // SYSTEM STOP MERGES reports success for a table that does not exist, so an - // unmatched name would leave merges running and silently restore the race - // this call exists to remove. + // SYSTEM STOP MERGES reports success for a table that does not exist. var exists uint64 require.NoError(t, chDB.QueryRow(ctx, "SELECT count() FROM system.tables WHERE database = ? AND name = ?", @@ -186,14 +178,11 @@ func TestEventDedup(t *testing.T) { chDB := setupClickHouseConnection(t) defer chDB.Close() - // Concrete type so the test uses the same table names the code under test does. + // Concrete type: the test needs the same table name the queries use. logStore := NewLogStore(chDB, "").(*logStoreImpl) - // The injected legacy duplicates below reuse the originals' (event_time, - // event_id), so a background merge would collapse them and the raw row - // count would drop from 9 to 3. Production reads unmerged parts — that is - // why the read path dedups at all — so this test asserts on the unmerged - // state and holds merges off to keep it. + // The injected duplicates below share (event_time, event_id) with the + // originals, so a merge would collapse the raw count from 9 to 3. stopMerges(t, chDB, logStore.eventsTable) tenantID := "dedup-tenant" @@ -323,13 +312,11 @@ func TestFetchAndDedupTruncation(t *testing.T) { chDB := setupClickHouseConnection(t) defer chDB.Close() - // Concrete type so the test uses the same table names the code under test does. + // Concrete type: the test needs the same table name the queries use. logStore := NewLogStore(chDB, "").(*logStoreImpl) - // evt-trunc-a is inserted twice at the same event_time, so a background - // merge would collapse it and the first batch would come back already - // deduplicated — the overshoot this test exists to cover would never - // happen and the assertion would pass without exercising anything. + // evt-trunc-a is inserted twice at one event_time; a merge would collapse it + // and the first batch would come back deduplicated, so no overshoot to catch. stopMerges(t, chDB, logStore.eventsTable) tenantID := "dedup-truncation"