From b1d5f16360ce3595325559e67167da53ec04cfef Mon Sep 17 00:00:00 2001 From: Rada Kamysheva Date: Fri, 21 Aug 2026 15:53:19 +0000 Subject: [PATCH 1/2] Add job_runs lifecycle.triggers.on_value_change Re-fire a run when a resolved expression changes, including resource refs that become deploy-graph edges. Removing the trigger is a state-only update. --- .../bundles/job-runs-on-value-change.md | 1 + .../job_runs/on_value_change/databricks.yml | 25 ++++ .../job_runs/on_value_change/out.test.toml | 2 + .../job_runs/on_value_change/output.txt | 140 ++++++++++++++++++ .../resources/job_runs/on_value_change/script | 40 +++++ .../on_value_change_ref/databricks.yml | 32 ++++ .../on_value_change_ref/out.test.toml | 3 + .../job_runs/on_value_change_ref/output.txt | 76 ++++++++++ .../job_runs/on_value_change_ref/script | 29 ++++ .../job_runs/on_value_change_ref/test.toml | 5 + .../mutator/validate_job_run_triggers.go | 23 ++- .../mutator/validate_job_run_triggers_test.go | 48 +++++- bundle/config/resources/job_run.go | 12 ++ bundle/config/resources/lifecycle.go | 17 ++- bundle/direct/bundle_plan.go | 1 + bundle/direct/dresources/job_run.go | 67 ++++++++- bundle/direct/dresources/job_run_test.go | 87 +++++++++++ bundle/internal/schema/annotations.yml | 3 + bundle/schema/jsonschema.json | 4 + 19 files changed, 601 insertions(+), 14 deletions(-) create mode 100644 .nextchanges/bundles/job-runs-on-value-change.md create mode 100644 acceptance/bundle/resources/job_runs/on_value_change/databricks.yml create mode 100644 acceptance/bundle/resources/job_runs/on_value_change/out.test.toml create mode 100644 acceptance/bundle/resources/job_runs/on_value_change/output.txt create mode 100644 acceptance/bundle/resources/job_runs/on_value_change/script create mode 100644 acceptance/bundle/resources/job_runs/on_value_change_ref/databricks.yml create mode 100644 acceptance/bundle/resources/job_runs/on_value_change_ref/out.test.toml create mode 100644 acceptance/bundle/resources/job_runs/on_value_change_ref/output.txt create mode 100644 acceptance/bundle/resources/job_runs/on_value_change_ref/script create mode 100644 acceptance/bundle/resources/job_runs/on_value_change_ref/test.toml diff --git a/.nextchanges/bundles/job-runs-on-value-change.md b/.nextchanges/bundles/job-runs-on-value-change.md new file mode 100644 index 00000000000..94d0b06290f --- /dev/null +++ b/.nextchanges/bundles/job-runs-on-value-change.md @@ -0,0 +1 @@ +direct: `resources.job_runs` can set `lifecycle.triggers.on_value_change` to an interpolated expression (for example `${resources.jobs.foo.id}`) to re-fire the run when that value changes. Removing the trigger does not recreate the existing run. diff --git a/acceptance/bundle/resources/job_runs/on_value_change/databricks.yml b/acceptance/bundle/resources/job_runs/on_value_change/databricks.yml new file mode 100644 index 00000000000..b0ed0322b9c --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change/databricks.yml @@ -0,0 +1,25 @@ +bundle: + name: job-runs-on-value-change + +variables: + stamp: + default: v1 + other: + default: a1 + +resources: + jobs: + my_job: + name: my-job + tasks: + - task_key: main + notebook_task: + notebook_path: /Workspace/test + + job_runs: + my_run: + job_id: ${resources.jobs.my_job.id} + lifecycle: + triggers: + - on_value_change: ${var.stamp} + - on_value_change: ${var.other} diff --git a/acceptance/bundle/resources/job_runs/on_value_change/out.test.toml b/acceptance/bundle/resources/job_runs/on_value_change/out.test.toml new file mode 100644 index 00000000000..0938e678987 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change/out.test.toml @@ -0,0 +1,2 @@ +Cloud = false +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] diff --git a/acceptance/bundle/resources/job_runs/on_value_change/output.txt b/acceptance/bundle/resources/job_runs/on_value_change/output.txt new file mode 100644 index 00000000000..c6591532e73 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change/output.txt @@ -0,0 +1,140 @@ + +=== first deploy triggers a run +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-on-value-change/default/files... +Output from job_runs.my_run: id=[MY_RUN_ID]: Run URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID]?o=[NUMID] +Output from job_runs.my_run: id=[MY_RUN_ID]: SUCCESS +Created job_runs.my_run +Created jobs.my_job +Files: 4 uploaded, 0 deleted +Resources: 2 created, 0 changed, 0 deleted, 0 unchanged + +>>> read_id.py my_job +[MY_JOB_ID] + +>>> print_requests.py //jobs/run-now +{ + "method": "POST", + "path": "/api/2.2/jobs/run-now", + "body": { + "idempotency_token": "[UUID]", + "job_id": [MY_JOB_ID] + } +} + +=== redeploy with unchanged values plans nothing +>>> [CLI] bundle plan +Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged + +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-on-value-change/default/files... +Files: 2 uploaded, 0 deleted +Resources: 0 created, 0 changed, 0 deleted, 2 unchanged + +>>> print_requests.py //jobs/run-now + +=== changing one value re-fires +>>> [CLI] bundle plan +recreate job_runs.my_run + +Plan: 1 to add, 0 to change, 1 to delete, 1 unchanged + +>>> changes +{ + "lifecycle.triggers.on_value_change.values": { + "action": "recreate", + "reason": "immutable", + "old": { + "a1": "a1", + "v1": "v1" + }, + "new": { + "a1": "a1", + "v2": "v2" + } + }, + "lifecycle.triggers.on_value_change.values['v1']": { + "action": "recreate", + "reason": "immutable", + "old": "v1" + }, + "lifecycle.triggers.on_value_change.values['v2']": { + "action": "recreate", + "reason": "immutable", + "new": "v2" + }, + "result_state": { + "action": "skip", + "reason": "remote_already_set", + "new": "SUCCESS", + "remote": "SUCCESS" + } +} + +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-on-value-change/default/files... +Output from job_runs.my_run: id=[MY_RUN_ID_2]: Run URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID_2]?o=[NUMID] +Output from job_runs.my_run: id=[MY_RUN_ID_2]: SUCCESS +Recreated job_runs.my_run +Files: 3 uploaded, 0 deleted +Resources: 1 created, 0 changed, 1 deleted, 1 unchanged + +>>> print_requests.py --keep //jobs/runs/delete +{ + "method": "POST", + "path": "/api/2.2/jobs/runs/delete", + "body": { + "run_id": [MY_RUN_ID] + } +} + +>>> print_requests.py //jobs/run-now +{ + "method": "POST", + "path": "/api/2.2/jobs/run-now", + "body": { + "idempotency_token": "[UUID]", + "job_id": [MY_JOB_ID] + } +} + +=== removing on_value_change rewrites state without a run +>>> [CLI] bundle plan +update job_runs.my_run + +Plan: 0 to add, 1 to change, 0 to delete, 1 unchanged + +>>> changes +{ + "lifecycle.triggers.on_value_change.values": { + "action": "update", + "reason": "trigger removed", + "old": { + "a1": "a1", + "v2": "v2" + } + }, + "result_state": { + "action": "skip", + "reason": "remote_already_set", + "new": "SUCCESS", + "remote": "SUCCESS" + } +} + +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-on-value-change/default/files... +Updated job_runs.my_run +Files: 3 uploaded, 0 deleted +Resources: 0 created, 1 changed, 0 deleted, 1 unchanged + +>>> print_requests.py //jobs/run-now + +>>> [CLI] bundle destroy --auto-approve +The following resources will be deleted: + delete resources.job_runs.my_run + delete resources.jobs.my_job + +All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/job-runs-on-value-change/default + +Destroy: 2 deleted diff --git a/acceptance/bundle/resources/job_runs/on_value_change/script b/acceptance/bundle/resources/job_runs/on_value_change/script new file mode 100644 index 00000000000..a8421e880dd --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change/script @@ -0,0 +1,40 @@ +cleanup() { + trace $CLI bundle destroy --auto-approve + rm -f out.requests.txt +} +trap cleanup EXIT + +changes() { + $CLI bundle plan -o json | jq '.plan["resources.job_runs.my_run"].changes' +} + +title "first deploy triggers a run" +trace $CLI bundle deploy +trace read_id.py my_job +read_id.py my_run > /dev/null +trace print_requests.py //jobs/run-now + +title "redeploy with unchanged values plans nothing" +trace $CLI bundle plan +trace $CLI bundle deploy +trace print_requests.py //jobs/run-now + +title "changing one value re-fires" +update_file.py databricks.yml "default: v1" "default: v2" +trace $CLI bundle plan +trace changes +trace $CLI bundle deploy +read_id.py my_run > /dev/null +trace print_requests.py --keep //jobs/runs/delete +trace print_requests.py //jobs/run-now + +title "removing on_value_change rewrites state without a run" +update_file.py databricks.yml ' lifecycle: + triggers: + - on_value_change: ${var.stamp} + - on_value_change: ${var.other} +' '' +trace $CLI bundle plan +trace changes +trace $CLI bundle deploy +trace print_requests.py //jobs/run-now diff --git a/acceptance/bundle/resources/job_runs/on_value_change_ref/databricks.yml b/acceptance/bundle/resources/job_runs/on_value_change_ref/databricks.yml new file mode 100644 index 00000000000..84de0fdaf82 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change_ref/databricks.yml @@ -0,0 +1,32 @@ +bundle: + name: job-runs-on-value-change-ref + +resources: + jobs: + my_job: + name: my-job + tasks: + - task_key: main + notebook_task: + notebook_path: /Workspace/test + + other: + name: other-job + tasks: + - task_key: main + notebook_task: + notebook_path: /Workspace/test + + extra: + name: extra-job + tasks: + - task_key: main + notebook_task: + notebook_path: /Workspace/test + + job_runs: + my_run: + job_id: ${resources.jobs.my_job.id} + lifecycle: + triggers: + - on_value_change: ${resources.jobs.other.id}-${resources.jobs.extra.id} diff --git a/acceptance/bundle/resources/job_runs/on_value_change_ref/out.test.toml b/acceptance/bundle/resources/job_runs/on_value_change_ref/out.test.toml new file mode 100644 index 00000000000..57b0f616850 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change_ref/out.test.toml @@ -0,0 +1,3 @@ +Cloud = false +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] +EnvMatrix.READPLAN = ["", "1"] diff --git a/acceptance/bundle/resources/job_runs/on_value_change_ref/output.txt b/acceptance/bundle/resources/job_runs/on_value_change_ref/output.txt new file mode 100644 index 00000000000..f1521167db3 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change_ref/output.txt @@ -0,0 +1,76 @@ + +=== plan depends on both jobs in the expression +>>> jq .plan["resources.job_runs.my_run"].depends_on tmp.plan.json +[ + { + "node": "resources.jobs.extra", + "label": "${resources.jobs.extra.id}" + }, + { + "node": "resources.jobs.my_job", + "label": "${resources.jobs.my_job.id}" + }, + { + "node": "resources.jobs.other", + "label": "${resources.jobs.other.id}" + } +] + +=== first deploy triggers a run +=== bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-on-value-change-ref/default/files... +Output from job_runs.my_run: id=[MY_RUN_ID]: Run URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID]?o=[NUMID] +Output from job_runs.my_run: id=[MY_RUN_ID]: SUCCESS +Created job_runs.my_run +Created jobs.extra +Created jobs.my_job +Created jobs.other +Files: 6 uploaded, 0 deleted +Resources: 4 created, 0 changed, 0 deleted, 0 unchanged + +>>> read_id.py my_job +[MY_JOB_ID] + +>>> read_id.py other +[OTHER_ID] + +>>> read_id.py extra +[EXTRA_ID] + +>>> print_requests.py //jobs/run-now +{ + "method": "POST", + "path": "/api/2.2/jobs/run-now", + "body": { + "idempotency_token": "[UUID]", + "job_id": [MY_JOB_ID] + } +} + +=== state records the concatenated resolved ids +>>> print_state.py +{ + "${resources.jobs.other.id}-${resources.jobs.extra.id}": "[OTHER_ID]-[EXTRA_ID]" +} + +=== redeploy with unchanged ids plans nothing +>>> [CLI] bundle plan +Plan: 0 to add, 0 to change, 0 to delete, 4 unchanged + +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-on-value-change-ref/default/files... +Files: 2 uploaded, 0 deleted +Resources: 0 created, 0 changed, 0 deleted, 4 unchanged + +>>> print_requests.py //jobs/run-now + +>>> [CLI] bundle destroy --auto-approve +The following resources will be deleted: + delete resources.job_runs.my_run + delete resources.jobs.extra + delete resources.jobs.my_job + delete resources.jobs.other + +All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/job-runs-on-value-change-ref/default + +Destroy: 4 deleted diff --git a/acceptance/bundle/resources/job_runs/on_value_change_ref/script b/acceptance/bundle/resources/job_runs/on_value_change_ref/script new file mode 100644 index 00000000000..e4377eb5bfb --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change_ref/script @@ -0,0 +1,29 @@ +cleanup() { + trace $CLI bundle destroy --auto-approve + rm -f out.requests.txt +} +trap cleanup EXIT + +title "plan depends on both jobs in the expression" +$CLI bundle plan -o json > tmp.plan.json +trace jq '.plan["resources.job_runs.my_run"].depends_on' tmp.plan.json | contains.py \ + 'resources.jobs.other' '${resources.jobs.other.id}' \ + 'resources.jobs.extra' '${resources.jobs.extra.id}' + +title "first deploy triggers a run" +# Not traced: readplanarg makes the command line differ per READPLAN variant. +title "bundle deploy\n" +$CLI bundle deploy $(readplanarg tmp.plan.json) +trace read_id.py my_job +trace read_id.py other +trace read_id.py extra +read_id.py my_run > /dev/null +trace print_requests.py //jobs/run-now + +title "state records the concatenated resolved ids" +trace print_state.py | jq '.state["resources.job_runs.my_run"].state.lifecycle.triggers.on_value_change.values' + +title "redeploy with unchanged ids plans nothing" +trace $CLI bundle plan +trace $CLI bundle deploy +trace print_requests.py //jobs/run-now diff --git a/acceptance/bundle/resources/job_runs/on_value_change_ref/test.toml b/acceptance/bundle/resources/job_runs/on_value_change_ref/test.toml new file mode 100644 index 00000000000..a698f7a65d1 --- /dev/null +++ b/acceptance/bundle/resources/job_runs/on_value_change_ref/test.toml @@ -0,0 +1,5 @@ +# Deploy both by re-planning and from a plan saved on disk, so resource refs +# in on_value_change survive plan serialization. +EnvMatrix.READPLAN = ["", "1"] + +Ignore = ["tmp.plan.json"] diff --git a/bundle/config/mutator/validate_job_run_triggers.go b/bundle/config/mutator/validate_job_run_triggers.go index c185d3b5781..fa62b3110f6 100644 --- a/bundle/config/mutator/validate_job_run_triggers.go +++ b/bundle/config/mutator/validate_job_run_triggers.go @@ -26,28 +26,28 @@ func (*validateJobRunTriggers) Apply(_ context.Context, b *bundle.Bundle) diag.D if jr == nil || jr.Lifecycle == nil { continue } - // Recreate-every-deploy cannot coexist with prevent_destroy. - if (jr.HasOnBundleDeploy() || jr.HasOnFileChange()) && jr.Lifecycle.PreventDestroy { + if (jr.HasOnBundleDeploy() || jr.HasOnFileChange() || jr.HasOnValueChange()) && jr.Lifecycle.PreventDestroy { diags = diags.Append(diag.Diagnostic{ Severity: diag.Error, - Summary: "lifecycle.triggers.on_bundle_deploy or on_file_change is incompatible with lifecycle.prevent_destroy", + Summary: "lifecycle.triggers.on_bundle_deploy, on_file_change, or on_value_change is incompatible with lifecycle.prevent_destroy", Locations: b.Config.GetLocations(fmt.Sprintf("resources.job_runs.%s.lifecycle", name)), }) } for i, t := range jr.Lifecycle.Triggers { path := fmt.Sprintf("resources.job_runs.%s.lifecycle.triggers[%d]", name, i) - if t.OnBundleDeploy == nil && t.OnFileChange == nil { + switch t.ArmedCount() { + case 0: diags = diags.Append(diag.Diagnostic{ Severity: diag.Error, - Summary: "lifecycle.triggers entry must set on_bundle_deploy or on_file_change", + Summary: "lifecycle.triggers entry must set on_bundle_deploy, on_file_change, or on_value_change", Locations: b.Config.GetLocations(path), }) continue - } - if t.OnBundleDeploy != nil && t.OnFileChange != nil { + case 1: + default: diags = diags.Append(diag.Diagnostic{ Severity: diag.Error, - Summary: "lifecycle.triggers entry must set only one of on_bundle_deploy or on_file_change", + Summary: "lifecycle.triggers entry must set only one of on_bundle_deploy, on_file_change, or on_value_change", Locations: b.Config.GetLocations(path), }) continue @@ -66,6 +66,13 @@ func (*validateJobRunTriggers) Apply(_ context.Context, b *bundle.Bundle) diag.D Locations: b.Config.GetLocations(path + ".on_file_change"), }) } + if t.OnValueChange != nil && strings.TrimSpace(*t.OnValueChange) == "" { + diags = diags.Append(diag.Diagnostic{ + Severity: diag.Error, + Summary: "lifecycle.triggers.on_value_change must be non-empty when set", + Locations: b.Config.GetLocations(path + ".on_value_change"), + }) + } } } return diags diff --git a/bundle/config/mutator/validate_job_run_triggers_test.go b/bundle/config/mutator/validate_job_run_triggers_test.go index 7e8772d8d8b..e6b429f0f5a 100644 --- a/bundle/config/mutator/validate_job_run_triggers_test.go +++ b/bundle/config/mutator/validate_job_run_triggers_test.go @@ -18,6 +18,10 @@ func TestValidateJobRunTriggers(t *testing.T) { emptyFile := "" whitespaceFile := " \t" + valueChange := "${resources.jobs.foo.id}" + emptyValue := "" + whitespaceValue := " \t" + tests := []struct { name string triggers []resources.JobRunTrigger @@ -41,6 +45,7 @@ func TestValidateJobRunTriggers(t *testing.T) { triggers: []resources.JobRunTrigger{ {OnFileChange: &fileChange}, {OnBundleDeploy: &trueVal}, + {OnValueChange: &valueChange}, }, }, { @@ -48,14 +53,14 @@ func TestValidateJobRunTriggers(t *testing.T) { triggers: []resources.JobRunTrigger{ {}, }, - summary: "lifecycle.triggers entry must set on_bundle_deploy or on_file_change", + summary: "lifecycle.triggers entry must set on_bundle_deploy, on_file_change, or on_value_change", }, { name: "both keys on one entry", triggers: []resources.JobRunTrigger{ {OnBundleDeploy: &trueVal, OnFileChange: &fileChange}, }, - summary: "lifecycle.triggers entry must set only one of on_bundle_deploy or on_file_change", + summary: "lifecycle.triggers entry must set only one of on_bundle_deploy, on_file_change, or on_value_change", }, { name: "on_bundle_deploy false", @@ -84,7 +89,7 @@ func TestValidateJobRunTriggers(t *testing.T) { {OnBundleDeploy: &trueVal}, }, preventDestroy: true, - summary: "lifecycle.triggers.on_bundle_deploy or on_file_change is incompatible with lifecycle.prevent_destroy", + summary: "lifecycle.triggers.on_bundle_deploy, on_file_change, or on_value_change is incompatible with lifecycle.prevent_destroy", }, { name: "on_file_change with prevent_destroy", @@ -92,12 +97,47 @@ func TestValidateJobRunTriggers(t *testing.T) { {OnFileChange: &fileChange}, }, preventDestroy: true, - summary: "lifecycle.triggers.on_bundle_deploy or on_file_change is incompatible with lifecycle.prevent_destroy", + summary: "lifecycle.triggers.on_bundle_deploy, on_file_change, or on_value_change is incompatible with lifecycle.prevent_destroy", }, { name: "prevent_destroy alone", preventDestroy: true, }, + { + name: "on_value_change set", + triggers: []resources.JobRunTrigger{ + {OnValueChange: &valueChange}, + }, + }, + { + name: "on_value_change empty", + triggers: []resources.JobRunTrigger{ + {OnValueChange: &emptyValue}, + }, + summary: "lifecycle.triggers.on_value_change must be non-empty when set", + }, + { + name: "on_value_change whitespace", + triggers: []resources.JobRunTrigger{ + {OnValueChange: &whitespaceValue}, + }, + summary: "lifecycle.triggers.on_value_change must be non-empty when set", + }, + { + name: "on_value_change with prevent_destroy", + triggers: []resources.JobRunTrigger{ + {OnValueChange: &valueChange}, + }, + preventDestroy: true, + summary: "lifecycle.triggers.on_bundle_deploy, on_file_change, or on_value_change is incompatible with lifecycle.prevent_destroy", + }, + { + name: "on_value_change and on_file_change on one entry", + triggers: []resources.JobRunTrigger{ + {OnFileChange: &fileChange, OnValueChange: &valueChange}, + }, + summary: "lifecycle.triggers entry must set only one of on_bundle_deploy, on_file_change, or on_value_change", + }, } for _, tt := range tests { diff --git a/bundle/config/resources/job_run.go b/bundle/config/resources/job_run.go index f4b16963453..4200a65a2ef 100644 --- a/bundle/config/resources/job_run.go +++ b/bundle/config/resources/job_run.go @@ -58,6 +58,18 @@ func (r *JobRun) HasOnFileChange() bool { return false } +func (r *JobRun) HasOnValueChange() bool { + if r.Lifecycle == nil { + return false + } + for _, t := range r.Lifecycle.Triggers { + if t.OnValueChange != nil { + return true + } + } + return false +} + func (r *JobRun) UnmarshalJSON(b []byte) error { return marshal.Unmarshal(b, r) } diff --git a/bundle/config/resources/lifecycle.go b/bundle/config/resources/lifecycle.go index a4e881d298c..9bed0aaa641 100644 --- a/bundle/config/resources/lifecycle.go +++ b/bundle/config/resources/lifecycle.go @@ -37,5 +37,20 @@ type JobRunLifecycle struct { // JobRunTrigger is one lifecycle.triggers entry. type JobRunTrigger struct { OnBundleDeploy *bool `json:"on_bundle_deploy,omitempty"` - OnFileChange *string `json:"on_file_change,omitempty"` // path or glob under sync root + OnFileChange *string `json:"on_file_change,omitempty"` // path or glob under sync root + OnValueChange *string `json:"on_value_change,omitempty"` // interpolated expr; re-fire when the resolved value changes +} + +func (t JobRunTrigger) ArmedCount() int { + n := 0 + if t.OnBundleDeploy != nil { + n++ + } + if t.OnFileChange != nil { + n++ + } + if t.OnValueChange != nil { + n++ + } + return n } diff --git a/bundle/direct/bundle_plan.go b/bundle/direct/bundle_plan.go index 5b8829e3f59..11c34d5cdee 100644 --- a/bundle/direct/bundle_plan.go +++ b/bundle/direct/bundle_plan.go @@ -1007,6 +1007,7 @@ func (b *DeploymentBundle) makePlan(ctx context.Context, configRoot *config.Root } maps.Copy(refs, inputStructVar.Refs) + dresources.DropJobRunValueChangeConfigRefs(refs) var dependsOn []deployplan.DependsOnEntry for _, reference := range refs { diff --git a/bundle/direct/dresources/job_run.go b/bundle/direct/dresources/job_run.go index a300318dabe..5487bffb598 100644 --- a/bundle/direct/dresources/job_run.go +++ b/bundle/direct/dresources/job_run.go @@ -14,8 +14,11 @@ import ( "github.com/databricks/cli/bundle/config/resources" "github.com/databricks/cli/bundle/deployplan" "github.com/databricks/cli/libs/cmdio" + "github.com/databricks/cli/libs/dyn" + "github.com/databricks/cli/libs/dyn/dynvar" "github.com/databricks/cli/libs/log" "github.com/databricks/cli/libs/structs/structpath" + "github.com/databricks/cli/libs/structs/structvar" "github.com/databricks/cli/libs/workspaceurls" "github.com/databricks/databricks-sdk-go" "github.com/databricks/databricks-sdk-go/marshal" @@ -32,6 +35,7 @@ const jobRunTimeout = 24 * time.Hour var jobRunTriggerLocalPaths = []string{ "lifecycle.triggers.on_bundle_deploy", "lifecycle.triggers.on_file_change.files", + "lifecycle.triggers.on_value_change.values", } func isJobRunTriggerPath(path string) bool { @@ -52,7 +56,8 @@ type JobRunTriggersState struct { // Fresh UUID each plan while armed so Old!=New forces recreate. OnBundleDeploy string `json:"on_bundle_deploy,omitempty"` // Content hashes from ResolveJobRunFileTriggers; any change recreates. - OnFileChange JobRunFileTriggerState `json:"on_file_change"` + OnFileChange JobRunFileTriggerState `json:"on_file_change"` + OnValueChange JobRunValueTriggerState `json:"on_value_change"` } // JobRunFileTriggerState wraps the hashes so turning the trigger off (Files nil) @@ -61,6 +66,12 @@ type JobRunFileTriggerState struct { Files map[string]string `json:"files,omitempty"` } +// JobRunValueTriggerState wraps expr → value so turning the trigger off +// (Values nil) diffs at one path, distinct from a single expression changing. +type JobRunValueTriggerState struct { + Values map[string]string `json:"values,omitempty"` +} + // UnmarshalJSON accepts the wrapped map and the older path-to-hash map. func (s *JobRunFileTriggerState) UnmarshalJSON(b []byte) error { var raw map[string]json.RawMessage @@ -92,6 +103,9 @@ func emptyJobRunLifecycleState() JobRunLifecycleState { OnFileChange: JobRunFileTriggerState{ Files: nil, }, + OnValueChange: JobRunValueTriggerState{ + Values: nil, + }, }, } } @@ -164,9 +178,59 @@ func (*ResourceJobRun) PrepareState(input *resources.JobRun) *JobRunState { if len(input.ResolvedFileTriggers) > 0 { state.Lifecycle.Triggers.OnFileChange.Files = input.ResolvedFileTriggers } + if values := jobRunValueChangeState(input); len(values) > 0 { + state.Lifecycle.Triggers.OnValueChange.Values = values + } return state } +func (*ResourceJobRun) PrepareInputConfig(input *resources.JobRun, _ string) (*structvar.StructVar, error) { + refs := map[string]string{} + for expr := range jobRunValueChangeState(input) { + if _, ok := dynvar.NewRef(dyn.V(expr)); !ok { + continue + } + path := structpath.NewStringKey(structpath.MustParsePath("lifecycle.triggers.on_value_change.values"), expr) + refs[path.String()] = expr + } + if len(refs) == 0 { + refs = nil + } + return &structvar.StructVar{Value: input, Refs: refs}, nil +} + +func jobRunValueChangeState(input *resources.JobRun) map[string]string { + if input.Lifecycle == nil { + return nil + } + out := make(map[string]string) + for _, t := range input.Lifecycle.Triggers { + if t.OnValueChange == nil { + continue + } + expr := strings.TrimSpace(*t.OnValueChange) + if expr == "" { + continue + } + out[expr] = expr + } + if len(out) == 0 { + return nil + } + return out +} + +// DropJobRunValueChangeConfigRefs drops lifecycle.triggers[N].on_value_change. +// ExtractReferences treats [0] on the triggers struct as a no-op, so that path +// is the wrapper, which cannot hold a resolved id. +func DropJobRunValueChangeConfigRefs(refs map[string]string) { + for k := range refs { + if strings.Contains(k, ".triggers[") && strings.HasSuffix(k, "].on_value_change") { + delete(refs, k) + } + } +} + // makeJobRunRemote maps the GetRun response into the RunNow-shaped remote: GET // nests the params under overriding_parameters and returns job_parameters as a // list, so both are flattened back into RunNow. @@ -450,6 +514,7 @@ func (*ResourceJobRun) OverrideChangeDesc(_ context.Context, path *structpath.Pa if isJobRunTriggerPath(pathString) { removed := pathString == "lifecycle.triggers.on_bundle_deploy" && (change.New == nil || change.New == "") removed = removed || pathString == "lifecycle.triggers.on_file_change.files" && change.New == nil + removed = removed || pathString == "lifecycle.triggers.on_value_change.values" && change.New == nil if removed { change.Action = deployplan.Update change.Reason = "trigger removed" diff --git a/bundle/direct/dresources/job_run_test.go b/bundle/direct/dresources/job_run_test.go index 78ec09d3604..02461041461 100644 --- a/bundle/direct/dresources/job_run_test.go +++ b/bundle/direct/dresources/job_run_test.go @@ -345,6 +345,71 @@ func TestJobRunReadOfUnfinishedRunReportsNoResult(t *testing.T) { assert.Empty(t, remote.ResultState) } +func TestJobRunPrepareStateOnValueChange(t *testing.T) { + expr := "${resources.jobs.foo.id}" + + t.Run("armed", func(t *testing.T) { + state := (&ResourceJobRun{}).PrepareState(&resources.JobRun{ + Lifecycle: &resources.JobRunLifecycle{ + Triggers: []resources.JobRunTrigger{{OnValueChange: &expr}}, + }, + }) + assert.Equal(t, map[string]string{expr: expr}, state.Lifecycle.Triggers.OnValueChange.Values) + assert.Empty(t, state.Lifecycle.Triggers.OnBundleDeploy) + assert.Nil(t, state.Lifecycle.Triggers.OnFileChange.Files) + }) + + t.Run("literal", func(t *testing.T) { + lit := "v1" + state := (&ResourceJobRun{}).PrepareState(&resources.JobRun{ + Lifecycle: &resources.JobRunLifecycle{ + Triggers: []resources.JobRunTrigger{{OnValueChange: &lit}}, + }, + }) + assert.Equal(t, map[string]string{"v1": "v1"}, state.Lifecycle.Triggers.OnValueChange.Values) + }) +} + +func TestJobRunPrepareInputConfigOnValueChange(t *testing.T) { + expr := "${resources.jobs.foo.id}" + input := &resources.JobRun{ + Lifecycle: &resources.JobRunLifecycle{ + Triggers: []resources.JobRunTrigger{{OnValueChange: &expr}}, + }, + } + sv, err := (&ResourceJobRun{}).PrepareInputConfig(input, "resources.job_runs.my_run") + require.NoError(t, err) + assert.Same(t, input, sv.Value) + path := structpath.NewStringKey(structpath.MustParsePath("lifecycle.triggers.on_value_change.values"), expr) + assert.Equal(t, map[string]string{path.String(): expr}, sv.Refs) +} + +func TestDropJobRunValueChangeConfigRefs(t *testing.T) { + expr := "${resources.jobs.other.id}" + valuesPath := structpath.NewStringKey(structpath.MustParsePath("lifecycle.triggers.on_value_change.values"), expr).String() + refs := map[string]string{ + "lifecycle.triggers[0].on_value_change": expr, + "job_id": "${resources.jobs.my_job.id}", + valuesPath: expr, + } + DropJobRunValueChangeConfigRefs(refs) + assert.Equal(t, map[string]string{ + "job_id": "${resources.jobs.my_job.id}", + valuesPath: expr, + }, refs) +} + +func TestJobRunPrepareInputConfigLiteralHasNoRefs(t *testing.T) { + lit := "v1" + sv, err := (&ResourceJobRun{}).PrepareInputConfig(&resources.JobRun{ + Lifecycle: &resources.JobRunLifecycle{ + Triggers: []resources.JobRunTrigger{{OnValueChange: &lit}}, + }, + }, "resources.job_runs.my_run") + require.NoError(t, err) + assert.Nil(t, sv.Refs) +} + // PrepareState records the outcome the run must reach, the same for every run, // so the planner has something to compare the remote against. func TestJobRunPrepareStateRequiresSuccess(t *testing.T) { @@ -352,6 +417,7 @@ func TestJobRunPrepareStateRequiresSuccess(t *testing.T) { assert.Equal(t, jobs.RunResultStateSuccess, state.ResultState) assert.Nil(t, state.Lifecycle.Triggers.OnFileChange.Files) + assert.Nil(t, state.Lifecycle.Triggers.OnValueChange.Values) } func TestJobRunPrepareStateOnBundleDeploy(t *testing.T) { @@ -494,6 +560,27 @@ func TestJobRunOverrideChangeDescTriggerRemoved(t *testing.T) { require.NoError(t, r.OverrideChangeDesc(t.Context(), structpath.MustParsePath("lifecycle.triggers.on_file_change.files['a.txt']"), change, nil)) assert.Equal(t, deployplan.Recreate, change.Action) }) + + t.Run("clearing on_value_change values downgrades to update", func(t *testing.T) { + change := &ChangeDesc{ + Action: deployplan.Recreate, + Old: map[string]string{"${resources.jobs.foo.id}": "1"}, + New: nil, + } + require.NoError(t, r.OverrideChangeDesc(t.Context(), structpath.MustParsePath("lifecycle.triggers.on_value_change.values"), change, nil)) + assert.Equal(t, deployplan.Update, change.Action) + assert.Equal(t, "trigger removed", change.Reason) + }) + + t.Run("changed on_value_change still recreates", func(t *testing.T) { + change := &ChangeDesc{ + Action: deployplan.Recreate, + Old: "1", + New: "2", + } + require.NoError(t, r.OverrideChangeDesc(t.Context(), structpath.MustParsePath("lifecycle.triggers.on_value_change.values['${resources.jobs.foo.id}']"), change, nil)) + assert.Equal(t, deployplan.Recreate, change.Action) + }) } // The planner diffs RemapState(remote) against PrepareState(config), so a run diff --git a/bundle/internal/schema/annotations.yml b/bundle/internal/schema/annotations.yml index 1fd6b452415..111edc74670 100644 --- a/bundle/internal/schema/annotations.yml +++ b/bundle/internal/schema/annotations.yml @@ -993,6 +993,9 @@ resources: "on_file_change": "description": |- Path or glob under the sync root. Re-fire the run when a matched file's content hash changes, or when the set of matches appears or disappears. Incompatible with lifecycle.prevent_destroy. + "on_value_change": + "description": |- + Interpolated expression. Re-fire the run when the resolved value changes. Resource references in the expression become deployment dependencies. Incompatible with lifecycle.prevent_destroy. "python_named_params": "description": |- PLACEHOLDER diff --git a/bundle/schema/jsonschema.json b/bundle/schema/jsonschema.json index 39d31e92167..4172c6a20e1 100644 --- a/bundle/schema/jsonschema.json +++ b/bundle/schema/jsonschema.json @@ -1304,6 +1304,10 @@ "on_file_change": { "description": "Path or glob under the sync root. Re-fire the run when a matched file's content hash changes, or when the set of matches appears or disappears. Incompatible with lifecycle.prevent_destroy.", "$ref": "#/$defs/string" + }, + "on_value_change": { + "description": "Interpolated expression. Re-fire the run when the resolved value changes. Resource references in the expression become deployment dependencies. Incompatible with lifecycle.prevent_destroy.", + "$ref": "#/$defs/string" } }, "additionalProperties": false From 0db1f4b8426d084e2fad3eb2039efbf3a6c9d8dd Mon Sep 17 00:00:00 2001 From: Rada Kamysheva Date: Fri, 21 Aug 2026 16:13:05 +0000 Subject: [PATCH 2/2] Regenerate job_runs reference schema fields Include the on_value_change input and state paths required by generated-file validation. --- acceptance/bundle/refschema/out.fields.txt | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/acceptance/bundle/refschema/out.fields.txt b/acceptance/bundle/refschema/out.fields.txt index 691c5589bcc..ad463164147 100644 --- a/acceptance/bundle/refschema/out.fields.txt +++ b/acceptance/bundle/refschema/out.fields.txt @@ -855,9 +855,13 @@ resources.job_runs.*.lifecycle.triggers.on_bundle_deploy string STATE resources.job_runs.*.lifecycle.triggers.on_file_change dresources.JobRunFileTriggerState STATE resources.job_runs.*.lifecycle.triggers.on_file_change.files map[string]string STATE resources.job_runs.*.lifecycle.triggers.on_file_change.files.* string STATE +resources.job_runs.*.lifecycle.triggers.on_value_change dresources.JobRunValueTriggerState STATE +resources.job_runs.*.lifecycle.triggers.on_value_change.values map[string]string STATE +resources.job_runs.*.lifecycle.triggers.on_value_change.values.* string STATE resources.job_runs.*.lifecycle.triggers[*] resources.JobRunTrigger INPUT resources.job_runs.*.lifecycle.triggers[*].on_bundle_deploy *bool INPUT resources.job_runs.*.lifecycle.triggers[*].on_file_change *string INPUT +resources.job_runs.*.lifecycle.triggers[*].on_value_change *string INPUT resources.job_runs.*.modified_status string INPUT resources.job_runs.*.notebook_params map[string]string ALL resources.job_runs.*.notebook_params.* string ALL