diff --git a/benchmarks/duckshop/edge_cases/orchestration/shared_inline_audits.json b/benchmarks/duckshop/edge_cases/orchestration/shared_inline_audits.json new file mode 100644 index 0000000000..410c38e177 --- /dev/null +++ b/benchmarks/duckshop/edge_cases/orchestration/shared_inline_audits.json @@ -0,0 +1,15 @@ +{ + "description": "Two models attach the same inline audit to a column of the same name, and the same explicitly named audit. Each attachment is its own snapshot node.", + "files": { + "sqlbuild_project.toml": "name = \"orders\"\nadapter = \"duckdb\"\n[connections.local]\ndatabase = \"warehouse.duckdb\"\n[targets.development]\nschema = \"main\"\n", + "models/orders.sql": "MODEL (\n materialized table,\n description \"Synthetic rows.\",\n columns (id (audits [not_null, unique, accepted_values (name \"id_known\", values [1, 2])])),\n);\n\nSELECT 1 AS id\n", + "models/customers.sql": "MODEL (\n materialized table,\n description \"Synthetic rows.\",\n columns (id (audits [not_null, accepted_values (name \"id_known\", values [1, 2])])),\n);\n\nSELECT 2 AS id\n" + }, + "expected_audits": [ + "audit:id_known:model:customers:id", + "audit:id_known:model:orders:id", + "audit:not_null:model:customers:id", + "audit:not_null:model:orders:id", + "audit:unique:model:orders:id" + ] +} diff --git a/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot.rs b/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot.rs index ff338a5e1f..089dfedbfc 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot.rs +++ b/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot.rs @@ -75,9 +75,12 @@ pub(crate) fn insert_node( key: String, node: SnapshotNode, ) -> PyResult> { - if nodes.insert(key, node).is_some() { - return Err(PyValueError::new_err("duplicate snapshot node identity")); + if nodes.contains_key(&key) { + return Err(PyValueError::new_err(format!( + "duplicate snapshot node identity {key}" + ))); } + nodes.insert(key, node); Ok(nodes) } diff --git a/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot_resources.rs b/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot_resources.rs index 213bcc9739..959edffc0e 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot_resources.rs +++ b/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot_resources.rs @@ -4,6 +4,7 @@ use std::collections::BTreeMap; use pyo3::prelude::{PyAnyMethods, PyResult, Python}; use serde_json::{Value, json}; +use sqlbuild_core::nested_progress::main::audit_resource_id::audit_resource_id; use sqlbuild_orchestration::snapshots::models::SnapshotNode; use crate::bindings::_helpers::compiled_project::project::NativeCompiledProject; @@ -101,9 +102,17 @@ pub(crate) fn resource_nodes( let target: Option = attached.target.as_ref().map( |target| json!({"kind": target.kind, "name": target.name, "column": target.column}), ); + let key = match &attached.target { + Some(target) => audit_resource_id( + attached.compiled_name(), + &target.kind, + (Some(&target.name), target.column.as_deref()), + ), + None => format!("audit:{}", attached.compiled_name()), + }; nodes = insert_node( nodes, - format!("audit:{}", attached.compiled_name()), + key, SnapshotNode { kind: "audit".into(), definition: json!({ diff --git a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/_test_types.py b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/_test_types.py index c2cb0e5bcd..c90b11f1d8 100644 --- a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/_test_types.py +++ b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/_test_types.py @@ -313,6 +313,15 @@ class SchedulerCrashTestCase: expected_minimum_kills: int +@dataclass(frozen=True) +class SharedAuditPublicationTestCase: + description: str + create_project: Callable[[Path], None] + target: str + expected_audits: frozenset[str] + expected_count: int + + @dataclass(frozen=True) class ProjectRolesTestCase: description: str diff --git a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/helpers.py b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/helpers.py index b3b7be5ff5..8d93e83735 100644 --- a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/helpers.py +++ b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/helpers.py @@ -1221,7 +1221,7 @@ def scheduler_crash_history(directory: Path, fixture: dict[str, Any]) -> dict[st def publish_generation( - project: Path, api: SnapshotApi, generation: int + project: Path, api: SnapshotApi, generation: int, target: str = "development" ) -> subprocess.CompletedProcess[str]: """Publish one generation of a project through the installed location command.""" return subprocess.run( @@ -1234,7 +1234,7 @@ def publish_generation( "--location-id", api.location_id, "--target", - "development", + target, "--server", api.url, "--generation", @@ -1326,3 +1326,29 @@ def request_job_run(worker: ScheduledWorker, job: str) -> dict[str, Any]: ) assert requested.returncode == 0, requested.stderr return wait_for_run(worker.api, worker.process, worker.log, run_id=run_id) + + +SHARED_AUDITS: Path = Path("benchmarks/duckshop/edge_cases/orchestration/shared_inline_audits.json") + + +def create_playground(project: Path) -> None: + """Create the default playground project with the installed CLI.""" + created: subprocess.CompletedProcess[str] = subprocess.run( + [str(Path(sys.executable).with_name("sqb")), "playground", str(project)], + capture_output=True, + text=True, + timeout=60, + check=False, + ) + assert created.returncode == 0, created.stderr + + +def write_shared_audits(project: Path) -> None: + """Write the DuckShop project whose models share inline and named audits.""" + write_files(project_dir=project, files=json.loads(SHARED_AUDITS.read_text())["files"]) + + +def audit_node_keys(api: SnapshotApi, target: str) -> frozenset[str]: + """Keys of the current snapshot's audit nodes.""" + nodes: dict[str, Any] = current_snapshot(api, target)["nodes"] + return frozenset(filter(lambda key: nodes[key]["kind"] == "audit", nodes)) diff --git a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_shared_audits.py b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_shared_audits.py new file mode 100644 index 0000000000..31c2f4166d --- /dev/null +++ b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_shared_audits.py @@ -0,0 +1,69 @@ +"""Audits sharing a name across attachments publish as one snapshot node per attachment.""" + +import json +import subprocess +from pathlib import Path + +import pytest + +from tests.integration.src.sqlbuild.compiler.compiled_project.publication._test_types import ( + SharedAuditPublicationTestCase, +) +from tests.integration.src.sqlbuild.compiler.compiled_project.publication.helpers import ( + SHARED_AUDITS, + audit_node_keys, + create_playground, + current_snapshot, + publish_generation, + snapshot_api, + write_shared_audits, +) + + +@pytest.mark.parametrize( + "test_case", + [ + SharedAuditPublicationTestCase( + description="the default playground publishes every inline audit attachment", + create_project=create_playground, + target="dev", + expected_audits=frozenset( + { + "audit:not_null:model:stg_customers:customer_id", + "audit:not_null:model:stg_orders:order_id", + "audit:unique:model:stg_customers:customer_id", + "audit:unique:seed:waffle_types:waffle_type_id", + "audit:revenue_is_non_negative:model:daily_revenue", + } + ), + expected_count=30, + ), + SharedAuditPublicationTestCase( + description="DuckShop shared inline and named audits publish once per attachment", + create_project=write_shared_audits, + target="development", + expected_audits=frozenset(json.loads(SHARED_AUDITS.read_text())["expected_audits"]), + expected_count=5, + ), + ], + ids=lambda case: case.description, +) +def test_given_audits_sharing_a_name_when_published_then_each_attachment_is_a_stable_node( + test_case: SharedAuditPublicationTestCase, tmp_path: Path +) -> None: + project: Path = tmp_path / "project" + test_case.create_project(project) + with snapshot_api(tmp_path / "control") as api: + first: subprocess.CompletedProcess[str] = publish_generation( + project, api, 1, test_case.target + ) + assert first.returncode == 0, first.stderr + audits: frozenset[str] = audit_node_keys(api, test_case.target) + assert test_case.expected_audits <= audits + assert len(audits) == test_case.expected_count + second: subprocess.CompletedProcess[str] = publish_generation( + project, api, 2, test_case.target + ) + assert second.returncode == 0, second.stderr + assert audit_node_keys(api, test_case.target) == audits + assert current_snapshot(api, test_case.target)["changes"] == []