Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions NEXT_CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
1 change: 1 addition & 0 deletions docs/resources/pipeline.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
26 changes: 19 additions & 7 deletions pipelines/resource_pipeline.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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{
Expand Down
70 changes: 70 additions & 0 deletions pipelines/resource_pipeline_test.go
Original file line number Diff line number Diff line change
@@ -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"

Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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"
Expand Down
Loading