From d0a7690c2039433e9e3a9198ea6a1e4209bb8f8c Mon Sep 17 00:00:00 2001 From: Grigory Panov Date: Thu, 23 Jul 2026 18:31:42 +0200 Subject: [PATCH 1/2] localenv: target a specific job task, not the whole job (--job-task) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the whole-job --job-id flag with a task-scoped --job-task .. A job can bind multiple tasks to different serverless environment versions (Workflows notebook-environments PRD), so a bare job ID cannot resolve to a single environment; the target must be a specific task. Behavior (per [P0] CLI Changes spec, task D5): - --job-task . resolves the named task's compute. The value is split on the FIRST dot, so task keys may contain dots. - A serverless task's environment version is read directly from the environment it binds (environment_key -> spec version), so the job path no longer applies the serverless-vN default fallback — that now belongs solely to the bundle path. A task whose environment records no version is an E_RESOLVE, not a guess. - A classic task resolves from its own new_cluster (or existing_cluster_id, via the Clusters API). - A bare --job-task (no task key) is E_USAGE, listing the job's task keys so the user can pick one. Because listing requires a Jobs API call, this E_USAGE is emitted at the resolve phase; the error-code annotation in result.go is updated to note E_USAGE can also come from resolve for this case. - An unknown task key is E_RESOLVE, also listing the available keys. The ComputeClient seam gains JobTaskEnvironment (replacing GetJobSparkVersion) and a typed ErrTaskKeyRequired so ResolveTarget can classify the missing-key case as E_USAGE vs a genuine E_RESOLVE. Acceptance tests reworked: job-classic-check and job-serverless-check now bind a task; the old whole-job ambiguity cases (both-compute, multi-cluster mismatch, serverless-version mismatch) are obsolete under task scoping and are replaced by job-task-missing-key (E_USAGE), job-task-unknown (E_RESOLVE), and job-task-unpinned (E_RESOLVE, no fallback). serverless-default-check is removed: the job path no longer defaults, and the bundle default remains unit-tested. Co-authored-by: Isaac --- acceptance/localenv/help/output.txt | 2 +- .../localenv/job-ambiguous-compute/script | 1 - .../localenv/job-ambiguous-compute/test.toml | 24 ---- .../localenv/job-classic-check/output.txt | 2 +- acceptance/localenv/job-classic-check/script | 2 +- .../localenv/job-classic-check/test.toml | 6 +- .../localenv/job-multicluster-mismatch/script | 1 - .../job-multicluster-mismatch/test.toml | 22 ---- .../localenv/job-serverless-check/output.txt | 2 +- .../localenv/job-serverless-check/script | 2 +- .../localenv/job-serverless-check/test.toml | 8 +- .../job-serverless-version-mismatch/script | 1 - .../job-serverless-version-mismatch/test.toml | 22 ---- .../out.test.toml | 0 .../output.txt | 2 +- .../localenv/job-task-missing-key/script | 1 + .../localenv/job-task-missing-key/test.toml | 26 +++++ .../out.test.toml | 0 .../output.txt | 2 +- acceptance/localenv/job-task-unknown/script | 1 + .../localenv/job-task-unknown/test.toml | 23 ++++ .../out.test.toml | 0 .../output.txt | 2 +- acceptance/localenv/job-task-unpinned/script | 1 + .../localenv/job-task-unpinned/test.toml | 25 +++++ acceptance/localenv/json-error/output.txt | 2 +- acceptance/localenv/no-target/output.txt | 2 +- .../serverless-default-check/out.test.toml | 3 - .../serverless-default-check/output.txt | 13 --- .../localenv/serverless-default-check/script | 1 - .../serverless-default-check/test.toml | 39 ------- cmd/environments/compute.go | 106 ++++++++++-------- cmd/environments/sync.go | 10 +- libs/localenv/result.go | 2 +- libs/localenv/target.go | 77 +++++++++---- libs/localenv/target_test.go | 86 +++++++++----- 36 files changed, 274 insertions(+), 245 deletions(-) delete mode 100644 acceptance/localenv/job-ambiguous-compute/script delete mode 100644 acceptance/localenv/job-ambiguous-compute/test.toml delete mode 100644 acceptance/localenv/job-multicluster-mismatch/script delete mode 100644 acceptance/localenv/job-multicluster-mismatch/test.toml delete mode 100644 acceptance/localenv/job-serverless-version-mismatch/script delete mode 100644 acceptance/localenv/job-serverless-version-mismatch/test.toml rename acceptance/localenv/{job-ambiguous-compute => job-task-missing-key}/out.test.toml (100%) rename acceptance/localenv/{job-serverless-version-mismatch => job-task-missing-key}/output.txt (54%) create mode 100644 acceptance/localenv/job-task-missing-key/script create mode 100644 acceptance/localenv/job-task-missing-key/test.toml rename acceptance/localenv/{job-multicluster-mismatch => job-task-unknown}/out.test.toml (100%) rename acceptance/localenv/{job-multicluster-mismatch => job-task-unknown}/output.txt (52%) create mode 100644 acceptance/localenv/job-task-unknown/script create mode 100644 acceptance/localenv/job-task-unknown/test.toml rename acceptance/localenv/{job-serverless-version-mismatch => job-task-unpinned}/out.test.toml (100%) rename acceptance/localenv/{job-ambiguous-compute => job-task-unpinned}/output.txt (51%) create mode 100644 acceptance/localenv/job-task-unpinned/script create mode 100644 acceptance/localenv/job-task-unpinned/test.toml delete mode 100644 acceptance/localenv/serverless-default-check/out.test.toml delete mode 100644 acceptance/localenv/serverless-default-check/output.txt delete mode 100644 acceptance/localenv/serverless-default-check/script delete mode 100644 acceptance/localenv/serverless-default-check/test.toml diff --git a/acceptance/localenv/help/output.txt b/acceptance/localenv/help/output.txt index f96a2db80cf..b024e671819 100644 --- a/acceptance/localenv/help/output.txt +++ b/acceptance/localenv/help/output.txt @@ -15,7 +15,7 @@ Flags: --constraints-only apply the Python version and constraints without adding the databricks-connect dependency --dry-run compute the plan without writing files or provisioning -h, --help help for setup-local - --job-id string job ID to use as the compute target + --job-task string job task to use as the compute target, as . (the task key is required) --serverless-version string serverless version to use as the compute target (e.g. 5) Global Flags: diff --git a/acceptance/localenv/job-ambiguous-compute/script b/acceptance/localenv/job-ambiguous-compute/script deleted file mode 100644 index bd37e81d924..00000000000 --- a/acceptance/localenv/job-ambiguous-compute/script +++ /dev/null @@ -1 +0,0 @@ -musterr $CLI environments setup-local --job-id 12345 --dry-run diff --git a/acceptance/localenv/job-ambiguous-compute/test.toml b/acceptance/localenv/job-ambiguous-compute/test.toml deleted file mode 100644 index 704ed044370..00000000000 --- a/acceptance/localenv/job-ambiguous-compute/test.toml +++ /dev/null @@ -1,24 +0,0 @@ -EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] - -[Env] -DATABRICKS_LOCALENV_CONSTRAINT_SOURCE = "$DATABRICKS_HOST" - -# A job that declares both serverless environments and classic job clusters is -# ambiguous: its tasks can run on different compute, so resolution must refuse -# rather than guess serverless. -[[Server]] -Pattern = "GET /api/2.2/jobs/get" -Response.Body = ''' -{ - "job_id": 12345, - "settings": { - "name": "mixed-compute-job", - "environments": [ - {"environment_key": "default", "spec": {"client": "3"}} - ], - "job_clusters": [ - {"job_cluster_key": "main", "new_cluster": {"spark_version": "15.4.x-scala2.12", "num_workers": 1}} - ] - } -} -''' diff --git a/acceptance/localenv/job-classic-check/output.txt b/acceptance/localenv/job-classic-check/output.txt index 9253f7a267e..b3fa6654679 100644 --- a/acceptance/localenv/job-classic-check/output.txt +++ b/acceptance/localenv/job-classic-check/output.txt @@ -1,5 +1,5 @@ ->>> [CLI] environments setup-local --job-id 12345 --dry-run +>>> [CLI] environments setup-local --job-task 12345.ingest --dry-run preflight ok check resolve ok source=job envKey=dbr/15.4.x-scala2.12 fetch ok source=[DATABRICKS_URL]/dbr/15.4.x-scala2.12/pyproject.toml fromCache=false diff --git a/acceptance/localenv/job-classic-check/script b/acceptance/localenv/job-classic-check/script index 07e14a5386f..c7c56c97f7b 100644 --- a/acceptance/localenv/job-classic-check/script +++ b/acceptance/localenv/job-classic-check/script @@ -1 +1 @@ -trace $CLI environments setup-local --job-id 12345 --dry-run +trace $CLI environments setup-local --job-task 12345.ingest --dry-run diff --git a/acceptance/localenv/job-classic-check/test.toml b/acceptance/localenv/job-classic-check/test.toml index b61e90a2ba4..33fc90a5e37 100644 --- a/acceptance/localenv/job-classic-check/test.toml +++ b/acceptance/localenv/job-classic-check/test.toml @@ -3,7 +3,7 @@ EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] [Env] DATABRICKS_LOCALENV_CONSTRAINT_SOURCE = "$DATABRICKS_HOST" -# A classic-compute job (a single job cluster, no serverless environments) +# A classic task carries its own new_cluster, so --job-task . # resolves to that cluster's DBR-derived environment key. [[Server]] Pattern = "GET /api/2.2/jobs/get" @@ -12,8 +12,8 @@ Response.Body = ''' "job_id": 12345, "settings": { "name": "classic-job", - "job_clusters": [ - {"job_cluster_key": "main", "new_cluster": {"spark_version": "15.4.x-scala2.12", "num_workers": 1}} + "tasks": [ + {"task_key": "ingest", "new_cluster": {"spark_version": "15.4.x-scala2.12", "num_workers": 1}} ] } } diff --git a/acceptance/localenv/job-multicluster-mismatch/script b/acceptance/localenv/job-multicluster-mismatch/script deleted file mode 100644 index bd37e81d924..00000000000 --- a/acceptance/localenv/job-multicluster-mismatch/script +++ /dev/null @@ -1 +0,0 @@ -musterr $CLI environments setup-local --job-id 12345 --dry-run diff --git a/acceptance/localenv/job-multicluster-mismatch/test.toml b/acceptance/localenv/job-multicluster-mismatch/test.toml deleted file mode 100644 index 3f41f4648b0..00000000000 --- a/acceptance/localenv/job-multicluster-mismatch/test.toml +++ /dev/null @@ -1,22 +0,0 @@ -EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] - -[Env] -DATABRICKS_LOCALENV_CONSTRAINT_SOURCE = "$DATABRICKS_HOST" - -# A job whose job clusters declare differing spark_version is ambiguous: tasks -# can reference any job_cluster_key, so resolution must refuse rather than -# silently provision for the first cluster's runtime. -[[Server]] -Pattern = "GET /api/2.2/jobs/get" -Response.Body = ''' -{ - "job_id": 12345, - "settings": { - "name": "multi-cluster-job", - "job_clusters": [ - {"job_cluster_key": "a", "new_cluster": {"spark_version": "15.4.x-scala2.12", "num_workers": 1}}, - {"job_cluster_key": "b", "new_cluster": {"spark_version": "14.3.x-scala2.12", "num_workers": 1}} - ] - } -} -''' diff --git a/acceptance/localenv/job-serverless-check/output.txt b/acceptance/localenv/job-serverless-check/output.txt index b386001e3f7..a86c32d2cb3 100644 --- a/acceptance/localenv/job-serverless-check/output.txt +++ b/acceptance/localenv/job-serverless-check/output.txt @@ -1,5 +1,5 @@ ->>> [CLI] environments setup-local --job-id 12345 --dry-run +>>> [CLI] environments setup-local --job-task 12345.transform --dry-run preflight ok check resolve ok source=job envKey=serverless/serverless-v3 fetch ok source=[DATABRICKS_URL]/serverless/serverless-v3/pyproject.toml fromCache=false diff --git a/acceptance/localenv/job-serverless-check/script b/acceptance/localenv/job-serverless-check/script index 07e14a5386f..572709bfadf 100644 --- a/acceptance/localenv/job-serverless-check/script +++ b/acceptance/localenv/job-serverless-check/script @@ -1 +1 @@ -trace $CLI environments setup-local --job-id 12345 --dry-run +trace $CLI environments setup-local --job-task 12345.transform --dry-run diff --git a/acceptance/localenv/job-serverless-check/test.toml b/acceptance/localenv/job-serverless-check/test.toml index 63b18a4bb57..89ef2ce5794 100644 --- a/acceptance/localenv/job-serverless-check/test.toml +++ b/acceptance/localenv/job-serverless-check/test.toml @@ -3,8 +3,9 @@ EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] [Env] DATABRICKS_LOCALENV_CONSTRAINT_SOURCE = "$DATABRICKS_HOST" -# A serverless job pins its environment_version on the environment spec, so the -# target resolves to that serverless-vN (here v3) rather than defaulting to v5. +# A serverless task binds an environment_key to one of the job's environments; +# that environment pins the version, so --job-task . resolves to +# the matching serverless-vN (here v3) directly — no default fallback. [[Server]] Pattern = "GET /api/2.2/jobs/get" Response.Body = ''' @@ -12,6 +13,9 @@ Response.Body = ''' "job_id": 12345, "settings": { "name": "serverless-job", + "tasks": [ + {"task_key": "transform", "environment_key": "default"} + ], "environments": [ {"environment_key": "default", "spec": {"environment_version": "3"}} ] diff --git a/acceptance/localenv/job-serverless-version-mismatch/script b/acceptance/localenv/job-serverless-version-mismatch/script deleted file mode 100644 index bd37e81d924..00000000000 --- a/acceptance/localenv/job-serverless-version-mismatch/script +++ /dev/null @@ -1 +0,0 @@ -musterr $CLI environments setup-local --job-id 12345 --dry-run diff --git a/acceptance/localenv/job-serverless-version-mismatch/test.toml b/acceptance/localenv/job-serverless-version-mismatch/test.toml deleted file mode 100644 index 029bab1fb99..00000000000 --- a/acceptance/localenv/job-serverless-version-mismatch/test.toml +++ /dev/null @@ -1,22 +0,0 @@ -EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] - -[Env] -DATABRICKS_LOCALENV_CONSTRAINT_SOURCE = "$DATABRICKS_HOST" - -# Tasks can reference any environment_key, so a job whose serverless environments -# declare differing versions is ambiguous: resolution must refuse rather than -# guess from the first environment. -[[Server]] -Pattern = "GET /api/2.2/jobs/get" -Response.Body = ''' -{ - "job_id": 12345, - "settings": { - "name": "serverless-mismatch-job", - "environments": [ - {"environment_key": "a", "spec": {"environment_version": "3"}}, - {"environment_key": "b", "spec": {"environment_version": "4"}} - ] - } -} -''' diff --git a/acceptance/localenv/job-ambiguous-compute/out.test.toml b/acceptance/localenv/job-task-missing-key/out.test.toml similarity index 100% rename from acceptance/localenv/job-ambiguous-compute/out.test.toml rename to acceptance/localenv/job-task-missing-key/out.test.toml diff --git a/acceptance/localenv/job-serverless-version-mismatch/output.txt b/acceptance/localenv/job-task-missing-key/output.txt similarity index 54% rename from acceptance/localenv/job-serverless-version-mismatch/output.txt rename to acceptance/localenv/job-task-missing-key/output.txt index 5abae68d78c..75f44248d43 100644 --- a/acceptance/localenv/job-serverless-version-mismatch/output.txt +++ b/acceptance/localenv/job-task-missing-key/output.txt @@ -1,5 +1,5 @@ preflight ok check -resolve error resolving job 12345: job 12345 has serverless environments with differing versions; pass --serverless-version explicitly to disambiguate +resolve error specify a job task: job 12345 has multiple tasks; specify one: --job-task 12345. (available: ingest, transform) fetch pending merge pending provision pending diff --git a/acceptance/localenv/job-task-missing-key/script b/acceptance/localenv/job-task-missing-key/script new file mode 100644 index 00000000000..266ee6163aa --- /dev/null +++ b/acceptance/localenv/job-task-missing-key/script @@ -0,0 +1 @@ +musterr $CLI environments setup-local --job-task 12345 --dry-run diff --git a/acceptance/localenv/job-task-missing-key/test.toml b/acceptance/localenv/job-task-missing-key/test.toml new file mode 100644 index 00000000000..048d3a6f478 --- /dev/null +++ b/acceptance/localenv/job-task-missing-key/test.toml @@ -0,0 +1,26 @@ +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] + +[Env] +DATABRICKS_LOCALENV_CONSTRAINT_SOURCE = "$DATABRICKS_HOST" + +# A bare --job-task (no .) is a usage error: a job can bind +# tasks to different environment versions, so the CLI will not guess which one. +# The error lists the job's task keys. Reported as E_USAGE at the resolve phase. +[[Server]] +Pattern = "GET /api/2.2/jobs/get" +Response.Body = ''' +{ + "job_id": 12345, + "settings": { + "name": "multi-task-job", + "tasks": [ + {"task_key": "ingest", "environment_key": "e1"}, + {"task_key": "transform", "environment_key": "e2"} + ], + "environments": [ + {"environment_key": "e1", "spec": {"environment_version": "3"}}, + {"environment_key": "e2", "spec": {"environment_version": "4"}} + ] + } +} +''' diff --git a/acceptance/localenv/job-multicluster-mismatch/out.test.toml b/acceptance/localenv/job-task-unknown/out.test.toml similarity index 100% rename from acceptance/localenv/job-multicluster-mismatch/out.test.toml rename to acceptance/localenv/job-task-unknown/out.test.toml diff --git a/acceptance/localenv/job-multicluster-mismatch/output.txt b/acceptance/localenv/job-task-unknown/output.txt similarity index 52% rename from acceptance/localenv/job-multicluster-mismatch/output.txt rename to acceptance/localenv/job-task-unknown/output.txt index a62881c188c..7bd03a24d26 100644 --- a/acceptance/localenv/job-multicluster-mismatch/output.txt +++ b/acceptance/localenv/job-task-unknown/output.txt @@ -1,5 +1,5 @@ preflight ok check -resolve error resolving job 12345: job 12345 has job clusters with differing spark_version; pass --cluster-id or --serverless-version explicitly to disambiguate +resolve error resolving job task 12345.nope: job 12345 has no task "nope" (available: ingest) fetch pending merge pending provision pending diff --git a/acceptance/localenv/job-task-unknown/script b/acceptance/localenv/job-task-unknown/script new file mode 100644 index 00000000000..6d43be42211 --- /dev/null +++ b/acceptance/localenv/job-task-unknown/script @@ -0,0 +1 @@ +musterr $CLI environments setup-local --job-task 12345.nope --dry-run diff --git a/acceptance/localenv/job-task-unknown/test.toml b/acceptance/localenv/job-task-unknown/test.toml new file mode 100644 index 00000000000..2a5cd68d54c --- /dev/null +++ b/acceptance/localenv/job-task-unknown/test.toml @@ -0,0 +1,23 @@ +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] + +[Env] +DATABRICKS_LOCALENV_CONSTRAINT_SOURCE = "$DATABRICKS_HOST" + +# A --job-task naming a task key that the job does not define fails at resolve +# (E_RESOLVE); the message lists the job's actual task keys. +[[Server]] +Pattern = "GET /api/2.2/jobs/get" +Response.Body = ''' +{ + "job_id": 12345, + "settings": { + "name": "job-with-tasks", + "tasks": [ + {"task_key": "ingest", "environment_key": "default"} + ], + "environments": [ + {"environment_key": "default", "spec": {"environment_version": "3"}} + ] + } +} +''' diff --git a/acceptance/localenv/job-serverless-version-mismatch/out.test.toml b/acceptance/localenv/job-task-unpinned/out.test.toml similarity index 100% rename from acceptance/localenv/job-serverless-version-mismatch/out.test.toml rename to acceptance/localenv/job-task-unpinned/out.test.toml diff --git a/acceptance/localenv/job-ambiguous-compute/output.txt b/acceptance/localenv/job-task-unpinned/output.txt similarity index 51% rename from acceptance/localenv/job-ambiguous-compute/output.txt rename to acceptance/localenv/job-task-unpinned/output.txt index 9f846c1c409..b9a0f8ebc4b 100644 --- a/acceptance/localenv/job-ambiguous-compute/output.txt +++ b/acceptance/localenv/job-task-unpinned/output.txt @@ -1,5 +1,5 @@ preflight ok check -resolve error resolving job 12345: job 12345 has both serverless environments and job clusters; pass --cluster-id or --serverless-version explicitly to disambiguate +resolve error resolving job task 12345.transform: task "transform" of job 12345 binds environment "default", which records no environment version fetch pending merge pending provision pending diff --git a/acceptance/localenv/job-task-unpinned/script b/acceptance/localenv/job-task-unpinned/script new file mode 100644 index 00000000000..6bec20c0289 --- /dev/null +++ b/acceptance/localenv/job-task-unpinned/script @@ -0,0 +1 @@ +musterr $CLI environments setup-local --job-task 12345.transform --dry-run diff --git a/acceptance/localenv/job-task-unpinned/test.toml b/acceptance/localenv/job-task-unpinned/test.toml new file mode 100644 index 00000000000..292dae64b5a --- /dev/null +++ b/acceptance/localenv/job-task-unpinned/test.toml @@ -0,0 +1,25 @@ +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] + +[Env] +DATABRICKS_LOCALENV_CONSTRAINT_SOURCE = "$DATABRICKS_HOST" + +# A task whose bound serverless environment records no version cannot be resolved: +# the job-task path reads the version directly and never falls back to a default +# (the serverless-vN fallback applies only to the bundle-target path). Fails at +# resolve (E_RESOLVE); nothing is fetched. +[[Server]] +Pattern = "GET /api/2.2/jobs/get" +Response.Body = ''' +{ + "job_id": 12345, + "settings": { + "name": "serverless-job-unpinned", + "tasks": [ + {"task_key": "transform", "environment_key": "default"} + ], + "environments": [ + {"environment_key": "default", "spec": {}} + ] + } +} +''' diff --git a/acceptance/localenv/json-error/output.txt b/acceptance/localenv/json-error/output.txt index e99122aa349..b2e5a0ba43e 100644 --- a/acceptance/localenv/json-error/output.txt +++ b/acceptance/localenv/json-error/output.txt @@ -35,7 +35,7 @@ "error": { "code": "E_NO_TARGET", "failurePhase": "resolve", - "message": "No compute target is selected. Select a cluster or serverless target, or pass --cluster-id / --cluster-name / --serverless-version / --job-id", + "message": "No compute target is selected. Select a cluster or serverless target, or pass --cluster-id / --cluster-name / --serverless-version / --job-task", "diskMutated": false }, "durationMs": 0 diff --git a/acceptance/localenv/no-target/output.txt b/acceptance/localenv/no-target/output.txt index 46c5fa2199e..eba9aad3918 100644 --- a/acceptance/localenv/no-target/output.txt +++ b/acceptance/localenv/no-target/output.txt @@ -1,5 +1,5 @@ preflight ok uv [UV_VERSION] -resolve error No compute target is selected. Select a cluster or serverless target, or pass --cluster-id / --cluster-name / --serverless-version / --job-id +resolve error No compute target is selected. Select a cluster or serverless target, or pass --cluster-id / --cluster-name / --serverless-version / --job-task fetch pending merge pending provision pending diff --git a/acceptance/localenv/serverless-default-check/out.test.toml b/acceptance/localenv/serverless-default-check/out.test.toml deleted file mode 100644 index e90b6d5d1ba..00000000000 --- a/acceptance/localenv/serverless-default-check/out.test.toml +++ /dev/null @@ -1,3 +0,0 @@ -Local = true -Cloud = false -EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] diff --git a/acceptance/localenv/serverless-default-check/output.txt b/acceptance/localenv/serverless-default-check/output.txt deleted file mode 100644 index 96512162951..00000000000 --- a/acceptance/localenv/serverless-default-check/output.txt +++ /dev/null @@ -1,13 +0,0 @@ - ->>> [CLI] environments setup-local --job-id 12345 --dry-run -preflight ok check -resolve ok source=job envKey=serverless/serverless-v5 -fetch ok source=[DATABRICKS_URL]/serverless/serverless-v5/pyproject.toml fromCache=false -merge ok -provision ok -validate ok -Plan: [TEST_TMP_DIR]/pyproject.toml - changed region: requires-python - changed region: tool.uv.constraint-dependencies - changed region: databricks-connect -Check complete. No files were modified. diff --git a/acceptance/localenv/serverless-default-check/script b/acceptance/localenv/serverless-default-check/script deleted file mode 100644 index 07e14a5386f..00000000000 --- a/acceptance/localenv/serverless-default-check/script +++ /dev/null @@ -1 +0,0 @@ -trace $CLI environments setup-local --job-id 12345 --dry-run diff --git a/acceptance/localenv/serverless-default-check/test.toml b/acceptance/localenv/serverless-default-check/test.toml deleted file mode 100644 index efb58c3fdb0..00000000000 --- a/acceptance/localenv/serverless-default-check/test.toml +++ /dev/null @@ -1,39 +0,0 @@ -EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] - -[Env] -DATABRICKS_LOCALENV_CONSTRAINT_SOURCE = "$DATABRICKS_HOST" - -# A serverless job that pins no environment_version exercises the default-version -# path: ResolveTarget falls back to defaultServerlessVersion (v5), so the target -# resolves to serverless-v5. This is the end-to-end coverage that the v5 default -# actually resolves and fetches, distinct from job-serverless-check which pins v3. -[[Server]] -Pattern = "GET /api/2.2/jobs/get" -Response.Body = ''' -{ - "job_id": 12345, - "settings": { - "name": "serverless-job-unpinned", - "environments": [ - {"environment_key": "default", "spec": {}} - ] - } -} -''' - -[[Server]] -Pattern = "GET /serverless/serverless-v5/pyproject.toml" -Response.Body = ''' -[project] -requires-python = ">=3.12" - -[dependency-groups] -dev = ["databricks-connect~=17.2.0"] - -[tool.uv] -constraint-dependencies = ["pyarrow<19", "pandas<3"] -''' - -[[Repls]] -Old = 'uv uv \S+(?: \([^)]+\))?' -New = 'uv [UV_VERSION]' diff --git a/cmd/environments/compute.go b/cmd/environments/compute.go index 97a29ef56f1..95266e6da69 100644 --- a/cmd/environments/compute.go +++ b/cmd/environments/compute.go @@ -4,7 +4,9 @@ import ( "context" "fmt" "strconv" + "strings" + localenv "github.com/databricks/cli/libs/localenv" databricks "github.com/databricks/databricks-sdk-go" "github.com/databricks/databricks-sdk-go/service/compute" "github.com/databricks/databricks-sdk-go/service/jobs" @@ -70,18 +72,19 @@ func (c sdkCompute) GetClusterByName(ctx context.Context, name string) (string, } } -// GetJobSparkVersion inspects the job's configuration to determine compute type. +// JobTaskEnvironment resolves a single job task's compute to an environment. // -// A job is considered serverless when it has non-empty Environments (JobEnvironment -// entries), which signals the Databricks serverless runtime. A job with classic compute -// uses JobClusters; we read SparkVersion from the first job cluster's NewCluster spec. +// A job can bind multiple tasks to different serverless environment versions, so +// the target is a specific task, not the whole job. taskKey selects it: +// - an empty taskKey is an enumerate request — it returns *localenv.ErrTaskKeyRequired +// carrying the job's task keys, so the caller emits an actionable E_USAGE; +// - an unknown taskKey returns an error listing the available keys. // -// Task-level compute (tasks[].new_cluster / tasks[].existing_cluster_id with no -// job-level job_clusters) is not resolved here: it may vary per task and an -// existing_cluster_id would need a second lookup, which is out of scope for the -// initial job support. Such a job returns an actionable error rather than a wrong -// guess; use --cluster-id or --serverless-version explicitly instead. -func (c sdkCompute) GetJobSparkVersion(ctx context.Context, jobID string) (sparkVersion string, isServerless bool, version string, err error) { +// A serverless task binds an environment_key to one of the job's environments; +// its version is read directly from that environment's spec (no fallback). A +// classic task carries its own new_cluster (or an existing_cluster_id, resolved +// via the Clusters API). +func (c sdkCompute) JobTaskEnvironment(ctx context.Context, jobID, taskKey string) (sparkVersion string, isServerless bool, version string, err error) { id, err := strconv.ParseInt(jobID, 10, 64) if err != nil { return "", false, "", fmt.Errorf("invalid job ID %q: must be an integer: %w", jobID, err) @@ -91,54 +94,62 @@ func (c sdkCompute) GetJobSparkVersion(ctx context.Context, jobID string) (spark if err != nil { return "", false, "", fmt.Errorf("get job %d: %w", id, err) } - if job.Settings == nil { return "", false, "", fmt.Errorf("job %d has no settings", id) } - // A job that declares both serverless environments and classic job clusters is - // ambiguous: its tasks can run on different compute, so there is no single - // correct local environment to provision. Refuse rather than guess serverless. - if len(job.Settings.Environments) > 0 && len(job.Settings.JobClusters) > 0 { - return "", false, "", fmt.Errorf("job %d has both serverless environments and job clusters; pass --cluster-id or --serverless-version explicitly to disambiguate", id) + taskKeys := make([]string, 0, len(job.Settings.Tasks)) + for _, t := range job.Settings.Tasks { + taskKeys = append(taskKeys, t.TaskKey) } - // Serverless jobs have Environments populated; classic compute uses JobClusters. - if len(job.Settings.Environments) > 0 { - // The serverless environment version (e.g. "4") is recorded on the job's - // environment spec, unlike the bundle path where it is unavailable. Return - // it so ResolveTarget pins the matching serverless-vN instead of defaulting - // to v5. An empty version (older jobs) falls back to v5 in ResolveTarget. - version := environmentVersion(job.Settings.Environments[0]) - // Tasks can reference any environment_key, so if the job's environments do - // not all share one version there is no single correct local environment - // (mirrors the job-cluster check below). Refuse rather than guess from the - // first. A pinned-vs-unpinned mix is also ambiguous, so compare raw values. - for _, e := range job.Settings.Environments[1:] { - if environmentVersion(e) != version { - return "", false, "", fmt.Errorf("job %d has serverless environments with differing versions; pass --serverless-version explicitly to disambiguate", id) - } - } - return "", true, version, nil + // No task key given: ask the caller to pick one, listing what is available. + // This is a usage error, not a resolve failure — see ResolveTarget. + if taskKey == "" { + return "", false, "", &localenv.ErrTaskKeyRequired{JobID: jobID, TaskKeys: taskKeys} } - if len(job.Settings.JobClusters) > 0 { - sv := job.Settings.JobClusters[0].NewCluster.SparkVersion - if sv == "" { - return "", false, "", fmt.Errorf("could not determine compute for job %d: first job cluster has no spark_version", id) + var task *jobs.Task + for i := range job.Settings.Tasks { + if job.Settings.Tasks[i].TaskKey == taskKey { + task = &job.Settings.Tasks[i] + break } - // Tasks can reference any job_cluster_key, so if the job's clusters do not - // all share one Spark version there is no single correct local environment. - // Refuse rather than silently provisioning for the first cluster. - for _, jc := range job.Settings.JobClusters[1:] { - if jc.NewCluster.SparkVersion != sv { - return "", false, "", fmt.Errorf("job %d has job clusters with differing spark_version; pass --cluster-id or --serverless-version explicitly to disambiguate", id) + } + if task == nil { + return "", false, "", fmt.Errorf("job %s has no task %q (available: %s)", jobID, taskKey, strings.Join(taskKeys, ", ")) + } + + // Serverless task: it references one of the job's environments by key; the + // version lives on that environment's spec and is used directly. + if task.EnvironmentKey != "" { + for _, e := range job.Settings.Environments { + if e.EnvironmentKey == task.EnvironmentKey { + v := environmentVersion(e) + if v == "" { + return "", false, "", fmt.Errorf("task %q of job %s binds environment %q, which records no environment version", taskKey, jobID, task.EnvironmentKey) + } + return "", true, v, nil } } + return "", false, "", fmt.Errorf("task %q of job %s references environment %q, which the job does not define", taskKey, jobID, task.EnvironmentKey) + } + + // Classic task with an inline cluster spec. + if task.NewCluster != nil && task.NewCluster.SparkVersion != "" { + return task.NewCluster.SparkVersion, false, task.NewCluster.SparkVersion, nil + } + + // Classic task pinned to an existing cluster: resolve its Spark version. + if task.ExistingClusterId != "" { + sv, cerr := c.GetClusterSparkVersion(ctx, task.ExistingClusterId) + if cerr != nil { + return "", false, "", fmt.Errorf("resolving existing cluster %s for task %q of job %s: %w", task.ExistingClusterId, taskKey, jobID, cerr) + } return sv, false, sv, nil } - return "", false, "", fmt.Errorf("could not determine compute for job %d from its environments or job clusters (task-level compute is not supported); pass --cluster-id or --serverless-version explicitly", id) + return "", false, "", fmt.Errorf("task %q of job %s has no serverless environment or resolvable cluster; pass --cluster-id or --serverless-version explicitly", taskKey, jobID) } // environmentVersion returns the serverless environment version recorded on a @@ -146,10 +157,9 @@ func (c sdkCompute) GetJobSparkVersion(ctx context.Context, jobID string) (spark // // The version can arrive in either of two fields. environment_version is the // current one; client is its deprecated predecessor ("Use environment_version -// instead") and is still what some jobs pin. Reading both means the v5 fallback -// and the divergence guard observe whichever field actually carries the pin, -// rather than treating a client-pinned job as unversioned. base_environment is -// deliberately ignored: it is a path/ID, not a version. +// instead") and is still what some jobs pin. Reading both means a client-pinned +// task is not treated as unversioned. base_environment is deliberately ignored: +// it is a path/ID, not a version. func environmentVersion(e jobs.JobEnvironment) string { if e.Spec == nil { return "" diff --git a/cmd/environments/sync.go b/cmd/environments/sync.go index 441e1be64b0..3c0bf2880fc 100644 --- a/cmd/environments/sync.go +++ b/cmd/environments/sync.go @@ -50,7 +50,7 @@ func addTargetFlags(cmd *cobra.Command) { cmd.Flags().String("cluster-id", "", "cluster ID to use as the compute target") cmd.Flags().String("cluster-name", "", "cluster name to use as the compute target (resolved to an ID via the Clusters API)") cmd.Flags().String("serverless-version", "", "serverless version to use as the compute target (e.g. 5)") - cmd.Flags().String("job-id", "", "job ID to use as the compute target") + cmd.Flags().String("job-task", "", "job task to use as the compute target, as . (the task key is required)") cmd.Flags().Bool("constraints-only", false, "apply the Python version and constraints without adding the databricks-connect dependency") cmd.Flags().Bool("dry-run", false, "compute the plan without writing files or provisioning") cmd.Flags().String("constraint-source-url", "", "URL for the constraint source (overrides "+envConstraintSource+")") @@ -69,7 +69,7 @@ func runPipeline(cmd *cobra.Command) error { cluster, _ := cmd.Flags().GetString("cluster-id") clusterName, _ := cmd.Flags().GetString("cluster-name") serverless, _ := cmd.Flags().GetString("serverless-version") - job, _ := cmd.Flags().GetString("job-id") + jobTask, _ := cmd.Flags().GetString("job-task") constraintsOnly, _ := cmd.Flags().GetBool("constraints-only") check, _ := cmd.Flags().GetBool("dry-run") constraintSource, _ := cmd.Flags().GetString("constraint-source-url") @@ -78,7 +78,7 @@ func runPipeline(cmd *cobra.Command) error { Cluster: cluster, ClusterName: clusterName, Serverless: serverless, - Job: job, + JobTask: jobTask, } // Flag validation (including mutual exclusivity) happens in the pipeline's // preflight, so a conflict is reported as E_USAGE through the phase/JSON @@ -103,11 +103,11 @@ func runPipeline(cmd *cobra.Command) error { cacheDir = filepath.Join(cacheDir, "databricks", "localenv") // The bundle is only a fallback: ResolveTarget consults it solely when no - // explicit --cluster-id/--cluster-name/--serverless-version/--job-id flag is set. Skip the bundle load + // explicit --cluster-id/--cluster-name/--serverless-version/--job-task flag is set. Skip the bundle load // entirely when a flag is present — it would otherwise re-run TryConfigureBundle // (a second full load) and re-print any bundle load-time diagnostics for nothing. var bt libslocalenv.BundleTarget - if cluster == "" && clusterName == "" && serverless == "" && job == "" { + if cluster == "" && clusterName == "" && serverless == "" && jobTask == "" { bt = bundleTarget(cmd) } diff --git a/libs/localenv/result.go b/libs/localenv/result.go index 834801a47b4..53e50a6b919 100644 --- a/libs/localenv/result.go +++ b/libs/localenv/result.go @@ -74,7 +74,7 @@ const ( type ErrorCode string const ( - ErrUsage ErrorCode = "E_USAGE" // preflight: incompatible flags + ErrUsage ErrorCode = "E_USAGE" // preflight: incompatible flags; resolve: --job-task names a job but no task ErrManagerUnsupported ErrorCode = "E_MANAGER_UNSUPPORTED" // preflight: manager is not uv ErrNotWritable ErrorCode = "E_NOT_WRITABLE" // preflight: project dir not writable ErrUvMissing ErrorCode = "E_UV_MISSING" // preflight: uv not found / install failed diff --git a/libs/localenv/target.go b/libs/localenv/target.go index 119bc7a0538..b73575cc1fa 100644 --- a/libs/localenv/target.go +++ b/libs/localenv/target.go @@ -2,6 +2,7 @@ package localenv import ( "context" + "errors" "fmt" "strings" ) @@ -14,9 +15,33 @@ type ComputeClient interface { // errors when the name is unknown or ambiguous (more than one cluster shares // the name), so the caller can surface an actionable E_RESOLVE. GetClusterByName(ctx context.Context, name string) (clusterID, sparkVersion string, err error) - // GetJobSparkVersion returns either a Spark version (isServerless=false) or a - // serverless marker (isServerless=true) for a job, plus a recorded version string. - GetJobSparkVersion(ctx context.Context, jobID string) (sparkVersion string, isServerless bool, version string, err error) + // JobTaskEnvironment resolves a single job task's compute to an environment. + // + // A job can bind multiple tasks to different environment versions, so the + // target is a specific task, not the whole job. taskKey selects it; an empty + // taskKey is a request to enumerate — the method returns ErrTaskKeyRequired + // wrapping the job's task keys so the caller can prompt for one. An unknown + // taskKey returns an error naming the available keys. + // + // For a serverless task, isServerless is true and version is the task's + // recorded serverless environment version (read directly — no fallback). For + // a classic task, isServerless is false and sparkVersion is the task cluster's + // runtime. + JobTaskEnvironment(ctx context.Context, jobID, taskKey string) (sparkVersion string, isServerless bool, version string, err error) +} + +// ErrTaskKeyRequired is returned by JobTaskEnvironment when --job-task names a +// job but no task, so ResolveTarget can classify it as E_USAGE (the user must +// pick a task) rather than E_RESOLVE. It carries the job's task keys so the +// error message can list them. +type ErrTaskKeyRequired struct { + JobID string + TaskKeys []string +} + +func (e *ErrTaskKeyRequired) Error() string { + return fmt.Sprintf("job %s has multiple tasks; specify one: --job-task %s. (available: %s)", + e.JobID, e.JobID, strings.Join(e.TaskKeys, ", ")) } // TargetFlags holds the mutually-exclusive compute target flags from the CLI. @@ -24,7 +49,10 @@ type TargetFlags struct { Cluster string ClusterName string Serverless string - Job string + // JobTask is "." (or a bare "", which resolves to + // an E_USAGE listing the job's task keys). A job can bind tasks to different + // environment versions, so the target is a specific task, not the whole job. + JobTask string } // BundleTarget is the three-state result of reading the bundle's configured @@ -37,7 +65,7 @@ type BundleTarget struct { // noTargetMessage is the actionable message shown when no target is selected, // matching spec §2.3. -const noTargetMessage = "No compute target is selected. Select a cluster or serverless target, or pass --cluster-id / --cluster-name / --serverless-version / --job-id" +const noTargetMessage = "No compute target is selected. Select a cluster or serverless target, or pass --cluster-id / --cluster-name / --serverless-version / --job-task" // ValidateTargetFlags returns an error if more than one of the target flags is set. // Cobra marks them mutually exclusive too; this guards the library path. @@ -52,8 +80,8 @@ func ValidateTargetFlags(f TargetFlags) error { if f.Serverless != "" { set = append(set, "--serverless-version") } - if f.Job != "" { - set = append(set, "--job-id") + if f.JobTask != "" { + set = append(set, "--job-task") } if len(set) > 1 { return fmt.Errorf("flags %s are mutually exclusive; specify at most one", strings.Join(set, " and ")) @@ -62,7 +90,7 @@ func ValidateTargetFlags(f TargetFlags) error { } // ResolveTarget resolves the compute target using ordered precedence: -// --cluster-id → --cluster-name → --serverless-version → --job-id → bundle target. +// --cluster-id → --cluster-name → --serverless-version → --job-task → bundle target. // PythonVersion is left empty; it is filled later from constraint data. // // Incompatible flags are rejected up front: without this a library caller that @@ -114,27 +142,32 @@ func ResolveTarget(ctx context.Context, f TargetFlags, c ComputeClient, bt Bundl }, nil } - if f.Job != "" { - sparkVersion, isServerless, version, err := c.GetJobSparkVersion(ctx, f.Job) + if f.JobTask != "" { + // Split "." on the FIRST dot: task keys may themselves + // contain dots, so everything after the first separator is the task key. + // A bare "" (no dot) leaves taskKey empty, which JobTaskEnvironment + // treats as an enumerate request and reports back via ErrTaskKeyRequired. + jobID, taskKey, _ := strings.Cut(f.JobTask, ".") + sparkVersion, isServerless, version, err := c.JobTaskEnvironment(ctx, jobID, taskKey) if err != nil { - return nil, NewError(ErrResolve, err, "resolving job %s", f.Job) + // A missing task key is a usage error (the user must pick a task), + // distinct from a genuine resolve failure (unknown key, API error). + if _, ok := errors.AsType[*ErrTaskKeyRequired](err); ok { + return nil, NewError(ErrUsage, err, "specify a job task") + } + return nil, NewError(ErrResolve, err, "resolving job task %s", f.JobTask) } if isServerless { - // Use the job's recorded serverless environment version when present; - // fall back to the default when the job did not pin one (documented - // stand-in, spec §4.3). - v := version - if v == "" { - v = defaultServerlessVersion - } + // The task's serverless environment version is read directly from its + // bound environment, so there is no version to guess: unlike the bundle + // path, the job-task path never applies the serverless-vN fallback. return &TargetInfo{ Source: "job", - ServerlessVersion: NormalizeServerless(v), - EnvKey: EnvKeyForServerless(v), + ServerlessVersion: NormalizeServerless(version), + EnvKey: EnvKeyForServerless(version), }, nil } - // Classic compute: the Spark version is the first return per the - // GetJobSparkVersion contract, not the recorded-version third return. + // Classic compute: sparkVersion is the task cluster's runtime. return &TargetInfo{ Source: "job", SparkVersion: sparkVersion, diff --git a/libs/localenv/target_test.go b/libs/localenv/target_test.go index a202cac683c..6e0d64e9abd 100644 --- a/libs/localenv/target_test.go +++ b/libs/localenv/target_test.go @@ -25,7 +25,7 @@ func (s stubCompute) GetClusterByName(_ context.Context, _ string) (string, stri return s.byNameID, s.byNameVersion, s.byNameErr } -func (s stubCompute) GetJobSparkVersion(_ context.Context, _ string) (string, bool, string, error) { +func (s stubCompute) JobTaskEnvironment(_ context.Context, _, _ string) (string, bool, string, error) { return "", false, "", nil } @@ -120,59 +120,91 @@ func TestResolveBundleServerless(t *testing.T) { assert.Equal(t, "serverless/serverless-v5", ti.EnvKey) } -// jobStubCompute returns distinct values for the first (sparkVersion) and third -// (recorded version) results of GetJobSparkVersion so the classic-compute branch -// can be checked against the documented contract (it must use the first). +// jobStubCompute stubs JobTaskEnvironment. It records the jobID and taskKey it +// was called with (so tests can assert the "." split) and +// returns configurable results, including an error for the bare-job / unknown-key +// paths. The sparkVersion/version fields are distinct so the classic branch can +// be checked against the contract (it must use the Spark-version return). type jobStubCompute struct { sparkVersion string isServerless bool version string + err error + + gotJobID string + gotTaskKey string } -func (jobStubCompute) GetClusterSparkVersion(_ context.Context, _ string) (string, error) { +func (*jobStubCompute) GetClusterSparkVersion(_ context.Context, _ string) (string, error) { return "", nil } -func (jobStubCompute) GetClusterByName(_ context.Context, _ string) (string, string, error) { +func (*jobStubCompute) GetClusterByName(_ context.Context, _ string) (string, string, error) { return "", "", nil } -func (s jobStubCompute) GetJobSparkVersion(_ context.Context, _ string) (string, bool, string, error) { +func (s *jobStubCompute) JobTaskEnvironment(_ context.Context, jobID, taskKey string) (string, bool, string, error) { + s.gotJobID = jobID + s.gotTaskKey = taskKey + if s.err != nil { + return "", false, "", s.err + } return s.sparkVersion, s.isServerless, s.version, nil } -func TestResolveJobClassicUsesSparkVersionReturn(t *testing.T) { - // Contract: for a classic-compute job the Spark version is the FIRST return. - // The third "recorded version" return differs here to catch use of the wrong one. - c := jobStubCompute{sparkVersion: "15.4.x-scala2.12", isServerless: false, version: "wrong-recorded"} - ti, err := ResolveTarget(t.Context(), TargetFlags{Job: "42"}, c, BundleTarget{}) +func TestResolveJobTaskClassicUsesSparkVersion(t *testing.T) { + // A classic task resolves to its cluster's Spark version → dbr/ env key. + c := &jobStubCompute{sparkVersion: "15.4.x-scala2.12"} + ti, err := ResolveTarget(t.Context(), TargetFlags{JobTask: "42.ingest"}, c, BundleTarget{}) require.NoError(t, err) assert.Equal(t, "job", ti.Source) assert.Equal(t, "15.4.x-scala2.12", ti.SparkVersion) assert.Equal(t, "dbr/15.4.x-scala2.12", ti.EnvKey) + // The value is split on the first dot into job ID and task key. + assert.Equal(t, "42", c.gotJobID) + assert.Equal(t, "ingest", c.gotTaskKey) +} + +func TestResolveJobTaskSplitsOnFirstDot(t *testing.T) { + // Task keys may themselves contain dots, so only the first dot separates the + // job ID from the task key. + c := &jobStubCompute{sparkVersion: "15.4.x-scala2.12"} + _, err := ResolveTarget(t.Context(), TargetFlags{JobTask: "42.a.b.c"}, c, BundleTarget{}) + require.NoError(t, err) + assert.Equal(t, "42", c.gotJobID) + assert.Equal(t, "a.b.c", c.gotTaskKey) } -func TestResolveJobServerlessUsesRecordedVersion(t *testing.T) { - // A serverless job (isServerless=true) pins its serverless version via the - // third "recorded version" return; ResolveTarget must map it to the matching - // serverless-vN rather than the classic dbr path. - c := jobStubCompute{isServerless: true, version: "3"} - ti, err := ResolveTarget(t.Context(), TargetFlags{Job: "42"}, c, BundleTarget{}) +func TestResolveJobTaskServerlessUsesTaskVersion(t *testing.T) { + // A serverless task's environment version is read directly (no fallback): + // it maps to the matching serverless-vN, not the classic dbr path. + c := &jobStubCompute{isServerless: true, version: "3"} + ti, err := ResolveTarget(t.Context(), TargetFlags{JobTask: "42.transform"}, c, BundleTarget{}) require.NoError(t, err) assert.Equal(t, "job", ti.Source) assert.Empty(t, ti.SparkVersion) assert.Equal(t, "serverless/serverless-v3", ti.EnvKey) } -func TestResolveJobServerlessEmptyVersionFallsBackToDefault(t *testing.T) { - // When the job records no serverless version, ResolveTarget uses the default - // stand-in, matching the bundle serverless path. - c := jobStubCompute{isServerless: true, version: ""} - ti, err := ResolveTarget(t.Context(), TargetFlags{Job: "42"}, c, BundleTarget{}) - require.NoError(t, err) - // Concrete literal, not "serverless-"+defaultServerlessVersion: the default - // is v5, and asserting the constant against itself would pass for any value. - assert.Equal(t, "serverless/serverless-v5", ti.EnvKey) +func TestResolveJobTaskMissingKeyIsUsageError(t *testing.T) { + // A bare "" (no task key) is E_USAGE, not E_RESOLVE: the user must + // pick a task. The stub reports the enumerate request via ErrTaskKeyRequired. + c := &jobStubCompute{err: &ErrTaskKeyRequired{JobID: "42", TaskKeys: []string{"ingest", "transform"}}} + _, err := ResolveTarget(t.Context(), TargetFlags{JobTask: "42"}, c, BundleTarget{}) + var pe *PipelineError + require.ErrorAs(t, err, &pe) + assert.Equal(t, ErrUsage, pe.Code) + assert.Empty(t, c.gotTaskKey, "a bare job ID yields an empty task key") +} + +func TestResolveJobTaskUnknownKeyIsResolveError(t *testing.T) { + // An unknown task key is a genuine resolve failure (E_RESOLVE), distinct from + // the missing-key usage error above. + c := &jobStubCompute{err: errors.New(`job 42 has no task "nope" (available: ingest)`)} + _, err := ResolveTarget(t.Context(), TargetFlags{JobTask: "42.nope"}, c, BundleTarget{}) + var pe *PipelineError + require.ErrorAs(t, err, &pe) + assert.Equal(t, ErrResolve, pe.Code) } func TestValidateTargetFlagsMutuallyExclusive(t *testing.T) { From 087602d9cc0ae856c4eb951590e97cf426133eea Mon Sep 17 00:00:00 2001 From: Grigory Panov Date: Fri, 24 Jul 2026 10:20:21 +0200 Subject: [PATCH 2/2] Address review: resolve job_cluster_key and for_each tasks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex review of the --job-task change caught two task shapes JobTaskEnvironment did not handle, both of which resolve fine now that the target is a specific task: - job_cluster_key: a classic task commonly references a shared job_clusters entry by key instead of carrying an inline new_cluster. That fell through to "no resolvable cluster" — a regression from the old whole-job path, which read job_clusters[0]. Resolve the referenced job cluster's Spark version. - for_each_task: the outer task wraps the real per-iteration task, whose compute (environment_key / new_cluster / existing_cluster_id / job_cluster_key) lives on the nested task. Unwrap to it before resolving. Extract the per-task resolution into resolveTaskCompute so the for-each nested task reuses the same logic. Add acceptance coverage: job-task-jobcluster (shared job cluster -> dbr key) and job-task-foreach (nested serverless task -> serverless-vN). Co-authored-by: Isaac --- .../localenv/job-task-foreach/out.test.toml | 3 ++ .../localenv/job-task-foreach/output.txt | 13 +++++ acceptance/localenv/job-task-foreach/script | 1 + .../localenv/job-task-foreach/test.toml | 47 +++++++++++++++++++ .../job-task-jobcluster/out.test.toml | 3 ++ .../localenv/job-task-jobcluster/output.txt | 13 +++++ .../localenv/job-task-jobcluster/script | 1 + .../localenv/job-task-jobcluster/test.toml | 41 ++++++++++++++++ cmd/environments/compute.go | 37 +++++++++++++-- 9 files changed, 156 insertions(+), 3 deletions(-) create mode 100644 acceptance/localenv/job-task-foreach/out.test.toml create mode 100644 acceptance/localenv/job-task-foreach/output.txt create mode 100644 acceptance/localenv/job-task-foreach/script create mode 100644 acceptance/localenv/job-task-foreach/test.toml create mode 100644 acceptance/localenv/job-task-jobcluster/out.test.toml create mode 100644 acceptance/localenv/job-task-jobcluster/output.txt create mode 100644 acceptance/localenv/job-task-jobcluster/script create mode 100644 acceptance/localenv/job-task-jobcluster/test.toml diff --git a/acceptance/localenv/job-task-foreach/out.test.toml b/acceptance/localenv/job-task-foreach/out.test.toml new file mode 100644 index 00000000000..e90b6d5d1ba --- /dev/null +++ b/acceptance/localenv/job-task-foreach/out.test.toml @@ -0,0 +1,3 @@ +Local = true +Cloud = false +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] diff --git a/acceptance/localenv/job-task-foreach/output.txt b/acceptance/localenv/job-task-foreach/output.txt new file mode 100644 index 00000000000..6587b3e2135 --- /dev/null +++ b/acceptance/localenv/job-task-foreach/output.txt @@ -0,0 +1,13 @@ + +>>> [CLI] environments setup-local --job-task 12345.fanout --dry-run +preflight ok check +resolve ok source=job envKey=serverless/serverless-v3 +fetch ok source=[DATABRICKS_URL]/serverless/serverless-v3/pyproject.toml fromCache=false +merge ok +provision ok +validate ok +Plan: [TEST_TMP_DIR]/pyproject.toml + changed region: requires-python + changed region: tool.uv.constraint-dependencies + changed region: databricks-connect +Check complete. No files were modified. diff --git a/acceptance/localenv/job-task-foreach/script b/acceptance/localenv/job-task-foreach/script new file mode 100644 index 00000000000..46ed21b157e --- /dev/null +++ b/acceptance/localenv/job-task-foreach/script @@ -0,0 +1 @@ +trace $CLI environments setup-local --job-task 12345.fanout --dry-run diff --git a/acceptance/localenv/job-task-foreach/test.toml b/acceptance/localenv/job-task-foreach/test.toml new file mode 100644 index 00000000000..8c7e836acca --- /dev/null +++ b/acceptance/localenv/job-task-foreach/test.toml @@ -0,0 +1,47 @@ +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] + +[Env] +DATABRICKS_LOCALENV_CONSTRAINT_SOURCE = "$DATABRICKS_HOST" + +# A for_each_task wraps the real per-iteration task; its compute lives on the +# nested task. --job-task resolves against that nested task (here a serverless +# environment binding), not the empty outer wrapper. +[[Server]] +Pattern = "GET /api/2.2/jobs/get" +Response.Body = ''' +{ + "job_id": 12345, + "settings": { + "name": "foreach-job", + "tasks": [ + { + "task_key": "fanout", + "for_each_task": { + "inputs": "[1,2,3]", + "task": {"task_key": "inner", "environment_key": "default"} + } + } + ], + "environments": [ + {"environment_key": "default", "spec": {"environment_version": "3"}} + ] + } +} +''' + +[[Server]] +Pattern = "GET /serverless/serverless-v3/pyproject.toml" +Response.Body = ''' +[project] +requires-python = ">=3.11" + +[dependency-groups] +dev = ["databricks-connect~=15.4.0"] + +[tool.uv] +constraint-dependencies = ["pyarrow<19"] +''' + +[[Repls]] +Old = 'uv uv \S+(?: \([^)]+\))?' +New = 'uv [UV_VERSION]' diff --git a/acceptance/localenv/job-task-jobcluster/out.test.toml b/acceptance/localenv/job-task-jobcluster/out.test.toml new file mode 100644 index 00000000000..e90b6d5d1ba --- /dev/null +++ b/acceptance/localenv/job-task-jobcluster/out.test.toml @@ -0,0 +1,3 @@ +Local = true +Cloud = false +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] diff --git a/acceptance/localenv/job-task-jobcluster/output.txt b/acceptance/localenv/job-task-jobcluster/output.txt new file mode 100644 index 00000000000..b3fa6654679 --- /dev/null +++ b/acceptance/localenv/job-task-jobcluster/output.txt @@ -0,0 +1,13 @@ + +>>> [CLI] environments setup-local --job-task 12345.ingest --dry-run +preflight ok check +resolve ok source=job envKey=dbr/15.4.x-scala2.12 +fetch ok source=[DATABRICKS_URL]/dbr/15.4.x-scala2.12/pyproject.toml fromCache=false +merge ok +provision ok +validate ok +Plan: [TEST_TMP_DIR]/pyproject.toml + changed region: requires-python + changed region: tool.uv.constraint-dependencies + changed region: databricks-connect +Check complete. No files were modified. diff --git a/acceptance/localenv/job-task-jobcluster/script b/acceptance/localenv/job-task-jobcluster/script new file mode 100644 index 00000000000..c7c56c97f7b --- /dev/null +++ b/acceptance/localenv/job-task-jobcluster/script @@ -0,0 +1 @@ +trace $CLI environments setup-local --job-task 12345.ingest --dry-run diff --git a/acceptance/localenv/job-task-jobcluster/test.toml b/acceptance/localenv/job-task-jobcluster/test.toml new file mode 100644 index 00000000000..7eb76463420 --- /dev/null +++ b/acceptance/localenv/job-task-jobcluster/test.toml @@ -0,0 +1,41 @@ +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] + +[Env] +DATABRICKS_LOCALENV_CONSTRAINT_SOURCE = "$DATABRICKS_HOST" + +# A classic task references a shared job cluster by job_cluster_key (rather than +# an inline new_cluster), so --job-task resolves the version from the job's +# job_clusters entry. +[[Server]] +Pattern = "GET /api/2.2/jobs/get" +Response.Body = ''' +{ + "job_id": 12345, + "settings": { + "name": "jobcluster-job", + "tasks": [ + {"task_key": "ingest", "job_cluster_key": "shared"} + ], + "job_clusters": [ + {"job_cluster_key": "shared", "new_cluster": {"spark_version": "15.4.x-scala2.12", "num_workers": 1}} + ] + } +} +''' + +[[Server]] +Pattern = "GET /dbr/15.4.x-scala2.12/pyproject.toml" +Response.Body = ''' +[project] +requires-python = ">=3.11" + +[dependency-groups] +dev = ["databricks-connect~=15.4.0"] + +[tool.uv] +constraint-dependencies = ["pyarrow<19"] +''' + +[[Repls]] +Old = 'uv uv \S+(?: \([^)]+\))?' +New = 'uv [UV_VERSION]' diff --git a/cmd/environments/compute.go b/cmd/environments/compute.go index 95266e6da69..65c69825072 100644 --- a/cmd/environments/compute.go +++ b/cmd/environments/compute.go @@ -82,8 +82,10 @@ func (c sdkCompute) GetClusterByName(ctx context.Context, name string) (string, // // A serverless task binds an environment_key to one of the job's environments; // its version is read directly from that environment's spec (no fallback). A -// classic task carries its own new_cluster (or an existing_cluster_id, resolved -// via the Clusters API). +// classic task resolves from its new_cluster, its job_cluster_key (a shared +// job_clusters entry), or its existing_cluster_id (via the Clusters API). A +// for_each_task is unwrapped to its nested task, whose compute is resolved the +// same way. func (c sdkCompute) JobTaskEnvironment(ctx context.Context, jobID, taskKey string) (sparkVersion string, isServerless bool, version string, err error) { id, err := strconv.ParseInt(jobID, 10, 64) if err != nil { @@ -120,10 +122,25 @@ func (c sdkCompute) JobTaskEnvironment(ctx context.Context, jobID, taskKey strin return "", false, "", fmt.Errorf("job %s has no task %q (available: %s)", jobID, taskKey, strings.Join(taskKeys, ", ")) } + // A for_each_task wraps the real per-iteration task: its compute + // (environment_key / new_cluster / existing_cluster_id / job_cluster_key) + // lives on the nested task, not the outer one. Resolve against that. + if task.ForEachTask != nil { + task = &task.ForEachTask.Task + } + + return c.resolveTaskCompute(ctx, job.Settings, task, jobID, taskKey) +} + +// resolveTaskCompute resolves one task's compute to an environment. It reads the +// serverless environment version directly (no fallback) or the classic cluster's +// Spark version, consulting the job-level environments and job_clusters the task +// references by key. +func (c sdkCompute) resolveTaskCompute(ctx context.Context, settings *jobs.JobSettings, task *jobs.Task, jobID, taskKey string) (sparkVersion string, isServerless bool, version string, err error) { // Serverless task: it references one of the job's environments by key; the // version lives on that environment's spec and is used directly. if task.EnvironmentKey != "" { - for _, e := range job.Settings.Environments { + for _, e := range settings.Environments { if e.EnvironmentKey == task.EnvironmentKey { v := environmentVersion(e) if v == "" { @@ -140,6 +157,20 @@ func (c sdkCompute) JobTaskEnvironment(ctx context.Context, jobID, taskKey strin return task.NewCluster.SparkVersion, false, task.NewCluster.SparkVersion, nil } + // Classic task referencing a shared job cluster by key: read that cluster's + // Spark version from the job's job_clusters (the common classic-task shape). + if task.JobClusterKey != "" { + for _, jc := range settings.JobClusters { + if jc.JobClusterKey == task.JobClusterKey { + if jc.NewCluster.SparkVersion == "" { + return "", false, "", fmt.Errorf("task %q of job %s uses job cluster %q, which has no spark_version", taskKey, jobID, task.JobClusterKey) + } + return jc.NewCluster.SparkVersion, false, jc.NewCluster.SparkVersion, nil + } + } + return "", false, "", fmt.Errorf("task %q of job %s references job cluster %q, which the job does not define", taskKey, jobID, task.JobClusterKey) + } + // Classic task pinned to an existing cluster: resolve its Spark version. if task.ExistingClusterId != "" { sv, cerr := c.GetClusterSparkVersion(ctx, task.ExistingClusterId)