Add support for schemas in deployment bind/unbind commands (#2406)

## Changes
1. Changed `FindResourceByConfigKey` to return schema resources
2. Implemented the `Exists` method for schema resources

## Why
This PR adds support for schema resources in deployment operations,
enabling users to:
- Bind schemas using `databricks bundle deployment bind <myschema_key>
<schema_full_name>`
- Unbind schemas using `databricks bundle deployment unbind
<myschema_key>`

Where:
- `myschema_key` is a resource key defined in the bundle's .yml file
- `schema_full_name` references an existing schema in the Databricks
workspace

These capabilities allow for more flexible resource management of
schemas within bundles.

## Tests

Added a new integration test that tests bind and unbind methods together
with bundle deployment and destruction.

bundle/config/resources.go

@@ -112,6 +112,12 @@ func (r *Resources) FindResourceByConfigKey(key string) (ConfigResource, error)
 		}
 	}
 
+	for k := range r.Schemas {
+		if k == key {
+			found = append(found, r.Schemas[k])
+		}
+	}
+
 	if len(found) == 0 {
 		return nil, fmt.Errorf("no such resource: %s", key)
 	}

bundle/config/resources/schema.go

@@ -6,6 +6,10 @@ import (
 	"net/url"
 	"strings"
 
+	"github.com/databricks/databricks-sdk-go/apierr"
+
+	"github.com/databricks/cli/libs/log"
+
 	"github.com/databricks/databricks-sdk-go"
 	"github.com/databricks/databricks-sdk-go/marshal"
 	"github.com/databricks/databricks-sdk-go/service/catalog"
@@ -25,8 +29,23 @@ type Schema struct {
 	URL            string         `json:"url,omitempty" bundle:"internal"`
 }
 
-func (s *Schema) Exists(ctx context.Context, w *databricks.WorkspaceClient, id string) (bool, error) {
-	return false, errors.New("schema.Exists() is not supported")
+func (s *Schema) Exists(ctx context.Context, w *databricks.WorkspaceClient, fullName string) (bool, error) {
+	log.Tracef(ctx, "Checking if schema with fullName=%s exists", fullName)
+
+	_, err := w.Schemas.GetByFullName(ctx, fullName)
+	if err != nil {
+		log.Debugf(ctx, "schema with full name %s does not exist: %v", fullName, err)
+
+		var aerr *apierr.APIError
+		if errors.As(err, &aerr) {
+			if aerr.StatusCode == 404 {
+				return false, nil
+			}
+		}
+
+		return false, err
+	}
+	return true, nil
 }
 
 func (s *Schema) TerraformResourceName() string {

bundle/config/resources/schema_test.go

@@ -0,0 +1,26 @@
+package resources
+
+import (
+	"context"
+	"testing"
+
+	"github.com/databricks/databricks-sdk-go/apierr"
+	"github.com/databricks/databricks-sdk-go/experimental/mocks"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+)
+
+func TestSchemaNotFound(t *testing.T) {
+	ctx := context.Background()
+
+	m := mocks.NewMockWorkspaceClient(t)
+	m.GetMockSchemasAPI().On("GetByFullName", mock.Anything, "non-existent-schema").Return(nil, &apierr.APIError{
+		StatusCode: 404,
+	})
+
+	s := &Schema{}
+	exists, err := s.Exists(ctx, m.WorkspaceClient, "non-existent-schema")
+
+	require.Falsef(t, exists, "Exists should return false when getting a 404 response from Workspace")
+	require.NoErrorf(t, err, "Exists should not return an error when getting a 404 response from Workspace")
+}

integration/bundle/bind_resource_test.go

@@ -15,8 +15,62 @@ import (
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+
+	"github.com/databricks/databricks-sdk-go/service/catalog"
 )
 
+func TestBindSchemaToExistingSchema(t *testing.T) {
+	ctx, wt := acc.UcWorkspaceTest(t)
+
+	// create a pre-defined schema:
+	uniqueId := uuid.New().String()
+	predefinedSchema, err := wt.W.Schemas.Create(ctx, catalog.CreateSchema{
+		CatalogName: "main",
+		Name:        "test-schema-" + uniqueId,
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		err := wt.W.Schemas.DeleteByFullName(ctx, predefinedSchema.FullName)
+		require.NoError(t, err)
+	})
+
+	// setup the bundle:
+	bundleRoot := initTestTemplate(t, ctx, "uc_schema_only", map[string]any{
+		"unique_id": uniqueId,
+	})
+	ctx = env.Set(ctx, "BUNDLE_ROOT", bundleRoot)
+
+	// run the bind command:
+	c := testcli.NewRunner(t, ctx, "bundle", "deployment", "bind", "schema1", predefinedSchema.FullName, "--auto-approve")
+	_, _, err = c.Run()
+	require.NoError(t, err)
+
+	// deploy the bundle:
+	deployBundle(t, ctx, bundleRoot)
+
+	// Check that predefinedSchema is updated with config from bundle
+	w, err := databricks.NewWorkspaceClient()
+	require.NoError(t, err)
+
+	updatedSchema, err := w.Schemas.GetByFullName(ctx, predefinedSchema.FullName)
+	require.NoError(t, err)
+	require.Equal(t, updatedSchema.SchemaId, predefinedSchema.SchemaId)
+	require.Equal(t, "This schema was created from DABs", updatedSchema.Comment)
+
+	// unbind the schema:
+	c = testcli.NewRunner(t, ctx, "bundle", "deployment", "unbind", "schema1")
+	_, _, err = c.Run()
+	require.NoError(t, err)
+
+	// destroy the bundle:
+	destroyBundle(t, ctx, bundleRoot)
+
+	// Check that schema is unbound and exists after bundle is destroyed
+	postDestroySchema, err := w.Schemas.GetByFullName(ctx, predefinedSchema.FullName)
+	require.NoError(t, err)
+	require.Equal(t, postDestroySchema.SchemaId, predefinedSchema.SchemaId)
+}
+
 func TestBindJobToExistingJob(t *testing.T) {
 	ctx, wt := acc.WorkspaceTest(t)
 	gt := &generateJobTest{T: wt, w: wt.W}

integration/bundle/bundles/uc_schema_only/databricks_template_schema.json

@@ -0,0 +1,8 @@
+{
+    "properties": {
+        "unique_id": {
+            "type": "string",
+            "description": "Unique ID for the schema name"
+        }
+    }
+}

integration/bundle/bundles/uc_schema_only/template/databricks.yml.tmpl

@@ -0,0 +1,13 @@
+bundle:
+  name: uc-schema-only
+
+workspace:
+  root_path: "~/.bundle/{{.unique_id}}"
+
+resources:
+  schemas:
+    schema1:
+      name: test-schema-{{.unique_id}}
+      catalog_name: main
+      comment: This schema was created from DABs
+