Bundle packaging modes + ordered multi-bundle deploy by zanitarahimi · Pull Request #73 · databricks-solutions/flowx

zanitarahimi · 2026-09-11T13:11:24Z

Bundle packaging modes + ordered multi-bundle deploy

Implements the bundle-packaging design and the ordered auto-deploy follow-up.

Packaging modes

--packaging-mode: per-pipeline (default) / single / per-group; per-group supports --group-by inferred (Run Pipeline call graph) and --group-by spec.
Run Pipeline dependencies tracked in bundler/pipeline_graph.py.
A single top-level DEPLOY.md records the suggested callees-first deploy order for every mode.
Within each bundle: generated code + setup scripts de-duped, one SETUP.md.

Ordered deploy

flowx deploy (bundler/deployer.py) discovers the bundles, orders them callees-first, deploys each with databricks bundle deploy, reads each deployed job's numeric id from bundle summary, and injects it into callers via --var <callee>_job_id=<id> — no manual job-id wiring.

Global-param hoisting across grouped workflows

When several workflows share one bundle (single / per-group), global-parameter hoisting is union at the bundle level (databricks.yml variables: + SETUP.md) and per-workflow at each job (a widget binds to ${var.X} only in pipelines that declare X). Single-workflow bundles are byte-identical to before. Adds TestHoistedGlobalsAcrossGroupedWorkflows.

Testing

make fmt (ruff + mypy) clean; make test and make integration green.
Ordered-deploy validated on a real multi-pipeline factory: callees deploy first and each caller's ${var.<callee>_job_id} resolves to the deployed job id.

ExecutePipeline emits run_job_task.job_id = ${resources.jobs.X.id}, which only resolves when X is a job in this bundle. In a multi-pipeline migration each ADF pipeline becomes its own bundle, so a ref to a sibling pipeline points at a node that does not exist here and `bundle deploy` fails with "no such node resources.jobs.X". Add _rewrite_cross_bundle_run_job_refs: before databricks.yml/resource YAML are written, rewrite run_job_task refs to out-of-bundle jobs into ${var.X} and register X in _cross_bundle_variables (which the existing YAML builder declares). Operator supplies the numeric job id at deploy via --var, as SETUP.md documents. Recurses into for_each_task bodies. This is the stopgap that #10 (ordered cross-pipeline deploy from control lineage) builds on and keeps as the fallback for unresolved callees. Closes #23 Co-authored-by: Isaac

The rewrite count was returned but discarded at both call sites. The function's real output is its in-place mutation of cross_bundle_variables (declared in databricks.yml and surfaced in SETUP.md), so return None and remove the dead counter. Co-authored-by: Isaac

- Remove _rewrite_cross_bundle_run_job_refs from dab_writer: it was replaced by _rewrite_cross_bundle_job_references during the main rebase and had no remaining callers. Drop its now-unused CROSS_BUNDLE_JOB_ID_REF import. - Fix stale comments that named the removed function and claimed the old bare ${var.X} scheme (pipeline_graph.py, deployer.py docstring); the live scheme is ${var.X_job_id}. - deployer.run(): check for the `databricks` CLI on PATH up front and return an actionable error instead of an uncaught FileNotFoundError. Skipped for --dry-run, which never shells out. Add tests.

Declare the union of every workflow's hoisted globals in databricks.yml + SETUP.md, but pass each job only its own workflow's globals. Single-workflow bundles unchanged. Adds TestHoistedGlobalsAcrossGroupedWorkflows.

ghanse

This looks good. I left a few comments. It would be good to add documentation for this feature.

ghanse · 2026-09-14T13:08:36Z

+def _namespace_bundle_artifacts(workflow: PreparedWorkflow, prefix: str) -> None:
+    """Namespaces a workflow's notebooks and inner ForEach job keys by *prefix*, in place.
+
+    When several pipelines share one bundle (``single`` / ``per-group`` modes), their notebook file
+    paths (derived from activity names) and inner ForEach job keys (``<task_key>_inner_tasks``,
+    unique only within a pipeline) can collide in the shared ``resources/`` and ``src/`` dirs — two
+    pipelines writing the same path, the second silently overwriting the first while both jobs still
+    reference it. Prefixing every such artifact with the owning pipeline key makes them unique.
+
+    Rewrites, consistently:
+      * each notebook ``relative_path`` (``notebooks/x.py`` -> ``notebooks/<prefix>/x.py``) and every
+        ``../src/notebooks/x.py`` task reference to it;
+      * each inner ForEach job ``name`` (so its resource key becomes ``<prefix>__<key>``) and every
+        ``${resources.jobs.<key>.id}`` ref to it.
+
+    Pipeline-level job keys are NOT namespaced: pipeline names are unique within a migration, so they
+    never collide, and namespacing them would break cross-bundle ``${var.<callee>}`` wiring.
+    """
+    replacements: dict[str, str] = {}
+
+    # 1. Inner ForEach job keys: rename inner.name, map old resources.jobs ref -> new.
+    seen_new_keys: dict[str, str] = {}
+    for inner in workflow.inner_workflows:
+        old_key = normalize_task_key(inner.name)
+        inner.name = f"{prefix}__{inner.name}"
+        new_key = normalize_task_key(inner.name)
+        # normalize_task_key collapses the "__" separator to "_", so a pipeline name containing "__"
+        # could make two inner jobs land on the same namespaced key (e.g. prefix "a" + inner "b__c" vs
+        # prefix "a__b" + inner "c" both -> "a_b_c"), silently overwriting one resource file. Refuse
+        # rather than ship a corrupt bundle — this requires pathological ADF names but is cheap to catch.
+        if new_key in seen_new_keys:
+            raise ValueError(
+                f"Namespacing inner ForEach jobs under prefix '{prefix}' produced a duplicate resource "
+                f"key '{new_key}' (from inner jobs '{seen_new_keys[new_key]}' and '{inner.name}'). "
+                "Rename the offending pipeline/activity so keys don't collide after normalization."
+            )
+        seen_new_keys[new_key] = inner.name
+        if old_key != new_key:
+            replacements[f"${{resources.jobs.{old_key}.id}}"] = f"${{resources.jobs.{new_key}.id}}"
+
+    # 2. Notebook paths: prefix each unique relative_path with the pipeline key as a subdirectory.
+    for wf in (workflow, *workflow.inner_workflows):
+        for notebook in wf.notebooks:
+            old_path = notebook.relative_path
+            new_path = _prefixed_notebook_relative_path(old_path, prefix)
+            if old_path != new_path:
+                notebook.relative_path = new_path
+                replacements[f"../src/{old_path}"] = f"../src/{new_path}"
+                # Some generated bodies reference their own bundle-relative path (e.g. the Spark-Python
+                # placeholder's `databricks fs cp ... src/<path>` download hint). Rewrite that too so an
+                # operator following the instruction downloads to where the task now looks.
+                if f"src/{old_path}" in notebook.content:
+                    notebook.content = notebook.content.replace(f"src/{old_path}", f"src/{new_path}")
+
+    # 3. Rewrite every matching ref across the parent and inner task trees.
+    for wf in (workflow, *workflow.inner_workflows):
+        _rewrite_task_string_values(wf.tasks, replacements)


I think this is possibly missing dbt factory files. These may also collide during consolidation.

It may also be best to separate this function into submodules for each artifact type.

dbt-factory artifacts only come from Airflow, and multi-DAG Airflow already goes through a different namespacer (_namespace_workflow_assets) that handles them, so this ADF path never actually receives dbt today. But you're right the two namespacers have drifted apart and this one would mishandle dbt if it ever did.
Let's create a separate issue for separating this function into submodules for each artifact type.

Sounds good, I opened #101 for this.

ghanse · 2026-09-14T13:09:31Z

dab_writer.py is getting quite big. Let's open an issue to split it into a package similar to #54 . This can be done as a separate PR.

ghanse · 2026-09-14T13:11:19Z

+class PipelineCycleError(Exception):
+    """Raised when the Run Pipeline dependency graph contains a cycle (cannot be ordered)."""


Let's open an issue to move error classes into an errors.py module. This can be done in a separate PR.

ghanse · 2026-09-18T02:48:13Z

+        bundle_dir, pipeline_keys = groups[0]
+        lines += [
+            "All pipelines are packaged into a **single bundle** at the migration output root. Deploy it "
+            "directly — there is no cross-bundle ordering to worry about:",


This says there's no cross-bundle ordering to worry about, but a single bundle can still carry a ${var.<callee>_job_id} reference to a pipeline outside the migration (declared with no default). A plain databricks bundle deploy then fails on the unset var, and flowx deploy errors with MissingDependencyError unless --allow-missing-deps. I think it's worth a note here pointing users with external references at SETUP.md.

ghanse · 2026-09-18T02:49:41Z

+    return graph
+
+
+def _topo_sort(graph: dict[str, list[str]]) -> list[str]:


This is a second Kahn's-algorithm topo sort next to pipeline_graph.topo_order. Since the errors.py refactor is already planned, worth folding these together so the two copies don't drift.

ghanse · 2026-09-18T02:50:19Z

+        assert len(groups) == 1
+        assert groups[0][0] == "z_root"
+
+    def test_per_group_spec_honors_mapping(self):


Could you also add tests for the loud-failure guards the PR builds — the spec-absent pipeline whose key collides with an explicit group name, and the per-group + --group-by spec missing --group-spec error? Cheap to add and they lock in the intended failures.

ghanse · 2026-09-18T02:52:05Z

+    return _VAR_JOB_ID_SUFFIX.sub("", var_name)
+
+
+class CycleError(Exception):


CycleError here duplicates PipelineCycleError in pipeline_graph.py — same job, two classes. Worth consolidating as part of the planned errors.py refactor.

This comment was generated with GitHub MCP.

ghanse

Strong, well-tested PR overall. One change I'd like before approving:

deploy_writer.py — single-bundle DEPLOY.md. It tells the operator there's no cross-bundle ordering to worry about and to just run databricks bundle deploy, but a single bundle can still carry a ${var.<callee>_job_id} reference to a pipeline outside the migration (declared with no default). In that case a plain deploy fails on the unset var, and flowx deploy errors with MissingDependencyError unless --allow-missing-deps. Please add a note covering that external-reference case (see the inline comment).

The other inline comments — the duplicate topo sort / CycleError, and the loud-failure test gaps — are non-blocking.

matthewmoorcroft and others added 8 commits September 7, 2026 17:55

initial commit, introduce flowx deploy skill

1b22101

address review: mode-switch safety, non-mutating writer, cleaner errors

8254192

minor reformatting

22f45e5

Fix global-param hoisting across grouped workflows

b9a83cf

Declare the union of every workflow's hoisted globals in databricks.yml + SETUP.md, but pass each job only its own workflow's globals. Single-workflow bundles unchanged. Adds TestHoistedGlobalsAcrossGroupedWorkflows.

Fix missing-CLI guard breaking deployer tests without databricks on PATH

c57bfaa

zanitarahimi requested review from ghanse and lorenzorubi-db September 11, 2026 14:22

ghanse requested changes Sep 14, 2026

View reviewed changes

ghanse added the Needs changes label Sep 14, 2026

add documentation and remove unrelated changes

0eaab1d

zanitarahimi requested a review from ghanse September 14, 2026 16:50

This was referenced Sep 18, 2026

[FEATURE]: Decompose DAB writer #101

Open

[FEATURE]: Centralize error classes #102

Open

ghanse reviewed Sep 18, 2026

View reviewed changes

ghanse requested changes Sep 18, 2026

View reviewed changes

Navigation Menu

Bundle packaging modes + ordered multi-bundle deploy - #73

Bundle packaging modes + ordered multi-bundle deploy#73
zanitarahimi wants to merge 9 commits into
mainfrom
port/pr32-ordered-deploy

zanitarahimi commented Sep 11, 2026 •

edited

Loading

Uh oh!

ghanse left a comment

Uh oh!

Uh oh!

Uh oh!

ghanse Sep 14, 2026

Uh oh!

zanitarahimi Sep 14, 2026

Uh oh!

ghanse Sep 18, 2026

Uh oh!

ghanse Sep 14, 2026

Uh oh!

ghanse Sep 14, 2026

Uh oh!

ghanse Sep 18, 2026 •

edited

Loading

Uh oh!

ghanse Sep 18, 2026 •

edited

Loading

Uh oh!

ghanse Sep 18, 2026 •

edited

Loading

Uh oh!

ghanse Sep 18, 2026

Uh oh!

ghanse left a comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

		class PipelineCycleError(Exception):
		"""Raised when the Run Pipeline dependency graph contains a cycle (cannot be ordered)."""

		return graph


		def _topo_sort(graph: dict[str, list[str]]) -> list[str]:

		return _VAR_JOB_ID_SUFFIX.sub("", var_name)


		class CycleError(Exception):

Conversation

zanitarahimi commented Sep 11, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!