diff --git a/agents/drivers/h2/src/main/java/com/dbx/agent/h2/H2Agent.java b/agents/drivers/h2/src/main/java/com/dbx/agent/h2/H2Agent.java
index e3c64be7a8..ce2abcc751 100644
--- a/agents/drivers/h2/src/main/java/com/dbx/agent/h2/H2Agent.java
+++ b/agents/drivers/h2/src/main/java/com/dbx/agent/h2/H2Agent.java
@@ -28,8 +28,20 @@ import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
+import java.util.regex.Pattern;
 
 public class H2Agent extends AbstractJdbcAgent {
+    private static final String QUOTED_IDENTIFIER = "\"(?:\"\"|[^\"])*\"";
+    private static final String QUALIFIED_IDENTIFIER = QUOTED_IDENTIFIER + "\\." + QUOTED_IDENTIFIER;
+    private static final Pattern CREATE_SEQUENCE = Pattern.compile("^CREATE SEQUENCE (" + QUALIFIED_IDENTIFIER + ")\\s");
+    private static final Pattern SEQUENCE_REFERENCE = Pattern.compile(
+        "'(?:''|[^'])*'|" + QUOTED_IDENTIFIER + "|--[^\\r\\n]*|/\\*.*?\\*/|\\b(?:NEXT|CURRENT)\\s+VALUE\\s+FOR\\s+(" + QUALIFIED_IDENTIFIER + ")",
+        Pattern.DOTALL
+    );
+    private static final Pattern CREATE_INDEX = Pattern.compile(
+        "^CREATE (?:UNIQUE(?: NULLS (?:DISTINCT|NOT DISTINCT|ALL DISTINCT))? )?(?:HASH |SPATIAL )?INDEX .*",
+        Pattern.DOTALL
+    );
     private String databaseName = "";
     private H2DriverLoader.LoadedDriver loadedDriver;
     private H2DriverVersion driverVersion = H2DriverVersion.V3;
@@ -322,6 +334,55 @@ public class H2Agent extends AbstractJdbcAgent {
         });
     }
 
+    @Override
+    public String getTableDdl(String schema, String table) {
+        return unchecked(() -> {
+            String tableRef = JdbcIdentifiers.INSTANCE.doubleQuote(resolveSchema(schema)) + "."
+                + JdbcIdentifiers.INSTANCE.doubleQuote(table);
+            List<String> statements = new ArrayList<>();
+            List<String> sequences = new ArrayList<>();
+            // Let H2 preserve identity/computed columns, constraints and index ordering.
+            // Exclude database-wide settings, users, schemas, grants and triggers.
+            try (var stmt = requireConnected().createStatement();
+                 ResultSet rs = stmt.executeQuery("SCRIPT NODATA NOPASSWORDS NOSETTINGS TABLE " + tableRef)) {
+                while (rs.next()) {
+                    String sql = rs.getString(1).trim();
+                    if (sql.startsWith("CREATE SEQUENCE ")) {
+                        sequences.add(sql);
+                    } else if (sql.startsWith("CREATE MEMORY TABLE ") || sql.startsWith("CREATE CACHED TABLE ")
+                        || sql.startsWith("CREATE TABLE ") || sql.startsWith("ALTER TABLE ")
+                        || CREATE_INDEX.matcher(sql).matches()) {
+                        statements.add(sql);
+                    }
+                }
+            }
+            if (statements.isEmpty()) {
+                throw new IllegalStateException("H2 returned no table DDL for " + tableRef);
+            }
+            // SCRIPT TABLE can include every sequence in the schema. Keep only real
+            // references, skipping quoted strings, identifiers and comments as tokens.
+            Set<String> referenced = new HashSet<>();
+            for (String statement : statements) {
+                var references = SEQUENCE_REFERENCE.matcher(statement);
+                while (references.find()) {
+                    if (references.group(1) != null) {
+                        referenced.add(references.group(1));
+                    }
+                }
+            }
+            List<String> ddl = new ArrayList<>();
+            for (String sequence : sequences) {
+                var declaration = CREATE_SEQUENCE.matcher(sequence);
+                if (declaration.find() && referenced.contains(declaration.group(1))) {
+                    // Several selected tables may share one explicit sequence.
+                    ddl.add("CREATE SEQUENCE IF NOT EXISTS " + sequence.substring("CREATE SEQUENCE ".length()));
+                }
+            }
+            ddl.addAll(statements);
+            return String.join("\n", ddl);
+        });
+    }
+
     @Override
     public List<ColumnInfo> getColumns(String schema, String table) {
         return unchecked(() -> {
@@ -349,7 +410,10 @@ public class H2Agent extends AbstractJdbcAgent {
 
             List<ColumnInfo> result = new ArrayList<>();
             String typeColumn = isVersion2OrLater() ? "DATA_TYPE" : "TYPE_NAME";
-            String columnSql = "SELECT COLUMN_NAME, " + typeColumn + " AS DBX_DATA_TYPE, IS_NULLABLE, COLUMN_DEFAULT, "
+            String extraColumns = isVersion2OrLater()
+                ? "IS_GENERATED, IS_IDENTITY, IDENTITY_GENERATION, "
+                : "IS_COMPUTED, SEQUENCE_NAME, ";
+            String columnSql = "SELECT COLUMN_NAME, " + typeColumn + " AS DBX_DATA_TYPE, IS_NULLABLE, COLUMN_DEFAULT, " + extraColumns
                 + "NUMERIC_PRECISION, NUMERIC_SCALE, CHARACTER_MAXIMUM_LENGTH, REMARKS "
                 + "FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? ORDER BY ORDINAL_POSITION";
             try (var stmt = requireConnected().prepareStatement(columnSql)) {
@@ -358,14 +422,26 @@ public class H2Agent extends AbstractJdbcAgent {
                 try (ResultSet rs = stmt.executeQuery()) {
                     while (rs.next()) {
                         String columnName = rs.getString("COLUMN_NAME");
+                        String extra = null;
+                        if (isVersion2OrLater()) {
+                            if ("ALWAYS".equals(rs.getString("IS_GENERATED"))) {
+                                extra = "computed";
+                            } else if ("YES".equals(rs.getString("IS_IDENTITY"))) {
+                                extra = "generated " + rs.getString("IDENTITY_GENERATION").toLowerCase(Locale.ROOT) + " as identity";
+                            }
+                        } else if (rs.getBoolean("IS_COMPUTED")) {
+                            extra = "computed";
+                        } else if (rs.getString("SEQUENCE_NAME") != null) {
+                            extra = "identity";
+                        }
                         result.add(new ColumnInfo(
                             columnName,
                             rs.getString("DBX_DATA_TYPE"),
                             "YES".equals(rs.getString("IS_NULLABLE")),
                             rs.getString("COLUMN_DEFAULT"),
                             primaryKeys.contains(columnName),
+                            extra,
                             rs.getString("REMARKS"),
-                            null,
                             intOrNull(rs, "NUMERIC_PRECISION"),
                             intOrNull(rs, "NUMERIC_SCALE"),
                             intOrNull(rs, "CHARACTER_MAXIMUM_LENGTH")
@@ -587,10 +663,7 @@ public class H2Agent extends AbstractJdbcAgent {
     }
 
     private static String resolveSchema(String schema) {
-        if ("PUBLIC".equalsIgnoreCase(schema) || "INFORMATION_SCHEMA".equalsIgnoreCase(schema)) {
-            return schema.toUpperCase(Locale.ROOT);
-        }
-        return "PUBLIC";
+        return schema == null || schema.isBlank() ? "PUBLIC" : schema;
     }
 
     private static Integer intOrNull(ResultSet rs, String column) throws Exception {
diff --git a/agents/drivers/h2/src/test/java/com/dbx/agent/h2/H2AgentTest.java b/agents/drivers/h2/src/test/java/com/dbx/agent/h2/H2AgentTest.java
index 1079849654..50829aadde 100644
--- a/agents/drivers/h2/src/test/java/com/dbx/agent/h2/H2AgentTest.java
+++ b/agents/drivers/h2/src/test/java/com/dbx/agent/h2/H2AgentTest.java
@@ -355,6 +355,128 @@ class H2AgentMigrationTest {
 }
 
 class H2VersionMatrixTest {
+    @Test
+    void tableDdlPreservesGeneratedColumnsConstraintsAndComments() {
+        for (String profile : List.of("h2-v1", "h2-v2", "h2-v3")) {
+            ConnectParams params = new ConnectParams("", 0, "mem:ddl-" + profile, "sa", "", "", "", false);
+            params.setDriver_profile(profile);
+            H2Agent agent = new H2Agent();
+            agent.connect(params);
+            try {
+                ExecuteQueryOptions options = new ExecuteQueryOptions();
+                agent.executeQuery("CREATE SCHEMA \"Source\"; CREATE SCHEMA \"Target\"", null, options);
+                agent.executeQuery("CREATE SEQUENCE \"Source\".UNUSED; CREATE SEQUENCE \"Source\".SHARED START WITH 10", null, options);
+                agent.executeQuery("CREATE TABLE \"Source\".PARENT (ID INT PRIMARY KEY, TOKEN BIGINT DEFAULT NEXT VALUE FOR \"Source\".SHARED)", null, options);
+                agent.executeQuery("CREATE TABLE \"Source\".CHILD (ID INT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, "
+                    + "PARENT_ID INT, QTY INT DEFAULT 1 CHECK (QTY > 0), TOTAL INT AS (QTY * 2), RAW_BYTES BINARY(8), "
+                    + "TOKEN BIGINT DEFAULT NEXT VALUE FOR \"Source\".SHARED, NOTE VARCHAR(128) DEFAULT 'NEXT VALUE FOR \"Source\".\"UNUSED\"', "
+                    + "CONSTRAINT FK_PARENT FOREIGN KEY (PARENT_ID) REFERENCES \"Source\".PARENT(ID) ON DELETE CASCADE)", null, options);
+                agent.executeQuery("COMMENT ON COLUMN \"Source\".CHILD.QTY IS 'quantity'; COMMENT ON TABLE \"Source\".CHILD IS 'child table'", null, options);
+                agent.executeQuery("CREATE INDEX \"Source\".IDX_QTY ON \"Source\".CHILD(QTY DESC)", null, options);
+                List<ColumnInfo> columns = agent.getColumns("Source", "CHILD");
+                Assertions.assertTrue(columns.get(0).getExtra().contains("identity"), profile);
+                Assertions.assertEquals("computed", columns.get(3).getExtra(), profile);
+                Assertions.assertEquals("quantity", columns.get(2).getComment(), profile);
+
+                for (String table : List.of("PARENT", "CHILD")) {
+                    String ddl = agent.getTableDdl("Source", table);
+                    Assertions.assertFalse(ddl.contains("CREATE USER"), ddl);
+                    Assertions.assertFalse(ddl.contains("CREATE SCHEMA"), ddl);
+                    Assertions.assertFalse(ddl.contains("INSERT INTO"), ddl);
+                    Assertions.assertFalse(ddl.contains("CREATE SEQUENCE IF NOT EXISTS \"Source\".\"UNUSED\""), ddl);
+                    if (table.equals("PARENT")) {
+                        Assertions.assertFalse(ddl.contains("SYSTEM_SEQUENCE_"), ddl);
+                    }
+                    agent.executeQuery(ddl.replace("\"Source\".", "\"Target\"."), null, options);
+                }
+                agent.executeQuery("INSERT INTO \"Target\".PARENT(ID) VALUES (1)", null, options);
+                agent.executeQuery("INSERT INTO \"Target\".CHILD(PARENT_ID) VALUES (1)", null, options);
+                Assertions.assertEquals(List.of(List.of(1, 1, 2)), agent.executeQuery("SELECT ID, QTY, TOTAL FROM \"Target\".CHILD", null, options).getRows(), profile);
+                Assertions.assertEquals("11", agent.executeQuery("SELECT TOKEN FROM \"Target\".CHILD", null, options).getRows().get(0).get(0).toString(), profile);
+                Assertions.assertThrows(RuntimeException.class, () -> agent.executeQuery("INSERT INTO \"Target\".CHILD(QTY) VALUES (-1)", null, options), profile);
+                Assertions.assertTrue(agent.getTableDdl("Target", "CHILD").contains("\"QTY\" DESC"), profile);
+                Assertions.assertEquals("quantity", agent.getColumns("Target", "CHILD").get(2).getComment(), profile);
+                agent.executeQuery("DELETE FROM \"Target\".PARENT", null, options);
+                Assertions.assertTrue(agent.executeQuery("SELECT * FROM \"Target\".CHILD", null, options).getRows().isEmpty(), profile);
+            } finally {
+                agent.disconnect();
+            }
+        }
+    }
+
+    @Test
+    void generatedAlwaysIdentityAcceptsExplicitInsertWithOverride() {
+        for (String profile : List.of("h2-v2", "h2-v3")) {
+            H2Agent agent = connect(profile);
+            try {
+                ExecuteQueryOptions options = new ExecuteQueryOptions();
+                agent.executeQuery("CREATE TABLE ALWAYS_ID (ID INT GENERATED ALWAYS AS IDENTITY PRIMARY KEY, V INT)", null, options);
+                Assertions.assertEquals("generated always as identity", agent.getColumns("PUBLIC", "ALWAYS_ID").get(0).getExtra());
+                agent.executeQuery("INSERT INTO ALWAYS_ID(ID, V) OVERRIDING SYSTEM VALUE VALUES (42, 7)", null, options);
+                Assertions.assertEquals(List.of(List.of(42, 7)), agent.executeQuery("SELECT ID, V FROM ALWAYS_ID", null, options).getRows());
+            } finally {
+                agent.disconnect();
+            }
+        }
+    }
+
+    @Test
+    void transferMetadataAndWritesRespectCustomSchemasAcrossBundledVersions() {
+        for (String profile : List.of("h2-v1", "h2-v2", "h2-v3")) {
+            ConnectParams params = new ConnectParams("", 0, "mem:transfer-" + profile, "sa", "", "", "", false);
+            params.setDriver_profile(profile);
+            H2Agent agent = new H2Agent();
+            agent.connect(params);
+            try {
+                ExecuteQueryOptions options = new ExecuteQueryOptions();
+                agent.executeQuery("CREATE SCHEMA \"TransferTarget\"", null, options);
+                agent.executeQuery("CREATE TABLE PUBLIC.ITEMS (PUBLIC_ONLY INT)", null, options);
+                agent.executeQuery(
+                    "CREATE TABLE \"TransferTarget\".ITEMS (ID INT, PART INT, NAME VARCHAR(1024), AMOUNT DECIMAL(12, 2), PRIMARY KEY (ID, PART))",
+                    null, options
+                );
+                agent.executeQuery("CREATE INDEX \"TransferTarget\".IDX_NAME ON \"TransferTarget\".ITEMS(NAME)", null, options);
+                agent.executeQuery("CREATE UNIQUE HASH INDEX \"TransferTarget\".IDX_PART ON \"TransferTarget\".ITEMS(PART)", null, options);
+
+                Assertions.assertTrue(agent.listSchemas().contains("TransferTarget"), profile);
+                Assertions.assertTrue(agent.listTables("TransferTarget").stream().anyMatch(table -> table.getName().equals("ITEMS")), profile);
+                Assertions.assertTrue(agent.listTables("MISSING_SCHEMA").isEmpty(), profile);
+                Assertions.assertEquals(List.of("PUBLIC_ONLY"), agent.getColumns(null, "ITEMS").stream().map(ColumnInfo::getName).toList(), profile);
+                List<ColumnInfo> columns = agent.getColumns("TransferTarget", "ITEMS");
+                Assertions.assertEquals(List.of("ID", "PART", "NAME", "AMOUNT"), columns.stream().map(ColumnInfo::getName).toList(), profile);
+                Assertions.assertEquals(1024, columns.get(2).getCharacter_maximum_length(), profile);
+                Assertions.assertEquals(12, columns.get(3).getNumeric_precision(), profile);
+                Assertions.assertEquals(2, columns.get(3).getNumeric_scale(), profile);
+                Assertions.assertTrue(agent.listIndexes("TransferTarget", "ITEMS").stream().anyMatch(index -> index.getName().equals("IDX_NAME")), profile);
+                agent.executeQuery("CREATE SCHEMA \"TransferCopy\"", null, options);
+                String ddl = agent.getTableDdl("TransferTarget", "ITEMS");
+                agent.executeQuery(ddl.replace("\"TransferTarget\".", "\"TransferCopy\"."), null, options);
+                Assertions.assertEquals(1024, agent.getColumns("TransferCopy", "ITEMS").get(2).getCharacter_maximum_length(), profile);
+                Assertions.assertTrue(agent.listIndexes("TransferCopy", "ITEMS").stream().anyMatch(index -> index.getName().equals("IDX_NAME")), profile);
+                Assertions.assertTrue(agent.listIndexes("TransferCopy", "ITEMS").stream().anyMatch(index -> index.getName().equals("IDX_PART") && index.getIs_unique()), profile);
+                agent.executeQuery("INSERT INTO \"TransferCopy\".ITEMS VALUES (1, 1, 'first', 1.00)", null, options);
+                Assertions.assertThrows(RuntimeException.class, () -> agent.executeQuery("INSERT INTO \"TransferCopy\".ITEMS VALUES (2, 1, 'duplicate', 2.00)", null, options), profile);
+
+                agent.executeQuery("INSERT INTO \"TransferTarget\".ITEMS VALUES (1, 1, 'old', 1.00)", null, options);
+                agent.executeQuery(
+                    "MERGE INTO \"TransferTarget\".\"ITEMS\" (\"ID\", \"PART\", \"NAME\", \"AMOUNT\") KEY (\"ID\", \"PART\") VALUES "
+                        + "(1, 1, 'updated', 12.34), (1, 2, 'new', 56.78)",
+                    null, options
+                );
+                QueryResult result = agent.executeQuery("SELECT NAME, CAST(AMOUNT AS VARCHAR) FROM \"TransferTarget\".ITEMS ORDER BY PART", null, options);
+                Assertions.assertEquals(List.of(List.of("updated", "12.34"), List.of("new", "56.78")), result.getRows(), profile);
+                agent.executeQuery("TRUNCATE TABLE \"TransferTarget\".ITEMS", null, options);
+                Assertions.assertTrue(agent.executeQuery("SELECT * FROM \"TransferTarget\".ITEMS", null, options).getRows().isEmpty(), profile);
+                agent.executeQuery("CREATE TABLE \"TransferTarget\".PAYLOADS (LABEL VARCHAR(128), DATA JSON, RAW_BYTES VARBINARY)", null, options);
+                agent.executeQuery("INSERT INTO \"TransferTarget\".PAYLOADS VALUES ('C:\\tmp\\o''clock', JSON '[1,2]', X'00ff')", null, options);
+                QueryResult payload = agent.executeQuery("SELECT * FROM \"TransferTarget\".PAYLOADS", null, options);
+                Assertions.assertEquals(List.of(List.of("C:\\tmp\\o'clock", "[1,2]", "0x00ff")), payload.getRows(), profile);
+            } finally {
+                agent.disconnect();
+            }
+        }
+    }
+
     @Test
     void metadataAndExecutionWorkAcrossBundledVersions() {
         for (String profile : List.of("h2-v1", "h2-v2", "h2-v3")) {
diff --git a/agents/test-support/src/main/java/com/dbx/agent/test/JdbcMetadataBehaviorTest.java b/agents/test-support/src/main/java/com/dbx/agent/test/JdbcMetadataBehaviorTest.java
index 9ef0b26669..b7ab16b9a7 100644
--- a/agents/test-support/src/main/java/com/dbx/agent/test/JdbcMetadataBehaviorTest.java
+++ b/agents/test-support/src/main/java/com/dbx/agent/test/JdbcMetadataBehaviorTest.java
@@ -60,7 +60,7 @@ public abstract class JdbcMetadataBehaviorTest extends JdbcConnectedAgentTest {
             for (String column : expectedColumnsInOrder()) {
                 assertTrue(ddl.contains("\"" + column + "\""), "DDL should include column " + column + ": " + ddl);
             }
-            assertTrue(ddl.contains("CREATE TABLE"), "DDL should include CREATE TABLE: " + ddl);
+            assertTrue(ddl.matches("(?s).*\\bCREATE (?:MEMORY |CACHED )?TABLE\\b.*"), "DDL should include CREATE TABLE: " + ddl);
             assertTrue(ddl.contains(metadataColumnsTable()), "DDL should include table name: " + ddl);
         });
     }
diff --git a/apps/desktop/src/components/transfer/__tests__/transferConnections.spec.ts b/apps/desktop/src/components/transfer/__tests__/transferConnections.spec.ts
new file mode 100644
index 0000000000..1a113c0566
--- /dev/null
+++ b/apps/desktop/src/components/transfer/__tests__/transferConnections.spec.ts
@@ -0,0 +1,103 @@
+// @vitest-environment happy-dom
+
+import { createApp, defineComponent, h, nextTick, reactive, type App } from "vue";
+import { afterEach, describe, expect, it, vi } from "vitest";
+import ConnectionTreeSelect from "@/components/connection/ConnectionTreeSelect.vue";
+import { supportsTransfer } from "@/lib/database/databaseCapabilities";
+import { transferDatabaseTypeForConnection } from "@/lib/database/jdbcDialect";
+import type { ConnectionConfig, SidebarLayout } from "@/types/database";
+
+vi.mock("@/components/ui/button", async () => {
+  const { createPassthroughStub } = await import("@/components/grid/__tests__/vueHostHarness");
+  return { Button: createPassthroughStub("Button", "button") };
+});
+
+vi.mock("@/components/ui/popover", async () => {
+  const { createPassthroughStub } = await import("@/components/grid/__tests__/vueHostHarness");
+  return {
+    Popover: createPassthroughStub("Popover"),
+    PopoverContent: createPassthroughStub("PopoverContent"),
+    PopoverTrigger: createPassthroughStub("PopoverTrigger"),
+  };
+});
+
+vi.mock("@/components/icons/DatabaseIcon.vue", () => ({ default: defineComponent({ setup: () => () => h("span") }) }));
+
+const connections: ConnectionConfig[] = [
+  { id: "mysql-source", name: "MySQL Source", db_type: "mysql", host: "localhost", port: 3306, username: "test", password: "" },
+  { id: "pg-source", name: "PostgreSQL Source", db_type: "postgres", host: "localhost", port: 5432, username: "test", password: "" },
+  { id: "h2-local-id", name: "Local H2", db_type: "h2", driver_profile: "h2-v3", host: "", port: 0, username: "sa", password: "", database: "mem:transfer-test" },
+  { id: "h2-jdbc-id", name: "JDBC H2", db_type: "jdbc", connection_string: "jdbc:h2:mem:transfer-test", host: "", port: 0, username: "sa", password: "" },
+  { id: "redis-id", name: "Redis", db_type: "redis", host: "localhost", port: 6379, username: "", password: "" },
+  { id: "unknown-jdbc-id", name: "Unknown JDBC", db_type: "jdbc", host: "localhost", port: 0, username: "", password: "" },
+];
+const layout: SidebarLayout = {
+  groups: [{ id: "embedded", name: "Embedded", collapsed: false }],
+  order: [{ type: "group", id: "embedded", children: [{ type: "connection", id: "h2-local-id" }] }],
+};
+const mountedApps: App[] = [];
+
+async function mountTransferPickers(source: string) {
+  const selection = reactive({ source, target: "" });
+  const eligible = connections.filter((connection) => supportsTransfer(transferDatabaseTypeForConnection(connection)));
+  const container = document.createElement("div");
+  document.body.append(container);
+  const app = createApp(
+    defineComponent({
+      setup: () => () =>
+        h(
+          "div",
+          (["source", "target"] as const).map((side) =>
+            h("section", { "data-side": side }, [
+              h(ConnectionTreeSelect, {
+                modelValue: selection[side],
+                connections: eligible,
+                layout,
+                placeholder: "Select connection",
+                searchPlaceholder: "Search",
+                emptyText: "No connections",
+                "onUpdate:modelValue": (value: string) => {
+                  selection[side] = value;
+                },
+              }),
+            ]),
+          ),
+        ),
+    }),
+  );
+  mountedApps.push(app);
+  app.mount(container);
+  await nextTick();
+  return { container, selection };
+}
+
+afterEach(() => {
+  for (const app of mountedApps.splice(0)) app.unmount();
+  document.body.innerHTML = "";
+});
+
+describe("data transfer connection choices", () => {
+  it.each(["mysql-source", "pg-source"])("offers H2 on both sides when opened from %s", async (source) => {
+    const { container, selection } = await mountTransferPickers(source);
+    for (const side of ["source", "target"]) {
+      const picker = container.querySelector(`[data-side="${side}"]`)!;
+      expect(picker.querySelector('[data-picker-connection="h2-local-id"]')?.textContent).toContain("Local H2");
+      expect(picker.querySelector('[data-picker-connection="h2-jdbc-id"]')?.textContent).toContain("JDBC H2");
+      expect(picker.querySelector('[data-picker-connection="redis-id"]')).toBeNull();
+      expect(picker.querySelector('[data-picker-connection="unknown-jdbc-id"]')).toBeNull();
+    }
+    const target = container.querySelector('[data-side="target"]')!;
+    (target.querySelector('[data-picker-connection="h2-local-id"]') as HTMLButtonElement).click();
+    await nextTick();
+    expect(selection.target).toBe("h2-local-id");
+    expect(target.querySelector("button")?.textContent).toContain("Local H2");
+    expect(selection.source).toBe(source);
+  });
+
+  it.each(["h2-local-id", "h2-jdbc-id"])("resolves the prefilled %s to its connection name", async (source) => {
+    const { container } = await mountTransferPickers(source);
+    const trigger = container.querySelector('[data-side="source"] button');
+    expect(trigger?.textContent).toContain(connections.find((connection) => connection.id === source)!.name);
+    expect(trigger?.textContent).not.toContain(source);
+  });
+});
diff --git a/apps/desktop/src/lib/database/__tests__/transferObjectKinds.spec.ts b/apps/desktop/src/lib/database/__tests__/transferObjectKinds.spec.ts
index 9308f95f3e..920ca0dd59 100644
--- a/apps/desktop/src/lib/database/__tests__/transferObjectKinds.spec.ts
+++ b/apps/desktop/src/lib/database/__tests__/transferObjectKinds.spec.ts
@@ -4,7 +4,7 @@ import { manifestDatabaseTypes } from "@/lib/database/databaseDriverManifest";
 import { supportsTransfer } from "@/lib/database/databaseFeatureSupport";
 import type { DatabaseType } from "@/types/database";
 
-const TABLE_ONLY_TRANSFER_DATABASES: DatabaseType[] = ["sqlite", "rqlite", "turso", "cloudflare-d1", "duckdb", "clickhouse", "mongodb", "highgo", "goldendb", "questdb", "hive", "argo", "kyuubi", "impala", "spark"];
+const TABLE_ONLY_TRANSFER_DATABASES: DatabaseType[] = ["sqlite", "rqlite", "turso", "cloudflare-d1", "duckdb", "clickhouse", "mongodb", "highgo", "goldendb", "questdb", "h2", "hive", "argo", "kyuubi", "impala", "spark"];
 
 describe("transferObjectKinds", () => {
   it("groups databases into transfer families", () => {
diff --git a/crates/dbx-core/assets/database-drivers.manifest.json b/crates/dbx-core/assets/database-drivers.manifest.json
index ae7994b534..8e484d4067 100644
--- a/crates/dbx-core/assets/database-drivers.manifest.json
+++ b/crates/dbx-core/assets/database-drivers.manifest.json
@@ -1692,7 +1692,7 @@
         "tableDataEdit": true,
         "tableStructureEdit": true,
         "tableImport": false,
-        "dataTransfer": false,
+        "dataTransfer": true,
         "sqlFileExecution": true,
         "databaseCreate": false,
         "fieldLineage": false,
diff --git a/crates/dbx-core/src/transfer.rs b/crates/dbx-core/src/transfer.rs
index 27879cd0f3..65cf6d767e 100644
--- a/crates/dbx-core/src/transfer.rs
+++ b/crates/dbx-core/src/transfer.rs
@@ -1293,6 +1293,27 @@ fn rewrite_double_quoted_schema_qualifier(ddl: &str, source_schema: &str, target
     map_sql_code_spans(ddl, false, |code| code.replace(&source, &target))
 }
 
+fn rewrite_h2_schema_qualifier(ddl: &str, source_schema: &str, target_schema: &str) -> String {
+    if source_schema == target_schema || source_schema.is_empty() {
+        return ddl.to_string();
+    }
+    let source = format!("\"{}\"", source_schema.replace('"', "\"\""));
+    let target = format!("\"{}\"", target_schema.replace('"', "\"\""));
+    let identifiers = Regex::new(r#""(?:""|[^"])*""#).expect("valid quoted identifier regex");
+    map_sql_code_spans(ddl, false, |code| {
+        identifiers
+            .replace_all(code, |captures: &regex::Captures| {
+                let identifier = captures.get(0).unwrap();
+                if identifier.as_str() == source && code[identifier.end()..].trim_start().starts_with('.') {
+                    target.clone()
+                } else {
+                    identifier.as_str().to_string()
+                }
+            })
+            .into_owned()
+    })
+}
+
 pub fn rewrite_oracle_schema_qualifier(ddl: &str, source_schema: &str, target_schema: &str) -> String {
     rewrite_double_quoted_schema_qualifier(ddl, source_schema, target_schema)
 }
@@ -1685,6 +1706,34 @@ fn target_column_can_be_omitted(column: &db::ColumnInfo, target_db_type: &Databa
         || (matches!(target_db_type, DatabaseType::SqlServer) && is_sqlserver_rowversion_type(&column.data_type))
 }
 
+fn h2_writable_transfer_columns(
+    source_columns: &[db::ColumnInfo],
+    target_columns: &[db::ColumnInfo],
+    mode: &TransferMode,
+    target_table: &str,
+) -> Result<Vec<db::ColumnInfo>, String> {
+    let mut columns = Vec::new();
+    for source in source_columns {
+        let target = target_columns.iter().find(|target| target.name == source.name);
+        if target.is_some_and(|target| target.extra.as_deref() == Some("computed")) {
+            continue;
+        }
+        if *mode == TransferMode::Upsert
+            && target.is_some_and(|target| is_postgres_generated_always_identity_extra(target.extra.as_deref()))
+        {
+            return Err(format!("H2 Upsert cannot assign GENERATED ALWAYS identity column '{}'. Use append/overwrite or change the target column to GENERATED BY DEFAULT.", source.name));
+        }
+        columns.push(source.clone());
+    }
+    if columns.is_empty() {
+        return Err("No writable H2 target columns found".to_string());
+    }
+    // Data-only overwrite must also fail before TRUNCATE, not after losing target rows.
+    let names = columns.iter().map(|column| column.name.clone()).collect::<Vec<_>>();
+    validate_preexisting_target_columns(target_columns, &names, &DatabaseType::H2, true, target_table)?;
+    Ok(columns)
+}
+
 fn required_unmapped_transfer_target_columns(
     target_columns: &[db::ColumnInfo],
     col_names: &[String],
@@ -1889,9 +1938,33 @@ fn postgres_column_type_sql(
             return rewrite_postgres_schema_qualified_references(trimmed, source_schema, target_schema);
         }
     }
+    if matches!(source_db, DatabaseType::H2) {
+        return map_column_type(&h2_transfer_column_type(column), source_db, target_db);
+    }
     map_column_type(&column.data_type, source_db, target_db)
 }
 
+fn h2_transfer_column_type(column: &db::ColumnInfo) -> String {
+    let data_type = column.data_type.trim();
+    if !data_type.contains('(') {
+        match data_type.to_ascii_lowercase().as_str() {
+            "varchar" | "character varying" | "char" | "character" | "binary varying" | "varbinary" | "binary" => {
+                if let Some(length) = column.character_maximum_length.filter(|length| *length > 0) {
+                    return format!("{data_type}({length})");
+                }
+            }
+            "decimal" | "numeric" => {
+                if let Some(precision) = column.numeric_precision.filter(|precision| *precision > 0) {
+                    let scale = column.numeric_scale.unwrap_or(0);
+                    return format!("{data_type}({precision},{scale})");
+                }
+            }
+            _ => {}
+        }
+    }
+    data_type.to_string()
+}
+
 fn clickhouse_temporal_column_type(
     column: &db::ColumnInfo,
     source_db: &DatabaseType,
@@ -2699,6 +2772,13 @@ pub fn escape_value(val: &serde_json::Value, db_type: &DatabaseType) -> String {
 }
 
 pub fn escape_value_typed(val: &serde_json::Value, db_type: &DatabaseType, column_type: Option<&str>) -> String {
+    if *db_type == DatabaseType::H2
+        && !val.is_null()
+        && column_type.is_some_and(|kind| kind.trim().eq_ignore_ascii_case("json"))
+    {
+        let json = val.as_str().map(str::to_string).unwrap_or_else(|| val.to_string());
+        return format!("JSON {}", quote_string_literal(&json));
+    }
     match val {
         serde_json::Value::Null => "NULL".to_string(),
         serde_json::Value::Bool(b) => match db_type {
@@ -2769,12 +2849,21 @@ pub fn escape_value_typed(val: &serde_json::Value, db_type: &DatabaseType, colum
             if let Some(temporal_literal) = format_oracle_temporal_sql_literal(s, db_type, column_type) {
                 return temporal_literal;
             }
+            if *db_type == DatabaseType::H2 && column_type.is_some_and(is_binary_transfer_column_type) {
+                if let Some(hex) = s.strip_prefix("0x").or_else(|| s.strip_prefix("0X")) {
+                    if hex.len() % 2 == 0 && hex.bytes().all(|byte| byte.is_ascii_hexdigit()) {
+                        return format!("X'{hex}'");
+                    }
+                }
+            }
 
             let literal = format_literal_string(s, db_type, column_type);
             if *db_type == DatabaseType::Postgres {
                 return quote_postgres_string_literal(&literal);
             }
-            let escaped = if is_postgres_family_target(db_type) || *db_type == DatabaseType::SqlServer {
+            let escaped = if is_postgres_family_target(db_type)
+                || matches!(db_type, DatabaseType::SqlServer | DatabaseType::H2)
+            {
                 literal.replace('\'', "''")
             } else {
                 literal.replace('\\', "\\\\").replace('\'', "''")
@@ -2800,6 +2889,9 @@ pub fn escape_value_typed(val: &serde_json::Value, db_type: &DatabaseType, colum
         }
         _ => {
             let s = val.to_string();
+            if *db_type == DatabaseType::H2 {
+                return quote_string_literal(&s);
+            }
             format!("'{}'", s.replace('\\', "\\\\").replace('\'', "''"))
         }
     }
@@ -3287,6 +3379,15 @@ pub fn map_column_type(source_type: &str, source_db: &DatabaseType, target_db: &
     }
     let t = source_type.to_lowercase();
     let mut base = t.split('(').next().unwrap_or(&t).trim();
+    if matches!(source_db, DatabaseType::H2) {
+        base = match base {
+            "character varying" => "varchar",
+            "character large object" => "clob",
+            "binary varying" => "varbinary",
+            "binary large object" => "blob",
+            _ => base,
+        };
+    }
     // Extract basic type, `bigint unsigned` -> `bigint`
     base = base.split(' ').next().unwrap_or(base).trim();
 
@@ -3350,7 +3451,7 @@ pub fn map_column_type(source_type: &str, source_db: &DatabaseType, target_db: &
         "decimal" | "numeric" | "number" => {
             if t.contains('(') {
                 match target_db {
-                    DatabaseType::Mysql | DatabaseType::SqlServer | DatabaseType::Oracle => {
+                    DatabaseType::Mysql | DatabaseType::SqlServer | DatabaseType::Oracle | DatabaseType::H2 => {
                         format!("DECIMAL{}", &t[t.find('(').unwrap()..])
                     }
                     target_db if is_postgres_transfer_dialect(target_db) => {
@@ -3751,11 +3852,12 @@ impl InsertSqlTemplate {
             .map(|column| transfer_column_identifier(column, db_type, quote_target_column_names))
             .collect::<Vec<_>>()
             .join(", ");
-        let overriding = if overrides_postgres_system_values && matches!(db_type, DatabaseType::Postgres) {
-            " OVERRIDING SYSTEM VALUE"
-        } else {
-            ""
-        };
+        let overriding =
+            if overrides_postgres_system_values && matches!(db_type, DatabaseType::Postgres | DatabaseType::H2) {
+                " OVERRIDING SYSTEM VALUE"
+            } else {
+                ""
+            };
         Self {
             standard_prefix: format!("INSERT INTO {full_table} ({col_list}){overriding} VALUES\n"),
             oracle_into_prefix: matches!(db_type, DatabaseType::Oracle)
@@ -3967,6 +4069,14 @@ fn generate_upsert_typed_for_transfer(
             }
             sql
         }
+        DatabaseType::H2 => {
+            let pk_list = pk_columns
+                .iter()
+                .map(|column| transfer_column_identifier(column, db_type, quote_target_column_names))
+                .collect::<Vec<_>>()
+                .join(", ");
+            format!("MERGE INTO {full_table} ({col_list}) KEY ({pk_list}) VALUES\n{}", value_rows.join(",\n"))
+        }
         db_type if uses_mysql_style_upsert(db_type) => {
             let mut sql = format!("INSERT INTO {full_table} ({col_list}) VALUES\n{}", value_rows.join(",\n"));
             if non_pk_columns.is_empty() {
@@ -4254,6 +4364,10 @@ fn rewrite_transfer_source_table_ddl(
         Some(rewrite_postgres_schema_qualified_references(sql, source_schema, target_schema))
     } else if matches!((source_db_type, target_db_type), (DatabaseType::Dameng, DatabaseType::Dameng)) {
         Some(strip_dameng_storage_clauses(&rewrite_double_quoted_schema_qualifier(sql, source_schema, target_schema)))
+    } else if matches!((source_db_type, target_db_type), (DatabaseType::H2, DatabaseType::H2)) {
+        let source_schema = if source_schema.trim().is_empty() { "PUBLIC" } else { source_schema };
+        let target_schema = if target_schema.trim().is_empty() { "PUBLIC" } else { target_schema };
+        Some(rewrite_h2_schema_qualifier(sql, source_schema, target_schema))
     } else if is_mysql_family_target(source_db_type) && is_mysql_family_target(target_db_type) {
         // The reused SHOW CREATE TABLE DDL carries the source table name; rewrite the
         // CREATE TABLE header when the transfer renames the table (name case conversion).
@@ -6099,6 +6213,16 @@ fn effective_transfer_database_type(config: &ConnectionConfig) -> DatabaseType {
         return DatabaseType::Oracle;
     }
 
+    let is_h2 = config.connection_string.as_deref().is_some_and(|url| url.to_ascii_lowercase().starts_with("jdbc:h2:"))
+        || config.jdbc_driver_class.as_deref().is_some_and(|class| class.eq_ignore_ascii_case("org.h2.Driver"))
+        || config.driver_profile.as_deref().is_some_and(|profile| {
+            let profile = profile.to_ascii_lowercase();
+            profile == "h2" || profile.starts_with("h2-")
+        });
+    if is_h2 {
+        return DatabaseType::H2;
+    }
+
     let jdbc_identity = [
         config.driver_profile.as_deref().unwrap_or(""),
         config.connection_string.as_deref().unwrap_or(""),
@@ -9019,8 +9143,8 @@ where
         return Err(format!("No writable columns found for table {table}"));
     }
 
-    let col_names: Vec<String> = writable_columns.iter().map(|c| c.name.clone()).collect();
-    let col_types: Vec<Option<String>> = writable_columns.iter().map(|c| Some(c.data_type.clone())).collect();
+    let mut col_names: Vec<String> = writable_columns.iter().map(|c| c.name.clone()).collect();
+    let mut col_types: Vec<Option<String>> = writable_columns.iter().map(|c| Some(c.data_type.clone())).collect();
     let primary_key_columns = transfer_key_columns(&writable_columns, source_db_type);
     if should_copy_data(&request.content) {
         log::info!("[transfer] {} has {} columns, counting rows...", table, columns.len());
@@ -9148,7 +9272,7 @@ where
                     | DatabaseType::Impala
                     | DatabaseType::Argo
             ))
-        || matches!(target_db_type, DatabaseType::Postgres | DatabaseType::Dameng);
+        || matches!(target_db_type, DatabaseType::Postgres | DatabaseType::Dameng | DatabaseType::H2);
     let target_columns = if needs_target_columns {
         get_columns_for_transfer(
             state,
@@ -9184,6 +9308,18 @@ where
         )?;
     }
 
+    if *target_db_type == DatabaseType::H2 {
+        let columns = h2_writable_transfer_columns(&writable_columns, &target_columns, &request.mode, &target_table)?;
+        col_names = columns.iter().map(|column| column.name.clone()).collect();
+        col_types = columns
+            .iter()
+            .map(|column| {
+                let target = target_columns.iter().find(|target| target.name == column.name).unwrap_or(column);
+                Some(target.data_type.clone())
+            })
+            .collect();
+    }
+
     // Truncate target if overwrite mode (only when not rebuilding the table).
     // When drop_target_before_create is true, the target table was just created
     // and is already empty, so TRUNCATE is unnecessary.
@@ -9229,7 +9365,7 @@ where
 
     let writes_identity_insert_columns = matches!(target_db_type, DatabaseType::Dameng | DatabaseType::SqlServer)
         && selected_columns_include_identity_columns(&col_names, &target_columns);
-    let overrides_postgres_system_values = matches!(target_db_type, DatabaseType::Postgres)
+    let overrides_postgres_system_values = matches!(target_db_type, DatabaseType::Postgres | DatabaseType::H2)
         && selected_columns_include_postgres_generated_always_identity_columns(&col_names, &target_columns);
     // Transfer data in batches
     let batch_size = if request.batch_size == 0 { 1000 } else { request.batch_size };
@@ -10469,6 +10605,190 @@ mod tests {
 
     use serde_json::json;
 
+    #[test]
+    fn h2_jdbc_uses_native_transfer_dialect() {
+        for config in [
+            jdbc_transfer_config("jdbc:h2:mem:transfer", "", ""),
+            jdbc_transfer_config("JDBC:H2:tcp://localhost/test", "", ""),
+            jdbc_transfer_config("", "org.h2.Driver", ""),
+            jdbc_transfer_config("", "", "h2-v3"),
+        ] {
+            assert_eq!(effective_transfer_database_type(&config), DatabaseType::H2);
+        }
+        assert_eq!(
+            effective_transfer_database_type(&jdbc_transfer_config("jdbc:other:test", "", "")),
+            DatabaseType::Jdbc
+        );
+    }
+
+    #[test]
+    fn h2_transfer_preserves_character_and_numeric_metadata() {
+        let columns = vec![
+            db::ColumnInfo { character_maximum_length: Some(1024), ..test_column("NAME", "CHARACTER VARYING") },
+            db::ColumnInfo { numeric_precision: Some(12), numeric_scale: Some(2), ..test_column("AMOUNT", "NUMERIC") },
+        ];
+        for target in [DatabaseType::Mysql, DatabaseType::Postgres, DatabaseType::H2] {
+            let ddl = generate_create_table_ddl(
+                &columns,
+                "ITEMS",
+                "PUBLIC",
+                "target",
+                &target,
+                &DatabaseType::H2,
+                None,
+                None,
+            );
+            assert!(ddl.contains("(1024)"), "{ddl}");
+            assert!(ddl.contains("(12,2)"), "{ddl}");
+            assert!(!ddl.contains("CHAR(1)"), "{ddl}");
+        }
+        assert_eq!(
+            map_column_type("CHARACTER VARYING(1024)", &DatabaseType::H2, &DatabaseType::Mysql),
+            "VARCHAR(1024)"
+        );
+        assert_eq!(map_column_type("CHARACTER LARGE OBJECT", &DatabaseType::H2, &DatabaseType::Postgres), "TEXT");
+        assert_eq!(map_column_type("BINARY LARGE OBJECT", &DatabaseType::H2, &DatabaseType::Postgres), "BYTEA");
+        assert_eq!(map_column_type("BINARY VARYING(64)", &DatabaseType::H2, &DatabaseType::Mysql), "BLOB");
+        assert_eq!(map_column_type("DECIMAL(12,2)", &DatabaseType::Mysql, &DatabaseType::H2), "DECIMAL(12,2)");
+    }
+
+    #[test]
+    fn h2_transfer_rewrites_schema_without_changing_literals_or_identifiers() {
+        let sql = r#"CREATE TABLE "Source"."ITEMS" ("NAME" VARCHAR(1024) DEFAULT '"Source".ITEMS');
+CREATE INDEX "IDX" ON "Source"."ITEMS" ("NAME");
+-- "Source".ITEMS stays in the comment
+/* "Source".ITEMS */
+COMMENT ON TABLE "Source"."ITEMS" IS '"Source".ITEMS';
+CREATE TABLE "Other"."prefix""Source"."NAME" ("ID" INT);"#;
+        let rewritten = rewrite_transfer_source_table_ddl(
+            sql,
+            "Source",
+            "Target",
+            &DatabaseType::H2,
+            &DatabaseType::H2,
+            "ITEMS",
+            "ITEMS",
+        )
+        .unwrap();
+        assert!(rewritten.contains(r#"CREATE TABLE "Target"."ITEMS""#));
+        assert!(rewritten.contains(r#"ON "Target"."ITEMS""#));
+        assert!(rewritten.contains(r#"DEFAULT '"Source".ITEMS'"#));
+        assert!(rewritten.contains(r#"-- "Source".ITEMS"#));
+        assert!(rewritten.contains(r#"/* "Source".ITEMS */"#));
+        assert!(rewritten.contains(r#""Other"."prefix""Source"."NAME""#));
+        assert_eq!(rewrite_h2_schema_qualifier(r#""A""B" . "T""#, "A\"B", "C\"D"), r#""C""D" . "T""#);
+        assert!(can_reuse_source_table_ddl(&DatabaseType::H2, &DatabaseType::H2, None, None, true));
+        assert!(!can_reuse_source_table_ddl(&DatabaseType::H2, &DatabaseType::H2, None, None, false));
+    }
+
+    #[test]
+    fn h2_upsert_merges_by_composite_primary_key() {
+        let columns = vec!["ID".into(), "PART".into(), "NAME".into()];
+        let rows = vec![vec![json!(1), json!(1), json!("o'clock")], vec![json!(1), json!(2), json!(null)]];
+        assert_eq!(
+            generate_upsert(&columns, &rows, "ITEMS", "TransferTarget", &DatabaseType::H2, &["ID".into(), "PART".into()]),
+            "MERGE INTO \"TransferTarget\".\"ITEMS\" (\"ID\", \"PART\", \"NAME\") KEY (\"ID\", \"PART\") VALUES\n(1, 1, 'o''clock'),\n(1, 2, NULL)"
+        );
+        assert_eq!(
+            generate_upsert(&["ID".into()], &[vec![json!(1)]], "ITEMS", "PUBLIC", &DatabaseType::H2, &["ID".into()]),
+            "MERGE INTO \"PUBLIC\".\"ITEMS\" (\"ID\") KEY (\"ID\") VALUES\n(1)"
+        );
+    }
+
+    #[test]
+    fn h2_transfer_skips_target_computed_columns_and_guards_always_identity_upsert() {
+        let source = vec![test_column("ID", "INTEGER"), test_column("QTY", "INTEGER"), test_column("TOTAL", "INTEGER")];
+        let mut target = source.clone();
+        target[0].extra = Some("generated always as identity".into());
+        target[2].extra = Some("computed".into());
+        for mode in [TransferMode::Append, TransferMode::Overwrite] {
+            assert_eq!(
+                h2_writable_transfer_columns(&source, &target, &mode, "T")
+                    .unwrap()
+                    .iter()
+                    .map(|column| column.name.as_str())
+                    .collect::<Vec<_>>(),
+                vec!["ID", "QTY"]
+            );
+        }
+        assert!(h2_writable_transfer_columns(&source, &target, &TransferMode::Upsert, "T")
+            .unwrap_err()
+            .contains("GENERATED ALWAYS"));
+        target[0].extra = Some("generated by default as identity".into());
+        assert!(h2_writable_transfer_columns(&source, &target, &TransferMode::Upsert, "T").is_ok());
+        // A generated source value can still be copied into an ordinary target column.
+        assert_eq!(h2_writable_transfer_columns(&target, &source, &TransferMode::Append, "T").unwrap().len(), 3);
+        assert_eq!(
+            InsertSqlTemplate::new(&["ID".into()], "T", "PUBLIC", &DatabaseType::H2, None, true)
+                .build(&["(42)".into()]),
+            "INSERT INTO \"PUBLIC\".\"T\" (\"ID\") OVERRIDING SYSTEM VALUE VALUES\n(42)"
+        );
+    }
+
+    #[test]
+    fn h2_transfer_defaults_empty_schemas_to_public() {
+        assert_eq!(
+            rewrite_transfer_source_table_ddl(
+                r#"CREATE TABLE "PUBLIC"."T" ("ID" INT)"#,
+                "",
+                "Target",
+                &DatabaseType::H2,
+                &DatabaseType::H2,
+                "T",
+                "T"
+            )
+            .unwrap(),
+            r#"CREATE TABLE "Target"."T" ("ID" INT)"#
+        );
+        assert_eq!(
+            rewrite_transfer_source_table_ddl(
+                r#"CREATE TABLE "Source"."T" ("ID" INT)"#,
+                "Source",
+                "",
+                &DatabaseType::H2,
+                &DatabaseType::H2,
+                "T",
+                "T"
+            )
+            .unwrap(),
+            r#"CREATE TABLE "PUBLIC"."T" ("ID" INT)"#
+        );
+    }
+
+    #[test]
+    fn h2_write_preflight_rejects_missing_and_required_target_columns() {
+        let source = vec![test_column("ID", "INTEGER"), test_column("NAME", "VARCHAR(32)")];
+        for mode in [TransferMode::Append, TransferMode::Overwrite, TransferMode::Upsert] {
+            let error = h2_writable_transfer_columns(&source, &source[..1], &mode, "ITEMS").unwrap_err();
+            assert!(error.contains("NAME"), "{error}");
+            assert!(error.contains("ITEMS"), "{error}");
+            let mut target = source.clone();
+            target.push(db::ColumnInfo { is_nullable: false, ..test_column("REQUIRED", "INTEGER") });
+            let error = h2_writable_transfer_columns(&source, &target, &mode, "ITEMS").unwrap_err();
+            assert!(error.contains("REQUIRED"), "{error}");
+            target[2].column_default = Some("1".into());
+            assert!(h2_writable_transfer_columns(&source, &target, &mode, "ITEMS").is_ok());
+        }
+    }
+
+    #[test]
+    fn h2_transfer_literals_preserve_json_binary_and_backslashes() {
+        assert_eq!(
+            escape_value_typed(&json!(r"C:\tmp\o'clock"), &DatabaseType::H2, Some("CHARACTER VARYING")),
+            r"'C:\tmp\o''clock'"
+        );
+        assert_eq!(escape_value_typed(&json!("0x00ff"), &DatabaseType::H2, Some("BINARY VARYING")), "X'00ff'");
+        assert_eq!(escape_value_typed(&json!("0x"), &DatabaseType::H2, Some("BLOB")), "X''");
+        assert_eq!(escape_value_typed(&json!("0x00ff"), &DatabaseType::H2, Some("VARCHAR")), "'0x00ff'");
+        assert_eq!(
+            escape_value_typed(&json!(r#"{"name":"o'clock"}"#), &DatabaseType::H2, Some("JSON")),
+            r#"JSON '{"name":"o''clock"}'"#
+        );
+        assert_eq!(escape_value_typed(&json!([1, 2]), &DatabaseType::H2, Some("JSON")), "JSON '[1,2]'");
+        assert_eq!(escape_value_typed(&json!(null), &DatabaseType::H2, Some("JSON")), "NULL");
+        assert_eq!(escape_value_typed(&json!("null"), &DatabaseType::H2, Some("JSON")), "JSON 'null'");
+    }
+
     fn test_column(name: &str, data_type: &str) -> db::ColumnInfo {
         db::ColumnInfo {
             name: name.to_string(),
diff --git a/crates/dbx-core/src/transfer/ddl_plan.rs b/crates/dbx-core/src/transfer/ddl_plan.rs
index 9d5c217647..8ea74ef67c 100644
--- a/crates/dbx-core/src/transfer/ddl_plan.rs
+++ b/crates/dbx-core/src/transfer/ddl_plan.rs
@@ -145,10 +145,12 @@ pub(super) async fn prepare_table_ddl(
             .await
             {
                 Ok(ddl) => (ddl, true),
-                Err(e) if rebuild => {
-                    // A rebuild cannot fall back to generated DDL when the source DDL
-                    // it promised to reproduce cannot be read.
-                    return Err(format!("Failed to read source DDL for table '{table}' before rebuilding: {e}"));
+                Err(e)
+                    if rebuild || matches!((source_db_type, target_db_type), (DatabaseType::H2, DatabaseType::H2)) =>
+                {
+                    // Rebuilds and H2-to-H2 transfers must not silently discard source
+                    // constraints or generated columns when native DDL cannot be read.
+                    return Err(format!("Failed to read source DDL for table '{table}' before creating target: {e}"));
                 }
                 Err(_) => (
                     generate_create_table_ddl_with_column_quoting(
@@ -375,6 +377,26 @@ mod tests {
         assert!(prepared.ddl.contains("CREATE TABLE"));
     }
 
+    #[tokio::test]
+    async fn h2_native_ddl_read_failure_does_not_fall_back_to_lossy_metadata() {
+        let (state, _directory) = state_fixture().await;
+        let error = prepare_table_ddl(
+            &state,
+            &request(false),
+            "child",
+            "child",
+            &DatabaseType::H2,
+            &DatabaseType::H2,
+            "missing-source-pool",
+            &columns(),
+            None,
+            &HashMap::new(),
+        )
+        .await
+        .unwrap_err();
+        assert!(error.contains("Failed to read source DDL"), "{error}");
+    }
+
     #[tokio::test]
     async fn deferred_foreign_key_uses_actual_target_and_case_converted_parent() {
         let (state, _directory) = state_fixture().await;
diff --git a/docs/public/screenshots/h2-transfer-target.png b/docs/public/screenshots/h2-transfer-target.png
new file mode 100644
index 0000000000..159a7895d2
Binary files /dev/null and b/docs/public/screenshots/h2-transfer-target.png differ
diff --git a/plugins/connection-types/h2.yaml b/plugins/connection-types/h2.yaml
index 449e1b6cfe..9361a96d54 100644
--- a/plugins/connection-types/h2.yaml
+++ b/plugins/connection-types/h2.yaml
@@ -50,7 +50,7 @@ capabilities:
   tableDataEdit: true
   tableStructureEdit: true
   tableImport: false
-  dataTransfer: false
+  dataTransfer: true
   sqlFileExecution: true
   databaseCreate: false
   fieldLineage: false