direct: resolve immutable fields during plan (#4417)

## Changes
If field is in recreate_on_changes resource setting then we assume it
does not change unless the resource is recreated and can resolve it
before deploying the resource (like we do with "id" field).

## Why
More precise plan. In some cases this can downgrade 'recreate' to
'update' or 'update' to 'skip'.

## Tests
New regression test.

Author: denik

acceptance/bundle/resource_deps/immutable_field_ref/databricks.yml

@@ -0,0 +1,18 @@
+bundle:
+  name: test-bundle
+
+resources:
+  schemas:
+    schema:
+      catalog_name: main
+      name: schema
+  volumes:
+    foo:
+      catalog_name: main
+      schema_name: schema
+      name: volume
+      comment: comment1
+  jobs:
+    bar:
+      name: job bar
+      description: storage_location is ${resources.volumes.foo.storage_location}, catalog_name is ${resources.volumes.foo.catalog_name}

acceptance/bundle/resource_deps/immutable_field_ref/out.test.toml

@@ -0,0 +1,13 @@
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+>>> update_file.py databricks.yml comment1 comment2
+
+>>> [CLI] bundle plan
+update volumes.foo
+
+Plan: 0 to add, 1 to change, 0 to delete, 2 unchanged

acceptance/bundle/resource_deps/immutable_field_ref/output.txt

@@ -0,0 +1,5 @@
+trace $CLI bundle deploy
+trace update_file.py databricks.yml "comment1" "comment2"
+# Even though we have remote reference in jobs.bar (to volumes.foo.storage_location), since that field is immutable
+# we skip updating the job.
+trace $CLI bundle plan | contains.py '1 to change, 0 to delete, 2 unchanged' 'update volumes.foo' '!update jobs' '!update schemas'

acceptance/bundle/resource_deps/immutable_field_ref/script

@@ -0,0 +1,2 @@
+RecordRequests = false
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]

acceptance/bundle/resource_deps/immutable_field_ref/test.toml

@@ -160,7 +160,7 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa
 	}
 }
 
-func (b *DeploymentBundle) LookupReferenceRemote(ctx context.Context, path *structpath.PathNode) (any, error) {
+func (b *DeploymentBundle) LookupReferencePostDeploy(ctx context.Context, path *structpath.PathNode) (any, error) {
 	// TODO: Prefix(3) assumes resources.jobs.foo but not resources.jobs.foo.permissions
 	targetResourceKey := path.Prefix(3).String()
 	fieldPath := path.SkipPrefix(3)

bundle/direct/bundle_apply.go

@@ -281,10 +281,9 @@ func (b *DeploymentBundle) CalculatePlan(ctx context.Context, client *databricks
 			action = getMaxAction(entry.Changes)
 		}
 
-		if action == deployplan.Skip {
-			// resource is not going to change, can use remoteState to resolve references
-			b.RemoteStateCache.Store(resourceKey, remoteState)
-		}
+		// Note, this unconditionally stores remoteState. However, it may updated post-deploy, so whether
+		// it can be used for variable resolution depends on several factors, see canReadRemoteCache in LookupReferencePreDeploy
+		b.RemoteStateCache.Store(resourceKey, remoteState)
 
 		// Validate that resources without DoUpdate don't have update actions
 		if action == deployplan.Update && !adapter.HasDoUpdate() {
@@ -519,8 +518,7 @@ func isEmptyStruct(values ...any) bool {
 	return true
 }
 
-// TODO: calling this "Local" is not right, it can resolve "id" and remote refrences for "skip" targets
-func (b *DeploymentBundle) LookupReferenceLocal(ctx context.Context, path *structpath.PathNode) (any, error) {
+func (b *DeploymentBundle) LookupReferencePreDeploy(ctx context.Context, path *structpath.PathNode) (any, error) {
 	// TODO: Prefix(3) assumes resources.jobs.foo but not resources.jobs.foo.permissions
 	targetResourceKey := path.Prefix(3).String()
 
@@ -598,15 +596,15 @@ func (b *DeploymentBundle) LookupReferenceLocal(ctx context.Context, path *struc
 		return value, nil
 	}
 
+	canReadRemoteCache := targetAction == deployplan.Skip || (targetAction.KeepsID() && adapter.IsFieldInRecreateOnChanges(fieldPath))
+
 	if configValidErr != nil && remoteValidErr == nil {
 		// The field is only present in remote state schema.
-		if targetAction != deployplan.Skip {
-			// The resource is going to be updated, so remoteState can change
-			return nil, errDelayed
-		}
-		remoteState, ok := b.RemoteStateCache.Load(targetResourceKey)
-		if ok {
-			return structaccess.Get(remoteState, fieldPath)
+		if canReadRemoteCache {
+			remoteState, ok := b.RemoteStateCache.Load(targetResourceKey)
+			if ok {
+				return structaccess.Get(remoteState, fieldPath)
+			}
 		}
 		return nil, errDelayed
 	}
@@ -619,7 +617,7 @@ func (b *DeploymentBundle) LookupReferenceLocal(ctx context.Context, path *struc
 		return value, nil
 	}
 
-	if targetAction == deployplan.Skip {
+	if canReadRemoteCache {
 		remoteState, ok := b.RemoteStateCache.Load(targetResourceKey)
 		if ok {
 			return structaccess.Get(remoteState, fieldPath)
@@ -640,8 +638,6 @@ func (b *DeploymentBundle) getStructVar(resourceKey string) (*structvar.StructVa
 }
 
 // resolveReferences processes all references in entry.NewState.Refs.
-// If isLocal is true, uses LookupReferenceLocal (for planning phase).
-// If isLocal is false, uses LookupReferenceRemote (for apply phase).
 func (b *DeploymentBundle) resolveReferences(ctx context.Context, resourceKey string, entry *deployplan.PlanEntry, errorPrefix string, isLocal bool) bool {
 	sv, err := b.getStructVar(resourceKey)
 	if err != nil {
@@ -667,7 +663,7 @@ func (b *DeploymentBundle) resolveReferences(ctx context.Context, resourceKey st
 
 			var value any
 			if isLocal {
-				value, err = b.LookupReferenceLocal(ctx, targetPath)
+				value, err = b.LookupReferencePreDeploy(ctx, targetPath)
 				if err != nil {
 					if errors.Is(err, errDelayed) {
 						continue
@@ -676,7 +672,7 @@ func (b *DeploymentBundle) resolveReferences(ctx context.Context, resourceKey st
 					return false
 				}
 			} else {
-				value, err = b.LookupReferenceRemote(ctx, targetPath)
+				value, err = b.LookupReferencePostDeploy(ctx, targetPath)
 				if err != nil {
 					logdiag.LogError(ctx, fmt.Errorf("%s: cannot resolve %q: %w", errorPrefix, ref, err))
 					return false

bundle/direct/bundle_plan.go

@@ -355,6 +355,15 @@ func (a *Adapter) ResourceConfig() *ResourceLifecycleConfig {
 	return a.resourceConfig
 }
 
+func (a *Adapter) IsFieldInRecreateOnChanges(path *structpath.PathNode) bool {
+	for _, p := range a.resourceConfig.RecreateOnChanges {
+		if path.HasPrefix(p) {
+			return true
+		}
+	}
+	return false
+}
+
 func (a *Adapter) PrepareState(input any) (any, error) {
 	outs, err := a.prepareState.Call(input)
 	if err != nil {