feat: add a named-selector limit Rule for orchestration jobs and sensors

benchmarks/duckshop/edge_cases/named_selectors/models/orders/int_orders.sql

@@ -0,0 +1,2 @@
+MODEL (description "Synthetic orders edge case int_orders");
+SELECT 1 AS order_id

benchmarks/duckshop/edge_cases/named_selectors/models/orders/order_totals.sql

@@ -0,0 +1,2 @@
+MODEL (description "Synthetic orders edge case order_totals");
+SELECT 1 AS order_id

benchmarks/duckshop/edge_cases/named_selectors/models/orders/stg_orders.sql

@@ -0,0 +1,2 @@
+MODEL (description "Synthetic orders edge case stg_orders");
+SELECT 1 AS order_id

benchmarks/duckshop/edge_cases/named_selectors/models/orders/stg_payments.sql

@@ -0,0 +1,2 @@
+MODEL (description "Synthetic orders edge case stg_payments");
+SELECT 1 AS order_id

benchmarks/duckshop/edge_cases/named_selectors/orchestration/jobs/orders_folder_job.toml

@@ -0,0 +1,4 @@
+# A folder, an expansion, a glob, an intersection's name, a job name and a repeated name: one named node.
+name = "orders_folder_job"
+description = "Selects the orders pipeline without listing it."
+select = "models/orders +order_totals stg_* order_totals,tag:orders orders_listed_job order_totals"

benchmarks/duckshop/edge_cases/named_selectors/orchestration/jobs/orders_listed_job.toml

@@ -0,0 +1,4 @@
+# Four nodes listed one by one: over the default limit of three.
+name = "orders_listed_job"
+description = "Lists every orders model by name."
+select = "stg_orders stg_payments int_orders order_totals"

benchmarks/duckshop/edge_cases/named_selectors/orchestration/jobs/orders_repair_job.toml

@@ -0,0 +1,4 @@
+# A deliberate list, kept by an exact rule exception in the test's [rules] table.
+name = "orders_repair_job"
+description = "Rebuilds the four orders models after a late payments load."
+select = "stg_orders stg_payments int_orders order_totals"

benchmarks/duckshop/edge_cases/named_selectors/orchestration/sensors/_orders_conditions.py

@@ -0,0 +1,14 @@
+"""Conditions the orders sensors import: one over the limit, a platform event and a builder."""
+
+from sqlbuild.orchestration import PlatformEventKind, node_event, platform_event
+
+ORDERS_LANDED = node_event(
+    select="stg_orders stg_payments int_orders order_totals", kinds=["success"]
+)
+TICK_OVERDUE = platform_event(kinds=[PlatformEventKind.TICK_OVERDUE])
+
+
+def landed_after(model: str) -> object:
+    """A success condition on one model, built when the sensor module loads."""
+
+    return node_event(select=model, kinds=["success"])

benchmarks/duckshop/edge_cases/named_selectors/orchestration/sensors/on_orders_built_sensor.py

@@ -0,0 +1,12 @@
+"""Refreshes totals after one model: a condition a function builds, which cannot be counted."""
+
+from orchestration.sensors._orders_conditions import landed_after
+
+from sqlbuild.orchestration import sensor
+
+on_orders_built = sensor(
+    "on_orders_built_sensor",
+    description="Refresh order totals after the totals model",
+    when=landed_after("order_totals"),
+    run="orders_folder_job",
+)

benchmarks/duckshop/edge_cases/named_selectors/orchestration/sensors/on_orders_computed_sensor.py

@@ -0,0 +1,12 @@
+"""A condition whose select is not a string literal cannot be counted and is reported."""
+
+from sqlbuild.orchestration import node_event, sensor
+
+ORDERS_SELECTION = "stg_orders"
+
+on_orders_computed = sensor(
+    "on_orders_computed_sensor",
+    description="Refresh totals when staged orders change",
+    when=node_event(select=ORDERS_SELECTION, kinds=["success"]),
+    run="orders_folder_job",
+)

benchmarks/duckshop/edge_cases/named_selectors/orchestration/sensors/on_orders_failure_sensor.py

@@ -0,0 +1,14 @@
+"""Starts the repair job when any of four models fails: one over the limit, bound to a name."""
+
+from sqlbuild.orchestration import node_event, sensor
+
+ORDERS_FAILED = node_event(
+    select="stg_orders stg_payments int_orders order_totals", kinds=["failure"]
+)
+
+on_orders_failure = sensor(
+    "on_orders_failure_sensor",
+    description="Repair orders after a failure",
+    when=ORDERS_FAILED,
+    run="orders_repair_job",
+)

benchmarks/duckshop/edge_cases/named_selectors/orchestration/sensors/on_orders_landed_sensor.py

@@ -0,0 +1,12 @@
+"""Refreshes totals when orders land: a condition imported from a project module, over the limit."""
+
+from orchestration.sensors._orders_conditions import ORDERS_LANDED
+
+from sqlbuild.orchestration import sensor
+
+on_orders_landed = sensor(
+    "on_orders_landed_sensor",
+    description="Refresh order totals after orders land",
+    when=ORDERS_LANDED,
+    run="orders_folder_job",
+)

benchmarks/duckshop/edge_cases/named_selectors/orchestration/sensors/on_tick_overdue_sensor.py

@@ -0,0 +1,12 @@
+"""Repairs orders when a tick is overdue: an imported platform event, which selects no nodes."""
+
+from orchestration.sensors._orders_conditions import TICK_OVERDUE
+
+from sqlbuild.orchestration import sensor
+
+on_tick_overdue = sensor(
+    "on_tick_overdue_sensor",
+    description="Repair orders after an overdue tick",
+    when=TICK_OVERDUE,
+    run="orders_repair_job",
+)

crates/sqlbuild-analysis/src/graph/main/mod.rs

@@ -7,6 +7,7 @@ pub mod edge_keys;
 pub mod graph_from_indexes;
 pub mod match_selector;
 pub mod model_layer_count;
+pub mod named_selector_atoms;
 pub mod parse_selector;
 pub mod reachable_graph_keys;
 pub mod reachable_keys;

crates/sqlbuild-analysis/src/graph/main/named_selector_atoms.rs

@@ -0,0 +1,28 @@
+use crate::graph::constants::{INTERSECTION_SEPARATOR, JOB_KIND, PATH_KIND, TAG_KIND};
+use crate::graph::models::ParsedSelector;
+
+/// The atoms of `selection` that name one node: bare or typed names, without `+`, globs, tags, paths or jobs.
+#[must_use]
+pub fn named_selector_atoms(selection: &str) -> Vec<String> {
+    selection
+        .split(char::is_whitespace)
+        .flat_map(|token| token.split(INTERSECTION_SEPARATOR))
+        .filter(|atom| names_one_node(atom))
+        .map(str::to_owned)
+        .collect()
+}
+
+fn names_one_node(atom: &str) -> bool {
+    match crate::graph::_helpers::selectors::parse(atom) {
+        Ok(ParsedSelector::Kind {
+            kind,
+            value,
+            upstream: false,
+            downstream: false,
+        }) => {
+            ![TAG_KIND, PATH_KIND, JOB_KIND].contains(&kind.as_str())
+                && !crate::graph::_helpers::selectors::is_pattern(&value)
+        }
+        _ => false,
+    }
+}

crates/sqlbuild-analysis/src/graph/tests.rs

@@ -4,6 +4,8 @@ mod graph;
 mod helpers;
 #[path = "tests/test_job_selectors.rs"]
 mod job_selectors;
+#[path = "tests/test_named_selector_atoms.rs"]
+mod named_selector_atoms;
 #[path = "tests/test_selector_scenarios.rs"]
 mod selector_scenarios;
 #[path = "tests/test_types.rs"]

crates/sqlbuild-analysis/src/graph/tests/test_named_selector_atoms.rs

@@ -0,0 +1,66 @@
+use crate::graph::main::named_selector_atoms::named_selector_atoms;
+use crate::graph::tests::test_types::NamedAtomsTestCase;
+
+#[test]
+fn given_a_selection_when_listing_named_atoms_then_only_single_node_names_count() {
+    let test_cases = [
+        NamedAtomsTestCase {
+            description: "bare names count",
+            selection: "refund_review fct_orders",
+            expected_atoms: &["refund_review", "fct_orders"],
+        },
+        NamedAtomsTestCase {
+            description: "a bare name inside an intersection counts, its tag does not",
+            selection: "fct_orders,tag:orders",
+            expected_atoms: &["fct_orders"],
+        },
+        NamedAtomsTestCase {
+            description: "typed names count",
+            selection: "task:refund_review source:raw_orders seed:countries test:orders_unique",
+            expected_atoms: &[
+                "task:refund_review",
+                "source:raw_orders",
+                "seed:countries",
+                "test:orders_unique",
+            ],
+        },
+        NamedAtomsTestCase {
+            description: "tags do not count",
+            selection: "tag:orders tag:finance",
+            expected_atoms: &[],
+        },
+        NamedAtomsTestCase {
+            description: "folder paths do not count",
+            selection: "models/orders python/orders path:models/finance",
+            expected_atoms: &[],
+        },
+        NamedAtomsTestCase {
+            description: "graph expansions and path selectors do not count",
+            selection: "+fct_orders fct_orders+ +fct_refunds+ stg_orders~fct_orders",
+            expected_atoms: &[],
+        },
+        NamedAtomsTestCase {
+            description: "globs do not count",
+            selection: "fct_orders_* task:refund_? fct_[ab]",
+            expected_atoms: &[],
+        },
+        NamedAtomsTestCase {
+            description: "job selectors do not count",
+            selection: "job:orders_nightly_job",
+            expected_atoms: &[],
+        },
+        NamedAtomsTestCase {
+            description: "extra whitespace adds nothing",
+            selection: "  fct_orders \n\t fct_refunds  ",
+            expected_atoms: &["fct_orders", "fct_refunds"],
+        },
+    ];
+    for test_case in test_cases {
+        assert_eq!(
+            named_selector_atoms(test_case.selection),
+            test_case.expected_atoms,
+            "{}",
+            test_case.description
+        );
+    }
+}

crates/sqlbuild-analysis/src/graph/tests/test_types.rs

@@ -73,3 +73,9 @@ pub(super) struct NeedsJobsTestCase {
     pub(super) also_known: &'static [&'static str],
     pub(super) expected_needs_jobs: bool,
 }
+
+pub(super) struct NamedAtomsTestCase {
+    pub(super) description: &'static str,
+    pub(super) selection: &'static str,
+    pub(super) expected_atoms: &'static [&'static str],
+}

crates/sqlbuild-discovery/src/orchestration_files/_helpers/python_modules.rs

@@ -148,5 +148,6 @@ pub(crate) fn module_declaration(
         position: name.position,
         job_references,
         targets: declaration.targets,
+        condition: declaration.condition,
     }))
 }

crates/sqlbuild-discovery/src/orchestration_files/models.rs

@@ -4,6 +4,8 @@ use std::collections::BTreeMap;
 
 use serde::Deserialize;
 
+use sqlbuild_pysource::orchestration_modules::models::ConditionSource;
+
 use crate::models::{DiscoveryFailure, ReadFailure, StageFailure};
 use crate::project_validation::models::OrchestrationNameFact;
 
@@ -107,6 +109,8 @@ pub struct NamedModule {
     pub job_references: Vec<(&'static str, Positioned<String>)>,
     /// A sensor's targets: empty for every target, `None` when they are not written as literals.
     pub targets: Option<Vec<String>>,
+    /// What a sensor's `when` selects, as far as its own module shows; `None` when it selects no nodes.
+    pub condition: Option<ConditionSource>,
 }
 
 /// The calendars, sensors and every module below their folders.

crates/sqlbuild-pysource/src/orchestration_modules/_helpers/calls.rs

@@ -8,12 +8,21 @@ use ruff_text_size::{Ranged, TextSize};
 use crate::cpython_tree::main::source_position::source_position;
 use crate::cpython_tree::models::SourceLines;
 use crate::orchestration_modules::constants::{
-    CALENDAR_CALL, ORCHESTRATION_MODULE, RUN_EVENT_CALL, SENSOR_CALL,
+    CALENDAR_CALL, ORCHESTRATION_MODULE, PLATFORM_EVENT_CALL, RUN_EVENT_CALL,
+    SELECTING_CONDITION_CALLS, SENSOR_CALL,
 };
 use crate::orchestration_modules::models::{LiteralArgument, SourcePosition};
 
-/// The calls a module's `calendar`, `sensor` and `run_event` names can make.
-const AUTHORING_CALLS: [&str; 3] = [CALENDAR_CALL, SENSOR_CALL, RUN_EVENT_CALL];
+/// The calls a module's `calendar`, `sensor` and condition names can make.
+const AUTHORING_CALLS: [&str; 7] = [
+    CALENDAR_CALL,
+    SENSOR_CALL,
+    RUN_EVENT_CALL,
+    PLATFORM_EVENT_CALL,
+    SELECTING_CONDITION_CALLS[0],
+    SELECTING_CONDITION_CALLS[1],
+    SELECTING_CONDITION_CALLS[2],
+];
 
 /// The 1-based line and column of `offset`.
 pub(crate) fn position(lines: &SourceLines, offset: TextSize) -> SourcePosition {
@@ -96,7 +105,7 @@ fn dotted(expression: &Expr) -> Option<String> {
     }
 }
 
-/// Which authoring call `call` is: an imported `calendar`, `sensor` or `run_event`, or one reached as an attribute of the orchestration module.
+/// Which authoring call `call` is: an imported `calendar`, `sensor` or condition, or one reached as an attribute of the orchestration module.
 pub(crate) fn authoring_call(
     call: &ExprCall,
     (imported, modules): (&BTreeMap<&str, &str>, &BTreeSet<String>),