Fix three ADF -> Databricks control-flow / codegen bugs by alexnastetsky · Pull Request #151 · databricks-solutions/flowx

alexnastetsky · 2026-10-12T18:29:01Z

Port of databricks-field-eng/flowx#17 onto the restructured codebase (translator/engine.py is now sources/adf/translate.py + ir_serde.py). Supersedes #40.

ForEach inputs-bridge notebooks carry the # Databricks notebook source marker so bundle validate accepts the notebook_task.
- Already fixed on main by the render_bridge_notebook consolidation. No source change here — this PR only adds a guard test.
A join after an IfCondition (or Switch, which lowers to condition tasks) uses run_if: NONE_FAILED so a failure in the taken branch propagates (AT_LEAST_ONE_SUCCESS let a sibling success mask it).
Synthesised _init_<var> tasks that nothing references are pruned.

Changes

Where pruning runs. #17 pruned during IR serialisation. Here it runs at the end of prepare_workflow, which both packaging paths share. Only at that point are all references final:

the agentic merge has happened, so an agent-written task may read {{tasks._init_X.values.X}};
the preparers have created references the IR never shows: AppendVariable's source_task_key, ForEach inner-job parameters, and taskValues.get(taskKey=...) reads in generated notebooks.

Pruning on the IR would drop an init that an AppendVariable still needs.

How the pruned tasks are identified.

The translator records the exact synthesised keys in Pipeline.synthesized_variable_init_keys.
That set is serialised into translation_report.json only when non-empty, so Airflow reports are unchanged.
It is rehydrated by pipeline_dict_to_ir.
Only those keys are eligible for pruning, so a customer-authored task is never touched.

What counts as a reference. Only these structured forms count:

a tasks.<key>.values token;
a depends_on edge;
a parameter whose whole value is the key;
a taskKey='<key>' read in notebook code.

So a sibling key like _init_x_2, or a path like /Shared/_init_x_report, never keeps a dead _init_x alive. An init read by another init stays live. A pruned init also drops its notebook and setup tasks.

A dead init still appears in translation_report.json; it is removed only from the generated bundle.

Tests

manually tested
added unit tests
added integration tests
tests/unit/test_control_flow_fixes.py covers:
- reference matching;
- end-to-end translate → report → package for: a dominated init (pruned along with its notebook), a read init, an init feeding an AppendVariable, and an init read by a merged agentic task.
Results:
- pytest tests/unit: 1470 passed. The 2 failures in test_airflow_provider_sync.py are environmental (gpg signing in the fixture's git commit) and also fail on main.
- pytest tests/integration -m "not slow and not integration": 141 passed.
- ruff and mypy are clean.
Before/after comparison: I packaged every ADF fixture in tests/resources/json on main and on this branch.
- 24 _init_ tasks and their notebooks are gone. A substring search of main's packaged output finds no other mention of any of them, so they were dead.
- Two joins changed from AT_LEAST_ONE_SUCCESS to NONE_FAILED.
- Nothing else differs.

This pull request and its description were written by Isaac.

Port of databricks-field-eng/flowx#17 onto the restructured codebase. 1. ForEach inputs-bridge notebooks carry the `# Databricks notebook source` marker. Already fixed on main by the render_bridge_notebook consolidation; this only adds a guard test. 2. A join after an IfCondition/Switch uses run_if NONE_FAILED so a failure in the taken branch propagates (AT_LEAST_ONE_SUCCESS let a sibling success mask it). 3. Synthesised _init_<var> tasks that nothing references are pruned. The translator records the exact synthesised keys on the Pipeline IR (serialised in translation_report.json), and pruning runs at the end of prepare_workflow -- after the agentic merge and after the preparers have materialised references the IR never shows (AppendVariable's source_task_key, ForEach inner-job params, taskValues reads in generated notebooks). Only structured reference forms count, so an unrelated key/path containing the key text never keeps a dead init alive. Pruned inits also drop their notebook and setup tasks. Co-authored-by: Isaac <no-reply@databricks.com>

ghanse

Looks good overall. The join fix and the report round-trip are solid. One gap in pruning plus a few test holes are noted inline.

ghanse · 2026-10-12T19:29:57Z

+        for task in _iter_workflow_tasks(tasks, artifacts.inner_workflows)
+    ]
+    dead_keys: set[str] = set()
+    for key in init_keys:


This is a single pass, so a dead chain only loses its tail. If _init_b reads _init_a and nothing reads _init_b, _init_b is pruned but _init_a stays, because _init_b still pointed at it. Could we loop until nothing new is dead, or work out what's reachable from the real tasks first?

ghanse · 2026-10-12T19:29:57Z

+    assert _find_dead(tasks, {"_init_runMode"}, inner_workflows=(inner,)) == frozenset()
+
+
+def test_find_dead_keeps_init_read_by_another_init() -> None:


This covers a chain where every init is used. Can we add one where the whole chain is unused (_init_b reads _init_a, nothing reads _init_b) and assert both get pruned?

ghanse · 2026-10-12T19:29:57Z

+    ]
+
+
+def test_condition_join_uses_none_failed_not_at_least_one_success() -> None:


Nit: only IfCondition is covered. A Switch case, plus a check that a non-join task keeps its run_if, would pin the scope down.

ghanse · 2026-10-12T19:29:57Z

+    assert _find_dead(tasks, {"_init_items"}) == frozenset()
+
+
+def test_find_dead_keeps_init_read_from_notebook_code() -> None:


Nit: _notebook_references also matches taskKey="...", but only single quotes are tested here. Worth adding the double-quote case.

ghanse · 2026-10-12T19:29:57Z

+    return any(notebook.relative_path.endswith(f"/{filename}") for notebook in workflow.notebooks)
+
+
+def test_dominated_init_is_pruned_with_its_notebook() -> None:


Nit: this checks the notebook is dropped, but not the init's setup tasks. A test where a real ForEach inner-job parameter keeps an init alive would help too.

ghanse reviewed Oct 12, 2026

View reviewed changes

ghanse added the Needs changes label Oct 12, 2026

Navigation Menu

Repository navigation

Fix three ADF -> Databricks control-flow / codegen bugs - #151

Fix three ADF -> Databricks control-flow / codegen bugs#151
alexnastetsky wants to merge 1 commit into
databricks-solutions:mainfrom
alexnastetsky:fix/adf-control-flow-codegen-bugs

alexnastetsky commented Oct 12, 2026

Uh oh!

ghanse left a comment

Uh oh!

ghanse Oct 12, 2026

Uh oh!

ghanse Oct 12, 2026

Uh oh!

ghanse Oct 12, 2026

Uh oh!

ghanse Oct 12, 2026

Uh oh!

ghanse Oct 12, 2026

Uh oh!

This branch has not been deployed

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

		assert _find_dead(tasks, {"_init_runMode"}, inner_workflows=(inner,)) == frozenset()


		def test_find_dead_keeps_init_read_by_another_init() -> None:

		]


		def test_condition_join_uses_none_failed_not_at_least_one_success() -> None:

		assert _find_dead(tasks, {"_init_items"}) == frozenset()


		def test_find_dead_keeps_init_read_from_notebook_code() -> None:

		return any(notebook.relative_path.endswith(f"/{filename}") for notebook in workflow.notebooks)


		def test_dominated_init_is_pruned_with_its_notebook() -> None:

Conversation

alexnastetsky commented Oct 12, 2026

Changes

Tests

Uh oh!

ghanse left a comment

Choose a reason for hiding this comment

Uh oh!

ghanse Oct 12, 2026

Choose a reason for hiding this comment

Uh oh!

ghanse Oct 12, 2026

Choose a reason for hiding this comment

Uh oh!

ghanse Oct 12, 2026

Choose a reason for hiding this comment

Uh oh!

ghanse Oct 12, 2026

Choose a reason for hiding this comment

Uh oh!

ghanse Oct 12, 2026

Choose a reason for hiding this comment

Uh oh!

This branch has not been deployed

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants