Skip to content

Commit e092a9c

Browse files
chenyuem-dbchenyuem-db
committed
Support file selectors in pipelines run
1 parent 3aa5c96 commit e092a9c

15 files changed

Lines changed: 614 additions & 35 deletions

File tree

‎.nextchanges/cli/pipeline-refresh-by-file.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
* Support source files in `pipelines run --refresh` and `--full-refresh` selectors. ([#7184](https://github.com/databricks/cli/pull/7184))

‎acceptance/pipelines/run/no-wait/output.txt‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -15,8 +15,3 @@ If you've made local changes, run 'databricks pipelines deploy' first to ensure
1515
Update URL: [DATABRICKS_URL]/#joblist/pipelines/[UUID]/updates/[UUID]
1616

1717
Update ID: [UUID]
18-
19-
Update for pipeline completed successfully.
20-
21-
Pipeline configurations for this update:
22-
• All tables are refreshed

‎acceptance/pipelines/run/refresh-flags/output.txt‎

Lines changed: 98 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@ Update ID: [UUID]
1919
Update for pipeline completed successfully.
2020

2121
Pipeline configurations for this update:
22-
• All tables are refreshed
22+
• Refreshed [table1, table2]
2323

2424
>>> print_requests.py //pipelines
2525
{
@@ -46,7 +46,7 @@ Update ID: [UUID]
4646
Update for pipeline completed successfully.
4747

4848
Pipeline configurations for this update:
49-
• All tables are refreshed
49+
• All tables are fully refreshed
5050

5151
>>> print_requests.py //pipelines
5252
{
@@ -70,7 +70,7 @@ Update ID: [UUID]
7070
Update for pipeline completed successfully.
7171

7272
Pipeline configurations for this update:
73-
• All tables are refreshed
73+
• Full refreshed [table1, table2]
7474

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

9999
Pipeline configurations for this update:
100-
• All tables are refreshed
100+
• Refreshed [table3, table4]
101+
• Full refreshed [table1, table2]
101102

102103
>>> print_requests.py //pipelines
103104
{
@@ -114,3 +115,96 @@ Pipeline configurations for this update:
114115
]
115116
}
116117
}
118+
119+
=== Refreshing tables defined in a source file
120+
>>> [CLI] pipelines run --refresh pipeline_file.py
121+
Recommendation: This command runs the last deployed version of the code
122+
123+
If you've made local changes, run 'databricks pipelines deploy' first to ensure they are included.
124+
125+
Update URL: [DATABRICKS_URL]/#joblist/pipelines/[UUID]/updates/[UUID]
126+
127+
Update ID: [UUID]
128+
129+
Update for pipeline completed successfully.
130+
131+
Pipeline configurations for this update:
132+
• Refreshed tables defined in [/Workspace/Users/[USERNAME]/.bundle/test-pipeline-run-flags/default/files/pipeline_file.py]
133+
134+
>>> print_requests.py //pipelines
135+
{
136+
"method": "POST",
137+
"path": "/api/2.0/pipelines/[UUID]/updates",
138+
"body": {
139+
"refresh_selection_by_path": [
140+
"/Workspace/Users/[USERNAME]/.bundle/test-pipeline-run-flags/default/files/pipeline_file.py"
141+
]
142+
}
143+
}
144+
145+
=== Fully refreshing tables defined in a source file
146+
>>> [CLI] pipelines run --full-refresh pipeline_file.py
147+
Recommendation: This command runs the last deployed version of the code
148+
149+
If you've made local changes, run 'databricks pipelines deploy' first to ensure they are included.
150+
151+
Update URL: [DATABRICKS_URL]/#joblist/pipelines/[UUID]/updates/[UUID]
152+
153+
Update ID: [UUID]
154+
155+
Update for pipeline completed successfully.
156+
157+
Pipeline configurations for this update:
158+
• Full refreshed tables defined in [/Workspace/Users/[USERNAME]/.bundle/test-pipeline-run-flags/default/files/pipeline_file.py]
159+
160+
>>> print_requests.py //pipelines
161+
{
162+
"method": "POST",
163+
"path": "/api/2.0/pipelines/[UUID]/updates",
164+
"body": {
165+
"full_refresh_selection_by_path": [
166+
"/Workspace/Users/[USERNAME]/.bundle/test-pipeline-run-flags/default/files/pipeline_file.py"
167+
]
168+
}
169+
}
170+
171+
=== Rejecting missing sources and incompatible refresh modes
172+
>>> musterr [CLI] pipelines run --refresh ./missing.sql --no-wait
173+
Recommendation: This command runs the last deployed version of the code
174+
175+
If you've made local changes, run 'databricks pipelines deploy' first to ensure they are included.
176+
177+
Error: invalid --refresh: source file "./missing.sql" not found; use a table name or an existing source file
178+
179+
>>> musterr [CLI] pipelines run --refresh pipeline_file.py --full-refresh-all --no-wait
180+
Recommendation: This command runs the last deployed version of the code
181+
182+
If you've made local changes, run 'databricks pipelines deploy' first to ensure they are included.
183+
184+
Error: cannot use --refresh or --full-refresh together with --full-refresh-all, these flags are mutually exclusive
185+
186+
>>> musterr [CLI] pipelines run --full-refresh ./missing.sql --restart --no-wait
187+
Recommendation: This command runs the last deployed version of the code
188+
189+
If you've made local changes, run 'databricks pipelines deploy' first to ensure they are included.
190+
191+
Error: invalid --full-refresh: source file "./missing.sql" not found; use a table name or an existing source file
192+
193+
>>> test ! -e [OUTPUT_DIR]/out.requests.txt
194+
195+
=== Rejecting mixed selectors before restarting the pipeline
196+
>>> musterr [CLI] pipelines run --refresh table1,pipeline_file.py --restart --no-wait
197+
Recommendation: This command runs the last deployed version of the code
198+
199+
If you've made local changes, run 'databricks pipelines deploy' first to ensure they are included.
200+
201+
Error: cannot mix table names and source files in --refresh; use only table names or only source files
202+
203+
>>> musterr [CLI] pipelines run --full-refresh table2,pipeline_file.py --restart --no-wait
204+
Recommendation: This command runs the last deployed version of the code
205+
206+
If you've made local changes, run 'databricks pipelines deploy' first to ensure they are included.
207+
208+
Error: cannot mix table names and source files in --full-refresh; use only table names or only source files
209+
210+
>>> test ! -e [OUTPUT_DIR]/out.requests.txt

‎acceptance/pipelines/run/refresh-flags/script‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,3 +16,22 @@ trace print_requests.py //pipelines
1616
title "Running pipeline with --full-refresh flag and --refresh flag"
1717
trace $CLI pipelines run --full-refresh table1,table2 --refresh table3,table4
1818
trace print_requests.py //pipelines
19+
20+
title "Refreshing tables defined in a source file"
21+
trace $CLI pipelines run --refresh pipeline_file.py
22+
trace print_requests.py //pipelines
23+
24+
title "Fully refreshing tables defined in a source file"
25+
trace $CLI pipelines run --full-refresh pipeline_file.py
26+
trace print_requests.py //pipelines
27+
28+
title "Rejecting missing sources and incompatible refresh modes"
29+
trace musterr $CLI pipelines run --refresh ./missing.sql --no-wait
30+
trace musterr $CLI pipelines run --refresh pipeline_file.py --full-refresh-all --no-wait
31+
trace musterr $CLI pipelines run --full-refresh ./missing.sql --restart --no-wait
32+
trace test ! -e "$OUT_REQUESTS"
33+
34+
title "Rejecting mixed selectors before restarting the pipeline"
35+
trace musterr $CLI pipelines run --refresh table1,pipeline_file.py --restart --no-wait
36+
trace musterr $CLI pipelines run --full-refresh table2,pipeline_file.py --restart --no-wait
37+
trace test ! -e "$OUT_REQUESTS"

‎bundle/run/pipeline_options.go‎

Lines changed: 20 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package run
22

33
import (
4+
"errors"
45
"fmt"
56
"strings"
67

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

21+
RefreshByPath []string
22+
2023
// Perform a full graph reset and recompute.
2124
FullRefreshAll bool
2225

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

29+
FullRefreshByPath []string
30+
2631
// Perform an update to validate graph correctness.
2732
ValidateOnly bool
2833
}
@@ -51,10 +56,18 @@ func (o *PipelineOptions) Validate(pipeline *resources.Pipeline) error {
5156

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

63+
// The API ignores file selections when table names are supplied for the same refresh mode.
64+
if len(o.Refresh) > 0 && len(o.RefreshByPath) > 0 {
65+
return errors.New("cannot mix table names and source files in --refresh; use only table names or only source files")
66+
}
67+
if len(o.FullRefresh) > 0 && len(o.FullRefreshByPath) > 0 {
68+
return errors.New("cannot mix table names and source files in --full-refresh; use only table names or only source files")
69+
}
70+
5871
return nil
5972
}
6073

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

6881
// Note: `RefreshAll` is implied if the fields below are not set.
69-
RefreshSelection: o.Refresh,
70-
FullRefresh: o.FullRefreshAll,
71-
FullRefreshSelection: o.FullRefresh,
72-
ValidateOnly: o.ValidateOnly,
82+
RefreshSelection: o.Refresh,
83+
FullRefresh: o.FullRefreshAll,
84+
FullRefreshSelection: o.FullRefresh,
85+
RefreshSelectionByPath: o.RefreshByPath,
86+
FullRefreshSelectionByPath: o.FullRefreshByPath,
87+
ValidateOnly: o.ValidateOnly,
7388

7489
// Only true is sent. When false the field is omitted, so the backend keeps
7590
// using the pipeline-level development property while it is being deprecated.

‎bundle/run/pipeline_options_test.go‎

Lines changed: 77 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,85 @@
11
package run
22

33
import (
4+
"encoding/json"
45
"testing"
56

67
flag "github.com/spf13/pflag"
78
"github.com/stretchr/testify/assert"
89
"github.com/stretchr/testify/require"
910
)
1011

12+
func TestPipelineOptionsPathSelectionsPayload(t *testing.T) {
13+
tests := []struct {
14+
name string
15+
opts PipelineOptions
16+
expected string
17+
}{
18+
{
19+
name: "file selections for both refresh modes",
20+
opts: PipelineOptions{
21+
RefreshByPath: []string{"/Workspace/Shared/source.sql"},
22+
FullRefreshByPath: []string{"/Workspace/Shared/notebook"},
23+
},
24+
expected: `{"refresh_selection_by_path":["/Workspace/Shared/source.sql"],"full_refresh_selection_by_path":["/Workspace/Shared/notebook"]}`,
25+
},
26+
{
27+
name: "table refresh and file full-refresh",
28+
opts: PipelineOptions{
29+
Refresh: []string{"catalog.schema.table1"},
30+
FullRefreshByPath: []string{"/Workspace/Shared/notebook"},
31+
},
32+
expected: `{"refresh_selection":["catalog.schema.table1"],"full_refresh_selection_by_path":["/Workspace/Shared/notebook"]}`,
33+
},
34+
{
35+
name: "file refresh and table full-refresh",
36+
opts: PipelineOptions{
37+
RefreshByPath: []string{"/Workspace/Shared/source.sql"},
38+
FullRefresh: []string{"catalog.schema.table2"},
39+
},
40+
expected: `{"refresh_selection_by_path":["/Workspace/Shared/source.sql"],"full_refresh_selection":["catalog.schema.table2"]}`,
41+
},
42+
}
43+
for _, tt := range tests {
44+
t.Run(tt.name, func(t *testing.T) {
45+
payload, err := tt.opts.toPayload(nil, "pipeline-id", false)
46+
require.NoError(t, err)
47+
data, err := json.Marshal(payload)
48+
require.NoError(t, err)
49+
assert.JSONEq(t, tt.expected, string(data))
50+
})
51+
}
52+
}
53+
54+
func TestPipelineOptionsRejectMixedSelectors(t *testing.T) {
55+
tests := []struct {
56+
name string
57+
opts PipelineOptions
58+
}{
59+
{
60+
name: "--refresh",
61+
opts: PipelineOptions{
62+
Refresh: []string{"table1"},
63+
RefreshByPath: []string{"/Workspace/Shared/source.sql"},
64+
},
65+
},
66+
{
67+
name: "--full-refresh",
68+
opts: PipelineOptions{
69+
FullRefresh: []string{"table1"},
70+
FullRefreshByPath: []string{"/Workspace/Shared/source.sql"},
71+
},
72+
},
73+
}
74+
for _, tt := range tests {
75+
t.Run(tt.name, func(t *testing.T) {
76+
payload, err := tt.opts.toPayload(nil, "pipeline-id", false)
77+
require.ErrorContains(t, err, "cannot mix table names and source files in "+tt.name)
78+
assert.Nil(t, payload)
79+
})
80+
}
81+
}
82+
1183
func setupPipelineOptions(t *testing.T) (*flag.FlagSet, *PipelineOptions) {
1284
var fs flag.FlagSet
1385
var opts PipelineOptions
@@ -115,6 +187,11 @@ func TestPipelineOptionsValidateFailureWithRefreshAndMutuallyExclusiveFlags(t *t
115187
require.NoError(t, err)
116188
err = opts.Validate(nil)
117189
assert.ErrorContains(t, err, "cannot use --refresh or --full-refresh together with")
190+
191+
opts.RefreshByPath, opts.Refresh = opts.Refresh, nil
192+
opts.FullRefreshByPath, opts.FullRefresh = opts.FullRefresh, nil
193+
err = opts.Validate(nil)
194+
assert.ErrorContains(t, err, "cannot use --refresh or --full-refresh together with")
118195
}
119196
}
120197
}

0 commit comments

Comments
 (0)