From 9d78b59880a35c1eeb73eca08dcc664bf374c1b6 Mon Sep 17 00:00:00 2001 From: "chenyue.ma" Date: Wed, 28 Oct 2026 22:30:25 +0000 Subject: [PATCH 1/4] Support file selectors in pipelines run --- .nextchanges/cli/pipeline-refresh-by-file.md | 1 + .../pipelines/run/refresh-flags/output.txt | 102 ++++++++++- acceptance/pipelines/run/refresh-flags/script | 19 ++ bundle/run/pipeline_options.go | 25 ++- bundle/run/pipeline_options_test.go | 77 ++++++++ cmd/pipelines/refresh.go | 105 +++++++++++ cmd/pipelines/refresh_test.go | 170 ++++++++++++++++++ cmd/pipelines/run.go | 26 ++- cmd/pipelines/run_test.go | 28 +++ cmd/pipelines/templates.go | 8 +- libs/testserver/fake_workspace.go | 4 +- libs/testserver/handlers.go | 2 +- libs/testserver/pipelines.go | 28 ++- libs/testserver/pipelines_test.go | 44 ++++- 14 files changed, 612 insertions(+), 27 deletions(-) create mode 100644 .nextchanges/cli/pipeline-refresh-by-file.md create mode 100644 cmd/pipelines/refresh.go create mode 100644 cmd/pipelines/refresh_test.go diff --git a/.nextchanges/cli/pipeline-refresh-by-file.md b/.nextchanges/cli/pipeline-refresh-by-file.md new file mode 100644 index 00000000000..c5e25a11058 --- /dev/null +++ b/.nextchanges/cli/pipeline-refresh-by-file.md @@ -0,0 +1 @@ +* Support source files in `pipelines run --refresh` and `--full-refresh` selectors. ([#7184](https://github.com/databricks/cli/pull/7184)) diff --git a/acceptance/pipelines/run/refresh-flags/output.txt b/acceptance/pipelines/run/refresh-flags/output.txt index 68dcd0c9a23..e8ac6594453 100644 --- a/acceptance/pipelines/run/refresh-flags/output.txt +++ b/acceptance/pipelines/run/refresh-flags/output.txt @@ -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 { @@ -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 { @@ -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 { @@ -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 { @@ -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 diff --git a/acceptance/pipelines/run/refresh-flags/script b/acceptance/pipelines/run/refresh-flags/script index b324074d59f..e2838af2063 100644 --- a/acceptance/pipelines/run/refresh-flags/script +++ b/acceptance/pipelines/run/refresh-flags/script @@ -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" diff --git a/bundle/run/pipeline_options.go b/bundle/run/pipeline_options.go index 63ff3e81289..164e8a2de61 100644 --- a/bundle/run/pipeline_options.go +++ b/bundle/run/pipeline_options.go @@ -1,6 +1,7 @@ package run import ( + "errors" "fmt" "strings" @@ -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 } @@ -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 } @@ -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. diff --git a/bundle/run/pipeline_options_test.go b/bundle/run/pipeline_options_test.go index 1eda4c8bae7..b2f7c0a3f81 100644 --- a/bundle/run/pipeline_options_test.go +++ b/bundle/run/pipeline_options_test.go @@ -1,6 +1,7 @@ package run import ( + "encoding/json" "testing" flag "github.com/spf13/pflag" @@ -8,6 +9,77 @@ import ( "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 @@ -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") } } } diff --git a/cmd/pipelines/refresh.go b/cmd/pipelines/refresh.go new file mode 100644 index 00000000000..d26da6e33a8 --- /dev/null +++ b/cmd/pipelines/refresh.go @@ -0,0 +1,105 @@ +package pipelines + +import ( + "context" + "errors" + "fmt" + "io/fs" + "os" + "path" + "path/filepath" + "strings" + + "github.com/databricks/cli/bundle" + "github.com/databricks/cli/libs/notebook" + "github.com/databricks/cli/libs/sync" +) + +// resolveRefreshSelectors separates table names from source files and translates local files to workspace paths. +func resolveRefreshSelectors(ctx context.Context, b *bundle.Bundle, selectors []string) (tables, paths []string, err error) { + var remoteNames map[string]string + for _, selector := range selectors { + if strings.HasPrefix(selector, "/Workspace/") { + paths = append(paths, selector) + continue + } + + // Quoted identifiers remain table selectors even when they contain path separators. + if strings.Contains(selector, "`") { + tables = append(tables, selector) + continue + } + + info, statErr := os.Stat(selector) + // A bare selector is a table unless it identifies an existing local file. + looksLikeFile := strings.ContainsAny(selector, `/\`) + if !looksLikeFile && (statErr != nil || info.IsDir()) { + tables = append(tables, selector) + continue + } + if errors.Is(statErr, fs.ErrNotExist) { + return nil, nil, fmt.Errorf("source file %q not found; use a table name or an existing source file", selector) + } + if statErr != nil { + return nil, nil, fmt.Errorf("cannot read refresh selector %q: %w", selector, statErr) + } + if info.IsDir() { + return nil, nil, fmt.Errorf("refresh selector %q is a directory; select a source file", selector) + } + + localPath, err := filepath.Abs(selector) + if err != nil { + return nil, nil, err + } + rel, err := filepath.Rel(b.SyncRootPath, localPath) + if err != nil { + return nil, nil, err + } + if !filepath.IsLocal(rel) { + return nil, nil, fmt.Errorf("source file %q is outside the bundle sync root", selector) + } + if remoteNames == nil { + remoteNames, err = refreshSourcePaths(ctx, b) + if err != nil { + return nil, nil, err + } + } + remoteName, ok := remoteNames[filepath.ToSlash(rel)] + if !ok { + if _, _, err := notebook.DetectWithFS(b.SyncRoot, filepath.ToSlash(rel)); err != nil { + return nil, nil, fmt.Errorf("cannot identify source file %q: %w", selector, err) + } + return nil, nil, fmt.Errorf("source file %q is excluded from bundle sync; include it and run 'databricks pipelines deploy' first", selector) + } + remoteRoot := b.Config.Workspace.FilePath + if b.IsSourceLinkedDeployment() { + remoteRoot = filepath.ToSlash(b.SyncRootPath) + } + if strings.Contains(remoteRoot, "${") { + return nil, nil, fmt.Errorf("cannot resolve the deployed path for %q; use its absolute /Workspace/ path", selector) + } + paths = append(paths, path.Join(remoteRoot, remoteName)) + } + return tables, paths, nil +} + +// refreshSourcePaths uses the sync mapping so notebook names and sync exclusions match deployment. +func refreshSourcePaths(ctx context.Context, b *bundle.Bundle) (map[string]string, error) { + includes, err := b.GetSyncIncludePatterns(ctx) + if err != nil { + return nil, err + } + list, err := sync.NewFileList(ctx, b.WorktreeRoot, b.SyncRoot, b.Config.Sync.Paths, includes, b.Config.Sync.Exclude) + if err != nil { + return nil, err + } + files, err := list.Files(ctx) + if err != nil { + return nil, err + } + state, err := sync.NewSnapshotState(files) + if err != nil { + return nil, err + } + return state.LocalToRemoteNames, nil +} diff --git a/cmd/pipelines/refresh_test.go b/cmd/pipelines/refresh_test.go new file mode 100644 index 00000000000..eb1824a5890 --- /dev/null +++ b/cmd/pipelines/refresh_test.go @@ -0,0 +1,170 @@ +package pipelines + +import ( + "os" + "path/filepath" + "strings" + "testing" + + "github.com/databricks/cli/bundle" + "github.com/databricks/cli/bundle/config" + "github.com/databricks/cli/libs/vfs" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestResolveRefreshSelectors(t *testing.T) { + longTable := strings.Repeat("c", 100) + "." + strings.Repeat("s", 100) + "." + strings.Repeat("t", 100) + tests := []struct { + name string + selectors []string + tables []string + paths []string + err string + }{ + {name: "no selection"}, + {name: "long qualified table", selectors: []string{longTable}, tables: []string{longTable}}, + { + name: "table identifiers", + selectors: []string{"table1", "schema.table2", "schema.sql", "catalog.schema.sql", "`schema`.`table/with/slashes`", "`schema`.`sql`"}, + tables: []string{"table1", "schema.table2", "schema.sql", "catalog.schema.sql", "`schema`.`table/with/slashes`", "`schema`.`sql`"}, + }, + { + name: "quoted table with the same name as a file", + selectors: []string{"`src/query.sql`"}, + tables: []string{"`src/query.sql`"}, + }, + { + name: "mixed table and file selectors", + selectors: []string{"table1", "src/query.sql", "src/module.py"}, + tables: []string{"table1"}, + paths: []string{"/Workspace/Shared/project/files/src/query.sql", "/Workspace/Shared/project/files/src/module.py"}, + }, + { + name: "notebook extension removed", + selectors: []string{"src/notebook.py", "src/jupyter.ipynb"}, + paths: []string{"/Workspace/Shared/project/files/src/notebook", "/Workspace/Shared/project/files/src/jupyter"}, + }, + { + name: "workspace path", + selectors: []string{"/Workspace/Shared/remote.sql"}, + paths: []string{"/Workspace/Shared/remote.sql"}, + }, + {name: "missing relative file", selectors: []string{"src/missing.sql"}, err: `source file "src/missing.sql" not found`}, + {name: "explicit missing filename", selectors: []string{"./missing.sql"}, err: `source file "./missing.sql" not found`}, + {name: "table named like a directory", selectors: []string{"src"}, tables: []string{"src"}}, + {name: "explicit directory", selectors: []string{"./src"}, err: "is a directory"}, + {name: "excluded file", selectors: []string{"src/excluded.sql"}, err: "is excluded from bundle sync"}, + {name: "invalid notebook", selectors: []string{"src/invalid.ipynb"}, err: "invalid Jupyter notebook file"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + root := t.TempDir() + require.NoError(t, os.MkdirAll(filepath.Join(root, "src"), 0o700)) + for name, content := range map[string]string{ + "query.sql": "SELECT 1;\n", + "module.py": "print('file')\n", + "notebook.py": "# Databricks notebook source\nprint('notebook')\n", + "jupyter.ipynb": `{"cells":[],"metadata":{},"nbformat":4,"nbformat_minor":5}`, + "excluded.sql": "SELECT 2;\n", + "invalid.ipynb": `{}`, + } { + require.NoError(t, os.WriteFile(filepath.Join(root, "src", name), []byte(content), 0o600)) + } + t.Chdir(root) + b := refreshTestBundle(root) + b.Config.Sync.Exclude = []string{"src/excluded.sql"} + tables, paths, err := resolveRefreshSelectors(t.Context(), b, tt.selectors) + if tt.err != "" { + require.ErrorContains(t, err, tt.err) + return + } + require.NoError(t, err) + assert.Equal(t, tt.tables, tables) + assert.Equal(t, tt.paths, paths) + }) + } +} + +func refreshTestBundle(root string) *bundle.Bundle { + return &bundle.Bundle{ + BundleRootPath: root, + SyncRootPath: root, + SyncRoot: vfs.MustNew(root), + WorktreeRoot: vfs.MustNew(root), + Config: config.Root{ + Bundle: config.Bundle{Name: "test", Target: "default"}, + Workspace: config.Workspace{FilePath: "/Workspace/Shared/project/files"}, + Sync: config.Sync{Paths: []string{"."}}, + }, + } +} + +func TestResolveRefreshSelectorsFromSubdirectory(t *testing.T) { + root := t.TempDir() + require.NoError(t, os.MkdirAll(filepath.Join(root, "project", "src"), 0o700)) + require.NoError(t, os.WriteFile(filepath.Join(root, "shared.sql"), []byte("SELECT 1;\n"), 0o600)) + t.Chdir(filepath.Join(root, "project", "src")) + b := refreshTestBundle(root) + b.BundleRootPath = filepath.Join(root, "project") + for _, selector := range []string{"../../shared.sql", filepath.Join(root, "shared.sql")} { + tables, paths, err := resolveRefreshSelectors(t.Context(), b, []string{selector}) + require.NoError(t, err) + assert.Empty(t, tables) + assert.Equal(t, []string{"/Workspace/Shared/project/files/shared.sql"}, paths) + } +} + +func TestResolveRefreshSelectorsOutsideSyncRoot(t *testing.T) { + root := t.TempDir() + require.NoError(t, os.Mkdir(filepath.Join(root, "project"), 0o700)) + require.NoError(t, os.WriteFile(filepath.Join(root, "outside.sql"), []byte("SELECT 1;\n"), 0o600)) + t.Chdir(filepath.Join(root, "project")) + _, _, err := resolveRefreshSelectors(t.Context(), refreshTestBundle(filepath.Join(root, "project")), []string{"../outside.sql"}) + require.ErrorContains(t, err, "is outside the bundle sync root") +} + +func TestResolveRefreshSelectorsDeploymentMode(t *testing.T) { + root := t.TempDir() + require.NoError(t, os.WriteFile(filepath.Join(root, "query.sql"), []byte("SELECT 1;\n"), 0o600)) + t.Chdir(root) + enabled, disabled := true, false + localPath := filepath.ToSlash(filepath.Join(root, "query.sql")) + remotePath := "/Workspace/Shared/project/files/query.sql" + tests := []struct { + name string + mode config.SyncMode + sourceLinked *bool + want string + }{ + {name: "default upload", want: remotePath}, + {name: "in-place", mode: config.SyncModeInPlace, want: localPath}, + {name: "legacy source-linked", sourceLinked: &enabled, want: localPath}, + {name: "overwrite overrides legacy setting", mode: config.SyncModeOverwrite, sourceLinked: &enabled, want: remotePath}, + {name: "in-place overrides legacy setting", mode: config.SyncModeInPlace, sourceLinked: &disabled, want: localPath}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + b := refreshTestBundle(root) + b.Config.Workspace.SyncMode = tt.mode + b.Config.Presets.SourceLinkedDeployment = tt.sourceLinked + tables, paths, err := resolveRefreshSelectors(t.Context(), b, []string{"query.sql"}) + require.NoError(t, err) + assert.Empty(t, tables) + assert.Equal(t, []string{tt.want}, paths) + }) + } +} + +func TestResolveRefreshSelectorsUnresolvedWorkspacePath(t *testing.T) { + root := t.TempDir() + require.NoError(t, os.WriteFile(filepath.Join(root, "query.sql"), []byte("SELECT 1;\n"), 0o600)) + t.Chdir(root) + b := refreshTestBundle(root) + b.Config.Workspace.FilePath = "${resources.internal_immutable_snapshots.immutable.full_path}/files" + b.Config.Workspace.SyncMode = config.SyncModeImmutable + enabled := true + b.Config.Presets.SourceLinkedDeployment = &enabled + _, _, err := resolveRefreshSelectors(t.Context(), b, []string{"query.sql"}) + require.ErrorContains(t, err, "use its absolute /Workspace/ path") +} diff --git a/cmd/pipelines/run.go b/cmd/pipelines/run.go index 9f81961b6f0..702a8392edf 100644 --- a/cmd/pipelines/run.go +++ b/cmd/pipelines/run.go @@ -233,7 +233,14 @@ func runCommand() *cobra.Command { Long: `Run the pipeline identified by KEY. KEY is the unique name of the pipeline to run, as defined in its YAML file. If there is only one pipeline in the project, KEY is optional and the pipeline will be auto-selected. -Refreshes all tables in the pipeline unless otherwise specified.`, +Refreshes all tables in the pipeline unless otherwise specified. + +Use --refresh or --full-refresh with table names, local source files, or absolute +/Workspace/ paths. Local files use the bundle sync configuration and paths are +relative to the current directory. Deploy local changes before running. +Each flag accepts either table names or source files, but cannot mix both. +Existing local files take precedence over unquoted table names. Use ./ to require +a file, or quote table identifiers with backticks to select an ambiguous table.`, } var refresh []string @@ -241,9 +248,9 @@ Refreshes all tables in the pipeline unless otherwise specified.`, var fullRefresh []string pipelineGroup := cmdgroup.NewFlagGroup("Pipeline Run") - pipelineGroup.FlagSet().StringSliceVar(&refresh, "refresh", nil, "List of tables to run.") + pipelineGroup.FlagSet().StringSliceVar(&refresh, "refresh", nil, "List of tables or source files to run. Local paths are relative to the current directory.") pipelineGroup.FlagSet().BoolVar(&fullRefreshAll, "full-refresh-all", false, "Perform a full graph reset and recompute.") - pipelineGroup.FlagSet().StringSliceVar(&fullRefresh, "full-refresh", nil, "List of tables to reset and recompute.") + pipelineGroup.FlagSet().StringSliceVar(&fullRefresh, "full-refresh", nil, "List of tables or source files to reset and recompute. Local paths are relative to the current directory.") wrappedCmd := cmdgroup.NewCommandWithGroupFlag(cmd) wrappedCmd.AddFlagGroup(pipelineGroup) @@ -280,12 +287,21 @@ Refreshes all tables in the pipeline unless otherwise specified.`, runOptions := run.Options{ Pipeline: run.PipelineOptions{ - Refresh: refresh, FullRefreshAll: fullRefreshAll, - FullRefresh: fullRefresh, }, NoWait: noWait, } + runOptions.Pipeline.Refresh, runOptions.Pipeline.RefreshByPath, err = resolveRefreshSelectors(ctx, b, refresh) + if err != nil { + return fmt.Errorf("invalid --refresh: %w", err) + } + runOptions.Pipeline.FullRefresh, runOptions.Pipeline.FullRefreshByPath, err = resolveRefreshSelectors(ctx, b, fullRefresh) + if err != nil { + return fmt.Errorf("invalid --full-refresh: %w", err) + } + if err := runOptions.Pipeline.Validate(nil); err != nil { + return err + } var runOutput bundlerunoutput.RunOutput if restart { diff --git a/cmd/pipelines/run_test.go b/cmd/pipelines/run_test.go index 9937cb83d0d..2c910669e34 100644 --- a/cmd/pipelines/run_test.go +++ b/cmd/pipelines/run_test.go @@ -99,6 +99,34 @@ Update for pipeline completed successfully. Pipeline configurations for this update: • Refreshed [table1, table2] • Full refreshed [table3] +`, + }, + { + name: "pipeline update with file refresh selections", + update: pipelines.UpdateInfo{ + RefreshSelectionByPath: []string{"/Workspace/Shared/source.sql"}, + FullRefreshSelectionByPath: []string{"/Workspace/Shared/notebook"}, + }, + expected: ` +Update for pipeline completed successfully. + +Pipeline configurations for this update: +• Refreshed tables defined in [/Workspace/Shared/source.sql] +• Full refreshed tables defined in [/Workspace/Shared/notebook] +`, + }, + { + name: "pipeline update with table refresh and file full refresh", + update: pipelines.UpdateInfo{ + RefreshSelection: []string{"table1"}, + FullRefreshSelectionByPath: []string{"/Workspace/Shared/source.sql"}, + }, + expected: ` +Update for pipeline completed successfully. + +Pipeline configurations for this update: +• Refreshed [table1] +• Full refreshed tables defined in [/Workspace/Shared/source.sql] `, }, { diff --git a/cmd/pipelines/templates.go b/cmd/pipelines/templates.go index 3c471b366bb..f0bc50eedf1 100644 --- a/cmd/pipelines/templates.go +++ b/cmd/pipelines/templates.go @@ -29,7 +29,7 @@ Update end time: {{ .LastEventTime }} Pipeline configurations for this update: {{- if .Update.FullRefresh }} • All tables are fully refreshed -{{- else if and (eq (len .Update.RefreshSelection) 0) (eq (len .Update.FullRefreshSelection) 0) }} +{{- else if and (eq (len .Update.RefreshSelection) 0) (eq (len .Update.FullRefreshSelection) 0) (eq (len .Update.RefreshSelectionByPath) 0) (eq (len .Update.FullRefreshSelectionByPath) 0) }} • All tables are refreshed {{- else }} {{- if gt (len .Update.RefreshSelection) 0 }} @@ -38,6 +38,12 @@ Pipeline configurations for this update: {{- if gt (len .Update.FullRefreshSelection) 0 }} • Full refreshed [{{ join .Update.FullRefreshSelection ", " }}] {{- end }} +{{- if gt (len .Update.RefreshSelectionByPath) 0 }} +• Refreshed tables defined in [{{ join .Update.RefreshSelectionByPath ", " }}] +{{- end }} +{{- if gt (len .Update.FullRefreshSelectionByPath) 0 }} +• Full refreshed tables defined in [{{ join .Update.FullRefreshSelectionByPath ", " }}] +{{- end }} {{- end }} {{- if .Update.Cause }} • Update cause: {{ .Update.Cause }} diff --git a/libs/testserver/fake_workspace.go b/libs/testserver/fake_workspace.go index dd5050f6f49..ad72686ced7 100644 --- a/libs/testserver/fake_workspace.go +++ b/libs/testserver/fake_workspace.go @@ -216,7 +216,7 @@ type FakeWorkspace struct { JobRunOutputs map[int64]jobs.RunOutput JobRunIdempotency map[string]int64 Pipelines map[string]pipelines.GetPipelineResponse - PipelineUpdates map[string]bool + PipelineUpdates map[string]pipelines.UpdateInfo Monitors map[string]catalog.MonitorInfo Apps map[string]apps.App Schemas map[string]catalog.SchemaInfo @@ -516,7 +516,7 @@ func NewFakeWorkspace(url, token string) *FakeWorkspace { JobRunIdempotency: map[string]int64{}, Grants: map[string][]catalog.PrivilegeAssignment{}, Pipelines: map[string]pipelines.GetPipelineResponse{}, - PipelineUpdates: map[string]bool{}, + PipelineUpdates: map[string]pipelines.UpdateInfo{}, Monitors: map[string]catalog.MonitorInfo{}, Apps: map[string]apps.App{}, Catalogs: map[string]catalog.CatalogInfo{}, diff --git a/libs/testserver/handlers.go b/libs/testserver/handlers.go index 50ff904a573..22d47fa2586 100644 --- a/libs/testserver/handlers.go +++ b/libs/testserver/handlers.go @@ -648,7 +648,7 @@ func AddDefaultHandlers(server *Server) { }) server.Handle("POST", "/api/2.0/pipelines/{pipeline_id}/updates", func(req Request) any { - return req.Workspace.PipelineStartUpdate(req.Vars["pipeline_id"]) + return req.Workspace.PipelineStartUpdate(req, req.Vars["pipeline_id"]) }) server.Handle("GET", "/api/2.0/pipelines/{pipeline_id}/events", func(req Request) any { diff --git a/libs/testserver/pipelines.go b/libs/testserver/pipelines.go index 0aedc44c5d3..6960aebf350 100644 --- a/libs/testserver/pipelines.go +++ b/libs/testserver/pipelines.go @@ -143,7 +143,7 @@ func (s *FakeWorkspace) PipelineUpdate(req Request, pipelineId string) Response return Response{} } -func (s *FakeWorkspace) PipelineStartUpdate(pipelineId string) Response { +func (s *FakeWorkspace) PipelineStartUpdate(req Request, pipelineId string) Response { defer s.LockUnlock()() _, exists := s.Pipelines[pipelineId] @@ -154,8 +154,25 @@ func (s *FakeWorkspace) PipelineStartUpdate(pipelineId string) Response { } } + var start pipelines.StartUpdate + if err := json.Unmarshal(req.Body, &start); err != nil { + return Response{ + Body: fmt.Sprintf("cannot unmarshal request body: %s", err), + StatusCode: 400, + } + } + updateId := nextUUID() - s.PipelineUpdates[updateId] = true + // GetUpdate echoes selectors; the execution engine resolves paths to tables. + s.PipelineUpdates[updateId] = pipelines.UpdateInfo{ + UpdateId: updateId, + State: pipelines.UpdateInfoStateCompleted, + FullRefresh: start.FullRefresh, + RefreshSelection: start.RefreshSelection, + FullRefreshSelection: start.FullRefreshSelection, + RefreshSelectionByPath: start.RefreshSelectionByPath, + FullRefreshSelectionByPath: start.FullRefreshSelectionByPath, + } return Response{ Body: pipelines.StartUpdateResponse{ @@ -194,7 +211,7 @@ func (s *FakeWorkspace) PipelineGetUpdate(pipelineId, updateId string) Response } // Check if the update exists - _, updateExists := s.PipelineUpdates[updateId] + update, updateExists := s.PipelineUpdates[updateId] if !updateExists { return Response{ StatusCode: 404, @@ -204,10 +221,7 @@ func (s *FakeWorkspace) PipelineGetUpdate(pipelineId, updateId string) Response return Response{ Body: pipelines.GetUpdateResponse{ - Update: &pipelines.UpdateInfo{ - UpdateId: updateId, - State: pipelines.UpdateInfoStateCompleted, - }, + Update: &update, }, } } diff --git a/libs/testserver/pipelines_test.go b/libs/testserver/pipelines_test.go index 39114714082..6bb1dadfbdb 100644 --- a/libs/testserver/pipelines_test.go +++ b/libs/testserver/pipelines_test.go @@ -51,7 +51,7 @@ func TestPipelineCreate_AllowsSingleSegmentTargetSchemaName(t *testing.T) { func TestPipelineStartUpdate_HandlesNonExistentPipeline(t *testing.T) { workspace := NewFakeWorkspace("http://test", "dbapi123") - response := workspace.PipelineStartUpdate("non-existent-pipeline") + response := workspace.PipelineStartUpdate(Request{Body: []byte(`{}`)}, "non-existent-pipeline") assert.Equal(t, 404, response.StatusCode) assert.Contains(t, response.Body.(map[string]string)["message"], "The specified pipeline non-existent-pipeline was not found") } @@ -74,7 +74,7 @@ func TestPipelineStop_AfterUpdate(t *testing.T) { pipelineId := createTestPipeline(t, workspace) - startResponse := workspace.PipelineStartUpdate(pipelineId) + startResponse := workspace.PipelineStartUpdate(Request{Body: []byte(`{}`)}, pipelineId) assert.Equal(t, 0, startResponse.StatusCode) stopResponse := workspace.PipelineStop(pipelineId) @@ -85,3 +85,43 @@ func TestPipelineStop_AfterUpdate(t *testing.T) { assert.Equal(t, pipelineId, stopBody.PipelineId) assert.Equal(t, pipelines.PipelineStateIdle, stopBody.State) } + +func TestPipelineGetUpdatePreservesRefreshSelections(t *testing.T) { + workspace := NewFakeWorkspace("http://workspace.test", "dbapi123") + pipelineId := createTestPipeline(t, workspace) + tests := []struct { + name string + body string + want pipelines.UpdateInfo + }{ + { + name: "file selections", + body: `{"refresh_selection_by_path":["/Workspace/src/orders.py","/Workspace/src/returns.py"],"full_refresh_selection_by_path":["/Workspace/src/customers.sql"]}`, + want: pipelines.UpdateInfo{ + RefreshSelectionByPath: []string{"/Workspace/src/orders.py", "/Workspace/src/returns.py"}, + FullRefreshSelectionByPath: []string{"/Workspace/src/customers.sql"}, + }, + }, + { + name: "table selections", + body: `{"refresh_selection":["orders"],"full_refresh_selection":["customers"]}`, + want: pipelines.UpdateInfo{ + RefreshSelection: []string{"orders"}, + FullRefreshSelection: []string{"customers"}, + }, + }, + } + for i := range tests { + response := workspace.PipelineStartUpdate(Request{Body: []byte(tests[i].body)}, pipelineId) + require.Zero(t, response.StatusCode) + tests[i].want.UpdateId = response.Body.(pipelines.StartUpdateResponse).UpdateId + tests[i].want.State = pipelines.UpdateInfoStateCompleted + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + response := workspace.PipelineGetUpdate(pipelineId, tt.want.UpdateId) + require.Zero(t, response.StatusCode) + assert.Equal(t, &tt.want, response.Body.(pipelines.GetUpdateResponse).Update) + }) + } +} From 53d21ce02f431d1ab75fea1bb63d5864643eeb8e Mon Sep 17 00:00:00 2001 From: "chenyue.ma" Date: Mon, 9 Nov 2026 04:50:38 +0000 Subject: [PATCH 2/4] Address refresh selector review feedback --- .nextchanges/cli/pipeline-refresh-by-file.md | 2 +- .../run/refresh-flags/databricks.yml | 21 +- .../pipelines/run/refresh-flags/out.test.toml | 2 +- .../pipelines/run/refresh-flags/output.txt | 197 +++++++++--------- .../run/refresh-flags/pipeline_file.py | 9 +- acceptance/pipelines/run/refresh-flags/script | 58 +++++- .../refresh-flags/src/pipeline_notebook.sql | 6 + .../pipelines/run/refresh-flags/test.toml | 2 + bundle/bundle.go | 21 ++ .../mutator/resolve_job_run_file_triggers.go | 11 +- .../mutator/resolve_variable_references.go | 2 +- bundle/config/mutator/translate_paths.go | 10 +- bundle/run/pipeline_options.go | 8 +- .../run/pipeline_refresh.go | 61 ++++-- .../run/pipeline_refresh_test.go | 40 +++- cmd/bundle/run.go | 6 + cmd/pipelines/run.go | 41 ++-- cmd/pipelines/run_test.go | 36 ++++ 18 files changed, 341 insertions(+), 192 deletions(-) create mode 100644 acceptance/pipelines/run/refresh-flags/src/pipeline_notebook.sql rename cmd/pipelines/refresh.go => bundle/run/pipeline_refresh.go (61%) rename cmd/pipelines/refresh_test.go => bundle/run/pipeline_refresh_test.go (77%) diff --git a/.nextchanges/cli/pipeline-refresh-by-file.md b/.nextchanges/cli/pipeline-refresh-by-file.md index c5e25a11058..bf3a0452e79 100644 --- a/.nextchanges/cli/pipeline-refresh-by-file.md +++ b/.nextchanges/cli/pipeline-refresh-by-file.md @@ -1 +1 @@ -* Support source files in `pipelines run --refresh` and `--full-refresh` selectors. ([#7184](https://github.com/databricks/cli/pull/7184)) +* Support source files in `pipelines run` and `bundle run` with `--refresh` and `--full-refresh`. ([#7184](https://github.com/databricks/cli/pull/7184)) diff --git a/acceptance/pipelines/run/refresh-flags/databricks.yml b/acceptance/pipelines/run/refresh-flags/databricks.yml index 37e89920ad4..7fe4b3f27b5 100644 --- a/acceptance/pipelines/run/refresh-flags/databricks.yml +++ b/acceptance/pipelines/run/refresh-flags/databricks.yml @@ -1,10 +1,29 @@ bundle: name: test-pipeline-run-flags +variables: + unique_name: + description: Unique name for test resources + +workspace: + root_path: ~/.bundle/${bundle.name}/${var.unique_name} + +presets: + pipelines_development: true + resources: + schemas: + test_schema: + name: pipeline_refresh_${var.unique_name} + catalog_name: main pipelines: my_pipeline: - name: test-pipeline-flags + name: test-pipeline-flags-${var.unique_name} + catalog: main + schema: ${resources.schemas.test_schema.name} + serverless: true libraries: - file: path: pipeline_file.py + - notebook: + path: src/pipeline_notebook.sql diff --git a/acceptance/pipelines/run/refresh-flags/out.test.toml b/acceptance/pipelines/run/refresh-flags/out.test.toml index 18b1a88417e..c7c6f58ed6e 100644 --- a/acceptance/pipelines/run/refresh-flags/out.test.toml +++ b/acceptance/pipelines/run/refresh-flags/out.test.toml @@ -1 +1 @@ -Cloud = false +Cloud = true diff --git a/acceptance/pipelines/run/refresh-flags/output.txt b/acceptance/pipelines/run/refresh-flags/output.txt index e8ac6594453..10ff83edfcf 100644 --- a/acceptance/pipelines/run/refresh-flags/output.txt +++ b/acceptance/pipelines/run/refresh-flags/output.txt @@ -1,75 +1,46 @@ >>> [CLI] pipelines deploy -Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-pipeline-run-flags/default/files... +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-pipeline-run-flags/[UNIQUE_NAME]/files... Created pipelines.my_pipeline -Files: 2 uploaded, 0 deleted -Resources: 1 created, 0 changed, 0 deleted, 0 unchanged +Created schemas.test_schema +Files: 3 uploaded, 0 deleted +Resources: 2 created, 0 changed, 0 deleted, 0 unchanged View your pipeline my_pipeline here: [DATABRICKS_URL]/pipelines/[UUID]?w=[WORKSPACE_ID] +View your schema test_schema here: [DATABRICKS_URL]/explore/data/main/pipeline_refresh_[UNIQUE_NAME]?w=[WORKSPACE_ID] -=== Running pipeline with --refresh flag and specific tables ->>> [CLI] pipelines run --refresh table1,table2 -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 [table1, table2] +=== Initialize all tables with --full-refresh-all +>>> run_pipeline --full-refresh-all +• All tables are fully refreshed >>> print_requests.py //pipelines { "method": "POST", "path": "/api/2.0/pipelines/[UUID]/updates", "body": { - "refresh_selection": [ - "table1", - "table2" - ] + "development": true, + "full_refresh": true } } -=== Running pipeline with --full-refresh-all flag ->>> [CLI] pipelines run --full-refresh-all -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: -• All tables are fully refreshed +=== Running pipeline with --refresh flag and specific tables +>>> run_pipeline --refresh table1,table2 +• Refreshed [table1, table2] >>> print_requests.py //pipelines { "method": "POST", "path": "/api/2.0/pipelines/[UUID]/updates", "body": { - "full_refresh": true + "development": true, + "refresh_selection": [ + "table1", + "table2" + ] } } === Running pipeline with --full-refresh flag and specific tables ->>> [CLI] pipelines run --full-refresh table1,table2 -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: +>>> run_pipeline --full-refresh table1,table2 • Full refreshed [table1, table2] >>> print_requests.py //pipelines @@ -77,6 +48,7 @@ Pipeline configurations for this update: "method": "POST", "path": "/api/2.0/pipelines/[UUID]/updates", "body": { + "development": true, "full_refresh_selection": [ "table1", "table2" @@ -85,18 +57,7 @@ Pipeline configurations for this update: } === Running pipeline with --full-refresh flag and --refresh flag ->>> [CLI] pipelines run --full-refresh table1,table2 --refresh table3,table4 -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: +>>> run_pipeline --full-refresh table1,table2 --refresh table3,table4 • Refreshed [table3, table4] • Full refreshed [table1, table2] @@ -105,6 +66,7 @@ Pipeline configurations for this update: "method": "POST", "path": "/api/2.0/pipelines/[UUID]/updates", "body": { + "development": true, "full_refresh_selection": [ "table1", "table2" @@ -117,94 +79,123 @@ 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 +>>> run_pipeline --refresh pipeline_file.py +• Refreshed tables defined in [/Workspace/Users/[USERNAME]/.bundle/test-pipeline-run-flags/[UNIQUE_NAME]/files/pipeline_file.py] -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] +>>> print_requests.py //pipelines +{ + "method": "POST", + "path": "/api/2.0/pipelines/[UUID]/updates", + "body": { + "development": true, + "refresh_selection_by_path": [ + "/Workspace/Users/[USERNAME]/.bundle/test-pipeline-run-flags/[UNIQUE_NAME]/files/pipeline_file.py" + ] + } +} -Update ID: [UUID] +=== Fully refreshing tables defined in a source file +>>> run_pipeline --full-refresh pipeline_file.py +• Full refreshed tables defined in [/Workspace/Users/[USERNAME]/.bundle/test-pipeline-run-flags/[UNIQUE_NAME]/files/pipeline_file.py] -Update for pipeline completed successfully. +>>> print_requests.py //pipelines +{ + "method": "POST", + "path": "/api/2.0/pipelines/[UUID]/updates", + "body": { + "development": true, + "full_refresh_selection_by_path": [ + "/Workspace/Users/[USERNAME]/.bundle/test-pipeline-run-flags/[UNIQUE_NAME]/files/pipeline_file.py" + ] + } +} -Pipeline configurations for this update: -• Refreshed tables defined in [/Workspace/Users/[USERNAME]/.bundle/test-pipeline-run-flags/default/files/pipeline_file.py] +=== Refreshing a source notebook +>>> run_pipeline --refresh src/pipeline_notebook.sql +• Refreshed tables defined in [/Workspace/Users/[USERNAME]/.bundle/test-pipeline-run-flags/[UNIQUE_NAME]/files/src/pipeline_notebook] >>> print_requests.py //pipelines { "method": "POST", "path": "/api/2.0/pipelines/[UUID]/updates", "body": { + "development": true, "refresh_selection_by_path": [ - "/Workspace/Users/[USERNAME]/.bundle/test-pipeline-run-flags/default/files/pipeline_file.py" + "/Workspace/Users/[USERNAME]/.bundle/test-pipeline-run-flags/[UNIQUE_NAME]/files/src/pipeline_notebook" ] } } -=== 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] +=== Refreshing an absolute workspace path +>>> run_pipeline --refresh /Workspace/Users/[USERNAME]/.bundle/test-pipeline-run-flags/[UNIQUE_NAME]/files/pipeline_file.py +• Refreshed tables defined in [/Workspace/Users/[USERNAME]/.bundle/test-pipeline-run-flags/[UNIQUE_NAME]/files/pipeline_file.py] -Update ID: [UUID] +>>> print_requests.py //pipelines +{ + "method": "POST", + "path": "/api/2.0/pipelines/[UUID]/updates", + "body": { + "development": true, + "refresh_selection_by_path": [ + "/Workspace/Users/[USERNAME]/.bundle/test-pipeline-run-flags/[UNIQUE_NAME]/files/pipeline_file.py" + ] + } +} -Update for pipeline completed successfully. +=== Refreshing from a subdirectory +>>> withdir src run_pipeline --refresh ../pipeline_file.py +• Refreshed tables defined in [/Workspace/Users/[USERNAME]/.bundle/test-pipeline-run-flags/[UNIQUE_NAME]/files/pipeline_file.py] -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": { + "development": true, + "refresh_selection_by_path": [ + "/Workspace/Users/[USERNAME]/.bundle/test-pipeline-run-flags/[UNIQUE_NAME]/files/pipeline_file.py" + ] + } +} +=== Bundle run supports both file flags >>> print_requests.py //pipelines { "method": "POST", "path": "/api/2.0/pipelines/[UUID]/updates", "body": { + "development": true, "full_refresh_selection_by_path": [ - "/Workspace/Users/[USERNAME]/.bundle/test-pipeline-run-flags/default/files/pipeline_file.py" + "/Workspace/Users/[USERNAME]/.bundle/test-pipeline-run-flags/[UNIQUE_NAME]/files/src/pipeline_notebook" + ], + "refresh_selection_by_path": [ + "/Workspace/Users/[USERNAME]/.bundle/test-pipeline-run-flags/[UNIQUE_NAME]/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 +>>> print_requests.py //pipelines === 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 +Error: cannot mix table names and source files in --full-refresh; use only table names or only source files -If you've made local changes, run 'databricks pipelines deploy' first to ensure they are included. +>>> print_requests.py //pipelines -Error: cannot mix table names and source files in --full-refresh; use only table names or only source files +>>> musterr [CLI] bundle run my_pipeline --refresh table1,pipeline_file.py --restart --no-wait +Error: cannot mix table names and source files in --refresh; use only table names or only source files ->>> test ! -e [OUTPUT_DIR]/out.requests.txt +>>> print_requests.py //pipelines diff --git a/acceptance/pipelines/run/refresh-flags/pipeline_file.py b/acceptance/pipelines/run/refresh-flags/pipeline_file.py index aa0c44e3f6d..51dcf9ed5d7 100644 --- a/acceptance/pipelines/run/refresh-flags/pipeline_file.py +++ b/acceptance/pipelines/run/refresh-flags/pipeline_file.py @@ -1,7 +1,14 @@ import dlt from pyspark.sql import SparkSession +spark = SparkSession.builder.getOrCreate() + @dlt.table -def my_table(): +def table1(): return spark.range(10) + + +@dlt.table +def table2(): + return spark.range(20) diff --git a/acceptance/pipelines/run/refresh-flags/script b/acceptance/pipelines/run/refresh-flags/script index e2838af2063..6ad3daf72a4 100644 --- a/acceptance/pipelines/run/refresh-flags/script +++ b/acceptance/pipelines/run/refresh-flags/script @@ -1,37 +1,75 @@ +export BUNDLE_VAR_unique_name="$UNIQUE_NAME" +cleanup() { + $CLI bundle destroy --auto-approve &> LOG.cleanup +} +trap cleanup EXIT + +run_log="$PWD/LOG.run" + +# Cloud event logs and durations vary between runs. +run_pipeline() { + $CLI pipelines run "$@" &> "$run_log" + contains.py "completed successfully." < "$run_log" > "$run_log.checked" + grep -E '^• (All tables are |Refreshed |Full refreshed )' "$run_log" +} + trace $CLI pipelines deploy +remote=$($CLI bundle summary -o json | jq -r .workspace.file_path) rm "$OUT_REQUESTS" -title "Running pipeline with --refresh flag and specific tables" -trace $CLI pipelines run --refresh table1,table2 +title "Initialize all tables with --full-refresh-all" +trace run_pipeline --full-refresh-all trace print_requests.py //pipelines -title "Running pipeline with --full-refresh-all flag" -trace $CLI pipelines run --full-refresh-all +title "Running pipeline with --refresh flag and specific tables" +trace run_pipeline --refresh table1,table2 trace print_requests.py //pipelines title "Running pipeline with --full-refresh flag and specific tables" -trace $CLI pipelines run --full-refresh table1,table2 +trace run_pipeline --full-refresh table1,table2 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 run_pipeline --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 run_pipeline --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 run_pipeline --full-refresh pipeline_file.py +trace print_requests.py //pipelines + +title "Refreshing a source notebook" +trace run_pipeline --refresh src/pipeline_notebook.sql +trace print_requests.py //pipelines + +title "Refreshing an absolute workspace path" +trace run_pipeline --refresh "$remote/pipeline_file.py" +trace print_requests.py //pipelines + +title "Refreshing from a subdirectory" +trace withdir src run_pipeline --refresh ../pipeline_file.py +trace print_requests.py //pipelines + +title "Bundle run supports both file flags" +trace $CLI bundle run my_pipeline --refresh pipeline_file.py --full-refresh src/pipeline_notebook.sql &> LOG.bundle 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" +touch "$OUT_REQUESTS" +trace print_requests.py //pipelines 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" +touch "$OUT_REQUESTS" +trace print_requests.py //pipelines + +trace musterr $CLI bundle run my_pipeline --refresh table1,pipeline_file.py --restart --no-wait +touch "$OUT_REQUESTS" +trace print_requests.py //pipelines diff --git a/acceptance/pipelines/run/refresh-flags/src/pipeline_notebook.sql b/acceptance/pipelines/run/refresh-flags/src/pipeline_notebook.sql new file mode 100644 index 00000000000..be956582f1c --- /dev/null +++ b/acceptance/pipelines/run/refresh-flags/src/pipeline_notebook.sql @@ -0,0 +1,6 @@ +-- Databricks notebook source +CREATE OR REFRESH MATERIALIZED VIEW table3 AS SELECT 3 AS id; + +-- COMMAND ---------- + +CREATE OR REFRESH MATERIALIZED VIEW table4 AS SELECT 4 AS id; diff --git a/acceptance/pipelines/run/refresh-flags/test.toml b/acceptance/pipelines/run/refresh-flags/test.toml index 159efe02696..a86fb37aa56 100644 --- a/acceptance/pipelines/run/refresh-flags/test.toml +++ b/acceptance/pipelines/run/refresh-flags/test.toml @@ -1 +1,3 @@ +Cloud = true RecordRequests = true +Env.MSYS_NO_PATHCONV = "1" diff --git a/bundle/bundle.go b/bundle/bundle.go index 6762b38b91e..39f67f100bb 100644 --- a/bundle/bundle.go +++ b/bundle/bundle.go @@ -393,6 +393,27 @@ func (b *Bundle) InternalDir(ctx context.Context) (string, error) { return dir, nil } +// GetSyncFiles lists files matching the bundle sync configuration. +func (b *Bundle) GetSyncFiles(ctx context.Context) ([]fileset.File, error) { + includes, err := b.GetSyncIncludePatterns(ctx) + if err != nil { + return nil, err + } + list, err := libsync.NewFileList(ctx, b.WorktreeRoot, b.SyncRoot, b.Config.Sync.Paths, includes, b.Config.Sync.Exclude) + if err != nil { + return nil, err + } + return list.Files(ctx) +} + +// RemoteFilePath returns the initialized sync destination. +func (b *Bundle) RemoteFilePath() string { + if b.IsSourceLinkedDeployment() { + return filepath.ToSlash(b.SyncRootPath) + } + return b.Config.Workspace.FilePath +} + // GetSyncIncludePatterns returns a list of user defined includes // And also adds InternalDir folder to include list for sync command // so this folder is always synced diff --git a/bundle/config/mutator/resolve_job_run_file_triggers.go b/bundle/config/mutator/resolve_job_run_file_triggers.go index 35cffd0dcbe..32dd9c90ab7 100644 --- a/bundle/config/mutator/resolve_job_run_file_triggers.go +++ b/bundle/config/mutator/resolve_job_run_file_triggers.go @@ -15,7 +15,6 @@ import ( "github.com/databricks/cli/bundle" "github.com/databricks/cli/bundle/config/resources" "github.com/databricks/cli/libs/diag" - libsync "github.com/databricks/cli/libs/sync" ) // Every diagnostic below is reported against an on_file_change entry. @@ -85,15 +84,7 @@ func listSyncableRelPaths(ctx context.Context, b *bundle.Bundle) ([]string, erro // additionally drops files whose notebook type it cannot determine, such as a // malformed .ipynb. Those stay watched on purpose - editing one is exactly when the // run should re-fire - so a pattern can hash a file that this deploy will not upload. - includes, err := b.GetSyncIncludePatterns(ctx) - if err != nil { - return nil, err - } - fl, err := libsync.NewFileList(ctx, b.WorktreeRoot, b.SyncRoot, b.Config.Sync.Paths, includes, b.Config.Sync.Exclude) - if err != nil { - return nil, err - } - files, err := fl.Files(ctx) + files, err := b.GetSyncFiles(ctx) if err != nil { return nil, err } diff --git a/bundle/config/mutator/resolve_variable_references.go b/bundle/config/mutator/resolve_variable_references.go index 42c22b61533..37736ea8504 100644 --- a/bundle/config/mutator/resolve_variable_references.go +++ b/bundle/config/mutator/resolve_variable_references.go @@ -114,7 +114,7 @@ func ResolveVolumePathReferencesOnlyResources() bundle.Mutator { func lookup(v dyn.Value, path dyn.Path, b *bundle.Bundle) (dyn.Value, error) { if b.IsSourceLinkedDeployment() { if path.String() == "workspace.file_path" { - return dyn.V(b.SyncRootPath), nil + return dyn.V(b.RemoteFilePath()), nil } } // Future opportunity: if we lookup this path in both the given root diff --git a/bundle/config/mutator/translate_paths.go b/bundle/config/mutator/translate_paths.go index 956d010ec95..a2a32d14c06 100644 --- a/bundle/config/mutator/translate_paths.go +++ b/bundle/config/mutator/translate_paths.go @@ -15,7 +15,6 @@ import ( "github.com/databricks/cli/bundle/libraries" "github.com/databricks/cli/bundle" - "github.com/databricks/cli/bundle/config/resources" "github.com/databricks/cli/libs/diag" "github.com/databricks/cli/libs/dyn" "github.com/databricks/cli/libs/notebook" @@ -320,8 +319,7 @@ func (t *translateContext) rewriteValue(ctx context.Context, p dyn.Path, v dyn.V } func applyTranslations(ctx context.Context, b *bundle.Bundle, t *translateContext, translations []func(context.Context, dyn.Value) (dyn.Value, error)) diag.Diagnostics { - switch { - case b.IsImmutableFolder(): + if b.IsImmutableFolder() { // Reject an explicit workspace.file_path: immutable bundles set it // automatically to the content-addressed snapshot location. A user-supplied // value would be silently discarded during path translation, so we error early. @@ -332,12 +330,8 @@ func applyTranslations(ctx context.Context, b *bundle.Bundle, t *translateContex Locations: []diag.Location{loc}, }} } - t.remoteRoot = resources.SnapshotFullPathRef + "/files" - case t.b.IsSourceLinkedDeployment(): - t.remoteRoot = t.b.SyncRootPath - default: - t.remoteRoot = t.b.Config.Workspace.FilePath } + t.remoteRoot = b.RemoteFilePath() err := b.Config.Mutate(func(v dyn.Value) (dyn.Value, error) { var err error diff --git a/bundle/run/pipeline_options.go b/bundle/run/pipeline_options.go index 164e8a2de61..e6a2a6753f9 100644 --- a/bundle/run/pipeline_options.go +++ b/bundle/run/pipeline_options.go @@ -15,7 +15,7 @@ type PipelineOptions struct { // Perform a full graph update. RefreshAll bool - // List of tables to update. + // List of tables or source files to update. Refresh []string RefreshByPath []string @@ -23,7 +23,7 @@ type PipelineOptions struct { // Perform a full graph reset and recompute. FullRefreshAll bool - // List of tables to reset and recompute. + // List of tables or source files to reset and recompute. FullRefresh []string FullRefreshByPath []string @@ -34,9 +34,9 @@ type PipelineOptions struct { func (o *PipelineOptions) Define(fs *flag.FlagSet) { fs.BoolVar(&o.RefreshAll, "refresh-all", false, "Perform a full graph update.") - fs.StringSliceVar(&o.Refresh, "refresh", nil, "List of tables to update.") + fs.StringSliceVar(&o.Refresh, "refresh", nil, "List of tables or source files to update. Local paths are relative to the current directory.") fs.BoolVar(&o.FullRefreshAll, "full-refresh-all", false, "Perform a full graph reset and recompute.") - fs.StringSliceVar(&o.FullRefresh, "full-refresh", nil, "List of tables to reset and recompute.") + fs.StringSliceVar(&o.FullRefresh, "full-refresh", nil, "List of tables or source files to reset and recompute. Local paths are relative to the current directory.") fs.BoolVar(&o.ValidateOnly, "validate-only", false, "Perform an update to validate graph correctness.") } diff --git a/cmd/pipelines/refresh.go b/bundle/run/pipeline_refresh.go similarity index 61% rename from cmd/pipelines/refresh.go rename to bundle/run/pipeline_refresh.go index d26da6e33a8..997579f8e44 100644 --- a/cmd/pipelines/refresh.go +++ b/bundle/run/pipeline_refresh.go @@ -1,4 +1,4 @@ -package pipelines +package run import ( "context" @@ -11,14 +11,36 @@ import ( "strings" "github.com/databricks/cli/bundle" + "github.com/databricks/cli/libs/dyn" "github.com/databricks/cli/libs/notebook" "github.com/databricks/cli/libs/sync" ) -// resolveRefreshSelectors separates table names from source files and translates local files to workspace paths. -func resolveRefreshSelectors(ctx context.Context, b *bundle.Bundle, selectors []string) (tables, paths []string, err error) { - var remoteNames map[string]string +// ResolveRefreshSelectors resolves source files to workspace paths. +func (o *PipelineOptions) ResolveRefreshSelectors(ctx context.Context, b *bundle.Bundle) error { + r := refreshSelectorResolver{bundle: b} + var err error + o.Refresh, o.RefreshByPath, err = r.resolve(ctx, o.Refresh) + if err != nil { + return fmt.Errorf("invalid --refresh: %w", err) + } + o.FullRefresh, o.FullRefreshByPath, err = r.resolve(ctx, o.FullRefresh) + if err != nil { + return fmt.Errorf("invalid --full-refresh: %w", err) + } + // Validate before Restart stops an active update. + return o.Validate(nil) +} + +type refreshSelectorResolver struct { + bundle *bundle.Bundle + remoteNames map[string]string +} + +func (r *refreshSelectorResolver) resolve(ctx context.Context, selectors []string) (tables, paths []string, err error) { + b := r.bundle for _, selector := range selectors { + // /Workspace/ distinguishes remote paths from local paths like /Users/... on macOS. if strings.HasPrefix(selector, "/Workspace/") { paths = append(paths, selector) continue @@ -58,23 +80,26 @@ func resolveRefreshSelectors(ctx context.Context, b *bundle.Bundle, selectors [] if !filepath.IsLocal(rel) { return nil, nil, fmt.Errorf("source file %q is outside the bundle sync root", selector) } - if remoteNames == nil { - remoteNames, err = refreshSourcePaths(ctx, b) + if r.remoteNames == nil { + // Refresh uses the deployed snapshot. + if b.IsImmutableFolder() { + if err := b.DeploymentBundle.ResolveConfigAgainstState(&b.Config, dyn.MustPathFromString("workspace.file_path")); err != nil { + return nil, nil, err + } + } + r.remoteNames, err = refreshSourcePaths(ctx, b) if err != nil { return nil, nil, err } } - remoteName, ok := remoteNames[filepath.ToSlash(rel)] + remoteName, ok := r.remoteNames[filepath.ToSlash(rel)] if !ok { if _, _, err := notebook.DetectWithFS(b.SyncRoot, filepath.ToSlash(rel)); err != nil { return nil, nil, fmt.Errorf("cannot identify source file %q: %w", selector, err) } - return nil, nil, fmt.Errorf("source file %q is excluded from bundle sync; include it and run 'databricks pipelines deploy' first", selector) - } - remoteRoot := b.Config.Workspace.FilePath - if b.IsSourceLinkedDeployment() { - remoteRoot = filepath.ToSlash(b.SyncRootPath) + return nil, nil, fmt.Errorf("source file %q is excluded from bundle sync; include it and deploy first", selector) } + remoteRoot := b.RemoteFilePath() if strings.Contains(remoteRoot, "${") { return nil, nil, fmt.Errorf("cannot resolve the deployed path for %q; use its absolute /Workspace/ path", selector) } @@ -83,17 +108,9 @@ func resolveRefreshSelectors(ctx context.Context, b *bundle.Bundle, selectors [] return tables, paths, nil } -// refreshSourcePaths uses the sync mapping so notebook names and sync exclusions match deployment. +// refreshSourcePaths maps local files to their deployed names. func refreshSourcePaths(ctx context.Context, b *bundle.Bundle) (map[string]string, error) { - includes, err := b.GetSyncIncludePatterns(ctx) - if err != nil { - return nil, err - } - list, err := sync.NewFileList(ctx, b.WorktreeRoot, b.SyncRoot, b.Config.Sync.Paths, includes, b.Config.Sync.Exclude) - if err != nil { - return nil, err - } - files, err := list.Files(ctx) + files, err := b.GetSyncFiles(ctx) if err != nil { return nil, err } diff --git a/cmd/pipelines/refresh_test.go b/bundle/run/pipeline_refresh_test.go similarity index 77% rename from cmd/pipelines/refresh_test.go rename to bundle/run/pipeline_refresh_test.go index eb1824a5890..27b9c098bd1 100644 --- a/cmd/pipelines/refresh_test.go +++ b/bundle/run/pipeline_refresh_test.go @@ -1,4 +1,4 @@ -package pipelines +package run import ( "os" @@ -8,6 +8,8 @@ import ( "github.com/databricks/cli/bundle" "github.com/databricks/cli/bundle/config" + "github.com/databricks/cli/bundle/config/resources" + "github.com/databricks/cli/bundle/direct/dstate" "github.com/databricks/cli/libs/vfs" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -74,7 +76,7 @@ func TestResolveRefreshSelectors(t *testing.T) { t.Chdir(root) b := refreshTestBundle(root) b.Config.Sync.Exclude = []string{"src/excluded.sql"} - tables, paths, err := resolveRefreshSelectors(t.Context(), b, tt.selectors) + tables, paths, err := (&refreshSelectorResolver{bundle: b}).resolve(t.Context(), tt.selectors) if tt.err != "" { require.ErrorContains(t, err, tt.err) return @@ -108,7 +110,7 @@ func TestResolveRefreshSelectorsFromSubdirectory(t *testing.T) { b := refreshTestBundle(root) b.BundleRootPath = filepath.Join(root, "project") for _, selector := range []string{"../../shared.sql", filepath.Join(root, "shared.sql")} { - tables, paths, err := resolveRefreshSelectors(t.Context(), b, []string{selector}) + tables, paths, err := (&refreshSelectorResolver{bundle: b}).resolve(t.Context(), []string{selector}) require.NoError(t, err) assert.Empty(t, tables) assert.Equal(t, []string{"/Workspace/Shared/project/files/shared.sql"}, paths) @@ -120,7 +122,7 @@ func TestResolveRefreshSelectorsOutsideSyncRoot(t *testing.T) { require.NoError(t, os.Mkdir(filepath.Join(root, "project"), 0o700)) require.NoError(t, os.WriteFile(filepath.Join(root, "outside.sql"), []byte("SELECT 1;\n"), 0o600)) t.Chdir(filepath.Join(root, "project")) - _, _, err := resolveRefreshSelectors(t.Context(), refreshTestBundle(filepath.Join(root, "project")), []string{"../outside.sql"}) + _, _, err := (&refreshSelectorResolver{bundle: refreshTestBundle(filepath.Join(root, "project"))}).resolve(t.Context(), []string{"../outside.sql"}) require.ErrorContains(t, err, "is outside the bundle sync root") } @@ -148,7 +150,7 @@ func TestResolveRefreshSelectorsDeploymentMode(t *testing.T) { b := refreshTestBundle(root) b.Config.Workspace.SyncMode = tt.mode b.Config.Presets.SourceLinkedDeployment = tt.sourceLinked - tables, paths, err := resolveRefreshSelectors(t.Context(), b, []string{"query.sql"}) + tables, paths, err := (&refreshSelectorResolver{bundle: b}).resolve(t.Context(), []string{"query.sql"}) require.NoError(t, err) assert.Empty(t, tables) assert.Equal(t, []string{tt.want}, paths) @@ -165,6 +167,32 @@ func TestResolveRefreshSelectorsUnresolvedWorkspacePath(t *testing.T) { b.Config.Workspace.SyncMode = config.SyncModeImmutable enabled := true b.Config.Presets.SourceLinkedDeployment = &enabled - _, _, err := resolveRefreshSelectors(t.Context(), b, []string{"query.sql"}) + b.DeploymentBundle.StateDB.OpenWithData("state", dstate.Database{}) + _, _, err := (&refreshSelectorResolver{bundle: b}).resolve(t.Context(), []string{"query.sql"}) require.ErrorContains(t, err, "use its absolute /Workspace/ path") } + +func TestResolveRefreshSelectorsDeployedSnapshot(t *testing.T) { + root := t.TempDir() + for _, name := range []string{"query.sql", "full_query.sql"} { + require.NoError(t, os.WriteFile(filepath.Join(root, name), []byte("SELECT 2;\n"), 0o600)) + } + t.Chdir(root) + b := refreshTestBundle(root) + b.Config.Workspace.SyncMode = config.SyncModeImmutable + b.Config.Workspace.FilePath = resources.SnapshotFullPathRef + "/files" + b.DeploymentBundle.StateDB.OpenWithData("state", dstate.Database{ + State: map[string]dstate.ResourceEntry{ + resources.SnapshotKey: { + ID: "deployed-snapshot", + State: []byte(`{"full_path":"/Workspace/Shared/snapshots/deployed/snapshot"}`), + }, + }, + }) + opts := PipelineOptions{Refresh: []string{"query.sql"}, FullRefresh: []string{"full_query.sql"}} + require.NoError(t, opts.ResolveRefreshSelectors(t.Context(), b)) + assert.Empty(t, opts.Refresh) + assert.Empty(t, opts.FullRefresh) + assert.Equal(t, []string{"/Workspace/Shared/snapshots/deployed/snapshot/files/query.sql"}, opts.RefreshByPath) + assert.Equal(t, []string{"/Workspace/Shared/snapshots/deployed/snapshot/files/full_query.sql"}, opts.FullRefreshByPath) +} diff --git a/cmd/bundle/run.go b/cmd/bundle/run.go index 17199a6cabf..d27c3a12695 100644 --- a/cmd/bundle/run.go +++ b/cmd/bundle/run.go @@ -206,6 +206,12 @@ Example usage: return err } + if ref.Description.SingularName == "pipeline" { + if err := runOptions.Pipeline.ResolveRefreshSelectors(ctx, b); err != nil { + return err + } + } + runOptions.NoWait = noWait var runOutput output.RunOutput if restart { diff --git a/cmd/pipelines/run.go b/cmd/pipelines/run.go index 702a8392edf..f86fedb3b70 100644 --- a/cmd/pipelines/run.go +++ b/cmd/pipelines/run.go @@ -17,6 +17,7 @@ import ( bundleresources "github.com/databricks/cli/bundle/resources" "github.com/databricks/cli/bundle/run" bundlerunoutput "github.com/databricks/cli/bundle/run/output" + "github.com/databricks/cli/bundle/statemgmt" "github.com/databricks/cli/cmd/bundle/utils" "github.com/databricks/cli/cmd/root" "github.com/databricks/cli/libs/cmdgroup" @@ -179,15 +180,18 @@ func fetchAndDisplayPipelineUpdate(ctx context.Context, w *databricks.WorkspaceC latestUpdate := *getUpdateResponse.Update - params := &PipelineEventsQueryParams{ - Filter: fmt.Sprintf("update_id='%s' AND event_type='update_progress'", updateId), - OrderBy: "timestamp asc", - } - - events, err := fetchAllPipelineEvents(ctx, w, pipelineId, params) + // The public events API rejects event_type filters. + events, err := w.Pipelines.ListPipelineEventsAll(ctx, pipelines.ListPipelineEventsRequest{ + PipelineId: pipelineId, + Filter: fmt.Sprintf("update_id='%s'", updateId), + OrderBy: []string{"timestamp asc"}, + }) if err != nil { return err } + events = slices.DeleteFunc(events, func(event pipelines.PipelineEvent) bool { + return event.EventType != "update_progress" + }) err = displayPipelineUpdate(ctx, latestUpdate, pipelineId, events) if err != nil { @@ -264,6 +268,10 @@ a file, or quote table identifiers with backticks to select an ambiguous table.` ctx := logdiag.InitContext(cmd.Context()) cmd.SetContext(ctx) + runOptions := run.Options{ + Pipeline: run.PipelineOptions{Refresh: refresh, FullRefresh: fullRefresh, FullRefreshAll: fullRefreshAll}, + NoWait: noWait, + } var key string b, err := utils.ProcessBundle(cmd, utils.ProcessOptions{ PostInitFunc: func(ctx context.Context, b *bundle.Bundle) error { @@ -271,6 +279,9 @@ a file, or quote table identifiers with backticks to select an ambiguous table.` key, _, err = resolveRunArgument(ctx, b, args) return err }, + PostStateFunc: func(ctx context.Context, b *bundle.Bundle, _ *statemgmt.StateDesc) error { + return runOptions.Pipeline.ResolveRefreshSelectors(ctx, b) + }, ErrorOnEmptyState: true, SkipInitContext: true, }) @@ -285,24 +296,6 @@ a file, or quote table identifiers with backticks to select an ambiguous table.` return err } - runOptions := run.Options{ - Pipeline: run.PipelineOptions{ - FullRefreshAll: fullRefreshAll, - }, - NoWait: noWait, - } - runOptions.Pipeline.Refresh, runOptions.Pipeline.RefreshByPath, err = resolveRefreshSelectors(ctx, b, refresh) - if err != nil { - return fmt.Errorf("invalid --refresh: %w", err) - } - runOptions.Pipeline.FullRefresh, runOptions.Pipeline.FullRefreshByPath, err = resolveRefreshSelectors(ctx, b, fullRefresh) - if err != nil { - return fmt.Errorf("invalid --full-refresh: %w", err) - } - if err := runOptions.Pipeline.Validate(nil); err != nil { - return err - } - var runOutput bundlerunoutput.RunOutput if restart { runOutput, err = runner.Restart(ctx, &runOptions) diff --git a/cmd/pipelines/run_test.go b/cmd/pipelines/run_test.go index 2c910669e34..246380a8c8e 100644 --- a/cmd/pipelines/run_test.go +++ b/cmd/pipelines/run_test.go @@ -5,10 +5,46 @@ import ( "time" "github.com/databricks/cli/libs/cmdio" + "github.com/databricks/cli/libs/testserver" + "github.com/databricks/databricks-sdk-go" "github.com/databricks/databricks-sdk-go/service/pipelines" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) +func TestFetchAndDisplayPipelineUpdateFiltersProgressAcrossPages(t *testing.T) { + server := testserver.New(t) + server.Handle("GET", "/api/2.0/pipelines/pipeline-1/updates/update-1", func(req testserver.Request) any { + return pipelines.GetUpdateResponse{Update: &pipelines.UpdateInfo{UpdateId: "update-1"}} + }) + server.Handle("GET", "/api/2.0/pipelines/pipeline-1/events", func(req testserver.Request) any { + if req.URL.Query().Get("page_token") == "" { + assert.Equal(t, "update_id='update-1'", req.URL.Query().Get("filter")) + return pipelines.ListPipelineEventsResponse{ + NextPageToken: "next-page", + Events: []pipelines.PipelineEvent{ + {EventType: "update_progress", Timestamp: "2026-01-01T00:00:00Z", Message: "Update is WAITING_FOR_RESOURCES."}, + {EventType: "flow_progress", Timestamp: "2026-01-01T00:00:05Z", Message: "Flow is PLANNING."}, + }, + } + } + assert.Equal(t, "next-page", req.URL.Query().Get("page_token")) + return pipelines.ListPipelineEventsResponse{ + Events: []pipelines.PipelineEvent{ + {EventType: "update_progress", Timestamp: "2026-01-01T00:00:10Z", Message: "Update is COMPLETED."}, + }, + } + }) + w, err := databricks.NewWorkspaceClient(&databricks.Config{Host: server.URL, Token: "test-token"}) + require.NoError(t, err) + ctx, stdout := cmdio.NewTestContextWithStdout(t.Context()) + + err = fetchAndDisplayPipelineUpdate(ctx, w, "pipeline-1", "update-1") + require.NoError(t, err) + assert.Contains(t, stdout.String(), "WAITING_FOR_RESOURCES") + assert.Contains(t, stdout.String(), "10.0s") +} + func TestDisplayPipelineUpdate(t *testing.T) { tests := []struct { name string From 3b8bb1f93a068cb76c3dab9dc6830c09e8122cb0 Mon Sep 17 00:00:00 2001 From: "chenyue.ma" Date: Mon, 9 Nov 2026 06:26:41 +0000 Subject: [PATCH 3/4] Fix Windows refresh acceptance paths --- acceptance/pipelines/run/refresh-flags/script | 2 +- acceptance/pipelines/run/refresh-flags/test.toml | 1 - 2 files changed, 1 insertion(+), 2 deletions(-) diff --git a/acceptance/pipelines/run/refresh-flags/script b/acceptance/pipelines/run/refresh-flags/script index 6ad3daf72a4..2df84af88c7 100644 --- a/acceptance/pipelines/run/refresh-flags/script +++ b/acceptance/pipelines/run/refresh-flags/script @@ -8,7 +8,7 @@ run_log="$PWD/LOG.run" # Cloud event logs and durations vary between runs. run_pipeline() { - $CLI pipelines run "$@" &> "$run_log" + MSYS_NO_PATHCONV=1 $CLI pipelines run "$@" &> "$run_log" contains.py "completed successfully." < "$run_log" > "$run_log.checked" grep -E '^• (All tables are |Refreshed |Full refreshed )' "$run_log" } diff --git a/acceptance/pipelines/run/refresh-flags/test.toml b/acceptance/pipelines/run/refresh-flags/test.toml index a86fb37aa56..ff9f96e2553 100644 --- a/acceptance/pipelines/run/refresh-flags/test.toml +++ b/acceptance/pipelines/run/refresh-flags/test.toml @@ -1,3 +1,2 @@ Cloud = true RecordRequests = true -Env.MSYS_NO_PATHCONV = "1" From d2509d27f8f5b802dc2a986f85792001638fd9c7 Mon Sep 17 00:00:00 2001 From: "chenyue.ma" Date: Mon, 9 Nov 2026 06:55:56 +0000 Subject: [PATCH 4/4] Update bundle run refresh help snapshot --- acceptance/bundle/help/bundle-run/output.txt | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/acceptance/bundle/help/bundle-run/output.txt b/acceptance/bundle/help/bundle-run/output.txt index a6b453c8834..62857b1a5c9 100644 --- a/acceptance/bundle/help/bundle-run/output.txt +++ b/acceptance/bundle/help/bundle-run/output.txt @@ -54,9 +54,9 @@ Job Task Flags: --sql-params stringToString A map from keys to values for jobs with SQL tasks. (default []) Pipeline Flags: - --full-refresh strings List of tables to reset and recompute. + --full-refresh strings List of tables or source files to reset and recompute. Local paths are relative to the current directory. --full-refresh-all Perform a full graph reset and recompute. - --refresh strings List of tables to update. + --refresh strings List of tables or source files to update. Local paths are relative to the current directory. --refresh-all Perform a full graph update. --validate-only Perform an update to validate graph correctness.