Fix yaml formatting in config-remote-sync (#4468)

## Changes

Fix cases where the config-remote-sync command formats nodes in [Flow
style](https://www.yaml.info/learn/flowstyle.html)

## Why

It happens when a neighbor node is removed during sync, and new fields
are then added to the same parent. Empty nodes can't be expressed in
block style, therefore, yaml patcher falls back to using Flow
formatting, which is then inherited by created child nodes

TODO: A good alternative is to remove parent nodes during the Resolve
phase, when all of their keys/items are removed, but this is a separate
task and needs to be tested for edge cases. I'll clean up this fix if
the parent removal is a valid option

## Tests
Added acceptance tests

<!-- If your PR needs to be included in the release notes for next
release,
add a separate entry in NEXT_CHANGELOG.md as part of your PR. -->

Author: ilyakuz-db

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

@@ -23,3 +23,8 @@ resources:
       tags:
         env: dev # environment tag
         team: data-eng
+
+      # Flow-style formatting (should be preserved)
+      parameters:
+        - {name: catalog, default: main}
+        - {name: schema, default: dev}

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

@@ -31,12 +31,17 @@ Resource: resources.jobs.my_job
 +      max_concurrent_runs: 5
        # Task configuration
        tasks:
-@@ -19,5 +17,4 @@
+@@ -19,10 +17,8 @@
              node_type_id: [NODE_TYPE_ID]
              num_workers: 1 # inline comment about workers
 -
        # Tags for categorization
        tags:
+         env: dev # environment tag
+         team: data-eng
+-
+       # Flow-style formatting (should be preserved)
+       parameters:
 
 >>> [CLI] bundle destroy --auto-approve
 The following resources will be deleted:

acceptance/bundle/config-remote-sync/job_fields/out.test.toml

@@ -14,7 +14,9 @@ Resource: resources.jobs.my_job
   email_notifications.on_failure: add
   parameters: replace
   tags['team']: add
-  trigger.periodic.interval: replace
+  trigger.pause_status: add
+  trigger.periodic: remove
+  trigger.table_update: add
 
 
 
@@ -29,7 +31,7 @@ Resource: resources.jobs.my_job
 -
  resources:
    jobs:
-@@ -8,12 +7,17 @@
+@@ -8,13 +7,19 @@
          on_success:
            - success@example.com
 +        no_alert_for_skipped_runs: true
@@ -47,12 +49,16 @@ Resource: resources.jobs.my_job
 +        - default: us-east-1
 +          name: region
        trigger:
-         periodic:
+-        periodic:
 -          interval: 1
-+          interval: 2
-           unit: DAYS
+-          unit: DAYS
++        pause_status: UNPAUSED
++        table_update:
++          table_names:
++            - samples.nyctaxi.trips
        environments:
-@@ -31,5 +35,6 @@
+         - environment_key: default
+@@ -31,5 +36,6 @@
              node_type_id: [NODE_TYPE_ID]
              num_workers: 1
 -

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

@@ -18,7 +18,8 @@ edit_resource.py jobs $job_id <<EOF
 r["email_notifications"]["on_failure"] = ["failure@example.com"]
 r["email_notifications"]["no_alert_for_skipped_runs"] = True
 r["parameters"].append({"name": "region", "default": "us-east-1"})
-r["trigger"]["periodic"]["interval"] = 2
+r["trigger"] = {"pause_status": "UNPAUSED", "table_update": {"table_names": ["samples.nyctaxi.trips"]}}
+
 if "tags" not in r:
     r["tags"] = {}
 r["tags"]["team"] = "data"

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

@@ -1,4 +1,5 @@
 Cloud = true
+RequiresUnityCatalog = true
 
 RecordRequests = false
 Ignore = [".databricks", "dummy.whl", "databricks.yml", "databricks.yml.backup"]

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

@@ -1,6 +1,7 @@
 package configsync
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
@@ -14,12 +15,14 @@ import (
 	"github.com/databricks/cli/libs/structs/structpath"
 	"github.com/palantir/pkg/yamlpatch/gopkgv3yamlpatcher"
 	"github.com/palantir/pkg/yamlpatch/yamlpatch"
+	"go.yaml.in/yaml/v3"
 )
 
 // ApplyChangesToYAML generates YAML files for the given field changes.
 func ApplyChangesToYAML(ctx context.Context, b *bundle.Bundle, fieldChanges []FieldChange) ([]FileChange, error) {
 	originalFiles := make(map[string][]byte)
 	modifiedFiles := make(map[string][]byte)
+	fileFieldChanges := make(map[string][]FieldChange)
 
 	for _, fieldChange := range fieldChanges {
 		filePath := fieldChange.FilePath
@@ -39,14 +42,22 @@ func ApplyChangesToYAML(ctx context.Context, b *bundle.Bundle, fieldChanges []Fi
 		}
 
 		modifiedFiles[filePath] = modifiedContent
+		fileFieldChanges[filePath] = append(fileFieldChanges[filePath], fieldChange)
 	}
 
 	var result []FileChange
 	for filePath := range modifiedFiles {
+		// TODO: A good alternative approach is to remove parent nodes during the Resolve phase,
+		// when all of their keys/items are removed, but this should be tested for edge cases.
+		// In this case flow style will never appear because empty nodes are never serialized and we won't need clearAddedFlowStyle
+		normalized, err := clearAddedFlowStyle(modifiedFiles[filePath], fileFieldChanges[filePath])
+		if err != nil {
+			return nil, fmt.Errorf("failed to normalize YAML style in %s: %w", filePath, err)
+		}
 		result = append(result, FileChange{
 			Path:            filePath,
 			OriginalContent: string(originalFiles[filePath]),
-			ModifiedContent: string(modifiedFiles[filePath]),
+			ModifiedContent: string(normalized),
 		})
 	}
 
@@ -262,3 +273,81 @@ func strPathToJSONPointer(pathStr string) (string, error) {
 	}
 	return "/" + strings.Join(parts, "/"), nil
 }
+
+// clearAddedFlowStyle clears FlowStyle on YAML nodes along the changed field paths.
+// This prevents flow-style formatting (e.g. {key: value}) that yaml.v3 introduces
+// when empty mappings are serialized as "{}" during patch operations
+func clearAddedFlowStyle(content []byte, fieldChanges []FieldChange) ([]byte, error) {
+	var doc yaml.Node
+	if err := yaml.Unmarshal(content, &doc); err != nil {
+		return content, nil
+	}
+	for _, fc := range fieldChanges {
+		for _, candidate := range fc.FieldCandidates {
+			clearFlowStyleAlongPath(&doc, candidate)
+		}
+	}
+	var buf bytes.Buffer
+	enc := yaml.NewEncoder(&buf)
+	enc.SetIndent(2)
+	if err := enc.Encode(&doc); err != nil {
+		return nil, err
+	}
+	return buf.Bytes(), enc.Close()
+}
+
+// clearFlowStyleAlongPath navigates the YAML tree along the given structpath,
+// clearing FlowStyle on every node from root to leaf (inclusive).
+func clearFlowStyleAlongPath(doc *yaml.Node, pathStr string) {
+	node, err := structpath.ParsePath(pathStr)
+	if err != nil {
+		return
+	}
+
+	current := doc
+	if current.Kind == yaml.DocumentNode && len(current.Content) > 0 {
+		current = current.Content[0]
+	}
+
+	for _, n := range node.AsSlice() {
+		current.Style &^= yaml.FlowStyle
+
+		if key, ok := n.StringKey(); ok {
+			if current.Kind != yaml.MappingNode {
+				return
+			}
+			found := false
+			// current.Content: [key1, val1, key2, val2, ...]
+			for i := 0; i+1 < len(current.Content); i += 2 {
+				if current.Content[i].Value == key {
+					current = current.Content[i+1]
+					found = true
+					break
+				}
+			}
+			if !found {
+				return
+			}
+			continue
+		}
+
+		if idx, ok := n.Index(); ok {
+			if current.Kind != yaml.SequenceNode || idx < 0 || idx >= len(current.Content) {
+				return
+			}
+			current = current.Content[idx]
+			continue
+		}
+
+		return
+	}
+
+	clearFlowStyleNodes(current)
+}
+
+func clearFlowStyleNodes(node *yaml.Node) {
+	node.Style &^= yaml.FlowStyle
+	for _, child := range node.Content {
+		clearFlowStyleNodes(child)
+	}
+}