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..0099ed1190e 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
 {
@@ -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,7 +97,8 @@ 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
 {
@@ -105,12 +106,103 @@ Pipeline configurations for this update:
   "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
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 b324074d59f..70fad2ba1ce 100644
--- a/acceptance/pipelines/run/refresh-flags/script
+++ b/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"
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..d9647265a14 100644
--- a/bundle/run/pipeline_options_test.go
+++ b/bundle/run/pipeline_options_test.go
@@ -1,6 +1,8 @@
 package run
 
 import (
+	"encoding/json"
+	"fmt"
 	"testing"
 
 	flag "github.com/spf13/pflag"
@@ -8,6 +10,108 @@ 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)
+			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
diff --git a/cmd/pipelines/refresh.go b/cmd/pipelines/refresh.go
new file mode 100644
index 00000000000..60957bc2e05
--- /dev/null
+++ b/cmd/pipelines/refresh.go
@@ -0,0 +1,109 @@
+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)
+		if statErr != nil && !errors.Is(statErr, fs.ErrNotExist) {
+			return nil, nil, fmt.Errorf("cannot read refresh selector %q: %w", selector, statErr)
+		}
+		// Preserve table identifiers such as schema.sql when no local file exists.
+		looksLikeFile := strings.ContainsAny(selector, `/\`)
+		if errors.Is(statErr, fs.ErrNotExist) && !looksLikeFile {
+			tables = append(tables, selector)
+			continue
+		}
+		if statErr != nil {
+			return nil, nil, fmt.Errorf("source file %q not found; use a table name or an existing source file", selector)
+		}
+		if info.IsDir() {
+			if !looksLikeFile {
+				tables = append(tables, selector)
+				continue
+			}
+			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..4ec0900dc4c
--- /dev/null
+++ b/cmd/pipelines/refresh_test.go
@@ -0,0 +1,167 @@
+package pipelines
+
+import (
+	"os"
+	"path/filepath"
+	"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) {
+	tests := []struct {
+		name      string
+		selectors []string
+		tables    []string
+		paths     []string
+		err       string
+	}{
+		{name: "no selection"},
+		{
+			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..394810fb5c8 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)
@@ -286,6 +293,17 @@ Refreshes all tables in the pipeline unless otherwise specified.`,
 			},
 			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..df52b09444a 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 and file refresh selections",
+			update: pipelines.UpdateInfo{
+				RefreshSelection:       []string{"table1"},
+				RefreshSelectionByPath: []string{"/Workspace/Shared/source.sql"},
+			},
+			expected: `
+Update for pipeline completed successfully.
+
+Pipeline configurations for this update:
+• Refreshed [table1]
+• 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 647bd4c38cd..78eed0b8a45 100644
--- a/libs/testserver/fake_workspace.go
+++ b/libs/testserver/fake_workspace.go
@@ -217,7 +217,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
@@ -517,7 +517,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 73839619c3a..7e34b9fabf4 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)
+		})
+	}
+}