From ea5dd042914e1cb222d29b79eab0b7413802f395 Mon Sep 17 00:00:00 2001 From: Gregory Gogin Date: Tue, 18 Aug 2026 04:44:46 +0200 Subject: [PATCH] fix(ai-cost): scope the invoice layers' incremental boundary to one source MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #2607 replaced the table-wide `_version > max(_version)` on the AI silver classes with `silver_incremental_watermark`, because a class is written by every connector feeding it and each producer stamps `_version` from its own clock: whichever commits first raises the boundary above another producer's rows, and those rows stay below it forever, silently. `class_ai_invoice` landed from a branch that predated that fix, so it is the only AI class still carrying the pattern. The same boundary sits one layer down, in `claude_team__ai_invoice`, and fixing only the class would have changed nothing observable: a second instance's rows are dropped at staging before silver ever sees them. Two instances of one connector do write to one staging table — that is why `insight_source_id` is part of every unique_key and why `check_connection` refuses an empty one. Both layers now take the same macro, rather than a second implementation of it, and staging's empty-table guard goes with it: a `coalesce` to the epoch already admits an instance the table has never seen, which is what the guard stood in for. The e2e gains a second source instance read BEFORE everything the class already holds — the shape a per-table boundary silently drops. Reverting either layer alone fails it. `assert_ai_staging_rows_reach_silver` deliberately does NOT gain this class. Its `ref()` calls sit inside a loop over `materialised_models_for_tag`, which is empty at parse time, so dbt records no dependencies and the test errors with "unable to infer all dependencies" on any deployment where a listed class has a materialised contributor. Adding a fourth entry would only widen the set of stands where a scheduled data-quality run turns red. Signed-off-by: Gregory Gogin --- .../dbt/claude_team__ai_invoice.sql | 30 +++++++++----- src/ingestion/silver/ai/class_ai_invoice.sql | 8 ++-- .../e2e/metrics/test_ai_invoice_silver.py | 39 ++++++++++++++++++- 3 files changed, 61 insertions(+), 16 deletions(-) diff --git a/src/ingestion/connectors/ai/claude-team-invoices/dbt/claude_team__ai_invoice.sql b/src/ingestion/connectors/ai/claude-team-invoices/dbt/claude_team__ai_invoice.sql index 2bca8aa66..69372ef77 100644 --- a/src/ingestion/connectors/ai/claude-team-invoices/dbt/claude_team__ai_invoice.sql +++ b/src/ingestion/connectors/ai/claude-team-invoices/dbt/claude_team__ai_invoice.sql @@ -91,14 +91,24 @@ SELECT FROM latest_per_key {% if is_incremental() %} -- A row only ever changes because a newer read produced it, so rows read since - -- the last build are the only ones that can carry anything new. The empty-table - -- guard mirrors the sibling models: over an empty `this` the max is the epoch - -- and every row would be filtered out. - WHERE ( - (SELECT count() FROM {{ this }}) = 0 - OR _airbyte_extracted_at > ( - SELECT coalesce(max(collected_at), toDateTime64('1970-01-01 00:00:00', 3)) - FROM {{ this }} - ) - ) + -- the last build are the only ones that can carry anything new. + -- + -- Scoped to ONE source instance, for the reason silver_incremental_watermark + -- gives at the class above: two instances of this connector write here, each + -- stamping its own extraction clock, and a boundary taken over the whole table + -- lets whichever ran first put the other's rows below it FOREVER. `coalesce` + -- to the epoch is what admits an instance the table has never seen — comparing + -- against NULL would drop every row of every new one. + LEFT JOIN ( + SELECT + insight_tenant_id, + source_id AS watermark_source_id, + max(collected_at) AS max_collected + FROM {{ this }} + GROUP BY insight_tenant_id, source_id + ) AS watermarks + ON latest_per_key.tenant_id = watermarks.insight_tenant_id + AND latest_per_key.source_id = watermarks.watermark_source_id + WHERE latest_per_key._airbyte_extracted_at + > coalesce(watermarks.max_collected, toDateTime64('1970-01-01 00:00:00', 3)) {% endif %} diff --git a/src/ingestion/silver/ai/class_ai_invoice.sql b/src/ingestion/silver/ai/class_ai_invoice.sql index 6c922cb78..af28db377 100644 --- a/src/ingestion/silver/ai/class_ai_invoice.sql +++ b/src/ingestion/silver/ai/class_ai_invoice.sql @@ -19,9 +19,7 @@ -- -- depends_on: {{ ref('claude_team__ai_invoice') }} -SELECT * FROM ( +SELECT candidate.* FROM ( {{ union_by_tag('silver:class_ai_invoice') }} -) -{% if is_incremental() %} -WHERE _version > (SELECT max(_version) FROM {{ this }}) -{% endif %} +) AS candidate +{{ silver_incremental_watermark(['insight_tenant_id', 'source_id']) }} diff --git a/src/ingestion/tests/e2e/metrics/test_ai_invoice_silver.py b/src/ingestion/tests/e2e/metrics/test_ai_invoice_silver.py index 4f89564aa..5b4e4a632 100644 --- a/src/ingestion/tests/e2e/metrics/test_ai_invoice_silver.py +++ b/src/ingestion/tests/e2e/metrics/test_ai_invoice_silver.py @@ -35,6 +35,7 @@ # recovery pair is only legible on its own. SOURCE_ONE_BUILD = "claude-team-invoices-recovered-one-build" SOURCE_TWO_BUILDS = "claude-team-invoices-recovered-two-builds" +SOURCE_SECOND_INSTANCE = "claude-team-invoices-second-instance" # 2026-08-01 and 2026-09-01 UTC — the window a monthly line charges for. AUG_START, SEP_START = 1785542400, 1788220800 @@ -42,7 +43,7 @@ # dating by this instead of by the period would file the row in July. RAISED_AT = 1785456000 -# Staging admits only rows read strictly after what it already holds, and that +# Staging admits only rows read strictly after what it already holds, and THAT # watermark is table-wide. So every instant below is distinct and ascends in the # order the fixtures run: reuse one and the watermark, not the model, decides which # rows arrive — an assertion about the model would then hold with the model deleted. @@ -52,6 +53,11 @@ TWO_BUILDS_BROKEN_AT = "2026-09-05T00:00:00Z" TWO_BUILDS_RECOVERED_AT = "2026-09-06T00:00:00Z" +# The class's own boundary is per source instance, so this one deliberately does +# NOT ascend: it is read before every instant above, which is what a second +# connector instance stamping `_version` from its own clock looks like. +SECOND_INSTANCE_READ_AT = "2026-08-20T00:00:00Z" + def _base(source_id: str, read_at: str) -> dict: """Every column of the bronze table, all absent but the envelope. @@ -388,3 +394,34 @@ def test_a_recovery_across_two_builds_leaves_no_gap_behind(recovered_across_two_ assert invoice["invoice_id"] == "in_RECOVERED", "the gap row became the enriched one" assert invoice["invoice_net_cents"] == 16_000 assert recovered_across_two_builds[1]["invoice_net_cents"] is None, "counted once, not twice" + + +@pytest.fixture +def second_instance_silver( + ch_migrations_applied: SessionConfig, ch_seeder: CHSeeder, dbt_runner: DbtRunner, worker_ctx: WorkerContext +) -> list[dict]: + """A second source instance, read BEFORE everything the class already holds.""" + rows = [ + _invoice_row( + "pi_second", + 4_200, + 4_000, + source_id=SOURCE_SECOND_INSTANCE, + read_at=SECOND_INSTANCE_READ_AT, + invoice_id="in_SECOND", + period_start_ts=AUG_START, + period_end_ts=SEP_START, + ) + ] + _seed_and_build(ch_seeder, dbt_runner, worker_ctx, rows) + return _read_class(ch_migrations_applied, SOURCE_SECOND_INSTANCE) + + +def test_a_second_source_instance_is_not_shut_out_by_the_first(second_instance_silver): + """The class is written by every connector feeding it, each stamping `_version` + from its own clock. A boundary taken over the whole table would leave this + instance's rows below whatever the first instance committed — forever, and + silently. The boundary is per instance, so an instance the class has never + seen has no boundary to clear.""" + assert [r["invoice_id"] for r in second_instance_silver] == ["in_SECOND"] + assert second_instance_silver[0]["invoice_net_cents"] == 4_000