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
1 change: 1 addition & 0 deletions .nextchanges/cli/pipeline-refresh-by-file.md
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
* Support source files in `pipelines run --refresh` and `--full-refresh` selectors. ([#7184](https://github.com/databricks/cli/pull/7184))
102 changes: 98 additions & 4 deletions acceptance/pipelines/run/refresh-flags/output.txt
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ Update ID: [UUID]
Update for pipeline completed successfully.

Pipeline configurations for this update:
• All tables are refreshed
• Refreshed [table1, table2]

>>> print_requests.py //pipelines
{
Expand All @@ -46,7 +46,7 @@ Update ID: [UUID]
Update for pipeline completed successfully.

Pipeline configurations for this update:
• All tables are refreshed
• All tables are fully refreshed

>>> print_requests.py //pipelines
{
Expand All @@ -70,7 +70,7 @@ Update ID: [UUID]
Update for pipeline completed successfully.

Pipeline configurations for this update:
• All tables are refreshed
• Full refreshed [table1, table2]

>>> print_requests.py //pipelines
{
Expand All @@ -97,7 +97,8 @@ Update ID: [UUID]
Update for pipeline completed successfully.

Pipeline configurations for this update:
• All tables are refreshed
• Refreshed [table3, table4]
• Full refreshed [table1, table2]

>>> print_requests.py //pipelines
{
Expand All @@ -114,3 +115,96 @@ Pipeline configurations for this update:
]
}
}

=== Refreshing tables defined in a source file
>>> [CLI] pipelines run --refresh pipeline_file.py
Recommendation: This command runs the last deployed version of the code

If you've made local changes, run 'databricks pipelines deploy' first to ensure they are included.

Update URL: [DATABRICKS_URL]/#joblist/pipelines/[UUID]/updates/[UUID]

Update ID: [UUID]

Update for pipeline completed successfully.

Pipeline configurations for this update:
• Refreshed tables defined in [/Workspace/Users/[USERNAME]/.bundle/test-pipeline-run-flags/default/files/pipeline_file.py]

>>> print_requests.py //pipelines
{
"method": "POST",
"path": "/api/2.0/pipelines/[UUID]/updates",
"body": {
"refresh_selection_by_path": [
"/Workspace/Users/[USERNAME]/.bundle/test-pipeline-run-flags/default/files/pipeline_file.py"
]
}
}

=== Fully refreshing tables defined in a source file
>>> [CLI] pipelines run --full-refresh pipeline_file.py
Recommendation: This command runs the last deployed version of the code

If you've made local changes, run 'databricks pipelines deploy' first to ensure they are included.

Update URL: [DATABRICKS_URL]/#joblist/pipelines/[UUID]/updates/[UUID]

Update ID: [UUID]

Update for pipeline completed successfully.

Pipeline configurations for this update:
• Full refreshed tables defined in [/Workspace/Users/[USERNAME]/.bundle/test-pipeline-run-flags/default/files/pipeline_file.py]

>>> print_requests.py //pipelines
{
"method": "POST",
"path": "/api/2.0/pipelines/[UUID]/updates",
"body": {
"full_refresh_selection_by_path": [
"/Workspace/Users/[USERNAME]/.bundle/test-pipeline-run-flags/default/files/pipeline_file.py"
]
}
}

=== Rejecting missing sources and incompatible refresh modes
>>> musterr [CLI] pipelines run --refresh ./missing.sql --no-wait
Recommendation: This command runs the last deployed version of the code

If you've made local changes, run 'databricks pipelines deploy' first to ensure they are included.

Error: invalid --refresh: source file "./missing.sql" not found; use a table name or an existing source file

>>> musterr [CLI] pipelines run --refresh pipeline_file.py --full-refresh-all --no-wait
Recommendation: This command runs the last deployed version of the code

If you've made local changes, run 'databricks pipelines deploy' first to ensure they are included.

Error: cannot use --refresh or --full-refresh together with --full-refresh-all, these flags are mutually exclusive

>>> musterr [CLI] pipelines run --full-refresh ./missing.sql --restart --no-wait
Recommendation: This command runs the last deployed version of the code

If you've made local changes, run 'databricks pipelines deploy' first to ensure they are included.

Error: invalid --full-refresh: source file "./missing.sql" not found; use a table name or an existing source file

>>> test ! -e [OUTPUT_DIR]/out.requests.txt

=== Rejecting mixed selectors before restarting the pipeline
>>> musterr [CLI] pipelines run --refresh table1,pipeline_file.py --restart --no-wait
Recommendation: This command runs the last deployed version of the code

If you've made local changes, run 'databricks pipelines deploy' first to ensure they are included.

Error: cannot mix table names and source files in --refresh; use only table names or only source files

>>> musterr [CLI] pipelines run --full-refresh table2,pipeline_file.py --restart --no-wait
Recommendation: This command runs the last deployed version of the code

If you've made local changes, run 'databricks pipelines deploy' first to ensure they are included.

Error: cannot mix table names and source files in --full-refresh; use only table names or only source files

>>> test ! -e [OUTPUT_DIR]/out.requests.txt
19 changes: 19 additions & 0 deletions acceptance/pipelines/run/refresh-flags/script
Original file line number Diff line number Diff line change
Expand Up @@ -16,3 +16,22 @@ trace print_requests.py //pipelines
title "Running pipeline with --full-refresh flag and --refresh flag"
trace $CLI pipelines run --full-refresh table1,table2 --refresh table3,table4
trace print_requests.py //pipelines

title "Refreshing tables defined in a source file"
trace $CLI pipelines run --refresh pipeline_file.py
trace print_requests.py //pipelines

title "Fully refreshing tables defined in a source file"
trace $CLI pipelines run --full-refresh pipeline_file.py
trace print_requests.py //pipelines

title "Rejecting missing sources and incompatible refresh modes"
trace musterr $CLI pipelines run --refresh ./missing.sql --no-wait
trace musterr $CLI pipelines run --refresh pipeline_file.py --full-refresh-all --no-wait
trace musterr $CLI pipelines run --full-refresh ./missing.sql --restart --no-wait
trace test ! -e "$OUT_REQUESTS"

title "Rejecting mixed selectors before restarting the pipeline"
trace musterr $CLI pipelines run --refresh table1,pipeline_file.py --restart --no-wait
trace musterr $CLI pipelines run --full-refresh table2,pipeline_file.py --restart --no-wait
trace test ! -e "$OUT_REQUESTS"
25 changes: 20 additions & 5 deletions bundle/run/pipeline_options.go
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
package run

import (
"errors"
"fmt"
"strings"

Expand All @@ -17,12 +18,16 @@ type PipelineOptions struct {
// List of tables to update.
Refresh []string

RefreshByPath []string

// Perform a full graph reset and recompute.
FullRefreshAll bool

// List of tables to reset and recompute.
FullRefresh []string

FullRefreshByPath []string

// Perform an update to validate graph correctness.
ValidateOnly bool
}
Expand Down Expand Up @@ -51,10 +56,18 @@ func (o *PipelineOptions) Validate(pipeline *resources.Pipeline) error {

if len(mutuallyExclusiveFlags) > 1 {
return fmt.Errorf("%s pipeline run flags are mutually exclusive", strings.Join(mutuallyExclusiveFlags, ", "))
} else if len(mutuallyExclusiveFlags) == 1 && (len(o.Refresh) > 0 || len(o.FullRefresh) > 0) {
} else if len(mutuallyExclusiveFlags) == 1 && (len(o.Refresh) > 0 || len(o.FullRefresh) > 0 || len(o.RefreshByPath) > 0 || len(o.FullRefreshByPath) > 0) {
return fmt.Errorf("cannot use --refresh or --full-refresh together with %s, these flags are mutually exclusive", strings.Join(mutuallyExclusiveFlags, ""))
}

// The API ignores file selections when table names are supplied for the same refresh mode.
if len(o.Refresh) > 0 && len(o.RefreshByPath) > 0 {
return errors.New("cannot mix table names and source files in --refresh; use only table names or only source files")
}
if len(o.FullRefresh) > 0 && len(o.FullRefreshByPath) > 0 {
return errors.New("cannot mix table names and source files in --full-refresh; use only table names or only source files")
}

return nil
}

Expand All @@ -66,10 +79,12 @@ func (o *PipelineOptions) toPayload(pipeline *resources.Pipeline, pipelineID str
PipelineId: pipelineID,

// Note: `RefreshAll` is implied if the fields below are not set.
RefreshSelection: o.Refresh,
FullRefresh: o.FullRefreshAll,
FullRefreshSelection: o.FullRefresh,
ValidateOnly: o.ValidateOnly,
RefreshSelection: o.Refresh,
FullRefresh: o.FullRefreshAll,
FullRefreshSelection: o.FullRefresh,
RefreshSelectionByPath: o.RefreshByPath,
FullRefreshSelectionByPath: o.FullRefreshByPath,
ValidateOnly: o.ValidateOnly,

// Only true is sent. When false the field is omitted, so the backend keeps
// using the pipeline-level development property while it is being deprecated.
Expand Down
77 changes: 77 additions & 0 deletions bundle/run/pipeline_options_test.go
Original file line number Diff line number Diff line change
@@ -1,13 +1,85 @@
package run

import (
"encoding/json"
"testing"

flag "github.com/spf13/pflag"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)

func TestPipelineOptionsPathSelectionsPayload(t *testing.T) {
tests := []struct {
name string
opts PipelineOptions
expected string
}{
{
name: "file selections for both refresh modes",
opts: PipelineOptions{
RefreshByPath: []string{"/Workspace/Shared/source.sql"},
FullRefreshByPath: []string{"/Workspace/Shared/notebook"},
},
expected: `{"refresh_selection_by_path":["/Workspace/Shared/source.sql"],"full_refresh_selection_by_path":["/Workspace/Shared/notebook"]}`,
},
{
name: "table refresh and file full-refresh",
opts: PipelineOptions{
Refresh: []string{"catalog.schema.table1"},
FullRefreshByPath: []string{"/Workspace/Shared/notebook"},
},
expected: `{"refresh_selection":["catalog.schema.table1"],"full_refresh_selection_by_path":["/Workspace/Shared/notebook"]}`,
},
{
name: "file refresh and table full-refresh",
opts: PipelineOptions{
RefreshByPath: []string{"/Workspace/Shared/source.sql"},
FullRefresh: []string{"catalog.schema.table2"},
},
expected: `{"refresh_selection_by_path":["/Workspace/Shared/source.sql"],"full_refresh_selection":["catalog.schema.table2"]}`,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
payload, err := tt.opts.toPayload(nil, "pipeline-id", false)
require.NoError(t, err)
data, err := json.Marshal(payload)
require.NoError(t, err)
assert.JSONEq(t, tt.expected, string(data))
})
}
}

func TestPipelineOptionsRejectMixedSelectors(t *testing.T) {
tests := []struct {
name string
opts PipelineOptions
}{
{
name: "--refresh",
opts: PipelineOptions{
Refresh: []string{"table1"},
RefreshByPath: []string{"/Workspace/Shared/source.sql"},
},
},
{
name: "--full-refresh",
opts: PipelineOptions{
FullRefresh: []string{"table1"},
FullRefreshByPath: []string{"/Workspace/Shared/source.sql"},
},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
payload, err := tt.opts.toPayload(nil, "pipeline-id", false)
require.ErrorContains(t, err, "cannot mix table names and source files in "+tt.name)
assert.Nil(t, payload)
})
}
}

func setupPipelineOptions(t *testing.T) (*flag.FlagSet, *PipelineOptions) {
var fs flag.FlagSet
var opts PipelineOptions
Expand Down Expand Up @@ -115,6 +187,11 @@ func TestPipelineOptionsValidateFailureWithRefreshAndMutuallyExclusiveFlags(t *t
require.NoError(t, err)
err = opts.Validate(nil)
assert.ErrorContains(t, err, "cannot use --refresh or --full-refresh together with")

opts.RefreshByPath, opts.Refresh = opts.Refresh, nil
opts.FullRefreshByPath, opts.FullRefresh = opts.FullRefresh, nil
err = opts.Validate(nil)
assert.ErrorContains(t, err, "cannot use --refresh or --full-refresh together with")
}
}
}
Loading
Loading