diff --git a/benchmarks/duckshop/edge_cases/orchestration/worker_freshness_skip.yml b/benchmarks/duckshop/edge_cases/orchestration/worker_freshness_skip.yml new file mode 100644 index 000000000..0d8f9d076 --- /dev/null +++ b/benchmarks/duckshop/edge_cases/orchestration/worker_freshness_skip.yml @@ -0,0 +1,16 @@ +# A scheduled build where raw_orders fails its freshness check and raw_returns has none. +# The planner skips the models that read raw_orders before execution; the worker must never +# report those skips as started, and models that read raw_returns must still build. +sources: + - name: raw_orders + description: Raw orders. + expression: SELECT 1 AS order_id + freshness: + strategy: sql + type: timestamp + query: SELECT CAST('2000-01-01 00:00:00' AS TIMESTAMP) AS data_version + age_policy: + error_after: 1h + - name: raw_returns + description: Raw returns. + expression: SELECT 2 AS return_id diff --git a/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot_execution.rs b/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot_execution.rs index e117942e2..45463f934 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot_execution.rs +++ b/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot_execution.rs @@ -213,10 +213,16 @@ impl WorkerBuild { self.start_key(py, &format!("{kind}:{name}")) } - /// Report a selected node as started; the deterministic event ID makes a repeat a no-op. + /// Report a selected node as started unless its outcome is final, such as a planner skip. pub(crate) fn start_key(&self, py: Python<'_>, key: &str) -> PyResult<()> { self.check_cancelled(py)?; - if self.execution.claim.request.selection.contains(key) { + if self.execution.claim.request.selection.contains(key) + && !self + .completed + .lock() + .map_err(|_| poisoned())? + .contains_key(key) + { self.worker.bind(py).get().change( py, &format!("node-started/{key}"), diff --git a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_worker.py b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_worker.py index fc0a00128..66bab82ad 100644 --- a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_worker.py +++ b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_worker.py @@ -193,6 +193,71 @@ def test_given_a_scheduled_group_when_workers_execute_then_dependencies_and_fail assert warehouse.sql(test_case.warehouse_query).fetchall() == test_case.expected_rows +@pytest.mark.parametrize( + "test_case", + [ + WorkerExecutionTestCase( + "a stale source blocks only its own branch", + "SELECT 11 AS amount", + "SELECT amount FROM orders UNION ALL SELECT return_id FROM returns ORDER BY 1", + "partial", + [(2,), (11,)], + files={ + "sources/raw.yml": Path( + "benchmarks/duckshop/edge_cases/orchestration/worker_freshness_skip.yml" + ).read_text(), + "models/stale_orders.sql": ( + "MODEL (description 'Stale orders.', materialized table);\n" + 'SELECT order_id FROM __source("raw_orders")\n' + ), + "models/late_orders.sql": ( + "MODEL (description 'Late orders.', materialized table);\n" + 'SELECT order_id FROM __source("raw_orders")\n' + ), + "models/returns.sql": ( + "MODEL (description 'Returns.', materialized table);\n" + 'SELECT return_id FROM __source("raw_returns")\n' + ), + "orchestration/orders.py": ( + "from sqlbuild.orchestration import job, schedule\n" + 'annual_orders = job("annual_orders", ' + 'select="orders stale_orders late_orders returns", ' + 'description="Build orders.", ' + 'schedule=schedule(cron="0 0 * * *", timezone="UTC", ' + 'starts_at="2000-01-01T00:00:00Z"))\n' + ), + }, + ) + ], + ids=lambda case: case.description, +) +def test_given_a_stale_source_when_a_worker_builds_then_blocked_nodes_finish_skipped_unstarted( + test_case: WorkerExecutionTestCase, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + with scheduled_worker(tmp_path, monkeypatch, test_case) as worker: + run: dict[str, Any] = wait_for_run( + worker.api, worker.process, worker.log, job_key="annual_orders" + ) + assert run["status"] == test_case.expected_status, worker.log.read_text() + with sqlite3.connect(worker.control_store) as control: + nodes: list[tuple[str, str, bool]] = [ + (key, status, started is not None) + for key, status, started in control.execute( + "SELECT node_key, status, started_at FROM sqb_run_nodes " + "WHERE run_id=? AND kind='node' ORDER BY node_key", + (run["run_id"],), + ) + ] + assert nodes == [ + ("model:late_orders", "skipped", False), + ("model:orders", "success", True), + ("model:returns", "success", True), + ("model:stale_orders", "skipped", False), + ], worker.log.read_text() + with duckdb.connect(str(worker.project / "warehouse.duckdb"), read_only=True) as warehouse: + assert warehouse.sql(test_case.warehouse_query).fetchall() == test_case.expected_rows + + @pytest.mark.parametrize( "test_case", [