Resolve $resources refs before deployment if set by user (#3423)

## Changes
Resolve ${resources} references internally, before the deployment.

This is v2 of #3370 which was
reverted in #3421 because it
interacted poorly with other resources transformations that are done by
DABs. By moving this resolution later we ensure that all transformation
are visible when we do reference replacement.

## Why
Currently we resolve these via terraform, but that's not going to work
with direct deployment. To minimize the changes between TF and direct
and surface possible issues with this approach, we do this for both
backends.

---------

Co-authored-by: shreyas-goenka <[email protected]>

Author: denik

acceptance/bundle/resource_deps/presets/databricks.yml

@@ -0,0 +1,18 @@
+resources:
+  schemas:
+    bar:
+      catalog_name: mycatalog
+      name: myschema
+  volumes:
+    foo:
+      catalog_name: ${resources.schemas.bar.catalog_name}
+      schema_name: ${resources.schemas.bar.name}
+      name: myname
+  pipelines:
+    mypipeline:
+      # When creating the pipeline we should see dev prefix in schema_name here
+      name: pipeline for ${resources.volumes.foo.catalog_name}.${resources.volumes.foo.schema_name}.${resources.volumes.foo.name}
+
+targets:
+  dev:
+    mode: development

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

@@ -0,0 +1,5 @@
+Local = true
+Cloud = false
+
+[EnvMatrix]
+  DATABRICKS_CLI_DEPLOYMENT = ["terraform", "direct-exp"]

acceptance/bundle/resource_deps/presets/output.txt

@@ -0,0 +1,43 @@
+
+>>> [CLI] bundle validate -t dev -o json
+{
+  "pipelines": {
+    "mypipeline": {
+      "channel": "CURRENT",
+      "deployment": {
+        "kind": "BUNDLE",
+        "metadata_file_path": "/Workspace/Users/[USERNAME]/.bundle/test-bundle/dev/state/metadata.json"
+      },
+      "development": true,
+      "edition": "ADVANCED",
+      "name": "[dev [USERNAME]] pipeline for ${resources.volumes.foo.catalog_name}.${resources.volumes.foo.schema_name}.${resources.volumes.foo.name}",
+      "permissions": [],
+      "tags": {
+        "dev": "[USERNAME]"
+      }
+    }
+  },
+  "schemas": {
+    "bar": {
+      "catalog_name": "mycatalog",
+      "name": "dev_[USERNAME]_myschema"
+    }
+  },
+  "volumes": {
+    "foo": {
+      "catalog_name": "${resources.schemas.bar.catalog_name}",
+      "name": "myname",
+      "schema_name": "${resources.schemas.bar.name}",
+      "volume_type": "MANAGED"
+    }
+  }
+}
+
+>>> [CLI] bundle deploy -t dev
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/dev/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+>>> jq -s .[] | select(.path=="/api/2.0/pipelines") | .body.name out.requests.txt
+"[dev [USERNAME]] pipeline for mycatalog.dev_[USERNAME]_myschema.myname"

acceptance/bundle/resource_deps/presets/script

@@ -0,0 +1,4 @@
+trace $CLI bundle validate -t dev -o json | jq .resources
+trace $CLI bundle deploy -t dev
+trace jq -s '.[] | select(.path=="/api/2.0/pipelines") | .body.name' out.requests.txt
+rm out.requests.txt

acceptance/bundle/resource_deps/presets/test.toml

@@ -0,0 +1 @@
+RecordRequests = true

bundle/config/mutator/resolve_variable_references.go

@@ -229,8 +229,24 @@ func (m *resolveVariableReferences) resolveOnce(b *bundle.Bundle, prefixes []dyn
 				// Perform resolution only if the path starts with one of the specified prefixes.
 				for _, prefix := range prefixes {
 					if path.HasPrefix(prefix) {
-						hasUpdates = true
-						return m.lookupFn(normalized, path, b)
+						isOpt := prefix[0].Key() == "resources"
+						var value dyn.Value
+						var err error
+						if isOpt {
+							// We don't want injected zero value when resolving $resources.
+							// We only want entries that are explicitly provided by users, so we're using root not normalized here.
+							value, err = m.lookupFn(root, path, b)
+							if !value.IsValid() {
+								// Not having a value is not an error in this case, it might be resolved at deploy time. For example, output only fields.
+								// TODO: we still could check whether it's part of the schema or not. If latter, we can reject it right away.
+								// TODO: This might be better done after we got rid of TF.
+								return dyn.InvalidValue, dynvar.ErrSkipResolution
+							}
+						} else {
+							value, err = m.lookupFn(normalized, path, b)
+						}
+						hasUpdates = hasUpdates || (err == nil && value.IsValid())
+						return value, err
 					}
 				}
 

bundle/config/mutator/resourcemutator/resource_mutator.go

@@ -154,11 +154,6 @@ func applyNormalizeMutators(ctx context.Context, b *bundle.Bundle) {
 		// Updates (dynamic): resources.apps.*.resources (merges app resources with the same name)
 		MergeApps(),
 
-		// Reads (typed): resources.pipelines.*.{catalog,schema,target}, resources.volumes.*.{catalog_name,schema_name} (checks for schema references)
-		// Updates (typed): resources.pipelines.*.{schema,target}, resources.volumes.*.schema_name (converts implicit schema references to explicit ${resources.schemas.<schema_key>.name} syntax)
-		// Translates implicit schema references in DLT pipelines or UC Volumes to explicit syntax to capture dependencies
-		CaptureSchemaDependency(),
-
 		// Reads (dynamic): resources.dashboards.*.file_path
 		// Updates (dynamic): resources.dashboards.*.serialized_dashboard
 		// Drops (dynamic): resources.dashboards.*.file_path

bundle/phases/deploy.go

@@ -9,6 +9,7 @@ import (
 	"github.com/databricks/cli/bundle/artifacts"
 	"github.com/databricks/cli/bundle/config"
 	"github.com/databricks/cli/bundle/config/mutator"
+	"github.com/databricks/cli/bundle/config/mutator/resourcemutator"
 	"github.com/databricks/cli/bundle/deploy"
 	"github.com/databricks/cli/bundle/deploy/files"
 	"github.com/databricks/cli/bundle/deploy/lock"
@@ -173,6 +174,17 @@ func deployPrepare(ctx context.Context, b *bundle.Bundle) {
 		// TransformWheelTask depends on it and planning also depends on it.
 		libraries.Upload(),
 		trampoline.TransformWheelTask(),
+
+		mutator.ResolveVariableReferencesOnlyResources(
+			"resources",
+		),
+
+		// Reads (typed): resources.pipelines.*.{catalog,schema,target}, resources.volumes.*.{catalog_name,schema_name} (checks for schema references)
+		// Updates (typed): resources.pipelines.*.{schema,target}, resources.volumes.*.schema_name (converts implicit schema references to explicit ${resources.schemas.<schema_key>.name} syntax)
+		// Translates implicit schema references in DLT pipelines or UC Volumes to explicit syntax to capture dependencies
+		// Needs to be run after ${resources} resolution since otherwise that undoes the change here.
+		// TODO: one we have depends_on support we should leverage that here and move this back to initialize phase.
+		resourcemutator.CaptureSchemaDependency(),
 	)
 }