diff --git a/src/flowx/bundler/dab_writer.py b/src/flowx/bundler/dab_writer.py index 7e04af93..93aa72dd 100644 --- a/src/flowx/bundler/dab_writer.py +++ b/src/flowx/bundler/dab_writer.py @@ -2000,7 +2000,9 @@ def expand_terminals(condition_key: str, seen: set[str]) -> list[str]: seen_keys.add(key) deduped.append(dep) task["depends_on"] = deduped - task.setdefault("run_if", "AT_LEAST_ONE_SUCCESS") + # The untaken branch's terminals are SKIPPED, so NONE_FAILED lets the join run yet still propagates a + # taken-branch failure (ADF dependsOn [Succeeded]); AT_LEAST_ONE_SUCCESS let a sibling success mask it. + task.setdefault("run_if", "NONE_FAILED") _TASK_VALUE_REF = re.compile(r"\{\{tasks\.([^.]+)\.values\.[^}]+\}\}") @@ -2671,6 +2673,7 @@ def pipeline_dict_to_ir(pipeline_dict: dict[str, Any]) -> tuple[Pipeline, list[d reconciliation_status=pipeline_dict.get("reconciliation_status"), migration_status=pipeline_dict.get("migration_status", "included"), audit=dict(pipeline_dict.get("audit") or {}), + synthesized_variable_init_keys=frozenset(pipeline_dict.get("synthesized_variable_init_keys") or ()), ) return pipeline, parameters diff --git a/src/flowx/ir_serde.py b/src/flowx/ir_serde.py index 00f35bb9..da587285 100644 --- a/src/flowx/ir_serde.py +++ b/src/flowx/ir_serde.py @@ -70,6 +70,8 @@ def pipeline_to_dict(pipeline: Pipeline) -> dict[str, Any]: } if pipeline.bundle_variables: result["bundle_variables"] = pipeline.bundle_variables + if pipeline.synthesized_variable_init_keys: + result["synthesized_variable_init_keys"] = sorted(pipeline.synthesized_variable_init_keys) if pipeline.description is not None: result["description"] = pipeline.description if pipeline.timeout_seconds is not None: diff --git a/src/flowx/models/ir.py b/src/flowx/models/ir.py index cc440fba..9b407e31 100644 --- a/src/flowx/models/ir.py +++ b/src/flowx/models/ir.py @@ -632,6 +632,10 @@ class Pipeline: audit: Source-audit counts and transformation ledger. bundle_variables: DAB bundle-variable declarations (name -> ``{"description", "default"}``) for factory globals hoisted under the ``bundle_variable`` resolution policy. + synthesized_variable_init_keys: Task keys of the SetVariable tasks the + translator synthesises for default-valued variables. Tracked + explicitly so dead-init pruning acts only on translator-generated + tasks and never touches a customer-authored task. reconciliation_status: Source-audit result for this pipeline. migration_status: Whether the pipeline is included or explicitly excluded. audit: Source-audit counts and transformation ledger. @@ -651,6 +655,7 @@ class Pipeline: audit: dict[str, Any] = field(default_factory=dict) translation_configuration: TranslationConfiguration | None = None bundle_variables: dict[str, dict[str, Any]] = field(default_factory=dict) + synthesized_variable_init_keys: frozenset[str] = field(default_factory=frozenset) @dataclass(frozen=True, slots=True) diff --git a/src/flowx/motifs/collapser.py b/src/flowx/motifs/collapser.py index 49230000..7d8649b9 100644 --- a/src/flowx/motifs/collapser.py +++ b/src/flowx/motifs/collapser.py @@ -77,6 +77,7 @@ def collapse_motifs( migration_status=pipeline.migration_status, audit=dict(pipeline.audit), translation_configuration=pipeline.translation_configuration, + synthesized_variable_init_keys=pipeline.synthesized_variable_init_keys, ) diff --git a/src/flowx/preparer/workflow_preparer.py b/src/flowx/preparer/workflow_preparer.py index d79c3a7a..95e93bde 100644 --- a/src/flowx/preparer/workflow_preparer.py +++ b/src/flowx/preparer/workflow_preparer.py @@ -2,8 +2,9 @@ from __future__ import annotations +import json from collections.abc import Iterable -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace from typing import Any from flowx.models.dab import DabNotebook, ParameterApproximation, SecretInstruction, SetupTask @@ -336,9 +337,12 @@ def prepare_workflow(pipeline: Pipeline) -> PreparedWorkflow: # Updated in pipeline-declaration order so AppendVariable sees the # most-recent prior writer of each variable. variable_task_keys_map: dict[str, str] = {} + prepared_inits: dict[str, PreparedActivity] = {} for activity in pipeline.tasks: prepared = prepare_activity(activity, scope=scope, variable_task_keys=variable_task_keys_map) + if activity.task_key in pipeline.synthesized_variable_init_keys: + prepared_inits[activity.task_key] = prepared all_tasks.append(prepared.task) all_tasks.extend(prepared.extra_tasks) artifacts = merge_prepared_artifacts(artifacts, prepared) @@ -359,6 +363,8 @@ def prepare_workflow(pipeline: Pipeline) -> PreparedWorkflow: if original_key in task_key_remap: dep["task_key"] = task_key_remap[original_key] + all_tasks, artifacts = _prune_dead_variable_inits(all_tasks, artifacts, prepared_inits) + seen_secrets: set[tuple[str, str]] = set() unique_secrets: list[SecretInstruction] = [] for secret in artifacts.secrets: @@ -450,6 +456,133 @@ def prepare_workflow(pipeline: Pipeline) -> PreparedWorkflow: ) +def _prune_dead_variable_inits( + tasks: list[dict[str, Any]], + artifacts: PreparedArtifacts, + prepared_inits: dict[str, PreparedActivity], +) -> tuple[list[dict[str, Any]], PreparedArtifacts]: + """Drops synthesised variable-init tasks that no real task reaches, with their notebooks and setup tasks. + + Args: + tasks: Prepared parent-job task dicts. + artifacts: Artifacts accumulated while preparing *tasks*. + prepared_inits: Prepared output of each translator-synthesised ``_init_`` task, by task key. + + Returns: + The ``(tasks, artifacts)`` pair with every dead init task removed. + + Notes: + The ADF translator emits an init task per default-valued variable so every read has a + writer to bind to; once an explicit SetVariable dominates every read, the init task is dead + weight that still runs and reads as a duplicate of the real setter. Pruning runs here, on + the fully prepared tasks, because only now are all references final: the agentic merge has + happened, and preparers have materialised references the IR never shows (AppendVariable's + ``source_task_key``, ForEach inner-job parameters, notebook dispatch stubs). Only + translator-synthesised keys are eligible, so a customer-authored task is never removed. + """ + dead_keys = find_dead_variable_inits(tasks, artifacts, prepared_inits) + if not dead_keys: + return tasks, artifacts + dead_notebooks = {id(notebook) for key in dead_keys for notebook in prepared_inits[key].notebooks} + dead_setup_tasks = {id(setup_task) for key in dead_keys for setup_task in prepared_inits[key].setup_tasks} + pruned_artifacts = replace( + artifacts, + notebooks=tuple(notebook for notebook in artifacts.notebooks if id(notebook) not in dead_notebooks), + setup_tasks=tuple(setup_task for setup_task in artifacts.setup_tasks if id(setup_task) not in dead_setup_tasks), + ) + return [task for task in tasks if task.get("task_key") not in dead_keys], pruned_artifacts + + +def find_dead_variable_inits( + tasks: list[dict[str, Any]], + artifacts: PreparedArtifacts, + prepared_inits: dict[str, PreparedActivity], +) -> frozenset[str]: + """Returns the synthesised variable-init task keys that no real task reaches. + + Args: + tasks: Prepared parent-job task dicts. + artifacts: Artifacts accumulated while preparing *tasks*; their notebooks and inner + workflows (e.g. ForEach sub-jobs) are scanned too. + prepared_inits: Prepared output of each translator-synthesised ``_init_`` task, by task key. + + Returns: + The init task keys present in *tasks* that are safe to drop. + + Notes: + A reference is any of the forms the preparers emit: an anchored task-value token + (``tasks..values``), a ``depends_on`` edge, a task parameter whose whole value is the + key (AppendVariable's ``source_task_key``), or a ``taskKey=''`` read in generated + notebook code. Matching these forms rather than a bare substring avoids a false keep when + the key text merely appears inside an unrelated key, path, or metadata field. An init is + live when a real (non-init) task or notebook references it, directly or through a chain of + live inits, so a chain of inits that only feed each other is dropped as a whole. + """ + init_keys = {task["task_key"] for task in tasks if task.get("task_key") in prepared_inits} + notebook_owners = {id(notebook): key for key in init_keys for notebook in prepared_inits[key].notebooks} + references = [ + (task.get("task_key"), _task_references(task, init_keys)) + for task in _iter_workflow_tasks(tasks, artifacts.inner_workflows) + ] + [ + (notebook_owners.get(id(notebook)), {key for key in init_keys if _notebook_references(notebook.content, key)}) + for notebook in artifacts.notebooks + ] + live_keys: set[str] = set() + pending = [key for owner, referenced in references if owner not in init_keys for key in referenced] + while pending: + key = pending.pop() + if key not in live_keys: + live_keys.add(key) + pending.extend(ref for owner, referenced in references if owner == key for ref in referenced) + return frozenset(init_keys - live_keys) + + +_PARAMETER_FIELDS = ( + ("notebook_task", "base_parameters"), + ("run_job_task", "job_parameters"), + ("spark_python_task", "parameters"), + ("spark_jar_task", "parameters"), + ("python_wheel_task", "parameters"), + ("python_wheel_task", "named_parameters"), + ("sql_task", "parameters"), +) + + +def _task_references(task: dict[str, Any], init_keys: set[str]) -> set[str]: + """Returns the init keys that *task* reads through a task-value token, a dependency, or a parameter.""" + serialized_task = json.dumps(task, default=str) + whole_values = {value for value in _iter_reference_values(task) if isinstance(value, str)} + return {key for key in init_keys if f"tasks.{key}.values" in serialized_task or key in whole_values} + + +def _iter_reference_values(task: dict[str, Any]) -> Iterable[Any]: + """Yields *task*'s ``depends_on`` keys and parameter values, including a ForEach body task's.""" + for dependency in task.get("depends_on") or []: + yield dependency.get("task_key") + for task_type, field_name in _PARAMETER_FIELDS: + parameters = (task.get(task_type) or {}).get(field_name) or {} + yield from parameters.values() if isinstance(parameters, dict) else parameters + body_task = (task.get("for_each_task") or {}).get("task") + if body_task: + yield from _iter_reference_values(body_task) + + +def _notebook_references(content: str, task_key: str) -> bool: + """Reports whether generated notebook *content* reads a task value published by *task_key*.""" + return any( + token in content for token in (f"tasks.{task_key}.values", f"taskKey='{task_key}'", f'taskKey="{task_key}"') + ) + + +def _iter_workflow_tasks( + tasks: list[dict[str, Any]], inner_workflows: Iterable[PreparedWorkflow] +) -> Iterable[dict[str, Any]]: + """Yields *tasks* followed by every task of the (recursively nested) inner workflows.""" + yield from tasks + for workflow in inner_workflows: + yield from _iter_workflow_tasks(workflow.tasks, workflow.inner_workflows) + + def _detect_cross_foreach_variable_reads(activities: list[Activity]) -> list[SetupTask]: """Returns SetupTasks for variables mutated inside a ForEach but read outside. diff --git a/src/flowx/sources/adf/translate.py b/src/flowx/sources/adf/translate.py index 1b322fdd..7e24ce4c 100644 --- a/src/flowx/sources/adf/translate.py +++ b/src/flowx/sources/adf/translate.py @@ -141,7 +141,7 @@ def translate_pipeline( # C-05 (VAREX-002): synthesise init SetVariable tasks for variables with a defaultValue, seeding # variable_cache so @variables('X') resolves to the init task's value instead of a self-referential dangler. - init_variable_activities, context = _build_variable_init_activities(pipeline, context) + init_variable_activities, context, synthesized_init_keys = _build_variable_init_activities(pipeline, context) translated_activities: list[Activity] = list(init_variable_activities) sorted_activities = _topological_visit(pipeline.activities) @@ -183,6 +183,7 @@ def translate_pipeline( tasks=translated_activities, tags={"source": "adf", "pipeline": pipeline.name}, schedule=schedule, + synthesized_variable_init_keys=synthesized_init_keys, ) # Whole-IR expression rewrite: catches @{...} tokens the per-activity translators missed (raw SQL @@ -772,7 +773,7 @@ def _normalize_timezone(tz: Any) -> str: def _build_variable_init_activities( pipeline: AdfPipeline, context: TranslationContext, -) -> tuple[list[Activity], TranslationContext]: +) -> tuple[list[Activity], TranslationContext, frozenset[str]]: """Synthesise init SetVariable IR tasks for variables carrying defaultValue. C-05 (VAREX-002): without an explicit ADF SetVariable activity, a @@ -781,13 +782,18 @@ def _build_variable_init_activities( ``{{tasks.X.values.X}}`` reference. This helper emits an init task per default-valued variable so the variable_cache carries a real setter task_key. + + Also returns the exact set of synthesised init task keys so the + package-time pruning in :func:`prepare_workflow` can identify these tasks + explicitly and never act on a customer-authored task. """ from flowx.parser.expression_parser import resolve_expression if not pipeline.variables: - return [], context + return [], context, frozenset() init_tasks: list[Activity] = [] + init_keys: set[str] = set() for var_name, var_def in pipeline.variables.items(): default = var_def.default_value if default is None: @@ -833,12 +839,13 @@ def _build_variable_init_activities( required_parameters=required_parameters, ) init_tasks.append(init_activity) + init_keys.add(task_key) # Register the synthesised setter so @variables('X') resolves to {{tasks._init_X.values.X}}; when # the value is itself a DAB ref (e.g. @utcNow()), inline it directly. dab_ref_value = variable_value if value_kind == "dab_ref" else None context = context.with_variable(var_name, task_key, dab_ref_value=dab_ref_value) context = context.with_activity(init_activity.name, init_activity) - return init_tasks, context + return init_tasks, context, frozenset(init_keys) def _topological_visit(activities: list[AdfActivity]) -> list[AdfActivity]: diff --git a/tests/integration/azure/adf/cases/all-activity-types/expected/bundles/resources/all_activity_types.yml b/tests/integration/azure/adf/cases/all-activity-types/expected/bundles/resources/all_activity_types.yml index 5b97b01f..d0ee6904 100644 --- a/tests/integration/azure/adf/cases/all-activity-types/expected/bundles/resources/all_activity_types.yml +++ b/tests/integration/azure/adf/cases/all-activity-types/expected/bundles/resources/all_activity_types.yml @@ -3,12 +3,6 @@ resources: all_activity_types: name: all-activity-types tasks: - - task_key: _init_status - notebook_task: - notebook_path: ../src/notebooks/init_status.py - base_parameters: - variable_name: status - value: new - task_key: append-variable notebook_task: notebook_path: ../src/notebooks/append_variable.py diff --git a/tests/integration/azure/adf/cases/all-activity-types/expected/bundles/src/notebooks/init_status.py b/tests/integration/azure/adf/cases/all-activity-types/expected/bundles/src/notebooks/init_status.py deleted file mode 100644 index 48cf2b44..00000000 --- a/tests/integration/azure/adf/cases/all-activity-types/expected/bundles/src/notebooks/init_status.py +++ /dev/null @@ -1,21 +0,0 @@ -# Databricks notebook source -# MAGIC %md -# MAGIC # Set Variable: _init_status -# MAGIC -# MAGIC *Auto-generated by flowx. Do not edit manually unless necessary.* - -# COMMAND ---------- - -import json - -variable_name = "status" - -# Read value from widget parameter (set via base_parameters). -# DAB resolves dynamic references (e.g. {{job.run_id}}) before passing. -value = dbutils.widgets.get("value") - -# Set task value so downstream tasks can reference it via -# {{tasks._init_status.values.status}}. -dbutils.jobs.taskValues.set(key=variable_name, value=value) -print(f"Set task value '{variable_name}' = '{value}'") - diff --git a/tests/integration/azure/adf/cases/all-activity-types/expected/ir/all-activity-types.json b/tests/integration/azure/adf/cases/all-activity-types/expected/ir/all-activity-types.json index b90119b6..1b74f872 100644 --- a/tests/integration/azure/adf/cases/all-activity-types/expected/ir/all-activity-types.json +++ b/tests/integration/azure/adf/cases/all-activity-types/expected/ir/all-activity-types.json @@ -21,6 +21,9 @@ "pause_status": "PAUSED", "url": "" }, + "synthesized_variable_init_keys": [ + "_init_status" + ], "tags": { "pipeline": "all-activity-types", "source": "adf" diff --git a/tests/integration/azure/adf/cases/complex-orchestration/expected/bundles/resources/complex_orchestration.yml b/tests/integration/azure/adf/cases/complex-orchestration/expected/bundles/resources/complex_orchestration.yml index f7131ed5..100641a9 100644 --- a/tests/integration/azure/adf/cases/complex-orchestration/expected/bundles/resources/complex_orchestration.yml +++ b/tests/integration/azure/adf/cases/complex-orchestration/expected/bundles/resources/complex_orchestration.yml @@ -3,12 +3,6 @@ resources: complex_orchestration: name: complex-orchestration tasks: - - task_key: _init_state - notebook_task: - notebook_path: ../src/notebooks/init_state.py - base_parameters: - variable_name: state - value: start - task_key: launch-child run_job_task: job_id: ${var.child_pipeline_job_id} diff --git a/tests/integration/azure/adf/cases/complex-orchestration/expected/bundles/src/notebooks/init_state.py b/tests/integration/azure/adf/cases/complex-orchestration/expected/bundles/src/notebooks/init_state.py deleted file mode 100644 index 9a63d077..00000000 --- a/tests/integration/azure/adf/cases/complex-orchestration/expected/bundles/src/notebooks/init_state.py +++ /dev/null @@ -1,21 +0,0 @@ -# Databricks notebook source -# MAGIC %md -# MAGIC # Set Variable: _init_state -# MAGIC -# MAGIC *Auto-generated by flowx. Do not edit manually unless necessary.* - -# COMMAND ---------- - -import json - -variable_name = "state" - -# Read value from widget parameter (set via base_parameters). -# DAB resolves dynamic references (e.g. {{job.run_id}}) before passing. -value = dbutils.widgets.get("value") - -# Set task value so downstream tasks can reference it via -# {{tasks._init_state.values.state}}. -dbutils.jobs.taskValues.set(key=variable_name, value=value) -print(f"Set task value '{variable_name}' = '{value}'") - diff --git a/tests/integration/azure/adf/cases/complex-orchestration/expected/ir/complex-orchestration.json b/tests/integration/azure/adf/cases/complex-orchestration/expected/ir/complex-orchestration.json index 97a2eab2..17971fde 100644 --- a/tests/integration/azure/adf/cases/complex-orchestration/expected/ir/complex-orchestration.json +++ b/tests/integration/azure/adf/cases/complex-orchestration/expected/ir/complex-orchestration.json @@ -11,6 +11,9 @@ "quartz_cron_expression": "0 0 0 * * ?", "timezone_id": "UTC" }, + "synthesized_variable_init_keys": [ + "_init_state" + ], "tags": { "pipeline": "complex-orchestration", "source": "adf" diff --git a/tests/integration/azure/adf/cases/execute-pipeline-pair/expected/bundles/complex_orchestration/resources/complex_orchestration.yml b/tests/integration/azure/adf/cases/execute-pipeline-pair/expected/bundles/complex_orchestration/resources/complex_orchestration.yml index f7131ed5..100641a9 100644 --- a/tests/integration/azure/adf/cases/execute-pipeline-pair/expected/bundles/complex_orchestration/resources/complex_orchestration.yml +++ b/tests/integration/azure/adf/cases/execute-pipeline-pair/expected/bundles/complex_orchestration/resources/complex_orchestration.yml @@ -3,12 +3,6 @@ resources: complex_orchestration: name: complex-orchestration tasks: - - task_key: _init_state - notebook_task: - notebook_path: ../src/notebooks/init_state.py - base_parameters: - variable_name: state - value: start - task_key: launch-child run_job_task: job_id: ${var.child_pipeline_job_id} diff --git a/tests/integration/azure/adf/cases/execute-pipeline-pair/expected/bundles/complex_orchestration/src/notebooks/init_state.py b/tests/integration/azure/adf/cases/execute-pipeline-pair/expected/bundles/complex_orchestration/src/notebooks/init_state.py deleted file mode 100644 index 9a63d077..00000000 --- a/tests/integration/azure/adf/cases/execute-pipeline-pair/expected/bundles/complex_orchestration/src/notebooks/init_state.py +++ /dev/null @@ -1,21 +0,0 @@ -# Databricks notebook source -# MAGIC %md -# MAGIC # Set Variable: _init_state -# MAGIC -# MAGIC *Auto-generated by flowx. Do not edit manually unless necessary.* - -# COMMAND ---------- - -import json - -variable_name = "state" - -# Read value from widget parameter (set via base_parameters). -# DAB resolves dynamic references (e.g. {{job.run_id}}) before passing. -value = dbutils.widgets.get("value") - -# Set task value so downstream tasks can reference it via -# {{tasks._init_state.values.state}}. -dbutils.jobs.taskValues.set(key=variable_name, value=value) -print(f"Set task value '{variable_name}' = '{value}'") - diff --git a/tests/integration/azure/adf/cases/execute-pipeline-pair/expected/ir/complex-orchestration.json b/tests/integration/azure/adf/cases/execute-pipeline-pair/expected/ir/complex-orchestration.json index 97a2eab2..17971fde 100644 --- a/tests/integration/azure/adf/cases/execute-pipeline-pair/expected/ir/complex-orchestration.json +++ b/tests/integration/azure/adf/cases/execute-pipeline-pair/expected/ir/complex-orchestration.json @@ -11,6 +11,9 @@ "quartz_cron_expression": "0 0 0 * * ?", "timezone_id": "UTC" }, + "synthesized_variable_init_keys": [ + "_init_state" + ], "tags": { "pipeline": "complex-orchestration", "source": "adf" diff --git a/tests/integration/azure/adf/cases/full-factory/expected/bundles/all_activity_types/resources/all_activity_types.yml b/tests/integration/azure/adf/cases/full-factory/expected/bundles/all_activity_types/resources/all_activity_types.yml index 5b97b01f..d0ee6904 100644 --- a/tests/integration/azure/adf/cases/full-factory/expected/bundles/all_activity_types/resources/all_activity_types.yml +++ b/tests/integration/azure/adf/cases/full-factory/expected/bundles/all_activity_types/resources/all_activity_types.yml @@ -3,12 +3,6 @@ resources: all_activity_types: name: all-activity-types tasks: - - task_key: _init_status - notebook_task: - notebook_path: ../src/notebooks/init_status.py - base_parameters: - variable_name: status - value: new - task_key: append-variable notebook_task: notebook_path: ../src/notebooks/append_variable.py diff --git a/tests/integration/azure/adf/cases/full-factory/expected/bundles/all_activity_types/src/notebooks/init_status.py b/tests/integration/azure/adf/cases/full-factory/expected/bundles/all_activity_types/src/notebooks/init_status.py deleted file mode 100644 index 48cf2b44..00000000 --- a/tests/integration/azure/adf/cases/full-factory/expected/bundles/all_activity_types/src/notebooks/init_status.py +++ /dev/null @@ -1,21 +0,0 @@ -# Databricks notebook source -# MAGIC %md -# MAGIC # Set Variable: _init_status -# MAGIC -# MAGIC *Auto-generated by flowx. Do not edit manually unless necessary.* - -# COMMAND ---------- - -import json - -variable_name = "status" - -# Read value from widget parameter (set via base_parameters). -# DAB resolves dynamic references (e.g. {{job.run_id}}) before passing. -value = dbutils.widgets.get("value") - -# Set task value so downstream tasks can reference it via -# {{tasks._init_status.values.status}}. -dbutils.jobs.taskValues.set(key=variable_name, value=value) -print(f"Set task value '{variable_name}' = '{value}'") - diff --git a/tests/integration/azure/adf/cases/full-factory/expected/bundles/complex_orchestration/resources/complex_orchestration.yml b/tests/integration/azure/adf/cases/full-factory/expected/bundles/complex_orchestration/resources/complex_orchestration.yml index f7131ed5..100641a9 100644 --- a/tests/integration/azure/adf/cases/full-factory/expected/bundles/complex_orchestration/resources/complex_orchestration.yml +++ b/tests/integration/azure/adf/cases/full-factory/expected/bundles/complex_orchestration/resources/complex_orchestration.yml @@ -3,12 +3,6 @@ resources: complex_orchestration: name: complex-orchestration tasks: - - task_key: _init_state - notebook_task: - notebook_path: ../src/notebooks/init_state.py - base_parameters: - variable_name: state - value: start - task_key: launch-child run_job_task: job_id: ${var.child_pipeline_job_id} diff --git a/tests/integration/azure/adf/cases/full-factory/expected/bundles/complex_orchestration/src/notebooks/init_state.py b/tests/integration/azure/adf/cases/full-factory/expected/bundles/complex_orchestration/src/notebooks/init_state.py deleted file mode 100644 index 9a63d077..00000000 --- a/tests/integration/azure/adf/cases/full-factory/expected/bundles/complex_orchestration/src/notebooks/init_state.py +++ /dev/null @@ -1,21 +0,0 @@ -# Databricks notebook source -# MAGIC %md -# MAGIC # Set Variable: _init_state -# MAGIC -# MAGIC *Auto-generated by flowx. Do not edit manually unless necessary.* - -# COMMAND ---------- - -import json - -variable_name = "state" - -# Read value from widget parameter (set via base_parameters). -# DAB resolves dynamic references (e.g. {{job.run_id}}) before passing. -value = dbutils.widgets.get("value") - -# Set task value so downstream tasks can reference it via -# {{tasks._init_state.values.state}}. -dbutils.jobs.taskValues.set(key=variable_name, value=value) -print(f"Set task value '{variable_name}' = '{value}'") - diff --git a/tests/integration/azure/adf/cases/full-factory/expected/ir/all-activity-types.json b/tests/integration/azure/adf/cases/full-factory/expected/ir/all-activity-types.json index b90119b6..1b74f872 100644 --- a/tests/integration/azure/adf/cases/full-factory/expected/ir/all-activity-types.json +++ b/tests/integration/azure/adf/cases/full-factory/expected/ir/all-activity-types.json @@ -21,6 +21,9 @@ "pause_status": "PAUSED", "url": "" }, + "synthesized_variable_init_keys": [ + "_init_status" + ], "tags": { "pipeline": "all-activity-types", "source": "adf" diff --git a/tests/integration/azure/adf/cases/full-factory/expected/ir/complex-orchestration.json b/tests/integration/azure/adf/cases/full-factory/expected/ir/complex-orchestration.json index 97a2eab2..17971fde 100644 --- a/tests/integration/azure/adf/cases/full-factory/expected/ir/complex-orchestration.json +++ b/tests/integration/azure/adf/cases/full-factory/expected/ir/complex-orchestration.json @@ -11,6 +11,9 @@ "quartz_cron_expression": "0 0 0 * * ?", "timezone_id": "UTC" }, + "synthesized_variable_init_keys": [ + "_init_state" + ], "tags": { "pipeline": "complex-orchestration", "source": "adf" diff --git a/tests/unit/test_control_flow_fixes.py b/tests/unit/test_control_flow_fixes.py new file mode 100644 index 00000000..e788ca02 --- /dev/null +++ b/tests/unit/test_control_flow_fixes.py @@ -0,0 +1,448 @@ +"""Regression tests for three ADF -> Databricks translation fixes. + +1. ForEach inputs-bridge notebooks must carry the ``# Databricks notebook + source`` marker or ``bundle validate`` rejects the notebook_task. +2. A join after an ``IfCondition`` must use ``run_if: NONE_FAILED`` so a + failure in the *taken* branch propagates (``AT_LEAST_ONE_SUCCESS`` masks it). +3. Synthesised ``_init_`` tasks that nothing references must be pruned at + package time, after the agentic merge and after preparation. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import yaml + +from flowx import ir_serde +from flowx.bundler.dab_writer import pipeline_dict_to_ir, write_bundle +from flowx.models.adf_ast import AdfActivity, AdfDefinitions, AdfDependency, AdfPipeline, AdfVariable +from flowx.models.dab import DabNotebook +from flowx.preparer.activity_preparers.naming import notebook_filename +from flowx.preparer.code_generator import render_bridge_notebook +from flowx.preparer.workflow_preparer import ( + PreparedActivity, + PreparedArtifacts, + PreparedWorkflow, + find_dead_variable_inits, + prepare_workflow, +) +from flowx.sources.adf.translate import translate_pipeline + +_DEFS = AdfDefinitions(pipelines=[], datasets={}, linked_services={}, triggers=[]) + +# --------------------------------------------------------------------------- +# Bug 1 — inputs-bridge notebook header +# --------------------------------------------------------------------------- + + +def test_inputs_bridge_starts_with_notebook_source_marker() -> None: + source = render_bridge_notebook( + "['a']", + [], + [], + "items", + title="ForEach inputs bridge: Loop", + ) + assert source.splitlines()[0] == "# Databricks notebook source" + + +def test_inputs_bridge_marker_precedes_imports() -> None: + source = render_bridge_notebook( + "[1, 2]", + ["import json"], + ["item"], + "items", + title="ForEach inputs bridge: Loop", + ) + lines = source.splitlines() + assert lines[0] == "# Databricks notebook source" + assert "dbutils.jobs.taskValues.set(key='items', value=_bridge_value)" in source + + +# --------------------------------------------------------------------------- +# Bug 2 — IfCondition join run_if +# --------------------------------------------------------------------------- + + +def _condition_join_tasks() -> list[dict]: + """A condition task, a true-branch terminal, and a join that depended on the + condition plus an always-on sibling (mirrors a ForEach inputs-bridge).""" + return [ + {"task_key": "CheckRunMode", "condition_task": {"op": "EQUAL_TO"}}, + { + "task_key": "NotifyFull", + "depends_on": [{"task_key": "CheckRunMode", "outcome": "true"}], + }, + {"task_key": "items_bridge"}, + { + "task_key": "ProcessItems", + "depends_on": [ + {"task_key": "CheckRunMode"}, + {"task_key": "items_bridge"}, + ], + }, + ] + + +def _bundled_tasks(tasks: list[dict[str, Any]], output_dir: Path) -> dict[str, dict[str, Any]]: + workflow = PreparedWorkflow(name="job", tasks=tasks, notebooks=[], secrets=[], setup_tasks=[]) + write_bundle(workflow, output_dir) + job = yaml.safe_load((output_dir / "resources" / "job.yml").read_text())["resources"]["jobs"]["job"] + return {task["task_key"]: task for task in job["tasks"]} + + +def _dependency_keys(task: dict[str, Any]) -> set[str]: + return {dependency["task_key"] for dependency in task["depends_on"]} + + +def test_condition_join_uses_none_failed_not_at_least_one_success(tmp_path: Path) -> None: + join = _bundled_tasks(_condition_join_tasks(), tmp_path)["ProcessItems"] + + assert join["run_if"] == "NONE_FAILED" + assert _dependency_keys(join) == {"NotifyFull", "items_bridge"} + + +def test_switch_case_join_uses_none_failed(tmp_path: Path) -> None: + tasks: list[dict[str, Any]] = [ + {"task_key": "RouteMode", "condition_task": {"op": "EQUAL_TO"}}, + { + "task_key": "RouteMode_case_full", + "condition_task": {"op": "EQUAL_TO"}, + "depends_on": [{"task_key": "RouteMode", "outcome": "false"}], + }, + {"task_key": "LoadDelta", "depends_on": [{"task_key": "RouteMode", "outcome": "true"}]}, + {"task_key": "LoadFull", "depends_on": [{"task_key": "RouteMode_case_full", "outcome": "true"}]}, + {"task_key": "Publish", "depends_on": [{"task_key": "RouteMode"}]}, + ] + join = _bundled_tasks(tasks, tmp_path)["Publish"] + + assert join["run_if"] == "NONE_FAILED" + assert _dependency_keys(join) == {"LoadDelta", "LoadFull"} + + +def test_non_join_task_keeps_its_run_if(tmp_path: Path) -> None: + tasks = _condition_join_tasks() + tasks.append({"task_key": "Cleanup", "depends_on": [{"task_key": "items_bridge"}], "run_if": "ALL_DONE"}) + tasks.append({"task_key": "Audit", "depends_on": [{"task_key": "items_bridge"}]}) + by_key = _bundled_tasks(tasks, tmp_path) + + assert by_key["Cleanup"]["run_if"] == "ALL_DONE" + assert "run_if" not in by_key["Audit"] + assert "run_if" not in by_key["NotifyFull"] + + +def test_condition_join_keeps_explicit_run_if(tmp_path: Path) -> None: + tasks = _condition_join_tasks() + next(task for task in tasks if task["task_key"] == "ProcessItems")["run_if"] = "ALL_DONE" + + assert _bundled_tasks(tasks, tmp_path)["ProcessItems"]["run_if"] == "ALL_DONE" + + +# --------------------------------------------------------------------------- +# Bug 3 — dead _init_ pruning: reference matching +# --------------------------------------------------------------------------- + + +def _find_dead( + tasks: list[dict[str, Any]], + init_keys: set[str], + *, + notebooks: tuple[DabNotebook, ...] = (), + inner_workflows: tuple[PreparedWorkflow, ...] = (), +) -> frozenset[str]: + prepared_inits = {key: PreparedActivity(task={"task_key": key}) for key in init_keys} + artifacts = PreparedArtifacts(notebooks=notebooks, inner_workflows=inner_workflows) + return find_dead_variable_inits(tasks, artifacts, prepared_inits) + + +def test_find_dead_reports_init_when_explicit_setter_dominates() -> None: + tasks = [ + {"task_key": "_init_runMode"}, + {"task_key": "SetRunMode"}, + { + "task_key": "CheckRunMode", + "condition_task": {"left": "{{tasks.SetRunMode.values.runMode}}", "right": "full"}, + }, + ] + assert _find_dead(tasks, {"_init_runMode"}) == {"_init_runMode"} + + +def test_find_dead_keeps_init_when_value_is_read() -> None: + tasks = [ + {"task_key": "_init_runMode"}, + { + "task_key": "CheckRunMode", + "condition_task": {"left": "{{tasks._init_runMode.values.runMode}}", "right": "full"}, + }, + ] + assert _find_dead(tasks, {"_init_runMode"}) == frozenset() + + +def test_find_dead_keeps_init_when_referenced_by_depends_on() -> None: + tasks = [ + {"task_key": "_init_runMode"}, + {"task_key": "Next", "depends_on": [{"task_key": "_init_runMode"}]}, + ] + assert _find_dead(tasks, {"_init_runMode"}) == frozenset() + + +def test_find_dead_keeps_init_named_as_a_whole_parameter_value() -> None: + tasks = [ + {"task_key": "_init_items"}, + {"task_key": "AddItem", "notebook_task": {"base_parameters": {"source_task_key": "_init_items"}}}, + ] + assert _find_dead(tasks, {"_init_items"}) == frozenset() + + +def test_find_dead_keeps_init_read_from_notebook_code() -> None: + tasks = [{"task_key": "_init_runMode"}, {"task_key": "Web"}] + notebook = DabNotebook( + relative_path="src/notebooks/web.py", + content="body = f\"{dbutils.jobs.taskValues.get(taskKey='_init_runMode', key='runMode')}\"", + ) + assert _find_dead(tasks, {"_init_runMode"}, notebooks=(notebook,)) == frozenset() + + +def test_find_dead_keeps_init_read_from_inner_workflow() -> None: + tasks = [{"task_key": "_init_runMode"}, {"task_key": "Loop"}] + inner = PreparedWorkflow( + name="Loop_inner_tasks", + tasks=[{"task_key": "Body", "notebook_task": {"base_parameters": {"v": "{{tasks._init_runMode.values.v}}"}}}], + notebooks=[], + secrets=[], + setup_tasks=[], + ) + assert _find_dead(tasks, {"_init_runMode"}, inner_workflows=(inner,)) == frozenset() + + +def test_find_dead_keeps_init_read_by_another_init() -> None: + tasks = [ + {"task_key": "_init_base"}, + {"task_key": "_init_derived", "notebook_task": {"base_parameters": {"v": "{{tasks._init_base.values.base}}"}}}, + {"task_key": "Reader", "notebook_task": {"base_parameters": {"v": "{{tasks._init_derived.values.derived}}"}}}, + ] + assert _find_dead(tasks, {"_init_base", "_init_derived"}) == frozenset() + + +def test_find_dead_reports_whole_unused_init_chain() -> None: + tasks = [ + {"task_key": "_init_a"}, + {"task_key": "_init_b", "notebook_task": {"base_parameters": {"v": "{{tasks._init_a.values.a}}"}}}, + {"task_key": "Reader"}, + ] + assert _find_dead(tasks, {"_init_a", "_init_b"}) == {"_init_a", "_init_b"} + + +def test_find_dead_keeps_init_read_from_notebook_code_with_double_quotes() -> None: + tasks = [{"task_key": "_init_runMode"}, {"task_key": "Web"}] + notebook = DabNotebook( + relative_path="src/notebooks/web.py", + content='mode = dbutils.jobs.taskValues.get(taskKey="_init_runMode", key="runMode")', + ) + assert _find_dead(tasks, {"_init_runMode"}, notebooks=(notebook,)) == frozenset() + + +def test_find_dead_keeps_init_named_in_foreach_inner_job_parameters() -> None: + tasks = [ + {"task_key": "_init_items"}, + { + "task_key": "Loop", + "for_each_task": { + "inputs": "[1, 2]", + "task": { + "task_key": "Loop_iteration", + "run_job_task": {"job_id": "1", "job_parameters": {"source_task_key": "_init_items"}}, + }, + }, + }, + ] + assert _find_dead(tasks, {"_init_items"}) == frozenset() + + +def test_find_dead_ignores_unrelated_field_equal_to_key() -> None: + tasks = [ + {"task_key": "_init_runMode"}, + { + "task_key": "Reader", + "description": "_init_runMode", + "tags": {"origin": "_init_runMode"}, + "sql_task": {"parameters": {"config": {"origin": "_init_runMode"}}}, + }, + ] + assert _find_dead(tasks, {"_init_runMode"}) == {"_init_runMode"} + + +def test_find_dead_only_touches_synthesized_keys() -> None: + """A task the translator did not synthesise is never reported, even when unreferenced.""" + tasks = [{"task_key": "_init_report"}, {"task_key": "Downstream"}] + assert _find_dead(tasks, set()) == frozenset() + + +def test_find_dead_reports_init_despite_prefix_collision() -> None: + """A dead ``_init_x`` is reported even when its key text appears inside an + unrelated key or path; a bare substring match must not keep it alive.""" + tasks = [ + {"task_key": "_init_x"}, + {"task_key": "_init_x_2"}, + { + "task_key": "Reader", + "notebook_task": {"notebook_path": "/Shared/_init_x_report"}, + "condition_task": {"left": "{{tasks._init_x_2.values.x2}}", "right": "2"}, + }, + ] + assert _find_dead(tasks, {"_init_x", "_init_x_2"}) == {"_init_x"} + + +# --------------------------------------------------------------------------- +# Bug 3 — dead _init_ pruning: translate -> report -> package +# --------------------------------------------------------------------------- + + +def _notebook_reading(name: str, variable: str, depends_on: str | None = None) -> AdfActivity: + return AdfActivity( + name=name, + type="DatabricksNotebook", + depends_on=[AdfDependency(activity=depends_on)] if depends_on else None, + type_properties={ + "notebookPath": "/Shared/reader", + "baseParameters": {"v": {"value": f"@variables('{variable}')", "type": "Expression"}}, + }, + ) + + +def _report_dict(pipeline: AdfPipeline) -> dict[str, Any]: + return ir_serde.pipeline_to_dict(translate_pipeline(pipeline, _DEFS).pipeline) + + +def _package(report: dict[str, Any]) -> PreparedWorkflow: + pipeline, _parameters = pipeline_dict_to_ir(report) + return prepare_workflow(pipeline) + + +def _task_keys(workflow: PreparedWorkflow) -> set[str]: + return {task["task_key"] for task in workflow.tasks} + + +def _has_init_notebook(workflow: PreparedWorkflow, init_key: str) -> bool: + filename = notebook_filename(init_key, init_key) + return any(notebook.relative_path.endswith(f"/{filename}") for notebook in workflow.notebooks) + + +def test_dominated_init_is_pruned_with_its_notebook() -> None: + pipeline = AdfPipeline( + name="p", + variables={"runMode": AdfVariable(type="String", default_value="full")}, + activities=[ + AdfActivity( + name="SetRunMode", + type="SetVariable", + type_properties={"variableName": "runMode", "value": "delta"}, + ), + _notebook_reading("Reader", "runMode", depends_on="SetRunMode"), + ], + ) + report = _report_dict(pipeline) + assert "_init_runMode" in {task["task_key"] for task in report["tasks"]} + assert report["synthesized_variable_init_keys"] == ["_init_runMode"] + + workflow = _package(report) + assert "_init_runMode" not in _task_keys(workflow) + assert "SetRunMode" in _task_keys(workflow) + assert not _has_init_notebook(workflow, "_init_runMode") + assert _has_init_notebook(workflow, "SetRunMode") + + +def _unresolved_init_report(*activities: AdfActivity) -> dict[str, Any]: + report = _report_dict( + AdfPipeline( + name="p", + variables={"runMode": AdfVariable(type="String", default_value="full")}, + activities=list(activities), + ) + ) + init_task = next(task for task in report["tasks"] if task["task_key"] == "_init_runMode") + init_task.update(value_kind="unresolved", raw_expression="@pipeline().globalParameters.mode") + return report + + +def _setup_task_keys(workflow: PreparedWorkflow) -> set[str]: + return {setup_task.config.get("task_key") for setup_task in workflow.setup_tasks} + + +def test_dominated_init_is_pruned_with_its_setup_tasks() -> None: + workflow = _package( + _unresolved_init_report( + AdfActivity( + name="SetRunMode", type="SetVariable", type_properties={"variableName": "runMode", "value": "x"} + ), + _notebook_reading("Reader", "runMode", depends_on="SetRunMode"), + ) + ) + assert "_init_runMode" not in _task_keys(workflow) + assert "_init_runMode" not in _setup_task_keys(workflow) + + +def test_read_init_keeps_its_setup_tasks() -> None: + workflow = _package(_unresolved_init_report(_notebook_reading("Reader", "runMode"))) + assert "_init_runMode" in _task_keys(workflow) + assert "_init_runMode" in _setup_task_keys(workflow) + + +def test_read_init_is_kept() -> None: + pipeline = AdfPipeline( + name="p", + variables={"runMode": AdfVariable(type="String", default_value="full")}, + activities=[_notebook_reading("Reader", "runMode")], + ) + workflow = _package(_report_dict(pipeline)) + assert "_init_runMode" in _task_keys(workflow) + assert _has_init_notebook(workflow, "_init_runMode") + + +def test_init_feeding_append_variable_is_kept() -> None: + """AppendVariable binds its prior writer only at prepare time, so IR-level pruning would drop this init.""" + pipeline = AdfPipeline( + name="p", + variables={"items": AdfVariable(type="Array", default_value=[])}, + activities=[ + AdfActivity(name="AddItem", type="AppendVariable", type_properties={"variableName": "items", "value": "a"}), + ], + ) + workflow = _package(_report_dict(pipeline)) + assert "_init_items" in _task_keys(workflow) + + +def test_init_read_by_merged_agentic_task_is_kept() -> None: + pipeline = AdfPipeline( + name="p", + variables={"runMode": AdfVariable(type="String", default_value="full")}, + activities=[ + AdfActivity( + name="SetRunMode", + type="SetVariable", + type_properties={"variableName": "runMode", "value": "delta"}, + ), + ], + ) + report = _report_dict(pipeline) + report["tasks"].append( + { + "name": "AgentTask", + "task_key": "AgentTask", + "type": "NotebookActivity", + "notebook_path": "/Shared/agent", + "base_parameters": {"v": "{{tasks._init_runMode.values.runMode}}"}, + } + ) + workflow = _package(report) + assert "_init_runMode" in _task_keys(workflow) + + +def test_report_omits_init_keys_without_default_valued_variables() -> None: + pipeline = AdfPipeline( + name="p", + activities=[AdfActivity(name="W", type="Wait", type_properties={"waitTimeInSeconds": 1})], + ) + assert "synthesized_variable_init_keys" not in _report_dict(pipeline)