Support file selectors in pipelines run

.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))

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
 {
@@ -85,7 +85,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
+>>> [CLI] pipelines run --full-refresh table1 --refresh 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.
@@ -97,20 +97,112 @@ Update ID: [UUID]
 Update for pipeline completed successfully.
 
 Pipeline configurations for this update:
-• All tables are refreshed
+• Refreshed [table2]
+• Full refreshed [table1]
 
 >>> print_requests.py //pipelines
 {
   "method": "POST",
   "path": "/api/2.0/pipelines/[UUID]/updates",
   "body": {
     "full_refresh_selection": [
-      "table1",
-      "table2"
+      "table1"
     ],
     "refresh_selection": [
-      "table3",
-      "table4"
+      "table2"
+    ]
+  }
+}
+
+=== 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

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)

acceptance/pipelines/run/refresh-flags/script

@@ -14,5 +14,24 @@ trace $CLI pipelines run --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 $CLI pipelines run --full-refresh table1 --refresh table2
 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"

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.

bundle/run/pipeline_options_test.go

@@ -1,13 +1,117 @@
 package run
 
 import (
+	"encoding/json"
+	"fmt"
 	"testing"
 
 	flag "github.com/spf13/pflag"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
+func TestPipelineOptionsPathSelectionsPayload(t *testing.T) {
+	tests := []struct {
+		name     string
+		opts     PipelineOptions
+		expected string
+	}{
+		{
+			name: "file selections for both refresh modes",
+			opts: PipelineOptions{
+				RefreshByPath:     []string{"/Workspace/Shared/source.sql"},
+				FullRefreshByPath: []string{"/Workspace/Shared/notebook"},
+			},
+			expected: `{
+				"refresh_selection_by_path": ["/Workspace/Shared/source.sql"],
+				"full_refresh_selection_by_path": ["/Workspace/Shared/notebook"]
+			}`,
+		},
+		{
+			name: "table refresh and file full-refresh",
+			opts: PipelineOptions{
+				Refresh:           []string{"catalog.schema.table1"},
+				FullRefreshByPath: []string{"/Workspace/Shared/notebook"},
+			},
+			expected: `{
+				"refresh_selection": ["catalog.schema.table1"],
+				"full_refresh_selection_by_path": ["/Workspace/Shared/notebook"]
+			}`,
+		},
+		{
+			name: "file refresh and table full-refresh",
+			opts: PipelineOptions{
+				RefreshByPath: []string{"/Workspace/Shared/source.sql"},
+				FullRefresh:   []string{"catalog.schema.table2"},
+			},
+			expected: `{
+				"refresh_selection_by_path": ["/Workspace/Shared/source.sql"],
+				"full_refresh_selection": ["catalog.schema.table2"]
+			}`,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			payload, err := tt.opts.toPayload(nil, "pipeline-id", false)
+			require.NoError(t, err)
+			assert.Equal(t, "pipeline-id", payload.PipelineId)
+			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 TestPipelineOptionsValidatePathSelections(t *testing.T) {
+	for _, full := range []bool{false, true} {
+		for _, exclusive := range []string{"refresh-all", "full-refresh-all", "validate-only"} {
+			t.Run(fmt.Sprintf("full=%t/%s", full, exclusive), func(t *testing.T) {
+				opts := PipelineOptions{
+					RefreshAll:     exclusive == "refresh-all",
+					FullRefreshAll: exclusive == "full-refresh-all",
+					ValidateOnly:   exclusive == "validate-only",
+				}
+				if full {
+					opts.FullRefreshByPath = []string{"/Workspace/Shared/source.sql"}
+				} else {
+					opts.RefreshByPath = []string{"/Workspace/Shared/source.sql"}
+				}
+				_, err := opts.toPayload(nil, "pipeline-id", false)
+				assert.ErrorContains(t, err, "cannot use --refresh or --full-refresh together with --"+exclusive)
+			})
+		}
+	}
+}
+
 func setupPipelineOptions(t *testing.T) (*flag.FlagSet, *PipelineOptions) {
 	var fs flag.FlagSet
 	var opts PipelineOptions