Used KeyedSllice for job fields in bundle plan

acceptance/bundle/artifacts/whl_dynamic/out.plan_update.direct.json

@@ -159,7 +159,7 @@
           "reason": "empty",
           "remote": {}
         },
-        "environments[0].spec.dependencies[0]": {
+        "environments[environment_key='test_env'].spec.dependencies[0]": {
           "action": "update",
           "old": "/Workspace/Users/[USERNAME]/.bundle/test-bundle/default/artifacts/.internal/my_test_code-0.0.1+[UNIX_TIME_NANOS][2]-py3-none-any.whl",
           "new": "/Workspace/Users/[USERNAME]/.bundle/test-bundle/default/artifacts/.internal/my_test_code-0.0.1+[UNIX_TIME_NANOS][0]-py3-none-any.whl",

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

@@ -12,8 +12,9 @@ Detected changes in 1 resource(s):
 Resource: resources.jobs.my_job
   email_notifications.no_alert_for_skipped_runs: add
   email_notifications.on_failure: add
-  job_clusters[0].job_cluster_key: replace
-  parameters: replace
+  job_clusters[job_cluster_key='test_cluster']: remove
+  job_clusters[job_cluster_key='test_cluster_renamed']: add
+  parameters[name='region']: add
   tags['team']: add
   tasks[task_key='inline_cluster_task'].new_cluster.num_workers: replace
   tasks[task_key='main'].job_cluster_key: replace
@@ -28,21 +29,17 @@ Resource: resources.jobs.my_job
 >>> diff.py databricks.yml.backup databricks.yml
 --- databricks.yml.backup
 +++ databricks.yml
-@@ -8,13 +8,19 @@
+@@ -8,4 +8,7 @@
          on_success:
            - success@example.com
 +        no_alert_for_skipped_runs: true
 +        on_failure:
 +          - failure@example.com
        parameters:
--        - name: catalog
--          default: main
--        - name: env
--          default: dev
-+        - default: main
-+          name: catalog
-+        - default: dev
-+          name: env
+         - name: catalog
+@@ -13,8 +16,11 @@
+         - name: env
+           default: dev
 +        - default: us-east-1
 +          name: region
        trigger:
@@ -55,14 +52,18 @@ Resource: resources.jobs.my_job
 +            - samples.nyctaxi.trips
        environments:
          - environment_key: default
-@@ -24,5 +30,5 @@
+@@ -24,14 +30,14 @@
                - ./*.whl
        job_clusters:
 -        - job_cluster_key: test_cluster
 +        - job_cluster_key: test_cluster_renamed
            new_cluster:
-             spark_version: [[DEFAULT_SPARK_VERSION]]
-@@ -33,5 +39,5 @@
+-            spark_version: [[DEFAULT_SPARK_VERSION]]
+             node_type_id: [NODE_TYPE_ID]
+             num_workers: 1
++            spark_version: [[DEFAULT_SPARK_VERSION]]
+       tasks:
+         - task_key: main
            notebook_task:
              notebook_path: /Users/{{workspace_user_name}}/notebook
 -          job_cluster_key: test_cluster

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

@@ -0,0 +1,25 @@
+bundle:
+  name: test-bundle-$UNIQUE_NAME
+
+variables:
+  catalog:
+    default: main
+  schema:
+    default: dev
+
+resources:
+  jobs:
+    my_job:
+      parameters:
+        - name: catalog
+          default: ${var.catalog}
+        - name: schema
+          default: ${var.schema}
+      tasks:
+        - task_key: main
+          notebook_task:
+            notebook_path: /Users/{{workspace_user_name}}/notebook
+
+targets:
+  default:
+    mode: development

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

@@ -0,0 +1,35 @@
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle-[UNIQUE_NAME]/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+=== Add new parameter remotely
+=== Detect and save changes
+Detected changes in 1 resource(s):
+
+Resource: resources.jobs.my_job
+  parameters[name='region']: add
+
+
+
+=== Configuration changes
+
+>>> diff.py databricks.yml.backup databricks.yml
+--- databricks.yml.backup
++++ databricks.yml
+@@ -16,4 +16,6 @@
+         - name: schema
+           default: ${var.schema}
++        - default: us-east-1
++          name: region
+       tasks:
+         - task_key: main
+
+>>> [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
+
+Deleting files...
+Destroy complete!

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

@@ -0,0 +1,26 @@
+#!/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 "Add new parameter remotely"
+edit_resource.py jobs $job_id <<EOF
+r["parameters"].append({"name": "region", "default": "us-east-1"})
+EOF
+
+title "Detect and save 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/job_params_variables/test.toml

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

acceptance/bundle/resource_deps/jobs_update_remote/out.plan_update.direct.json

@@ -125,7 +125,7 @@
           "reason": "empty",
           "remote": {}
         },
-        "job_clusters[0].new_cluster.num_workers": {
+        "job_clusters[job_cluster_key='key'].new_cluster.num_workers": {
           "action": "skip",
           "reason": "empty",
           "old": 0,

bundle/direct/dresources/job.go

@@ -59,9 +59,24 @@ func getTaskKey(x jobs.Task) (string, string) {
 	return "task_key", x.TaskKey
 }
 
+func getParameterName(x jobs.JobParameterDefinition) (string, string) {
+	return "name", x.Name
+}
+
+func getJobClusterKey(x jobs.JobCluster) (string, string) {
+	return "job_cluster_key", x.JobClusterKey
+}
+
+func getEnvironmentKey(x jobs.JobEnvironment) (string, string) {
+	return "environment_key", x.EnvironmentKey
+}
+
 func (*ResourceJob) KeyedSlices() map[string]any {
 	return map[string]any{
-		"tasks": getTaskKey,
+		"tasks":        getTaskKey,
+		"parameters":   getParameterName,
+		"job_clusters": getJobClusterKey,
+		"environments": getEnvironmentKey,
 	}
 }