bundle: validate recorded state size during bundle validate

acceptance/bundle/validate/state_size_limit/databricks.yml.tmpl

@@ -0,0 +1,13 @@
+bundle:
+  name: state_size_limit
+
+experimental:
+  record_deployment_history: true
+
+resources:
+  jobs:
+    oversized:
+      name: oversized
+      # DESCRIPTION is generated in 'script' to keep this fixture readable. It is
+      # large enough to push the recorded state past the 64 KiB limit.
+      description: $DESCRIPTION

acceptance/bundle/validate/state_size_limit/output.txt

@@ -0,0 +1,18 @@
+
+>>> [CLI] bundle validate
+Error: resources.jobs.oversized has a serialized state of 70278 bytes, which exceeds the 65536 byte limit for recording deployment history
+  at resources.jobs.oversized
+  in databricks.yml:10:7
+
+Deployment history records the state of each resource, and this resource is too large to record.
+The limit applies per resource, so split this job into multiple jobs, each with fewer tasks.
+
+Name: state_size_limit
+Target: default
+Workspace:
+  User: [USERNAME]
+  Path: /Workspace/Users/[USERNAME]/.bundle/state_size_limit/default
+
+Found 1 error
+
+Exit code: 1

acceptance/bundle/validate/state_size_limit/script

@@ -0,0 +1,7 @@
+# A description over the 64 KiB state limit, generated so the fixture stays small.
+export DESCRIPTION=$(printf 'x%.0s' $(seq 1 70000))
+envsubst < databricks.yml.tmpl > databricks.yml
+
+trace $CLI bundle validate
+
+rm databricks.yml

acceptance/bundle/validate/state_size_limit/test.toml

@@ -0,0 +1,7 @@
+Local = true
+Cloud = false
+
+Ignore = [".databricks"]
+
+[EnvMatrix]
+  DATABRICKS_BUNDLE_ENGINE = ["direct"]

bundle/config/validate/fast_validate.go

@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/config/engine"
 	"github.com/databricks/cli/libs/diag"
 )
 
@@ -14,10 +15,13 @@ import (
 // 2. The validation is blocking for bundle deployments.
 //
 // The full suite of validation mutators is available in the [Validate] mutator.
-type fastValidate struct{ bundle.RO }
+type fastValidate struct {
+	bundle.RO
+	engine engine.EngineType
+}
 
-func FastValidate() bundle.ReadOnlyMutator {
-	return &fastValidate{}
+func FastValidate(e engine.EngineType) bundle.ReadOnlyMutator {
+	return &fastValidate{engine: e}
 }
 
 func (f *fastValidate) Name() string {
@@ -29,6 +33,7 @@ func (f *fastValidate) Apply(ctx context.Context, rb *bundle.Bundle) diag.Diagno
 		// Fast mutators with only in-memory checks
 		JobClusterKeyDefined(),
 		JobTaskClusterSpec(),
+		ValidateStateSize(f.engine),
 
 		// Blocking mutators. Deployments will fail if these checks fail.
 		ValidateArtifactPath(),

bundle/config/validate/state_size.go

@@ -0,0 +1,138 @@
+package validate
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"slices"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/config"
+	"github.com/databricks/cli/bundle/config/engine"
+	"github.com/databricks/cli/bundle/direct/dresources"
+	"github.com/databricks/cli/libs/diag"
+	"github.com/databricks/cli/libs/dyn"
+)
+
+// MaxStateSizeBytes is the largest serialized resource state the deployment
+// metadata service accepts, matching MAX_STATE_BYTES on the service side. A
+// resource above it is rejected there with InvalidArgument, so we check it up
+// front to fail during validate rather than midway through a deploy.
+const MaxStateSizeBytes = 64 * 1024
+
+type validateStateSize struct {
+	bundle.RO
+	engine engine.EngineType
+}
+
+// ValidateStateSize reports resources whose serialized state exceeds
+// MaxStateSizeBytes and so cannot be recorded as deployment history.
+func ValidateStateSize(e engine.EngineType) bundle.ReadOnlyMutator {
+	return &validateStateSize{engine: e}
+}
+
+func (v *validateStateSize) Name() string {
+	return "validate:state_size"
+}
+
+func (v *validateStateSize) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
+	// Only the direct engine records state with the metadata service, and only
+	// when the bundle opted in. Terraform deployments never upload this state.
+	if !v.engine.IsDirect() {
+		return nil
+	}
+	if b.Config.Experimental == nil || !b.Config.Experimental.RecordDeploymentHistory {
+		return nil
+	}
+
+	// The adapters only hold the client, so a nil one is enough to reach
+	// PrepareState. Nothing here issues a request, which keeps this check in
+	// the fast, in-memory half of validation.
+	adapters, err := dresources.InitAll(nil)
+	if err != nil {
+		return diag.FromErr(err)
+	}
+
+	var diags diag.Diagnostics
+	for _, key := range sortedResourceKeys(b) {
+		resourceType := config.GetResourceTypeFromKey(key)
+		adapter, ok := adapters[resourceType]
+		if !ok {
+			// Resource types the direct engine does not support are rejected
+			// during planning, with a better message than we could give here.
+			continue
+		}
+
+		size, err := stateSize(b, adapter, key)
+		if err != nil {
+			return diag.FromErr(err)
+		}
+		if size <= MaxStateSizeBytes {
+			continue
+		}
+
+		diags = diags.Append(diag.Diagnostic{
+			Severity:  diag.Error,
+			Summary:   fmt.Sprintf("%s has a serialized state of %d bytes, which exceeds the %d byte limit for recording deployment history", key, size, MaxStateSizeBytes),
+			Detail:    "Deployment history records the state of each resource, and this resource is too large to record.\n" + sizeAdvice(resourceType),
+			Locations: b.Config.GetLocations(key),
+			Paths:     []dyn.Path{dyn.MustPathFromString(key)},
+		})
+	}
+
+	return diags
+}
+
+// sizeAdvice returns the remediation hint for an oversized resource of the given
+// type. The limit is per resource, so the way out is always to split the
+// definition into smaller ones; the wording names the split that applies.
+//
+// Note that file_path on alerts, dashboards and genie spaces is deliberately not
+// suggested: those mutators inline the file into the resource during initialize,
+// so the recorded state is the same size either way.
+func sizeAdvice(resourceType string) string {
+	switch resourceType {
+	case "jobs":
+		return "The limit applies per resource, so split this job into multiple jobs, each with fewer tasks."
+	case "alerts":
+		return "The limit applies per resource, so split this alert into multiple alerts, each covering fewer conditions."
+	case "pipelines":
+		return "The limit applies per resource, so split this pipeline into multiple pipelines, each with fewer libraries."
+	default:
+		return "The limit applies per resource, so split this resource into multiple smaller resources."
+	}
+}
+
+// stateSize returns the size of the state that would be recorded for the
+// resource at key: the state prepared from config, serialized as JSON.
+func stateSize(b *bundle.Bundle, adapter *dresources.Adapter, key string) (int, error) {
+	inputConfig, err := b.Config.GetResourceConfig(key)
+	if err != nil {
+		return 0, fmt.Errorf("cannot read config for %s: %w", key, err)
+	}
+
+	state, err := adapter.PrepareState(inputConfig)
+	if err != nil {
+		return 0, fmt.Errorf("cannot prepare state for %s: %w", key, err)
+	}
+
+	raw, err := json.Marshal(state)
+	if err != nil {
+		return 0, fmt.Errorf("cannot serialize state for %s: %w", key, err)
+	}
+
+	return len(raw), nil
+}
+
+// sortedResourceKeys returns the bundle's resource keys ("resources.jobs.foo")
+// in a stable order, so diagnostics do not depend on map iteration order.
+func sortedResourceKeys(b *bundle.Bundle) []string {
+	var keys []string
+	for _, group := range b.Config.Resources.AllResources() {
+		for name := range group.Resources {
+			keys = append(keys, "resources."+group.Description.PluralName+"."+name)
+		}
+	}
+	slices.Sort(keys)
+	return keys
+}

bundle/config/validate/state_size_test.go

@@ -0,0 +1,139 @@
+package validate
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/config"
+	"github.com/databricks/cli/bundle/config/engine"
+	"github.com/databricks/cli/bundle/config/resources"
+	"github.com/databricks/cli/bundle/internal/bundletest"
+	"github.com/databricks/cli/libs/diag"
+	"github.com/databricks/cli/libs/dyn"
+	"github.com/databricks/databricks-sdk-go/service/jobs"
+	"github.com/databricks/databricks-sdk-go/service/sql"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// stateSizeBundle returns a bundle with one job whose description is n bytes,
+// which lands in the recorded state and so drives its size.
+func stateSizeBundle(t *testing.T, n int) *bundle.Bundle {
+	t.Helper()
+	b := &bundle.Bundle{
+		Config: config.Root{
+			Experimental: &config.Experimental{RecordDeploymentHistory: true},
+			Resources: config.Resources{
+				Jobs: map[string]*resources.Job{
+					"big": {JobSettings: jobs.JobSettings{Name: "big"}},
+				},
+			},
+		},
+	}
+	description := strings.Repeat("x", n)
+	b.Config.Resources.Jobs["big"].Description = description
+	bundletest.Mutate(t, b, func(v dyn.Value) (dyn.Value, error) {
+		return dyn.Set(v, "resources.jobs.big.description", dyn.V(description))
+	})
+	return b
+}
+
+func TestValidateStateSizeUnderLimit(t *testing.T) {
+	b := stateSizeBundle(t, 1024)
+
+	diags := ValidateStateSize(engine.EngineDirect).Apply(t.Context(), b)
+	require.Empty(t, diags)
+}
+
+func TestValidateStateSizeOverLimit(t *testing.T) {
+	b := stateSizeBundle(t, MaxStateSizeBytes+1)
+
+	diags := ValidateStateSize(engine.EngineDirect).Apply(t.Context(), b)
+	require.Len(t, diags, 1)
+	assert.Equal(t, diag.Error, diags[0].Severity)
+	assert.Contains(t, diags[0].Summary, "resources.jobs.big")
+	assert.Contains(t, diags[0].Summary, "exceeds the 65536 byte limit")
+	assert.Equal(t, []dyn.Path{dyn.MustPathFromString("resources.jobs.big")}, diags[0].Paths)
+	assert.Contains(t, diags[0].Detail, "split this job into multiple jobs")
+}
+
+// The advice names the split that applies to the resource type, since "split it
+// up" means a different edit for a job than for an alert.
+func TestSizeAdvice(t *testing.T) {
+	assert.Contains(t, sizeAdvice("jobs"), "split this job into multiple jobs")
+	assert.Contains(t, sizeAdvice("alerts"), "split this alert into multiple alerts")
+	assert.Contains(t, sizeAdvice("pipelines"), "split this pipeline into multiple pipelines")
+	assert.Contains(t, sizeAdvice("schemas"), "split this resource into multiple smaller resources")
+}
+
+// The state is only uploaded by the direct engine, so a terraform deployment of
+// the same oversized resource must not be blocked.
+func TestValidateStateSizeSkippedForTerraform(t *testing.T) {
+	b := stateSizeBundle(t, MaxStateSizeBytes+1)
+
+	diags := ValidateStateSize(engine.EngineTerraform).Apply(t.Context(), b)
+	require.Empty(t, diags)
+}
+
+func TestValidateStateSizeSkippedWithoutOptIn(t *testing.T) {
+	b := stateSizeBundle(t, MaxStateSizeBytes+1)
+	b.Config.Experimental.RecordDeploymentHistory = false
+
+	diags := ValidateStateSize(engine.EngineDirect).Apply(t.Context(), b)
+	require.Empty(t, diags)
+}
+
+func TestValidateStateSizeNoExperimentalBlock(t *testing.T) {
+	b := stateSizeBundle(t, MaxStateSizeBytes+1)
+	b.Config.Experimental = nil
+
+	diags := ValidateStateSize(engine.EngineDirect).Apply(t.Context(), b)
+	require.Empty(t, diags)
+}
+
+func TestValidateStateSizeAlertAdvice(t *testing.T) {
+	queryText := strings.Repeat("s", MaxStateSizeBytes+1)
+	b := &bundle.Bundle{
+		Config: config.Root{
+			Experimental: &config.Experimental{RecordDeploymentHistory: true},
+			Resources: config.Resources{
+				Alerts: map[string]*resources.Alert{
+					"noisy": {AlertV2: sql.AlertV2{QueryText: queryText}},
+				},
+			},
+		},
+	}
+	bundletest.Mutate(t, b, func(v dyn.Value) (dyn.Value, error) {
+		return dyn.Set(v, "resources.alerts.noisy", dyn.V(map[string]dyn.Value{
+			"query_text": dyn.V(queryText),
+		}))
+	})
+
+	diags := ValidateStateSize(engine.EngineDirect).Apply(t.Context(), b)
+	require.Len(t, diags, 1)
+	assert.Contains(t, diags[0].Summary, "resources.alerts.noisy")
+	assert.Contains(t, diags[0].Detail, "split this alert into multiple alerts")
+}
+
+// Every oversized resource is reported, so one deploy attempt surfaces them all.
+func TestValidateStateSizeReportsEveryResource(t *testing.T) {
+	b := stateSizeBundle(t, MaxStateSizeBytes+1)
+	description := strings.Repeat("y", MaxStateSizeBytes+1)
+	b.Config.Resources.Jobs["also_big"] = &resources.Job{
+		JobSettings: jobs.JobSettings{Name: "also_big", Description: description},
+	}
+	bundletest.Mutate(t, b, func(v dyn.Value) (dyn.Value, error) {
+		return dyn.Set(v, "resources.jobs.also_big", dyn.V(map[string]dyn.Value{
+			"name":        dyn.V("also_big"),
+			"description": dyn.V(description),
+		}))
+	})
+
+	diags := ValidateStateSize(engine.EngineDirect).Apply(t.Context(), b)
+	require.Len(t, diags, 2)
+	// Diagnostics are ordered by resource key so output does not depend on map
+	// iteration order.
+	assert.Contains(t, diags[0].Summary, "resources.jobs.also_big")
+	assert.Contains(t, diags[1].Summary, "resources.jobs.big")
+}

bundle/config/validate/validate.go

@@ -4,11 +4,12 @@ import (
 	"context"
 
 	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/config/engine"
 )
 
-func Validate(ctx context.Context, b *bundle.Bundle) {
+func Validate(ctx context.Context, b *bundle.Bundle, e engine.EngineType) {
 	bundle.ApplyParallel(ctx, b,
-		FastValidate(),
+		FastValidate(e),
 
 		// Slow mutators that require network or file i/o. These are only
 		// run in the `bundle validate` command.

cmd/bundle/utils/process.go

@@ -283,7 +283,7 @@ func ProcessBundleRet(cmd *cobra.Command, opts ProcessOptions) (b *bundle.Bundle
 
 	if opts.FastValidate {
 		t1 := time.Now()
-		bundle.ApplyContext(ctx, b, validate.FastValidate())
+		bundle.ApplyContext(ctx, b, validate.FastValidate(requiredEngine.Type))
 		b.Metrics.ExecutionTimes = append(b.Metrics.ExecutionTimes, protos.IntMapEntry{
 			Key:   "validate.FastValidate",
 			Value: time.Since(t1).Milliseconds(),
@@ -303,7 +303,7 @@ func ProcessBundleRet(cmd *cobra.Command, opts ProcessOptions) (b *bundle.Bundle
 	}
 
 	if opts.Validate {
-		validate.Validate(ctx, b)
+		validate.Validate(ctx, b, requiredEngine.Type)
 		if logdiag.HasError(ctx) {
 			return b, stateDesc, root.ErrAlreadyPrinted
 		}