From d411fdb4a990aacfa58a329c914787a1c7080b85 Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Mon, 9 Nov 2026 13:12:09 +0000 Subject: [PATCH 1/4] feat(orchestration): jobs, groups, sensors and calendars join the global unique-name check Publication now rejects a job, job group, sensor or calendar schedule whose name is already used by any other project resource or orchestration declaration, with the same D007 conflict discovery raises for models, sources and the rest. The narrower duplicate-job and duplicate-trigger checks are removed; the object-identity check stays. build_orchestration_snapshot and compile_snapshot take the discovered project inputs so publication can build the full name set. Fixtures whose jobs reused model names are renamed, and DuckShop gains a name-collision edge case. --- .../orchestration/calendar_orders.py | 2 +- .../orchestration/conflicting_settings.py | 2 +- .../orchestration/invalid_settings.py | 2 +- .../orchestration/name_collision.py | 5 + .../scheduler_crash_recovery.json | 6 +- .../_helpers/resource_names.rs | 9 +- .../src/project_validation/main/mod.rs | 1 + .../main/validate_orchestration_names.rs | 9 ++ .../src/project_validation/models.rs | 10 ++ .../src/project_validation/tests.rs | 2 + .../src/project_validation/tests/helpers.rs | 27 ++++- .../test_validate_orchestration_names.rs | 85 +++++++++++++ .../compiled_project/snapshot_declarations.rs | 16 +-- .../_helpers/discovery/project_facts.rs | 25 ++++ .../_helpers/discovery/project_validation.rs | 14 ++- .../_helpers/location_command/command.rs | 19 +-- .../_helpers/location_command/snapshot.rs | 63 ++++++++-- .../_helpers/native_host/constants.rs | 8 ++ .../native_host/orchestration_declarations.rs | 76 +++++++++++- .../native_host/orchestration_sensors.rs | 2 +- .../native_host/orchestration_triggers.rs | 10 -- deploy/project/orchestration/orders.py | 2 +- src/sqlbuild/_native.pyi | 1 + .../main/compile_snapshot.py | 13 +- .../commands/main/test/test_select_by_name.py | 2 +- .../publication/test_backfill.py | 2 +- .../publication/test_calendar.py | 4 +- .../publication/test_publication.py | 14 ++- .../publication/test_rerun.py | 2 +- .../publication/test_warehouse_metadata.py | 5 +- .../publication/test_worker.py | 21 ++-- .../test_orchestration_snapshot.py | 114 ++++++++++++++---- .../sqlbuild/compiler/discovery/test_main.py | 2 +- .../docs/concepts/python-nodes/overview.mdx | 2 +- 34 files changed, 474 insertions(+), 103 deletions(-) create mode 100644 benchmarks/duckshop/edge_cases/orchestration/name_collision.py create mode 100644 crates/sqlbuild-discovery/src/project_validation/main/validate_orchestration_names.rs create mode 100644 crates/sqlbuild-discovery/src/project_validation/tests/test_validate_orchestration_names.rs diff --git a/benchmarks/duckshop/edge_cases/orchestration/calendar_orders.py b/benchmarks/duckshop/edge_cases/orchestration/calendar_orders.py index 644b92c5d6..80d336fbbb 100644 --- a/benchmarks/duckshop/edge_cases/orchestration/calendar_orders.py +++ b/benchmarks/duckshop/edge_cases/orchestration/calendar_orders.py @@ -2,7 +2,7 @@ from sqlbuild.orchestration import calendar_schedule, job -orders = job("orders", description="Build order reporting.", select="orders") +orders = job("order_reporting", description="Build order reporting.", select="orders") @calendar_schedule( diff --git a/benchmarks/duckshop/edge_cases/orchestration/conflicting_settings.py b/benchmarks/duckshop/edge_cases/orchestration/conflicting_settings.py index ae5a4c9dec..e864cf753d 100644 --- a/benchmarks/duckshop/edge_cases/orchestration/conflicting_settings.py +++ b/benchmarks/duckshop/edge_cases/orchestration/conflicting_settings.py @@ -3,7 +3,7 @@ from sqlbuild.orchestration import job orders = job( - "orders", + "refresh_orders", description="Refresh orders.", select="orders", settings={"load_sources": False, "reload": True}, diff --git a/benchmarks/duckshop/edge_cases/orchestration/invalid_settings.py b/benchmarks/duckshop/edge_cases/orchestration/invalid_settings.py index 248b1b4ce1..621d038079 100644 --- a/benchmarks/duckshop/edge_cases/orchestration/invalid_settings.py +++ b/benchmarks/duckshop/edge_cases/orchestration/invalid_settings.py @@ -3,7 +3,7 @@ from sqlbuild.orchestration import job orders = job( - "orders", + "refresh_orders", description="Refresh orders.", select="orders", settings={"select": "customers"}, diff --git a/benchmarks/duckshop/edge_cases/orchestration/name_collision.py b/benchmarks/duckshop/edge_cases/orchestration/name_collision.py new file mode 100644 index 0000000000..fb0e024e9c --- /dev/null +++ b/benchmarks/duckshop/edge_cases/orchestration/name_collision.py @@ -0,0 +1,5 @@ +"""A job may not reuse a model's name: jobs share the project's one resource namespace.""" + +from sqlbuild.orchestration import job + +orders = job("orders", description="Build orders.", select="orders") diff --git a/benchmarks/duckshop/edge_cases/orchestration/scheduler_crash_recovery.json b/benchmarks/duckshop/edge_cases/orchestration/scheduler_crash_recovery.json index 12eb4f25d5..ab9c620d90 100644 --- a/benchmarks/duckshop/edge_cases/orchestration/scheduler_crash_recovery.json +++ b/benchmarks/duckshop/edge_cases/orchestration/scheduler_crash_recovery.json @@ -1,9 +1,9 @@ { "description": "Three minutely cron jobs start ten hours back with the default latest catch-up. Two installed schedulers catch up while being SIGKILLed at seeded random moments; every lapse of a killed leader's lease is simulated. After a final uninterrupted scheduler, each trigger has ticked its latest due minute exactly once and nothing off schedule, each tick requests or is covered by exactly one run, and every requested run has one request event.", "jobs": [ - "orders", - "customers", - "inventory" + "refresh_orders", + "refresh_customers", + "refresh_inventory" ], "backlog_minutes": 600, "rounds": 16, diff --git a/crates/sqlbuild-discovery/src/project_validation/_helpers/resource_names.rs b/crates/sqlbuild-discovery/src/project_validation/_helpers/resource_names.rs index 7f2032bafa..c9b65c04f6 100644 --- a/crates/sqlbuild-discovery/src/project_validation/_helpers/resource_names.rs +++ b/crates/sqlbuild-discovery/src/project_validation/_helpers/resource_names.rs @@ -1,4 +1,4 @@ -//! Python's `_validate_unique_project_resource_names`: one namespace across resource kinds. +//! One namespace across resource kinds, orchestration declarations included. use std::collections::{HashMap, HashSet}; @@ -16,8 +16,8 @@ pub(crate) fn validate_unique_project_resource_names( return Err(conflict(format!( "Project resource name '{name}' is declared as both {existing_kind} in \ {existing_path} and {kind} in {path}; model, source, seed, function, loader, \ - task, asset, check, provider, hook, unit test, and scenario names must be \ - globally unique" + task, asset, check, provider, hook, unit test, scenario, job, job group, sensor, \ + and calendar schedule names must be globally unique" ))); } seen.insert(name.to_owned(), (kind.to_owned(), path.to_owned())); @@ -69,5 +69,8 @@ pub(crate) fn validate_unique_project_resource_names( declare(name, "unit test", &test.file.path)?; } } + for declaration in &facts.orchestration { + declare(&declaration.name, declaration.kind, &declaration.path)?; + } Ok(()) } diff --git a/crates/sqlbuild-discovery/src/project_validation/main/mod.rs b/crates/sqlbuild-discovery/src/project_validation/main/mod.rs index 2c19ca312a..9eaf797a34 100644 --- a/crates/sqlbuild-discovery/src/project_validation/main/mod.rs +++ b/crates/sqlbuild-discovery/src/project_validation/main/mod.rs @@ -1,2 +1,3 @@ pub mod at_most_one_csv_data_row; +pub mod validate_orchestration_names; pub mod validate_project_inputs; diff --git a/crates/sqlbuild-discovery/src/project_validation/main/validate_orchestration_names.rs b/crates/sqlbuild-discovery/src/project_validation/main/validate_orchestration_names.rs new file mode 100644 index 0000000000..ac3a340b34 --- /dev/null +++ b/crates/sqlbuild-discovery/src/project_validation/main/validate_orchestration_names.rs @@ -0,0 +1,9 @@ +//! Orchestration declarations share the project's one resource namespace. + +use crate::project_validation::_helpers::resource_names::validate_unique_project_resource_names; +use crate::project_validation::models::{ProjectFacts, ValidationFailure}; + +/// Reject the first name a job, group, sensor or calendar schedule shares with anything else. +pub fn validate_orchestration_names(facts: &ProjectFacts) -> Result<(), ValidationFailure> { + validate_unique_project_resource_names(facts) +} diff --git a/crates/sqlbuild-discovery/src/project_validation/models.rs b/crates/sqlbuild-discovery/src/project_validation/models.rs index 58c65bb869..a583cc66ce 100644 --- a/crates/sqlbuild-discovery/src/project_validation/models.rs +++ b/crates/sqlbuild-discovery/src/project_validation/models.rs @@ -178,6 +178,16 @@ pub struct ProjectFacts { pub project_model_paths: Option>, /// Python's `csv.field_size_limit()`, or `None` for its default. pub csv_field_limit: Option, + /// Jobs, job groups, sensors and calendar schedules, known only when a location publishes. + pub orchestration: Vec, +} + +/// A named orchestration declaration: its name, its kind as conflicts word it, and its module. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct OrchestrationNameFact { + pub name: String, + pub kind: &'static str, + pub path: String, } /// Why discovered inputs are rejected: a discovery error, or a seed CSV Python cannot read. diff --git a/crates/sqlbuild-discovery/src/project_validation/tests.rs b/crates/sqlbuild-discovery/src/project_validation/tests.rs index d7aebec514..ba01477136 100644 --- a/crates/sqlbuild-discovery/src/project_validation/tests.rs +++ b/crates/sqlbuild-discovery/src/project_validation/tests.rs @@ -6,5 +6,7 @@ mod csv_header; mod helpers; #[path = "tests/test_types.rs"] mod test_types; +#[path = "tests/test_validate_orchestration_names.rs"] +mod validate_orchestration_names; #[path = "tests/test_validate_project_inputs.rs"] mod validate_project_inputs; diff --git a/crates/sqlbuild-discovery/src/project_validation/tests/helpers.rs b/crates/sqlbuild-discovery/src/project_validation/tests/helpers.rs index dc141fbc95..601bc978ea 100644 --- a/crates/sqlbuild-discovery/src/project_validation/tests/helpers.rs +++ b/crates/sqlbuild-discovery/src/project_validation/tests/helpers.rs @@ -1,9 +1,11 @@ +use crate::models::FailureKind; use crate::project_validation::_helpers::csv_header::csv_header; use crate::project_validation::main::validate_project_inputs::validate_project_inputs; use crate::project_validation::models::{ - CsvSettings, DependencyFact, ModelFact, NamedPath, NodeFact, NodeKind, ProjectFacts, - SchemaFileFact, SeedEntryFact, SeedFileFact, ValidationFailure, + CsvSettings, DependencyFact, ModelFact, NamedPath, NodeFact, NodeKind, OrchestrationNameFact, + ProjectFacts, SchemaFileFact, SeedEntryFact, SeedFileFact, ValidationFailure, }; +use crate::project_validation::tests::test_types::failure; /// The header the reader returns, or its error's message. pub(super) fn read_header( @@ -88,3 +90,24 @@ pub(super) fn path_default_facts(paths: &[&str], keys: &[&str]) -> ProjectFacts ..ProjectFacts::default() } } + +/// An orchestration declaration named `name`, of `kind`, in one module. +pub(super) fn declared(name: &str, kind: &'static str) -> OrchestrationNameFact { + OrchestrationNameFact { + name: name.to_owned(), + kind, + path: "orchestration/daily.py".to_owned(), + } +} + +/// A global unique-name conflict that `message` begins. +pub(super) fn collision(message: &str) -> Result<(), String> { + failure( + FailureKind::Conflict, + &format!( + "{message}; model, source, seed, function, loader, task, asset, check, provider, \ + hook, unit test, scenario, job, job group, sensor, and calendar schedule names must \ + be globally unique" + ), + ) +} diff --git a/crates/sqlbuild-discovery/src/project_validation/tests/test_validate_orchestration_names.rs b/crates/sqlbuild-discovery/src/project_validation/tests/test_validate_orchestration_names.rs new file mode 100644 index 0000000000..beb6d48ffa --- /dev/null +++ b/crates/sqlbuild-discovery/src/project_validation/tests/test_validate_orchestration_names.rs @@ -0,0 +1,85 @@ +use crate::project_validation::main::validate_orchestration_names::validate_orchestration_names; +use crate::project_validation::models::{ + ProjectFacts, TestBlockFact, TestFileFact, ValidationFailure, +}; +use crate::project_validation::tests::helpers::{collision, declared, model, named}; +use crate::project_validation::tests::test_types::ValidationTestCase; + +#[test] +fn given_orchestration_names_when_validating_then_they_share_the_project_namespace() { + let test_cases = [ + ValidationTestCase { + description: "distinct names pass", + facts: ProjectFacts { + models: vec![model("orders")], + orchestration: vec![ + declared("refresh_orders", "job"), + declared("nightly", "job group"), + declared("orders_arrived", "sensor"), + declared("fulfillment_calendar", "calendar schedule"), + ], + ..ProjectFacts::default() + }, + expected_failure: Ok(()), + }, + ValidationTestCase { + description: "a job named like a model", + facts: ProjectFacts { + models: vec![model("orders")], + orchestration: vec![declared("orders", "job")], + ..ProjectFacts::default() + }, + expected_failure: collision( + "Project resource name 'orders' is declared as both model in models/orders.sql \ + and job in orchestration/daily.py", + ), + }, + ValidationTestCase { + description: "a sensor named like a unit test", + facts: ProjectFacts { + tests: vec![TestFileFact { + file: named("orders_arrived", "tests/orders_arrived.sql"), + blocks: vec![TestBlockFact::default()], + }], + orchestration: vec![declared("orders_arrived", "sensor")], + ..ProjectFacts::default() + }, + expected_failure: collision( + "Project resource name 'orders_arrived' is declared as both unit test in \ + tests/orders_arrived.sql and sensor in orchestration/daily.py", + ), + }, + ValidationTestCase { + description: "a calendar schedule named like a provider", + facts: ProjectFacts { + providers: vec![named("fulfillment", "providers/fulfillment.py")], + orchestration: vec![declared("fulfillment", "calendar schedule")], + ..ProjectFacts::default() + }, + expected_failure: collision( + "Project resource name 'fulfillment' is declared as both provider in \ + providers/fulfillment.py and calendar schedule in orchestration/daily.py", + ), + }, + ValidationTestCase { + description: "a job group named like a job", + facts: ProjectFacts { + orchestration: vec![declared("daily", "job"), declared("daily", "job group")], + ..ProjectFacts::default() + }, + expected_failure: collision( + "Project resource name 'daily' is declared as both job in orchestration/daily.py \ + and job group in orchestration/daily.py", + ), + }, + ]; + for test_case in test_cases { + assert_eq!( + validate_orchestration_names(&test_case.facts) + .map_err(|failure: ValidationFailure| failure.to_string()), + test_case.expected_failure, + "{}", + test_case.description + ); + } +} diff --git a/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot_declarations.rs b/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot_declarations.rs index 0c2ac4c08b..882e3e8ba4 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot_declarations.rs +++ b/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot_declarations.rs @@ -31,7 +31,7 @@ use crate::bindings::_helpers::native_host::orchestration_triggers::{ pub(crate) fn discover_declarations<'py>( py: Python<'py>, project_dir: &Path, -) -> PyResult>> { +) -> PyResult, String)>> { let tree = ProjectTree::new(project_dir); let files = py .detach(|| { @@ -44,12 +44,12 @@ pub(crate) fn discover_declarations<'py>( .map_err(|error| { PyValueError::new_err(format!("orchestration discovery failed: {error:?}")) })?; - let mut declarations: Vec> = Vec::new(); + let mut declarations: Vec<(Bound<'py, PyAny>, String)> = Vec::new(); let mut seen: BTreeSet = BTreeSet::new(); for file in files { for job in module_jobs(py, &tree.absolute(&file), project_dir)? { if seen.insert(job.as_ptr() as usize) { - declarations.push(job); + declarations.push((job, file.clone())); } } } @@ -87,11 +87,7 @@ pub(crate) fn compile_declarations( } } let mut triggers: BTreeMap = BTreeMap::new(); - let mut names: BTreeSet = BTreeSet::new(); for job in authored_jobs { - if !names.insert(job.name.clone()) { - return Err(PyValueError::new_err(format!("duplicate job {}", job.name))); - } if !job.targets.is_empty() && !job.targets.contains(&input.target) { continue; } @@ -140,12 +136,6 @@ pub(crate) fn compile_declarations( } let mut groups: BTreeMap = BTreeMap::new(); for group in authored_groups { - if !names.insert(group.name.clone()) { - return Err(PyValueError::new_err(format!( - "duplicate job or group {}", - group.name - ))); - } if !group.targets.is_empty() && !group.targets.contains(&input.target) { continue; } diff --git a/crates/sqlbuild-python/src/bindings/_helpers/discovery/project_facts.rs b/crates/sqlbuild-python/src/bindings/_helpers/discovery/project_facts.rs index 5bfcaeb0d1..4e7d9f6e52 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/discovery/project_facts.rs +++ b/crates/sqlbuild-python/src/bindings/_helpers/discovery/project_facts.rs @@ -17,6 +17,13 @@ use sqlbuild_discovery::project_validation::models::{ const CSV_MODULE: &str = "csv"; /// `DeclarationKind.AUDIT`: a generic audit file. const GENERIC_AUDIT_KIND: &str = "audit"; +/// Each Python node kind and the discovered inputs field that lists it, in discovery order. +const PYTHON_NODE_FIELDS: [(&str, &str); 4] = [ + ("loader", "loader_functions"), + ("task", "task_functions"), + ("asset", "asset_functions"), + ("check", "check_functions"), +]; /// `ScopeKind.PRIVATE`. const PRIVATE_SCOPE: &str = "private"; @@ -72,9 +79,27 @@ pub(crate) fn project_facts( .collect::>()?, project_model_paths: injection().map(|injection| injection.project_model_paths.clone()), csv_field_limit: Some(csv_field_limit(py)?), + orchestration: Vec::new(), }) } +/// The discovered Python nodes by kind, name and path only, as the resource-name check reads them. +pub(crate) fn named_node_rows(inputs: &Bound<'_, PyAny>) -> PyResult> { + let mut rows: Vec = Vec::new(); + for (kind, name) in PYTHON_NODE_FIELDS { + rows.extend(each(inputs, name, |node| { + Ok(NodeRow { + kind: kind.to_owned(), + name: native_field(node, "name")?.extract()?, + path: text(node, "relative_path")?, + decorator_config: false, + depends_on: Vec::new(), + }) + })?); + } + Ok(rows) +} + /// `read` applied to each item of the iterable attribute `name` of `owner`. pub(crate) fn each( owner: &Bound<'_, PyAny>, diff --git a/crates/sqlbuild-python/src/bindings/_helpers/discovery/project_validation.rs b/crates/sqlbuild-python/src/bindings/_helpers/discovery/project_validation.rs index 92d7a38846..b3b1a096e7 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/discovery/project_validation.rs +++ b/crates/sqlbuild-python/src/bindings/_helpers/discovery/project_validation.rs @@ -33,10 +33,12 @@ fn validate_discovered_project_inputs( nodes: Vec, ) -> Result<(), PyErr> { let facts: ProjectFacts = project_facts(py, &inputs, nodes)?; - let Err(failure) = validate_project_inputs(&facts) else { - return Ok(()); - }; - Err(match failure { + validate_project_inputs(&facts).map_err(|failure| validation_error(py, failure)) +} + +/// The Python exception a failed project validation raises. +pub(crate) fn validation_error(py: Python<'_>, failure: ValidationFailure) -> PyErr { + match failure { ValidationFailure::Discovery(failure) => failure_error(py, failure), ValidationFailure::Unreadable { absolute_path, @@ -56,7 +58,7 @@ fn validate_discovered_project_inputs( ), ValidationFailure::CsvDialect(error) => { if !CSV_DIALECT_ERRORS.contains(&error.class.as_str()) { - return Err(PyKeyError::new_err(error.class.to_string())); + return PyKeyError::new_err(error.class.to_string()); } raised( py.import(BUILTINS_MODULE) @@ -64,7 +66,7 @@ fn validate_discovered_project_inputs( .and_then(|class| class.call1((error.message,))), ) } - }) + } } pub(crate) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> { diff --git a/crates/sqlbuild-python/src/bindings/_helpers/location_command/command.rs b/crates/sqlbuild-python/src/bindings/_helpers/location_command/command.rs index ccb2e600ba..92cc25b235 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/location_command/command.rs +++ b/crates/sqlbuild-python/src/bindings/_helpers/location_command/command.rs @@ -22,7 +22,9 @@ use crate::bindings::_helpers::compiled_project::snapshot::NativeOrchestrationSn use crate::bindings::_helpers::compiled_views::project_view::NativeProjectView; use crate::bindings::_helpers::location_command::constants::{EXECUTE_ACTION, PUBLISH_ACTION}; use crate::bindings::_helpers::location_command::runtime::run_runtime; -use crate::bindings::_helpers::location_command::snapshot::build_orchestration_snapshot; +use crate::bindings::_helpers::location_command::snapshot::{ + build_orchestration_snapshot, orchestration_snapshot, +}; use crate::bindings::_helpers::location_command::worker::execute_worker; use crate::bindings::_helpers::native_host::host::write_console; use crate::bindings::_helpers::query_command::project::{ @@ -136,15 +138,18 @@ fn compile_location<'py>( )?; let project = field(&compiled, "project")?; let view = project.cast::()?.borrow(); - let snapshot = build_orchestration_snapshot( + let snapshot = orchestration_snapshot( view, &serde_json::to_string(input).map_err(|error| PyValueError::new_err(error.to_string()))?, project_dir.extract()?, - arguments - .value("action")? - .extract::>()? - .is_some_and(|action| action == EXECUTE_ACTION) - .then(Vec::new), + ( + discovered.clone(), + arguments + .value("action")? + .extract::>()? + .is_some_and(|action| action == EXECUTE_ACTION) + .then(Vec::new), + ), )?; status(py, "Location snapshot compiled.")?; Ok((snapshot, discovered)) diff --git a/crates/sqlbuild-python/src/bindings/_helpers/location_command/snapshot.rs b/crates/sqlbuild-python/src/bindings/_helpers/location_command/snapshot.rs index 6e2353b734..a56c2c2e99 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/location_command/snapshot.rs +++ b/crates/sqlbuild-python/src/bindings/_helpers/location_command/snapshot.rs @@ -9,8 +9,10 @@ use pyo3::types::{PyDict, PyDictMethods}; use pyo3::{IntoPyObject, pyfunction}; use serde_json::json; use sqlbuild_core::models::DiagnosticSeverity; +use sqlbuild_discovery::project_validation::main::validate_orchestration_names::validate_orchestration_names; use sqlbuild_orchestration::snapshots::models::{SnapshotInput, SnapshotNode}; +use crate::bindings::_helpers::boundary::python_calls::KeywordArguments; use crate::bindings::_helpers::compiled_project::snapshot::NativeOrchestrationSnapshot; use crate::bindings::_helpers::compiled_project::snapshot_declarations::{ compile_declarations, discover_declarations, @@ -19,21 +21,55 @@ use crate::bindings::_helpers::compiled_project::snapshot_models::model_nodes; use crate::bindings::_helpers::compiled_project::snapshot_resources::resource_nodes; use crate::bindings::_helpers::compiled_project::snapshot_roles::compile_project_roles; use crate::bindings::_helpers::compiled_views::project_view::NativeProjectView; +use crate::bindings::_helpers::discovery::project_facts::{named_node_rows, project_facts}; +use crate::bindings::_helpers::discovery::project_validation::validation_error; use crate::bindings::_helpers::location_command::snapshot_lineage::column_lineage_document; -use crate::bindings::_helpers::native_host::orchestration_declarations::collect_declarations; +use crate::bindings::_helpers::native_host::orchestration_declarations::{ + collect_declarations, orchestration_names, +}; use crate::bindings::_helpers::native_host::orchestration_environment::environment_hash; use crate::bindings::_helpers::planner_identity::identities::direct_model_version_identities; use crate::bindings::_helpers::planner_scope::scope::build_planner_scope; use crate::bindings::_helpers::records::fields::field; +/// Where declarations passed to the snapshot directly are reported as declared. +const DECLARATIONS_ARGUMENT: &str = "declarations argument"; + /// Compile catalog nodes and identities in the location, never in the server. #[pyfunction] -#[pyo3(signature = (project, metadata_json, project_dir, *, declarations=None))] -pub(crate) fn build_orchestration_snapshot( - project: PyRef<'_, NativeProjectView>, +#[pyo3(signature = (project, metadata_json, project_dir, **options))] +pub(crate) fn build_orchestration_snapshot<'py>( + project: PyRef<'py, NativeProjectView>, + metadata_json: &str, + project_dir: PathBuf, + options: Option>, +) -> PyResult { + let arguments = KeywordArguments::read( + project.py(), + "build_orchestration_snapshot", + options, + (&["discovered_inputs"], &["declarations"]), + )?; + let declarations = arguments.value("declarations")?; + orchestration_snapshot( + project, + metadata_json, + project_dir, + ( + arguments.value("discovered_inputs")?, + (!declarations.is_none()) + .then(|| declarations.extract()) + .transpose()?, + ), + ) +} + +/// Compile the snapshot from the project, its discovery, and the declarations passed directly, if any. +pub(crate) fn orchestration_snapshot<'py>( + project: PyRef<'py, NativeProjectView>, metadata_json: &str, project_dir: PathBuf, - declarations: Option>>, + (discovered_inputs, declarations): (Bound<'py, PyAny>, Option>>), ) -> PyResult { let py = project.py(); let mut input: SnapshotInput = serde_json::from_str(metadata_json) @@ -66,12 +102,25 @@ pub(crate) fn build_orchestration_snapshot( input.nodes = resource_nodes(py, &project, &owner, input.nodes)?; drop(owner); let declarations = match declarations { - Some(declarations) => declarations, + Some(declarations) => declarations + .into_iter() + .map(|declaration| (declaration, DECLARATIONS_ARGUMENT.to_owned())) + .collect(), None => discover_declarations(py, &project_dir)?, }; let graph = project.lineage_graph(py)?; if !declarations.is_empty() { - let authored = collect_declarations(py, declarations)?; + let mut facts = + project_facts(py, &discovered_inputs, named_node_rows(&discovered_inputs)?)?; + facts.orchestration = orchestration_names(py, &declarations)?; + validate_orchestration_names(&facts).map_err(|failure| validation_error(py, failure))?; + let authored = collect_declarations( + py, + declarations + .into_iter() + .map(|(declaration, _)| declaration) + .collect(), + )?; input = compile_declarations( py, input, diff --git a/crates/sqlbuild-python/src/bindings/_helpers/native_host/constants.rs b/crates/sqlbuild-python/src/bindings/_helpers/native_host/constants.rs index db99dfcfe9..c5354c4914 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/native_host/constants.rs +++ b/crates/sqlbuild-python/src/bindings/_helpers/native_host/constants.rs @@ -11,3 +11,11 @@ pub(crate) const SOURCE_STALE_CONDITION: &str = "source_stale"; /// The sensor context is supplied by the host, not a provider. pub(crate) const SENSOR_CONTEXT_PARAMETER: &str = "ctx"; + +pub(crate) const AUTHORING_MODULE: &str = "sqlbuild.orchestration_declarations.models.authoring"; + +/// Kinds named in the global unique-name check. +pub(crate) const JOB_LABEL: &str = "job"; +pub(crate) const GROUP_LABEL: &str = "job group"; +pub(crate) const SENSOR_LABEL: &str = "sensor"; +pub(crate) const CALENDAR_LABEL: &str = "calendar schedule"; diff --git a/crates/sqlbuild-python/src/bindings/_helpers/native_host/orchestration_declarations.rs b/crates/sqlbuild-python/src/bindings/_helpers/native_host/orchestration_declarations.rs index 4bfe0de411..87651dcd21 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/native_host/orchestration_declarations.rs +++ b/crates/sqlbuild-python/src/bindings/_helpers/native_host/orchestration_declarations.rs @@ -8,9 +8,12 @@ use pyo3::exceptions::{PyTypeError, PyValueError}; use pyo3::prelude::{Bound, Py, PyAny, PyAnyMethods, PyResult, Python}; use pyo3::types::{PyDict, PyDictMethods, PyList, PyListMethods, PyTuple}; use serde_json::Value; +use sqlbuild_discovery::project_validation::models::OrchestrationNameFact; use sqlbuild_host::entries::models::UserCodeHost; -use crate::bindings::_helpers::native_host::constants::RUN_EVENT_CONDITION; +use crate::bindings::_helpers::native_host::constants::{ + AUTHORING_MODULE, CALENDAR_LABEL, GROUP_LABEL, JOB_LABEL, RUN_EVENT_CONDITION, SENSOR_LABEL, +}; use crate::bindings::_helpers::native_host::host::entry; #[derive(FromPyObject)] @@ -231,3 +234,74 @@ pub(crate) fn collect_declarations<'py>( } Ok(collected) } + +/// Every named declaration reachable from `roots`, labelled with the module of the root that reaches it first. +pub(crate) fn orchestration_names( + py: Python<'_>, + roots: &[(Bound<'_, PyAny>, String)], +) -> PyResult> { + let authoring = py.import(AUTHORING_MODULE)?; + let classes = [ + (authoring.getattr("Job")?, JOB_LABEL), + (authoring.getattr("JobGroup")?, GROUP_LABEL), + (authoring.getattr("Sensor")?, SENSOR_LABEL), + (authoring.getattr("CalendarSchedule")?, CALENDAR_LABEL), + ]; + let mut seen: HashSet = HashSet::new(); + let mut names: Vec = Vec::new(); + for (root, path) in roots { + let mut pending: Vec> = vec![root.clone()]; + while let Some(declaration) = pending.pop() { + if !seen.insert(declaration.as_ptr() as usize) { + continue; + } + let Some(kind) = classes + .iter() + .find(|(class, _)| declaration.is_instance(class).unwrap_or(false)) + .map(|(_, kind)| *kind) + else { + continue; + }; + pending.extend(nested_declarations(&declaration, kind)?); + let name: Option = if kind == CALENDAR_LABEL { + let function = declaration.getattr("function")?; + (!function.is_none()) + .then(|| function.getattr("__name__")?.extract()) + .transpose()? + } else { + Some(declaration.getattr("name")?.extract()?) + }; + names.extend(name.map(|name| OrchestrationNameFact { + name, + kind, + path: path.clone(), + })); + } + } + Ok(names) +} + +/// The declarations one declaration refers to: a group's jobs, a calendar's job, a sensor's job and run-event job. +fn nested_declarations<'py>( + declaration: &Bound<'py, PyAny>, + kind: &str, +) -> PyResult>> { + Ok(match kind { + GROUP_LABEL => declaration + .getattr("jobs")? + .try_iter()? + .collect::>>()?, + CALENDAR_LABEL => vec![declaration.getattr("job")?], + SENSOR_LABEL => { + let condition = declaration.getattr("when")?; + let mut nested = vec![declaration.getattr("run")?]; + if !condition.is_none() + && condition.getattr("kind")?.extract::()? == RUN_EVENT_CONDITION + { + nested.push(condition.getattr("job")?); + } + nested + } + _ => Vec::new(), + }) +} diff --git a/crates/sqlbuild-python/src/bindings/_helpers/native_host/orchestration_sensors.rs b/crates/sqlbuild-python/src/bindings/_helpers/native_host/orchestration_sensors.rs index 434c71c2fc..8a01736e93 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/native_host/orchestration_sensors.rs +++ b/crates/sqlbuild-python/src/bindings/_helpers/native_host/orchestration_sensors.rs @@ -35,7 +35,7 @@ pub(crate) fn evaluate_sensor( let authoring = py.import("sqlbuild.orchestration_declarations.models.authoring")?; let sensor_class = authoring.getattr("Sensor")?; let mut matches: Vec> = Vec::new(); - for declaration in discover_declarations(py, project_dir)? { + for (declaration, _) in discover_declarations(py, project_dir)? { if declaration.is_instance(&sensor_class)? && declaration.getattr("name")?.extract::()? == *name { diff --git a/crates/sqlbuild-python/src/bindings/_helpers/native_host/orchestration_triggers.rs b/crates/sqlbuild-python/src/bindings/_helpers/native_host/orchestration_triggers.rs index 532b5e0723..583ae015e2 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/native_host/orchestration_triggers.rs +++ b/crates/sqlbuild-python/src/bindings/_helpers/native_host/orchestration_triggers.rs @@ -233,9 +233,6 @@ pub(crate) fn compile_calendars( "calendar {name} needs a description and a published job" ))); } - if definitions.triggers.contains_key(&name) { - return Err(PyValueError::new_err(format!("duplicate trigger {name}"))); - } let schedule = calendar_times(calendar, &session, &by_name, start)?; let mut trigger = compile_trigger(calendar, &job_key, &description, schedule)?; trigger.starts_at.get_or_insert(start); @@ -256,14 +253,7 @@ pub(crate) fn compile_sensors( } else { String::new() }; - let mut names: BTreeSet = BTreeSet::new(); for sensor in authored { - if !names.insert(sensor.name.clone()) || definitions.triggers.contains_key(&sensor.name) { - return Err(PyValueError::new_err(format!( - "duplicate trigger {}", - sensor.name - ))); - } if !sensor.targets.is_empty() && !sensor.targets.iter().any(|value| value == &input.target) { continue; diff --git a/deploy/project/orchestration/orders.py b/deploy/project/orchestration/orders.py index 47aeb4c1f3..8023794654 100644 --- a/deploy/project/orchestration/orders.py +++ b/deploy/project/orchestration/orders.py @@ -3,7 +3,7 @@ from sqlbuild.orchestration import job, schedule orders = job( - "orders", + "refresh_orders", select="orders", description="Refresh orders every minute.", schedule=schedule(cron="* * * * *", timezone="UTC"), diff --git a/src/sqlbuild/_native.pyi b/src/sqlbuild/_native.pyi index d9dfe2d850..0e5f0ccaf2 100644 --- a/src/sqlbuild/_native.pyi +++ b/src/sqlbuild/_native.pyi @@ -1768,6 +1768,7 @@ def build_orchestration_snapshot( metadata_json: str, project_dir: Path, *, + discovered_inputs: DiscoveredProjectInputs, declarations: Sequence[object] | None = None, ) -> NativeOrchestrationSnapshot: ... def run_location_command( diff --git a/src/sqlbuild/orchestration_declarations/main/compile_snapshot.py b/src/sqlbuild/orchestration_declarations/main/compile_snapshot.py index 0f9fc8b052..fb42de5d94 100644 --- a/src/sqlbuild/orchestration_declarations/main/compile_snapshot.py +++ b/src/sqlbuild/orchestration_declarations/main/compile_snapshot.py @@ -5,7 +5,11 @@ from collections.abc import Sequence from pathlib import Path -from sqlbuild._native import NativeOrchestrationSnapshot, build_orchestration_snapshot +from sqlbuild._native import ( + DiscoveredProjectInputs, + NativeOrchestrationSnapshot, + build_orchestration_snapshot, +) from sqlbuild.compiler.compile.types import CompiledProjectView from sqlbuild.orchestration_declarations.models.authoring import ( CalendarSchedule, @@ -20,9 +24,14 @@ def compile_snapshot( project: CompiledProjectView, metadata_json: str, project_dir: Path, + discovered_inputs: DiscoveredProjectInputs, declarations: Sequence[Job | JobGroup | CalendarSchedule | Sensor] | None = None, ) -> NativeOrchestrationSnapshot: """Compile authoring records and retained project facts into a data-only snapshot.""" return build_orchestration_snapshot( - project, metadata_json, project_dir, declarations=declarations + project, + metadata_json, + project_dir, + discovered_inputs=discovered_inputs, + declarations=declarations, ) diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/test/test_select_by_name.py b/tests/e2e/src/sqlbuild/cli/commands/main/test/test_select_by_name.py index baf6b94d53..cecb5fdc91 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/test/test_select_by_name.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/test/test_select_by_name.py @@ -152,7 +152,7 @@ def test_given_test_names_when_selecting_then_exactly_those_tests_run( expected_output_fragments=( "Project resource name 'orders' is declared as both model in models/orders.sql " "and unit test in tests/unit/orders.sql", - "unit test, and scenario names must be globally unique", + "and calendar schedule names must be globally unique", ), ), ), diff --git a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_backfill.py b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_backfill.py index 9d54256fc6..512db91881 100644 --- a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_backfill.py +++ b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_backfill.py @@ -75,7 +75,7 @@ def test_given_explicit_backfill_bounds_when_running_and_rerunning_a_group_then_ "sources/raw_orders.yml": "sources:\n - name: raw_orders\n description: Synthetic order windows.\n schema: main\n table: raw_orders\n", "orchestration/orders.py": ( "from sqlbuild.orchestration import job, job_group\n" - "orders = job('orders', description='Windowed orders.', select='orders', " + "orders = job('windowed_orders', description='Windowed orders.', select='orders', " "settings={'start_cursor': '0', 'end_cursor': '1'})\n" "annual_orders = job_group('annual_orders', description='Windowed group.', jobs=[orders])\n" ), diff --git a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_calendar.py b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_calendar.py index 9f3a98f241..f7d26b84fa 100644 --- a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_calendar.py +++ b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_calendar.py @@ -74,7 +74,7 @@ def test_given_invalid_calendar_results_when_publishing_through_the_cli_then_val "orchestration/orders.py": ( "from datetime import datetime, timedelta\nfrom itertools import repeat\n" "from sqlbuild.orchestration import calendar_schedule, job\n" - "orders = job('orders', description='Order reporting.', select='orders')\n" + "orders = job('order_reporting', description='Order reporting.', select='orders')\n" f"@calendar_schedule(job=orders, description='Delivery times.', horizon={test_case.horizon!r}, timezone={test_case.timezone!r})\n" f"def deliveries(window):\n return {test_case.expression}\n" ), @@ -275,7 +275,7 @@ def test_given_a_published_calendar_when_a_real_worker_compiles_then_it_does_not | { "orchestration/orders.py": ( "from pathlib import Path\nfrom sqlbuild.orchestration import calendar_schedule, job, schedule\n" - "orders = job('orders', description='Order reporting.', select='orders', schedule=schedule(cron='0 0 * * *', timezone='UTC', starts_at='2000-01-01T00:00:00Z'))\n" + "orders = job('order_reporting', description='Order reporting.', select='orders', schedule=schedule(cron='0 0 * * *', timezone='UTC', starts_at='2000-01-01T00:00:00Z'))\n" "@calendar_schedule(job=orders, description='Upcoming deliveries.', horizon='1d', timezone='UTC')\n" "def deliveries(window):\n" f" marker = Path({str(evaluated)!r})\n" diff --git a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_publication.py b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_publication.py index a0a60c70c7..49b4a75a47 100644 --- a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_publication.py +++ b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_publication.py @@ -13,6 +13,7 @@ from _pytest.capture import CaptureResult from sqlbuild._native import ( + DiscoveredProjectInputs, NativeOrchestrationSnapshot, build_compiled_project, build_orchestration_snapshot, @@ -67,8 +68,9 @@ def test_given_retained_compiler_data_when_publishing_then_http_updates_atomical ), } write_files(project_dir=project_dir, files=files) + discovered: DiscoveredProjectInputs = discover_project_inputs(project_dir=project_dir) project: CompiledProjectView = build_compiled_project( - discovered_inputs=discover_project_inputs(project_dir=project_dir), + discovered_inputs=discovered, adapter=DuckDbAdapter(), resolved_connection={"database": "warehouse", "schema": "main"}, ) @@ -79,7 +81,7 @@ def test_given_retained_compiler_data_when_publishing_then_http_updates_atomical "target": test_case.target, } first: NativeOrchestrationSnapshot = build_orchestration_snapshot( - project, json.dumps(metadata), project_dir + project, json.dumps(metadata), project_dir, discovered_inputs=discovered ) record: dict[str, Any] = json.loads(first.publish(api.url, api.bearer)) assert record["generation"] == 1 @@ -93,13 +95,17 @@ def test_given_retained_compiler_data_when_publishing_then_http_updates_atomical ) } write_files(project_dir=project_dir, files=edited) + discovered = discover_project_inputs(project_dir=project_dir) changed: CompiledProjectView = build_compiled_project( - discovered_inputs=discover_project_inputs(project_dir=project_dir), + discovered_inputs=discovered, adapter=DuckDbAdapter(), resolved_connection={"database": "warehouse", "schema": "main"}, ) next_snapshot: NativeOrchestrationSnapshot = build_orchestration_snapshot( - changed, json.dumps(metadata | {"generation": 2}), project_dir + changed, + json.dumps(metadata | {"generation": 2}), + project_dir, + discovered_inputs=discovered, ) updated: dict[str, Any] = json.loads(next_snapshot.publish(api.url, api.bearer)) assert updated["generation"] == test_case.expected_generation diff --git a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_rerun.py b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_rerun.py index 1d1193a3a9..4682b6bb22 100644 --- a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_rerun.py +++ b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_rerun.py @@ -63,7 +63,7 @@ def test_given_a_partial_group_when_retrying_from_failure_then_successful_nodes_ "from sqlbuild.orchestration import job, job_group, schedule\n" "staging = job('staging', description='Stage orders.', select='orders broken_orders')\n" "marts = job('marts', description='Build totals.', select='totals')\n" - "inventory = job('inventory', description='Build inventory.', select='inventory')\n" + "inventory = job('stock', description='Build inventory.', select='inventory')\n" "annual_orders = job_group('annual_orders', description='Retry boundaries.', " "jobs=[inventory, marts, staging], schedule=schedule(cron='0 0 * * *', " "timezone='UTC', starts_at='2000-01-01T00:00:00Z'))\n" diff --git a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_warehouse_metadata.py b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_warehouse_metadata.py index 1ebf21518f..4d466244e5 100644 --- a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_warehouse_metadata.py +++ b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_warehouse_metadata.py @@ -18,6 +18,7 @@ import pytest from sqlbuild._native import ( + DiscoveredProjectInputs, Fingerprint, SourceFreshnessRecord, SourceFreshnessRenderers, @@ -670,8 +671,9 @@ def test_given_external_warehouse_retention_when_a_complete_scan_arrives_then_mi adapter.close(connection) declarations: Path = tmp_path / "declarations" write_files(project_dir=declarations, files=TYPE_PROOF_FILES) + discovered: DiscoveredProjectInputs = discover_project_inputs(project_dir=declarations) compiled: CompiledProjectView = build_compiled_project( - discovered_inputs=discover_project_inputs(project_dir=declarations), + discovered_inputs=discovered, adapter=DuckDbAdapter(), resolved_connection={"database": "warehouse", "schema": "main"}, ) @@ -680,6 +682,7 @@ def test_given_external_warehouse_retention_when_a_complete_scan_arrives_then_mi compiled, json.dumps(METADATA | {"location_id": api.location_id, "target": "development"}), declarations, + discovered_inputs=discovered, ).publish(api.url, api.bearer) session_id: str = str(uuid4()) heartbeat: Request = Request( 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 ae32ee21e4..0fb2cbabee 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 @@ -16,6 +16,7 @@ import pytest from sqlbuild._native import ( + DiscoveredProjectInputs, NativeOrchestrationSnapshot, build_compiled_project, build_orchestration_snapshot, @@ -286,15 +287,16 @@ def test_given_a_retained_baseline_when_starting_a_worker_then_verification_and_ ) -> None: project_dir: Path = tmp_path / "project" write_files(project_dir=project_dir, files=TYPE_PROOF_FILES) + discovered: DiscoveredProjectInputs = discover_project_inputs(project_dir=project_dir) project: CompiledProjectView = build_compiled_project( - discovered_inputs=discover_project_inputs(project_dir=project_dir), + discovered_inputs=discovered, adapter=DuckDbAdapter(), resolved_connection={"database": "warehouse", "schema": "main"}, ) with snapshot_api(tmp_path / "control") as api: metadata: dict[str, Any] = METADATA | {"location_id": api.location_id} baseline: NativeOrchestrationSnapshot = build_orchestration_snapshot( - project, json.dumps(metadata), project_dir + project, json.dumps(metadata), project_dir, discovered_inputs=discovered ) publication: dict[str, Any] = json.loads(baseline.publish(api.url, api.bearer)) claim: ClaimedWorker = claimed_worker(tmp_path / "control", api, publication) @@ -305,13 +307,14 @@ def test_given_a_retained_baseline_when_starting_a_worker_then_verification_and_ ) } write_files(project_dir=project_dir, files=changed) + discovered = discover_project_inputs(project_dir=project_dir) drifted: CompiledProjectView = build_compiled_project( - discovered_inputs=discover_project_inputs(project_dir=project_dir), + discovered_inputs=discovered, adapter=DuckDbAdapter(), resolved_connection={"database": "warehouse", "schema": "main"}, ) wrong: NativeOrchestrationSnapshot = build_orchestration_snapshot( - drifted, json.dumps(metadata), project_dir + drifted, json.dumps(metadata), project_dir, discovered_inputs=discovered ) with pytest.raises(RuntimeError, match=test_case.expected_error): _ = wrong.start_worker(api.url, claim.bearer, identity) @@ -846,8 +849,8 @@ def test_given_an_executing_scheduled_worker_when_cancelled_then_the_statement_s ), "orchestration/orders.py": ( "from sqlbuild.orchestration import job, job_group, schedule\n" - "orders = job('orders', description='Long query.', select='orders')\n" - "totals = job('totals', description='Waiting query.', select='totals')\n" + "orders = job('long_orders', description='Long query.', select='orders')\n" + "totals = job('waiting_totals', description='Waiting query.', select='totals')\n" "annual_orders = job_group('annual_orders', description='Cancellable group.', " "jobs=[totals, orders], schedule=schedule(cron='0 0 * * *', timezone='UTC', " "starts_at='2000-01-01T00:00:00Z'))\n" @@ -869,7 +872,7 @@ def test_given_an_executing_group_when_parent_cancelled_then_all_children_stop_b job_key="annual_orders", ) assert parent["status"] == "running", worker.log.read_text() - cancel_running_statement(worker, job_key="orders", run_id=parent["run_id"]) + cancel_running_statement(worker, job_key="long_orders", run_id=parent["run_id"]) completed: dict[str, Any] = wait_for_run( worker.api, worker.process, worker.log, job_key="annual_orders" ) @@ -879,8 +882,8 @@ def test_given_an_executing_group_when_parent_cancelled_then_all_children_stop_b "SELECT job_key, status FROM sqb_runs ORDER BY job_key" ).fetchall() == [ ("annual_orders", "cancelled"), - ("orders", "cancelled"), - ("totals", "cancelled"), + ("long_orders", "cancelled"), + ("waiting_totals", "cancelled"), ] assert control.execute( "SELECT count(*) FROM sqb_runs WHERE parent_run_id=? AND ended_at>?", diff --git a/tests/integration/src/sqlbuild/compiler/compiled_project/test_orchestration_snapshot.py b/tests/integration/src/sqlbuild/compiler/compiled_project/test_orchestration_snapshot.py index 9824426203..972081f9a0 100644 --- a/tests/integration/src/sqlbuild/compiler/compiled_project/test_orchestration_snapshot.py +++ b/tests/integration/src/sqlbuild/compiler/compiled_project/test_orchestration_snapshot.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import re from dataclasses import replace from pathlib import Path from typing import Any @@ -10,6 +11,8 @@ import pytest from sqlbuild._native import ( + DiscoveredProjectInputs, + DiscoveryConflictError, NativeOrchestrationSnapshot, PythonNodeDiscoveryError, build_compiled_project, @@ -47,6 +50,11 @@ ) from tests.integration.src.sqlbuild.compiler.graph.helpers import write_files +GLOBALLY_UNIQUE: str = ( + "; model, source, seed, function, loader, task, asset, check, provider, hook, unit test, " + "scenario, job, job group, sensor, and calendar schedule names must be globally unique" +) + @pytest.mark.parametrize( "test_case", @@ -172,14 +180,15 @@ def test_given_real_projects_when_building_snapshots_then_catalogs_are_stable_an metadata: str = json.dumps(METADATA) for directory in (tmp_path / "original", tmp_path / "relocated"): write_files(project_dir=directory, files=test_case.files) + discovered: DiscoveredProjectInputs = discover_project_inputs(project_dir=directory) project: CompiledProjectView = build_compiled_project( - discovered_inputs=discover_project_inputs(project_dir=directory), + discovered_inputs=discovered, adapter=DuckDbAdapter(), resolved_connection={"database": "warehouse", "schema": "main"}, ) assert not any(item.severity == "error" for item in project.diagnostics) snapshot: NativeOrchestrationSnapshot = build_orchestration_snapshot( - project, metadata, directory + project, metadata, directory, discovered_inputs=discovered ) publication: dict[str, Any] = json.loads(snapshot.to_json()) publications.append(publication) @@ -251,14 +260,17 @@ def test_given_semantic_edits_when_publishing_then_local_identities_keep_planner for label, inputs in (("original", files), ("changed", changed)): directory: Path = tmp_path / label write_files(project_dir=directory, files=inputs) + discovered: DiscoveredProjectInputs = discover_project_inputs(project_dir=directory) project: CompiledProjectView = build_compiled_project( - discovered_inputs=discover_project_inputs(project_dir=directory), + discovered_inputs=discovered, adapter=DuckDbAdapter(), resolved_connection={"database": "warehouse", "schema": "main"}, ) assert not any(item.severity == "error" for item in project.diagnostics) snapshot: dict[str, Any] = json.loads( - build_orchestration_snapshot(project, json.dumps(METADATA), directory).to_json() + build_orchestration_snapshot( + project, json.dumps(METADATA), directory, discovered_inputs=discovered + ).to_json() ) hashes: dict[str, str] = {} for key in ( @@ -310,8 +322,9 @@ def test_given_invalid_compilation_or_forged_nodes_when_publishing_then_no_snaps tmp_path: Path, ) -> None: write_files(project_dir=tmp_path, files=test_case.files) + discovered: DiscoveredProjectInputs = discover_project_inputs(project_dir=tmp_path) project: CompiledProjectView = build_compiled_project( - discovered_inputs=discover_project_inputs(project_dir=tmp_path), + discovered_inputs=discovered, adapter=DuckDbAdapter(), resolved_connection={"database": "warehouse", "schema": "main"}, ) @@ -324,6 +337,7 @@ def test_given_invalid_compilation_or_forged_nodes_when_publishing_then_no_snaps project, json.dumps(METADATA | {"nodes": test_case.nodes}), tmp_path, + discovered_inputs=discovered, ) @@ -365,13 +379,16 @@ def test_given_a_compiled_project_when_publishing_then_its_column_lineage_is_a_s tmp_path: Path, ) -> None: write_files(project_dir=tmp_path, files=test_case.files) + discovered: DiscoveredProjectInputs = discover_project_inputs(project_dir=tmp_path) project: CompiledProjectView = build_compiled_project( - discovered_inputs=discover_project_inputs(project_dir=tmp_path), + discovered_inputs=discovered, adapter=DuckDbAdapter(), resolved_connection={"database": "warehouse", "schema": "main"}, ) publication: dict[str, Any] = json.loads( - build_orchestration_snapshot(project, json.dumps(METADATA), tmp_path).to_json() + build_orchestration_snapshot( + project, json.dumps(METADATA), tmp_path, discovered_inputs=discovered + ).to_json() ) edges: list[list[str]] = publication["documents"]["lineage"]["edges"] assert sorted(edges) == test_case.expected_edges @@ -397,21 +414,28 @@ def test_given_a_dependency_lock_when_it_changes_then_environment_identity_chang tmp_path: Path, ) -> None: write_files(project_dir=tmp_path, files=TYPE_PROOF_FILES) + discovered: DiscoveredProjectInputs = discover_project_inputs(project_dir=tmp_path) project: CompiledProjectView = build_compiled_project( - discovered_inputs=discover_project_inputs(project_dir=tmp_path), + discovered_inputs=discovered, adapter=DuckDbAdapter(), resolved_connection={"database": "warehouse", "schema": "main"}, ) installed: dict[str, Any] = json.loads( - build_orchestration_snapshot(project, json.dumps(METADATA), tmp_path).to_json() + build_orchestration_snapshot( + project, json.dumps(METADATA), tmp_path, discovered_inputs=discovered + ).to_json() ) (tmp_path / "uv.lock").write_text(test_case.first_lockfile) locked: dict[str, Any] = json.loads( - build_orchestration_snapshot(project, json.dumps(METADATA), tmp_path).to_json() + build_orchestration_snapshot( + project, json.dumps(METADATA), tmp_path, discovered_inputs=discovered + ).to_json() ) (tmp_path / "uv.lock").write_text(test_case.second_lockfile) changed: dict[str, Any] = json.loads( - build_orchestration_snapshot(project, json.dumps(METADATA), tmp_path).to_json() + build_orchestration_snapshot( + project, json.dumps(METADATA), tmp_path, discovered_inputs=discovered + ).to_json() ) assert ( len( @@ -421,7 +445,9 @@ def test_given_a_dependency_lock_when_it_changes_then_environment_identity_chang ) assert installed["nodes"] == locked["nodes"] == changed["nodes"] assert changed == json.loads( - build_orchestration_snapshot(project, json.dumps(METADATA), tmp_path).to_json() + build_orchestration_snapshot( + project, json.dumps(METADATA), tmp_path, discovered_inputs=discovered + ).to_json() ) @@ -459,8 +485,9 @@ def test_given_authored_jobs_when_compiling_then_native_snapshots_resolve_select tmp_path: Path, test_case: SnapshotScheduleTestCase ) -> None: write_files(project_dir=tmp_path, files=TYPE_PROOF_FILES) + discovered: DiscoveredProjectInputs = discover_project_inputs(project_dir=tmp_path) project: CompiledProjectView = build_compiled_project( - discovered_inputs=discover_project_inputs(project_dir=tmp_path), + discovered_inputs=discovered, adapter=DuckDbAdapter(), resolved_connection={"database": "warehouse", "schema": "main"}, ) @@ -482,6 +509,7 @@ def test_given_authored_jobs_when_compiling_then_native_snapshots_resolve_select metadata_json=json.dumps(METADATA), project_dir=tmp_path, declarations=[nightly, excluded], + discovered_inputs=discovered, ).to_json() ) assert first["documents"]["jobs"] == { @@ -512,6 +540,7 @@ def test_given_authored_jobs_when_compiling_then_native_snapshots_resolve_select metadata_json=json.dumps(METADATA), project_dir=tmp_path, declarations=[edited], + discovered_inputs=discovered, ).to_json() ) assert second["documents"]["triggers"].keys() == first["documents"]["triggers"].keys() @@ -547,7 +576,7 @@ def test_given_authored_jobs_when_compiling_then_native_snapshots_resolve_select expected_message="starts_at", ), SnapshotDeclarationRejectionTestCase( - description="duplicate names", copies=2, expected_message="duplicate job orders" + description="duplicate names", copies=2, expected_message="duplicate job daily_build" ), SnapshotDeclarationRejectionTestCase( description="unknown setting", @@ -607,14 +636,15 @@ def test_given_invalid_declarations_when_compiling_then_native_validation_reject tmp_path: Path, test_case: SnapshotDeclarationRejectionTestCase ) -> None: write_files(project_dir=tmp_path, files=TYPE_PROOF_FILES) + discovered: DiscoveredProjectInputs = discover_project_inputs(project_dir=tmp_path) project: CompiledProjectView = build_compiled_project( - discovered_inputs=discover_project_inputs(project_dir=tmp_path), + discovered_inputs=discovered, adapter=DuckDbAdapter(), resolved_connection={"database": "warehouse", "schema": "main"}, ) declarations: list[Job] = [ job( - "orders", + "daily_build", description=test_case.job_description, select=test_case.select, settings=test_case.settings, @@ -629,6 +659,7 @@ def test_given_invalid_declarations_when_compiling_then_native_validation_reject metadata_json=json.dumps(METADATA), project_dir=tmp_path, declarations=declarations, + discovered_inputs=discovered, ) @@ -921,8 +952,9 @@ def test_given_declaration_modules_when_compiling_then_native_discovery_collects tmp_path: Path, test_case: SnapshotDiscoveryTestCase ) -> None: write_files(project_dir=tmp_path, files=TYPE_PROOF_FILES | test_case.files) + discovered: DiscoveredProjectInputs = discover_project_inputs(project_dir=tmp_path) project: CompiledProjectView = build_compiled_project( - discovered_inputs=discover_project_inputs(project_dir=tmp_path), + discovered_inputs=discovered, adapter=DuckDbAdapter(), resolved_connection={"database": "warehouse", "schema": "main"}, ) @@ -932,6 +964,7 @@ def test_given_declaration_modules_when_compiling_then_native_discovery_collects metadata_json=json.dumps(METADATA), project_dir=tmp_path, declarations=test_case.declarations, + discovered_inputs=discovered, ).to_json() ) assert tuple(publication["documents"].get("jobs", {})) == test_case.expected_jobs @@ -967,7 +1000,7 @@ def test_given_declaration_modules_when_compiling_then_native_discovery_collects files={ "orchestration/orders.py": ( "from sqlbuild.orchestration import job, sensor\n" - "orders = job('orders', description='Build orders.', select='orders')\n" + "orders = job('refresh_orders', description='Build orders.', select='orders')\n" "poll = sensor('poll', description='Poll orders.', run=orders)\n" ), }, @@ -979,7 +1012,7 @@ def test_given_declaration_modules_when_compiling_then_native_discovery_collects files={ "orchestration/orders.py": ( "from sqlbuild.orchestration import job, sensor, platform_event, PlatformEventKind\n" - "orders = job('orders', description='Build orders.', select='orders')\n" + "orders = job('refresh_orders', description='Build orders.', select='orders')\n" "@sensor('poll', description='Poll orders.', run=orders, " "when=platform_event(kinds=[PlatformEventKind.TICK_OVERDUE]))\n" "def poll(ctx):\n return []\n" @@ -1024,8 +1057,39 @@ def test_given_declaration_modules_when_compiling_then_native_discovery_collects "jobs = [job('daily', description='Orders.', select='orders') for _ in range(2)]\n" ), }, - expected_error=ValueError, - expected_message="duplicate job daily", + expected_error=DiscoveryConflictError, + expected_message=re.escape( + "Project resource name 'daily' is declared as both job in orchestration/jobs.py " + "and job in orchestration/jobs.py" + GLOBALLY_UNIQUE + ), + ), + SnapshotDiscoveryRejectionTestCase( + description="DuckShop jobs cannot reuse a model name", + files={ + "orchestration/jobs.py": Path( + "benchmarks/duckshop/edge_cases/orchestration/name_collision.py" + ).read_text() + }, + expected_error=DiscoveryConflictError, + expected_message=re.escape( + "Project resource name 'orders' is declared as both model in models/orders.sql " + "and job in orchestration/jobs.py" + GLOBALLY_UNIQUE + ), + ), + SnapshotDiscoveryRejectionTestCase( + description="a sensor cannot reuse its job's name", + files={ + "orchestration/jobs.py": ( + "from sqlbuild.orchestration import job, sensor\n" + "daily = job('daily', description='Orders.', select='orders')\n" + "poll = sensor('daily', description='Poll orders.', run=daily)\n" + ), + }, + expected_error=DiscoveryConflictError, + expected_message=re.escape( + "Project resource name 'daily' is declared as both sensor in orchestration/jobs.py " + "and job in orchestration/jobs.py" + GLOBALLY_UNIQUE + ), ), SnapshotDiscoveryRejectionTestCase( description="broken imports fail on every retry", @@ -1289,13 +1353,17 @@ def test_given_broken_declaration_modules_when_compiling_then_no_partial_snapsho tmp_path: Path, test_case: SnapshotDiscoveryRejectionTestCase ) -> None: write_files(project_dir=tmp_path, files=TYPE_PROOF_FILES | test_case.files) + discovered: DiscoveredProjectInputs = discover_project_inputs(project_dir=tmp_path) project: CompiledProjectView = build_compiled_project( - discovered_inputs=discover_project_inputs(project_dir=tmp_path), + discovered_inputs=discovered, adapter=DuckDbAdapter(), resolved_connection={"database": "warehouse", "schema": "main"}, ) for _ in range(2): with pytest.raises(test_case.expected_error, match=test_case.expected_message): compile_snapshot( - project=project, metadata_json=json.dumps(METADATA), project_dir=tmp_path + project=project, + metadata_json=json.dumps(METADATA), + project_dir=tmp_path, + discovered_inputs=discovered, ) diff --git a/tests/unit/src/sqlbuild/compiler/discovery/test_main.py b/tests/unit/src/sqlbuild/compiler/discovery/test_main.py index 20a56de1a2..79ec931f6c 100644 --- a/tests/unit/src/sqlbuild/compiler/discovery/test_main.py +++ b/tests/unit/src/sqlbuild/compiler/discovery/test_main.py @@ -1059,7 +1059,7 @@ def test_given_parameterized_test_when_discovering_inputs_then_cases_are_not_dup expected_error_fragment=( "Project resource name 'orders' is declared as both model in " "models/marts/orders.sql and unit test in tests/unit/orders.sql; .*unit test, " - "and scenario names must be globally unique" + "scenario, job, job group, sensor, and calendar schedule names must be globally unique" ), ), DiscoverProjectInputsErrorTestCase( diff --git a/website/src/content/docs/docs/concepts/python-nodes/overview.mdx b/website/src/content/docs/docs/concepts/python-nodes/overview.mdx index 7348bbf664..3f20e3ab26 100644 --- a/website/src/content/docs/docs/concepts/python-nodes/overview.mdx +++ b/website/src/content/docs/docs/concepts/python-nodes/overview.mdx @@ -146,7 +146,7 @@ sqb build --select tag:exports # by tag sqb build --select +orders_export # with upstreams ``` -Names are globally unique across models, sources, seeds, functions, loaders, tasks, assets, checks, providers, hooks, unit tests, and scenarios. +Names are globally unique across models, sources, seeds, functions, loaders, tasks, assets, checks, providers, hooks, unit tests, scenarios, jobs, job groups, sensors, and calendar schedules. ## Lifecycle: run, build, check From 5f31bb481f3339a28e1acda338469ee9ccabd350 Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Mon, 9 Nov 2026 13:26:17 +0000 Subject: [PATCH 2/4] docs: sync the docs skill with the python nodes page --- .../sqlbuild/references/docs/concepts/python-nodes/overview.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/python-nodes/overview.md b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/python-nodes/overview.md index d2343d6641..17cdf8438d 100644 --- a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/python-nodes/overview.md +++ b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/python-nodes/overview.md @@ -162,7 +162,7 @@ sqb build --select tag:exports # by tag sqb build --select +orders_export # with upstreams ``` -Names are globally unique across models, sources, seeds, functions, loaders, tasks, assets, checks, providers, hooks, unit tests, and scenarios. +Names are globally unique across models, sources, seeds, functions, loaders, tasks, assets, checks, providers, hooks, unit tests, scenarios, jobs, job groups, sensors, and calendar schedules. ## Lifecycle: run, build, check From 7986d70c1de9e7ce60a18b89382f5b21c7da3b34 Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Mon, 9 Nov 2026 13:34:12 +0000 Subject: [PATCH 3/4] feat(orchestration): say how to rename a colliding declaration A name conflict already names both declarations and their files. It now ends with how to resolve it, and when a job, job group, sensor or calendar schedule is involved, says where that name comes from: a job, group or sensor takes it from its first argument, a calendar schedule from the function it decorates. --- .../_helpers/resource_names.rs | 90 ++++++++++++++----- .../src/project_validation/constants.rs | 4 + .../src/project_validation/tests/helpers.rs | 4 +- .../test_validate_orchestration_names.rs | 23 +++++ .../test_orchestration_snapshot.py | 12 +-- 5 files changed, 102 insertions(+), 31 deletions(-) diff --git a/crates/sqlbuild-discovery/src/project_validation/_helpers/resource_names.rs b/crates/sqlbuild-discovery/src/project_validation/_helpers/resource_names.rs index c9b65c04f6..3f9c07638b 100644 --- a/crates/sqlbuild-discovery/src/project_validation/_helpers/resource_names.rs +++ b/crates/sqlbuild-discovery/src/project_validation/_helpers/resource_names.rs @@ -4,27 +4,20 @@ use std::collections::{HashMap, HashSet}; use crate::project_validation::_helpers::conflicts::{conflict, schema_seeds}; use crate::project_validation::_helpers::identities::block_or_file_name; +use crate::project_validation::constants::ORCHESTRATION_RENAME_ADVICE; use crate::project_validation::models::{NodeKind, ProjectFacts, ValidationFailure}; /// Reject the first resource name declared twice across models, sources, seeds and the rest. pub(crate) fn validate_unique_project_resource_names( facts: &ProjectFacts, ) -> Result<(), ValidationFailure> { - let mut seen: HashMap = HashMap::new(); - let mut declare = |name: &str, kind: &str, path: &str| -> Result<(), ValidationFailure> { - if let Some((existing_kind, existing_path)) = seen.get(name) { - return Err(conflict(format!( - "Project resource name '{name}' is declared as both {existing_kind} in \ - {existing_path} and {kind} in {path}; model, source, seed, function, loader, \ - task, asset, check, provider, hook, unit test, scenario, job, job group, sensor, \ - and calendar schedule names must be globally unique" - ))); - } - seen.insert(name.to_owned(), (kind.to_owned(), path.to_owned())); - Ok(()) - }; + let mut seen: HashMap = HashMap::new(); for model in &facts.models { - declare(&model.file.name, "model", &model.file.path)?; + declare( + &mut seen, + (&model.file.name, "model", &model.file.path), + false, + )?; } let mut managed_sources: HashSet<&str> = HashSet::new(); for file in &facts.sources { @@ -32,32 +25,48 @@ pub(crate) fn validate_unique_project_resource_names( if entry.managed { managed_sources.insert(&entry.name); } - declare(&entry.name, "source", &file.path)?; + declare(&mut seen, (&entry.name, "source", &file.path), false)?; } } for seed in schema_seeds(facts) { - declare(&seed.name, "seed", &seed.path)?; + declare(&mut seen, (&seed.name, "seed", &seed.path), false)?; } for function in facts.sql_functions.iter().chain(&facts.python_functions) { - declare(&function.name, "function", &function.path)?; + declare( + &mut seen, + (&function.name, "function", &function.path), + false, + )?; } for node in &facts.nodes { if node.kind == NodeKind::Loader && managed_sources.contains(node.name.as_str()) { continue; } - declare(&node.name, node.kind.label(), &node.path)?; + declare( + &mut seen, + (&node.name, node.kind.label(), &node.path), + false, + )?; } for provider in &facts.providers { - declare(&provider.name, "provider", &provider.path)?; + declare( + &mut seen, + (&provider.name, "provider", &provider.path), + false, + )?; } for hook in &facts.sql_hooks { - declare(&hook.name, "sql hook", &hook.path)?; + declare(&mut seen, (&hook.name, "sql hook", &hook.path), false)?; } for hook in &facts.hook_functions { - declare(&hook.name, "python hook", &hook.path)?; + declare(&mut seen, (&hook.name, "python hook", &hook.path), false)?; } for scenario in &facts.scenarios { - declare(&scenario.name, "scenario", &scenario.path)?; + declare( + &mut seen, + (&scenario.name, "scenario", &scenario.path), + false, + )?; } let mut test_names: HashSet<&str> = HashSet::new(); for test in &facts.tests { @@ -66,11 +75,44 @@ pub(crate) fn validate_unique_project_resource_names( if !test_names.insert(name) { continue; } - declare(name, "unit test", &test.file.path)?; + declare(&mut seen, (name, "unit test", &test.file.path), false)?; } } for declaration in &facts.orchestration { - declare(&declaration.name, declaration.kind, &declaration.path)?; + declare( + &mut seen, + (&declaration.name, declaration.kind, &declaration.path), + true, + )?; + } + Ok(()) +} + +/// A name's first declaration: its kind, its file and whether it is an orchestration declaration. +type Declared = (String, String, bool); + +/// Record one declaration, or name both declarations and say how to rename one of them. +fn declare( + seen: &mut HashMap, + (name, kind, path): (&str, &str, &str), + orchestration: bool, +) -> Result<(), ValidationFailure> { + if let Some((existing_kind, existing_path, existing_orchestration)) = seen.get(name) { + let advice: &str = if orchestration || *existing_orchestration { + ORCHESTRATION_RENAME_ADVICE + } else { + "" + }; + return Err(conflict(format!( + "Project resource name '{name}' is declared as both {existing_kind} in \ + {existing_path} and {kind} in {path}; model, source, seed, function, loader, task, \ + asset, check, provider, hook, unit test, scenario, job, job group, sensor, and \ + calendar schedule names must be globally unique; rename one of them{advice}" + ))); } + seen.insert( + name.to_owned(), + (kind.to_owned(), path.to_owned(), orchestration), + ); Ok(()) } diff --git a/crates/sqlbuild-discovery/src/project_validation/constants.rs b/crates/sqlbuild-discovery/src/project_validation/constants.rs index e03d600d5a..e2dcb6a95a 100644 --- a/crates/sqlbuild-discovery/src/project_validation/constants.rs +++ b/crates/sqlbuild-discovery/src/project_validation/constants.rs @@ -12,3 +12,7 @@ pub(crate) const RESERVED_MODEL_NAMES: [&str; 1] = ["_chain_"]; pub(crate) const INITIAL_SEED_READ_BYTES: u64 = 64 * 1024; /// A header and two data rows: enough to know a seed holds more than one data row. pub(crate) const RECORDS_FOR_MANY_ROWS: usize = 3; + +/// How a colliding orchestration declaration is renamed, added to its name conflict. +pub(crate) const ORCHESTRATION_RENAME_ADVICE: &str = " (a job, job group or sensor takes its name \ + from its first argument, and a calendar schedule from the function it decorates)"; diff --git a/crates/sqlbuild-discovery/src/project_validation/tests/helpers.rs b/crates/sqlbuild-discovery/src/project_validation/tests/helpers.rs index 601bc978ea..64c9bc7474 100644 --- a/crates/sqlbuild-discovery/src/project_validation/tests/helpers.rs +++ b/crates/sqlbuild-discovery/src/project_validation/tests/helpers.rs @@ -101,13 +101,13 @@ pub(super) fn declared(name: &str, kind: &'static str) -> OrchestrationNameFact } /// A global unique-name conflict that `message` begins. -pub(super) fn collision(message: &str) -> Result<(), String> { +pub(super) fn collision(message: &str, advice: &str) -> Result<(), String> { failure( FailureKind::Conflict, &format!( "{message}; model, source, seed, function, loader, task, asset, check, provider, \ hook, unit test, scenario, job, job group, sensor, and calendar schedule names must \ - be globally unique" + be globally unique; rename one of them{advice}" ), ) } diff --git a/crates/sqlbuild-discovery/src/project_validation/tests/test_validate_orchestration_names.rs b/crates/sqlbuild-discovery/src/project_validation/tests/test_validate_orchestration_names.rs index beb6d48ffa..eb5505f446 100644 --- a/crates/sqlbuild-discovery/src/project_validation/tests/test_validate_orchestration_names.rs +++ b/crates/sqlbuild-discovery/src/project_validation/tests/test_validate_orchestration_names.rs @@ -5,6 +5,9 @@ use crate::project_validation::models::{ use crate::project_validation::tests::helpers::{collision, declared, model, named}; use crate::project_validation::tests::test_types::ValidationTestCase; +const RENAME_DECLARATION: &str = " (a job, job group or sensor takes its name from its first \ + argument, and a calendar schedule from the function it decorates)"; + #[test] fn given_orchestration_names_when_validating_then_they_share_the_project_namespace() { let test_cases = [ @@ -32,6 +35,7 @@ fn given_orchestration_names_when_validating_then_they_share_the_project_namespa expected_failure: collision( "Project resource name 'orders' is declared as both model in models/orders.sql \ and job in orchestration/daily.py", + RENAME_DECLARATION, ), }, ValidationTestCase { @@ -47,6 +51,7 @@ fn given_orchestration_names_when_validating_then_they_share_the_project_namespa expected_failure: collision( "Project resource name 'orders_arrived' is declared as both unit test in \ tests/orders_arrived.sql and sensor in orchestration/daily.py", + RENAME_DECLARATION, ), }, ValidationTestCase { @@ -59,6 +64,23 @@ fn given_orchestration_names_when_validating_then_they_share_the_project_namespa expected_failure: collision( "Project resource name 'fulfillment' is declared as both provider in \ providers/fulfillment.py and calendar schedule in orchestration/daily.py", + RENAME_DECLARATION, + ), + }, + ValidationTestCase { + description: "a unit test named like a model gets no orchestration advice", + facts: ProjectFacts { + models: vec![model("orders")], + tests: vec![TestFileFact { + file: named("orders", "tests/orders.sql"), + blocks: vec![TestBlockFact::default()], + }], + ..ProjectFacts::default() + }, + expected_failure: collision( + "Project resource name 'orders' is declared as both model in models/orders.sql \ + and unit test in tests/orders.sql", + "", ), }, ValidationTestCase { @@ -70,6 +92,7 @@ fn given_orchestration_names_when_validating_then_they_share_the_project_namespa expected_failure: collision( "Project resource name 'daily' is declared as both job in orchestration/daily.py \ and job group in orchestration/daily.py", + RENAME_DECLARATION, ), }, ]; diff --git a/tests/integration/src/sqlbuild/compiler/compiled_project/test_orchestration_snapshot.py b/tests/integration/src/sqlbuild/compiler/compiled_project/test_orchestration_snapshot.py index 972081f9a0..9b1ba8d315 100644 --- a/tests/integration/src/sqlbuild/compiler/compiled_project/test_orchestration_snapshot.py +++ b/tests/integration/src/sqlbuild/compiler/compiled_project/test_orchestration_snapshot.py @@ -50,9 +50,11 @@ ) from tests.integration.src.sqlbuild.compiler.graph.helpers import write_files -GLOBALLY_UNIQUE: str = ( +ORCHESTRATION_NAME_CONFLICT: str = ( "; model, source, seed, function, loader, task, asset, check, provider, hook, unit test, " - "scenario, job, job group, sensor, and calendar schedule names must be globally unique" + "scenario, job, job group, sensor, and calendar schedule names must be globally unique; " + "rename one of them (a job, job group or sensor takes its name from its first argument, and " + "a calendar schedule from the function it decorates)" ) @@ -1060,7 +1062,7 @@ def test_given_declaration_modules_when_compiling_then_native_discovery_collects expected_error=DiscoveryConflictError, expected_message=re.escape( "Project resource name 'daily' is declared as both job in orchestration/jobs.py " - "and job in orchestration/jobs.py" + GLOBALLY_UNIQUE + "and job in orchestration/jobs.py" + ORCHESTRATION_NAME_CONFLICT ), ), SnapshotDiscoveryRejectionTestCase( @@ -1073,7 +1075,7 @@ def test_given_declaration_modules_when_compiling_then_native_discovery_collects expected_error=DiscoveryConflictError, expected_message=re.escape( "Project resource name 'orders' is declared as both model in models/orders.sql " - "and job in orchestration/jobs.py" + GLOBALLY_UNIQUE + "and job in orchestration/jobs.py" + ORCHESTRATION_NAME_CONFLICT ), ), SnapshotDiscoveryRejectionTestCase( @@ -1088,7 +1090,7 @@ def test_given_declaration_modules_when_compiling_then_native_discovery_collects expected_error=DiscoveryConflictError, expected_message=re.escape( "Project resource name 'daily' is declared as both sensor in orchestration/jobs.py " - "and job in orchestration/jobs.py" + GLOBALLY_UNIQUE + "and job in orchestration/jobs.py" + ORCHESTRATION_NAME_CONFLICT ), ), SnapshotDiscoveryRejectionTestCase( From 67e90e0c9c7ede0e32f85d80cc8def0335a96ddf Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Mon, 9 Nov 2026 13:42:47 +0000 Subject: [PATCH 4/4] refactor(orchestration): collect declared names before scanning for conflicts The name check now gathers every declaration as a named record and then scans them once, instead of threading a mutable map through a helper. The first conflict found is the same. --- .../_helpers/resource_names.rs | 167 +++++++++++------- .../src/project_validation/models.rs | 10 ++ 2 files changed, 109 insertions(+), 68 deletions(-) diff --git a/crates/sqlbuild-discovery/src/project_validation/_helpers/resource_names.rs b/crates/sqlbuild-discovery/src/project_validation/_helpers/resource_names.rs index 3f9c07638b..2d10e91f6a 100644 --- a/crates/sqlbuild-discovery/src/project_validation/_helpers/resource_names.rs +++ b/crates/sqlbuild-discovery/src/project_validation/_helpers/resource_names.rs @@ -5,19 +5,35 @@ use std::collections::{HashMap, HashSet}; use crate::project_validation::_helpers::conflicts::{conflict, schema_seeds}; use crate::project_validation::_helpers::identities::block_or_file_name; use crate::project_validation::constants::ORCHESTRATION_RENAME_ADVICE; -use crate::project_validation::models::{NodeKind, ProjectFacts, ValidationFailure}; +use crate::project_validation::models::{ + DeclaredName, NamedPath, NodeKind, ProjectFacts, ValidationFailure, +}; /// Reject the first resource name declared twice across models, sources, seeds and the rest. pub(crate) fn validate_unique_project_resource_names( facts: &ProjectFacts, ) -> Result<(), ValidationFailure> { - let mut seen: HashMap = HashMap::new(); + let seeds: Vec = schema_seeds(facts); + let mut seen: HashMap<&str, DeclaredName<'_>> = HashMap::new(); + for declared in declarations(facts, &seeds) { + if let Some(first) = seen.get(declared.name) { + return Err(name_conflict(first, &declared)); + } + seen.insert(declared.name, declared); + } + Ok(()) +} + +/// Every declared name in the order conflicts are reported. +fn declarations<'a>(facts: &'a ProjectFacts, seeds: &'a [NamedPath]) -> Vec> { + let mut declared: Vec> = Vec::new(); for model in &facts.models { - declare( - &mut seen, - (&model.file.name, "model", &model.file.path), - false, - )?; + declared.push(DeclaredName { + name: &model.file.name, + kind: "model", + path: &model.file.path, + orchestration: false, + }); } let mut managed_sources: HashSet<&str> = HashSet::new(); for file in &facts.sources { @@ -25,94 +41,109 @@ pub(crate) fn validate_unique_project_resource_names( if entry.managed { managed_sources.insert(&entry.name); } - declare(&mut seen, (&entry.name, "source", &file.path), false)?; + declared.push(DeclaredName { + name: &entry.name, + kind: "source", + path: &file.path, + orchestration: false, + }); } } - for seed in schema_seeds(facts) { - declare(&mut seen, (&seed.name, "seed", &seed.path), false)?; + for seed in seeds { + declared.push(DeclaredName { + name: &seed.name, + kind: "seed", + path: &seed.path, + orchestration: false, + }); } for function in facts.sql_functions.iter().chain(&facts.python_functions) { - declare( - &mut seen, - (&function.name, "function", &function.path), - false, - )?; + declared.push(DeclaredName { + name: &function.name, + kind: "function", + path: &function.path, + orchestration: false, + }); } for node in &facts.nodes { if node.kind == NodeKind::Loader && managed_sources.contains(node.name.as_str()) { continue; } - declare( - &mut seen, - (&node.name, node.kind.label(), &node.path), - false, - )?; + declared.push(DeclaredName { + name: &node.name, + kind: node.kind.label(), + path: &node.path, + orchestration: false, + }); } for provider in &facts.providers { - declare( - &mut seen, - (&provider.name, "provider", &provider.path), - false, - )?; + declared.push(DeclaredName { + name: &provider.name, + kind: "provider", + path: &provider.path, + orchestration: false, + }); } for hook in &facts.sql_hooks { - declare(&mut seen, (&hook.name, "sql hook", &hook.path), false)?; + declared.push(DeclaredName { + name: &hook.name, + kind: "sql hook", + path: &hook.path, + orchestration: false, + }); } for hook in &facts.hook_functions { - declare(&mut seen, (&hook.name, "python hook", &hook.path), false)?; + declared.push(DeclaredName { + name: &hook.name, + kind: "python hook", + path: &hook.path, + orchestration: false, + }); } for scenario in &facts.scenarios { - declare( - &mut seen, - (&scenario.name, "scenario", &scenario.path), - false, - )?; + declared.push(DeclaredName { + name: &scenario.name, + kind: "scenario", + path: &scenario.path, + orchestration: false, + }); } let mut test_names: HashSet<&str> = HashSet::new(); for test in &facts.tests { for block in &test.blocks { let name: &str = block_or_file_name(block.name.as_deref(), &test.file.name); - if !test_names.insert(name) { - continue; + if test_names.insert(name) { + declared.push(DeclaredName { + name, + kind: "unit test", + path: &test.file.path, + orchestration: false, + }); } - declare(&mut seen, (name, "unit test", &test.file.path), false)?; } } for declaration in &facts.orchestration { - declare( - &mut seen, - (&declaration.name, declaration.kind, &declaration.path), - true, - )?; + declared.push(DeclaredName { + name: &declaration.name, + kind: declaration.kind, + path: &declaration.path, + orchestration: true, + }); } - Ok(()) + declared } -/// A name's first declaration: its kind, its file and whether it is an orchestration declaration. -type Declared = (String, String, bool); - -/// Record one declaration, or name both declarations and say how to rename one of them. -fn declare( - seen: &mut HashMap, - (name, kind, path): (&str, &str, &str), - orchestration: bool, -) -> Result<(), ValidationFailure> { - if let Some((existing_kind, existing_path, existing_orchestration)) = seen.get(name) { - let advice: &str = if orchestration || *existing_orchestration { - ORCHESTRATION_RENAME_ADVICE - } else { - "" - }; - return Err(conflict(format!( - "Project resource name '{name}' is declared as both {existing_kind} in \ - {existing_path} and {kind} in {path}; model, source, seed, function, loader, task, \ - asset, check, provider, hook, unit test, scenario, job, job group, sensor, and \ - calendar schedule names must be globally unique; rename one of them{advice}" - ))); - } - seen.insert( - name.to_owned(), - (kind.to_owned(), path.to_owned(), orchestration), - ); - Ok(()) +/// Name both declarations and say how to rename one of them. +fn name_conflict(first: &DeclaredName<'_>, second: &DeclaredName<'_>) -> ValidationFailure { + let advice: &str = if first.orchestration || second.orchestration { + ORCHESTRATION_RENAME_ADVICE + } else { + "" + }; + conflict(format!( + "Project resource name '{}' is declared as both {} in {} and {} in {}; model, source, seed, function, loader, task, asset, check, provider, \ + hook, unit test, scenario, job, job group, sensor, and calendar schedule names must be \ + globally unique; rename one of them{advice}", + first.name, first.kind, first.path, second.kind, second.path + )) } diff --git a/crates/sqlbuild-discovery/src/project_validation/models.rs b/crates/sqlbuild-discovery/src/project_validation/models.rs index a583cc66ce..2b185d25ab 100644 --- a/crates/sqlbuild-discovery/src/project_validation/models.rs +++ b/crates/sqlbuild-discovery/src/project_validation/models.rs @@ -182,6 +182,16 @@ pub struct ProjectFacts { pub orchestration: Vec, } +/// One name in the project's resource namespace, as a name conflict reports it. +#[derive(Debug)] +pub(crate) struct DeclaredName<'a> { + pub(crate) name: &'a str, + pub(crate) kind: &'a str, + pub(crate) path: &'a str, + /// A job, job group, sensor or calendar schedule, whose conflicts say how it is named. + pub(crate) orchestration: bool, +} + /// A named orchestration declaration: its name, its kind as conflicts word it, and its module. #[derive(Clone, Debug, PartialEq, Eq)] pub struct OrchestrationNameFact {