Config sync migrate to resource yml

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

@@ -26,6 +26,7 @@ Resource: resources.jobs.my_job
   email_notifications.on_failure[0]: replace
   max_concurrent_runs: replace
   tags['env']: remove
+  timeout_seconds: remove
 
 
 
@@ -41,15 +42,16 @@ Resource: resources.jobs.my_job
 +              - remote-failure@example.com
            parameters:
              - name: catalog
-@@ -35,7 +35,6 @@
+@@ -35,8 +35,6 @@
                unit: DAYS
            tags:
 -            env: config-production
              team: data-team
 -          max_concurrent_runs: 3
+-          timeout_seconds: 3600
 +          max_concurrent_runs: 5
-           timeout_seconds: 3600
            environments:
+             - environment_key: default
 
 >>> [CLI] bundle destroy --auto-approve
 The following resources will be deleted:

bundle/configsync/defaults.go

@@ -1,177 +1,26 @@
 package configsync
 
-import (
-	"reflect"
-	"strings"
-)
+import "github.com/databricks/cli/libs/structs/structpath"
 
-type (
-	skipAlways           struct{}
-	skipIfZeroOrNil      struct{}
-	skipIfEmptyOrDefault struct {
-		defaults map[string]any
-	}
-)
-
-var (
-	alwaysSkip              = skipAlways{}
-	zeroOrNil               = skipIfZeroOrNil{}
-	emptyEmailNotifications = skipIfEmptyOrDefault{defaults: map[string]any{"no_alert_for_skipped_runs": false}}
-)
-
-// serverSideDefaults contains all hardcoded server-side defaults.
-// This is a temporary solution until the bundle plan issue is resolved.
-// Fields mapped to alwaysSkip are always considered defaults regardless of value.
-// Other fields are compared using reflect.DeepEqual.
-var serverSideDefaults = map[string]any{
-	// Job-level fields
-	"resources.jobs.*.timeout_seconds":       zeroOrNil,
-	"resources.jobs.*.email_notifications":   emptyEmailNotifications,
-	"resources.jobs.*.webhook_notifications": map[string]any{},
-	"resources.jobs.*.edit_mode":             alwaysSkip, // set by CLI
-	"resources.jobs.*.performance_target":    "PERFORMANCE_OPTIMIZED",
-
-	// Task-level fields
-	"resources.jobs.*.tasks[*].run_if":                     "ALL_SUCCESS",
-	"resources.jobs.*.tasks[*].disabled":                   false,
-	"resources.jobs.*.tasks[*].timeout_seconds":            zeroOrNil,
-	"resources.jobs.*.tasks[*].notebook_task.source":       "WORKSPACE",
-	"resources.jobs.*.tasks[*].email_notifications":        emptyEmailNotifications,
-	"resources.jobs.*.tasks[*].webhook_notifications":      map[string]any{},
-	"resources.jobs.*.tasks[*].pipeline_task.full_refresh": false,
-
-	"resources.jobs.*.tasks[*].for_each_task.task.run_if":                "ALL_SUCCESS",
-	"resources.jobs.*.tasks[*].for_each_task.task.disabled":              false,
-	"resources.jobs.*.tasks[*].for_each_task.task.timeout_seconds":       zeroOrNil,
-	"resources.jobs.*.tasks[*].for_each_task.task.notebook_task.source":  "WORKSPACE",
-	"resources.jobs.*.tasks[*].for_each_task.task.email_notifications":   emptyEmailNotifications,
-	"resources.jobs.*.tasks[*].for_each_task.task.webhook_notifications": map[string]any{},
-
-	// Cluster fields (tasks)
-	"resources.jobs.*.tasks[*].new_cluster.aws_attributes":      alwaysSkip,
-	"resources.jobs.*.tasks[*].new_cluster.azure_attributes":    alwaysSkip,
-	"resources.jobs.*.tasks[*].new_cluster.gcp_attributes":      alwaysSkip,
-	"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,
-
-	// Cluster fields (job_clusters)
-	"resources.jobs.*.job_clusters[*].new_cluster.aws_attributes":      alwaysSkip,
-	"resources.jobs.*.job_clusters[*].new_cluster.azure_attributes":    alwaysSkip,
-	"resources.jobs.*.job_clusters[*].new_cluster.gcp_attributes":      alwaysSkip,
-	"resources.jobs.*.job_clusters[*].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.*.job_clusters[*].new_cluster.enable_elastic_disk": alwaysSkip,    // deprecated field
-	"resources.jobs.*.job_clusters[*].new_cluster.single_user_name":    alwaysSkip,
-
-	// Standalone cluster fields
-	"resources.clusters.*.aws_attributes":      alwaysSkip,
-	"resources.clusters.*.azure_attributes":    alwaysSkip,
-	"resources.clusters.*.gcp_attributes":      alwaysSkip,
-	"resources.clusters.*.data_security_mode":  "SINGLE_USER",
-	"resources.clusters.*.driver_node_type_id": alwaysSkip,
-	"resources.clusters.*.enable_elastic_disk": alwaysSkip,
-	"resources.clusters.*.single_user_name":    alwaysSkip,
-
-	// Experiment fields
-	"resources.experiments.*.artifact_location": alwaysSkip,
-
-	// Registered model fields
-	"resources.registered_models.*.full_name":        alwaysSkip,
-	"resources.registered_models.*.metastore_id":     alwaysSkip,
-	"resources.registered_models.*.owner":            alwaysSkip,
-	"resources.registered_models.*.storage_location": alwaysSkip,
-
-	// Volume fields
-	"resources.volumes.*.storage_location": alwaysSkip,
-
-	// SQL warehouse fields
-	"resources.sql_warehouses.*.creator_name":     alwaysSkip,
-	"resources.sql_warehouses.*.min_num_clusters": int64(1),
-	"resources.sql_warehouses.*.warehouse_type":   "CLASSIC",
-
-	// Terraform defaults
-	"resources.jobs.*.run_as": alwaysSkip,
-
-	// Pipeline fields
-	"resources.pipelines.*.storage":    alwaysSkip,
-	"resources.pipelines.*.continuous": false,
-}
-
-func shouldSkipField(path string, value any) bool {
-	for pattern, expected := range serverSideDefaults {
-		if matchPattern(pattern, path) {
-			if _, ok := expected.(skipAlways); ok {
-				return true
-			}
-			if _, ok := expected.(skipIfZeroOrNil); ok {
-				return value == nil || value == int64(0)
-			}
-			if marker, ok := expected.(skipIfEmptyOrDefault); ok {
-				m, ok := value.(map[string]any)
-				if !ok {
-					return false
-				}
-				if len(m) == 0 {
-					return true
-				}
-				return reflect.DeepEqual(m, marker.defaults)
-			}
-			return reflect.DeepEqual(value, expected)
-		}
-	}
-	return false
-}
-
-func matchPattern(pattern, path string) bool {
-	patternParts := strings.Split(pattern, ".")
-	pathParts := strings.Split(path, ".")
-	return matchParts(patternParts, pathParts)
-}
-
-func matchParts(patternParts, pathParts []string) bool {
-	if len(patternParts) == 0 && len(pathParts) == 0 {
-		return true
-	}
-	if len(patternParts) == 0 || len(pathParts) == 0 {
-		return false
-	}
-
-	patternPart := patternParts[0]
-	pathPart := pathParts[0]
-
-	if patternPart == "*" {
-		return matchParts(patternParts[1:], pathParts[1:])
-	}
-
-	if strings.Contains(patternPart, "[*]") {
-		prefix := strings.Split(patternPart, "[*]")[0]
-
-		if strings.HasPrefix(pathPart, prefix) && strings.Contains(pathPart, "[") {
-			return matchParts(patternParts[1:], pathParts[1:])
-		}
-		return false
-	}
-
-	if patternPart == pathPart {
-		return matchParts(patternParts[1:], pathParts[1:])
-	}
-
-	return false
+type resetRule struct {
+	field *structpath.PatternNode
+	value any
 }
 
-// resetValues contains all values that should be used to reset CLI-defaulted fields.
-// If CLI-defaulted field is changed on remote and should be disabled (e.g. queueing disabled -> remote field is nil)
-// we can't define it in the config as "null" because CLI default will be applied again.
-var resetValues = map[string]any{
-	"resources.jobs.*.queue": map[string]any{
-		"enabled": false,
+// resetValues defines values that should replace CLI-defaulted fields.
+// If a CLI-defaulted field is changed on remote and should be disabled
+// (e.g. queueing disabled → remote field is nil), we can't define it
+// in the config as "null" because the CLI default will be applied again.
+var resetValues = map[string][]resetRule{
+	"jobs": {
+		{field: structpath.MustParsePattern("queue"), value: map[string]any{"enabled": false}},
 	},
 }
 
-func resetValueIfNeeded(path string, value any) any {
-	for pattern, expected := range resetValues {
-		if matchPattern(pattern, path) {
-			return expected
+func resetValueIfNeeded(resourceType string, path *structpath.PathNode, value any) any {
+	for _, rule := range resetValues[resourceType] {
+		if path.HasPatternPrefix(rule.field) {
+			return rule.value
 		}
 	}
 	return value

bundle/configsync/diff.go

@@ -10,13 +10,16 @@ import (
 	"path/filepath"
 
 	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/config"
 	"github.com/databricks/cli/bundle/config/engine"
 	"github.com/databricks/cli/bundle/deploy"
 	"github.com/databricks/cli/bundle/deployplan"
 	"github.com/databricks/cli/bundle/direct"
+	"github.com/databricks/cli/bundle/direct/dresources"
 	"github.com/databricks/cli/libs/dyn"
 	"github.com/databricks/cli/libs/dyn/convert"
 	"github.com/databricks/cli/libs/log"
+	"github.com/databricks/cli/libs/structs/structpath"
 )
 
 type OperationType string
@@ -47,16 +50,26 @@ func normalizeValue(v any) (any, error) {
 	return dynValue.AsAny(), nil
 }
 
-func filterEntityDefaults(basePath string, value any) any {
+// shouldClassifySkip checks if a value should be skipped using ClassifyChange
+// with a synthetic ChangeDesc (Old=nil, New=nil, Remote=value).
+func shouldClassifySkip(ctx context.Context, adapter *dresources.Adapter, path *structpath.PathNode, value, remoteState any) bool {
+	ch := &deployplan.ChangeDesc{Old: nil, New: nil, Remote: value}
+	if err := direct.ClassifyChange(ctx, adapter, path, ch, remoteState); err != nil {
+		return false
+	}
+	return ch.Action == deployplan.Skip
+}
+
+func filterEntityDefaults(ctx context.Context, adapter *dresources.Adapter, basePath *structpath.PathNode, value, remoteState any) any {
 	if value == nil {
 		return nil
 	}
 
 	if arr, ok := value.([]any); ok {
 		result := make([]any, 0, len(arr))
 		for i, elem := range arr {
-			elementPath := fmt.Sprintf("%s[%d]", basePath, i)
-			result = append(result, filterEntityDefaults(elementPath, elem))
+			elemPath := structpath.NewIndex(basePath, i)
+			result = append(result, filterEntityDefaults(ctx, adapter, elemPath, elem, remoteState))
 		}
 		return result
 	}
@@ -68,37 +81,43 @@ func filterEntityDefaults(basePath string, value any) any {
 
 	result := make(map[string]any)
 	for key, val := range m {
-		fieldPath := basePath + "." + key
-
-		if shouldSkipField(fieldPath, val) {
+		fieldPath := structpath.NewDotString(basePath, key)
+		if shouldClassifySkip(ctx, adapter, fieldPath, val, remoteState) {
 			continue
 		}
-
 		if nestedMap, ok := val.(map[string]any); ok {
-			result[key] = filterEntityDefaults(fieldPath, nestedMap)
+			filtered := filterEntityDefaults(ctx, adapter, fieldPath, nestedMap, remoteState)
+			if filtered != nil {
+				result[key] = filtered
+			}
 		} else {
 			result[key] = val
 		}
 	}
 
+	if len(result) == 0 {
+		return nil
+	}
+
 	return result
 }
 
-func convertChangeDesc(path string, cd *deployplan.ChangeDesc) (*ConfigChangeDesc, error) {
-	hasConfigValue := cd.Old != nil || cd.New != nil
+func convertChangeDesc(ctx context.Context, adapter *dresources.Adapter, resourceType, fieldPath string, cd *deployplan.ChangeDesc, remoteState any) (*ConfigChangeDesc, error) {
+	pathNode, err := structpath.ParsePath(fieldPath)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse path %q: %w", fieldPath, err)
+	}
+
+	hasConfigValue := cd.New != nil
 	normalizedValue, err := normalizeValue(cd.Remote)
 	if err != nil {
 		return nil, fmt.Errorf("failed to normalize remote value: %w", err)
 	}
 
-	if shouldSkipField(path, normalizedValue) {
-		return &ConfigChangeDesc{
-			Operation: OperationSkip,
-		}, nil
-	}
+	// Recursive nested entity filtering using ClassifyChange.
+	normalizedValue = filterEntityDefaults(ctx, adapter, pathNode, normalizedValue, remoteState)
 
-	normalizedValue = filterEntityDefaults(path, normalizedValue)
-	normalizedValue = resetValueIfNeeded(path, normalizedValue)
+	normalizedValue = resetValueIfNeeded(resourceType, pathNode, normalizedValue)
 
 	var op OperationType
 	if normalizedValue == nil && hasConfigValue {
@@ -143,13 +162,20 @@ func DetectChanges(ctx context.Context, b *bundle.Bundle, engine engine.EngineTy
 	for resourceKey, entry := range plan.Plan {
 		resourceChanges := make(ResourceChanges)
 
+		resourceType := config.GetResourceTypeFromKey(resourceKey)
+
+		adapter, ok := deployBundle.Adapters[resourceType]
+		if !ok {
+			return nil, fmt.Errorf("no adapter for resource type %q", resourceType)
+		}
+
 		if entry.Changes != nil {
 			for path, changeDesc := range entry.Changes {
 				if changeDesc.Action == deployplan.Skip {
 					continue
 				}
 
-				change, err := convertChangeDesc(resourceKey+"."+path, changeDesc)
+				change, err := convertChangeDesc(ctx, adapter, resourceType, path, changeDesc, entry.RemoteState)
 				if err != nil {
 					return nil, fmt.Errorf("failed to compute config change for path %s: %w", path, err)
 				}