From f4bbb9bd84c9daa9cd60de7f20edd0bbea5356ea Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Wed, 29 Jul 2026 10:29:25 -0700 Subject: [PATCH] [tests] Fix racy barrier in Iceberg snapshot tests test_iceberg_ordered_snapshot_ingests_all_rows failed in CI with "assert 0 == 60" on python-multihost. The tests waited on input_connector_iceberg_phase == 2 and then queried the table, but the phase gauge is a reader-side signal: crates/iceberg/src/input.rs sets it one statement after queueing the last records, and push_entry only appends to a mutex-guarded deque that the circuit thread drains later. Multihost loses this race where single-host wins it: the coordinator advances last_committed_step, the only step an ad-hoc query may lease, only after every host reports a committed transaction, whereas a single-host pipeline republishes the snapshot in the thread that just committed. Fix: Wait for total_completed_records to catch up with total_input_records. Signed-off-by: Leonid Ryzhyk --- python/tests/platform/test_iceberg_input.py | 31 +++++++++++++++++++++ 1 file changed, 31 insertions(+) diff --git a/python/tests/platform/test_iceberg_input.py b/python/tests/platform/test_iceberg_input.py index 735f3d88ef4..d1808e40254 100644 --- a/python/tests/platform/test_iceberg_input.py +++ b/python/tests/platform/test_iceberg_input.py @@ -335,13 +335,44 @@ def _row_count(pipeline) -> int: return int(rows[0]["c"]) +def _all_records_completed(pipeline) -> bool: + """Has every record the connector handed over been processed to completion? + + Both counters come from one ``/stats`` response, so they are a consistent + pair. ``total_input_records`` counts a record when the connector buffers it, + which the connector does before it reports the completed phase, so the + target is already final by the time this is polled. + """ + metrics = pipeline.stats().global_metrics + return metrics.total_completed_records >= metrics.total_input_records + + def _wait_for_completed(pipeline, pipeline_name: str, timeout_s: float = 120.0) -> None: + """Wait until the snapshot has been read *and* is queryable. + + The phase gauge is a reader-side signal: the connector sets it right after + queueing its last records, so it can read 2 while the circuit has published + nothing. Under ``transaction_mode = snapshot`` the queued records stay + invisible to queries until the transaction commits, so a query issued on the + gauge alone can see zero rows. + + ``total_completed_records`` supplies the missing half of the barrier. It + advances only at a transaction boundary, the same point at which the circuit + republishes the ad-hoc snapshot, so once it catches up with the records the + connector handed over, a query sees all of them. + """ wait_for_condition( "iceberg snapshot completed (phase == 2)", lambda: _phase_from_metrics(pipeline_name) == 2, timeout_s=timeout_s, poll_interval_s=0.2, ) + wait_for_condition( + "every ingested record processed to completion", + lambda: _all_records_completed(pipeline), + timeout_s=timeout_s, + poll_interval_s=0.2, + ) # ─── tests ───────────────────────────────────────────────────────────────