Allow configuring cascade deletion in DABs

NEXT_CHANGELOG.md

@@ -7,6 +7,7 @@
 ### CLI
 
 ### Bundles
+* Added a `cascade` field to the pipeline resource to control whether destroying a pipeline also cascades to its datasets. When unset, the server default applies; set `cascade: false` to retain the datasets on destroy.
 
 ### Dependency updates
 

acceptance/bundle/refschema/out.fields.txt

@@ -2273,6 +2273,7 @@ resources.models.*.permissions[*].service_principal_name	string	ALL
 resources.models.*.permissions[*].user_name	string	ALL
 resources.pipelines.*.allow_duplicate_names	bool	ALL
 resources.pipelines.*.budget_policy_id	string	ALL
+resources.pipelines.*.cascade	*bool	ALL
 resources.pipelines.*.catalog	string	ALL
 resources.pipelines.*.cause	string	REMOTE
 resources.pipelines.*.channel	string	ALL

acceptance/pipelines/destroy/destroy-pipeline-cascade/databricks.yml

@@ -0,0 +1,7 @@
+bundle:
+  name: test-pipeline-destroy-cascade
+resources:
+  pipelines:
+    my_pipeline:
+      name: test-pipeline-cascade
+      cascade: false

acceptance/pipelines/destroy/destroy-pipeline-cascade/output.txt

@@ -0,0 +1,56 @@
+
+=== Deploy a pipeline configured with cascade: false
+>>> [CLI] pipelines deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-pipeline-destroy-cascade/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+View your pipeline my_pipeline here: [DATABRICKS_URL]/pipelines/[UUID]?w=[NUMID]
+
+>>> print_requests.py //api/2.0/pipelines
+{
+  "method": "POST",
+  "path": "/api/2.0/pipelines",
+  "body": {
+    "channel": "CURRENT",
+    "deployment": {
+      "kind": "BUNDLE",
+      "metadata_file_path": "/Workspace/Users/[USERNAME]/.bundle/test-pipeline-destroy-cascade/default/state/metadata.json"
+    },
+    "edition": "ADVANCED",
+    "name": "test-pipeline-cascade"
+  }
+}
+
+=== Change only cascade (false -> true): expect a state-only update, no pipeline update API call
+>>> [CLI] pipelines deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-pipeline-destroy-cascade/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+View your pipeline my_pipeline here: [DATABRICKS_URL]/pipelines/[UUID]?w=[NUMID]
+
+>>> print_requests.py //api/2.0/pipelines
+
+=== Destroy: delete reads the updated cascade=true from persisted state
+>>> [CLI] pipelines destroy --auto-approve
+The following resources will be deleted:
+  delete resources.pipelines.my_pipeline
+
+This action will result in the deletion of the following Lakeflow Spark Declarative Pipelines along with the
+Streaming Tables (STs) and Materialized Views (MVs) managed by them:
+  delete resources.pipelines.my_pipeline
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/test-pipeline-destroy-cascade/default
+
+Deleting files...
+Destroy complete!
+
+>>> print_requests.py //api/2.0/pipelines
+{
+  "method": "DELETE",
+  "path": "/api/2.0/pipelines/[UUID]",
+  "q": {
+    "cascade": "true"
+  }
+}

acceptance/pipelines/destroy/destroy-pipeline-cascade/script

@@ -0,0 +1,15 @@
+title "Deploy a pipeline configured with cascade: false"
+trace $CLI pipelines deploy
+
+# Flush the deploy requests: cascade is delete-time-only, so the create request must not carry it.
+trace print_requests.py //api/2.0/pipelines
+
+title "Change only cascade (false -> true): expect a state-only update, no pipeline update API call"
+update_file.py databricks.yml "cascade: false" "cascade: true"
+trace $CLI pipelines deploy
+# No POST/PATCH to /pipelines is expected here; the cascade-only change is persisted to state.
+trace print_requests.py //api/2.0/pipelines
+
+title "Destroy: delete reads the updated cascade=true from persisted state"
+trace $CLI pipelines destroy --auto-approve
+trace print_requests.py //api/2.0/pipelines

acceptance/pipelines/destroy/destroy-pipeline-cascade/test.toml

@@ -0,0 +1,5 @@
+# --cascade only affects the direct engine's delete request; the terraform engine deletes
+# pipelines through the provider and ignores it. Restrict to direct so the recorded request
+# is meaningful and the single variant does not diverge.
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+RecordRequests = true

bundle/config/resources/pipeline.go

@@ -16,6 +16,10 @@ type Pipeline struct {
 	pipelines.CreatePipeline //nolint CreatePipeline also defines Id field with the same json tag "id"
 
 	Permissions []PipelinePermission `json:"permissions,omitempty"`
+
+	// Optional parameter that controls whether destroying the pipeline also cascades to datasets.
+	// By default, the server will cascade the pipeline deletion to its datasets.
+	Cascade *bool `json:"cascade,omitempty"`
 }
 
 func (p *Pipeline) UnmarshalJSON(b []byte) error {

bundle/deploy/terraform/tfdyn/convert_pipeline.go

@@ -21,7 +21,8 @@ func convertPipelineResource(ctx context.Context, vin dyn.Value) (dyn.Value, err
 		return dyn.InvalidValue, err
 	}
 
-	vout, err = dyn.DropKeys(vout, []string{"dry_run"})
+	// Current Terraform provider does not support the cascade attribute yet
+	vout, err = dyn.DropKeys(vout, []string{"dry_run", "cascade"})
 	if err != nil {
 		return dyn.InvalidValue, err
 	}

bundle/direct/dresources/pipeline.go

@@ -10,11 +10,32 @@ import (
 	"github.com/databricks/databricks-sdk-go/service/pipelines"
 )
 
+// PipelineState is the state type for Pipeline resources. It extends CreatePipeline with
+// the Cascade field, a delete-time setting that is not part of the pipeline spec
+type PipelineState struct {
+	pipelines.CreatePipeline
+
+	// Cascade controls whether deleting the pipeline also deletes its datasets (MVs, STs,
+	// Views). Nil means the server default (cascade) applies. Read from persisted state at
+	// delete time; never sent on create/update.
+	Cascade *bool `json:"cascade,omitempty"`
+}
+
+func (s *PipelineState) UnmarshalJSON(b []byte) error {
+	return marshal.Unmarshal(b, s)
+}
+
+func (s PipelineState) MarshalJSON() ([]byte, error) {
+	return marshal.Marshal(s)
+}
+
 // PipelineRemote is the return type for DoRead. It embeds CreatePipeline so that all
 // paths in StateType are valid paths in RemoteType.
 type PipelineRemote struct {
 	pipelines.CreatePipeline
 
+	Cascade *bool `json:"cascade,omitempty"`
+
 	// Remote-specific fields from pipelines.GetPipelineResponse
 	Cause                   string                              `json:"cause,omitempty"`
 	ClusterId               string                              `json:"cluster_id,omitempty"`
@@ -49,12 +70,18 @@ func (*ResourcePipeline) New(client *databricks.WorkspaceClient) *ResourcePipeli
 	}
 }
 
-func (*ResourcePipeline) PrepareState(input *resources.Pipeline) *pipelines.CreatePipeline {
-	return &input.CreatePipeline
+func (*ResourcePipeline) PrepareState(input *resources.Pipeline) *PipelineState {
+	return &PipelineState{
+		CreatePipeline: input.CreatePipeline,
+		Cascade:        input.Cascade,
+	}
 }
 
-func (*ResourcePipeline) RemapState(remote *PipelineRemote) *pipelines.CreatePipeline {
-	return &remote.CreatePipeline
+func (*ResourcePipeline) RemapState(remote *PipelineRemote) *PipelineState {
+	return &PipelineState{
+		CreatePipeline: remote.CreatePipeline,
+		Cascade:        remote.Cascade,
+	}
 }
 
 func (r *ResourcePipeline) DoRead(ctx context.Context, id string) (*PipelineRemote, error) {
@@ -109,7 +136,9 @@ func makePipelineRemote(p *pipelines.GetPipelineResponse) *PipelineRemote {
 		}
 	}
 	return &PipelineRemote{
-		CreatePipeline:          createPipeline,
+		CreatePipeline: createPipeline,
+		// cascade is input-only; the GET response never carries it, so leave it nil.
+		Cascade:                 nil,
 		Cause:                   p.Cause,
 		ClusterId:               p.ClusterId,
 		CreatorUserName:         p.CreatorUserName,
@@ -124,15 +153,22 @@ func makePipelineRemote(p *pipelines.GetPipelineResponse) *PipelineRemote {
 	}
 }
 
-func (r *ResourcePipeline) DoCreate(ctx context.Context, config *pipelines.CreatePipeline) (string, *PipelineRemote, error) {
-	response, err := r.client.Pipelines.Create(ctx, *config)
+func (r *ResourcePipeline) DoCreate(ctx context.Context, config *PipelineState) (string, *PipelineRemote, error) {
+	response, err := r.client.Pipelines.Create(ctx, config.CreatePipeline)
 	if err != nil {
 		return "", nil, err
 	}
 	return response.PipelineId, nil, nil
 }
 
-func (r *ResourcePipeline) DoUpdate(ctx context.Context, id string, config *pipelines.CreatePipeline, _ *PlanEntry) (*PipelineRemote, error) {
+func (r *ResourcePipeline) DoUpdate(ctx context.Context, id string, config *PipelineState, entry *PlanEntry) (*PipelineRemote, error) {
+	// cascade is a delete-time-only setting with no update API, so a change to it alone must
+	// persist to state without a pipeline Update call. This mirrors sql_warehouse's handling
+	// of its non-spec lifecycle field.
+	if !entry.Changes.HasChangeExcept("cascade") {
+		return nil, nil
+	}
+
 	request := pipelines.EditPipeline{
 		AllowDuplicateNames:  config.AllowDuplicateNames,
 		BudgetPolicyId:       config.BudgetPolicyId,
@@ -174,8 +210,17 @@ func (r *ResourcePipeline) DoUpdate(ctx context.Context, id string, config *pipe
 	return nil, r.client.Pipelines.Update(ctx, request)
 }
 
-func (r *ResourcePipeline) DoDelete(ctx context.Context, id string, _ *pipelines.CreatePipeline) error {
-	return r.client.Pipelines.DeleteByPipelineId(ctx, id)
+func (r *ResourcePipeline) DoDelete(ctx context.Context, id string, state *PipelineState) error {
+	if state.Cascade == nil {
+		// No explicit cascade in config: preserve the backend default (cascade).
+		return r.client.Pipelines.DeleteByPipelineId(ctx, id)
+	}
+	return r.client.Pipelines.Delete(ctx, pipelines.DeletePipelineRequest{
+		PipelineId:      id,
+		Cascade:         *state.Cascade,
+		Force:           false,
+		ForceSendFields: []string{"Cascade"},
+	})
 }
 
 // Note, terraform provider either

bundle/direct/dresources/resources.yml

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

bundle/internal/schema/annotations.yml

@@ -1229,6 +1229,9 @@ resources:
                       path: ./pipeline.py
           ```
       "$fields":
+        "cascade":
+          "description": |-
+            Whether destroying the pipeline also cascades to its datasets. Defaults to true. Set to false to keep the datasets when the pipeline is deleted. Only affects the delete operation.
         "ingestion_definition":
           "$fields":
             "objects":