WIP new migration

acceptance/bundle/help/bundle-deployment-migrate/output.txt

@@ -13,7 +13,7 @@ Usage:
 
 Flags:
   -h, --help          help for migrate
-      --noplancheck   Skip running bundle plan before migration.
+      --noplancheck   No-op (kept for compatibility).
 
 Global Flags:
       --debug            enable debug logging

acceptance/bundle/migrate/basic/output.txt

@@ -11,8 +11,6 @@ Updating deployment state...
 Deployment complete!
 
 >>> [CLI] bundle deployment migrate
-Note: Migration should be done after a full deploy. Running plan now to verify that deployment was done:
-Plan: 0 to add, 0 to change, 0 to delete, 3 unchanged
 Success! Migrated 3 resources to direct engine state file: [TEST_TMP_DIR]/.databricks/bundle/dev/resources.json
 
 Validate the migration by running "databricks bundle plan", there should be no actions planned.

acceptance/bundle/migrate/dashboards/output.txt

@@ -6,8 +6,6 @@ Updating deployment state...
 Deployment complete!
 
 >>> [CLI] bundle deployment migrate
-Note: Migration should be done after a full deploy. Running plan now to verify that deployment was done:
-Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged
 Success! Migrated 1 resources to direct engine state file: [TEST_TMP_DIR]/.databricks/bundle/default/resources.json
 
 Validate the migration by running "databricks bundle plan", there should be no actions planned.

acceptance/bundle/migrate/default-python/output.txt

@@ -24,7 +24,6 @@ Deployment complete!
 
 >>> musterr [CLI] bundle deployment migrate
 Building python_artifact...
-Note: Migration should be done after a full deploy. Running plan now to verify that deployment was done:
 Building python_artifact...
 update jobs.sample_job
 

acceptance/bundle/migrate/grants/output.txt

@@ -6,8 +6,6 @@ Updating deployment state...
 Deployment complete!
 
 >>> [CLI] bundle deployment migrate
-Note: Migration should be done after a full deploy. Running plan now to verify that deployment was done:
-Plan: 0 to add, 0 to change, 0 to delete, 6 unchanged
 Success! Migrated 6 resources to direct engine state file: [TEST_TMP_DIR]/.databricks/bundle/default/resources.json
 
 Validate the migration by running "databricks bundle plan", there should be no actions planned.

acceptance/bundle/migrate/permissions/output.txt

@@ -6,8 +6,6 @@ Updating deployment state...
 Deployment complete!
 
 >>> [CLI] bundle deployment migrate
-Note: Migration should be done after a full deploy. Running plan now to verify that deployment was done:
-Plan: 0 to add, 0 to change, 0 to delete, 4 unchanged
 Success! Migrated 4 resources to direct engine state file: [TEST_TMP_DIR]/.databricks/bundle/default/resources.json
 
 Validate the migration by running "databricks bundle plan", there should be no actions planned.

acceptance/bundle/migrate/profile_arg/output.txt

@@ -6,8 +6,6 @@ Updating deployment state...
 Deployment complete!
 
 >>> [CLI] bundle deployment migrate -p non_existent321
-Note: Migration should be done after a full deploy. Running plan now to verify that deployment was done:
-Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged
 Success! Migrated 1 resources to direct engine state file: [TEST_TMP_DIR]/.databricks/bundle/dev/resources.json
 
 Validate the migration by running "databricks bundle plan -p non_existent321", there should be no actions planned.
@@ -24,8 +22,6 @@ Updating deployment state...
 Deployment complete!
 
 >>> [CLI] bundle deployment migrate -p non_existent321 -t prod
-Note: Migration should be done after a full deploy. Running plan now to verify that deployment was done:
-Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged
 Success! Migrated 2 resources to direct engine state file: [TEST_TMP_DIR]/.databricks/bundle/prod/resources.json
 
 Validate the migration by running "databricks bundle plan -t prod -p non_existent321", there should be no actions planned.

acceptance/bundle/migrate/runas/output.txt

@@ -81,7 +81,6 @@ Consider using a adding a top-level permissions section such as the following:
 See https://docs.databricks.com/dev-tools/bundles/permissions.html to learn more about permission configuration.
   in databricks.yml:5:3
 
-Note: Migration should be done after a full deploy. Running plan now to verify that deployment was done:
 Recommendation: permissions section should explicitly include the current deployment identity '[USERNAME]' or one of its groups
 If it is not included, CAN_MANAGE permissions are only applied if the present identity is used to deploy.
 

acceptance/bundle/migrate/var_arg/output.txt

@@ -6,8 +6,6 @@ Updating deployment state...
 Deployment complete!
 
 >>> [CLI] bundle deployment migrate --var=job_name=Custom Job Name
-Note: Migration should be done after a full deploy. Running plan now to verify that deployment was done:
-Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged
 Success! Migrated 1 resources to direct engine state file: [TEST_TMP_DIR]/.databricks/bundle/dev/resources.json
 
 Validate the migration by running "databricks bundle plan --var 'job_name=Custom Job Name'", there should be no actions planned.
@@ -39,8 +37,6 @@ Updating deployment state...
 Deployment complete!
 
 >>> [CLI] bundle deployment migrate --var job_name=Custom Job Name
-Note: Migration should be done after a full deploy. Running plan now to verify that deployment was done:
-Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged
 Success! Migrated 1 resources to direct engine state file: [TEST_TMP_DIR]/.databricks/bundle/dev/resources.json
 
 Validate the migration by running "databricks bundle plan --var 'job_name=Custom Job Name'", there should be no actions planned.

bundle/direct/bundle_apply.go

@@ -15,9 +15,7 @@ import (
 	"github.com/databricks/databricks-sdk-go"
 )
 
-type MigrateMode bool
-
-func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.WorkspaceClient, plan *deployplan.Plan, migrateMode MigrateMode) {
+func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.WorkspaceClient, plan *deployplan.Plan) {
 	if plan == nil {
 		panic("Planning is not done")
 	}
@@ -52,9 +50,6 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa
 
 		action := entry.Action
 		errorPrefix := fmt.Sprintf("cannot %s %s", action, resourceKey)
-		if migrateMode {
-			errorPrefix = "cannot migrate " + resourceKey
-		}
 
 		if action == deployplan.Undefined {
 			logdiag.LogError(ctx, fmt.Errorf("cannot deploy %s: unknown action %q", resourceKey, action))
@@ -82,10 +77,6 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa
 		}
 
 		if action == deployplan.Delete {
-			if migrateMode {
-				logdiag.LogError(ctx, fmt.Errorf("%s: Unexpected delete action during migration", errorPrefix))
-				return false
-			}
 			err = d.Destroy(ctx, &b.StateDB)
 			if err != nil {
 				logdiag.LogError(ctx, fmt.Errorf("%s: %w", errorPrefix, err))
@@ -113,19 +104,8 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa
 				return false
 			}
 
-			if migrateMode {
-				// In migration mode we're reading resources in DAG order so that we have fully resolved config snapshots stored
-				id := b.StateDB.GetResourceID(resourceKey)
-				if id == "" {
-					logdiag.LogError(ctx, fmt.Errorf("state entry not found for %q", resourceKey))
-					return false
-				}
-				err = b.StateDB.SaveState(resourceKey, id, sv.Value, entry.DependsOn)
-			} else {
-				// TODO: redo calcDiff to downgrade planned action if possible (?)
-				err = d.Deploy(ctx, &b.StateDB, sv.Value, action, entry)
-			}
-
+			// TODO: redo calcDiff to downgrade planned action if possible (?)
+			err = d.Deploy(ctx, &b.StateDB, sv.Value, action, entry)
 			if err != nil {
 				logdiag.LogError(ctx, fmt.Errorf("%s: %w", errorPrefix, err))
 				return false

bundle/direct/bundle_plan.go

@@ -931,6 +931,12 @@ func (b *DeploymentBundle) makePlan(ctx context.Context, configRoot *config.Root
 	return p, nil
 }
 
+// ExtractReferences extracts all variable references from the config subtree rooted at node.
+// Returns a map from structpath string (field path within the resource) to template string.
+func ExtractReferences(root dyn.Value, node string) (map[string]string, error) {
+	return extractReferences(root, node)
+}
+
 func extractReferences(root dyn.Value, node string) (map[string]string, error) {
 	nodeType := config.GetResourceTypeFromKey(node)
 	refs := make(map[string]string)

bundle/migrate/build_state.go

@@ -0,0 +1,179 @@
+package migrate
+
+import (
+	"context"
+	"fmt"
+	"maps"
+	"strings"
+
+	"github.com/databricks/cli/bundle/config"
+	"github.com/databricks/cli/bundle/config/resources"
+	"github.com/databricks/cli/bundle/deploy/terraform"
+	"github.com/databricks/cli/bundle/direct"
+	"github.com/databricks/cli/bundle/direct/dresources"
+	"github.com/databricks/cli/bundle/direct/dstate"
+	"github.com/databricks/cli/libs/dyn"
+	"github.com/databricks/cli/libs/log"
+	"github.com/databricks/cli/libs/structs/structaccess"
+	"github.com/databricks/cli/libs/structs/structpath"
+	"github.com/databricks/cli/libs/structs/structvar"
+)
+
+// BuildStateFromTF iterates over bundle resources, resolves cross-resource
+// references using TF state attributes, and writes each resource's state entry.
+// configRoot should be an un-interpolated config (with ${resources.*} references).
+func BuildStateFromTF(
+	ctx context.Context,
+	configRoot *config.Root,
+	adapters map[string]*dresources.Adapter,
+	stateDB *dstate.DeploymentState,
+	tfAttrs TFStateAttrs,
+	tfIDs terraform.ExportedResourcesMap,
+	etags map[string]string,
+) error {
+	// Collect all resource nodes (same patterns as makePlan).
+	var nodes []string
+	patterns := []dyn.Pattern{
+		dyn.NewPattern(dyn.Key("resources"), dyn.AnyKey(), dyn.AnyKey()),
+		dyn.NewPattern(dyn.Key("resources"), dyn.AnyKey(), dyn.AnyKey(), dyn.Key("permissions")),
+		dyn.NewPattern(dyn.Key("resources"), dyn.AnyKey(), dyn.AnyKey(), dyn.Key("grants")),
+	}
+	for _, pat := range patterns {
+		_, err := dyn.MapByPattern(
+			configRoot.Value(),
+			pat,
+			func(p dyn.Path, v dyn.Value) (dyn.Value, error) {
+				nodes = append(nodes, p.String())
+				return dyn.InvalidValue, nil
+			},
+		)
+		if err != nil {
+			return err
+		}
+	}
+
+	for _, node := range nodes {
+		idEntry, ok := tfIDs[node]
+		if !ok {
+			// Resource is in config but not in TF state (new resource); skip.
+			continue
+		}
+
+		group := config.GetResourceTypeFromKey(node)
+		if group == "" {
+			return fmt.Errorf("cannot determine resource type for %q", node)
+		}
+
+		adapter, ok := adapters[group]
+		if !ok {
+			log.Warnf(ctx, "unsupported resource type %q for %s, skipping", group, node)
+			continue
+		}
+
+		inputConfig, err := configRoot.GetResourceConfig(node)
+		if err != nil {
+			return fmt.Errorf("%s: getting config: %w", node, err)
+		}
+
+		baseRefs := map[string]string{}
+
+		switch {
+		case strings.HasSuffix(node, ".permissions"):
+			var sv *structvar.StructVar
+			if strings.HasPrefix(node, "resources.secret_scopes.") {
+				typedConfig, ok := inputConfig.(*[]resources.SecretScopePermission)
+				if !ok {
+					return fmt.Errorf("%s: expected *[]resources.SecretScopePermission, got %T", node, inputConfig)
+				}
+				sv, err = dresources.PrepareSecretScopeAclsInputConfig(*typedConfig, node)
+				if err != nil {
+					return fmt.Errorf("%s: preparing secret scope ACLs config: %w", node, err)
+				}
+			} else {
+				sv, err = dresources.PreparePermissionsInputConfig(inputConfig, node)
+				if err != nil {
+					return fmt.Errorf("%s: preparing permissions config: %w", node, err)
+				}
+			}
+			inputConfig = sv.Value
+			baseRefs = sv.Refs
+
+		case strings.HasSuffix(node, ".grants"):
+			sv, err := dresources.PrepareGrantsInputConfig(inputConfig, node)
+			if err != nil {
+				return fmt.Errorf("%s: preparing grants config: %w", node, err)
+			}
+			inputConfig = sv.Value
+			baseRefs = sv.Refs
+		}
+
+		newStateValue, err := adapter.PrepareState(inputConfig)
+		if err != nil {
+			return fmt.Errorf("%s: PrepareState: %w", node, err)
+		}
+
+		refs, err := direct.ExtractReferences(configRoot.Value(), node)
+		if err != nil {
+			return fmt.Errorf("%s: extracting references: %w", node, err)
+		}
+		maps.Copy(refs, baseRefs)
+
+		sv := structvar.NewStructVar(newStateValue, refs)
+
+		// Resolve each reference using TF state.
+		// node format: "resources.<group>.<name>" or "resources.<group>.<name>.permissions"
+		parts := strings.SplitN(node, ".", 4)
+		var srcGroup, srcName string
+		if len(parts) >= 3 {
+			srcGroup = parts[1]
+			srcName = parts[2]
+		}
+
+		// Collect all field paths that need resolution (avoid modifying map during iteration).
+		type refEntry struct {
+			fieldPathStr string
+			refTemplate  string
+		}
+		var pendingRefs []refEntry
+		for fieldPathStr, refTemplate := range sv.Refs {
+			pendingRefs = append(pendingRefs, refEntry{fieldPathStr, refTemplate})
+		}
+
+		for _, pending := range pendingRefs {
+			fieldPath, err := structpath.ParsePath(pending.fieldPathStr)
+			if err != nil {
+				return fmt.Errorf("%s: parsing field path %q: %w", node, pending.fieldPathStr, err)
+			}
+
+			// ResolveFieldRef returns the fully resolved value for this field,
+			// using either Method A (TF state lookup) or Method B (template evaluation).
+			value, err := ResolveFieldRef(ctx, tfAttrs, srcGroup, srcName, fieldPath, pending.refTemplate)
+			if err != nil {
+				return fmt.Errorf("%s: cannot resolve field %q (template %q): %w", node, pending.fieldPathStr, pending.refTemplate, err)
+			}
+
+			// Set the resolved value directly and remove the ref entry.
+			if err := structaccess.Set(sv.Value, fieldPath, value); err != nil {
+				return fmt.Errorf("%s: cannot set resolved value for field %q: %w", node, pending.fieldPathStr, err)
+			}
+			delete(sv.Refs, pending.fieldPathStr)
+		}
+
+		if len(sv.Refs) > 0 {
+			return fmt.Errorf("%s: unresolved references: %v", node, sv.Refs)
+		}
+
+		// Handle etag for dashboards.
+		if etag := etags[node]; etag != "" {
+			if err := structaccess.Set(sv.Value, structpath.NewStringKey(nil, "etag"), etag); err != nil {
+				return fmt.Errorf("%s: cannot set etag: %w", node, err)
+			}
+		}
+
+		if err := stateDB.SaveState(node, idEntry.ID, sv.Value, nil); err != nil {
+			return fmt.Errorf("%s: SaveState: %w", node, err)
+		}
+	}
+
+	return nil
+}