diff --git a/Makefile b/Makefile index e76ffecaab..15ac68a182 100644 --- a/Makefile +++ b/Makefile @@ -76,6 +76,7 @@ E2E_DUCKDB_BUILD_CORE_PATHS := \ tests/e2e/src/sqlbuild/cli/commands/main/build/test_no_tests_no_audits_flags.py \ tests/e2e/src/sqlbuild/cli/commands/main/build/test_plan_command_surface.py \ tests/e2e/src/sqlbuild/cli/commands/main/build/test_python_hooks.py \ + tests/e2e/src/sqlbuild/cli/commands/main/build/test_python_node_selection.py \ tests/e2e/src/sqlbuild/cli/commands/main/build/test_python_source_deferral.py \ tests/e2e/src/sqlbuild/cli/commands/main/build/test_sql_hook_reads.py \ tests/e2e/src/sqlbuild/cli/commands/main/build/test_python_node_identity.py \ 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 b0642c0854..95e5a8cef0 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 @@ -170,6 +170,12 @@ Python nodes run in two phases relative to SQL: - **Ingress (pre-SQL):** loaders, and tasks/assets that feed sources, run before SQL models are built. - **Read-side (post-SQL):** tasks/assets that read SQL run after their SQL dependencies are built. +Read-side tasks and assets follow the same selection rule as SQL models. A SQL dependency that is +part of the run gates the node: it waits for that model, seed, or source load, and is skipped if the +dependency fails or is skipped. A dependency that is not part of the run (not selected, or an +unmanaged source) is read as it currently exists, so `sqb build --select task:export_orders` runs +the task against the relations already in the warehouse. + The commands differ in what they include by default: | Command | SQL | Loaders / tasks / assets | Checks | Audits | diff --git a/src/sqlbuild/cli/commands/_helpers/build_python_nodes/python_lifecycle.py b/src/sqlbuild/cli/commands/_helpers/build_python_nodes/python_lifecycle.py index 3794f15de0..76c8389315 100644 --- a/src/sqlbuild/cli/commands/_helpers/build_python_nodes/python_lifecycle.py +++ b/src/sqlbuild/cli/commands/_helpers/build_python_nodes/python_lifecycle.py @@ -9,6 +9,7 @@ from sqlbuild.adapter.contract.classes.base_adapter import BaseAdapter from sqlbuild.cli.commands._helpers.build_python_nodes.python_lifecycle_selection import ( python_node_result_names, + scheduled_sql_resource_names, task_asset_python_node_names, ) from sqlbuild.cli.commands._helpers.build_python_nodes.python_node_output import ( @@ -177,6 +178,8 @@ def prepare_direct_python_lifecycle( read_side_tracker = create_read_side_python_execution_tracker( python_graph=python_graph, selected_python_names=read_side_names, + scheduled_sql_names=scheduled_sql_resource_names(plan_output) + | frozenset(result.source_name for result in ingress_load_results), runtime=PythonNodeRuntime( adapter=adapter, connection_config=connection_config, diff --git a/src/sqlbuild/cli/commands/_helpers/build_python_nodes/python_lifecycle_selection.py b/src/sqlbuild/cli/commands/_helpers/build_python_nodes/python_lifecycle_selection.py index 1eb17e6820..4d0571dbb0 100644 --- a/src/sqlbuild/cli/commands/_helpers/build_python_nodes/python_lifecycle_selection.py +++ b/src/sqlbuild/cli/commands/_helpers/build_python_nodes/python_lifecycle_selection.py @@ -23,6 +23,17 @@ def task_asset_python_node_names( return frozenset(name for name in selected_names if name in task_asset_names) +def scheduled_sql_resource_names(plan: PlanOutput) -> frozenset[str]: + """Return the models, seeds, and source loads this run executes and reports results for.""" + + scheduled_keys: frozenset[CompiledObjectKey] = frozenset(plan.execution_order) + return frozenset( + entry.name + for entry in (*plan.model_entries, *plan.seed_entries, *plan.source_load_entries) + if entry.key in scheduled_keys + ) + + def sql_loader_functions_for_lifecycle_handoff( *, discovered_inputs: DiscoveredProjectInputs, diff --git a/src/sqlbuild/cli/commands/_helpers/check/core.py b/src/sqlbuild/cli/commands/_helpers/check/core.py index c37597e83a..1427b6c497 100644 --- a/src/sqlbuild/cli/commands/_helpers/check/core.py +++ b/src/sqlbuild/cli/commands/_helpers/check/core.py @@ -36,7 +36,6 @@ PythonSqlRunLifecyclePlan, ) from sqlbuild.compiler.python_nodes.types import PythonNodeKind, PythonNodeStatus -from sqlbuild.executor.build.models import SeedExecutionResult from sqlbuild.executor.build.types import ExecutionStatus from sqlbuild.executor.load.models import LoadExecutionResult from sqlbuild.executor.python_nodes.main.read_side import create_read_side_python_execution_tracker @@ -46,12 +45,10 @@ PythonNodeRunState, PythonNodeRuntime, ) -from sqlbuild.executor.run.models import ModelExecutionResult from sqlbuild.presentation.classes.cli_style import CliStyle from sqlbuild.provider.main.runtime import ProviderContainer from sqlbuild.python_nodes.models import SqlResourceRef from sqlbuild.python_nodes.types import SqlResourceRefKind -from sqlbuild.runtime.contracts.types import ExecutionResourceKind from sqlbuild.spec.contracts.models import SourceEntry _IGNORED_EXCLUDE_SELECTOR_ERROR_CODES: frozenset[str] = frozenset({"S007", "S008", "S009"}) @@ -234,6 +231,7 @@ def run_check_read_side_dependencies( tracker: Any = create_read_side_python_execution_tracker( python_graph=python_graph, selected_python_names=read_side_names, + scheduled_sql_names=frozenset(), runtime=PythonNodeRuntime( adapter=adapter, connection_config=connection_config, @@ -249,25 +247,6 @@ def run_check_read_side_dependencies( providers=providers, ), ) - for sql_ref in sorted(read_side_refs, key=_sql_ref_sort_key): - if sql_ref.kind == SqlResourceRefKind.MODEL: - tracker.record_sql_result( - ModelExecutionResult(model_name=sql_ref.name, status=ExecutionStatus.SUCCESS) - ) - elif sql_ref.kind == SqlResourceRefKind.SEED: - tracker.record_sql_result( - SeedExecutionResult(seed_name=sql_ref.name, status=ExecutionStatus.SUCCESS) - ) - elif sql_ref.kind == SqlResourceRefKind.SOURCE: - tracker.record_sql_result( - LoadExecutionResult( - source_name=sql_ref.name, - loader_name=sql_ref.name, - status=ExecutionStatus.SUCCESS, - target=relation_targets[sql_ref], - resource_kind=ExecutionResourceKind.SOURCE, - ) - ) tracker.dispatch_ready_python_nodes() tracker.finalize_unrun_python_nodes() return tracker.results diff --git a/src/sqlbuild/executor/python_nodes/_helpers/read_side_execution.py b/src/sqlbuild/executor/python_nodes/_helpers/read_side_execution.py index 020e3a63a8..315636f578 100644 --- a/src/sqlbuild/executor/python_nodes/_helpers/read_side_execution.py +++ b/src/sqlbuild/executor/python_nodes/_helpers/read_side_execution.py @@ -19,6 +19,8 @@ def _sql_result_name(result: object) -> str | None: def _sql_result_failed(result: object) -> bool: - if isinstance(result, ModelExecutionResult | LoadExecutionResult | SeedExecutionResult): + if isinstance(result, ModelExecutionResult): + return result.status in {ExecutionStatus.FAILED, ExecutionStatus.SKIPPED} + if isinstance(result, LoadExecutionResult | SeedExecutionResult): return result.status == ExecutionStatus.FAILED return False diff --git a/src/sqlbuild/executor/python_nodes/classes/read_side_python_execution_tracker.py b/src/sqlbuild/executor/python_nodes/classes/read_side_python_execution_tracker.py index 825dcedefa..82c964c868 100644 --- a/src/sqlbuild/executor/python_nodes/classes/read_side_python_execution_tracker.py +++ b/src/sqlbuild/executor/python_nodes/classes/read_side_python_execution_tracker.py @@ -31,18 +31,20 @@ class ReadSidePythonExecutionTracker: - """Dispatch selected read-only Python nodes as SQL dependencies complete.""" + """Dispatch selected read-only Python nodes as SQL dependencies scheduled in this run finish.""" def __init__( self, *, python_graph: PythonNodeGraph, selected_python_names: frozenset[str], + scheduled_sql_names: frozenset[str], runtime: PythonNodeRuntime, identity_recorder: PythonIdentityRecorder | None = None, ) -> None: self._python_graph: PythonNodeGraph = python_graph self._selected_python_names: frozenset[str] = selected_python_names + self._scheduled_sql_names: frozenset[str] = scheduled_sql_names self._runtime: PythonNodeRuntime = runtime self._identity_recorder: PythonIdentityRecorder | None = identity_recorder self._result_store: Any | None = ( @@ -133,7 +135,10 @@ def _sql_dependencies_satisfied(self, node: DiscoveredPythonNode) -> bool: for sql_dep_name in (dependency.name for dependency in node.sql_deps): if sql_dep_name in self._failed_sql_names: return False - if sql_dep_name not in self._completed_sql_names: + if ( + sql_dep_name in self._scheduled_sql_names + and sql_dep_name not in self._completed_sql_names + ): return False return True @@ -191,7 +196,10 @@ def _unrun_reason(self, node: DiscoveredPythonNode) -> str: for sql_dep_name in (dependency.name for dependency in node.sql_deps): if sql_dep_name in self._failed_sql_names: return f"Upstream SQL resource did not succeed: {sql_dep_name}" - if sql_dep_name not in self._completed_sql_names: + if ( + sql_dep_name in self._scheduled_sql_names + and sql_dep_name not in self._completed_sql_names + ): return f"Upstream SQL resource did not complete: {sql_dep_name}" upstream_name: str for upstream_name in self._python_graph.upstream_deps.get(node.name, ()): diff --git a/src/sqlbuild/executor/python_nodes/main/read_side.py b/src/sqlbuild/executor/python_nodes/main/read_side.py index 7dbc8d9999..46725c23af 100644 --- a/src/sqlbuild/executor/python_nodes/main/read_side.py +++ b/src/sqlbuild/executor/python_nodes/main/read_side.py @@ -14,6 +14,7 @@ def create_read_side_python_execution_tracker( *, python_graph: PythonNodeGraph, selected_python_names: frozenset[str], + scheduled_sql_names: frozenset[str], runtime: PythonNodeRuntime, identity_recorder: PythonIdentityRecorder | None = None, ) -> ReadSidePythonExecutionTracker: @@ -22,6 +23,7 @@ def create_read_side_python_execution_tracker( return ReadSidePythonExecutionTracker( python_graph=python_graph, selected_python_names=selected_python_names, + scheduled_sql_names=scheduled_sql_names, runtime=runtime, identity_recorder=identity_recorder, ) diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/build/_test_types.py b/tests/e2e/src/sqlbuild/cli/commands/main/build/_test_types.py index e46aa02b7e..4d12fbf4dd 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/build/_test_types.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/build/_test_types.py @@ -1104,7 +1104,7 @@ class PythonSourceTaskOnlyE2ETestCase: description: str target_config: str - expected_task_pattern: str + expected_task_count: int @dataclass(frozen=True) @@ -1142,3 +1142,18 @@ class AwareCursorDateWatermarkE2ETestCase: plan_args: tuple[str, ...] expected_plan_fragment: str expected_event_ids: tuple[int, ...] + + +@dataclass(frozen=True) +class PythonNodeSelectionE2ETestCase: + """Python nodes selected with or without their SQL upstreams.""" + + description: str + select: tuple[str, ...] + expected_exit_code: int + expected_output_patterns: tuple[str, ...] + expected_row_counts: dict[str, int] + expected_node_counts: dict[str, int] + expected_missing_tables: tuple[str, ...] = () + orders_sql: str = 'MODEL (materialized table);\nSELECT order_id FROM __source("raw_orders")\n' + existing_sql: str = "" diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/build/helpers.py b/tests/e2e/src/sqlbuild/cli/commands/main/build/helpers.py index 33e16d2f86..2f47746609 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/build/helpers.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/build/helpers.py @@ -751,3 +751,68 @@ def explicit_reference_literal_loader(*, table: str) -> str: f' ctx.query("SELECT count(*) FROM {table}")\n' " return [{'id': 1}]\n" ) + + +PYTHON_NODE_SELECTION_DATABASE: str = "warehouse.duckdb" + + +def _counting_python_node(*, decorator: str, name: str, ref: str, returned: str) -> str: + return ( + f"@{decorator}(depends_on={ref})\n" + f"def {name}(ctx):\n" + f" relation = ctx.relation({ref})\n" + f' ctx.execute_sql(f"CREATE OR REPLACE TABLE main.{name}_result AS ' + f'SELECT count(*) AS n FROM {{relation}}")\n' + f" return {returned}\n\n\n" + ) + + +def python_node_selection_project_files(*, orders_sql: str) -> dict[str, str]: + """Return a DuckDB project whose tasks and assets count managed, unmanaged, and model rows.""" + + task_nodes: str = "".join( + _counting_python_node(decorator="task", name=name, ref=ref, returned="None") + for name, ref in ( + ("count_customers", 'source("raw_customers")'), + ("count_raw_orders", 'source("raw_orders")'), + ("count_orders", 'model("orders")'), + ("count_order_summary", 'model("order_summary")'), + ) + ) + asset_nodes: str = "".join( + _counting_python_node( + decorator="asset", name=name, ref=ref, returned="ctx.result(materialized=True)" + ) + for name, ref in ( + ("customers_extract", 'source("raw_customers")'), + ("orders_extract", 'model("orders")'), + ) + ) + return { + "sqlbuild_project.toml": ( + 'name = "order_nodes"\nadapter = "duckdb"\n\n' + f'[connection]\ndatabase = "{PYTHON_NODE_SELECTION_DATABASE}"\n' + ), + "sources/raw.yml": ( + "sources:\n" + " - name: raw_orders\n schema: main\n table: raw_orders\n" + " - name: raw_customers\n managed: true\n write_strategy: table\n" + " columns:\n - name: customer_id\n type: INTEGER\n" + ), + "python/loaders/customers.py": ( + "from sqlbuild.loaders import loader\n\n\n" + "@loader\ndef raw_customers(ctx):\n return [{'customer_id': 99}]\n" + ), + "python/tasks/counts.py": ( + "from sqlbuild.refs import model, source\nfrom sqlbuild.tasks import task\n\n\n" + + task_nodes + ), + "python/assets/extracts.py": ( + "from sqlbuild.assets import asset\nfrom sqlbuild.refs import model, source\n\n\n" + + asset_nodes + ), + "models/orders.sql": orders_sql, + "models/order_summary.sql": ( + 'MODEL (materialized table);\nSELECT count(*) AS order_count FROM __ref("orders")\n' + ), + } diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/build/test_python_node_selection.py b/tests/e2e/src/sqlbuild/cli/commands/main/build/test_python_node_selection.py new file mode 100644 index 0000000000..2dbec73617 --- /dev/null +++ b/tests/e2e/src/sqlbuild/cli/commands/main/build/test_python_node_selection.py @@ -0,0 +1,176 @@ +"""Real-CLI coverage of Python node selection against unselected and selected SQL upstreams.""" + +from __future__ import annotations + +import re +import subprocess +from pathlib import Path + +import pytest + +from tests.e2e.src.sqlbuild.cli.commands.main.build._test_types import ( + PythonNodeSelectionE2ETestCase, +) +from tests.e2e.src.sqlbuild.cli.commands.main.build.helpers import ( + PYTHON_NODE_SELECTION_DATABASE, + python_node_selection_project_files, +) +from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import ( + execute_duckdb, + prepare_inline_project, + query_duckdb, + run_sqb, + table_exists, +) + +_FAILING_ORDERS_SQL: str = ( + "MODEL (materialized table);\n" + "SELECT order_id FROM __source(\"raw_orders\") WHERE error('orders failed') IS NULL\n" +) +_EXISTING_ORDERS: str = "CREATE TABLE main.orders AS SELECT * FROM range(1, 6) AS t(order_id);" +_EXISTING_ROWS: str = ( + "CREATE TABLE main.raw_orders AS SELECT * FROM range(1, 3) AS t(order_id);" + "CREATE TABLE main.raw_customers AS SELECT * FROM range(1, 4) AS t(customer_id);" +) + + +@pytest.mark.parametrize( + "test_case", + [ + PythonNodeSelectionE2ETestCase( + description="task selected alone reads the existing managed source table", + select=("task:count_customers",), + expected_exit_code=0, + expected_output_patterns=(r"count_customers\s+OK",), + expected_row_counts={"raw_customers": 3}, + expected_node_counts={"count_customers": 3}, + ), + PythonNodeSelectionE2ETestCase( + description="asset selected alone reads the existing managed source table", + select=("asset:customers_extract",), + expected_exit_code=0, + expected_output_patterns=(r"customers_extract\s+OK",), + expected_row_counts={"raw_customers": 3}, + expected_node_counts={"customers_extract": 3}, + ), + PythonNodeSelectionE2ETestCase( + description="task on an unmanaged source runs when selected alone", + select=("task:count_raw_orders",), + expected_exit_code=0, + expected_output_patterns=(r"count_raw_orders\s+OK",), + expected_row_counts={}, + expected_node_counts={"count_raw_orders": 2}, + ), + PythonNodeSelectionE2ETestCase( + description="task on an unmanaged source runs when selected with upstreams", + select=("+task:count_raw_orders",), + expected_exit_code=0, + expected_output_patterns=(r"count_raw_orders\s+OK",), + expected_row_counts={}, + expected_node_counts={"count_raw_orders": 2}, + ), + PythonNodeSelectionE2ETestCase( + description="task selected alone reads the existing unselected model", + select=("task:count_orders",), + existing_sql=_EXISTING_ORDERS, + expected_exit_code=0, + expected_output_patterns=(r"count_orders\s+OK",), + expected_row_counts={"orders": 5}, + expected_node_counts={"count_orders": 5}, + ), + PythonNodeSelectionE2ETestCase( + description="asset selected alone reads the existing unselected model", + select=("asset:orders_extract",), + existing_sql=_EXISTING_ORDERS, + expected_exit_code=0, + expected_output_patterns=(r"orders_extract\s+OK",), + expected_row_counts={"orders": 5}, + expected_node_counts={"orders_extract": 5}, + ), + PythonNodeSelectionE2ETestCase( + description="task waits for its selected upstream model", + select=("+task:count_orders",), + expected_exit_code=0, + expected_output_patterns=(r"count_orders\s+OK",), + expected_row_counts={"orders": 2}, + expected_node_counts={"count_orders": 2}, + ), + PythonNodeSelectionE2ETestCase( + description="asset waits for its selected upstream model", + select=("+asset:orders_extract",), + expected_exit_code=0, + expected_output_patterns=(r"orders_extract\s+OK",), + expected_row_counts={"orders": 2}, + expected_node_counts={"orders_extract": 2}, + ), + PythonNodeSelectionE2ETestCase( + description="task is skipped when its selected upstream model fails", + select=("+task:count_orders",), + orders_sql=_FAILING_ORDERS_SQL, + expected_exit_code=1, + expected_output_patterns=( + r"count_orders\s+SKIP\s+Upstream SQL resource did not succeed: orders", + ), + expected_row_counts={}, + expected_node_counts={}, + expected_missing_tables=("count_orders_result",), + ), + PythonNodeSelectionE2ETestCase( + description="asset is skipped when its selected upstream model fails", + select=("+asset:orders_extract",), + orders_sql=_FAILING_ORDERS_SQL, + expected_exit_code=1, + expected_output_patterns=( + r"orders_extract\s+SKIP\s+Upstream SQL resource did not succeed: orders", + ), + expected_row_counts={}, + expected_node_counts={}, + expected_missing_tables=("orders_extract_result",), + ), + PythonNodeSelectionE2ETestCase( + description="task is skipped when its selected upstream model is skipped", + select=("+task:count_order_summary",), + orders_sql=_FAILING_ORDERS_SQL, + expected_exit_code=1, + expected_output_patterns=( + r"count_order_summary\s+SKIP\s+" + r"Upstream SQL resource did not succeed: order_summary", + ), + expected_row_counts={}, + expected_node_counts={}, + expected_missing_tables=("count_order_summary_result",), + ), + ], + ids=lambda case: case.description, +) +def test_given_python_node_selection_when_building_then_it_gates_only_on_scheduled_upstreams( + test_case: PythonNodeSelectionE2ETestCase, + tmp_path: Path, +) -> None: + project_dir: Path = prepare_inline_project( + tmp_path=tmp_path, + project_name="order_nodes", + repo_files=python_node_selection_project_files(orders_sql=test_case.orders_sql), + ) + database: Path = project_dir / PYTHON_NODE_SELECTION_DATABASE + execute_duckdb(db_path=database, sql=_EXISTING_ROWS + test_case.existing_sql) + + build: subprocess.CompletedProcess[str] = run_sqb( + command=("--no-color", "build", "--select", *test_case.select), project_dir=project_dir + ) + + output: str = build.stdout + build.stderr + assert build.returncode == test_case.expected_exit_code, output + assert all(re.search(pattern, output) for pattern in test_case.expected_output_patterns), output + assert { + table: query_duckdb(db_path=database, sql=f"SELECT count(*) FROM main.{table}")[0][0] + for table in test_case.expected_row_counts + } == test_case.expected_row_counts + assert { + node: query_duckdb(db_path=database, sql=f"SELECT n FROM main.{node}_result")[0][0] + for node in test_case.expected_node_counts + } == test_case.expected_node_counts + assert not any( + table_exists(db_path=database, table_name=table) + for table in test_case.expected_missing_tables + ) diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/build/test_python_source_deferral.py b/tests/e2e/src/sqlbuild/cli/commands/main/build/test_python_source_deferral.py index 54b3e96270..5db2a5626a 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/build/test_python_source_deferral.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/build/test_python_source_deferral.py @@ -74,9 +74,7 @@ "INSERT INTO raw_prod.raw_orders VALUES (1), (2);" ) _CHECK_PASSED: str = r"check\s+two_raw_orders\s+PASS" -_TASK_WAITS_FOR_SOURCE: str = ( - r"count_task\s+SKIP\s+Upstream SQL resource did not complete: raw_orders" -) +_TASK_RAN: str = r"count_task\s+OK" @pytest.mark.parametrize( @@ -129,17 +127,17 @@ def test_given_deferred_source_reads_when_python_reads_the_source_then_it_reads_ PythonSourceTaskOnlyE2ETestCase( description="with source deferral", target_config='defer_sources_to = "prod"\n', - expected_task_pattern=_TASK_WAITS_FOR_SOURCE, + expected_task_count=2, ), PythonSourceTaskOnlyE2ETestCase( description="without source deferral", target_config="", - expected_task_pattern=_TASK_WAITS_FOR_SOURCE, + expected_task_count=1, ), ], ids=lambda case: case.description, ) -def test_given_only_a_task_selected_when_building_then_its_declared_source_is_planned( +def test_given_only_a_task_selected_when_building_then_it_reads_its_source_like_sql( test_case: PythonSourceTaskOnlyE2ETestCase, tmp_path: Path, ) -> None: @@ -150,7 +148,8 @@ def test_given_only_a_task_selected_when_building_then_its_declared_source_is_pl project_dir: Path = prepare_inline_project( tmp_path=tmp_path, project_name="deferred_sources", repo_files=files ) - execute_duckdb(db_path=project_dir / _DATABASE, sql=_ENVIRONMENT_ROWS) + database: Path = project_dir / _DATABASE + execute_duckdb(db_path=database, sql=_ENVIRONMENT_ROWS) build: subprocess.CompletedProcess[str] = run_sqb( command=("--no-color", "build", "--select", "task:count_task"), project_dir=project_dir @@ -160,7 +159,11 @@ def test_given_only_a_task_selected_when_building_then_its_declared_source_is_pl assert build.returncode == 0, output assert "G000" not in output assert "missing from the resolved plan source map" not in output - assert re.search(test_case.expected_task_pattern, output), output + assert re.search(_TASK_RAN, output), output + assert ( + query_duckdb(db_path=database, sql="SELECT n FROM main.task_counts")[0][0] + == test_case.expected_task_count + ) if __name__ == "__main__": diff --git a/tests/unit/src/sqlbuild/executor/python_nodes/_helpers/_test_types.py b/tests/unit/src/sqlbuild/executor/python_nodes/_helpers/_test_types.py index f514111a80..bfd9e54a11 100644 --- a/tests/unit/src/sqlbuild/executor/python_nodes/_helpers/_test_types.py +++ b/tests/unit/src/sqlbuild/executor/python_nodes/_helpers/_test_types.py @@ -150,6 +150,7 @@ class ReadSidePythonTrackerTestCase: expected_statuses: tuple[PythonNodeStatus, ...] expected_skip_reasons: tuple[str | None, ...] = () failed_sql_names: tuple[str, ...] = () + scheduled_sql_names: frozenset[str] = frozenset({"stg_orders"}) @dataclass(frozen=True) diff --git a/tests/unit/src/sqlbuild/executor/python_nodes/_helpers/test_read_side_execution.py b/tests/unit/src/sqlbuild/executor/python_nodes/_helpers/test_read_side_execution.py index ac26e44d4e..be3f30a40b 100644 --- a/tests/unit/src/sqlbuild/executor/python_nodes/_helpers/test_read_side_execution.py +++ b/tests/unit/src/sqlbuild/executor/python_nodes/_helpers/test_read_side_execution.py @@ -40,7 +40,17 @@ expected_call_order=("profile_stg_orders", "export_stg_profile"), expected_statuses=(PythonNodeStatus.SUCCESS, PythonNodeStatus.SUCCESS), expected_skip_reasons=(None, None), - ) + ), + ReadSidePythonTrackerTestCase( + description="runs Python on an unscheduled SQL dependency without waiting", + selected_names=frozenset({"profile_stg_orders", "export_stg_profile"}), + completed_sql_names=(), + scheduled_sql_names=frozenset(), + expected_result_names=("profile_stg_orders", "export_stg_profile"), + expected_call_order=("profile_stg_orders", "export_stg_profile"), + expected_statuses=(PythonNodeStatus.SUCCESS, PythonNodeStatus.SUCCESS), + expected_skip_reasons=(None, None), + ), ], ids=lambda case: case.description, ) @@ -52,6 +62,7 @@ def test_given_sql_dep_completes_when_tracking_read_side_then_runs_ready_python_ tracker: ReadSidePythonExecutionTracker = ReadSidePythonExecutionTracker( python_graph=graph, selected_python_names=test_case.selected_names, + scheduled_sql_names=test_case.scheduled_sql_names, runtime=PythonNodeRuntime( adapter=PythonNodeContextTestAdapter(), connection_config={}, @@ -64,6 +75,7 @@ def test_given_sql_dep_completes_when_tracking_read_side_then_runs_ready_python_ ), ) + tracker.dispatch_ready_python_nodes() sql_name: str for sql_name in test_case.completed_sql_names: tracker.record_sql_result( @@ -93,7 +105,19 @@ def test_given_sql_dep_completes_when_tracking_read_side_then_runs_ready_python_ "Upstream Python node did not complete: profile_stg_orders", "Upstream SQL resource did not succeed: stg_orders", ), - ) + ), + ReadSidePythonTrackerTestCase( + description="finalizes Python nodes whose scheduled SQL never reported as skipped", + selected_names=frozenset({"profile_stg_orders", "export_stg_profile"}), + completed_sql_names=(), + expected_result_names=("export_stg_profile", "profile_stg_orders"), + expected_call_order=(), + expected_statuses=(PythonNodeStatus.SKIPPED, PythonNodeStatus.SKIPPED), + expected_skip_reasons=( + "Upstream Python node did not complete: profile_stg_orders", + "Upstream SQL resource did not complete: stg_orders", + ), + ), ], ids=lambda case: case.description, ) @@ -108,6 +132,7 @@ def test_given_sql_dep_fails_when_finalizing_read_side_then_skips_unrun_python_n tracker: ReadSidePythonExecutionTracker = ReadSidePythonExecutionTracker( python_graph=graph, selected_python_names=test_case.selected_names, + scheduled_sql_names=test_case.scheduled_sql_names, runtime=PythonNodeRuntime( adapter=PythonNodeContextTestAdapter(), connection_config={}, 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 732da0adec..9ac5a63176 100644 --- a/website/src/content/docs/docs/concepts/python-nodes/overview.mdx +++ b/website/src/content/docs/docs/concepts/python-nodes/overview.mdx @@ -154,6 +154,12 @@ Python nodes run in two phases relative to SQL: - **Ingress (pre-SQL):** loaders, and tasks/assets that feed sources, run before SQL models are built. - **Read-side (post-SQL):** tasks/assets that read SQL run after their SQL dependencies are built. +Read-side tasks and assets follow the same selection rule as SQL models. A SQL dependency that is +part of the run gates the node: it waits for that model, seed, or source load, and is skipped if the +dependency fails or is skipped. A dependency that is not part of the run (not selected, or an +unmanaged source) is read as it currently exists, so `sqb build --select task:export_orders` runs +the task against the relations already in the warehouse. + The commands differ in what they include by default: | Command | SQL | Loaders / tasks / assets | Checks | Audits |