Treat spark_env_vars as a backend default in config-remote-sync

acceptance/bundle/config-remote-sync/spark_env_vars_backend_default/databricks.yml.tmpl

@@ -0,0 +1,28 @@
+bundle:
+  name: test-bundle-$UNIQUE_NAME
+
+targets:
+  default:
+    mode: development
+
+resources:
+  jobs:
+    my_job:
+      max_concurrent_runs: 1
+      tasks:
+        - task_key: with_env
+          notebook_task:
+            notebook_path: /Users/{{workspace_user_name}}/notebook
+          new_cluster:
+            spark_version: $DEFAULT_SPARK_VERSION
+            node_type_id: $NODE_TYPE_ID
+            num_workers: 1
+            spark_env_vars:
+              PYSPARK_PYTHON: /databricks/python3/bin/python3
+        - task_key: without_env
+          notebook_task:
+            notebook_path: /Users/{{workspace_user_name}}/notebook
+          new_cluster:
+            spark_version: $DEFAULT_SPARK_VERSION
+            node_type_id: $NODE_TYPE_ID
+            num_workers: 1

acceptance/bundle/config-remote-sync/spark_env_vars_backend_default/output.txt

@@ -0,0 +1,45 @@
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle-[UNIQUE_NAME]/default/files...
+Created jobs.my_job
+Files: 5 uploaded, 0 deleted
+Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
+
+=== Simulate a cluster policy injecting spark_env_vars
+spark_env_vars the user did not set are backend defaults and must not be synced back.
+A change to a user-declared key and an unrelated edit still sync.
+
+=== Detect and save all changes
+Detected changes in 1 resource(s):
+
+Resource: resources.jobs.my_job
+  max_concurrent_runs: replace
+  tasks[task_key='with_env'].new_cluster.spark_env_vars['PYSPARK_PYTHON']: replace
+
+
+
+=== Configuration changes
+
+>>> diff.py databricks.yml.backup databricks.yml
+--- databricks.yml.backup
++++ databricks.yml
+@@ -9,5 +9,5 @@
+   jobs:
+     my_job:
+-      max_concurrent_runs: 1
++      max_concurrent_runs: 5
+       tasks:
+         - task_key: with_env
+@@ -19,5 +19,5 @@
+             num_workers: 1
+             spark_env_vars:
+-              PYSPARK_PYTHON: /databricks/python3/bin/python3
++              PYSPARK_PYTHON: /databricks/python3/bin/python3.11
+         - task_key: without_env
+           notebook_task:
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.jobs.my_job
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/test-bundle-[UNIQUE_NAME]/default
+
+Destroy: 1 deleted

acceptance/bundle/config-remote-sync/spark_env_vars_backend_default/script

@@ -0,0 +1,37 @@
+#!/bin/bash
+
+envsubst < databricks.yml.tmpl > databricks.yml
+
+cleanup() {
+  trace $CLI bundle destroy --auto-approve
+}
+trap cleanup EXIT
+
+$CLI bundle deploy
+job_id="$(read_id.py my_job)"
+
+title "Simulate a cluster policy injecting spark_env_vars"
+echo
+echo "spark_env_vars the user did not set are backend defaults and must not be synced back."
+echo "A change to a user-declared key and an unrelated edit still sync."
+edit_resource.py jobs $job_id <<'EOF'
+r["max_concurrent_runs"] = 5
+
+for task in r.get("tasks", []):
+    cluster = task["new_cluster"]
+    if "spark_env_vars" in cluster:
+        cluster["spark_env_vars"]["PYSPARK_PYTHON"] = "/databricks/python3/bin/python3.11"
+        cluster["spark_env_vars"]["INJECTED_VAR"] = "backend-value"
+    else:
+        cluster["spark_env_vars"] = {"INJECTED_VAR": "backend-value"}
+EOF
+
+title "Detect and save all changes"
+echo
+cp databricks.yml databricks.yml.backup
+$CLI bundle config-remote-sync --save
+
+title "Configuration changes"
+echo
+trace diff.py databricks.yml.backup databricks.yml
+rm databricks.yml.backup

acceptance/bundle/config-remote-sync/spark_env_vars_backend_default/test.toml

@@ -0,0 +1,8 @@
+RecordRequests = false
+Ignore = [".databricks", "databricks.yml", "databricks.yml.backup"]
+
+[Env]
+DATABRICKS_BUNDLE_ENABLE_EXPERIMENTAL_YAML_SYNC = "true"
+
+[EnvMatrix]
+DATABRICKS_BUNDLE_ENGINE = ["direct", "terraform"]

bundle/configsync/defaults.go

@@ -60,13 +60,17 @@ var serverSideDefaults = map[string]any{
 	"resources.jobs.*.tasks[*].new_cluster.data_security_mode":  "SINGLE_USER", // TODO this field is computed on some workspaces in integration tests, check why and if we can skip it
 	"resources.jobs.*.tasks[*].new_cluster.enable_elastic_disk": alwaysSkip,    // deprecated field
 	"resources.jobs.*.tasks[*].new_cluster.single_user_name":    alwaysSkip,
-	// custom_tags and cluster_log_conf are commonly injected by cluster policies
-	// when the user omits them, so they exist only remotely. Syncing them back leaks
-	// one environment's policy values into (often shared) config and breaks deploys in
-	// other environments. TODO: move to backend_defaults in resources.yml once
-	// configsync filtering is migrated to the direct engine lifecycle metadata.
-	"resources.jobs.*.tasks[*].new_cluster.custom_tags":      backendDefault,
-	"resources.jobs.*.tasks[*].new_cluster.cluster_log_conf": backendDefault,
+	// custom_tags, cluster_log_conf and spark_env_vars are commonly injected by
+	// cluster policies when the user omits them, so they exist only remotely.
+	// Syncing them back leaks one environment's policy values into (often shared)
+	// config and breaks deploys in other environments; spark_env_vars may also carry
+	// credentials. The "[*]" variants cover per-key injection into an existing block.
+	// TODO: move to backend_defaults in resources.yml once configsync filtering is
+	// migrated to the direct engine lifecycle metadata.
+	"resources.jobs.*.tasks[*].new_cluster.custom_tags":       backendDefault,
+	"resources.jobs.*.tasks[*].new_cluster.cluster_log_conf":  backendDefault,
+	"resources.jobs.*.tasks[*].new_cluster.spark_env_vars":    backendDefault,
+	"resources.jobs.*.tasks[*].new_cluster.spark_env_vars[*]": backendDefault,
 
 	// Cluster fields (job_clusters)
 	"resources.jobs.*.job_clusters[*].new_cluster.aws_attributes":      alwaysSkip,
@@ -77,6 +81,8 @@ var serverSideDefaults = map[string]any{
 	"resources.jobs.*.job_clusters[*].new_cluster.single_user_name":    alwaysSkip,
 	"resources.jobs.*.job_clusters[*].new_cluster.custom_tags":         backendDefault, // see tasks[*].new_cluster.custom_tags
 	"resources.jobs.*.job_clusters[*].new_cluster.cluster_log_conf":    backendDefault, // see tasks[*].new_cluster.cluster_log_conf
+	"resources.jobs.*.job_clusters[*].new_cluster.spark_env_vars":      backendDefault, // see tasks[*].new_cluster.spark_env_vars
+	"resources.jobs.*.job_clusters[*].new_cluster.spark_env_vars[*]":   backendDefault, // see tasks[*].new_cluster.spark_env_vars
 
 	// Standalone cluster fields
 	"resources.clusters.*.aws_attributes":      alwaysSkip,
@@ -88,6 +94,8 @@ var serverSideDefaults = map[string]any{
 	"resources.clusters.*.single_user_name":    alwaysSkip,
 	"resources.clusters.*.custom_tags":         backendDefault, // see jobs.*.tasks[*].new_cluster.custom_tags
 	"resources.clusters.*.cluster_log_conf":    backendDefault, // see jobs.*.tasks[*].new_cluster.cluster_log_conf
+	"resources.clusters.*.spark_env_vars":      backendDefault, // see jobs.*.tasks[*].new_cluster.spark_env_vars
+	"resources.clusters.*.spark_env_vars[*]":   backendDefault, // see jobs.*.tasks[*].new_cluster.spark_env_vars
 
 	// Experiment fields
 	"resources.experiments.*.artifact_location": alwaysSkip,