From 22c35c399aa0e02a48e54fc3b263486df1b2ff33 Mon Sep 17 00:00:00 2001 From: Yuri Barreira Date: Wed, 14 Oct 2026 15:57:58 -0300 Subject: [PATCH 1/4] Serialize complex variables used in for_each_task.inputs A complex variable substituted into for_each_task.inputs was dropped by normalization with 'expected string, found sequence', leaving the required field unset. Serialize pure variable references to complex values into the JSON string the Jobs API expects, before variable resolution runs. Fixes #6901. --- .nextchanges/bundles/6901.md | 1 + .../complex-foreach-inputs/databricks.yml | 34 +++ .../variables/complex-foreach-inputs/hello.py | 1 + .../complex-foreach-inputs/out.test.toml | 2 + .../complex-foreach-inputs/out.validate.json | 75 ++++++ .../complex-foreach-inputs/output.txt | 5 + .../variables/complex-foreach-inputs/script | 2 + .../complex-foreach-inputs/test.toml | 1 + .../process_static_resources.go | 6 + .../resourcemutator/resource_mutator.go | 6 + .../serialize_for_each_task_inputs.go | 145 +++++++++++ .../serialize_for_each_task_inputs_test.go | 227 ++++++++++++++++++ 12 files changed, 505 insertions(+) create mode 100644 .nextchanges/bundles/6901.md create mode 100644 acceptance/bundle/variables/complex-foreach-inputs/databricks.yml create mode 100644 acceptance/bundle/variables/complex-foreach-inputs/hello.py create mode 100644 acceptance/bundle/variables/complex-foreach-inputs/out.test.toml create mode 100644 acceptance/bundle/variables/complex-foreach-inputs/out.validate.json create mode 100644 acceptance/bundle/variables/complex-foreach-inputs/output.txt create mode 100644 acceptance/bundle/variables/complex-foreach-inputs/script create mode 100644 acceptance/bundle/variables/complex-foreach-inputs/test.toml create mode 100644 bundle/config/mutator/resourcemutator/serialize_for_each_task_inputs.go create mode 100644 bundle/config/mutator/resourcemutator/serialize_for_each_task_inputs_test.go diff --git a/.nextchanges/bundles/6901.md b/.nextchanges/bundles/6901.md new file mode 100644 index 00000000000..a657235ec10 --- /dev/null +++ b/.nextchanges/bundles/6901.md @@ -0,0 +1 @@ +* Fixed `type: complex` variables failing with `expected string, found sequence` when referenced from `for_each_task.inputs`; the CLI now serializes the referenced value to the JSON string the Jobs API expects (see [#6901](https://github.com/databricks/cli/issues/6901)). diff --git a/acceptance/bundle/variables/complex-foreach-inputs/databricks.yml b/acceptance/bundle/variables/complex-foreach-inputs/databricks.yml new file mode 100644 index 00000000000..9eda6e9ba90 --- /dev/null +++ b/acceptance/bundle/variables/complex-foreach-inputs/databricks.yml @@ -0,0 +1,34 @@ +bundle: + name: complex-foreach-inputs + +variables: + table_list: + type: complex + description: "Tables to load from source" + default: + - task_id: load_customers + table: customers + source_type: parquet + - task_id: load_orders + table: orders + source_type: json + +resources: + jobs: + my_job: + job_clusters: + - job_cluster_key: c + new_cluster: + spark_version: 15.4.x-scala2.12 + node_type_id: i3.xlarge + num_workers: 1 + tasks: + - task_key: process_tables + for_each_task: + inputs: ${var.table_list} + concurrency: 2 + task: + task_key: load_table + job_cluster_key: c + spark_python_task: + python_file: ./hello.py diff --git a/acceptance/bundle/variables/complex-foreach-inputs/hello.py b/acceptance/bundle/variables/complex-foreach-inputs/hello.py new file mode 100644 index 00000000000..11b15b1a458 --- /dev/null +++ b/acceptance/bundle/variables/complex-foreach-inputs/hello.py @@ -0,0 +1 @@ +print("hello") diff --git a/acceptance/bundle/variables/complex-foreach-inputs/out.test.toml b/acceptance/bundle/variables/complex-foreach-inputs/out.test.toml new file mode 100644 index 00000000000..a927a5fbc06 --- /dev/null +++ b/acceptance/bundle/variables/complex-foreach-inputs/out.test.toml @@ -0,0 +1,2 @@ +Cloud = false +EnvMatrix.DMS = ["", "true"] diff --git a/acceptance/bundle/variables/complex-foreach-inputs/out.validate.json b/acceptance/bundle/variables/complex-foreach-inputs/out.validate.json new file mode 100644 index 00000000000..47effd1b490 --- /dev/null +++ b/acceptance/bundle/variables/complex-foreach-inputs/out.validate.json @@ -0,0 +1,75 @@ +{ + "resources": { + "jobs": { + "my_job": { + "deployment": { + "kind": "BUNDLE", + "metadata_file_path": "/Workspace/Users/[USERNAME]/.bundle/complex-foreach-inputs/default/state/metadata.json" + }, + "edit_mode": "UI_LOCKED", + "format": "MULTI_TASK", + "job_clusters": [ + { + "job_cluster_key": "c", + "new_cluster": { + "node_type_id": "[NODE_TYPE_ID]", + "num_workers": 1, + "spark_version": "15.4.x-scala2.12" + } + } + ], + "max_concurrent_runs": 1, + "name": "Untitled", + "queue": { + "enabled": true + }, + "tasks": [ + { + "for_each_task": { + "concurrency": 2, + "inputs": "[{\"source_type\":\"parquet\",\"table\":\"customers\",\"task_id\":\"load_customers\"},{\"source_type\":\"json\",\"table\":\"orders\",\"task_id\":\"load_orders\"}]", + "task": { + "job_cluster_key": "c", + "spark_python_task": { + "python_file": "/Workspace/Users/[USERNAME]/.bundle/complex-foreach-inputs/default/files/hello.py" + }, + "task_key": "load_table" + } + }, + "task_key": "process_tables" + } + ] + } + } + }, + "variables": { + "table_list": { + "default": [ + { + "source_type": "parquet", + "table": "customers", + "task_id": "load_customers" + }, + { + "source_type": "json", + "table": "orders", + "task_id": "load_orders" + } + ], + "description": "Tables to load from source", + "type": "complex", + "value": [ + { + "source_type": "parquet", + "table": "customers", + "task_id": "load_customers" + }, + { + "source_type": "json", + "table": "orders", + "task_id": "load_orders" + } + ] + } + } +} diff --git a/acceptance/bundle/variables/complex-foreach-inputs/output.txt b/acceptance/bundle/variables/complex-foreach-inputs/output.txt new file mode 100644 index 00000000000..e521b6d9df4 --- /dev/null +++ b/acceptance/bundle/variables/complex-foreach-inputs/output.txt @@ -0,0 +1,5 @@ + +>>> [CLI] bundle validate -o json + +>>> jq -r .resources.jobs.my_job.tasks[0].for_each_task.inputs out.validate.json +[{"source_type":"parquet","table":"customers","task_id":"load_customers"},{"source_type":"json","table":"orders","task_id":"load_orders"}] diff --git a/acceptance/bundle/variables/complex-foreach-inputs/script b/acceptance/bundle/variables/complex-foreach-inputs/script new file mode 100644 index 00000000000..8f75dc13b0c --- /dev/null +++ b/acceptance/bundle/variables/complex-foreach-inputs/script @@ -0,0 +1,2 @@ +trace $CLI bundle validate -o json | jq '{resources,variables}' > out.validate.json +trace jq -r .resources.jobs.my_job.tasks[0].for_each_task.inputs out.validate.json | grep -F '[{"source_type":"parquet","table":"customers","task_id":"load_customers"},{"source_type":"json","table":"orders","task_id":"load_orders"}]' diff --git a/acceptance/bundle/variables/complex-foreach-inputs/test.toml b/acceptance/bundle/variables/complex-foreach-inputs/test.toml new file mode 100644 index 00000000000..601384fdf96 --- /dev/null +++ b/acceptance/bundle/variables/complex-foreach-inputs/test.toml @@ -0,0 +1 @@ +Ignore = [".databricks"] diff --git a/bundle/config/mutator/resourcemutator/process_static_resources.go b/bundle/config/mutator/resourcemutator/process_static_resources.go index b1f689808f6..015afa662f1 100644 --- a/bundle/config/mutator/resourcemutator/process_static_resources.go +++ b/bundle/config/mutator/resourcemutator/process_static_resources.go @@ -42,6 +42,12 @@ func (p processStaticResources) Apply(ctx context.Context, b *bundle.Bundle) dia bundle.ApplySeqContext( ctx, b, + // Reads (dynamic): resources.jobs.*.tasks for_each_task.inputs, variables.*.value + // Updates (dynamic): resources.jobs.*.tasks for_each_task.inputs (complex variable reference -> JSON string) + // Must run before variable resolution: a substituted sequence would be + // dropped by normalization with an "expected string" warning. + SerializeForEachTaskInputs(), + // Reads (dynamic): * (strings) (searches for variable references in string values) // Updates (dynamic): resources.* (strings) (resolves variable references to their actual values) // Resolves variable references in 'resources' using bundle, workspace, and variables prefixes diff --git a/bundle/config/mutator/resourcemutator/resource_mutator.go b/bundle/config/mutator/resourcemutator/resource_mutator.go index 4d0b3bd47ab..1f3cdc25de5 100644 --- a/bundle/config/mutator/resourcemutator/resource_mutator.go +++ b/bundle/config/mutator/resourcemutator/resource_mutator.go @@ -143,6 +143,12 @@ func applyNormalizeMutators(ctx context.Context, b *bundle.Bundle) { validate.SingleNodeCluster(), + // Reads (dynamic): resources.jobs.*.tasks for_each_task.inputs, variables.*.value + // Updates (dynamic): resources.jobs.*.tasks for_each_task.inputs (complex variable reference -> JSON string) + // Must run before variable resolution: a substituted sequence would be + // dropped by normalization with an "expected string" warning. + SerializeForEachTaskInputs(), + // Reads (dynamic): * (strings) (searches for variable references in string values) // Updates (dynamic): resources.* (strings) (resolves variable references to their actual values) // Resolves variable references in 'resources' using bundle, workspace, and variables prefixes diff --git a/bundle/config/mutator/resourcemutator/serialize_for_each_task_inputs.go b/bundle/config/mutator/resourcemutator/serialize_for_each_task_inputs.go new file mode 100644 index 00000000000..0ef76181037 --- /dev/null +++ b/bundle/config/mutator/resourcemutator/serialize_for_each_task_inputs.go @@ -0,0 +1,145 @@ +package resourcemutator + +import ( + "context" + "encoding/json" + "fmt" + + "github.com/databricks/cli/bundle" + "github.com/databricks/cli/libs/diag" + "github.com/databricks/cli/libs/dyn" + "github.com/databricks/cli/libs/dyn/dynvar" +) + +// Suffix matching for_each_task.inputs at any nesting depth within a job. +var forEachTaskInputsSuffix = dyn.NewPath(dyn.Key("for_each_task"), dyn.Key("inputs")) + +var varShorthandPrefix = dyn.NewPath(dyn.Key("var")) + +type serializeForEachTaskInputs struct{} + +// SerializeForEachTaskInputs converts a complex variable referenced by +// for_each_task.inputs into the JSON string the Jobs API expects. +// +// The inputs field is typed as a string, so a substituted sequence or map +// would be dropped by normalization with an "expected string, found sequence" +// warning. Serializing here, before variable resolution runs, keeps the value +// intact. Anything that is not a pure reference to a complex variable value +// is left untouched for the regular resolution and validation to handle. +func SerializeForEachTaskInputs() bundle.Mutator { + return &serializeForEachTaskInputs{} +} + +func (s serializeForEachTaskInputs) Name() string { + return "SerializeForEachTaskInputs" +} + +func (s serializeForEachTaskInputs) Apply(_ context.Context, b *bundle.Bundle) diag.Diagnostics { + var diags diag.Diagnostics + + pattern := dyn.NewPattern( + dyn.Key("resources"), + dyn.Key("jobs"), + dyn.AnyKey(), + ) + + err := b.Config.Mutate(func(root dyn.Value) (dyn.Value, error) { + return dyn.MapByPattern(root, pattern, func(_ dyn.Path, job dyn.Value) (dyn.Value, error) { + return dyn.Walk(job, func(p dyn.Path, v dyn.Value) (dyn.Value, error) { + if !p.HasSuffix(forEachTaskInputsSuffix) { + return v, nil + } + return serializeInputsValue(root, v) + }) + }) + }) + + diags = diags.Extend(diag.FromErr(err)) + return diags +} + +func serializeInputsValue(root, v dyn.Value) (dyn.Value, error) { + s, ok := v.AsString() + if !ok { + return v, nil + } + + path, ok := dynvar.PureReferenceToPath(s) + if !ok { + return v, nil + } + + path, ok = variableValuePath(path) + if !ok { + return v, nil + } + + target, ok := lookupVariableValue(root, path) + if !ok { + return v, nil + } + + switch target.Kind() { + case dyn.KindSequence, dyn.KindMap: + default: + return v, nil + } + + // Only serialize fully resolved values. Baking an unresolved reference + // into JSON would change its meaning from structural substitution to + // string interpolation when a later resolution round picks it up. + if containsVariableReference(target) { + return v, nil + } + + jsonBytes, err := json.Marshal(target.AsAny()) + if err != nil { + return dyn.InvalidValue, fmt.Errorf("failed to serialize for_each_task.inputs: %w", err) + } + + return dyn.NewValue(string(jsonBytes), v.Locations()), nil +} + +// variableValuePath maps a pure reference to the variable value it points to. +// ${var.foo} is shorthand for ${variables.foo.value}; longer paths address +// into the value itself. References to anything else are not variable values. +func variableValuePath(path dyn.Path) (dyn.Path, bool) { + if path.HasPrefix(varShorthandPrefix) { + if len(path) < 2 { + return nil, false + } + out := dyn.NewPath(dyn.Key("variables"), path[1], dyn.Key("value")) + if len(path) > 2 { + out = out.Append(path[2:]...) + } + return out, true + } + + if len(path) > 0 && path[0].Key() == "variables" { + return path, true + } + + return nil, false +} + +// lookupVariableValue resolves a variable value path against the current +// configuration. A missing reference is not an error here; the regular +// variable resolution reports it with suggestions. +func lookupVariableValue(root dyn.Value, path dyn.Path) (dyn.Value, bool) { + target, err := dyn.GetByPath(root, path) + if err != nil || !target.IsValid() { + return dyn.InvalidValue, false + } + return target, true +} + +func containsVariableReference(v dyn.Value) bool { + found := false + _ = dyn.WalkReadOnly(v, func(_ dyn.Path, v dyn.Value) error { + if s, ok := v.AsString(); ok && dynvar.ContainsVariableReference(s) { + found = true + } + return nil + }) + return found +} diff --git a/bundle/config/mutator/resourcemutator/serialize_for_each_task_inputs_test.go b/bundle/config/mutator/resourcemutator/serialize_for_each_task_inputs_test.go new file mode 100644 index 00000000000..3d760c38b57 --- /dev/null +++ b/bundle/config/mutator/resourcemutator/serialize_for_each_task_inputs_test.go @@ -0,0 +1,227 @@ +package resourcemutator_test + +import ( + "testing" + + "github.com/databricks/cli/bundle" + "github.com/databricks/cli/bundle/config" + "github.com/databricks/cli/bundle/config/mutator" + "github.com/databricks/cli/bundle/config/mutator/resourcemutator" + "github.com/databricks/cli/bundle/config/resources" + "github.com/databricks/cli/bundle/config/variable" + "github.com/databricks/databricks-sdk-go/service/jobs" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// forEachTestBundle builds a bundle with a single job whose for_each_task.inputs +// is the given value, plus the given variables. +func forEachTestBundle(inputs string, vars map[string]*variable.Variable) *bundle.Bundle { + return &bundle.Bundle{ + Config: config.Root{ + Variables: vars, + Resources: config.Resources{ + Jobs: map[string]*resources.Job{ + "my_job": { + JobSettings: jobs.JobSettings{ + Tasks: []jobs.Task{ + { + TaskKey: "process_tables", + ForEachTask: &jobs.ForEachTask{ + Inputs: inputs, + Task: jobs.Task{ + TaskKey: "load_table", + }, + }, + }, + }, + }, + }, + }, + }, + }, + } +} + +func forEachInputs(t *testing.T, b *bundle.Bundle) string { + t.Helper() + tasks := b.Config.Resources.Jobs["my_job"].Tasks + require.Len(t, tasks, 1) + require.NotNil(t, tasks[0].ForEachTask) + return tasks[0].ForEachTask.Inputs +} + +func TestSerializeForEachTaskInputs(t *testing.T) { + complexValue := []any{ + map[string]any{"task_id": "load_customers", "table": "customers", "source_type": "parquet"}, + map[string]any{"task_id": "load_orders", "table": "orders", "source_type": "json"}, + } + + tests := []struct { + name string + inputs string + vars map[string]*variable.Variable + // wantJSON is compared semantically against the resulting inputs string. + // Empty means inputs must be left unchanged. + wantJSON string + }{ + { + name: "complex variable array of objects is serialized", + inputs: "${var.table_list}", + vars: map[string]*variable.Variable{ + "table_list": {Type: variable.VariableTypeComplex, Value: complexValue}, + }, + wantJSON: `[{"source_type":"parquet","table":"customers","task_id":"load_customers"},{"source_type":"json","table":"orders","task_id":"load_orders"}]`, + }, + { + name: "long variable reference form is serialized", + inputs: "${variables.table_list.value}", + vars: map[string]*variable.Variable{ + "table_list": {Type: variable.VariableTypeComplex, Value: complexValue}, + }, + wantJSON: `[{"source_type":"parquet","table":"customers","task_id":"load_customers"},{"source_type":"json","table":"orders","task_id":"load_orders"}]`, + }, + { + name: "empty array is serialized", + inputs: "${var.table_list}", + vars: map[string]*variable.Variable{ + "table_list": {Type: variable.VariableTypeComplex, Value: []any{}}, + }, + wantJSON: `[]`, + }, + { + name: "escaping and nested values are serialized", + inputs: "${var.table_list}", + vars: map[string]*variable.Variable{ + "table_list": {Type: variable.VariableTypeComplex, Value: []any{ + map[string]any{ + "task_id": `load "quoted" \\ ünïcodé`, + "nested": map[string]any{"list": []any{float64(1), true, nil, "x\ny"}}, + }, + }}, + }, + wantJSON: `[{"nested":{"list":[1,true,null,"x\ny"]},"task_id":"load \"quoted\" \\\\ ünïcodé"}]`, + }, + { + name: "existing JSON string is left unchanged", + inputs: `[{"task_id":"load_customers"}]`, + vars: map[string]*variable.Variable{ + "table_list": {Type: variable.VariableTypeComplex, Value: complexValue}, + }, + }, + { + name: "scalar variable reference is left for normal resolution", + inputs: "${var.table_name}", + vars: map[string]*variable.Variable{ + "table_name": {Value: "customers"}, + }, + }, + { + name: "non-variable reference is left unchanged", + inputs: "${bundle.name}", + vars: map[string]*variable.Variable{ + "table_list": {Type: variable.VariableTypeComplex, Value: complexValue}, + }, + }, + { + name: "interpolated reference is left unchanged", + inputs: `prefix-${var.table_list}`, + vars: map[string]*variable.Variable{ + "table_list": {Type: variable.VariableTypeComplex, Value: complexValue}, + }, + }, + { + name: "missing variable is left for normal resolution", + inputs: "${var.does_not_exist}", + vars: map[string]*variable.Variable{ + "table_list": {Type: variable.VariableTypeComplex, Value: complexValue}, + }, + }, + { + name: "complex value with nested reference is left unchanged", + inputs: "${var.table_list}", + vars: map[string]*variable.Variable{ + "table_list": {Type: variable.VariableTypeComplex, Value: []any{ + map[string]any{"task_id": "${resources.jobs.other.name}"}, + }}, + }, + }, + { + name: "subpath into complex variable is serialized", + inputs: "${var.wrapped.items}", + vars: map[string]*variable.Variable{ + "wrapped": {Type: variable.VariableTypeComplex, Value: map[string]any{"items": complexValue}}, + }, + wantJSON: `[{"source_type":"parquet","table":"customers","task_id":"load_customers"},{"source_type":"json","table":"orders","task_id":"load_orders"}]`, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + b := forEachTestBundle(tt.inputs, tt.vars) + + diags := bundle.Apply(t.Context(), b, resourcemutator.SerializeForEachTaskInputs()) + require.NoError(t, diags.Error()) + + got := forEachInputs(t, b) + if tt.wantJSON == "" { + assert.Equal(t, tt.inputs, got) + return + } + assert.JSONEq(t, tt.wantJSON, got) + }) + } +} + +func TestSerializeForEachTaskInputsNestedForEach(t *testing.T) { + b := forEachTestBundle("${var.outer}", map[string]*variable.Variable{ + "outer": {Type: variable.VariableTypeComplex, Value: []any{map[string]any{"a": "b"}}}, + "inner": {Type: variable.VariableTypeComplex, Value: []any{map[string]any{"c": "d"}}}, + }) + // Nest a second for_each_task inside the nested task. + outer := &b.Config.Resources.Jobs["my_job"].Tasks[0] + outer.ForEachTask.Task = jobs.Task{ + TaskKey: "middle", + ForEachTask: &jobs.ForEachTask{Inputs: "${var.inner}"}, + } + + diags := bundle.Apply(t.Context(), b, resourcemutator.SerializeForEachTaskInputs()) + require.NoError(t, diags.Error()) + + assert.JSONEq(t, `[{"a":"b"}]`, forEachInputs(t, b)) + assert.JSONEq(t, `[{"c":"d"}]`, b.Config.Resources.Jobs["my_job"].Tasks[0].ForEachTask.Task.ForEachTask.Inputs) +} + +// Complex variables substituted into unrelated string fields must keep failing +// validation as before; the serialization only applies to for_each_task.inputs. +func TestSerializeForEachTaskInputsPreservesUnrelatedValidation(t *testing.T) { + b := &bundle.Bundle{ + Config: config.Root{ + Variables: map[string]*variable.Variable{ + "table_list": {Type: variable.VariableTypeComplex, Value: []any{ + map[string]any{"task_id": "load_customers"}, + }}, + }, + Resources: config.Resources{ + Jobs: map[string]*resources.Job{ + "my_job": { + JobSettings: jobs.JobSettings{ + Name: "${var.table_list}", + }, + }, + }, + }, + }, + } + + diags := bundle.Apply(t.Context(), b, resourcemutator.SerializeForEachTaskInputs()) + require.NoError(t, diags.Error()) + assert.Equal(t, "${var.table_list}", b.Config.Resources.Jobs["my_job"].Name) + + diags = bundle.Apply(t.Context(), b, mutator.ResolveVariableReferencesOnlyResources()) + summaries := make([]string, 0, len(diags)) + for _, d := range diags { + summaries = append(summaries, d.Summary) + } + assert.Contains(t, summaries, "expected string, found sequence") +} From 73c12e3b111dc99516ee9421c4f3ac5f02118e4a Mon Sep 17 00:00:00 2001 From: Yuri Barreira Date: Wed, 14 Oct 2026 16:00:00 -0300 Subject: [PATCH 2/4] Link PR in changelog fragment --- .nextchanges/bundles/6901.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.nextchanges/bundles/6901.md b/.nextchanges/bundles/6901.md index a657235ec10..df39f027534 100644 --- a/.nextchanges/bundles/6901.md +++ b/.nextchanges/bundles/6901.md @@ -1 +1 @@ -* Fixed `type: complex` variables failing with `expected string, found sequence` when referenced from `for_each_task.inputs`; the CLI now serializes the referenced value to the JSON string the Jobs API expects (see [#6901](https://github.com/databricks/cli/issues/6901)). +* Fixed `type: complex` variables failing with `expected string, found sequence` when referenced from `for_each_task.inputs`; the CLI now serializes the referenced value to the JSON string the Jobs API expects (see [#6901](https://github.com/databricks/cli/issues/6901)). ([#7067](https://github.com/databricks/cli/pull/7067)) From 5c9491a7eb89ee240820e359f41018b93f8c9f9f Mon Sep 17 00:00:00 2001 From: Yuri Barreira Date: Wed, 14 Oct 2026 16:28:33 -0300 Subject: [PATCH 3/4] Move for_each_task.inputs serialization into variable resolution Serializing before resolution missed inputs nested inside variable-defined tasks, and the reference guard did not cover map keys. Serialize after substitution in resolveOnce instead, and scan keys too. --- .../complex-foreach-inputs/databricks.yml | 21 + .../complex-foreach-inputs/out.validate.json | 97 +++++ .../complex-foreach-inputs/output.txt | 3 + .../variables/complex-foreach-inputs/script | 1 + bundle/config/mutator/for_each_task_inputs.go | 87 ++++ .../mutator/for_each_task_inputs_test.go | 381 ++++++++++++++++++ .../mutator/resolve_variable_references.go | 8 + .../process_static_resources.go | 6 - .../resourcemutator/resource_mutator.go | 6 - .../serialize_for_each_task_inputs.go | 145 ------- .../serialize_for_each_task_inputs_test.go | 227 ----------- 11 files changed, 598 insertions(+), 384 deletions(-) create mode 100644 bundle/config/mutator/for_each_task_inputs.go create mode 100644 bundle/config/mutator/for_each_task_inputs_test.go delete mode 100644 bundle/config/mutator/resourcemutator/serialize_for_each_task_inputs.go delete mode 100644 bundle/config/mutator/resourcemutator/serialize_for_each_task_inputs_test.go diff --git a/acceptance/bundle/variables/complex-foreach-inputs/databricks.yml b/acceptance/bundle/variables/complex-foreach-inputs/databricks.yml index 9eda6e9ba90..6412b3185f9 100644 --- a/acceptance/bundle/variables/complex-foreach-inputs/databricks.yml +++ b/acceptance/bundle/variables/complex-foreach-inputs/databricks.yml @@ -12,6 +12,19 @@ variables: - task_id: load_orders table: orders source_type: json + job_tasks: + type: complex + description: "Shared tasks with a nested inputs reference" + default: + - task_key: process_tables + for_each_task: + inputs: ${var.table_list} + concurrency: 2 + task: + task_key: load_table + job_cluster_key: c + spark_python_task: + python_file: ./hello.py resources: jobs: @@ -32,3 +45,11 @@ resources: job_cluster_key: c spark_python_task: python_file: ./hello.py + var_tasks_job: + job_clusters: + - job_cluster_key: c + new_cluster: + spark_version: 15.4.x-scala2.12 + node_type_id: i3.xlarge + num_workers: 1 + tasks: ${var.job_tasks} diff --git a/acceptance/bundle/variables/complex-foreach-inputs/out.validate.json b/acceptance/bundle/variables/complex-foreach-inputs/out.validate.json index 47effd1b490..72b1f7c86db 100644 --- a/acceptance/bundle/variables/complex-foreach-inputs/out.validate.json +++ b/acceptance/bundle/variables/complex-foreach-inputs/out.validate.json @@ -39,10 +39,107 @@ "task_key": "process_tables" } ] + }, + "var_tasks_job": { + "deployment": { + "kind": "BUNDLE", + "metadata_file_path": "/Workspace/Users/[USERNAME]/.bundle/complex-foreach-inputs/default/state/metadata.json" + }, + "edit_mode": "UI_LOCKED", + "format": "MULTI_TASK", + "job_clusters": [ + { + "job_cluster_key": "c", + "new_cluster": { + "node_type_id": "[NODE_TYPE_ID]", + "num_workers": 1, + "spark_version": "15.4.x-scala2.12" + } + } + ], + "max_concurrent_runs": 1, + "name": "Untitled", + "queue": { + "enabled": true + }, + "tasks": [ + { + "for_each_task": { + "concurrency": 2, + "inputs": "[{\"source_type\":\"parquet\",\"table\":\"customers\",\"task_id\":\"load_customers\"},{\"source_type\":\"json\",\"table\":\"orders\",\"task_id\":\"load_orders\"}]", + "task": { + "job_cluster_key": "c", + "spark_python_task": { + "python_file": "/Workspace/Users/[USERNAME]/.bundle/complex-foreach-inputs/default/files/hello.py" + }, + "task_key": "load_table" + } + }, + "task_key": "process_tables" + } + ] } } }, "variables": { + "job_tasks": { + "default": [ + { + "for_each_task": { + "concurrency": 2, + "inputs": [ + { + "source_type": "parquet", + "table": "customers", + "task_id": "load_customers" + }, + { + "source_type": "json", + "table": "orders", + "task_id": "load_orders" + } + ], + "task": { + "job_cluster_key": "c", + "spark_python_task": { + "python_file": "./hello.py" + }, + "task_key": "load_table" + } + }, + "task_key": "process_tables" + } + ], + "description": "Shared tasks with a nested inputs reference", + "type": "complex", + "value": [ + { + "for_each_task": { + "concurrency": 2, + "inputs": [ + { + "source_type": "parquet", + "table": "customers", + "task_id": "load_customers" + }, + { + "source_type": "json", + "table": "orders", + "task_id": "load_orders" + } + ], + "task": { + "job_cluster_key": "c", + "spark_python_task": { + "python_file": "./hello.py" + }, + "task_key": "load_table" + } + }, + "task_key": "process_tables" + } + ] + }, "table_list": { "default": [ { diff --git a/acceptance/bundle/variables/complex-foreach-inputs/output.txt b/acceptance/bundle/variables/complex-foreach-inputs/output.txt index e521b6d9df4..716e386a09b 100644 --- a/acceptance/bundle/variables/complex-foreach-inputs/output.txt +++ b/acceptance/bundle/variables/complex-foreach-inputs/output.txt @@ -3,3 +3,6 @@ >>> jq -r .resources.jobs.my_job.tasks[0].for_each_task.inputs out.validate.json [{"source_type":"parquet","table":"customers","task_id":"load_customers"},{"source_type":"json","table":"orders","task_id":"load_orders"}] + +>>> jq -r .resources.jobs.var_tasks_job.tasks[0].for_each_task.inputs out.validate.json +[{"source_type":"parquet","table":"customers","task_id":"load_customers"},{"source_type":"json","table":"orders","task_id":"load_orders"}] diff --git a/acceptance/bundle/variables/complex-foreach-inputs/script b/acceptance/bundle/variables/complex-foreach-inputs/script index 8f75dc13b0c..cf5f112e5d0 100644 --- a/acceptance/bundle/variables/complex-foreach-inputs/script +++ b/acceptance/bundle/variables/complex-foreach-inputs/script @@ -1,2 +1,3 @@ trace $CLI bundle validate -o json | jq '{resources,variables}' > out.validate.json trace jq -r .resources.jobs.my_job.tasks[0].for_each_task.inputs out.validate.json | grep -F '[{"source_type":"parquet","table":"customers","task_id":"load_customers"},{"source_type":"json","table":"orders","task_id":"load_orders"}]' +trace jq -r .resources.jobs.var_tasks_job.tasks[0].for_each_task.inputs out.validate.json | grep -F '[{"source_type":"parquet","table":"customers","task_id":"load_customers"},{"source_type":"json","table":"orders","task_id":"load_orders"}]' diff --git a/bundle/config/mutator/for_each_task_inputs.go b/bundle/config/mutator/for_each_task_inputs.go new file mode 100644 index 00000000000..9ad05131271 --- /dev/null +++ b/bundle/config/mutator/for_each_task_inputs.go @@ -0,0 +1,87 @@ +package mutator + +import ( + "encoding/json" + "fmt" + + "github.com/databricks/cli/libs/dyn" + "github.com/databricks/cli/libs/dyn/dynvar" +) + +// forEachTaskInputsSuffix matches for_each_task.inputs at any nesting depth. +var forEachTaskInputsSuffix = dyn.NewPath(dyn.Key("for_each_task"), dyn.Key("inputs")) + +// serializeForEachTaskInputs converts complex values substituted into +// for_each_task.inputs into the JSON string the Jobs API expects. +// +// The inputs field is typed as a string, so a substituted sequence or map +// would be dropped by normalization with an "expected string, found sequence" +// warning. Serializing here, after substitution and before normalization, +// keeps the value intact. This covers both direct references +// (inputs: ${var.table_list}) and references nested inside variable-defined +// structures (tasks: ${var.job_tasks}), which only materialize once the +// enclosing reference is substituted. +// +// Values still containing references are left untouched. Baking an unresolved +// reference into JSON would change its meaning from structural substitution +// to string interpolation, which coerces non-string values to text and can +// corrupt the JSON when the result contains quotes. Normalization then +// reports such values as before. +func serializeForEachTaskInputs(root dyn.Value) (dyn.Value, error) { + pattern := dyn.NewPattern( + dyn.Key("resources"), + dyn.Key("jobs"), + dyn.AnyKey(), + ) + + return dyn.MapByPattern(root, pattern, func(_ dyn.Path, job dyn.Value) (dyn.Value, error) { + return dyn.Walk(job, func(p dyn.Path, v dyn.Value) (dyn.Value, error) { + if !p.HasSuffix(forEachTaskInputsSuffix) { + return v, nil + } + return serializeInputsValue(v) + }) + }) +} + +func serializeInputsValue(v dyn.Value) (dyn.Value, error) { + switch v.Kind() { + case dyn.KindSequence, dyn.KindMap: + default: + return v, nil + } + + if containsVariableReference(v) { + return v, nil + } + + jsonBytes, err := json.Marshal(v.AsAny()) + if err != nil { + return dyn.InvalidValue, fmt.Errorf("failed to serialize for_each_task.inputs: %w", err) + } + + return dyn.NewValue(string(jsonBytes), v.Locations()), nil +} + +// containsVariableReference reports whether any string in the value, +// including map keys, holds an unescaped variable reference. Map keys are +// never resolved elsewhere, so a key holding a reference must keep the value +// out of serialization: the marshaled JSON would otherwise expose the key to +// interpolation by later resolution rounds. +func containsVariableReference(v dyn.Value) bool { + found := false + _, _ = dyn.Walk(v, func(_ dyn.Path, n dyn.Value) (dyn.Value, error) { + if s, ok := n.AsString(); ok && dynvar.ContainsVariableReference(s) { + found = true + } + if n.Kind() == dyn.KindMap { + for _, pair := range n.MustMap().Pairs() { + if s, ok := pair.Key.AsString(); ok && dynvar.ContainsVariableReference(s) { + found = true + } + } + } + return n, nil + }) + return found +} diff --git a/bundle/config/mutator/for_each_task_inputs_test.go b/bundle/config/mutator/for_each_task_inputs_test.go new file mode 100644 index 00000000000..70777f9b58c --- /dev/null +++ b/bundle/config/mutator/for_each_task_inputs_test.go @@ -0,0 +1,381 @@ +package mutator + +import ( + "testing" + + "github.com/databricks/cli/bundle" + "github.com/databricks/cli/bundle/config" + "github.com/databricks/cli/bundle/config/resources" + "github.com/databricks/cli/bundle/config/variable" + "github.com/databricks/cli/libs/diag" + "github.com/databricks/cli/libs/dyn" + "github.com/databricks/databricks-sdk-go/service/jobs" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestSerializeInputsValue(t *testing.T) { + obj := func(pairs map[string]dyn.Value) dyn.Value { + return dyn.V(pairs) + } + + tests := []struct { + name string + in dyn.Value + // wantJSON is compared semantically against the resulting string. + // Empty means the value must be returned unchanged. + wantJSON string + }{ + { + name: "sequence of objects is serialized", + in: dyn.V([]dyn.Value{ + obj(map[string]dyn.Value{ + "task_id": dyn.V("load_customers"), + "table": dyn.V("customers"), + }), + }), + wantJSON: `[{"table":"customers","task_id":"load_customers"}]`, + }, + { + name: "map is serialized", + in: obj(map[string]dyn.Value{"table": dyn.V("customers")}), + wantJSON: `{"table":"customers"}`, + }, + { + name: "empty sequence is serialized", + in: dyn.V([]dyn.Value{}), + wantJSON: `[]`, + }, + { + name: "escaping and nested values are serialized", + in: dyn.V([]dyn.Value{ + obj(map[string]dyn.Value{ + "task_id": dyn.V(`load "quoted" \\ ünïcodé`), + "nested": obj(map[string]dyn.Value{ + "list": dyn.V([]dyn.Value{dyn.V(int64(1)), dyn.V(true), dyn.NilValue, dyn.V("x\ny")}), + }), + }), + }), + wantJSON: `[{"nested":{"list":[1,true,null,"x\ny"]},"task_id":"load \"quoted\" \\\\ ünïcodé"}]`, + }, + { + name: "string is left unchanged", + in: dyn.V(`[{"task_id":"load_customers"}]`), + }, + { + name: "int is left unchanged", + in: dyn.V(int64(123)), + }, + { + name: "reference in value is left unchanged", + in: dyn.V([]dyn.Value{ + obj(map[string]dyn.Value{"task_id": dyn.V("${resources.jobs.other.name}")}), + }), + }, + { + name: "reference in map key is left unchanged", + in: dyn.V([]dyn.Value{ + obj(map[string]dyn.Value{"${var.key_name}": dyn.V("payload")}), + }), + }, + { + name: "escaped reference in map key is serialized verbatim", + in: dyn.V([]dyn.Value{ + obj(map[string]dyn.Value{"$${var.key_name}": dyn.V("payload")}), + }), + wantJSON: `[{"$${var.key_name}":"payload"}]`, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := serializeInputsValue(tt.in) + require.NoError(t, err) + if tt.wantJSON == "" { + assert.Equal(t, tt.in.Kind(), got.Kind()) + assert.Equal(t, tt.in.AsAny(), got.AsAny()) + return + } + s, ok := got.AsString() + require.True(t, ok) + assert.JSONEq(t, tt.wantJSON, s) + }) + } +} + +func TestSerializeInputsValuePreservesLocations(t *testing.T) { + locs := []diag.Location{{File: "databricks.yml", Line: 3, Column: 11}} + in := dyn.NewValue([]dyn.Value{dyn.V("a")}, locs) + + got, err := serializeInputsValue(in) + require.NoError(t, err) + assert.Equal(t, locs, got.Locations()) +} + +// forEachTestBundle builds a bundle with a single job whose for_each_task.inputs +// is the given value, plus the given variables. +func forEachTestBundle(inputs string, vars map[string]*variable.Variable) *bundle.Bundle { + return &bundle.Bundle{ + Config: config.Root{ + Variables: vars, + Resources: config.Resources{ + Jobs: map[string]*resources.Job{ + "my_job": { + JobSettings: jobs.JobSettings{ + Tasks: []jobs.Task{ + { + TaskKey: "process_tables", + ForEachTask: &jobs.ForEachTask{ + Inputs: inputs, + Task: jobs.Task{ + TaskKey: "load_table", + }, + }, + }, + }, + }, + }, + }, + }, + }, + } +} + +func forEachInputs(t *testing.T, b *bundle.Bundle) string { + t.Helper() + tasks := b.Config.Resources.Jobs["my_job"].Tasks + require.Len(t, tasks, 1) + require.NotNil(t, tasks[0].ForEachTask) + return tasks[0].ForEachTask.Inputs +} + +func diagSummaries(diags diag.Diagnostics) []string { + out := make([]string, 0, len(diags)) + for _, d := range diags { + out = append(out, d.Summary) + } + return out +} + +func TestSerializeForEachTaskInputsPipeline(t *testing.T) { + complexValue := []any{ + map[string]any{"task_id": "load_customers", "table": "customers", "source_type": "parquet"}, + map[string]any{"task_id": "load_orders", "table": "orders", "source_type": "json"}, + } + complexJSON := `[{"source_type":"parquet","table":"customers","task_id":"load_customers"},{"source_type":"json","table":"orders","task_id":"load_orders"}]` + + tests := []struct { + name string + inputs string + vars map[string]*variable.Variable + // want is the resulting inputs value; wantJSON compares semantically + // instead when set. warn expects the type-mismatch warning (and a + // dropped inputs), errContains expects a resolution error. + want string + wantJSON string + warn bool + errContains string + }{ + { + name: "complex variable array of objects is serialized", + inputs: "${var.table_list}", + vars: map[string]*variable.Variable{ + "table_list": {Type: variable.VariableTypeComplex, Value: complexValue}, + }, + wantJSON: complexJSON, + }, + { + name: "long variable reference form is serialized", + inputs: "${variables.table_list.value}", + vars: map[string]*variable.Variable{ + "table_list": {Type: variable.VariableTypeComplex, Value: complexValue}, + }, + wantJSON: complexJSON, + }, + { + name: "subpath into complex variable is serialized", + inputs: "${var.wrapped.items}", + vars: map[string]*variable.Variable{ + "wrapped": {Type: variable.VariableTypeComplex, Value: map[string]any{"items": complexValue}}, + }, + wantJSON: complexJSON, + }, + { + name: "existing JSON string keeps working", + inputs: `[{"task_id":"load_customers"}]`, + want: `[{"task_id":"load_customers"}]`, + }, + { + name: "scalar variable resolves as before", + inputs: "${var.table_name}", + vars: map[string]*variable.Variable{ + "table_name": {Value: "customers"}, + }, + want: "customers", + }, + { + name: "interpolated complex reference still errors", + inputs: `prefix-${var.table_list}`, + vars: map[string]*variable.Variable{ + "table_list": {Type: variable.VariableTypeComplex, Value: complexValue}, + }, + errContains: "cannot interpolate non-primitive value", + }, + { + name: "missing variable still errors", + inputs: "${var.does_not_exist}", + errContains: "reference does not exist", + }, + { + // Complex values containing resource references are intentionally + // not serialized (see serializeForEachTaskInputs): normalization + // reports them exactly as it did before this feature. + name: "complex value with nested resource reference keeps old diagnostics", + inputs: "${var.table_list}", + vars: map[string]*variable.Variable{ + "table_list": {Type: variable.VariableTypeComplex, Value: []any{ + map[string]any{"task_id": "${resources.jobs.other.name}"}, + }}, + }, + warn: true, + }, + { + // Map keys are never resolved anywhere in bundle configuration. + // The missing key_name variable must therefore not produce a + // resolution error; serialization is skipped and normalization + // warns as before. + name: "reference in map key is not interpolated", + inputs: "${var.table_list}", + vars: map[string]*variable.Variable{ + "table_list": {Type: variable.VariableTypeComplex, Value: []any{ + map[string]any{"${var.key_name}": "payload"}, + }}, + }, + warn: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + b := forEachTestBundle(tt.inputs, tt.vars) + + diags := bundle.Apply(t.Context(), b, ResolveVariableReferencesOnlyResources()) + + if tt.errContains != "" { + require.Error(t, diags.Error()) + assert.ErrorContains(t, diags.Error(), tt.errContains) + return + } + require.NoError(t, diags.Error()) + + summaries := diagSummaries(diags) + if tt.warn { + assert.Contains(t, summaries, "expected string, found sequence") + assert.NotContains(t, summaries, "reference does not exist: ${var.key_name}") + assert.Empty(t, forEachInputs(t, b)) + return + } + + assert.NotContains(t, summaries, "expected string, found sequence") + got := forEachInputs(t, b) + if tt.wantJSON != "" { + assert.JSONEq(t, tt.wantJSON, got) + return + } + assert.Equal(t, tt.want, got) + }) + } +} + +func TestSerializeForEachTaskInputsNestedForEach(t *testing.T) { + b := forEachTestBundle("${var.outer}", map[string]*variable.Variable{ + "outer": {Type: variable.VariableTypeComplex, Value: []any{map[string]any{"a": "b"}}}, + "inner": {Type: variable.VariableTypeComplex, Value: []any{map[string]any{"c": "d"}}}, + }) + // Nest a second for_each_task inside the nested task. + outer := &b.Config.Resources.Jobs["my_job"].Tasks[0] + outer.ForEachTask.Task = jobs.Task{ + TaskKey: "middle", + ForEachTask: &jobs.ForEachTask{Inputs: "${var.inner}"}, + } + + diags := bundle.Apply(t.Context(), b, ResolveVariableReferencesOnlyResources()) + require.NoError(t, diags.Error()) + + assert.JSONEq(t, `[{"a":"b"}]`, forEachInputs(t, b)) + assert.JSONEq(t, `[{"c":"d"}]`, b.Config.Resources.Jobs["my_job"].Tasks[0].ForEachTask.Task.ForEachTask.Inputs) +} + +// A for_each_task.inputs reference nested inside a variable-defined task list +// (tasks: ${var.job_tasks}) only materializes once the enclosing reference is +// substituted; serialization must still happen for it. +func TestSerializeForEachTaskInputsVariableDefinedTasks(t *testing.T) { + root, diags := config.LoadFromBytes("databricks.yml", []byte(` +bundle: + name: variable-defined-tasks + +variables: + table_list: + type: complex + default: + - task_id: load_customers + table: customers + source_type: parquet + - task_id: load_orders + table: orders + source_type: json + job_tasks: + type: complex + default: + - task_key: process_tables + for_each_task: + inputs: ${var.table_list} + concurrency: 2 + task: + task_key: load_table + +resources: + jobs: + my_job: + tasks: ${var.job_tasks} +`)) + require.NoError(t, diags.Error()) + + b := &bundle.Bundle{Config: *root} + diags = bundle.ApplySeq(t.Context(), b, + SetVariables(), + ResolveVariableReferencesOnlyResources(), + ) + require.NoError(t, diags.Error()) + assert.NotContains(t, diagSummaries(diags), "expected string, found sequence") + assert.JSONEq(t, + `[{"source_type":"parquet","table":"customers","task_id":"load_customers"},{"source_type":"json","table":"orders","task_id":"load_orders"}]`, + forEachInputs(t, b)) +} + +// Complex variables substituted into unrelated string fields must keep failing +// validation as before; the serialization only applies to for_each_task.inputs. +func TestSerializeForEachTaskInputsPreservesUnrelatedValidation(t *testing.T) { + b := &bundle.Bundle{ + Config: config.Root{ + Variables: map[string]*variable.Variable{ + "table_list": {Type: variable.VariableTypeComplex, Value: []any{ + map[string]any{"task_id": "load_customers"}, + }}, + }, + Resources: config.Resources{ + Jobs: map[string]*resources.Job{ + "my_job": { + JobSettings: jobs.JobSettings{ + Name: "${var.table_list}", + }, + }, + }, + }, + }, + } + + diags := bundle.Apply(t.Context(), b, ResolveVariableReferencesOnlyResources()) + require.NoError(t, diags.Error()) + assert.Contains(t, diagSummaries(diags), "expected string, found sequence") +} diff --git a/bundle/config/mutator/resolve_variable_references.go b/bundle/config/mutator/resolve_variable_references.go index 257309c55f8..7c68826a781 100644 --- a/bundle/config/mutator/resolve_variable_references.go +++ b/bundle/config/mutator/resolve_variable_references.go @@ -265,6 +265,14 @@ func (m *resolveVariableReferences) resolveOnce(b *bundle.Bundle, prefixes []dyn return dyn.InvalidValue, err } + // Serialize complex values substituted into for_each_task.inputs before + // normalization: the field is a JSON string, so a substituted sequence + // or map would otherwise be dropped as a type mismatch. + root, err = serializeForEachTaskInputs(root) + if err != nil { + return dyn.InvalidValue, err + } + // Normalize the result because variable resolution may have been applied to non-string fields. // For example, a variable reference may have been resolved to a integer. root, normaliseDiags := convert.Normalize(b.Config, root) diff --git a/bundle/config/mutator/resourcemutator/process_static_resources.go b/bundle/config/mutator/resourcemutator/process_static_resources.go index 015afa662f1..b1f689808f6 100644 --- a/bundle/config/mutator/resourcemutator/process_static_resources.go +++ b/bundle/config/mutator/resourcemutator/process_static_resources.go @@ -42,12 +42,6 @@ func (p processStaticResources) Apply(ctx context.Context, b *bundle.Bundle) dia bundle.ApplySeqContext( ctx, b, - // Reads (dynamic): resources.jobs.*.tasks for_each_task.inputs, variables.*.value - // Updates (dynamic): resources.jobs.*.tasks for_each_task.inputs (complex variable reference -> JSON string) - // Must run before variable resolution: a substituted sequence would be - // dropped by normalization with an "expected string" warning. - SerializeForEachTaskInputs(), - // Reads (dynamic): * (strings) (searches for variable references in string values) // Updates (dynamic): resources.* (strings) (resolves variable references to their actual values) // Resolves variable references in 'resources' using bundle, workspace, and variables prefixes diff --git a/bundle/config/mutator/resourcemutator/resource_mutator.go b/bundle/config/mutator/resourcemutator/resource_mutator.go index 1f3cdc25de5..4d0b3bd47ab 100644 --- a/bundle/config/mutator/resourcemutator/resource_mutator.go +++ b/bundle/config/mutator/resourcemutator/resource_mutator.go @@ -143,12 +143,6 @@ func applyNormalizeMutators(ctx context.Context, b *bundle.Bundle) { validate.SingleNodeCluster(), - // Reads (dynamic): resources.jobs.*.tasks for_each_task.inputs, variables.*.value - // Updates (dynamic): resources.jobs.*.tasks for_each_task.inputs (complex variable reference -> JSON string) - // Must run before variable resolution: a substituted sequence would be - // dropped by normalization with an "expected string" warning. - SerializeForEachTaskInputs(), - // Reads (dynamic): * (strings) (searches for variable references in string values) // Updates (dynamic): resources.* (strings) (resolves variable references to their actual values) // Resolves variable references in 'resources' using bundle, workspace, and variables prefixes diff --git a/bundle/config/mutator/resourcemutator/serialize_for_each_task_inputs.go b/bundle/config/mutator/resourcemutator/serialize_for_each_task_inputs.go deleted file mode 100644 index 0ef76181037..00000000000 --- a/bundle/config/mutator/resourcemutator/serialize_for_each_task_inputs.go +++ /dev/null @@ -1,145 +0,0 @@ -package resourcemutator - -import ( - "context" - "encoding/json" - "fmt" - - "github.com/databricks/cli/bundle" - "github.com/databricks/cli/libs/diag" - "github.com/databricks/cli/libs/dyn" - "github.com/databricks/cli/libs/dyn/dynvar" -) - -// Suffix matching for_each_task.inputs at any nesting depth within a job. -var forEachTaskInputsSuffix = dyn.NewPath(dyn.Key("for_each_task"), dyn.Key("inputs")) - -var varShorthandPrefix = dyn.NewPath(dyn.Key("var")) - -type serializeForEachTaskInputs struct{} - -// SerializeForEachTaskInputs converts a complex variable referenced by -// for_each_task.inputs into the JSON string the Jobs API expects. -// -// The inputs field is typed as a string, so a substituted sequence or map -// would be dropped by normalization with an "expected string, found sequence" -// warning. Serializing here, before variable resolution runs, keeps the value -// intact. Anything that is not a pure reference to a complex variable value -// is left untouched for the regular resolution and validation to handle. -func SerializeForEachTaskInputs() bundle.Mutator { - return &serializeForEachTaskInputs{} -} - -func (s serializeForEachTaskInputs) Name() string { - return "SerializeForEachTaskInputs" -} - -func (s serializeForEachTaskInputs) Apply(_ context.Context, b *bundle.Bundle) diag.Diagnostics { - var diags diag.Diagnostics - - pattern := dyn.NewPattern( - dyn.Key("resources"), - dyn.Key("jobs"), - dyn.AnyKey(), - ) - - err := b.Config.Mutate(func(root dyn.Value) (dyn.Value, error) { - return dyn.MapByPattern(root, pattern, func(_ dyn.Path, job dyn.Value) (dyn.Value, error) { - return dyn.Walk(job, func(p dyn.Path, v dyn.Value) (dyn.Value, error) { - if !p.HasSuffix(forEachTaskInputsSuffix) { - return v, nil - } - return serializeInputsValue(root, v) - }) - }) - }) - - diags = diags.Extend(diag.FromErr(err)) - return diags -} - -func serializeInputsValue(root, v dyn.Value) (dyn.Value, error) { - s, ok := v.AsString() - if !ok { - return v, nil - } - - path, ok := dynvar.PureReferenceToPath(s) - if !ok { - return v, nil - } - - path, ok = variableValuePath(path) - if !ok { - return v, nil - } - - target, ok := lookupVariableValue(root, path) - if !ok { - return v, nil - } - - switch target.Kind() { - case dyn.KindSequence, dyn.KindMap: - default: - return v, nil - } - - // Only serialize fully resolved values. Baking an unresolved reference - // into JSON would change its meaning from structural substitution to - // string interpolation when a later resolution round picks it up. - if containsVariableReference(target) { - return v, nil - } - - jsonBytes, err := json.Marshal(target.AsAny()) - if err != nil { - return dyn.InvalidValue, fmt.Errorf("failed to serialize for_each_task.inputs: %w", err) - } - - return dyn.NewValue(string(jsonBytes), v.Locations()), nil -} - -// variableValuePath maps a pure reference to the variable value it points to. -// ${var.foo} is shorthand for ${variables.foo.value}; longer paths address -// into the value itself. References to anything else are not variable values. -func variableValuePath(path dyn.Path) (dyn.Path, bool) { - if path.HasPrefix(varShorthandPrefix) { - if len(path) < 2 { - return nil, false - } - out := dyn.NewPath(dyn.Key("variables"), path[1], dyn.Key("value")) - if len(path) > 2 { - out = out.Append(path[2:]...) - } - return out, true - } - - if len(path) > 0 && path[0].Key() == "variables" { - return path, true - } - - return nil, false -} - -// lookupVariableValue resolves a variable value path against the current -// configuration. A missing reference is not an error here; the regular -// variable resolution reports it with suggestions. -func lookupVariableValue(root dyn.Value, path dyn.Path) (dyn.Value, bool) { - target, err := dyn.GetByPath(root, path) - if err != nil || !target.IsValid() { - return dyn.InvalidValue, false - } - return target, true -} - -func containsVariableReference(v dyn.Value) bool { - found := false - _ = dyn.WalkReadOnly(v, func(_ dyn.Path, v dyn.Value) error { - if s, ok := v.AsString(); ok && dynvar.ContainsVariableReference(s) { - found = true - } - return nil - }) - return found -} diff --git a/bundle/config/mutator/resourcemutator/serialize_for_each_task_inputs_test.go b/bundle/config/mutator/resourcemutator/serialize_for_each_task_inputs_test.go deleted file mode 100644 index 3d760c38b57..00000000000 --- a/bundle/config/mutator/resourcemutator/serialize_for_each_task_inputs_test.go +++ /dev/null @@ -1,227 +0,0 @@ -package resourcemutator_test - -import ( - "testing" - - "github.com/databricks/cli/bundle" - "github.com/databricks/cli/bundle/config" - "github.com/databricks/cli/bundle/config/mutator" - "github.com/databricks/cli/bundle/config/mutator/resourcemutator" - "github.com/databricks/cli/bundle/config/resources" - "github.com/databricks/cli/bundle/config/variable" - "github.com/databricks/databricks-sdk-go/service/jobs" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -// forEachTestBundle builds a bundle with a single job whose for_each_task.inputs -// is the given value, plus the given variables. -func forEachTestBundle(inputs string, vars map[string]*variable.Variable) *bundle.Bundle { - return &bundle.Bundle{ - Config: config.Root{ - Variables: vars, - Resources: config.Resources{ - Jobs: map[string]*resources.Job{ - "my_job": { - JobSettings: jobs.JobSettings{ - Tasks: []jobs.Task{ - { - TaskKey: "process_tables", - ForEachTask: &jobs.ForEachTask{ - Inputs: inputs, - Task: jobs.Task{ - TaskKey: "load_table", - }, - }, - }, - }, - }, - }, - }, - }, - }, - } -} - -func forEachInputs(t *testing.T, b *bundle.Bundle) string { - t.Helper() - tasks := b.Config.Resources.Jobs["my_job"].Tasks - require.Len(t, tasks, 1) - require.NotNil(t, tasks[0].ForEachTask) - return tasks[0].ForEachTask.Inputs -} - -func TestSerializeForEachTaskInputs(t *testing.T) { - complexValue := []any{ - map[string]any{"task_id": "load_customers", "table": "customers", "source_type": "parquet"}, - map[string]any{"task_id": "load_orders", "table": "orders", "source_type": "json"}, - } - - tests := []struct { - name string - inputs string - vars map[string]*variable.Variable - // wantJSON is compared semantically against the resulting inputs string. - // Empty means inputs must be left unchanged. - wantJSON string - }{ - { - name: "complex variable array of objects is serialized", - inputs: "${var.table_list}", - vars: map[string]*variable.Variable{ - "table_list": {Type: variable.VariableTypeComplex, Value: complexValue}, - }, - wantJSON: `[{"source_type":"parquet","table":"customers","task_id":"load_customers"},{"source_type":"json","table":"orders","task_id":"load_orders"}]`, - }, - { - name: "long variable reference form is serialized", - inputs: "${variables.table_list.value}", - vars: map[string]*variable.Variable{ - "table_list": {Type: variable.VariableTypeComplex, Value: complexValue}, - }, - wantJSON: `[{"source_type":"parquet","table":"customers","task_id":"load_customers"},{"source_type":"json","table":"orders","task_id":"load_orders"}]`, - }, - { - name: "empty array is serialized", - inputs: "${var.table_list}", - vars: map[string]*variable.Variable{ - "table_list": {Type: variable.VariableTypeComplex, Value: []any{}}, - }, - wantJSON: `[]`, - }, - { - name: "escaping and nested values are serialized", - inputs: "${var.table_list}", - vars: map[string]*variable.Variable{ - "table_list": {Type: variable.VariableTypeComplex, Value: []any{ - map[string]any{ - "task_id": `load "quoted" \\ ünïcodé`, - "nested": map[string]any{"list": []any{float64(1), true, nil, "x\ny"}}, - }, - }}, - }, - wantJSON: `[{"nested":{"list":[1,true,null,"x\ny"]},"task_id":"load \"quoted\" \\\\ ünïcodé"}]`, - }, - { - name: "existing JSON string is left unchanged", - inputs: `[{"task_id":"load_customers"}]`, - vars: map[string]*variable.Variable{ - "table_list": {Type: variable.VariableTypeComplex, Value: complexValue}, - }, - }, - { - name: "scalar variable reference is left for normal resolution", - inputs: "${var.table_name}", - vars: map[string]*variable.Variable{ - "table_name": {Value: "customers"}, - }, - }, - { - name: "non-variable reference is left unchanged", - inputs: "${bundle.name}", - vars: map[string]*variable.Variable{ - "table_list": {Type: variable.VariableTypeComplex, Value: complexValue}, - }, - }, - { - name: "interpolated reference is left unchanged", - inputs: `prefix-${var.table_list}`, - vars: map[string]*variable.Variable{ - "table_list": {Type: variable.VariableTypeComplex, Value: complexValue}, - }, - }, - { - name: "missing variable is left for normal resolution", - inputs: "${var.does_not_exist}", - vars: map[string]*variable.Variable{ - "table_list": {Type: variable.VariableTypeComplex, Value: complexValue}, - }, - }, - { - name: "complex value with nested reference is left unchanged", - inputs: "${var.table_list}", - vars: map[string]*variable.Variable{ - "table_list": {Type: variable.VariableTypeComplex, Value: []any{ - map[string]any{"task_id": "${resources.jobs.other.name}"}, - }}, - }, - }, - { - name: "subpath into complex variable is serialized", - inputs: "${var.wrapped.items}", - vars: map[string]*variable.Variable{ - "wrapped": {Type: variable.VariableTypeComplex, Value: map[string]any{"items": complexValue}}, - }, - wantJSON: `[{"source_type":"parquet","table":"customers","task_id":"load_customers"},{"source_type":"json","table":"orders","task_id":"load_orders"}]`, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - b := forEachTestBundle(tt.inputs, tt.vars) - - diags := bundle.Apply(t.Context(), b, resourcemutator.SerializeForEachTaskInputs()) - require.NoError(t, diags.Error()) - - got := forEachInputs(t, b) - if tt.wantJSON == "" { - assert.Equal(t, tt.inputs, got) - return - } - assert.JSONEq(t, tt.wantJSON, got) - }) - } -} - -func TestSerializeForEachTaskInputsNestedForEach(t *testing.T) { - b := forEachTestBundle("${var.outer}", map[string]*variable.Variable{ - "outer": {Type: variable.VariableTypeComplex, Value: []any{map[string]any{"a": "b"}}}, - "inner": {Type: variable.VariableTypeComplex, Value: []any{map[string]any{"c": "d"}}}, - }) - // Nest a second for_each_task inside the nested task. - outer := &b.Config.Resources.Jobs["my_job"].Tasks[0] - outer.ForEachTask.Task = jobs.Task{ - TaskKey: "middle", - ForEachTask: &jobs.ForEachTask{Inputs: "${var.inner}"}, - } - - diags := bundle.Apply(t.Context(), b, resourcemutator.SerializeForEachTaskInputs()) - require.NoError(t, diags.Error()) - - assert.JSONEq(t, `[{"a":"b"}]`, forEachInputs(t, b)) - assert.JSONEq(t, `[{"c":"d"}]`, b.Config.Resources.Jobs["my_job"].Tasks[0].ForEachTask.Task.ForEachTask.Inputs) -} - -// Complex variables substituted into unrelated string fields must keep failing -// validation as before; the serialization only applies to for_each_task.inputs. -func TestSerializeForEachTaskInputsPreservesUnrelatedValidation(t *testing.T) { - b := &bundle.Bundle{ - Config: config.Root{ - Variables: map[string]*variable.Variable{ - "table_list": {Type: variable.VariableTypeComplex, Value: []any{ - map[string]any{"task_id": "load_customers"}, - }}, - }, - Resources: config.Resources{ - Jobs: map[string]*resources.Job{ - "my_job": { - JobSettings: jobs.JobSettings{ - Name: "${var.table_list}", - }, - }, - }, - }, - }, - } - - diags := bundle.Apply(t.Context(), b, resourcemutator.SerializeForEachTaskInputs()) - require.NoError(t, diags.Error()) - assert.Equal(t, "${var.table_list}", b.Config.Resources.Jobs["my_job"].Name) - - diags = bundle.Apply(t.Context(), b, mutator.ResolveVariableReferencesOnlyResources()) - summaries := make([]string, 0, len(diags)) - for _, d := range diags { - summaries = append(summaries, d.Summary) - } - assert.Contains(t, summaries, "expected string, found sequence") -} From ccaf63f4a14546f65ad4f82dd123df2129265fda Mon Sep 17 00:00:00 2001 From: Yuri Barreira Date: Mon, 19 Oct 2026 17:07:06 -0300 Subject: [PATCH 4/4] Use imperative voice in for_each_task.inputs changelog --- .nextchanges/bundles/6901.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.nextchanges/bundles/6901.md b/.nextchanges/bundles/6901.md index df39f027534..a18fb3a517a 100644 --- a/.nextchanges/bundles/6901.md +++ b/.nextchanges/bundles/6901.md @@ -1 +1 @@ -* Fixed `type: complex` variables failing with `expected string, found sequence` when referenced from `for_each_task.inputs`; the CLI now serializes the referenced value to the JSON string the Jobs API expects (see [#6901](https://github.com/databricks/cli/issues/6901)). ([#7067](https://github.com/databricks/cli/pull/7067)) +* Fix `type: complex` variables failing with `expected string, found sequence` when referenced from `for_each_task.inputs`; the CLI now serializes the referenced value to the JSON string the Jobs API expects (see [#6901](https://github.com/databricks/cli/issues/6901)). ([#7067](https://github.com/databricks/cli/pull/7067))