diff --git a/src/sqlbuild/cli/commands/_helpers/compile/target_writer.py b/src/sqlbuild/cli/commands/_helpers/compile/target_writer.py index bf74b5c7f2..583da81cc6 100644 --- a/src/sqlbuild/cli/commands/_helpers/compile/target_writer.py +++ b/src/sqlbuild/cli/commands/_helpers/compile/target_writer.py @@ -52,10 +52,8 @@ from sqlbuild.runtime.observability.classes.operation_lifecycle import OperationLifecycle _COMPILED_DIR: str = "compiled" -_RUN_DIR: str = "run" _MODELS_DIR: str = "models" _FUNCTIONS_DIR: str = "functions" -_SQL_FUNCTIONS_DIR: str = "sql" _AUDITS_DIR: str = "audits" _GENERIC_DIR: str = "generic" _SINGULAR_DIR: str = "singular" diff --git a/src/sqlbuild/cli/commands/_helpers/scenario_execution/runner.py b/src/sqlbuild/cli/commands/_helpers/scenario_execution/runner.py index d4dbc76af6..6f19611a40 100644 --- a/src/sqlbuild/cli/commands/_helpers/scenario_execution/runner.py +++ b/src/sqlbuild/cli/commands/_helpers/scenario_execution/runner.py @@ -83,8 +83,6 @@ ) _SCENARIO_NAME_WIDTH: int = 64 -_EXPECTATION_LABEL_WIDTH: int = 10 -_EXPECTATION_NAME_WIDTH: int = 50 _CAPTURE_RELATION_KIND_WIDTH: int = 8 _CAPTURE_RELATION_NAME_WIDTH: int = _SCENARIO_NAME_WIDTH - 4 - _CAPTURE_RELATION_KIND_WIDTH - 1 diff --git a/src/sqlbuild/cli/commands/classes/build_progress_callbacks.py b/src/sqlbuild/cli/commands/classes/build_progress_callbacks.py index 0e490475c9..1a7fe8c318 100644 --- a/src/sqlbuild/cli/commands/classes/build_progress_callbacks.py +++ b/src/sqlbuild/cli/commands/classes/build_progress_callbacks.py @@ -62,7 +62,6 @@ from sqlbuild.spec.contracts.models import SourceEntry _TYPE_WIDTH: int = 10 -_MAX_NAME_WIDTH: int = 60 _MIN_NAME_WIDTH: int = 20 _NAME_PADDING: int = 2 _SUB_INDENT: int = 2 diff --git a/src/sqlbuild/cli/output/_helpers/execution_result_document.py b/src/sqlbuild/cli/output/_helpers/execution_result_document.py index 724c45bed9..b2421d1c4c 100644 --- a/src/sqlbuild/cli/output/_helpers/execution_result_document.py +++ b/src/sqlbuild/cli/output/_helpers/execution_result_document.py @@ -67,11 +67,6 @@ from sqlbuild.virtual.executor.models import VirtualCloneResult _JSON_VERSION: int = 1 -_SUCCESS_STATUS: str = "success" -_FAILED_STATUS: str = "failed" -_SKIPPED_STATUS: str = "skipped" -_WARNING_STATUS: str = "warning" -_WARN_STATUS: str = "warn" _SCENARIO_RESOURCE_NAMESPACE: ContextVar[str | None] = ContextVar( "sqlbuild_scenario_resource_namespace", default=None ) diff --git a/src/sqlbuild/compiler/compile/_helpers/analysis/compact.py b/src/sqlbuild/compiler/compile/_helpers/analysis/compact.py index bf65ecd3bc..42e9da6806 100644 --- a/src/sqlbuild/compiler/compile/_helpers/analysis/compact.py +++ b/src/sqlbuild/compiler/compile/_helpers/analysis/compact.py @@ -3,7 +3,6 @@ from __future__ import annotations import logging -import re from collections.abc import Sequence from dataclasses import replace from typing import Any, cast @@ -61,13 +60,6 @@ InferredNullability, ) from sqlbuild.compiler.profiling.main.record import record_compile_timing -from sqlbuild.compiler.references.main._quoted_reference_call_pattern import ( - quoted_reference_call_pattern, -) -from sqlbuild.compiler.references.main.reference_call_prefix_pattern_text import ( - reference_call_prefix_pattern_text, -) -from sqlbuild.compiler.references.types import SqlReferenceKind from sqlbuild.compiler.sql_analysis.constants import ( POLYGLOT_ANALYSIS_BASE_TABLES as _POLYGLOT_ANALYSIS_BASE_TABLES, ) @@ -175,19 +167,6 @@ _LARGE_COMPACT_PROJECT_MODELS: int = 10_000 _LARGE_COMPACT_SQL_BYTES: int = 48 * 1024 * 1024 _NATIVE_LEGACY_FALLBACK: str = "native project type recovery requires legacy fallback" -_REF_PATTERN: re.Pattern[str] = quoted_reference_call_pattern(SqlReferenceKind.REF) -_SEED_PATTERN: re.Pattern[str] = quoted_reference_call_pattern(SqlReferenceKind.SEED) -_SOURCE_PATTERN: re.Pattern[str] = quoted_reference_call_pattern(SqlReferenceKind.SOURCE) -_DBT_REF_PATTERN: re.Pattern[str] = quoted_reference_call_pattern(SqlReferenceKind.DBT_REF) -_UDF_PATTERN: re.Pattern[str] = re.compile( - rf"{reference_call_prefix_pattern_text(SqlReferenceKind.UDF)}" - r'"([A-Za-z_][A-Za-z0-9_]*)"\)\s*(?=\()' -) -_TABLE_FUNCTION_PATTERN: re.Pattern[str] = re.compile( - rf"{reference_call_prefix_pattern_text(SqlReferenceKind.TABLE_FUNCTION)}" - r'"([A-Za-z_][A-Za-z0-9_]*)"\)\s*(?=\()' -) -_PLACEHOLDER_PATTERN: re.Pattern[str] = re.compile(r"@@@(\w+)") def infer_columns_with_sql_analysis( diff --git a/src/sqlbuild/compiler/compile/_helpers/attachment/audits.py b/src/sqlbuild/compiler/compile/_helpers/attachment/audits.py index 2bf965757d..1da79606c4 100644 --- a/src/sqlbuild/compiler/compile/_helpers/attachment/audits.py +++ b/src/sqlbuild/compiler/compile/_helpers/attachment/audits.py @@ -93,12 +93,6 @@ def remember_scoped_declarations( self.scoped_declarations[key] = declarations -_MODEL_HOOK_KEYS: frozenset[str] = frozenset({"pre_hooks", "post_hooks"}) -_HOOK_CONTEXT_PARAMETER_NAMES: frozenset[str] = frozenset( - {"ctx", "context", "_ctx", "hook_context"} -) - - def build_audit_inputs( *, discovered_inputs: DiscoveredProjectInputs, diff --git a/src/sqlbuild/compiler/compile/_helpers/attachment/functions.py b/src/sqlbuild/compiler/compile/_helpers/attachment/functions.py index a48b3c9dfb..83d65cf6e3 100644 --- a/src/sqlbuild/compiler/compile/_helpers/attachment/functions.py +++ b/src/sqlbuild/compiler/compile/_helpers/attachment/functions.py @@ -2,7 +2,6 @@ from __future__ import annotations -import re from dataclasses import dataclass from pathlib import Path from typing import Any, cast @@ -79,13 +78,6 @@ class _PythonFunctionBuildContext: python_functions_inherit_default_namespace: bool -_HOOK_TEMPLATE_PATTERN: re.Pattern[str] = re.compile(r"\$\{[^}]+\}") -_MODEL_HOOK_KEYS: frozenset[str] = frozenset({"pre_hooks", "post_hooks"}) -_HOOK_CONTEXT_PARAMETER_NAMES: frozenset[str] = frozenset( - {"ctx", "context", "_ctx", "hook_context"} -) - - def build_sql_function_inputs( *, discovered_inputs: DiscoveredProjectInputs, diff --git a/src/sqlbuild/compiler/compile/_helpers/attachment/references.py b/src/sqlbuild/compiler/compile/_helpers/attachment/references.py index 6a1da58202..99b05079d7 100644 --- a/src/sqlbuild/compiler/compile/_helpers/attachment/references.py +++ b/src/sqlbuild/compiler/compile/_helpers/attachment/references.py @@ -2,8 +2,6 @@ from __future__ import annotations -import re - from sqlbuild.compiler.compile.constants import TABLE_FUNCTION_RETURN_KEYS from sqlbuild.compiler.compile.exceptions import CompileInputError from sqlbuild.compiler.compile.models import ( @@ -19,12 +17,6 @@ ) from sqlbuild.compiler.references.types import ExternalSqlReferenceResolver, SqlReferenceKind -_HOOK_TEMPLATE_PATTERN: re.Pattern[str] = re.compile(r"\$\{[^}]+\}") -_MODEL_HOOK_KEYS: frozenset[str] = frozenset({"pre_hooks", "post_hooks"}) -_HOOK_CONTEXT_PARAMETER_NAMES: frozenset[str] = frozenset( - {"ctx", "context", "_ctx", "hook_context"} -) - def validate_model_references( *, diff --git a/src/sqlbuild/compiler/compile/_helpers/attachment/sources.py b/src/sqlbuild/compiler/compile/_helpers/attachment/sources.py index 55ca9f3889..417d45b0d9 100644 --- a/src/sqlbuild/compiler/compile/_helpers/attachment/sources.py +++ b/src/sqlbuild/compiler/compile/_helpers/attachment/sources.py @@ -2,7 +2,6 @@ from __future__ import annotations -import re from dataclasses import replace from pathlib import Path from typing import cast @@ -38,12 +37,6 @@ SourceEntry, ) -_HOOK_TEMPLATE_PATTERN: re.Pattern[str] = re.compile(r"\$\{[^}]+\}") -_MODEL_HOOK_KEYS: frozenset[str] = frozenset({"pre_hooks", "post_hooks"}) -_HOOK_CONTEXT_PARAMETER_NAMES: frozenset[str] = frozenset( - {"ctx", "context", "_ctx", "hook_context"} -) - def build_source_inputs( *, diff --git a/src/sqlbuild/compiler/compile/_helpers/attachment/sql_tests.py b/src/sqlbuild/compiler/compile/_helpers/attachment/sql_tests.py index 26a0eb28b2..f99cabbe12 100644 --- a/src/sqlbuild/compiler/compile/_helpers/attachment/sql_tests.py +++ b/src/sqlbuild/compiler/compile/_helpers/attachment/sql_tests.py @@ -2,7 +2,6 @@ from __future__ import annotations -import re from dataclasses import dataclass, replace from pathlib import Path @@ -76,12 +75,6 @@ ) from sqlbuild.compiler.scopes.types import ResourceKind, ScopeKind, UsageKind -_HOOK_TEMPLATE_PATTERN: re.Pattern[str] = re.compile(r"\$\{[^}]+\}") -_MODEL_HOOK_KEYS: frozenset[str] = frozenset({"pre_hooks", "post_hooks"}) -_HOOK_CONTEXT_PARAMETER_NAMES: frozenset[str] = frozenset( - {"ctx", "context", "_ctx", "hook_context"} -) - @dataclass(frozen=True) class _ExpandedSqlTest: diff --git a/src/sqlbuild/compiler/compile/main/_cursor_roles.py b/src/sqlbuild/compiler/compile/main/_cursor_roles.py index fc5b437af6..956b85fcec 100644 --- a/src/sqlbuild/compiler/compile/main/_cursor_roles.py +++ b/src/sqlbuild/compiler/compile/main/_cursor_roles.py @@ -45,8 +45,6 @@ def resolve_cursor_input_roles(*, model: CompiledModel) -> CursorInputRoles: return CursorInputRoles( filter_inputs=filter_inputs, watermark_inputs=watermark_inputs, - filter_field=CURSOR_INPUTS_CONFIG_KEY, - watermark_field=CURSOR_INPUTS_CONFIG_KEY, ) diff --git a/src/sqlbuild/compiler/compile/models.py b/src/sqlbuild/compiler/compile/models.py index d54e22ac83..ddddb2584f 100644 --- a/src/sqlbuild/compiler/compile/models.py +++ b/src/sqlbuild/compiler/compile/models.py @@ -218,8 +218,6 @@ class CursorInputRoles: filter_inputs: dict[str, str] watermark_inputs: dict[str, str] - filter_field: str - watermark_field: str @dataclass(frozen=True) diff --git a/src/sqlbuild/compiler/discovery/_helpers/sql/model_files.py b/src/sqlbuild/compiler/discovery/_helpers/sql/model_files.py index 5ba5185769..ab0d0cb15a 100644 --- a/src/sqlbuild/compiler/discovery/_helpers/sql/model_files.py +++ b/src/sqlbuild/compiler/discovery/_helpers/sql/model_files.py @@ -21,7 +21,6 @@ PythonHookEntry, SqlHookEntry, ) -from sqlbuild.compiler.references.types import SqlReferenceKind from sqlbuild.spec.contracts.models import SourceLocation from sqlbuild.sql_values.models import AuthoredSqlSet, AuthoredSqlValueCall @@ -29,43 +28,15 @@ _MODEL_HEADER_WORD_TOKEN: str = "word" _MODEL_HEADER_STRING_TOKEN: str = "string" _MODEL_HEADER_SYMBOL_TOKEN: str = "symbol" -_MODEL_HEADER_SYMBOLS: frozenset[str] = frozenset({"(", ")", "[", "]", "{", "}", ","}) _MODEL_HEADER_OPEN_PAREN: str = "(" _MODEL_HEADER_CLOSE_PAREN: str = ")" -_MODEL_HEADER_OPEN_BRACKET: str = "[" -_MODEL_HEADER_CLOSE_BRACKET: str = "]" -_MODEL_HEADER_OPEN_BRACE: str = "{" -_MODEL_HEADER_CLOSE_BRACE: str = "}" _MODEL_HEADER_COMMA: str = "," -_MODEL_HEADER_KEY_VALUE_SEPARATOR: str = ":" _MODEL_HEADER_QUOTE_NAMES: dict[str, str] = {"'": "single", '"': "double"} _MODEL_HEADER_ESCAPE_CHARACTER: str = "\\" _MODEL_HEADER_COLUMNS_KEY: str = "columns" -_MODEL_HEADER_TYPE_KEY: str = "type" -_MODEL_HEADER_RELATION_CALL_NAMES: frozenset[str] = frozenset( - { - SqlReferenceKind.REF.function_name, - SqlReferenceKind.SEED.function_name, - SqlReferenceKind.SOURCE.function_name, - } -) -_MODEL_HEADER_INLINE_SQL_HOOK_CALL: str = "inline_sql" -_MODEL_HEADER_NAMED_SQL_HOOK_CALL: str = "sql" -_MODEL_HEADER_PYTHON_HOOK_CALL: str = "python" -_MODEL_HEADER_HOOK_CALL_NAMES: frozenset[str] = frozenset( - { - _MODEL_HEADER_INLINE_SQL_HOOK_CALL, - _MODEL_HEADER_NAMED_SQL_HOOK_CALL, - _MODEL_HEADER_PYTHON_HOOK_CALL, - } -) -_MODEL_HEADER_HOOK_FIELD_NAMES: frozenset[str] = frozenset({"pre_hooks", "post_hooks"}) _MODEL_HEADER_TRUE_VALUE: str = "true" _MODEL_HEADER_FALSE_VALUE: str = "false" _MODEL_HEADER_NULL_VALUE: str = "null" -_MODEL_HEADER_TYPED_CONSTANT_CALL: str = "constant" -_MODEL_HEADER_THRESHOLDS_KEY: str = "thresholds" -_MODEL_HEADER_THRESHOLD_BOUND_KEYS: frozenset[str] = frozenset({"warn", "error"}) _SQL_IDENTIFIER_SEPARATOR: str = "_" _SQL_SELECT_KEYWORD: str = "SELECT" _SQL_UNION_KEYWORD: str = "UNION" diff --git a/src/sqlbuild/compiler/planner/_helpers/output/cursor_type_check.py b/src/sqlbuild/compiler/planner/_helpers/output/cursor_type_check.py index 96a6981e17..419b9b4505 100644 --- a/src/sqlbuild/compiler/planner/_helpers/output/cursor_type_check.py +++ b/src/sqlbuild/compiler/planner/_helpers/output/cursor_type_check.py @@ -14,23 +14,6 @@ from sqlbuild.diagnostics.main.log_debug_event import log_debug_event _DEBUG_LOGGER: logging.Logger = logging.getLogger("sqlbuild.planner") -_TIMESTAMP_SUBSTRINGS: frozenset[str] = frozenset( - { - "TIMESTAMP", - "DATETIME", - "DATE", - } -) - -_INTEGER_SUBSTRINGS: frozenset[str] = frozenset( - { - "BIGINT", - "SMALLINT", - "TINYINT", - "MEDIUMINT", - "INT", - } -) _POLYGLOT_TIMESTAMP_TYPE_NAMES: frozenset[str] = frozenset( { diff --git a/src/sqlbuild/compiler/planner/_helpers/output/plan_entry.py b/src/sqlbuild/compiler/planner/_helpers/output/plan_entry.py index 985719ee71..f9a6ea1a8a 100644 --- a/src/sqlbuild/compiler/planner/_helpers/output/plan_entry.py +++ b/src/sqlbuild/compiler/planner/_helpers/output/plan_entry.py @@ -9,6 +9,7 @@ from sqlbuild.adapter.contract.classes.base_adapter import BaseAdapter from sqlbuild.adapter.contract.models import ColumnInfo, RelationInfo +from sqlbuild.compiler.compile.constants import CURSOR_INPUTS_CONFIG_KEY from sqlbuild.compiler.compile.exceptions import CompileInputError from sqlbuild.compiler.compile.main._cursor_roles import resolve_cursor_input_roles from sqlbuild.compiler.compile.models import ( @@ -143,7 +144,6 @@ ) from sqlbuild.spec.contracts.types import MicrobatchLimitAction, TableType -_MODELS_DIR_PREFIX: str = "models/" _IDEMPOTENT_MICROBATCH_STRATEGIES: frozenset[IncrementalStrategy] = frozenset( (IncrementalStrategy.DELETE_INSERT, IncrementalStrategy.MERGE) ) @@ -1684,17 +1684,12 @@ def validate_source_cursor_input_columns( cursor_roles: CursorInputRoles = resolve_cursor_input_roles(model=model) effective_models_by_name: dict[str, CompiledModel] = models_by_name or {} effective_source_map: dict[str, SourceEntry] = source_map or {} - authored_inputs: tuple[tuple[str, str, str], ...] = tuple( - (cursor_roles.filter_field, name, column) - for name, column in cursor_roles.filter_inputs.items() - ) + tuple( - (cursor_roles.watermark_field, name, column) - for name, column in cursor_roles.watermark_inputs.items() - ) - config_field: str + authored_inputs: tuple[tuple[str, str], ...] = tuple( + cursor_roles.filter_inputs.items() + ) + tuple(cursor_roles.watermark_inputs.items()) input_name: str input_cursor_column: str - for config_field, input_name, input_cursor_column in authored_inputs: + for input_name, input_cursor_column in authored_inputs: ref: CompileSqlReference = resolve_lineage_reference( model=model, input_name=input_name, @@ -1715,7 +1710,8 @@ def validate_source_cursor_input_columns( continue declared_display: str = ", ".join(declared_names) or "none" raise PlannerInputError( - f"model '{model.name}': {config_field} references model '{ref.ref_name}' " + f"model '{model.name}': {CURSOR_INPUTS_CONFIG_KEY} " + f"references model '{ref.ref_name}' " f"column '{input_cursor_column}', but that model contract does not expose " f"the column. Known output columns: {declared_display}", code="S302", @@ -1729,7 +1725,8 @@ def validate_source_cursor_input_columns( continue declared_display = ", ".join(declared_names) or "none" raise PlannerInputError( - f"model '{model.name}': {config_field} references source '{ref.ref_name}' " + f"model '{model.name}': {CURSOR_INPUTS_CONFIG_KEY} " + f"references source '{ref.ref_name}' " f"column '{input_cursor_column}', but that source contract does not expose " f"the column. Declared contract columns: {declared_display}", code="S302", @@ -1742,7 +1739,7 @@ def validate_source_cursor_input_columns( continue known_display: str = ", ".join(col.name for col in known_columns) or "none" raise PlannerInputError( - f"model '{model.name}': {config_field} references source '{ref.ref_name}' " + f"model '{model.name}': {CURSOR_INPUTS_CONFIG_KEY} references source '{ref.ref_name}' " f"column '{input_cursor_column}', but that source does not expose the column. " f"Known source columns: {known_display}", code="S302", diff --git a/src/sqlbuild/compiler/planner/_helpers/output/strategy.py b/src/sqlbuild/compiler/planner/_helpers/output/strategy.py index 737f847de7..62c70aa62c 100644 --- a/src/sqlbuild/compiler/planner/_helpers/output/strategy.py +++ b/src/sqlbuild/compiler/planner/_helpers/output/strategy.py @@ -27,8 +27,6 @@ ) _DEFAULT_ON_SCHEMA_CHANGE: OnSchemaChange = OnSchemaChange.APPEND_NEW_COLUMNS -_DDL_MERGE_SOURCE_ALIAS: str = "__source" -_DDL_MERGE_TARGET_ALIAS: str = "target" def resolve_model_plan_action( diff --git a/src/sqlbuild/executor/python_nodes/_helpers/execution.py b/src/sqlbuild/executor/python_nodes/_helpers/execution.py index 3f242347f2..272468d85f 100644 --- a/src/sqlbuild/executor/python_nodes/_helpers/execution.py +++ b/src/sqlbuild/executor/python_nodes/_helpers/execution.py @@ -5,7 +5,6 @@ import logging import time from collections.abc import Callable -from dataclasses import dataclass, field from typing import Any from sqlbuild.adapter.contract.classes.statement_recorder import StatementRecorder @@ -28,29 +27,21 @@ from sqlbuild.executor.python_nodes.models import ( AssetContext, PythonNodeExecutionResult, - PythonNodeExecutorResult, PythonNodeFanInDecision, PythonNodeRunState, PythonNodeRuntime, TaskContext, ) from sqlbuild.executor.python_nodes.types import ExecutablePythonNode, OwnedResultCallback -from sqlbuild.executor.scheduling.main._build_in_degree import build_python_node_in_degree -from sqlbuild.executor.scheduling.main._build_ready_queue import build_python_node_ready_queue -from sqlbuild.executor.scheduling.main._unlock_downstream import unlock_downstream_python_nodes from sqlbuild.provider.main.runtime import ( ProviderContainer, _empty_provider_container, invoke_with_providers, ) from sqlbuild.python_nodes.main.calculate_retry_delay import calculate_retry_delay -from sqlbuild.python_nodes.main.read_asset_definition import read_asset_definition -from sqlbuild.python_nodes.main.read_task_definition import read_task_definition from sqlbuild.python_nodes.models import ( - AssetDefinition, RetryPolicy, SqlResourceRef, - TaskDefinition, ) from sqlbuild.runtime.observability.classes.operation_lifecycle import ( OperationLifecycle, @@ -61,101 +52,6 @@ ) -@dataclass -class _SerialResultAccumulator: - run_state: PythonNodeRunState - results_by_name: dict[str, PythonNodeExecutionResult] = field(default_factory=dict) - ordered_results: list[PythonNodeExecutionResult] = field(default_factory=list) - - def record(self, *, node: ExecutablePythonNode, result: PythonNodeExecutionResult) -> None: - self.run_state.record_result(node_function=node.function, result=result) - self.results_by_name[node.name] = result - self.ordered_results.append(result) - - -def execute_python_nodes( - *, - nodes: tuple[ExecutablePythonNode, ...], - runtime: PythonNodeRuntime, - statement_recorder: StatementRecorder, - logger: logging.Logger | None = None, - run_state: PythonNodeRunState | None = None, - sleep: Callable[[float], None] = time.sleep, - monotonic: Callable[[], float] = time.monotonic, -) -> PythonNodeExecutorResult: - """Execute task/asset nodes in dependency order within the current process.""" - - resolved_run_state: PythonNodeRunState = ( - run_state if run_state is not None else PythonNodeRunState() - ) - result_store: Any | None = ( - build_direct_node_result_store( - adapter=runtime.adapter, - connection=runtime.connection, - database=runtime.default_database, - schema=runtime.default_schema, - ) - if runtime.persist_node_results - else None - ) - node_by_name: dict[str, ExecutablePythonNode] = {node.name: node for node in nodes} - node_by_dependency_key: dict[object | tuple[str, str], ExecutablePythonNode] = { - node.function: node for node in nodes - } - for node in nodes: - node_by_dependency_key[("name", node.name)] = node - upstream_names: dict[str, tuple[str, ...]] = _build_upstream_names( - nodes=nodes, - node_by_dependency_key=node_by_dependency_key, - ) - downstream_names: dict[str, tuple[str, ...]] = _build_downstream_names( - node_names=tuple(node.name for node in nodes), - upstream_names=upstream_names, - ) - in_degree: dict[str, int] = build_python_node_in_degree( - node_names=tuple(node.name for node in nodes), - upstream_names=upstream_names, - ) - ready: list[str] = build_python_node_ready_queue( - node_names=tuple(node.name for node in nodes), - in_degree=in_degree, - ) - accumulator: _SerialResultAccumulator = _SerialResultAccumulator(resolved_run_state) - - while ready: - node_name: str = ready.pop(0) - node: ExecutablePythonNode = node_by_name[node_name] - - _ = _execute_ready_node( - node=node, - upstream_results=tuple( - accumulator.results_by_name[name] for name in upstream_names[node.name] - ), - runtime=runtime, - statement_recorder=statement_recorder, - logger=logger, - run_state=resolved_run_state, - result_store=result_store, - sleep=sleep, - monotonic=monotonic, - on_result=accumulator.record, - ) - newly_ready: tuple[str, ...] - in_degree, newly_ready = unlock_downstream_python_nodes( - completed_node_name=node.name, - in_degree=in_degree, - downstream_names=downstream_names, - ) - ready.extend(newly_ready) - - if len(accumulator.ordered_results) != len(nodes): - raise ExecutorInputError("Python node executor could not resolve all dependencies") - return PythonNodeExecutorResult( - results=tuple(accumulator.ordered_results), - run_state=resolved_run_state, - ) - - def execute_ready_python_node( *, node: ExecutablePythonNode, @@ -489,57 +385,6 @@ def _build_context( ) -def _build_upstream_names( - *, - nodes: tuple[ExecutablePythonNode, ...], - node_by_dependency_key: dict[object | tuple[str, str], ExecutablePythonNode], -) -> dict[str, tuple[str, ...]]: - upstream_names: dict[str, tuple[str, ...]] = {} - for node in nodes: - names: list[str] = [] - dependency: Callable[..., object] | SqlResourceRef - for dependency in node.depends_on: - if isinstance(dependency, SqlResourceRef): - continue - upstream_node: ExecutablePythonNode | None = node_by_dependency_key.get( - _python_node_dependency_key(dependency) - ) - if upstream_node is None: - raise ExecutorInputError( - f"Python node '{node.name}' depends on a node outside the executor selection" - ) - names.append(upstream_node.name) - upstream_names[node.name] = tuple(names) - return upstream_names - - -def _python_node_dependency_key(dependency: object) -> object | tuple[str, str]: - task_definition: TaskDefinition | None = ( - read_task_definition(dependency) if callable(dependency) else None - ) - if task_definition is not None: - return ("name", task_definition.name) - asset_definition: AssetDefinition | None = ( - read_asset_definition(dependency) if callable(dependency) else None - ) - if asset_definition is not None: - return ("name", asset_definition.name) - return dependency - - -def _build_downstream_names( - *, node_names: tuple[str, ...], upstream_names: dict[str, tuple[str, ...]] -) -> dict[str, tuple[str, ...]]: - downstream: dict[str, list[str]] = {node_name: [] for node_name in node_names} - node_name: str - upstreams: tuple[str, ...] - for node_name, upstreams in upstream_names.items(): - upstream_name: str - for upstream_name in upstreams: - downstream[upstream_name].append(node_name) - return {node_name: tuple(names) for node_name, names in downstream.items()} - - def _node_kind(node: ExecutablePythonNode) -> PythonNodeKind: if isinstance(node, DiscoveredAssetFunction): return PythonNodeKind.ASSET diff --git a/src/sqlbuild/executor/python_nodes/models.py b/src/sqlbuild/executor/python_nodes/models.py index 5687aeb6af..e4139cdbe2 100644 --- a/src/sqlbuild/executor/python_nodes/models.py +++ b/src/sqlbuild/executor/python_nodes/models.py @@ -122,14 +122,6 @@ class CursorWindow: from sqlbuild.executor.python_nodes.classes.run_state import PythonNodeRunState # noqa: E402 -@dataclass(frozen=True) -class PythonNodeExecutorResult: - """Result bundle for one in-process Python-node executor run.""" - - results: tuple[PythonNodeExecutionResult, ...] - run_state: PythonNodeRunState - - @dataclass(frozen=True) class PythonIngressLoaderExecutorResult: """Result bundle for pre-SQL Python/loader lifecycle execution.""" diff --git a/tests/unit/src/sqlbuild/executor/python_nodes/_helpers/helpers.py b/tests/unit/src/sqlbuild/executor/python_nodes/_helpers/helpers.py index ba0e813634..9d1cc43dcd 100644 --- a/tests/unit/src/sqlbuild/executor/python_nodes/_helpers/helpers.py +++ b/tests/unit/src/sqlbuild/executor/python_nodes/_helpers/helpers.py @@ -3,7 +3,9 @@ from __future__ import annotations import logging +import time from collections.abc import Callable, Iterator +from dataclasses import replace from datetime import UTC, datetime from itertools import chain, repeat from pathlib import Path @@ -34,9 +36,17 @@ from sqlbuild.cost.classes.cost_context import CostContext from sqlbuild.cost.models import CostResourceContext from sqlbuild.errors.contracts.exceptions import ExecutorInputError +from sqlbuild.executor.node_results.main._direct_store import build_direct_node_result_store from sqlbuild.executor.node_results.models import NodeResultEnvelope, NodeResultRecord +from sqlbuild.executor.python_nodes._helpers.execution import execute_ready_python_node from sqlbuild.executor.python_nodes.constants import MISSING_DEFAULT -from sqlbuild.executor.python_nodes.models import BasePythonNodeContext, PythonNodeRunState +from sqlbuild.executor.python_nodes.models import ( + BasePythonNodeContext, + PythonNodeExecutionResult, + PythonNodeRunState, + PythonNodeRuntime, +) +from sqlbuild.executor.python_nodes.types import ExecutablePythonNode from sqlbuild.executor.scheduling.main._unlock_downstream import unlock_downstream_python_nodes from sqlbuild.executor.scheduling.models import LifecycleExecutionNode from sqlbuild.observability import LifecycleEvent @@ -51,6 +61,43 @@ ) +def execute_ordered_test_nodes( + *, + nodes: tuple[ExecutablePythonNode, ...], + runtime: PythonNodeRuntime, + statement_recorder: StatementRecorder, + sleep: Callable[[float], None] = time.sleep, + monotonic: Callable[[], float] = time.monotonic, +) -> tuple[PythonNodeExecutionResult, ...]: + """Exercise the live ready-node entry point in fixture order, without a second scheduler.""" + + run_state: PythonNodeRunState = PythonNodeRunState() + runtime = replace( + runtime, + result_store=build_direct_node_result_store( + adapter=runtime.adapter, + connection=runtime.connection, + database=runtime.default_database, + schema=runtime.default_schema, + ), + ) + results: dict[object, PythonNodeExecutionResult] = {} + node: ExecutablePythonNode + for node in nodes: + result: PythonNodeExecutionResult = execute_ready_python_node( + node=node, + upstream_results=tuple(results[dependency] for dependency in node.depends_on), + runtime=runtime, + statement_recorder=statement_recorder, + run_state=run_state, + sleep=sleep, + monotonic=monotonic, + ) + run_state.record_result(node_function=node.function, result=result) + results[node.function] = result + return tuple(results.values()) + + class ExecutionSlackProvider(Provider): provider_name: ClassVar[str] = "slack_provider" label: str = "slack" diff --git a/tests/unit/src/sqlbuild/executor/python_nodes/_helpers/test_execution.py b/tests/unit/src/sqlbuild/executor/python_nodes/_helpers/test_execution.py index 5cac3fd20f..8ff6291176 100644 --- a/tests/unit/src/sqlbuild/executor/python_nodes/_helpers/test_execution.py +++ b/tests/unit/src/sqlbuild/executor/python_nodes/_helpers/test_execution.py @@ -15,12 +15,10 @@ from sqlbuild.executor.node_results.classes.direct_store import DirectNodeResultStore from sqlbuild.executor.node_results.main._direct_store import build_direct_node_result_store from sqlbuild.executor.python_nodes._helpers.execution import ( - execute_python_nodes, execute_ready_python_node, ) from sqlbuild.executor.python_nodes.models import ( PythonNodeExecutionResult, - PythonNodeExecutorResult, PythonNodeResult, PythonNodeRunState, PythonNodeRuntime, @@ -50,6 +48,7 @@ context_provider_asset, context_provider_task, cursor_window, + execute_ordered_test_nodes, export_after_failure, export_after_mixed_skip, export_after_skip, @@ -90,7 +89,7 @@ def malformed_task() -> PythonNodeResult: return PythonNodeResult(payload={"invalid": True}, materialized=True) with invocation_scope("inv-malformed-task"), dispatcher_scope(dispatcher): - result: PythonNodeExecutorResult = execute_python_nodes( + result: tuple[PythonNodeExecutionResult, ...] = execute_ordered_test_nodes( nodes=( DiscoveredTaskFunction( file_path=Path("/project/tasks/bad.py"), @@ -113,8 +112,8 @@ def malformed_task() -> PythonNodeResult: ) operation_events: tuple[LifecycleEvent, ...] = python_operation_events(events) - assert result.results[0].status == PythonNodeStatus.FAILED - assert test_case.expected_error_fragment in (result.results[0].error_message or "") + assert result[0].status == PythonNodeStatus.FAILED + assert test_case.expected_error_fragment in (result[0].error_message or "") assert tuple(event.event_type for event in operation_events) == ( "operation_started", "operation_failed", @@ -273,7 +272,7 @@ def test_given_task_asset_chain_when_executing_python_nodes_then_records_results ), ) - result: PythonNodeExecutorResult = execute_python_nodes( + result: tuple[PythonNodeExecutionResult, ...] = execute_ordered_test_nodes( nodes=nodes, statement_recorder=StatementRecorder(), runtime=PythonNodeRuntime( @@ -288,19 +287,13 @@ def test_given_task_asset_chain_when_executing_python_nodes_then_records_results ), ) - assert ( - tuple(node_result.node_name for node_result in result.results) == test_case.expected_names - ) - assert ( - tuple(node_result.status for node_result in result.results) == test_case.expected_statuses - ) - assert ( - tuple(node_result.payload for node_result in result.results) == test_case.expected_payloads - ) - assert tuple(node_result.materialized for node_result in result.results) == ( + assert tuple(node_result.node_name for node_result in result) == test_case.expected_names + assert tuple(node_result.status for node_result in result) == test_case.expected_statuses + assert tuple(node_result.payload for node_result in result) == test_case.expected_payloads + assert tuple(node_result.materialized for node_result in result) == ( test_case.expected_materialized ) - assert tuple(node_result.error_message for node_result in result.results) == ( + assert tuple(node_result.error_message for node_result in result) == ( test_case.expected_error_fragments ) @@ -343,7 +336,7 @@ def test_given_provider_parameters_when_executing_python_nodes_then_providers_ar {"slack_provider": ExecutionSlackProvider(label="slack")} ).providers - result: PythonNodeExecutorResult = execute_python_nodes( + result: tuple[PythonNodeExecutionResult, ...] = execute_ordered_test_nodes( nodes=nodes, statement_recorder=StatementRecorder(), runtime=PythonNodeRuntime( @@ -358,9 +351,9 @@ def test_given_provider_parameters_when_executing_python_nodes_then_providers_ar ), ) - assert tuple(item.node_name for item in result.results) == test_case.expected_names - assert tuple(item.status for item in result.results) == test_case.expected_statuses - assert tuple(item.payload for item in result.results) == test_case.expected_payloads + assert tuple(item.node_name for item in result) == test_case.expected_names + assert tuple(item.status for item in result) == test_case.expected_statuses + assert tuple(item.payload for item in result) == test_case.expected_payloads @pytest.mark.parametrize( @@ -401,7 +394,7 @@ def test_given_provider_container_when_executing_python_nodes_then_context_expos {"slack_provider": ExecutionSlackProvider(label="slack")} ).providers - result: PythonNodeExecutorResult = execute_python_nodes( + result: tuple[PythonNodeExecutionResult, ...] = execute_ordered_test_nodes( nodes=nodes, statement_recorder=StatementRecorder(), runtime=PythonNodeRuntime( @@ -416,9 +409,9 @@ def test_given_provider_container_when_executing_python_nodes_then_context_expos ), ) - assert tuple(item.node_name for item in result.results) == test_case.expected_names - assert tuple(item.status for item in result.results) == test_case.expected_statuses - assert tuple(item.payload for item in result.results) == test_case.expected_payloads + assert tuple(item.node_name for item in result) == test_case.expected_names + assert tuple(item.status for item in result) == test_case.expected_statuses + assert tuple(item.payload for item in result) == test_case.expected_payloads @pytest.mark.parametrize( @@ -458,7 +451,7 @@ def test_given_missing_provider_container_when_executing_python_nodes_then_failu ), ) - result: PythonNodeExecutorResult = execute_python_nodes( + result: tuple[PythonNodeExecutionResult, ...] = execute_ordered_test_nodes( nodes=nodes, statement_recorder=StatementRecorder(), runtime=PythonNodeRuntime( @@ -472,11 +465,11 @@ def test_given_missing_provider_container_when_executing_python_nodes_then_failu ), ) - assert tuple(item.node_name for item in result.results) == test_case.expected_names - assert tuple(item.status for item in result.results) == test_case.expected_statuses - assert tuple(item.payload for item in result.results) == test_case.expected_payloads - assert tuple(item.materialized for item in result.results) == test_case.expected_materialized - for item, expected in zip(result.results, test_case.expected_error_fragments, strict=True): + assert tuple(item.node_name for item in result) == test_case.expected_names + assert tuple(item.status for item in result) == test_case.expected_statuses + assert tuple(item.payload for item in result) == test_case.expected_payloads + assert tuple(item.materialized for item in result) == test_case.expected_materialized + for item, expected in zip(result, test_case.expected_error_fragments, strict=True): assert expected is not None assert expected in (item.error_message or "") @@ -500,7 +493,7 @@ def test_given_missing_provider_container_when_executing_python_nodes_then_failu def test_given_missing_context_provider_when_executing_python_node_then_failure_is_recorded( test_case: PythonNodeExecutorTestCase, ) -> None: - result: PythonNodeExecutorResult = execute_python_nodes( + result: tuple[PythonNodeExecutionResult, ...] = execute_ordered_test_nodes( nodes=( DiscoveredTaskFunction( file_path=Path("/project/tasks/provider.py"), @@ -521,12 +514,12 @@ def test_given_missing_context_provider_when_executing_python_node_then_failure_ ), ) - assert tuple(item.node_name for item in result.results) == test_case.expected_names - assert tuple(item.status for item in result.results) == test_case.expected_statuses - assert tuple(item.payload for item in result.results) == test_case.expected_payloads + assert tuple(item.node_name for item in result) == test_case.expected_names + assert tuple(item.status for item in result) == test_case.expected_statuses + assert tuple(item.payload for item in result) == test_case.expected_payloads expected_error_fragment: str | None = test_case.expected_error_fragments[0] assert expected_error_fragment is not None - assert expected_error_fragment in (result.results[0].error_message or "") + assert expected_error_fragment in (result[0].error_message or "") @pytest.mark.parametrize( @@ -649,7 +642,7 @@ def test_given_cursor_overrides_when_executing_python_nodes_then_context_receive ), ) - result: PythonNodeExecutorResult = execute_python_nodes( + result: tuple[PythonNodeExecutionResult, ...] = execute_ordered_test_nodes( nodes=nodes, statement_recorder=StatementRecorder(), runtime=PythonNodeRuntime( @@ -667,16 +660,10 @@ def test_given_cursor_overrides_when_executing_python_nodes_then_context_receive ), ) - assert ( - tuple(node_result.node_name for node_result in result.results) == test_case.expected_names - ) - assert ( - tuple(node_result.status for node_result in result.results) == test_case.expected_statuses - ) - assert ( - tuple(node_result.payload for node_result in result.results) == test_case.expected_payloads - ) - assert tuple(node_result.materialized for node_result in result.results) == ( + assert tuple(node_result.node_name for node_result in result) == test_case.expected_names + assert tuple(node_result.status for node_result in result) == test_case.expected_statuses + assert tuple(node_result.payload for node_result in result) == test_case.expected_payloads + assert tuple(node_result.materialized for node_result in result) == ( test_case.expected_materialized ) @@ -714,7 +701,7 @@ def test_given_hard_skipped_upstream_when_executing_python_nodes_then_skips_down ), ) - result: PythonNodeExecutorResult = execute_python_nodes( + result: tuple[PythonNodeExecutionResult, ...] = execute_ordered_test_nodes( nodes=nodes, statement_recorder=StatementRecorder(), runtime=PythonNodeRuntime( @@ -728,16 +715,10 @@ def test_given_hard_skipped_upstream_when_executing_python_nodes_then_skips_down ), ) - assert ( - tuple(node_result.node_name for node_result in result.results) == test_case.expected_names - ) - assert ( - tuple(node_result.status for node_result in result.results) == test_case.expected_statuses - ) - assert ( - tuple(node_result.payload for node_result in result.results) == test_case.expected_payloads - ) - assert tuple(node_result.materialized for node_result in result.results) == ( + assert tuple(node_result.node_name for node_result in result) == test_case.expected_names + assert tuple(node_result.status for node_result in result) == test_case.expected_statuses + assert tuple(node_result.payload for node_result in result) == test_case.expected_payloads + assert tuple(node_result.materialized for node_result in result) == ( test_case.expected_materialized ) @@ -825,7 +806,7 @@ def test_given_mixed_python_skips_when_executing_nodes_then_fan_in_matches_mode( ), ) - result: PythonNodeExecutorResult = execute_python_nodes( + result: tuple[PythonNodeExecutionResult, ...] = execute_ordered_test_nodes( nodes=nodes, statement_recorder=StatementRecorder(), runtime=PythonNodeRuntime( @@ -839,16 +820,10 @@ def test_given_mixed_python_skips_when_executing_nodes_then_fan_in_matches_mode( ), ) - assert ( - tuple(node_result.node_name for node_result in result.results) == test_case.expected_names - ) - assert ( - tuple(node_result.status for node_result in result.results) == test_case.expected_statuses - ) - assert ( - tuple(node_result.payload for node_result in result.results) == test_case.expected_payloads - ) - assert tuple(node_result.materialized for node_result in result.results) == ( + assert tuple(node_result.node_name for node_result in result) == test_case.expected_names + assert tuple(node_result.status for node_result in result) == test_case.expected_statuses + assert tuple(node_result.payload for node_result in result) == test_case.expected_payloads + assert tuple(node_result.materialized for node_result in result) == ( test_case.expected_materialized ) @@ -886,7 +861,7 @@ def test_given_failed_upstream_when_executing_python_nodes_then_blocks_downstrea ), ) - result: PythonNodeExecutorResult = execute_python_nodes( + result: tuple[PythonNodeExecutionResult, ...] = execute_ordered_test_nodes( nodes=nodes, statement_recorder=StatementRecorder(), runtime=PythonNodeRuntime( @@ -900,23 +875,17 @@ def test_given_failed_upstream_when_executing_python_nodes_then_blocks_downstrea ), ) - assert ( - tuple(node_result.node_name for node_result in result.results) == test_case.expected_names - ) - assert ( - tuple(node_result.status for node_result in result.results) == test_case.expected_statuses - ) - assert ( - tuple(node_result.payload for node_result in result.results) == test_case.expected_payloads - ) - assert tuple(node_result.materialized for node_result in result.results) == ( + assert tuple(node_result.node_name for node_result in result) == test_case.expected_names + assert tuple(node_result.status for node_result in result) == test_case.expected_statuses + assert tuple(node_result.payload for node_result in result) == test_case.expected_payloads + assert tuple(node_result.materialized for node_result in result) == ( test_case.expected_materialized ) - assert result.results[0].error_message == test_case.expected_error_fragments[0] - assert result.results[1].error_message is not None + assert result[0].error_message == test_case.expected_error_fragments[0] + assert result[1].error_message is not None expected_error_fragment: str | None = test_case.expected_error_fragments[1] assert expected_error_fragment is not None - assert expected_error_fragment in result.results[1].error_message + assert expected_error_fragment in result[1].error_message @pytest.mark.parametrize( @@ -971,7 +940,7 @@ def record_sleep(delay: float) -> None: phase="build", ), ): - result: PythonNodeExecutorResult = execute_python_nodes( + result: tuple[PythonNodeExecutionResult, ...] = execute_ordered_test_nodes( nodes=nodes, statement_recorder=StatementRecorder(), sleep=record_sleep, @@ -986,9 +955,9 @@ def record_sleep(delay: float) -> None: ), ) - assert result.results[0].status == test_case.expected_status - assert result.results[0].payload == test_case.expected_payload - assert result.results[0].error_message == test_case.expected_error_fragment + assert result[0].status == test_case.expected_status + assert result[0].payload == test_case.expected_payload + assert result[0].error_message == test_case.expected_error_fragment assert flaky_task.attempts == test_case.expected_attempts assert tuple(sleeps) == test_case.expected_sleeps assert tuple(flaky_task.cost_contexts) == tuple( @@ -1078,7 +1047,7 @@ def test_given_retry_policy_when_attempts_exhausted_then_records_final_exception ), ) - result: PythonNodeExecutorResult = execute_python_nodes( + result: tuple[PythonNodeExecutionResult, ...] = execute_ordered_test_nodes( nodes=nodes, statement_recorder=StatementRecorder(), sleep=sleeps.append, @@ -1093,12 +1062,12 @@ def test_given_retry_policy_when_attempts_exhausted_then_records_final_exception ), ) - assert result.results[0].status == test_case.expected_status - assert result.results[0].payload == test_case.expected_payload - assert result.results[0].error_message is not None + assert result[0].status == test_case.expected_status + assert result[0].payload == test_case.expected_payload + assert result[0].error_message is not None expected_error_fragment: str | None = test_case.expected_error_fragment assert expected_error_fragment is not None - assert expected_error_fragment in result.results[0].error_message + assert expected_error_fragment in result[0].error_message assert flaky_task.attempts == test_case.expected_attempts assert tuple(sleeps) == test_case.expected_sleeps @@ -1132,7 +1101,7 @@ def test_given_retry_policy_when_exception_is_not_selected_then_does_not_retry( ), ) - result: PythonNodeExecutorResult = execute_python_nodes( + result: tuple[PythonNodeExecutionResult, ...] = execute_ordered_test_nodes( nodes=nodes, statement_recorder=StatementRecorder(), sleep=sleeps.append, @@ -1147,12 +1116,12 @@ def test_given_retry_policy_when_exception_is_not_selected_then_does_not_retry( ), ) - assert result.results[0].status == test_case.expected_status - assert result.results[0].payload == test_case.expected_payload - assert result.results[0].error_message is not None + assert result[0].status == test_case.expected_status + assert result[0].payload == test_case.expected_payload + assert result[0].error_message is not None expected_error_fragment: str | None = test_case.expected_error_fragment assert expected_error_fragment is not None - assert expected_error_fragment in result.results[0].error_message + assert expected_error_fragment in result[0].error_message assert flaky_task.attempts == test_case.expected_attempts assert tuple(sleeps) == test_case.expected_sleeps @@ -1193,7 +1162,7 @@ def test_given_retry_policy_when_backoff_exceeds_cap_then_sleep_is_capped( ), ) - result: PythonNodeExecutorResult = execute_python_nodes( + result: tuple[PythonNodeExecutionResult, ...] = execute_ordered_test_nodes( nodes=nodes, statement_recorder=StatementRecorder(), sleep=sleeps.append, @@ -1208,8 +1177,8 @@ def test_given_retry_policy_when_backoff_exceeds_cap_then_sleep_is_capped( ), ) - assert result.results[0].status == test_case.expected_status - assert result.results[0].payload == test_case.expected_payload - assert result.results[0].error_message == test_case.expected_error_fragment + assert result[0].status == test_case.expected_status + assert result[0].payload == test_case.expected_payload + assert result[0].error_message == test_case.expected_error_fragment assert flaky_task.attempts == test_case.expected_attempts assert tuple(sleeps) == test_case.expected_sleeps