Address review feedback on cascade_on_destroy

- Add docs URL to the terraform-engine validation error so users can
  learn about direct deployment mode.
- Omit CascadeOnDestroy from PipelineRemote: it is an input-only,
  delete-time setting the GET response never returns, so drift is
  suppressed automatically (missing_in_remote). Matches the postgres
  input-only field pattern.
- Propagate errors from pipelineDeletionCascades instead of silently
  defaulting, and add unit coverage.

Co-authored-by: Isaac

Author: ronaldz-db

acceptance/pipelines/destroy/cascade-on-destroy-terraform-error/output.txt

@@ -4,5 +4,8 @@
 Error: cascade_on_destroy is only supported in direct deployment mode
   in databricks.yml:7:27
 
+cascade_on_destroy is not yet supported by the terraform provider. Set the DATABRICKS_BUNDLE_ENGINE environment variable to 'direct' or set 'bundle.engine: direct' in your databricks.yml to use it.
+Learn more at https://docs.databricks.com/dev-tools/bundles/direct
+
 
 Exit code: 1

bundle/config/mutator/validate_cascade_on_destroy.go

@@ -36,8 +36,11 @@ func (m *validateCascadeOnDestroy) Apply(_ context.Context, b *bundle.Bundle) di
 		}
 		path := "resources.pipelines." + key + ".cascade_on_destroy"
 		diags = diags.Append(diag.Diagnostic{
-			Severity:  diag.Error,
-			Summary:   "cascade_on_destroy is only supported in direct deployment mode",
+			Severity: diag.Error,
+			Summary:  "cascade_on_destroy is only supported in direct deployment mode",
+			Detail: "cascade_on_destroy is not yet supported by the terraform provider. " +
+				"Set the DATABRICKS_BUNDLE_ENGINE environment variable to 'direct' or set 'bundle.engine: direct' in your databricks.yml to use it.\n" +
+				"Learn more at https://docs.databricks.com/dev-tools/bundles/direct",
 			Locations: b.Config.GetLocations(path),
 		})
 	}

bundle/config/mutator/validate_cascade_on_destroy_test.go

@@ -65,4 +65,5 @@ func TestValidateCascadeOnDestroyTerraformEngineSetEmitsError(t *testing.T) {
 	diags := bundle.Apply(t.Context(), b, mutator.ValidateCascadeOnDestroy(engine.EngineTerraform))
 	assert.Len(t, diags, 1)
 	assert.Equal(t, "cascade_on_destroy is only supported in direct deployment mode", diags[0].Summary)
+	assert.Contains(t, diags[0].Detail, "https://docs.databricks.com/dev-tools/bundles/direct")
 }

bundle/direct/dresources/pipeline.go

@@ -31,11 +31,11 @@ func (s PipelineState) MarshalJSON() ([]byte, error) {
 
 // PipelineRemote is the return type for DoRead. It embeds CreatePipeline so that all
 // paths in StateType are valid paths in RemoteType.
+// Note that cascade_on_destroy is intentionally absent: it is a delete-time-only setting
+// that the GET response never returns, so the engine suppresses its drift automatically
 type PipelineRemote struct {
 	pipelines.CreatePipeline
 
-	CascadeOnDestroy *bool `json:"cascade_on_destroy,omitempty"`
-
 	// Remote-specific fields from pipelines.GetPipelineResponse
 	Cause                   string                              `json:"cause,omitempty"`
 	ClusterId               string                              `json:"cluster_id,omitempty"`
@@ -79,8 +79,9 @@ func (*ResourcePipeline) PrepareState(input *resources.Pipeline) *PipelineState
 
 func (*ResourcePipeline) RemapState(remote *PipelineRemote) *PipelineState {
 	return &PipelineState{
-		CreatePipeline:   remote.CreatePipeline,
-		CascadeOnDestroy: remote.CascadeOnDestroy,
+		CreatePipeline: remote.CreatePipeline,
+		// cascade_on_destroy is input-only and absent from PipelineRemote, so it stays nil here.
+		CascadeOnDestroy: nil,
 	}
 }
 
@@ -136,9 +137,7 @@ func makePipelineRemote(p *pipelines.GetPipelineResponse) *PipelineRemote {
 		}
 	}
 	return &PipelineRemote{
-		CreatePipeline: createPipeline,
-		// cascade_on_destroy is input-only; the GET response never carries it, so leave it nil.
-		CascadeOnDestroy:        nil,
+		CreatePipeline:          createPipeline,
 		Cause:                   p.Cause,
 		ClusterId:               p.ClusterId,
 		CreatorUserName:         p.CreatorUserName,

bundle/direct/dresources/resources.yml

@@ -180,10 +180,6 @@ resources:
         reason: "!drop"
       - field: run_as
         reason: input_only
-      # cascade_on_destroy is a delete-time-only setting
-      # it is never returned by GET, so suppress remote drift for it.
-      - field: cascade_on_destroy
-        reason: input_only
 
     ignore_local_changes:
       - field: deployment.version_id

bundle/direct/dresources/type_test.go

@@ -25,6 +25,11 @@ var knownMissingInRemoteType = map[string][]string{
 	"model_serving_endpoints": {
 		"rate_limits",
 	},
+	"pipelines": {
+		// Note that this is a deliberate omission: cascade_on_destroy is a client-side-only field.
+		// It has no remote counterpart, so it is never expected in RemoteType.
+		"cascade_on_destroy",
+	},
 	"quality_monitors": {
 		"skip_builtin_dashboard",
 		"warehouse_id",

bundle/phases/destroy.go

@@ -48,12 +48,16 @@ var destroyApprovalGroups = []approvalGroup{
 
 // logPipelineDeleteApproval prints the pipeline deletions. If cascade_on_destroy is true, we will include
 // a note that datasets will be deleted as well.
-func logPipelineDeleteApproval(ctx context.Context, b *bundle.Bundle, actions []deployplan.Action) {
+func logPipelineDeleteApproval(ctx context.Context, b *bundle.Bundle, actions []deployplan.Action) error {
 	pipelineDeletes := filterGroup(actions, "pipelines", deployplan.Delete)
 
 	var cascading, retaining []deployplan.Action
 	for _, a := range pipelineDeletes {
-		if pipelineDeletionCascades(b, a) {
+		cascade, err := pipelineDeletionCascades(b, a)
+		if err != nil {
+			return err
+		}
+		if cascade {
 			cascading = append(cascading, a)
 		} else {
 			retaining = append(retaining, a)
@@ -76,6 +80,7 @@ func logPipelineDeleteApproval(ctx context.Context, b *bundle.Bundle, actions []
 		}
 		cmdio.LogString(ctx, "")
 	}
+	return nil
 }
 
 func approvalForDestroy(ctx context.Context, b *bundle.Bundle, plan *deployplan.Plan) (bool, error) {
@@ -102,7 +107,9 @@ func approvalForDestroy(ctx context.Context, b *bundle.Bundle, plan *deployplan.
 	}
 
 	logApprovalGroups(ctx, deleteActions, destroyApprovalGroups, true, deployplan.Delete)
-	logPipelineDeleteApproval(ctx, b, deleteActions)
+	if err := logPipelineDeleteApproval(ctx, b, deleteActions); err != nil {
+		return false, err
+	}
 
 	cmdio.LogString(ctx, "All files and directories at the following location will be deleted: "+b.Config.Workspace.RootPath)
 	cmdio.LogString(ctx, "")

bundle/phases/plan.go

@@ -36,19 +36,22 @@ func PreDeployChecks(ctx context.Context, b *bundle.Bundle, isPlan bool, engine
 // cascade_on_destroy is explicitly set to false. A lookup miss (unset field, or a non-pipeline
 // resource) means the default applies, so it returns true.
 // Reads from config like checkForPreventDestroy.
-func pipelineDeletionCascades(b *bundle.Bundle, action deployplan.Action) bool {
+func pipelineDeletionCascades(b *bundle.Bundle, action deployplan.Action) (bool, error) {
 	path, err := dyn.NewPathFromString(action.ResourceKey)
 	if err != nil {
-		return true
+		return false, fmt.Errorf("failed to parse %q", action.ResourceKey)
 	}
 	path = append(path, dyn.Key("cascade_on_destroy"))
 
 	v, err := dyn.GetByPath(b.Config.Value(), path)
 	if err != nil {
-		return true
+		return true, nil
 	}
 	cascade, ok := v.AsBool()
-	return !ok || cascade
+	if !ok {
+		return false, fmt.Errorf("internal error: cascade_on_destroy is not a boolean for %s", action.ResourceKey)
+	}
+	return cascade, nil
 }
 
 // checkForPreventDestroy checks if the resource has lifecycle.prevent_destroy set, but the plan calls for this resource to be recreated or destroyed.

bundle/phases/plan_test.go

@@ -7,9 +7,75 @@ import (
 	"github.com/databricks/cli/bundle/config"
 	"github.com/databricks/cli/bundle/deployplan"
 	"github.com/databricks/cli/libs/dyn"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
+func TestPipelineDeletionCascades(t *testing.T) {
+	b := &bundle.Bundle{}
+	err := b.Config.Mutate(func(v dyn.Value) (dyn.Value, error) {
+		return dyn.Set(v, "resources", dyn.NewValue(map[string]dyn.Value{
+			"pipelines": dyn.NewValue(map[string]dyn.Value{
+				"cascade_true": dyn.NewValue(map[string]dyn.Value{
+					"cascade_on_destroy": dyn.NewValue(true, nil),
+				}, nil),
+				"cascade_false": dyn.NewValue(map[string]dyn.Value{
+					"cascade_on_destroy": dyn.NewValue(false, nil),
+				}, nil),
+				"cascade_unset": dyn.NewValue(map[string]dyn.Value{}, nil),
+				"cascade_not_bool": dyn.NewValue(map[string]dyn.Value{
+					"cascade_on_destroy": dyn.NewValue("yes", nil),
+				}, nil),
+			}, nil),
+		}, nil))
+	})
+	require.NoError(t, err)
+
+	tests := []struct {
+		name        string
+		resourceKey string
+		want        bool
+		wantErr     string
+	}{
+		{
+			name:        "cascade_on_destroy explicitly true",
+			resourceKey: "resources.pipelines.cascade_true",
+			want:        true,
+		},
+		{
+			name:        "cascade_on_destroy explicitly false",
+			resourceKey: "resources.pipelines.cascade_false",
+			want:        false,
+		},
+		{
+			name:        "cascade_on_destroy unset defaults to cascade",
+			resourceKey: "resources.pipelines.cascade_unset",
+			want:        true,
+		},
+		{
+			name:        "cascade_on_destroy not a bool errors",
+			resourceKey: "resources.pipelines.cascade_not_bool",
+			wantErr:     "internal error: cascade_on_destroy is not a boolean for resources.pipelines.cascade_not_bool",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			action := deployplan.Action{
+				ResourceKey: tt.resourceKey,
+				ActionType:  deployplan.Delete,
+			}
+			cascade, err := pipelineDeletionCascades(b, action)
+			if tt.wantErr != "" {
+				require.EqualError(t, err, tt.wantErr)
+				return
+			}
+			require.NoError(t, err)
+			assert.Equal(t, tt.want, cascade)
+		})
+	}
+}
+
 func TestCheckPreventDestroyForAllResources(t *testing.T) {
 	for resourceType := range config.SupportedResources() {
 		t.Run(resourceType, func(t *testing.T) {