Serialize complex variables used in for_each_task.inputs

.nextchanges/bundles/6901.md

@@ -0,0 +1 @@
+* 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))

acceptance/bundle/variables/complex-foreach-inputs/databricks.yml

@@ -0,0 +1,55 @@
+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
+  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:
+    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
+    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}

acceptance/bundle/variables/complex-foreach-inputs/hello.py

@@ -0,0 +1 @@
+print("hello")

acceptance/bundle/variables/complex-foreach-inputs/out.test.toml

@@ -0,0 +1,2 @@
+Cloud = false
+EnvMatrix.DMS = ["", "true"]

acceptance/bundle/variables/complex-foreach-inputs/out.validate.json

@@ -0,0 +1,172 @@
+{
+  "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"
+          }
+        ]
+      },
+      "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": [
+        {
+          "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"
+        }
+      ]
+    }
+  }
+}

acceptance/bundle/variables/complex-foreach-inputs/output.txt

@@ -0,0 +1,8 @@
+
+>>> [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"}]
+
+>>> 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"}]

acceptance/bundle/variables/complex-foreach-inputs/script

@@ -0,0 +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"}]'

acceptance/bundle/variables/complex-foreach-inputs/test.toml

@@ -0,0 +1 @@
+Ignore = [".databricks"]

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
+}