[draft] Reject unprefixed volume schema references during bundle validation

acceptance/bundle/variables/reference-typo/output.txt

@@ -97,3 +97,28 @@ Workspace:
   Path: /Workspace/Users/[USERNAME]/.bundle/reference-typo-double-nested-typo-multiple/default
 
 Found 1 error
+
+=== Volume schema and catalog references missing resources
+
+>>> [CLI] bundle validate
+Error: volume "ingest" field "schema_name": reference does not exist: ${schemas.test.name}
+  at resources.volumes.ingest.schema_name
+  in databricks.yml:12:20
+
+did you mean:
+  ${resources.schemas.test.name}
+
+Error: volume "ingest" field "catalog_name": reference does not exist: ${schemas.test.catalog_name}
+  at resources.volumes.ingest.catalog_name
+  in databricks.yml:13:21
+
+did you mean:
+  ${resources.schemas.test.catalog_name}
+
+Name: reference-typo-volume-missing-resources
+Target: default
+Workspace:
+  User: [USERNAME]
+  Path: /Workspace/Users/[USERNAME]/.bundle/reference-typo-volume-missing-resources/default
+
+Found 2 errors

acceptance/bundle/variables/reference-typo/script

@@ -15,3 +15,6 @@ withdir double-nested-typo musterr trace $CLI bundle validate
 
 title "Two typos in one reference, outermost close to two variables: multiple suggestions\n"
 withdir double-nested-typo-multiple musterr trace $CLI bundle validate
+
+title "Volume schema and catalog references missing resources\n"
+withdir volume-missing-resources musterr trace $CLI bundle validate

acceptance/bundle/variables/reference-typo/volume-missing-resources/databricks.yml

@@ -0,0 +1,13 @@
+bundle:
+  name: reference-typo-volume-missing-resources
+
+resources:
+  schemas:
+    test:
+      name: myschema
+      catalog_name: main
+  volumes:
+    ingest:
+      name: myvolume
+      schema_name: ${schemas.test.name}
+      catalog_name: ${schemas.test.catalog_name}

bundle/config/mutator/resolve_variable_references.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"maps"
 	"slices"
 	"strings"
 
@@ -15,6 +16,7 @@ import (
 	"github.com/databricks/cli/libs/dyn"
 	"github.com/databricks/cli/libs/dyn/convert"
 	"github.com/databricks/cli/libs/dyn/dynvar"
+	"github.com/databricks/cli/libs/structs/structpath"
 )
 
 /*
@@ -59,7 +61,8 @@ type resolveVariableReferences struct {
 	// outside of 'resources'.
 	includeResources bool
 
-	artifactsReferenceUsed bool
+	artifactsReferenceUsed   bool
+	validateVolumeSchemaRefs bool
 
 	// excludePaths lists variable reference paths (e.g. "workspace.file_path") whose
 	// resolution should be skipped. References to these paths remain unresolved so a
@@ -103,11 +106,12 @@ func ResolveVariableReferencesInLookup() bundle.Mutator {
 // ResolveVolumePathReferencesOnlyResources resolves only references to resources.volumes.*.volume_path.
 func ResolveVolumePathReferencesOnlyResources() bundle.Mutator {
 	return &resolveVariableReferences{
-		prefixes:         []string{"resources"},
-		lookupFn:         lookup,
-		allowPathFn:      isVolumePathReferencePath,
-		extraRounds:      maxResolutionRounds - 1,
-		includeResources: true,
+		prefixes:                 []string{"resources"},
+		lookupFn:                 lookup,
+		allowPathFn:              isVolumePathReferencePath,
+		extraRounds:              maxResolutionRounds - 1,
+		includeResources:         true,
+		validateVolumeSchemaRefs: true,
 	}
 }
 
@@ -169,6 +173,12 @@ func (m *resolveVariableReferences) Apply(ctx context.Context, b *bundle.Bundle)
 	varPath := dyn.NewPath(dyn.Key("var"))
 
 	var diags diag.Diagnostics
+	if m.validateVolumeSchemaRefs {
+		diags = diags.Extend(validateVolumeSchemaReferences(b))
+		if diags.HasError() {
+			return diags
+		}
+	}
 	maxRounds := 1 + m.extraRounds
 
 	for round := range maxRounds {
@@ -201,6 +211,35 @@ func (m *resolveVariableReferences) Apply(ctx context.Context, b *bundle.Bundle)
 	return diags
 }
 
+func validateVolumeSchemaReferences(b *bundle.Bundle) diag.Diagnostics {
+	var diags diag.Diagnostics
+	for _, volumeName := range slices.Sorted(maps.Keys(b.Config.Resources.Volumes)) {
+		volume := b.Config.Resources.Volumes[volumeName]
+		if volume == nil {
+			continue
+		}
+		for _, field := range []struct{ name, value string }{
+			{"schema_name", volume.SchemaName},
+			{"catalog_name", volume.CatalogName},
+		} {
+			fieldPath := structpath.NewPath(nil, "resources", "volumes", volumeName, field.name)
+			_, _ = dynvar.Resolve(dyn.V(field.value), func(path dyn.Path) (dyn.Value, error) {
+				if len(path) > 0 && path[0].Key() == "schemas" {
+					diags = diags.Append(diag.Diagnostic{
+						Severity:  diag.Error,
+						Summary:   fmt.Sprintf("volume %q field %q: reference does not exist: ${%s}", volumeName, field.name, path.String()),
+						Detail:    fmt.Sprintf("did you mean:\n  ${resources.%s}", path.String()),
+						Locations: b.Config.GetLocationsOf(fieldPath),
+						Paths:     []*structpath.PathNode{fieldPath},
+					})
+				}
+				return dyn.InvalidValue, dynvar.ErrSkipResolution
+			})
+		}
+	}
+	return diags
+}
+
 func (m *resolveVariableReferences) resolveOnce(b *bundle.Bundle, prefixes []dyn.Path, varPath dyn.Path) (bool, diag.Diagnostics) {
 	var diags diag.Diagnostics
 	hasUpdates := false

bundle/config/mutator/resolve_variable_references_test.go

@@ -105,6 +105,70 @@ func TestResolveVolumePathReferencesOnlyResources_MissingTarget(t *testing.T) {
 	require.ErrorContains(t, diags.Error(), "reference does not exist: ${resources.volumes.missing.volume_path}")
 }
 
+func TestResolveVolumePathReferencesOnlyResources_RejectsUnprefixedSchemaReferences(t *testing.T) {
+	root, loadDiags := config.LoadFromBytes("databricks.yml", []byte(`resources:
+  schemas:
+    test:
+      name: myschema
+      catalog_name: main
+  volumes:
+    archive:
+      name: archive
+      schema_name: ${schemas.test.name}
+      catalog_name: ${schemas.test.catalog_name}
+    ingest:
+      name: ingest
+      schema_name: ${schemas.test.name}
+      catalog_name: ${schemas.test.catalog_name}
+`))
+	require.NoError(t, loadDiags.Error())
+	b := &bundle.Bundle{Config: *root}
+
+	diags := bundle.Apply(t.Context(), b, ResolveVolumePathReferencesOnlyResources())
+	require.Len(t, diags, 4)
+	require.Error(t, diags.Error())
+	for i, want := range []struct {
+		volume, field, reference, suggestion string
+		line                                 int
+	}{
+		{"archive", "schema_name", "schemas.test.name", "resources.schemas.test.name", 9},
+		{"archive", "catalog_name", "schemas.test.catalog_name", "resources.schemas.test.catalog_name", 10},
+		{"ingest", "schema_name", "schemas.test.name", "resources.schemas.test.name", 13},
+		{"ingest", "catalog_name", "schemas.test.catalog_name", "resources.schemas.test.catalog_name", 14},
+	} {
+		d := diags[i]
+		require.Equal(t, "volume \""+want.volume+"\" field \""+want.field+"\": reference does not exist: ${"+want.reference+"}", d.Summary)
+		require.Equal(t, "did you mean:\n  ${"+want.suggestion+"}", d.Detail)
+		require.Len(t, d.Paths, 1)
+		require.Equal(t, "resources.volumes."+want.volume+"."+want.field, d.Paths[0].String())
+		require.Len(t, d.Locations, 1)
+		require.Equal(t, "databricks.yml", d.Locations[0].File)
+		require.Equal(t, want.line, d.Locations[0].Line)
+	}
+
+	b = &bundle.Bundle{
+		Config: config.Root{
+			Resources: config.Resources{
+				Schemas: map[string]*resources.Schema{"test": {}},
+				Volumes: map[string]*resources.Volume{"foo": {}},
+			},
+		},
+	}
+	schema := b.Config.Resources.Schemas["test"]
+	schema.Name = "myschema"
+	schema.CatalogName = "main"
+	volume := b.Config.Resources.Volumes["foo"]
+	volume.SchemaName = "${resources.schemas.test.name}"
+	volume.CatalogName = "${resources.schemas.test.catalog_name}"
+
+	diags = bundle.Apply(t.Context(), b, ResolveVolumePathReferencesOnlyResources())
+	require.NoError(t, diags.Error())
+	diags = bundle.Apply(t.Context(), b, ResolveVariableReferencesOnlyResources("resources"))
+	require.NoError(t, diags.Error())
+	require.Equal(t, "myschema", b.Config.Resources.Volumes["foo"].SchemaName)
+	require.Equal(t, "main", b.Config.Resources.Volumes["foo"].CatalogName)
+}
+
 func TestIsVolumePathReferencePath(t *testing.T) {
 	require.True(t, isVolumePathReferencePath(dyn.MustPathFromString("resources.volumes.foo.volume_path")))
 	require.False(t, isVolumePathReferencePath(dyn.MustPathFromString("resources.volumes.foo.name")))