Bundle packaging modes + ordered multi-bundle deploy - #73
Open
zanitarahimi wants to merge 9 commits into
Open
zanitarahimi wants to merge 9 commits into
zanitarahimi wants to merge 9 commits into
Conversation
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
requested changes
Sep 14, 2026
ghanse
left a comment
ghanse
left a comment
Collaborator
There was a problem hiding this comment.
This looks good. I left a few comments. It would be good to add documentation for this feature.
Comment on lines
+199
to
+255
| 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) |
Collaborator
There was a problem hiding this comment.
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.
Contributor
Author
There was a problem hiding this comment.
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.
Collaborator
There was a problem hiding this comment.
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.
Comment on lines
+39
to
+40
| class PipelineCycleError(Exception): | ||
| """Raised when the Run Pipeline dependency graph contains a cycle (cannot be ordered).""" |
Collaborator
There was a problem hiding this comment.
Let's open an issue to move error classes into an errors.py module. This can be done in a separate PR.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
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.bundler/pipeline_graph.py.DEPLOY.mdrecords the suggested callees-first deploy order for every mode.SETUP.md.Ordered deploy
flowx deploy(bundler/deployer.py) discovers the bundles, orders them callees-first, deploys each withdatabricks bundle deploy, reads each deployed job's numeric id frombundle summary, and injects it into callers via--var <callee>_job_id=<id>— no manual job-id wiring.Global-param hoisting across grouped workflows
single/per-group), global-parameter hoisting is union at the bundle level (databricks.ymlvariables:+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. AddsTestHoistedGlobalsAcrossGroupedWorkflows.Testing
make fmt(ruff + mypy) clean;make testandmake integrationgreen.${var.<callee>_job_id}resolves to the deployed job id.