Implement Cluster 1 Lakeflow control-flow fidelity - #24
Merged
engineeringmadness merged 6 commits intoSep 4, 2026
Conversation
Add dbutils.jobs.taskValues set/get backed by a runner-scoped store, seeded from base_parameters. Shared workflow tasks see values immediately; isolated notebook.run() buffers sets until the child returns. Co-authored-by: Karan Gupta <engineeringmadness@users.noreply.github.com>
Parse task-level run_if and depends_on outcome, skip ineligible tasks as SKIPPED, and continue the DAG after failures so ALL_FAILED / ALL_DONE downstream tasks can still run. Re-raise the first task error at the end. Co-authored-by: Karan Gupta <engineeringmadness@users.noreply.github.com>
Allow condition_task without a notebook, evaluate EQUAL/NOT_EQUAL/ GREATER_THAN (and Databricks EQUAL_TO aliases) over literals, taskValues, and job parameters, then skip the non-matching branch via depends_on outcome. DAG nodes are now task keys so condition tasks can participate in topology. Co-authored-by: Karan Gupta <engineeringmadness@users.noreply.github.com>
Parse for_each_task inputs (JSON list, literal list, or taskValues),
run the nested notebook once per input with {{input}} params, isolate
env between iterations, and fail the parent task on the first child error.
Concurrency is accepted but iterations run sequentially.
Co-authored-by: Karan Gupta <engineeringmadness@users.noreply.github.com>
Allow run_workflow(only=[...]) and run_workflow(from_task=...) to re-run a subgraph. Unselected tasks are treated as SUCCESS so depends_on still resolves. Document overwrite vs append repair idempotency in the README. Co-authored-by: Karan Gupta <engineeringmadness@users.noreply.github.com>
Parse max_retries, min_retry_interval_millis, and timeout_seconds from workflow JSON. Retry failed notebook and for_each tasks with optional sleep; log timeout_seconds without enforcing it. Document Cluster 1 control-flow extras in the README. Co-authored-by: Karan Gupta <engineeringmadness@users.noreply.github.com>
engineeringmadness
marked this pull request as ready for review
September 4, 2026 15:31
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
2 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.
Implement Cluster 1 from
docs/superpowers/specs/2026-09-03-spark-write-lakeflow-roadmap-design.md(T1–T6). Each item was implemented, covered with tests, and the full suite stayed green.T1 — taskValues:
dbutils.jobs.taskValues.set/.getwith a runner-scoped store, seeded frombase_parameters. Shared workflow tasks see values immediately; isolatednotebook.run()bufferssetuntil return.T2 — run_if / depends_on outcome:
ALL_SUCCESS,ALL_FAILED,AT_LEAST_ONE_SUCCESS,ALL_DONE,NONE_FAILED,AT_LEAST_ONE_FAILED. Ineligible tasks areSKIPPED. Failures no longer abort the scheduler immediately; remaining eligible tasks run, then the first error is re-raised.T3 — condition_task: Evaluate
EQUAL/EQUAL_TO,NOT_EQUAL,GREATER_THAN, etc. over literals,{{tasks.<key>.values.<name>}}, and{{job.parameters.<name>}}. Matchingdepends_on[].outcome(true/false) runs; the other branch is skipped. DAG nodes are task keys.T4 — for_each_task: Sequential expansion of JSON/list/
taskValuesinputs. Nested notebook runs once per input with{{input}}params and env isolation. First child failure fails the parent (no later iterations).concurrencyis accepted, not parallelized.T5 — repair-and-rerun:
run_workflow(only=[...])andfrom_task=. Unselected tasks are treated asSUCCESS. README documents overwrite vs append repair.T6 — retries / timeouts:
max_retriesandmin_retry_interval_millisretry notebook and for_each tasks.timeout_secondsis logged and not enforced.Verification:
python3.14 -m coverage run -m pytest tests/ -v→ 243 passed.