From 9d78b59880a35c1eeb73eca08dcc664bf374c1b6 Mon Sep 17 00:00:00 2001 From: "chenyue.ma" Date: Wed, 28 Oct 2026 22:30:25 +0000 Subject: [PATCH] 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) + }) + } +}