diff --git a/apps/desktop/src/components/sql-file/SqlFileExecutionDialog.retry.spec.ts b/apps/desktop/src/components/sql-file/SqlFileExecutionDialog.retry.spec.ts
index c515129bcc..c083fa55ba 100644
--- a/apps/desktop/src/components/sql-file/SqlFileExecutionDialog.retry.spec.ts
+++ b/apps/desktop/src/components/sql-file/SqlFileExecutionDialog.retry.spec.ts
@@ -12,6 +12,7 @@ const mocks = vi.hoisted(() => ({
   listenSqlFileProgress: vi.fn(),
   openFileDialog: vi.fn(),
   previewSqlFile: vi.fn(),
+  inspectSqlFileTables: vi.fn(),
   progressHandler: undefined as undefined | ((progress: Record<string, unknown>) => void),
   refreshDatabaseTreeNode: vi.fn(),
   requestConfirmation: vi.fn(),
@@ -41,7 +42,7 @@ vi.mock("@/composables/useExportTracker", () => ({
   useExportTracker: () => ({ addSqlFileTask: mocks.addSqlFileTask, updateSqlFileTask: mocks.updateSqlFileTask }),
 }));
 vi.mock("@/composables/useDatabaseOptions", () => ({ fetchSqlFileTargetOptions: mocks.fetchSqlFileTargetOptions }));
-vi.mock("@/lib/connection/connectionLevelDatabaseBootstrap", () => ({ requiresSqlFileTargetDatabaseSelection: () => false }));
+vi.mock("@/lib/connection/connectionLevelDatabaseBootstrap", () => ({ requiresSqlFileTargetDatabaseSelection: () => false, supportsConnectionLevelDatabaseBootstrap: (connection: any) => connection?.db_type === "mysql" }));
 vi.mock("@/lib/database/productionSafety", () => ({ productionContextForDatabase: () => ({ active: false, databases: [] }) }));
 vi.mock("@/stores/productionSafetyStore", () => ({
   useProductionSafetyStore: () => ({ requestConfirmation: mocks.requestConfirmation }),
@@ -59,6 +60,7 @@ vi.mock("@/lib/backend/api", () => ({
   executeSqlFiles: mocks.executeSqlFiles,
   listenSqlFileProgress: mocks.listenSqlFileProgress,
   previewSqlFile: mocks.previewSqlFile,
+  inspectSqlFileTables: mocks.inspectSqlFileTables,
 }));
 vi.mock("@lucide/vue", () => {
   const Icon = passthrough("span");
@@ -77,7 +79,15 @@ vi.mock("@/components/ui/tooltip", () => ({
   TooltipTrigger: passthrough("div"),
 }));
 vi.mock("@/components/ui/button", () => ({ Button: passthrough("button") }));
-vi.mock("@/components/ui/input", () => ({ Input: passthrough("input") }));
+vi.mock("@/components/ui/input", () => ({
+  Input: defineComponent({
+    props: ["modelValue"],
+    emits: ["update:modelValue"],
+    setup(props, { emit }) {
+      return () => h("input", { value: props.modelValue, onInput: (event: Event) => emit("update:modelValue", (event.target as HTMLInputElement).value) });
+    },
+  }),
+}));
 vi.mock("@/components/ui/label", () => ({ Label: passthrough("label") }));
 vi.mock("@/components/ui/select", () => ({
   Select: passthrough("div"),
@@ -181,6 +191,11 @@ beforeEach(() => {
     }
   });
   mocks.ensureConnected.mockResolvedValue(undefined);
+  mocks.inspectSqlFileTables.mockResolvedValue([
+    { database: "app", name: "users" },
+    { database: "app", name: "orders" },
+    { database: "archive", name: "users" },
+  ]);
   mocks.fetchSqlFileTargetOptions.mockResolvedValue([]);
   mocks.openFileDialog.mockResolvedValue(["/tmp/first.sql", "/tmp/second.sql"]);
   mocks.previewSqlFile.mockImplementation(async (filePath: string) => ({
@@ -296,3 +311,93 @@ describe("SqlFileExecutionDialog retries", () => {
     expect(root!.textContent).not.toContain("stale statement failure");
   });
 });
+
+describe("SqlFileExecutionDialog selected-table restore", () => {
+  async function mountBackup() {
+    root = document.createElement("div");
+    document.body.append(root);
+    app = createApp(SqlFileExecutionDialog, { open: true, prefillConnectionId: "mysql-1", prefillDatabase: "app", prefillFilePath: "/tmp/backup.sql.gz" });
+    app.mount(root);
+    await vi.waitFor(() => expect(findButton("sqlFile.execute").disabled).toBe(false));
+    root.querySelectorAll<HTMLInputElement>('input[type="radio"]')[1]!.click();
+    await nextTick();
+  }
+
+  it("scans the backup and sends only checked table identities", async () => {
+    await mountBackup();
+    await vi.waitFor(() => expect(root!.textContent).toContain("archive.users"));
+    expect(mocks.inspectSqlFileTables).toHaveBeenCalledWith("/tmp/backup.sql.gz");
+    expect(findButton("sqlFile.execute").disabled).toBe(true);
+    const label = Array.from(root!.querySelectorAll("label")).find((label) => label.textContent === "app.users")!;
+    label.querySelector<HTMLInputElement>("input")!.click();
+    await nextTick();
+    mocks.executeSqlFiles.mockImplementationOnce(async (request) => {
+      mocks.progressHandler?.(progress(request.executionId, "done", { successCount: 2 }));
+    });
+    findButton("sqlFile.execute").click();
+    await vi.waitFor(() => expect(mocks.executeSqlFiles).toHaveBeenCalledWith(expect.objectContaining({ selectedTables: [{ database: "app", name: "users" }] }), ["/tmp/backup.sql.gz"]));
+  });
+
+  it("selects only search matches and preserves selections outside the search", async () => {
+    await mountBackup();
+    await vi.waitFor(() => expect(root!.textContent).toContain("archive.users"));
+    const search = root!.querySelector<HTMLInputElement>('[aria-label="sqlFile.searchBackupTables"]')!;
+    search.value = "app.";
+    search.dispatchEvent(new Event("input"));
+    await nextTick();
+    root!.querySelector<HTMLInputElement>('[data-table-restore] input[type="checkbox"]')!.click();
+    await nextTick();
+    search.value = "archive";
+    search.dispatchEvent(new Event("input"));
+    await nextTick();
+    expect(root!.querySelector<HTMLInputElement>('[data-table-restore] input[type="checkbox"]')!.checked).toBe(false);
+    mocks.executeSqlFiles.mockImplementationOnce(async (request) => {
+      mocks.progressHandler?.(progress(request.executionId, "done"));
+    });
+    findButton("sqlFile.execute").click();
+    await vi.waitFor(() =>
+      expect(mocks.executeSqlFiles).toHaveBeenCalledWith(
+        expect.objectContaining({
+          selectedTables: [
+            { database: "app", name: "users" },
+            { database: "app", name: "orders" },
+          ],
+        }),
+        expect.any(Array),
+      ),
+    );
+  });
+
+  it("blocks execution on scan failure and allows switching back to full restore", async () => {
+    mocks.inspectSqlFileTables.mockRejectedValueOnce(new Error("unsupported dump statement"));
+    await mountBackup();
+    await vi.waitFor(() => expect(root!.querySelector('[role="alert"]')?.textContent).toContain("unsupported dump statement"));
+    expect(findButton("sqlFile.execute").disabled).toBe(true);
+    root!.querySelector<HTMLInputElement>('input[type="radio"]')!.click();
+    await nextTick();
+    expect(findButton("sqlFile.execute").disabled).toBe(false);
+  });
+
+  it("ignores a scan result after switching back to all contents", async () => {
+    let finish!: (tables: any[]) => void;
+    mocks.inspectSqlFileTables.mockImplementationOnce(
+      () =>
+        new Promise((resolve) => {
+          finish = resolve;
+        }),
+    );
+    await mountBackup();
+    expect(findButton("sqlFile.execute").disabled).toBe(true);
+    root!.querySelector<HTMLInputElement>('input[type="radio"]')!.click();
+    await nextTick();
+    finish([{ database: null, name: "stale" }]);
+    await nextTick();
+    expect(root!.textContent).not.toContain("stale");
+    mocks.executeSqlFiles.mockImplementationOnce(async (request) => {
+      mocks.progressHandler?.(progress(request.executionId, "done"));
+    });
+    findButton("sqlFile.execute").click();
+    await vi.waitFor(() => expect(mocks.executeSqlFiles).toHaveBeenCalled());
+    expect(mocks.executeSqlFiles.mock.calls[0]![0]).not.toHaveProperty("selectedTables");
+  });
+});
diff --git a/apps/desktop/src/components/sql-file/SqlFileExecutionDialog.vue b/apps/desktop/src/components/sql-file/SqlFileExecutionDialog.vue
index 9a2f52e4da..929f91a2ef 100644
--- a/apps/desktop/src/components/sql-file/SqlFileExecutionDialog.vue
+++ b/apps/desktop/src/components/sql-file/SqlFileExecutionDialog.vue
@@ -18,8 +18,8 @@ import { useProductionSafetyStore } from "@/stores/productionSafetyStore";
 import { productionContextForDatabase } from "@/lib/database/productionSafety";
 import { connectionIsEffectivelyReadOnly, ensureReadOnlyWriteAccess } from "@/lib/database/readOnlyWriteAccess";
 import { fetchSqlFileTargetOptions } from "@/composables/useDatabaseOptions";
-import { requiresSqlFileTargetDatabaseSelection } from "@/lib/connection/connectionLevelDatabaseBootstrap";
-import { cancelSqlFileExecution, executeSqlFiles, listenSqlFileProgress, previewSqlFile, type SqlFilePreview, type SqlFileProgress, type SqlFileStatus } from "@/lib/backend/api";
+import { requiresSqlFileTargetDatabaseSelection, supportsConnectionLevelDatabaseBootstrap } from "@/lib/connection/connectionLevelDatabaseBootstrap";
+import { cancelSqlFileExecution, executeSqlFiles, inspectSqlFileTables, listenSqlFileProgress, previewSqlFile, type SqlFilePreview, type SqlFileProgress, type SqlFileStatus, type SqlFileTable } from "@/lib/backend/api";
 import { buildDisplayFileNames, tooltipText as computeTooltipText } from "./sqlFilePreviewLabel";
 import { useExportTracker, type ExportTask } from "@/composables/useExportTracker";
 import { translateBackendError } from "@/i18n/backend-errors";
@@ -135,9 +135,59 @@ const sqlConnections = computed(() => store.connections.filter((c) => !["redis",
 
 const selectedConnection = computed(() => sqlConnections.value.find((c) => c.id === connectionId.value));
 
+const restoreSelectedTables = ref(false);
+const backupTables = ref<SqlFileTable[]>([]);
+const selectedTableKeys = ref(new Set<string>());
+const tableSearch = ref("");
+const loadingTables = ref(false);
+const tableScanError = ref("");
+let tableScanGeneration = 0;
+const canSelectTables = computed(() => previews.value.length === 1 && supportsConnectionLevelDatabaseBootstrap(selectedConnection.value));
+const tableKey = (table: SqlFileTable) => JSON.stringify([table.database, table.name]);
+const tableLabel = (table: SqlFileTable) => (table.database ? `${table.database}.${table.name}` : table.name);
+const filteredBackupTables = computed(() => backupTables.value.filter((table) => tableLabel(table).toLocaleLowerCase().includes(tableSearch.value.trim().toLocaleLowerCase())));
+const selectedTables = computed(() => backupTables.value.filter((table) => selectedTableKeys.value.has(tableKey(table))));
+const allFilteredTablesSelected = computed(() => filteredBackupTables.value.length > 0 && filteredBackupTables.value.every((table) => selectedTableKeys.value.has(tableKey(table))));
+
+function toggleTable(table: SqlFileTable) {
+  const key = tableKey(table);
+  if (selectedTableKeys.value.has(key)) selectedTableKeys.value.delete(key);
+  else selectedTableKeys.value.add(key);
+}
+
+function toggleFilteredTables() {
+  const deselect = allFilteredTablesSelected.value;
+  for (const table of filteredBackupTables.value) {
+    if (deselect) selectedTableKeys.value.delete(tableKey(table));
+    else selectedTableKeys.value.add(tableKey(table));
+  }
+}
+
+watch([restoreSelectedTables, canSelectTables, () => previews.value[0]?.filePath, connectionId, open], async () => {
+  const generation = ++tableScanGeneration;
+  backupTables.value = [];
+  selectedTableKeys.value = new Set();
+  tableSearch.value = "";
+  tableScanError.value = "";
+  loadingTables.value = false;
+  if (!canSelectTables.value) restoreSelectedTables.value = false;
+  if (!open.value || !restoreSelectedTables.value || !canSelectTables.value) return;
+  loadingTables.value = true;
+  try {
+    const tables = await inspectSqlFileTables(previews.value[0]!.filePath);
+    if (generation !== tableScanGeneration) return;
+    backupTables.value = tables;
+  } catch (error: any) {
+    if (generation === tableScanGeneration) tableScanError.value = error?.message || String(error);
+  } finally {
+    if (generation === tableScanGeneration) loadingTables.value = false;
+  }
+});
+
 const canStart = computed(() => {
   const connection = selectedConnection.value;
   if (previews.value.length === 0 || !connection || running.value || loadingPreview.value || loadingDatabases.value) return false;
+  if (restoreSelectedTables.value && (loadingTables.value || tableScanError.value || selectedTables.value.length === 0)) return false;
   let hasDatabaseContext = false;
   const canExecuteWithoutSelectedDatabase = previews.value.every((item) => {
     if (!hasDatabaseContext && !item.canExecuteWithoutSelectedDatabase) return false;
@@ -259,6 +309,7 @@ function resetState() {
   databaseOptions.value = [];
   loadingDatabases.value = false;
   continueOnError.value = false;
+  restoreSelectedTables.value = false;
   resetExecution();
 }
 
@@ -495,6 +546,7 @@ async function startExecution() {
           database: database.value.trim(),
           filePath: previews.value[0]!.filePath,
           continueOnError: continueOnError.value,
+          ...(restoreSelectedTables.value ? { selectedTables: selectedTables.value.map((table) => ({ ...table })) } : {}),
         },
         previews.value.map((item) => item.filePath),
       );
@@ -702,6 +754,33 @@ watch(
           </div>
         </div>
 
+        <div v-if="canSelectTables" class="min-w-0 space-y-2.5" data-table-restore>
+          <Label class="text-xs">{{ t("sqlFile.restoreScope") }}</Label>
+          <div class="flex items-center gap-4 text-xs">
+            <label class="flex items-center gap-2"><input v-model="restoreSelectedTables" type="radio" :value="false" :disabled="running" name="sql-file-restore-scope" />{{ t("sqlFile.restoreAll") }}</label>
+            <label class="flex items-center gap-2"><input v-model="restoreSelectedTables" type="radio" :value="true" :disabled="running" name="sql-file-restore-scope" />{{ t("sqlFile.restoreSelectedTables") }}</label>
+          </div>
+          <template v-if="restoreSelectedTables">
+            <p class="text-xs text-muted-foreground">{{ t("sqlFile.restoreTablesOnly") }}</p>
+            <div v-if="loadingTables" class="flex items-center gap-2 text-xs" role="status"><Loader2 class="h-3.5 w-3.5 animate-spin" />{{ t("sqlFile.scanningTables") }}</div>
+            <p v-else-if="tableScanError" class="break-words text-xs text-destructive" role="alert">{{ tableScanError }}</p>
+            <template v-else>
+              <Input v-model="tableSearch" :placeholder="t('sqlFile.searchBackupTables')" :aria-label="t('sqlFile.searchBackupTables')" :disabled="running" class="h-8 text-xs" />
+              <div class="flex items-center justify-between gap-3 text-xs">
+                <label class="flex items-center gap-2"><input type="checkbox" :checked="allFilteredTablesSelected" :disabled="running || filteredBackupTables.length === 0" @change="toggleFilteredTables" />{{ t("sqlFile.selectVisibleTables") }}</label>
+                <span>{{ t("sqlFile.selectedTableCount", { selected: selectedTables.length, total: backupTables.length }) }}</span>
+              </div>
+              <div class="max-h-44 overflow-y-auto border rounded-md p-2 text-xs">
+                <label v-for="table in filteredBackupTables" :key="tableKey(table)" class="flex min-w-0 items-center gap-2 py-1">
+                  <input type="checkbox" :checked="selectedTableKeys.has(tableKey(table))" :disabled="running" @change="toggleTable(table)" />
+                  <span class="min-w-0 break-all">{{ tableLabel(table) }}</span>
+                </label>
+                <p v-if="filteredBackupTables.length === 0" class="text-muted-foreground">{{ t("sqlFile.noBackupTables") }}</p>
+              </div>
+            </template>
+          </template>
+        </div>
+
         <div class="min-w-0 space-y-2.5">
           <div class="text-xs font-medium text-muted-foreground uppercase tracking-wider">
             {{ t("sqlFile.options") }}
diff --git a/apps/desktop/src/i18n/locales/az.ts b/apps/desktop/src/i18n/locales/az.ts
index eb242b1cb5..f988bcf6a1 100644
--- a/apps/desktop/src/i18n/locales/az.ts
+++ b/apps/desktop/src/i18n/locales/az.ts
@@ -6167,6 +6167,15 @@ export default withEnglishFallback({
     generateDataComingSoon: "Tezliklə",
   },
   sqlFile: {
+    restoreScope: "Bərpa əhatəsi",
+    restoreAll: "Bütün məzmun",
+    restoreSelectedTables: "Seçilmiş cədvəllər",
+    restoreTablesOnly: "Yalnız cədvəl strukturu və məlumatları bərpa olunur. Görünüşlər, saxlanılan prosedurlar, funksiyalar, triggerlər və hadisələr daxil edilmir. Mövcud xarici açarlar və triggerlər bərpaya təsir edə bilər.",
+    scanningTables: "Ehtiyat nüsxədəki cədvəllər skan edilir...",
+    searchBackupTables: "Ehtiyat nüsxədə cədvəl axtar...",
+    selectVisibleTables: "Görünən cədvəlləri seç",
+    selectedTableCount: "{selected} / {total} cədvəl seçilib",
+    noBackupTables: "Ehtiyat nüsxədə uyğun cədvəl yoxdur",
     title: "SQL faylını icra et",
     file: "Fayl",
     selectSqlFile: "SQL faylı seç",
diff --git a/apps/desktop/src/i18n/locales/en.ts b/apps/desktop/src/i18n/locales/en.ts
index 4dca25410e..8a5b0bd93d 100644
--- a/apps/desktop/src/i18n/locales/en.ts
+++ b/apps/desktop/src/i18n/locales/en.ts
@@ -6166,6 +6166,15 @@ export default {
     generateDataComingSoon: "Coming soon",
   },
   sqlFile: {
+    restoreScope: "Restore scope",
+    restoreAll: "All contents",
+    restoreSelectedTables: "Selected tables",
+    restoreTablesOnly: "Restore table structure and data only. Views, routines, triggers and events are excluded. Existing foreign keys and triggers may affect the restore.",
+    scanningTables: "Scanning backup tables...",
+    searchBackupTables: "Search backup tables...",
+    selectVisibleTables: "Select visible tables",
+    selectedTableCount: "{selected} / {total} tables selected",
+    noBackupTables: "No matching tables in the backup",
     title: "Execute SQL File",
     file: "File",
     selectSqlFile: "Select SQL file",
diff --git a/apps/desktop/src/i18n/locales/es.ts b/apps/desktop/src/i18n/locales/es.ts
index e40fd93ad3..152bfeeb8a 100644
--- a/apps/desktop/src/i18n/locales/es.ts
+++ b/apps/desktop/src/i18n/locales/es.ts
@@ -5793,6 +5793,15 @@ export default withEnglishFallback({
     generateDataComingSoon: "Próximamente",
   },
   sqlFile: {
+    restoreScope: "Alcance de la restauración",
+    restoreAll: "Todo el contenido",
+    restoreSelectedTables: "Tablas seleccionadas",
+    restoreTablesOnly: "Solo se restauran la estructura y los datos de las tablas. Se excluyen las vistas, los procedimientos almacenados, las funciones, los disparadores y los eventos. Las claves foráneas y los disparadores existentes pueden afectar a la restauración.",
+    scanningTables: "Analizando las tablas de la copia de seguridad...",
+    searchBackupTables: "Buscar tablas en la copia de seguridad...",
+    selectVisibleTables: "Seleccionar las tablas visibles",
+    selectedTableCount: "{selected} / {total} tablas seleccionadas",
+    noBackupTables: "No hay tablas coincidentes en la copia de seguridad",
     title: "Ejecutar archivo SQL",
     file: "Archivo",
     selectSqlFile: "Seleccionar archivo SQL",
diff --git a/apps/desktop/src/i18n/locales/it.ts b/apps/desktop/src/i18n/locales/it.ts
index 995180a7d4..8444118f9f 100644
--- a/apps/desktop/src/i18n/locales/it.ts
+++ b/apps/desktop/src/i18n/locales/it.ts
@@ -5791,6 +5791,15 @@ export default withEnglishFallback({
     generateDataComingSoon: "Prossimamente",
   },
   sqlFile: {
+    restoreScope: "Ambito del ripristino",
+    restoreAll: "Tutto il contenuto",
+    restoreSelectedTables: "Tabelle selezionate",
+    restoreTablesOnly: "Ripristina solo la struttura e i dati delle tabelle. Viste, procedure memorizzate, funzioni, trigger ed eventi sono esclusi. Le chiavi esterne e i trigger esistenti possono influire sul ripristino.",
+    scanningTables: "Scansione delle tabelle nel backup...",
+    searchBackupTables: "Cerca tabelle nel backup...",
+    selectVisibleTables: "Seleziona le tabelle visibili",
+    selectedTableCount: "{selected} / {total} tabelle selezionate",
+    noBackupTables: "Nessuna tabella corrispondente nel backup",
     title: "Esegui File SQL",
     file: "File",
     selectSqlFile: "Seleziona file SQL",
diff --git a/apps/desktop/src/i18n/locales/ja.ts b/apps/desktop/src/i18n/locales/ja.ts
index 87ed47726c..6e73638bfa 100644
--- a/apps/desktop/src/i18n/locales/ja.ts
+++ b/apps/desktop/src/i18n/locales/ja.ts
@@ -5829,6 +5829,15 @@ export default withEnglishFallback({
     generateDataComingSoon: "近日公開",
   },
   sqlFile: {
+    restoreScope: "復元範囲",
+    restoreAll: "すべての内容",
+    restoreSelectedTables: "指定したテーブル",
+    restoreTablesOnly: "テーブル構造とデータのみを復元します。ビュー、ストアドプロシージャ、関数、トリガー、イベントは含まれません。既存の外部キーやトリガーが復元に影響する場合があります。",
+    scanningTables: "バックアップ内のテーブルをスキャン中...",
+    searchBackupTables: "バックアップ内のテーブルを検索...",
+    selectVisibleTables: "表示中のテーブルを選択",
+    selectedTableCount: "{total} テーブル中 {selected} テーブルを選択",
+    noBackupTables: "バックアップ内に一致するテーブルがありません",
     runInBackground: "バックグラウンドで実行",
     title: "SQLファイルを実行",
     file: "ファイル",
diff --git a/apps/desktop/src/i18n/locales/ko.ts b/apps/desktop/src/i18n/locales/ko.ts
index a83f6bf002..ecd704f177 100644
--- a/apps/desktop/src/i18n/locales/ko.ts
+++ b/apps/desktop/src/i18n/locales/ko.ts
@@ -5498,6 +5498,15 @@ export default withEnglishFallback({
     generateDataComingSoon: "준비 중",
   },
   sqlFile: {
+    restoreScope: "복원 범위",
+    restoreAll: "모든 내용",
+    restoreSelectedTables: "선택한 테이블",
+    restoreTablesOnly: "테이블 구조와 데이터만 복원합니다. 뷰, 저장 프로시저, 함수, 트리거 및 이벤트는 제외됩니다. 기존 외래 키와 트리거가 복원에 영향을 줄 수 있습니다.",
+    scanningTables: "백업의 테이블을 검색하는 중...",
+    searchBackupTables: "백업의 테이블 검색...",
+    selectVisibleTables: "표시된 테이블 선택",
+    selectedTableCount: "테이블 {selected} / {total}개 선택됨",
+    noBackupTables: "백업에 일치하는 테이블이 없습니다",
     title: "SQL 파일 실행",
     file: "파일",
     selectSqlFile: "SQL 파일 선택",
diff --git a/apps/desktop/src/i18n/locales/pt-BR.ts b/apps/desktop/src/i18n/locales/pt-BR.ts
index fb66ab124c..ca3e32c948 100644
--- a/apps/desktop/src/i18n/locales/pt-BR.ts
+++ b/apps/desktop/src/i18n/locales/pt-BR.ts
@@ -5793,6 +5793,15 @@ export default withEnglishFallback({
     generateDataComingSoon: "Em breve",
   },
   sqlFile: {
+    restoreScope: "Escopo da restauração",
+    restoreAll: "Todo o conteúdo",
+    restoreSelectedTables: "Tabelas selecionadas",
+    restoreTablesOnly: "Restaura apenas a estrutura e os dados das tabelas. Views, procedimentos armazenados, funções, gatilhos e eventos não são incluídos. Chaves estrangeiras e gatilhos existentes podem afetar a restauração.",
+    scanningTables: "Verificando as tabelas do backup...",
+    searchBackupTables: "Pesquisar tabelas no backup...",
+    selectVisibleTables: "Selecionar as tabelas visíveis",
+    selectedTableCount: "{selected} / {total} tabelas selecionadas",
+    noBackupTables: "Nenhuma tabela correspondente no backup",
     title: "Executar Arquivo SQL",
     file: "Arquivo",
     selectSqlFile: "Selecionar arquivo SQL",
diff --git a/apps/desktop/src/i18n/locales/tr.ts b/apps/desktop/src/i18n/locales/tr.ts
index c9a8170e6a..07f963f02a 100644
--- a/apps/desktop/src/i18n/locales/tr.ts
+++ b/apps/desktop/src/i18n/locales/tr.ts
@@ -6049,6 +6049,15 @@ export default withEnglishFallback({
     generateDataComingSoon: "Yakında",
   },
   sqlFile: {
+    restoreScope: "Geri yükleme kapsamı",
+    restoreAll: "Tüm içerik",
+    restoreSelectedTables: "Seçili tablolar",
+    restoreTablesOnly: "Yalnızca tablo yapısı ve verileri geri yüklenir. Görünümler, saklı yordamlar, işlevler, tetikleyiciler ve olaylar dahil edilmez. Mevcut yabancı anahtarlar ve tetikleyiciler geri yüklemeyi etkileyebilir.",
+    scanningTables: "Yedekteki tablolar taranıyor...",
+    searchBackupTables: "Yedekteki tabloları ara...",
+    selectVisibleTables: "Görünür tabloları seç",
+    selectedTableCount: "{selected} / {total} tablo seçildi",
+    noBackupTables: "Yedekte eşleşen tablo yok",
     title: "SQL Dosyasını Çalıştır",
     file: "Dosya",
     selectSqlFile: "SQL dosyası seçin",
diff --git a/apps/desktop/src/i18n/locales/zh-CN.ts b/apps/desktop/src/i18n/locales/zh-CN.ts
index 7082d589e9..f30bc5fba0 100644
--- a/apps/desktop/src/i18n/locales/zh-CN.ts
+++ b/apps/desktop/src/i18n/locales/zh-CN.ts
@@ -6150,6 +6150,15 @@ export default withEnglishFallback({
     generateDataComingSoon: "即将推出",
   },
   sqlFile: {
+    restoreScope: "恢复范围",
+    restoreAll: "全部内容",
+    restoreSelectedTables: "指定表",
+    restoreTablesOnly: "仅恢复表结构与数据，不包含视图、存储过程、函数、触发器和事件。现有外键和触发器可能影响恢复。",
+    scanningTables: "正在扫描备份中的表...",
+    searchBackupTables: "搜索备份中的表...",
+    selectVisibleTables: "选择当前列表中的表",
+    selectedTableCount: "已选择 {selected} / {total} 张表",
+    noBackupTables: "备份中没有匹配的表",
     title: "执行 SQL 文件",
     file: "文件",
     selectSqlFile: "选择 SQL 文件",
diff --git a/apps/desktop/src/i18n/locales/zh-TW.ts b/apps/desktop/src/i18n/locales/zh-TW.ts
index 951e0fe4ff..6229cd4ee7 100644
--- a/apps/desktop/src/i18n/locales/zh-TW.ts
+++ b/apps/desktop/src/i18n/locales/zh-TW.ts
@@ -5113,6 +5113,15 @@ export default withEnglishFallback({
     generateDataComingSoon: "即將推出",
   },
   sqlFile: {
+    restoreScope: "還原範圍",
+    restoreAll: "全部內容",
+    restoreSelectedTables: "指定資料表",
+    restoreTablesOnly: "僅還原資料表結構與資料，不包含檢視、預存程序、函式、觸發程序和事件。現有的外部索引鍵與觸發程序可能影響還原。",
+    scanningTables: "正在掃描備份中的資料表...",
+    searchBackupTables: "搜尋備份中的資料表...",
+    selectVisibleTables: "選取目前清單中的資料表",
+    selectedTableCount: "已選取 {selected} / {total} 張資料表",
+    noBackupTables: "備份中沒有符合的資料表",
     title: "執行 SQL 檔案",
     file: "檔案",
     selectSqlFile: "選擇 SQL 檔案",
diff --git a/apps/desktop/src/lib/backend/api.ts b/apps/desktop/src/lib/backend/api.ts
index 355d596f4f..3111426bcf 100644
--- a/apps/desktop/src/lib/backend/api.ts
+++ b/apps/desktop/src/lib/backend/api.ts
@@ -384,6 +384,7 @@ export const previewSqlFile = forward("previewSqlFile");
 export const executeSqlFile = forward("executeSqlFile");
 export const executeSqlFiles = forward("executeSqlFiles");
 export const cancelSqlFileExecution = forward("cancelSqlFileExecution");
+export const inspectSqlFileTables = forward("inspectSqlFileTables");
 export const listenSqlFileProgress = forward("listenSqlFileProgress");
 export const pendingOpenSqlFiles = forward("pendingOpenSqlFiles");
 export const pendingOpenDbFiles = forward("pendingOpenDbFiles");
@@ -967,6 +968,7 @@ export type {
   SqlFileStatus,
   SqlFileRequest,
   SqlFilePreview,
+  SqlFileTable,
   SqlFileProgress,
   TransferRequest,
   TransferProgress,
diff --git a/apps/desktop/src/lib/backend/http.ts b/apps/desktop/src/lib/backend/http.ts
index fc3b443ad0..ba5b0a0cff 100644
--- a/apps/desktop/src/lib/backend/http.ts
+++ b/apps/desktop/src/lib/backend/http.ts
@@ -121,6 +121,7 @@ import type {
   HistoryConnectionOption,
   SqlFileRequest,
   SqlFilePreview,
+  SqlFileTable,
   SqlFileProgress,
   TransferRequest,
   TransferProgress,
@@ -2252,6 +2253,10 @@ export async function executeSqlFile(request: SqlFileRequest): Promise<void> {
   return post("/api/sql-file/execute", { request });
 }
 
+export async function inspectSqlFileTables(filePath: string): Promise<SqlFileTable[]> {
+  return post("/api/sql-file/tables", { filePath });
+}
+
 export async function executeSqlFiles(request: SqlFileRequest, filePaths: string[]): Promise<void> {
   return post("/api/sql-file/execute", { request, filePaths });
 }
diff --git a/apps/desktop/src/lib/backend/tauri.ts b/apps/desktop/src/lib/backend/tauri.ts
index d39f644776..1eef2b9a28 100644
--- a/apps/desktop/src/lib/backend/tauri.ts
+++ b/apps/desktop/src/lib/backend/tauri.ts
@@ -4583,6 +4583,16 @@ export interface SqlFileRequest {
   database: string;
   filePath: string;
   continueOnError: boolean;
+  selectedTables?: SqlFileTable[];
+}
+
+export interface SqlFileTable {
+  database: string | null;
+  name: string;
+}
+
+export async function inspectSqlFileTables(filePath: string): Promise<SqlFileTable[]> {
+  return invoke("inspect_sql_file_tables", { filePath });
 }
 
 export interface SqlFilePreview {
diff --git a/crates/dbx-core/src/sql.rs b/crates/dbx-core/src/sql.rs
index 559077e944..34da8c93c5 100644
--- a/crates/dbx-core/src/sql.rs
+++ b/crates/dbx-core/src/sql.rs
@@ -59,6 +59,8 @@ pub struct SqlFileRequest {
     pub database: String,
     pub file_path: String,
     pub continue_on_error: bool,
+    #[serde(default)]
+    pub selected_tables: Option<Vec<crate::sql_file_import::SqlFileTable>>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
diff --git a/crates/dbx-core/src/sql_file_import.rs b/crates/dbx-core/src/sql_file_import.rs
index 85deca3d68..4122619151 100644
--- a/crates/dbx-core/src/sql_file_import.rs
+++ b/crates/dbx-core/src/sql_file_import.rs
@@ -27,6 +27,10 @@ use crate::sql::{
 };
 use crate::types::QueryResult;
 
+mod table_restore;
+pub use table_restore::SqlFileTable;
+use table_restore::TableRestoreFilter;
+
 #[derive(Debug, Clone)]
 struct SqlFileImportTarget {
     db_type: DatabaseType,
@@ -340,11 +344,22 @@ pub async fn execute_sql_file_content(
     mut emit: impl FnMut(SqlFileProgress),
 ) -> Result<(), String> {
     let import_target = sql_file_import_target(state, &request.connection_id).await;
-    let statements = split_sql_file_import_statements_with_control(
+    validate_table_restore_target(request, import_target.as_ref(), 1)?;
+    let mut statements = split_sql_file_import_statements_with_control(
         file_content,
         import_target.as_ref().map(|target| target.db_type),
     );
 
+    if let Some(selected) = &request.selected_tables {
+        let mut scan = TableRestoreFilter::default();
+        for statement in &statements {
+            scan.inspect(&statement.sql)?;
+        }
+        let (tables, views) = scan.finish();
+        validate_selected_tables(selected, &tables)?;
+        filter_restore_statements(&mut statements, &mut TableRestoreFilter::with_views(views), selected)?;
+    }
+
     let planned_statements = optimize_controlled_sql_file_import_statements(
         &statements,
         import_target.as_ref().map(|target| target.db_type),
@@ -403,6 +418,29 @@ pub async fn execute_sql_file_paths(
         import_target.as_ref().map(|target| SqlParsingOptions::for_database_type(target.db_type)).unwrap_or_default();
     let database_type = import_target.as_ref().map(|target| target.db_type);
     let mut progress = SqlFileExecutionProgress::new();
+    // Validate the entire dump before any SQL reaches the database. This also
+    // identifies view placeholders whose final CREATE VIEW appears later.
+    let restore_filter = async {
+        validate_table_restore_target(request, import_target.as_ref(), file_paths.len())?;
+        let Some(selected) = &request.selected_tables else {
+            return Ok(None);
+        };
+        let (tables, views) = scan_sql_file_tables(file_paths[0], &token).await?;
+        validate_selected_tables(selected, &tables)?;
+        Ok::<_, String>(Some(TableRestoreFilter::with_views(views)))
+    }
+    .await;
+    let mut restore_filter = match restore_filter {
+        Ok(filter) => filter,
+        Err(error) => {
+            if token.is_cancelled() {
+                emit_sql_file_terminal_progress(request, &token, started_at, &progress, &mut emit);
+                return Ok(());
+            }
+            emit(sql_file_execution_error_progress(&request.execution_id, started_at, &progress, error.clone()));
+            return Err(error);
+        }
+    };
     let mut mysql_executor = match MySqlSqlFileExecutor::build(state, request, import_target.as_ref()).await {
         Ok(executor) => executor,
         Err(error) => {
@@ -475,7 +513,23 @@ pub async fn execute_sql_file_paths(
                 emit_sql_file_terminal_progress(request, &token, started_at, &progress, &mut emit);
                 return Ok(());
             }
-            pending_statements.extend(splitter.push_chunk(&chunk));
+            let mut next_statements = splitter.push_chunk(&chunk);
+            if let Some(filter) = restore_filter.as_mut() {
+                if let Err(error) = filter_restore_statements(
+                    &mut next_statements,
+                    filter,
+                    request.selected_tables.as_deref().unwrap_or_default(),
+                ) {
+                    emit(sql_file_execution_error_progress(
+                        &request.execution_id,
+                        started_at,
+                        &progress,
+                        error.clone(),
+                    ));
+                    return Err(error);
+                }
+            }
+            pending_statements.extend(next_statements);
             if pending_statements.len() < SQL_FILE_STATEMENT_BATCH_SIZE {
                 continue;
             }
@@ -493,7 +547,18 @@ pub async fn execute_sql_file_paths(
             .await?;
         }
 
-        pending_statements.extend(splitter.finish());
+        let mut next_statements = splitter.finish();
+        if let Some(filter) = restore_filter.as_mut() {
+            if let Err(error) = filter_restore_statements(
+                &mut next_statements,
+                filter,
+                request.selected_tables.as_deref().unwrap_or_default(),
+            ) {
+                emit(sql_file_execution_error_progress(&request.execution_id, started_at, &progress, error.clone()));
+                return Err(error);
+            }
+        }
+        pending_statements.extend(next_statements);
         execute_sql_file_statement_batch(
             state,
             request,
@@ -533,6 +598,79 @@ pub async fn execute_sql_file_paths(
     Ok(())
 }
 
+pub async fn inspect_sql_file_tables(file_path: &Path) -> Result<Vec<SqlFileTable>, String> {
+    scan_sql_file_tables(file_path, &CancellationToken::new()).await.map(|(tables, _)| tables)
+}
+
+async fn scan_sql_file_tables(
+    file_path: &Path,
+    token: &CancellationToken,
+) -> Result<(Vec<SqlFileTable>, std::collections::BTreeSet<SqlFileTable>), String> {
+    let mut decoder = SqlFileStreamDecoder::open_for_target(file_path, true).await?;
+    let mut splitter = StreamingSqlFileSplitter::new(Some(DatabaseType::Mysql), SqlParsingOptions::mysql_compatible());
+    let mut scan = TableRestoreFilter::default();
+    while let Some(chunk) = decoder.next_chunk().await? {
+        if token.is_cancelled() {
+            return Err("Table inspection cancelled".to_string());
+        }
+        for statement in splitter.push_chunk(&chunk) {
+            scan.inspect(&statement.sql)?;
+        }
+    }
+    for statement in splitter.finish() {
+        scan.inspect(&statement.sql)?;
+    }
+    Ok(scan.finish())
+}
+
+fn validate_table_restore_target(
+    request: &SqlFileRequest,
+    target: Option<&SqlFileImportTarget>,
+    file_count: usize,
+) -> Result<(), String> {
+    let Some(selected) = &request.selected_tables else {
+        return Ok(());
+    };
+    if selected.is_empty() {
+        return Err("Select at least one table to restore".to_string());
+    }
+    if file_count != 1 {
+        return Err("Selected-table restore requires exactly one SQL file".to_string());
+    }
+    if !target.is_some_and(|target| {
+        crate::sql::supports_connection_level_database_bootstrap_target(
+            &target.db_type,
+            target.driver_profile.as_deref(),
+        )
+    }) {
+        return Err("Selected-table restore is only supported for MySQL-compatible connections".to_string());
+    }
+    Ok(())
+}
+
+fn validate_selected_tables(selected: &[SqlFileTable], available: &[SqlFileTable]) -> Result<(), String> {
+    if selected.iter().any(|table| !available.contains(table)) {
+        return Err("A selected table is no longer present in the backup; scan the file again".to_string());
+    }
+    Ok(())
+}
+
+fn filter_restore_statements(
+    statements: &mut Vec<SqlStatementWithControl>,
+    filter: &mut TableRestoreFilter,
+    selected: &[SqlFileTable],
+) -> Result<(), String> {
+    let mut filtered = Vec::new();
+    for mut statement in std::mem::take(statements) {
+        if let Some(sql) = filter.filter(&statement.sql, selected)? {
+            statement.sql = sql;
+            filtered.push(statement);
+        }
+    }
+    *statements = filtered;
+    Ok(())
+}
+
 pub async fn read_sql_file_preview(file_path: &Path, max_chars: usize) -> Result<String, String> {
     let mut decoder =
         SqlFileStreamDecoder::open_with_detection_limit(file_path, Some(SQL_FILE_PREVIEW_ENCODING_SAMPLE_BYTES))
@@ -1916,6 +2054,74 @@ mod tests {
         path
     }
 
+    #[tokio::test]
+    async fn table_restore_scan_reads_beyond_preview_and_supports_gzip_and_utf16() {
+        let sql = format!(
+            "CREATE TABLE a (id INT); INSERT INTO a VALUES ('{}'); CREATE TABLE late_table (id INT);",
+            "x".repeat(1_100_000)
+        );
+        let mut utf16 = vec![0xff, 0xfe];
+        for unit in sql.encode_utf16() {
+            utf16.extend_from_slice(&unit.to_le_bytes());
+        }
+        for path in [
+            temporary_sql_file(sql.as_bytes()).await,
+            temporary_gzip_sql_file(sql.as_bytes()).await,
+            temporary_sql_file(&utf16).await,
+        ] {
+            let tables = inspect_sql_file_tables(&path).await.unwrap();
+            tokio::fs::remove_file(&path).await.unwrap();
+            assert_eq!(tables.iter().map(|table| table.name.as_str()).collect::<Vec<_>>(), vec!["a", "late_table"]);
+        }
+    }
+
+    #[test]
+    fn table_restore_stream_filters_each_statement_once_across_chunks() {
+        let selected = [
+            SqlFileTable { database: None, name: "a".to_string() },
+            SqlFileTable { database: Some("second".to_string()), name: "b".to_string() },
+        ];
+        let mut splitter =
+            StreamingSqlFileSplitter::new(Some(DatabaseType::Mysql), SqlParsingOptions::mysql_compatible());
+        let mut filter = TableRestoreFilter::default();
+        let mut statements = Vec::new();
+        for chunk in ["INSERT INTO a VALUES (1); USE second;", "INSERT INTO b VALUES (2); INSERT INTO a VALUES (3);"] {
+            let mut next = splitter.push_chunk(chunk);
+            filter_restore_statements(&mut next, &mut filter, &selected).unwrap();
+            statements.extend(next);
+        }
+        let mut next = splitter.finish();
+        filter_restore_statements(&mut next, &mut filter, &selected).unwrap();
+        statements.extend(next);
+        assert_eq!(
+            statements.iter().map(|statement| statement.sql.as_str()).collect::<Vec<_>>(),
+            vec!["INSERT INTO a VALUES (1)", "USE second", "INSERT INTO b VALUES (2)"]
+        );
+    }
+
+    #[test]
+    fn table_restore_request_is_opt_in_and_rejects_empty_or_missing_selections() {
+        let mut request: SqlFileRequest = serde_json::from_value(serde_json::json!({"executionId":"test", "connectionId":"conn", "database":"app", "filePath":"backup.sql", "continueOnError":false})).unwrap();
+        assert!(request.selected_tables.is_none());
+        assert!(validate_table_restore_target(&request, None, 1).is_ok());
+        request.selected_tables = Some(Vec::new());
+        assert!(validate_table_restore_target(&request, None, 1).is_err());
+        request.selected_tables = Some(vec![SqlFileTable { database: None, name: "a".to_string() }]);
+        assert!(validate_table_restore_target(
+            &request,
+            Some(&SqlFileImportTarget { db_type: DatabaseType::Postgres, driver_profile: None }),
+            1
+        )
+        .is_err());
+        assert!(validate_table_restore_target(
+            &request,
+            Some(&SqlFileImportTarget { db_type: DatabaseType::Mysql, driver_profile: None }),
+            2
+        )
+        .is_err());
+        assert!(validate_selected_tables(request.selected_tables.as_ref().unwrap(), &[]).is_err());
+    }
+
     fn test_progress(status: SqlFileStatus, statement_index: usize) -> SqlFileProgress {
         SqlFileProgress {
             execution_id: "test-execution".to_string(),
@@ -2302,6 +2508,7 @@ mod tests {
             database: String::new(),
             file_path: path.to_string_lossy().to_string(),
             continue_on_error: true,
+            selected_tables: None,
         };
         let mut progress = Vec::new();
 
diff --git a/crates/dbx-core/src/sql_file_import/table_restore.rs b/crates/dbx-core/src/sql_file_import/table_restore.rs
new file mode 100644
index 0000000000..e8a55e18e1
--- /dev/null
+++ b/crates/dbx-core/src/sql_file_import/table_restore.rs
@@ -0,0 +1,395 @@
+use std::collections::BTreeSet;
+
+use serde::{Deserialize, Serialize};
+use sqlparser::dialect::MySqlDialect;
+use sqlparser::tokenizer::{Token, Tokenizer};
+
+/// The database is the scope in the dump, not the destination selected in the UI.
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct SqlFileTable {
+    pub database: Option<String>,
+    pub name: String,
+}
+
+#[derive(Default)]
+pub(super) struct TableRestoreFilter {
+    database: Option<String>,
+    pub tables: BTreeSet<SqlFileTable>,
+    views: BTreeSet<SqlFileTable>,
+}
+
+enum Scope {
+    Table(SqlFileTable),
+    DropTables(Vec<SqlFileTable>, bool),
+    Session,
+    DatabaseSession(String),
+    Skip,
+}
+
+impl TableRestoreFilter {
+    pub fn inspect(&mut self, sql: &str) -> Result<(), String> {
+        match self.classify(sql)? {
+            Scope::Table(table) => {
+                self.tables.insert(table);
+            }
+            Scope::DropTables(tables, _) => self.tables.extend(tables),
+            _ => {}
+        }
+        Ok(())
+    }
+
+    pub fn with_views(views: BTreeSet<SqlFileTable>) -> Self {
+        Self { views, ..Self::default() }
+    }
+
+    pub fn finish(mut self) -> (Vec<SqlFileTable>, BTreeSet<SqlFileTable>) {
+        self.tables.retain(|table| !self.views.contains(table));
+        (self.tables.into_iter().collect(), self.views)
+    }
+
+    pub fn filter(&mut self, sql: &str, selected: &[SqlFileTable]) -> Result<Option<String>, String> {
+        let scope = self.classify(sql)?;
+        let includes = |table: &SqlFileTable| !self.views.contains(table) && selected.contains(table);
+        match scope {
+            Scope::Table(table) => {
+                let included = includes(&table);
+                self.tables.insert(table);
+                Ok(included.then(|| sql.to_string()))
+            }
+            Scope::DropTables(tables, if_exists) => {
+                let kept = tables.iter().filter(|table| includes(table)).map(quoted_table).collect::<Vec<_>>();
+                self.tables.extend(tables);
+                Ok((!kept.is_empty())
+                    .then(|| format!("DROP TABLE {}{}", if if_exists { "IF EXISTS " } else { "" }, kept.join(", "))))
+            }
+            Scope::Session => Ok(Some(sql.to_string())),
+            Scope::DatabaseSession(database) => {
+                Ok(selected.iter().any(|table| table.database.as_ref() == Some(&database)).then(|| sql.to_string()))
+            }
+            Scope::Skip => Ok(None),
+        }
+    }
+
+    fn classify(&mut self, sql: &str) -> Result<Scope, String> {
+        // INSERT payloads can be gigabytes. Only tokenize the header; the existing
+        // streaming splitter remains responsible for statement boundaries.
+        let limit = sql.char_indices().nth(16_384).map_or(sql.len(), |(offset, _)| offset);
+        let mut tokens = Vec::new();
+        let tokenized = Tokenizer::new(&MySqlDialect {}, &sql[..limit]).tokenize_with_location_into_buf(&mut tokens);
+        let tokens = tokens
+            .into_iter()
+            .map(|token| token.token)
+            .filter(|token| !matches!(token, Token::Whitespace(_) | Token::EOF))
+            .collect::<Vec<_>>();
+        let mut cursor = Tokens { tokens: &tokens, index: 0 };
+        if tokens.is_empty() && tokenized.is_ok() && limit == sql.len() {
+            return Ok(Scope::Skip);
+        }
+        if cursor.eat("INSERT") || cursor.eat("REPLACE") {
+            for modifier in ["LOW_PRIORITY", "DELAYED", "HIGH_PRIORITY", "IGNORE"] {
+                cursor.eat(modifier);
+            }
+            cursor.eat("INTO");
+            return cursor.table(&self.database).map(Scope::Table).ok_or_else(|| unsupported(sql));
+        }
+        // Object ownership is in the header even if a long definition was
+        // truncated inside a string literal by the tokenizer limit.
+        if cursor.eat("CREATE") {
+            if cursor.eat("TEMPORARY") {
+                return Err(unsupported(sql));
+            }
+            if cursor.eat("TABLE") {
+                cursor.eat_if_not_exists();
+                return cursor.table(&self.database).map(Scope::Table).ok_or_else(|| unsupported(sql));
+            }
+            // Dumps may use CREATE ... DEFINER ... SQL SECURITY ... VIEW.
+            // Track views to exclude mysqldump's temporary table placeholders too.
+            if let Some(index) = tokens.iter().take_while(|token| **token != Token::LParen).position(|token| {
+                ["VIEW", "PROCEDURE", "FUNCTION", "TRIGGER", "EVENT"].iter().any(|kind| word_is(token, kind))
+            }) {
+                if word_is(&tokens[index], "VIEW") {
+                    cursor.index = index + 1;
+                    self.views.insert(cursor.table(&self.database).ok_or_else(|| unsupported(sql))?);
+                }
+                return Ok(Scope::Skip);
+            }
+        }
+        tokenized.map_err(|_| unsupported(sql))?;
+        if limit != sql.len() {
+            return Err(unsupported(sql));
+        }
+        cursor.index = 0;
+        if cursor.eat("USE") {
+            let database = cursor.identifier().ok_or_else(|| unsupported(sql))?;
+            self.database = Some(database.clone());
+            return Ok(Scope::DatabaseSession(database));
+        }
+        if cursor.eat("SET") {
+            if cursor.eat("STATEMENT") {
+                // MariaDB's SET STATEMENT ... FOR can wrap arbitrary mutations.
+                return Err(unsupported(sql));
+            }
+            // Server-wide settings (including GTID_PURGED) are not part of a table restore.
+            return Ok(
+                if tokens.iter().any(|token| {
+                    word_is(token, "GLOBAL")
+                        || word_is(token, "@@GLOBAL")
+                        || word_is(token, "PERSIST")
+                        || word_is(token, "PERSIST_ONLY")
+                }) {
+                    Scope::Skip
+                } else {
+                    Scope::Session
+                },
+            );
+        }
+        if cursor.eat("BEGIN") {
+            cursor.eat("WORK");
+            cursor.eat_token(&Token::SemiColon);
+            return if cursor.index == tokens.len() { Ok(Scope::Session) } else { Err(unsupported(sql)) };
+        }
+        if cursor.eat("START") && cursor.eat("TRANSACTION")
+            || word_is(&tokens[0], "COMMIT")
+            || word_is(&tokens[0], "ROLLBACK")
+        {
+            return Ok(Scope::Session);
+        }
+        if word_is(&tokens[0], "LOCK") || word_is(&tokens[0], "UNLOCK") {
+            // The SQL-file executor already omits dump table locks.
+            return Ok(Scope::Skip);
+        }
+        cursor.index = 0;
+        if cursor.eat("CREATE") {
+            if cursor.eat("DATABASE") || cursor.eat("SCHEMA") {
+                // Keep only idempotent bootstrap, never destructive database DDL.
+                if cursor.eat("IF") && cursor.eat("NOT") && cursor.eat("EXISTS") {
+                    return Ok(Scope::DatabaseSession(cursor.identifier().ok_or_else(|| unsupported(sql))?));
+                }
+                return Ok(Scope::Skip);
+            }
+            cursor.eat("UNIQUE");
+            cursor.eat("FULLTEXT");
+            cursor.eat("SPATIAL");
+            if cursor.eat("INDEX") {
+                if let Some(index) = tokens.iter().position(|token| word_is(token, "ON")) {
+                    cursor.index = index + 1;
+                    return cursor.table(&self.database).map(Scope::Table).ok_or_else(|| unsupported(sql));
+                }
+            }
+        }
+        cursor.index = 0;
+        if cursor.eat("DROP") {
+            if cursor.eat("TEMPORARY") {
+                return Err(unsupported(sql));
+            }
+            if cursor.eat("TABLE") {
+                let if_exists = cursor.eat("IF") && cursor.eat("EXISTS");
+                let mut tables = vec![cursor.table(&self.database).ok_or_else(|| unsupported(sql))?];
+                while cursor.eat_token(&Token::Comma) {
+                    tables.push(cursor.table(&self.database).ok_or_else(|| unsupported(sql))?);
+                }
+                cursor.eat("RESTRICT");
+                cursor.eat("CASCADE");
+                cursor.eat_token(&Token::SemiColon);
+                if cursor.index != tokens.len() {
+                    return Err(unsupported(sql));
+                }
+                return Ok(Scope::DropTables(tables, if_exists));
+            }
+            if ["DATABASE", "SCHEMA", "VIEW", "PROCEDURE", "FUNCTION", "TRIGGER", "EVENT"]
+                .iter()
+                .any(|kind| cursor.eat(kind))
+            {
+                return Ok(Scope::Skip);
+            }
+        }
+        cursor.index = 0;
+        if cursor.eat("ALTER") && cursor.eat("TABLE") {
+            let table = cursor.table(&self.database).ok_or_else(|| unsupported(sql))?;
+            // Renaming/exchanging a table can also mutate an unselected table.
+            if tokens[cursor.index..].iter().any(|token| word_is(token, "RENAME") || word_is(token, "EXCHANGE")) {
+                return Err(unsupported(sql));
+            }
+            return Ok(Scope::Table(table));
+        }
+        Err(unsupported(sql))
+    }
+}
+
+fn unsupported(sql: &str) -> String {
+    format!("Cannot safely restore selected tables from this statement: {}", crate::sql::statement_summary(sql))
+}
+
+fn word_is(token: &Token, expected: &str) -> bool {
+    matches!(token, Token::Word(word) if word.quote_style.is_none() && word.value.eq_ignore_ascii_case(expected))
+}
+
+struct Tokens<'a> {
+    tokens: &'a [Token],
+    index: usize,
+}
+
+impl Tokens<'_> {
+    fn eat(&mut self, word: &str) -> bool {
+        if self.tokens.get(self.index).is_some_and(|token| word_is(token, word)) {
+            self.index += 1;
+            true
+        } else {
+            false
+        }
+    }
+
+    fn eat_token(&mut self, token: &Token) -> bool {
+        if self.tokens.get(self.index) == Some(token) {
+            self.index += 1;
+            true
+        } else {
+            false
+        }
+    }
+
+    fn eat_if_not_exists(&mut self) {
+        if self.eat("IF") {
+            self.eat("NOT");
+            self.eat("EXISTS");
+        }
+    }
+
+    fn identifier(&mut self) -> Option<String> {
+        let Token::Word(word) = self.tokens.get(self.index)? else {
+            return None;
+        };
+        self.index += 1;
+        Some(word.value.clone())
+    }
+
+    fn table(&mut self, database: &Option<String>) -> Option<SqlFileTable> {
+        let first = self.identifier()?;
+        if self.eat_token(&Token::Period) {
+            Some(SqlFileTable { database: Some(first), name: self.identifier()? })
+        } else {
+            Some(SqlFileTable { database: database.clone(), name: first })
+        }
+    }
+}
+
+fn quoted_table(table: &SqlFileTable) -> String {
+    let quote = |name: &str| format!("`{}`", name.replace('`', "``"));
+    match &table.database {
+        Some(database) => format!("{}.{}", quote(database), quote(&table.name)),
+        None => quote(&table.name),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::models::connection::DatabaseType;
+    use crate::sql::split_sql_statements_for_database;
+
+    fn table(database: Option<&str>, name: &str) -> SqlFileTable {
+        SqlFileTable { database: database.map(str::to_string), name: name.to_string() }
+    }
+
+    fn restore(sql: &str, selected: &[SqlFileTable]) -> Result<Vec<String>, String> {
+        let statements = split_sql_statements_for_database(sql, DatabaseType::Mysql);
+        let mut scan = TableRestoreFilter::default();
+        for statement in &statements {
+            scan.inspect(statement)?;
+        }
+        let (_, views) = scan.finish();
+        let mut filter = TableRestoreFilter::with_views(views);
+        statements.iter().filter_map(|sql| filter.filter(sql, selected).transpose()).collect()
+    }
+
+    #[test]
+    fn restores_only_selected_structure_and_data_with_session_context() {
+        let sql = "SET FOREIGN_KEY_CHECKS=0; CREATE TABLE `a` (id INT); INSERT INTO a VALUES (1, 'DROP TABLE b;'); CREATE TABLE b (id INT); INSERT INTO b VALUES (2); SET FOREIGN_KEY_CHECKS=1;";
+        assert_eq!(
+            restore(sql, &[table(None, "a")]).unwrap(),
+            vec![
+                "SET FOREIGN_KEY_CHECKS=0",
+                "CREATE TABLE `a` (id INT)",
+                "INSERT INTO a VALUES (1, 'DROP TABLE b;')",
+                "SET FOREIGN_KEY_CHECKS=1",
+            ]
+        );
+    }
+
+    #[test]
+    fn distinguishes_databases_and_escaped_identifiers() {
+        let sql = "USE first; INSERT INTO `a``b` VALUES (1); USE second; INSERT INTO `a``b` VALUES (2); INSERT INTO first.`a``b` VALUES (3);";
+        let statements = restore(sql, &[table(Some("first"), "a`b")]).unwrap();
+        assert_eq!(statements.iter().filter(|sql| sql.starts_with("INSERT")).count(), 2);
+        assert!(!statements.iter().any(|sql| sql.contains("(2)")));
+    }
+
+    #[test]
+    fn filters_multi_table_drop_without_dropping_unselected_tables() {
+        assert_eq!(
+            restore("DROP TABLE IF EXISTS a, b;", &[table(None, "a")]).unwrap(),
+            vec!["DROP TABLE IF EXISTS `a`"]
+        );
+    }
+
+    #[test]
+    fn excludes_view_placeholders_routines_and_global_changes() {
+        let sql = "CREATE TABLE v (id INT); DROP TABLE IF EXISTS v; CREATE VIEW v AS SELECT * FROM a; DROP DATABASE other; SET @@GLOBAL.GTID_PURGED='x'; CREATE TABLE a (id INT);";
+        assert_eq!(restore(sql, &[table(None, "a"), table(None, "v")]).unwrap(), vec!["CREATE TABLE a (id INT)"]);
+    }
+
+    #[test]
+    fn handles_version_comments_key_toggles_and_insert_modifiers() {
+        let sql = "/*!40000 ALTER TABLE a DISABLE KEYS */; INSERT IGNORE INTO a VALUES (1); REPLACE INTO b VALUES (2); /*!40000 ALTER TABLE a ENABLE KEYS */;";
+        assert_eq!(restore(sql, &[table(None, "a")]).unwrap().len(), 3);
+    }
+
+    #[test]
+    fn rejects_unknown_and_multi_target_mutations_before_execution() {
+        for sql in [
+            "CALL remove_all()",
+            "UPDATE a JOIN b SET b.id=1",
+            "ALTER TABLE a RENAME TO b",
+            "RENAME TABLE a TO b",
+            "DROP TEMPORARY TABLE a",
+            "CREATE TEMPORARY TABLE a (id INT)",
+            "SET STATEMENT max_statement_time=1 FOR DELETE FROM b",
+            "BEGIN NOT ATOMIC DELETE FROM b; END",
+        ] {
+            assert!(restore(sql, &[table(None, "a")]).is_err(), "{sql}");
+        }
+    }
+
+    #[test]
+    fn identifies_large_insert_without_tokenizing_its_values() {
+        let sql = format!("INSERT INTO a VALUES ('{}');", "x".repeat(1_000_000));
+        assert_eq!(restore(&sql, &[table(None, "a")]).unwrap().len(), 1);
+        assert!(restore(&sql, &[table(None, "b")]).unwrap().is_empty());
+    }
+
+    #[test]
+    fn identifies_large_definitions_without_parsing_their_bodies() {
+        let body = "x".repeat(100_000);
+        let ddl = format!("CREATE TABLE a (body TEXT COMMENT '{body}')");
+        let sql = format!("{ddl}; CREATE VIEW v AS SELECT '{body}';\nDELIMITER ;;\nCREATE PROCEDURE p() BEGIN SELECT '{body}'; END;;\nDELIMITER ;\n");
+        assert_eq!(restore(&sql, &[table(None, "a")]).unwrap(), vec![ddl]);
+    }
+
+    #[test]
+    fn skips_mysqldump_view_placeholders_and_delimited_triggers() {
+        let sql = "/*!40101 SET @OLD_CHARACTER_SET_CLIENT=@@CHARACTER_SET_CLIENT */;
+            CREATE TABLE a (id INT);
+            CREATE TABLE v (id INT);
+            DELIMITER ;;
+            /*!50003 CREATE*/ /*!50017 DEFINER=`root`@`localhost`*/ /*!50003 TRIGGER tr AFTER INSERT ON a FOR EACH ROW BEGIN INSERT INTO audit VALUES (NEW.id); END */;;
+            DELIMITER ;
+            DROP TABLE IF EXISTS v;
+            /*!50001 CREATE ALGORITHM=UNDEFINED */ /*!50013 DEFINER=`root`@`localhost` SQL SECURITY DEFINER */ /*!50001 VIEW v AS SELECT id FROM a */;
+            INSERT INTO a VALUES (1);
+            /*!40101 SET CHARACTER_SET_CLIENT=@OLD_CHARACTER_SET_CLIENT */;";
+        let statements = restore(sql, &[table(None, "a"), table(None, "v")]).unwrap();
+        assert_eq!(statements.len(), 4);
+        assert!(statements.iter().any(|sql| sql == "INSERT INTO a VALUES (1)"));
+    }
+}
diff --git a/crates/dbx-core/tests/live_mysql57.rs b/crates/dbx-core/tests/live_mysql57.rs
index 88ee0fe0f9..5a96ebe6d3 100644
--- a/crates/dbx-core/tests/live_mysql57.rs
+++ b/crates/dbx-core/tests/live_mysql57.rs
@@ -1173,6 +1173,7 @@ INSERT INTO install_check (id) VALUES (1), (2);
             .to_string_lossy()
             .into_owned(),
         continue_on_error: false,
+        selected_tables: None,
     };
 
     let _ = execute_sql_statement(
@@ -1258,6 +1259,7 @@ INSERT INTO children (parent_id) VALUES (LAST_INSERT_ID());
             .to_string_lossy()
             .into_owned(),
         continue_on_error: false,
+        selected_tables: None,
     };
 
     let _ = execute_sql_statement(
@@ -1327,6 +1329,7 @@ async fn live_sql_file_import_preserves_raw_mysql_binary_literal_bytes() {
         database: String::new(),
         file_path: std::env::temp_dir().join(format!("mysql-binary-dump-{suffix}.sql")).to_string_lossy().into_owned(),
         continue_on_error: false,
+        selected_tables: None,
     };
 
     tokio::fs::write(&request.file_path, script).await.unwrap();
diff --git a/crates/dbx-core/tests/live_mysql_database_export.rs b/crates/dbx-core/tests/live_mysql_database_export.rs
index 79e74da41f..a89f0bf0e3 100644
--- a/crates/dbx-core/tests/live_mysql_database_export.rs
+++ b/crates/dbx-core/tests/live_mysql_database_export.rs
@@ -39,6 +39,74 @@ fn live_mysql_config(id: &str) -> ConnectionConfig {
     .expect("live MySQL export config should deserialize")
 }
 
+#[tokio::test]
+#[ignore = "requires a disposable MySQL endpoint"]
+async fn live_mysql_selected_table_restore_preserves_unselected_tables() {
+    use dbx_core::sql_file_import::{inspect_sql_file_tables, SqlFileTable};
+    use futures::FutureExt;
+    use std::io::Write;
+
+    let suffix = uuid::Uuid::new_v4().simple().to_string();
+    let connection_id = format!("selected-restore-{suffix}");
+    let database = format!("dbx_restore_{suffix}");
+    let dir = std::env::temp_dir().join(format!("dbx-selected-restore-{suffix}"));
+    std::fs::create_dir_all(&dir).unwrap();
+    let storage = Storage::open(&dir.join("storage.db")).await.unwrap();
+    let state = AppState::new(storage);
+    state.configs.write().await.insert(connection_id.clone(), live_mysql_config(&connection_id));
+    execute_sql_statement(&state, &connection_id, "", &format!("CREATE DATABASE `{database}`"), None, None)
+        .await
+        .unwrap();
+
+    let outcome = std::panic::AssertUnwindSafe(async {
+        execute_sql_statement(&state, &connection_id, &database, "CREATE TABLE untouched (id INT PRIMARY KEY)", None, None).await?;
+        execute_sql_statement(&state, &connection_id, &database, "INSERT INTO untouched VALUES (99)", None, None).await?;
+        let script = format!("CREATE DATABASE IF NOT EXISTS `{database}`; USE `{database}`; SET FOREIGN_KEY_CHECKS=0; DROP TABLE IF EXISTS chosen, untouched; CREATE TABLE chosen (id INT PRIMARY KEY, body LONGTEXT); INSERT INTO chosen VALUES (7, '{}'); CREATE TABLE untouched (id INT PRIMARY KEY); INSERT INTO untouched VALUES (1); CREATE VIEW skipped_view AS SELECT id FROM chosen; SET FOREIGN_KEY_CHECKS=1;", "x".repeat(400_000));
+        for compressed in [false, true] {
+            let path = dir.join(if compressed { "backup.sql.gz" } else { "backup.sql" });
+            if compressed {
+                let mut writer = flate2::write::GzEncoder::new(std::fs::File::create(&path).map_err(|e| e.to_string())?, flate2::Compression::default());
+                writer.write_all(script.as_bytes()).map_err(|e| e.to_string())?;
+                writer.finish().map_err(|e| e.to_string())?;
+            } else {
+                std::fs::write(&path, &script).map_err(|e| e.to_string())?;
+            }
+            let tables = inspect_sql_file_tables(&path).await?;
+            assert_eq!(tables.len(), 2);
+            let request = SqlFileRequest {
+                execution_id: format!("restore-{suffix}-{compressed}"), connection_id: connection_id.clone(), database: database.clone(), file_path: path.display().to_string(), continue_on_error: false,
+                selected_tables: Some(vec![SqlFileTable { database: Some(database.clone()), name: "chosen".into() }]),
+            };
+            execute_sql_file_path(&state, &request, &path, CancellationToken::new(), std::time::Instant::now(), |_| {}).await?;
+            let rows = execute_sql_statement(&state, &connection_id, &database, "SELECT id, LENGTH(body) FROM chosen", None, None).await?;
+            assert_eq!(rows.rows, vec![vec![serde_json::json!("7"), serde_json::json!("400000")]]);
+            let untouched = execute_sql_statement(&state, &connection_id, &database, "SELECT id FROM untouched", None, None).await?;
+            assert_eq!(untouched.rows, vec![vec![serde_json::json!("99")]]);
+            let views = execute_sql_statement(&state, &connection_id, &database, "SHOW FULL TABLES WHERE Table_type = 'VIEW'", None, None).await?;
+            assert!(views.rows.is_empty());
+        }
+        let path = dir.join("unsupported.sql");
+        std::fs::write(&path, "DROP TABLE chosen; INSERT INTO chosen VALUES (1, 'bad'); CALL unexpected();").map_err(|e| e.to_string())?;
+        let request = SqlFileRequest {
+            execution_id: format!("invalid-{suffix}"), connection_id: connection_id.clone(), database: database.clone(), file_path: path.display().to_string(), continue_on_error: true,
+            selected_tables: Some(vec![SqlFileTable { database: None, name: "chosen".into() }]),
+        };
+        let mut events = Vec::new();
+        assert!(execute_sql_file_path(&state, &request, &path, CancellationToken::new(), std::time::Instant::now(), |event| events.push(event)).await.is_err());
+        assert_eq!(events.last().unwrap().success_count, 0);
+        let rows = execute_sql_statement(&state, &connection_id, &database, "SELECT id FROM chosen", None, None).await?;
+        assert_eq!(rows.rows, vec![vec![serde_json::json!("7")]]);
+        Ok::<_, String>(())
+    }).catch_unwind().await;
+
+    let cleanup =
+        execute_sql_statement(&state, &connection_id, "", &format!("DROP DATABASE `{database}`"), None, None).await;
+    drop(state);
+    std::fs::remove_dir_all(&dir).unwrap();
+    cleanup.unwrap();
+    outcome.unwrap().unwrap();
+}
+
 #[tokio::test]
 #[ignore = "requires a disposable MySQL endpoint"]
 async fn live_mysql_database_export_restores_dependent_views() {
@@ -102,6 +170,7 @@ async fn live_mysql_database_export_restores_dependent_views() {
             database: String::new(),
             file_path: file_path.to_string_lossy().to_string(),
             continue_on_error: false,
+            selected_tables: None,
         };
         execute_sql_file_path(
             &state,
diff --git a/crates/dbx-core/tests/live_postgres_all_schema_export.rs b/crates/dbx-core/tests/live_postgres_all_schema_export.rs
index 30f3982a46..06f390cf53 100644
--- a/crates/dbx-core/tests/live_postgres_all_schema_export.rs
+++ b/crates/dbx-core/tests/live_postgres_all_schema_export.rs
@@ -149,6 +149,7 @@ async fn live_postgres_all_schema_export_restores_one_sql_file() {
             database: target_database.clone(),
             file_path: export_path.display().to_string(),
             continue_on_error: false,
+            selected_tables: None,
         },
         &export_path,
         CancellationToken::new(),
diff --git a/crates/dbx-core/tests/live_sqlserver_completion.rs b/crates/dbx-core/tests/live_sqlserver_completion.rs
index 5d3bc9b96a..e594bbcbc7 100644
--- a/crates/dbx-core/tests/live_sqlserver_completion.rs
+++ b/crates/dbx-core/tests/live_sqlserver_completion.rs
@@ -1290,6 +1290,7 @@ async fn live_sqlserver_sql_file_import_executes_go_batches() {
         database: database.clone(),
         file_path: "fixture.sql".to_string(),
         continue_on_error: false,
+        selected_tables: None,
     };
     let done_seen = AtomicBool::new(false);
 
diff --git a/crates/dbx-web/src/main.rs b/crates/dbx-web/src/main.rs
index 9d69e660cb..ff1440490b 100644
--- a/crates/dbx-web/src/main.rs
+++ b/crates/dbx-web/src/main.rs
@@ -1021,6 +1021,7 @@ async fn main() {
                 .layer(DefaultBodyLimit::max(routes::sql_file::sql_file_upload_hard_cap_bytes())),
         )
         .route("/sql-file/execute", post(routes::sql_file::execute_sql_file))
+        .route("/sql-file/tables", post(routes::sql_file::inspect_sql_file_tables))
         .route("/sql-file/progress/{executionId}", get(routes::sql_file::sql_file_progress))
         .route("/sql-file/cancel", post(routes::sql_file::cancel_sql_file))
         // Table import
diff --git a/crates/dbx-web/src/routes/sql_file.rs b/crates/dbx-web/src/routes/sql_file.rs
index 70ec20868c..9bf1952720 100644
--- a/crates/dbx-web/src/routes/sql_file.rs
+++ b/crates/dbx-web/src/routes/sql_file.rs
@@ -5,7 +5,6 @@ use std::time::Duration;
 use axum::extract::{Multipart, Path as AxumPath, State};
 use axum::response::sse::{Event, Sse};
 use axum::Json;
-use dbx_core::sql;
 use dbx_core::sql::{SqlFileProgress, SqlFileRequest, SqlFileStatus};
 use dbx_core::sql_file_import::{
     execute_sql_file_paths, sql_file_error_progress, sql_file_progress as build_sql_file_progress,
@@ -63,6 +62,20 @@ pub struct CancelSqlFileRequest {
     pub execution_id: String,
 }
 
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct InspectSqlFileTablesRequest {
+    pub file_path: String,
+}
+
+pub async fn inspect_sql_file_tables(
+    State(state): State<Arc<WebState>>,
+    Json(body): Json<InspectSqlFileTablesRequest>,
+) -> Result<Json<Vec<dbx_core::sql_file_import::SqlFileTable>>, AppError> {
+    let path = validated_uploaded_sql_path(&state.data_dir, &body.file_path)?;
+    Ok(Json(dbx_core::sql_file_import::inspect_sql_file_tables(&path).await.map_err(AppError::from)?))
+}
+
 pub async fn preview_sql_file(
     State(state): State<Arc<WebState>>,
     mut multipart: Multipart,
@@ -83,7 +96,8 @@ pub async fn preview_sql_file(
         std::fs::write(&file_path, &data).map_err(|e| AppError::from(e.to_string()))?;
 
         let size_bytes = data.len() as u64;
-        let content = sql::decode_sql_file_bytes(&data).map_err(AppError::from)?;
+        let content =
+            dbx_core::sql_file_import::read_sql_file_preview(&file_path, 1_000_000).await.map_err(AppError::from)?;
         let preview: String = content.chars().take(20_000).collect();
         let bootstrap_analysis = dbx_core::sql_file_import::mysql_like_sql_file_bootstrap_analysis(&content);
 
@@ -212,6 +226,9 @@ fn safe_uploaded_sql_path(tmp_dir: &Path, file_name: &str) -> Result<PathBuf, Ap
     let extension =
         file_name.extension().and_then(|extension| extension.to_str()).filter(|extension| !extension.is_empty());
     let unique_name = match extension {
+        Some(extension) if extension.eq_ignore_ascii_case("gz") && stem.to_ascii_lowercase().ends_with(".sql") => {
+            format!("{}-{}.sql.gz", &stem[..stem.len() - 4], Uuid::new_v4())
+        }
         Some(extension) => format!("{stem}-{}.{}", Uuid::new_v4(), extension),
         None => format!("{stem}-{}", Uuid::new_v4()),
     };
@@ -334,6 +351,8 @@ mod tests {
         assert!(second.starts_with(&tmp_dir));
         assert_ne!(first, second);
         assert_eq!(first.extension().and_then(|extension| extension.to_str()), Some("sql"));
+        let compressed = safe_uploaded_sql_path(&tmp_dir, "backup.sql.gz").unwrap();
+        assert!(compressed.file_name().unwrap().to_string_lossy().ends_with(".sql.gz"));
         let _ = std::fs::remove_dir_all(data_dir);
     }
 
diff --git a/docs/content/docs/database-backup.cn.mdx b/docs/content/docs/database-backup.cn.mdx
index 16ce121efb..fa582e0f13 100644
--- a/docs/content/docs/database-backup.cn.mdx
+++ b/docs/content/docs/database-backup.cn.mdx
@@ -124,6 +124,10 @@ orders_202?
 
 ## 恢复演练
 
+MySQL 兼容连接支持从单个 `.sql` 或 `.sql.gz` 备份中恢复指定表。在备份历史中展开文件列表并点击**恢复**，将**恢复范围**切换为**指定表**。扫描完成后，可以搜索并勾选表，再执行恢复。列表来自完整备份文件，因此目标数据库尚不存在的表也能选择；跨数据库同名表会分别显示。
+
+指定表模式仅恢复所选表的结构、索引和数据，不恢复视图、存储过程、函数、触发器或事件。文件中的数据库上下文仍然生效，未显式指定数据库的表使用弹窗所选的目标数据库。现有外键或触发器可能影响恢复；需要完整恢复所有对象时选择**全部内容**。遇到无法确定作用范围的语句时，DBX 会在执行前拒绝选择性恢复。
+
 生成的 `.sql` 文件可以通过 [SQL 文件执行](/cn/docs/sql-file) 导入。建议定期在隔离环境验证：
 
 - 文件可以完整读取和执行
diff --git a/docs/content/docs/database-backup.mdx b/docs/content/docs/database-backup.mdx
index addc8d27be..2606e6f77c 100644
--- a/docs/content/docs/database-backup.mdx
+++ b/docs/content/docs/database-backup.mdx
@@ -124,6 +124,10 @@ Each schedule has a retention count. Older runs beyond the limit are pruned from
 
 ## Restore Drills
 
+For MySQL-compatible connections, restore individual tables from a single `.sql` or `.sql.gz` backup. Expand a backup's file list, click **Restore**, then choose **Selected tables** under **Restore scope**. Once the full file has been scanned, search and select the tables to restore. Tables do not need to exist in the destination, and identically named tables in different source databases are listed separately.
+
+Selected-table mode restores table structure, indexes, and data, excluding views, routines, triggers, and events. Database context in the dump still applies; tables without an explicit database use the destination chosen in the dialog. Existing foreign keys or triggers can affect the restore. Choose **All contents** for a complete restore. Statements whose scope cannot be determined cause selective restore to fail before execution.
+
 Generated `.sql` files can be imported through [SQL File Execution](/en/docs/sql-file). Regularly verify in an isolated environment that:
 
 - The complete file can be read and executed
diff --git a/issue-assets/issue-8729-search-tables.png b/issue-assets/issue-8729-search-tables.png
new file mode 100644
index 0000000000..204a388bc8
Binary files /dev/null and b/issue-assets/issue-8729-search-tables.png differ
diff --git a/issue-assets/issue-8729-select-tables.png b/issue-assets/issue-8729-select-tables.png
new file mode 100644
index 0000000000..1f81511b0c
Binary files /dev/null and b/issue-assets/issue-8729-select-tables.png differ
diff --git a/src-tauri/src/commands/sql_file.rs b/src-tauri/src/commands/sql_file.rs
index 07fe298e0a..76e94bdec7 100644
--- a/src-tauri/src/commands/sql_file.rs
+++ b/src-tauri/src/commands/sql_file.rs
@@ -30,6 +30,13 @@ struct SqlFileSummary {
     failed_statement_index: Option<usize>,
 }
 
+#[tauri::command]
+pub async fn inspect_sql_file_tables(
+    file_path: String,
+) -> Result<Vec<dbx_core::sql_file_import::SqlFileTable>, String> {
+    dbx_core::sql_file_import::inspect_sql_file_tables(std::path::Path::new(&file_path)).await
+}
+
 #[tauri::command]
 pub async fn preview_sql_file(file_path: String) -> Result<SqlFilePreview, String> {
     let path = PathBuf::from(&file_path);
diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs
index fc00fccb88..dbf59c791b 100644
--- a/src-tauri/src/lib.rs
+++ b/src-tauri/src/lib.rs
@@ -1960,6 +1960,7 @@ pub fn run() {
             commands::data_compare::prepare_data_compare_missing_target,
             commands::data_compare::build_data_compare_sync_plan,
             commands::sql_file::preview_sql_file,
+            commands::sql_file::inspect_sql_file_tables,
             commands::sql_file::execute_sql_file,
             commands::sql_file::execute_sql_files,
             commands::sql_file::cancel_sql_file_execution,