diff --git a/NEXT_CHANGELOG.md b/NEXT_CHANGELOG.md
index 46ac539d7d..319aaf6e3f 100644
--- a/NEXT_CHANGELOG.md
+++ b/NEXT_CHANGELOG.md
@@ -8,6 +8,8 @@
 
 ### New Features and Improvements
 
+* Added `cascade_on_destroy` attribute to `databricks_pipeline` to control whether destroying a pipeline also deletes its datasets (materialized views, streaming tables, and views). Defaults to `true`; set to `false` to preserve the datasets on destroy ([#5860](https://github.com/databricks/terraform-provider-databricks/pull/5860)).
+
 ### Bug Fixes
 
 ### Documentation
diff --git a/docs/resources/pipeline.md b/docs/resources/pipeline.md
index 18e593ccb2..c1506da397 100644
--- a/docs/resources/pipeline.md
+++ b/docs/resources/pipeline.md
@@ -99,6 +99,7 @@ The following arguments are supported:
 * `channel` - optional name of the release channel for Spark version used by Lakeflow Declarative Pipeline.  Supported values are: `CURRENT` (default) and `PREVIEW`.
 * `budget_policy_id` - optional string specifying ID of the budget policy for this Lakeflow Declarative Pipeline.
 * `allow_duplicate_names` - Optional boolean flag. If false, deployment will fail if name conflicts with that of another pipeline. default is `false`.
+* `cascade_on_destroy` - Optional boolean flag that controls whether destroying a pipeline also deletes its datasets (materialized views, streaming tables, and views). Defaults to `true`. Set to `false` to keep datasets when the pipeline is destroyed.
 * `deployment` - Deployment type of this pipeline. Supports following attributes:
   * `kind` - The deployment method that manages the pipeline.
   * `metadata_file_path` - The path to the file containing metadata about the deployment.
diff --git a/pipelines/resource_pipeline.go b/pipelines/resource_pipeline.go
index 65ff9ae879..977ead68b1 100644
--- a/pipelines/resource_pipeline.go
+++ b/pipelines/resource_pipeline.go
@@ -48,7 +48,7 @@ func Create(w *databricks.WorkspaceClient, ctx context.Context, d *schema.Resour
 	err = waitForState(w, ctx, id, timeout, pipelines.PipelineStateRunning)
 	if err != nil {
 		log.Printf("[INFO] Pipeline creation failed, attempting to clean up pipeline %s", id)
-		err2 := Delete(w, ctx, id, timeout)
+		err2 := Delete(w, ctx, id, d.Get("cascade_on_destroy").(bool), timeout)
 		if err2 != nil {
 			log.Printf("[WARN] Unable to delete pipeline %s; this resource needs to be manually cleaned up", id)
 			return fmt.Errorf("multiple errors occurred when creating pipeline. Error while waiting for creation: \"%v\"; error while attempting to clean up failed pipeline: \"%v\"", err, err2)
@@ -78,11 +78,13 @@ func Update(w *databricks.WorkspaceClient, ctx context.Context, d *schema.Resour
 	return waitForState(w, ctx, d.Id(), timeout, pipelines.PipelineStateRunning)
 }
 
-func Delete(w *databricks.WorkspaceClient, ctx context.Context, id string, timeout time.Duration) error {
-	err := w.Pipelines.Delete(ctx, pipelines.DeletePipelineRequest{
-		PipelineId: id,
-	})
-	if err != nil {
+func Delete(w *databricks.WorkspaceClient, ctx context.Context, id string, cascade bool, timeout time.Duration) error {
+	req := pipelines.DeletePipelineRequest{PipelineId: id}
+	if !cascade {
+		req.Cascade = false
+		req.ForceSendFields = append(req.ForceSendFields, "Cascade")
+	}
+	if err := w.Pipelines.Delete(ctx, req); err != nil {
 		return err
 	}
 	return retry.RetryContext(ctx, timeout,
@@ -251,6 +253,16 @@ func (Pipeline) CustomizeSchema(s *common.CustomizableSchema) *common.Customizab
 	s.SchemaPath("cluster", "gcp_attributes").RemoveField("use_preemptible_executors")
 	s.SchemaPath("cluster", "gcp_attributes").RemoveField("boot_disk_size")
 
+	// Delete-time only field. Not part of the pipeline spec, so it is added to the
+	// schema directly and read from state during deletion.
+	s.AddNewField("cascade_on_destroy", &schema.Schema{
+		Type:     schema.TypeBool,
+		Optional: true,
+		Default:  true,
+		Description: "Whether destroying a pipeline also deletes its datasets. Defaults to `true`. " +
+			"Set to `false` to keep datasets when the pipeline is destroyed. Only affects the delete operation.",
+	})
+
 	// Default values
 	s.SchemaPath("edition").SetDefault("ADVANCED")
 	s.SchemaPath("channel").SetDefault("CURRENT")
@@ -335,7 +347,7 @@ func ResourcePipeline() common.Resource {
 			if err != nil {
 				return err
 			}
-			return Delete(w, ctx, d.Id(), d.Timeout(schema.TimeoutDelete))
+			return Delete(w, ctx, d.Id(), d.Get("cascade_on_destroy").(bool), d.Timeout(schema.TimeoutDelete))
 
 		},
 		Timeouts: &schema.ResourceTimeout{
diff --git a/pipelines/resource_pipeline_test.go b/pipelines/resource_pipeline_test.go
index 7cd5c79cbc..7a7fd455bd 100644
--- a/pipelines/resource_pipeline_test.go
+++ b/pipelines/resource_pipeline_test.go
@@ -1,9 +1,11 @@
 package pipelines
 
 import (
+	"context"
 	"errors"
 	"testing"
 
+	"github.com/databricks/terraform-provider-databricks/common"
 	"github.com/databricks/terraform-provider-databricks/qa"
 	"github.com/stretchr/testify/assert"
 
@@ -266,6 +268,51 @@ func TestResourcePipelineCreate_ErrorWhenWaitingSuccessfulCleanup(t *testing.T)
 	}.ExpectError(t, "pipeline abcd has failed")
 }
 
+// When creation fails and the user configured cascade_on_destroy = false, the cleanup
+// delete must respect that value (sending Cascade=false via ForceSendFields) so datasets
+// are preserved rather than always cascading.
+func TestResourcePipelineCreate_ErrorWhenWaitingCleanupNoCascade(t *testing.T) {
+	qa.ResourceFixture{
+		MockWorkspaceClientFunc: func(w *mocks.MockWorkspaceClient) {
+			e := w.GetMockPipelinesAPI().EXPECT()
+			e.Create(mock.Anything, mock.Anything).Return(&pipelines.CreatePipelineResponse{
+				PipelineId: "abcd",
+			}, nil)
+
+			e.Get(mock.Anything, pipelines.GetPipelineRequest{
+				PipelineId: "abcd",
+			}).Return(&pipelines.GetPipelineResponse{
+				PipelineId: "abcd",
+				Name:       "test-pipeline",
+				State:      pipelines.PipelineStateFailed,
+			}, nil).Once()
+
+			e.Delete(mock.Anything, pipelines.DeletePipelineRequest{
+				PipelineId:      "abcd",
+				ForceSendFields: []string{"Cascade"},
+			}).Return(nil)
+
+			e.Get(mock.Anything, pipelines.GetPipelineRequest{
+				PipelineId: "abcd",
+			}).Return(nil, apierr.ErrNotFound)
+		},
+		Resource: ResourcePipeline(),
+		HCL: `name = "test"
+		storage = "/test/storage"
+		cascade_on_destroy = false
+		library {
+			notebook {
+				path = "/Test"
+			}
+		}
+		filters {
+			include = ["a"]
+		}
+		`,
+		Create: true,
+	}.ExpectError(t, "pipeline abcd has failed")
+}
+
 func TestResourcePipelineRead(t *testing.T) {
 	qa.ResourceFixture{
 		MockWorkspaceClientFunc: func(w *mocks.MockWorkspaceClient) {
@@ -558,6 +605,29 @@ func TestResourcePipelineDelete_Error(t *testing.T) {
 	assert.Equal(t, "abcd", d.Id())
 }
 
+// When cascade is false, Delete must send Cascade=false explicitly. Cascade is
+// tagged `url:"cascade,omitempty"`, so the false (zero) value would be dropped
+// during query-string encoding unless it is listed in ForceSendFields (honored for
+// query parameters as of databricks-sdk-go v0.152.0). cascade=true relies on the API
+// default and omits the parameter, which is covered by TestResourcePipelineDelete.
+func TestDeletePipelineNoCascade(t *testing.T) {
+	qa.MockWorkspaceApply(t, func(w *mocks.MockWorkspaceClient) {
+		e := w.GetMockPipelinesAPI().EXPECT()
+		e.Delete(mock.Anything, pipelines.DeletePipelineRequest{
+			PipelineId:      "abcd",
+			ForceSendFields: []string{"Cascade"},
+		}).Return(nil)
+		e.Get(mock.Anything, pipelines.GetPipelineRequest{
+			PipelineId: "abcd",
+		}).Return(nil, apierr.ErrNotFound)
+	}, func(ctx context.Context, client *common.DatabricksClient) {
+		w, err := client.WorkspaceClient()
+		require.NoError(t, err)
+		err = Delete(w, ctx, "abcd", false, DefaultTimeout)
+		require.NoError(t, err)
+	})
+}
+
 func TestStorageSuppressDiff(t *testing.T) {
 	k := "storage"
 	generated := "dbfs:/pipelines/c609bbb0-2e42-4bc8-bb4e-a1c26d6e9403"