Support file selectors in pipelines run

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

@@ -0,0 +1 @@
+* Support source files in `pipelines run` and `bundle run` with `--refresh` and `--full-refresh`. ([#7184](https://github.com/databricks/cli/pull/7184))

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.
 

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

acceptance/pipelines/run/refresh-flags/out.test.toml

@@ -1 +1 @@
-Cloud = false
+Cloud = true

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

@@ -1,116 +1,201 @@
 
 >>> [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]
+=== Initialize all tables with --full-refresh-all
+>>> run_pipeline --full-refresh-all
+• All tables are fully refreshed
 
-Update ID: [UUID]
-
-Update for pipeline completed successfully.
+>>> print_requests.py //pipelines
+{
+  "method": "POST",
+  "path": "/api/2.0/pipelines/[UUID]/updates",
+  "body": {
+    "development": true,
+    "full_refresh": true
+  }
+}
 
-Pipeline configurations for this update:
-• All tables are 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": {
+    "development": true,
     "refresh_selection": [
       "table1",
       "table2"
     ]
   }
 }
 
-=== 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 refreshed
+=== Running pipeline with --full-refresh flag and specific tables
+>>> run_pipeline --full-refresh table1,table2
+• Full refreshed [table1, table2]
 
 >>> print_requests.py //pipelines
 {
   "method": "POST",
   "path": "/api/2.0/pipelines/[UUID]/updates",
   "body": {
-    "full_refresh": true
+    "development": true,
+    "full_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.
+=== Running pipeline with --full-refresh flag and --refresh flag
+>>> run_pipeline --full-refresh table1,table2 --refresh table3,table4
+• Refreshed [table3, table4]
+• Full refreshed [table1, table2]
 
-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,
+    "full_refresh_selection": [
+      "table1",
+      "table2"
+    ],
+    "refresh_selection": [
+      "table3",
+      "table4"
+    ]
+  }
+}
 
-Update ID: [UUID]
+=== Refreshing tables defined in a source file
+>>> run_pipeline --refresh pipeline_file.py
+• 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,
+    "refresh_selection_by_path": [
+      "/Workspace/Users/[USERNAME]/.bundle/test-pipeline-run-flags/[UNIQUE_NAME]/files/pipeline_file.py"
+    ]
+  }
+}
 
-Pipeline configurations for this update:
-• All tables are refreshed
+=== 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]
 
 >>> print_requests.py //pipelines
 {
   "method": "POST",
   "path": "/api/2.0/pipelines/[UUID]/updates",
   "body": {
-    "full_refresh_selection": [
-      "table1",
-      "table2"
+    "development": true,
+    "full_refresh_selection_by_path": [
+      "/Workspace/Users/[USERNAME]/.bundle/test-pipeline-run-flags/[UNIQUE_NAME]/files/pipeline_file.py"
     ]
   }
 }
 
-=== 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
+=== 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]
 
-If you've made local changes, run 'databricks pipelines deploy' first to ensure they are included.
+>>> 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/src/pipeline_notebook"
+    ]
+  }
+}
 
-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:
-• All tables are refreshed
+>>> 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": {
-    "full_refresh_selection": [
-      "table1",
-      "table2"
+    "development": true,
+    "full_refresh_selection_by_path": [
+      "/Workspace/Users/[USERNAME]/.bundle/test-pipeline-run-flags/[UNIQUE_NAME]/files/src/pipeline_notebook"
     ],
-    "refresh_selection": [
-      "table3",
-      "table4"
+    "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
+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
+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
+Error: invalid --full-refresh: source file "./missing.sql" not found; use a table name or an existing source file
+
+>>> print_requests.py //pipelines
+
+=== Rejecting mixed selectors before restarting the pipeline
+>>> musterr [CLI] pipelines run --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
+
+>>> musterr [CLI] pipelines run --full-refresh table2,pipeline_file.py --restart --no-wait
+Error: cannot mix table names and source files in --full-refresh; use only table names or only source files
+
+>>> print_requests.py //pipelines
+
+>>> 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
+
+>>> print_requests.py //pipelines

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)