fix(transfer): enable IRIS and Cache source and target data transfers

Closes #10660
2026-10-21 03:21:53 +00:00 · 2026-10-02 23:00:28 +00:00
parent ef9bbe2405
commit 9338e69fb7
8 changed files with 422 additions and 33 deletions
@@ -5,6 +5,7 @@ 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 { supportsTransferUpsert } from "../transferStrategy";
 import type { ConnectionConfig, SidebarLayout } from "@/types/database";

 vi.mock("@/components/ui/button", async () => {
@@ -29,6 +30,8 @@ const connections: ConnectionConfig[] = [
  { 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: "yashandb-id", name: "YashanDB", db_type: "yashandb", host: "localhost", port: 1688, username: "test", password: "" },
+  { id: "iris-id", name: "InterSystems IRIS", db_type: "iris", host: "localhost", port: 1972, username: "test", password: "" },
+  { id: "cache-id", name: "InterSystems Caché", db_type: "iris", driver_profile: "cache", host: "localhost", port: 1972, username: "test", 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: "" },
 ];
@@ -78,6 +81,22 @@ afterEach(() => {
 });

 describe("data transfer connection choices", () => {
+  it.each(["iris-id", "cache-id"])("offers %s as source and target without upsert", async (id) => {
+    const { container, selection } = await mountTransferPickers("mysql-source");
+    const connection = connections.find((connection) => connection.id === id)!;
+    const databaseType = transferDatabaseTypeForConnection(connection);
+    expect(databaseType).toBe("iris");
+    expect(supportsTransfer(databaseType)).toBe(true);
+    expect(supportsTransferUpsert(databaseType)).toBe(false);
+    for (const side of ["source", "target"] as const) {
+      const option = container.querySelector(`[data-side="${side}"] [data-picker-connection="${id}"]`) as HTMLButtonElement;
+      expect(option?.textContent).toContain(connection.name);
+      option.click();
+      await nextTick();
+      expect(selection[side]).toBe(id);
+    }
+  });
+
  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"]) {
@@ -96,7 +115,7 @@ describe("data transfer connection choices", () => {
    expect(selection.source).toBe(source);
  });

-  it.each(["h2-local-id", "h2-jdbc-id", "yashandb-id"])("resolves the prefilled %s to its connection name", async (source) => {
+  it.each(["h2-local-id", "h2-jdbc-id", "yashandb-id", "iris-id", "cache-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);
@@ -16,7 +16,7 @@ export function transferStrategyOptions(strategy: TransferStrategy): Pick<Transf
 const REBUILD_TARGET_TYPES = new Set<DatabaseType>(["mysql", "postgres", "sqlserver", "kingbase", "gaussdb", "opengauss", "kwdb", "goldendb", "sqlite", "duckdb", "cloudflare-d1"]);

 export function supportsTransferUpsert(targetType: DatabaseType | undefined): boolean {
-  return targetType !== "db2";
+  return targetType !== "db2" && targetType !== "iris";
 }

 export function rebuildUnavailableReason(content: TransferContent, targetType: DatabaseType | undefined): "dataOnly" | "unsupported" | undefined {
@@ -2803,7 +2803,7 @@
        "tableDataEdit": false,
        "tableStructureEdit": true,
        "tableImport": false,
-        "dataTransfer": false,
+        "dataTransfer": true,
        "sqlFileExecution": true,
        "databaseCreate": false,
        "fieldLineage": false,
@@ -17,6 +17,10 @@ use futures::{SinkExt, StreamExt};
 #[path = "transfer/rebuild_tests.rs"]
 mod rebuild_tests;

+#[cfg(test)]
+#[path = "transfer/iris_tests.rs"]
+mod iris_tests;
+
 mod db2;
 mod ddl_plan;

@@ -33,7 +37,8 @@ use crate::query::{
 };
 use crate::sql::{split_sql_statements, split_sql_statements_for_database};
 use crate::sql_dialect::{
-    normalize_len_params, qualified_transfer_table, quote_transfer_identifier, transfer_column_identifier,
+    build_iris_table_select_sql, normalize_len_params, qualified_transfer_table, quote_transfer_identifier,
+    transfer_column_identifier,
 };

 static CANCELLED: std::sync::LazyLock<RwLock<HashSet<String>>> =
@@ -123,6 +128,7 @@ impl SqlBatchLimits {
        let max_rows = requested_max_rows.max(1).min(match db_type {
            DatabaseType::SqlServer => MAX_SQLSERVER_INSERT_ROWS,
            DatabaseType::Oracle | DatabaseType::OceanbaseOracle => MAX_ORACLE_INSERT_ALL_ROWS,
+            DatabaseType::Iris => 1,
            DatabaseType::Transwarp => 100,
            _ => usize::MAX,
        });
@@ -4479,6 +4485,7 @@ fn generate_upsert_typed_for_transfer(

 fn max_transfer_write_rows(db_type: &DatabaseType, mode: &TransferMode) -> usize {
    match (db_type, mode) {
+        (DatabaseType::Iris, _) => 1,
        (DatabaseType::SqlServer, TransferMode::Append | TransferMode::Overwrite) => MAX_SQLSERVER_INSERT_ROWS,
        (
            DatabaseType::Hive
@@ -4913,6 +4920,7 @@ pub(crate) fn generate_insert_typed_sql_batches_from_value_rows_with_options(
    let max_rows = limits.max_rows.max(1).min(match db_type {
        DatabaseType::SqlServer => MAX_SQLSERVER_INSERT_ROWS,
        DatabaseType::Oracle | DatabaseType::OceanbaseOracle => MAX_ORACLE_INSERT_ALL_ROWS,
+        DatabaseType::Iris => 1,
        _ => usize::MAX,
    });
    let target_sql_bytes = limits.target_sql_bytes.max(1);
@@ -5094,6 +5102,7 @@ fn generate_insert_sql_batches_from_value_rows(
    let max_rows = limits.max_rows.max(1).min(match db_type {
        DatabaseType::SqlServer => MAX_SQLSERVER_INSERT_ROWS,
        DatabaseType::Oracle | DatabaseType::OceanbaseOracle => MAX_ORACLE_INSERT_ALL_ROWS,
+        DatabaseType::Iris => 1,
        _ => usize::MAX,
    });
    let target_sql_bytes = limits.target_sql_bytes.max(1);
@@ -5280,6 +5289,7 @@ pub fn pagination_sql(
    let col_list = columns.iter().map(|c| quote_identifier(c, db_type)).collect::<Vec<_>>().join(", ");

    match db_type {
+        DatabaseType::Iris => build_iris_table_select_sql(&col_list, &full_table, "", " ORDER BY %ID", limit, offset),
        DatabaseType::Oracle | DatabaseType::OceanbaseOracle => {
            let base_sql = format!("SELECT {col_list} FROM {full_table}");
            oracle_rownum_page_sql(&col_list, base_sql, offset, limit)
@@ -5324,6 +5334,10 @@ pub fn pagination_sql_with_order(
    let order_expression = postgres_order_by_expression(order_by_columns, db_type);

    match db_type {
+        DatabaseType::Iris => {
+            let order_by = format!(" ORDER BY {}", order_expression.as_deref().unwrap_or("%ID"));
+            build_iris_table_select_sql(&col_list, &full_table, "", &order_by, limit, offset)
+        }
        DatabaseType::Oracle | DatabaseType::OceanbaseOracle => {
            let order_by = order_expression.map(|value| format!(" ORDER BY {value}")).unwrap_or_default();
            let base_sql = format!("SELECT {col_list} FROM {full_table}{order_by}");
@@ -5412,6 +5426,10 @@ pub fn pagination_sql_with_filter_order_and_identifier_quote(
        });

    match db_type {
+        DatabaseType::Iris => {
+            let order_by = format!(" ORDER BY {}", order_expression.as_deref().unwrap_or("%ID"));
+            build_iris_table_select_sql(&col_list, &full_table, &where_clause, &order_by, limit, offset)
+        }
        DatabaseType::Oracle | DatabaseType::OceanbaseOracle => {
            let order_by = order_expression.map(|value| format!(" ORDER BY {value}")).unwrap_or_default();
            let base_sql = format!("SELECT {col_list} FROM {full_table}{where_clause}{order_by}");
@@ -9229,6 +9247,19 @@ fn uses_agent_transfer_cursor(db_type: &DatabaseType) -> bool {
    matches!(db_type, DatabaseType::Kyuubi | DatabaseType::Impala | DatabaseType::Transwarp | DatabaseType::Db2)
 }

+fn transfer_upsert_falls_back_to_append(db_type: &DatabaseType) -> bool {
+    matches!(
+        db_type,
+        DatabaseType::ClickHouse
+            | DatabaseType::Hive
+            | DatabaseType::Kyuubi
+            | DatabaseType::Impala
+            | DatabaseType::Argo
+            | DatabaseType::Transwarp
+            | DatabaseType::Iris
+    )
+}
+
 fn transfer_clear_table_sql(table: &str, schema: &str, db_type: &DatabaseType, catalog: Option<&str>) -> String {
    let full_table = qualified_table(table, schema, db_type, catalog);
    match db_type {
@@ -10146,16 +10177,7 @@ where
    // identity values with 544.
    let needs_target_columns = default_rows_only
        || target_table_preexisting
-        || (request.mode == TransferMode::Upsert
-            && !matches!(
-                target_db_type,
-                DatabaseType::ClickHouse
-                    | DatabaseType::Hive
-                    | DatabaseType::Kyuubi
-                    | DatabaseType::Impala
-                    | DatabaseType::Argo
-                    | DatabaseType::Transwarp
-            ))
+        || (request.mode == TransferMode::Upsert && !transfer_upsert_falls_back_to_append(target_db_type))
        || matches!(
            target_db_type,
            DatabaseType::Postgres | DatabaseType::Dameng | DatabaseType::H2 | DatabaseType::SqlServer
@@ -10268,15 +10290,7 @@ where

    // Determine effective mode and PK columns for upsert
    let (effective_mode, pk_columns) = if request.mode == TransferMode::Upsert {
-        if matches!(
-            target_db_type,
-            DatabaseType::ClickHouse
-                | DatabaseType::Hive
-                | DatabaseType::Kyuubi
-                | DatabaseType::Impala
-                | DatabaseType::Argo
-                | DatabaseType::Transwarp
-        ) {
+        if transfer_upsert_falls_back_to_append(target_db_type) {
            log::warn!("[transfer] upsert not supported for {:?}, falling back to append", target_db_type);
            (TransferMode::Append, vec![])
        } else {
@@ -0,0 +1,357 @@
+use super::*;
+use serde_json::json;
+
+#[test]
+fn iris_transfer_pages_advance_with_top_and_vid() {
+    let columns = vec!["id".into(), "name".into()];
+    assert_eq!(
+        pagination_sql(&columns, "items", "SQLUSER", &DatabaseType::Iris, 0, 2),
+        "SELECT TOP 2 \"id\", \"name\" FROM \"SQLUSER\".\"items\" ORDER BY %ID"
+    );
+    for offset in [2, 4] {
+        assert_eq!(
+            pagination_sql_with_order(
+                &columns, "items", "SQLUSER", &DatabaseType::Iris, offset, 2, &["id".into()], None,
+            ),
+            format!("SELECT * FROM (SELECT TOP {} \"id\", \"name\" FROM \"SQLUSER\".\"items\" ORDER BY \"id\") WHERE %VID > {offset}", offset + 2)
+        );
+    }
+    assert_eq!(
+        pagination_sql_with_order(&columns, "items", "SQLUSER", &DatabaseType::Iris, 2, 2, &[], None),
+        "SELECT * FROM (SELECT TOP 4 \"id\", \"name\" FROM \"SQLUSER\".\"items\" ORDER BY %ID) WHERE %VID > 2"
+    );
+}
+
+#[test]
+fn iris_transfer_filtered_pages_preserve_predicates_order_and_quoting() {
+    assert_eq!(
+        pagination_sql_with_filter_order(
+            &["item\"id".into()], "item\"s", "team space", &DatabaseType::Iris,
+            2, 2, Some("WHERE active = 1"), Some("\"item\"\"id\" DESC"), &[],
+        ),
+        "SELECT * FROM (SELECT TOP 4 \"item\"\"id\" FROM \"team space\".\"item\"\"s\" WHERE (active = 1) ORDER BY \"item\"\"id\" DESC) WHERE %VID > 2"
+    );
+}
+
+#[test]
+fn iris_transfer_append_and_overwrite_use_single_row_inserts() {
+    for mode in [TransferMode::Append, TransferMode::Overwrite] {
+        let statements = generate_transfer_write_sql_batches(
+            &mode,
+            &["id".into(), "name".into()],
+            &[Some("INTEGER".into()), Some("VARCHAR".into())],
+            &[vec![json!(1), json!("O'Hara")], vec![json!(2), json!(null)]],
+            "item\"s",
+            "SQLUSER",
+            &DatabaseType::Iris,
+            &[],
+            None,
+            false,
+            false,
+        )
+        .unwrap();
+        assert_eq!(
+            statements,
+            vec![
+                "INSERT INTO \"SQLUSER\".\"item\"\"s\" (\"id\", \"name\") VALUES\n(1, 'O''Hara')",
+                "INSERT INTO \"SQLUSER\".\"item\"\"s\" (\"id\", \"name\") VALUES\n(2, NULL)",
+            ]
+        );
+        assert!(generate_transfer_write_sql_batches(
+            &mode,
+            &["id".into()],
+            &[],
+            &[],
+            "items",
+            "SQLUSER",
+            &DatabaseType::Iris,
+            &[],
+            None,
+            false,
+            false,
+        )
+        .unwrap()
+        .is_empty());
+    }
+}
+
+#[test]
+fn iris_transfer_limits_and_unsupported_upsert_keep_existing_policies() {
+    assert!(transfer_upsert_falls_back_to_append(&DatabaseType::Iris));
+    assert!(!supports_primary_key_upsert(&DatabaseType::Iris));
+    assert!(transfer_upsert_falls_back_to_append(&DatabaseType::Hive));
+    assert!(!transfer_upsert_falls_back_to_append(&DatabaseType::Postgres));
+    for requested in [0, 1, 1000] {
+        assert_eq!(SqlBatchLimits::for_database(&DatabaseType::Iris, requested).max_rows, 1);
+    }
+    let rows = ["(1)".into(), "(2)".into()];
+    // Even callers supplying their own limits must obey the dialect's row limit.
+    let limits = SqlBatchLimits { max_rows: 1000, target_sql_bytes: 1024, hard_sql_bytes: None };
+    for (db_type, expected_batches) in [(DatabaseType::Iris, 2), (DatabaseType::Postgres, 1), (DatabaseType::Mysql, 1)]
+    {
+        let batches = generate_insert_sql_batches_from_value_rows(
+            &["id".into()],
+            &rows,
+            "items",
+            "",
+            &db_type,
+            None,
+            limits,
+            false,
+            true,
+        )
+        .unwrap();
+        assert_eq!(batches.len(), expected_batches);
+    }
+    assert!(generate_insert_sql_batches_from_value_rows(
+        &["id".into()],
+        &rows,
+        "items",
+        "",
+        &DatabaseType::Iris,
+        None,
+        limits.with_hard_sql_bytes(Some(1)),
+        false,
+        true,
+    )
+    .unwrap_err()
+    .contains("exceeds the 1 byte hard limit"));
+}
+
+const IRIS_TRANSFER_AGENT: &str = r#"
+import json, pathlib, re, sys
+trace = pathlib.Path(sys.argv[1])
+row_count = int(sys.argv[2])
+failure = sys.argv[3]
+has_key = sys.argv[4] == 'true'
+
+def query(rows=None, columns=None):
+    return {'columns': columns or [], 'column_types': [], 'column_sortables': [],
+            'rows': rows or [], 'affected_rows': 1, 'execution_time_ms': 0,
+            'truncated': False, 'session_id': None, 'has_more': False}
+
+print(json.dumps({'ready': True}), flush=True)
+for line in sys.stdin:
+    req = json.loads(line)
+    method = req['method']
+    params = req.get('params', {})
+    sql = params.get('sql', '')
+    with trace.open('a', encoding='utf-8') as output:
+        output.write(json.dumps({'method': method, 'sql': sql, 'params': params}) + '\n')
+    try:
+        if method == 'handshake':
+            result = {'protocolVersion': 2, 'agentProtocolVersion': 2,
+                      'capabilities': ['multi_session', 'query', 'metadata']}
+        elif method == 'list_tables':
+            result = [{'name': 'items', 'table_type': 'TABLE'}]
+        elif method == 'get_columns':
+            result = [{'name': name, 'data_type': kind, 'is_nullable': False,
+                       'is_primary_key': has_key and name == 'id'}
+                      for name, kind in [('id', 'INTEGER'), ('name', 'VARCHAR')]]
+        elif method == 'execute_query':
+            if sql.startswith('SELECT COUNT('):
+                if failure == 'count':
+                    raise ValueError('injected count failure')
+                result = query([[row_count]], ['count'])
+            elif sql.startswith('SELECT'):
+                top = re.search(r'SELECT TOP (\d+)', sql)
+                if not top or ' LIMIT ' in sql or ' OFFSET ' in sql:
+                    raise ValueError('expected TOP/VID pagination: ' + sql)
+                expected_order = 'ORDER BY "id"' if has_key else 'ORDER BY %ID'
+                if expected_order not in sql:
+                    raise ValueError('expected stable table order: ' + sql)
+                offset = re.search(r'%VID > (\d+)', sql)
+                offset = int(offset[1]) if offset else 0
+                if failure == 'source' and offset > 0:
+                    raise ValueError('injected source failure')
+                rows = [[n, 'row-' + str(n)] for n in range(1, row_count + 1)]
+                result = query(rows[offset:int(top[1])], ['id', 'name'])
+            elif sql.startswith('INSERT INTO'):
+                if '),\n(' in sql or 'MERGE' in sql or 'ON CONFLICT' in sql:
+                    raise ValueError('expected single-row INSERT: ' + sql)
+                if failure == 'target' and 'VALUES\n(3,' in sql:
+                    raise ValueError('injected target failure')
+                result = query()
+            elif sql.startswith('TRUNCATE TABLE'):
+                result = query()
+            else:
+                raise ValueError('unexpected transfer SQL: ' + sql)
+        else:
+            result = {'ok': True}
+        response = {'jsonrpc': '2.0', 'id': req['id'], 'result': result}
+    except Exception as error:
+        response = {'jsonrpc': '2.0', 'id': req['id'], 'error': {'code': -1, 'message': str(error)}}
+    print(json.dumps(response), flush=True)
+"#;
+
+async fn iris_transfer_fixture(
+    profile: Option<&str>,
+    rows: usize,
+    failure: &str,
+    has_key: bool,
+) -> (tempfile::TempDir, Arc<AppState>, TransferRequest, String, String) {
+    let directory = tempfile::tempdir().unwrap();
+    let storage = crate::persistence::test_storage::open(&directory.path().join("storage.db")).await.unwrap();
+    let state = Arc::new(AppState::new_with_plugin_and_agent_dir_and_app_version(
+        storage,
+        directory.path().join("plugins"),
+        directory.path().join("agents"),
+        "0.0.0-test",
+    ));
+    let script = directory.path().join("agent.py");
+    std::fs::write(&script, IRIS_TRANSFER_AGENT).unwrap();
+    let launch = state.agent_manager.driver_launch_config_path(profile.unwrap_or("iris"));
+    std::fs::create_dir_all(launch.parent().unwrap()).unwrap();
+    std::fs::write(
+        launch,
+        serde_json::to_vec(&json!({
+            "command": if cfg!(windows) { "python" } else { "python3" },
+            "args": [script, directory.path().join("trace.jsonl"), rows.to_string(), failure, has_key.to_string()],
+        }))
+        .unwrap(),
+    )
+    .unwrap();
+    for id in ["source", "target"] {
+        let config: ConnectionConfig = serde_json::from_value(json!({
+            "id": id, "name": id, "db_type": "iris", "driver_profile": profile,
+            "host": "127.0.0.1", "port": 1972, "username": "test", "password": "",
+            "database": "USER", "one_time": false, "save_password": false,
+            "keepalive_interval_secs": 0,
+        }))
+        .unwrap();
+        assert_eq!(effective_transfer_database_type(&config), DatabaseType::Iris);
+        state.configs.write().await.insert(id.to_string(), config);
+    }
+    let source = ensure_transfer_pool(&state, "source", "USER", None).await.unwrap();
+    let target = ensure_transfer_pool(&state, "target", "USER", None).await.unwrap();
+    let request = serde_json::from_value(json!({
+        "transferId": uuid::Uuid::new_v4().to_string(),
+        "sourceConnectionId": "source", "sourceDatabase": "USER", "sourceSchema": "SOURCE",
+        "targetConnectionId": "target", "targetDatabase": "USER", "targetSchema": "TARGET",
+        "tables": ["items"], "createTable": false, "content": "dataOnly", "mode": "append", "batchSize": 2,
+    }))
+    .unwrap();
+    (directory, state, request, source, target)
+}
+
+fn iris_transfer_trace(directory: &std::path::Path) -> Vec<serde_json::Value> {
+    std::fs::read_to_string(directory.join("trace.jsonl"))
+        .unwrap()
+        .lines()
+        .map(|line| serde_json::from_str(line).unwrap())
+        .collect()
+}
+
+#[tokio::test]
+async fn iris_transfer_agent_covers_profiles_pages_empty_tables_and_modes() {
+    for profile in [None, Some("cache")] {
+        for (rows, mode, has_key) in [
+            (0, TransferMode::Append, true),
+            (4, TransferMode::Append, true),
+            (5, TransferMode::Overwrite, false),
+            (5, TransferMode::Upsert, true),
+        ] {
+            let (directory, state, mut request, source, target) =
+                iris_transfer_fixture(profile, rows, "", has_key).await;
+            request.mode = mode.clone();
+            let mut progress = Vec::new();
+            let result = transfer_table(
+                &state,
+                &request,
+                "items",
+                0,
+                &DatabaseType::Iris,
+                &DatabaseType::Iris,
+                &source,
+                &target,
+                &HashMap::new(),
+                &mut Vec::new(),
+                None,
+                |value| progress.push(value.rows_transferred),
+            )
+            .await;
+            state.shutdown(std::time::Duration::from_secs(1)).await;
+            assert_eq!(result.unwrap(), rows as u64, "profile={profile:?}, mode={mode:?}");
+            let trace = iris_transfer_trace(directory.path());
+            let sql: Vec<&str> = trace.iter().filter_map(|entry| entry["sql"].as_str()).collect();
+            let inserts: Vec<&&str> = sql.iter().filter(|sql| sql.starts_with("INSERT INTO")).collect();
+            assert_eq!(inserts.len(), rows);
+            for (index, statement) in inserts.iter().enumerate() {
+                assert_eq!(
+                    **statement,
+                    format!(
+                        "INSERT INTO \"TARGET\".\"items\" (\"id\", \"name\") VALUES\n({}, 'row-{}')",
+                        index + 1,
+                        index + 1,
+                    )
+                );
+            }
+            let pages: Vec<&&str> = sql.iter().filter(|sql| sql.contains("SELECT TOP")).collect();
+            assert_eq!(pages.len(), rows / 2 + 1);
+            for (page, statement) in pages.iter().enumerate().skip(1) {
+                assert!(statement.ends_with(&format!("WHERE %VID > {}", page * 2)));
+            }
+            assert_eq!(
+                sql.iter().filter(|sql| sql.starts_with("TRUNCATE TABLE")).count(),
+                usize::from(mode == TransferMode::Overwrite)
+            );
+            if mode == TransferMode::Overwrite {
+                let clear = sql.iter().position(|sql| sql.starts_with("TRUNCATE TABLE")).unwrap();
+                let write = sql.iter().position(|sql| sql.starts_with("INSERT INTO")).unwrap();
+                assert!(clear < write);
+            }
+            assert_eq!(progress.last().copied().unwrap_or(0), rows as u64);
+        }
+    }
+}
+
+#[tokio::test]
+async fn iris_transfer_agent_preserves_errors_count_fallback_and_cancellation() {
+    for failure in ["source", "target", "count", "cancel"] {
+        let (directory, state, request, source, target) = iris_transfer_fixture(None, 5, failure, true).await;
+        let mut totals = Vec::new();
+        let result = Box::pin(transfer_table_inner(
+            &state,
+            &request,
+            "items",
+            0,
+            &DatabaseType::Iris,
+            &DatabaseType::Iris,
+            &source,
+            &target,
+            &HashMap::new(),
+            &mut Vec::new(),
+            None,
+            |progress| {
+                totals.push(progress.total_rows);
+                if failure == "cancel" {
+                    CANCELLED.try_write().unwrap().insert(request.transfer_id.clone());
+                }
+            },
+        ))
+        .await;
+        clear_cancelled(&request.transfer_id).await;
+        state.shutdown(std::time::Duration::from_secs(1)).await;
+        if failure == "count" {
+            assert_eq!(result.unwrap(), 5);
+            assert!(totals.iter().all(Option::is_none));
+        } else {
+            let error = result.unwrap_err();
+            if failure == "cancel" {
+                assert_eq!(error, "Cancelled");
+            } else {
+                assert!(error.contains(&format!("injected {failure} failure")), "{error}");
+            }
+            let trace = iris_transfer_trace(directory.path());
+            let inserts = trace
+                .iter()
+                .filter(|entry| entry["sql"].as_str().is_some_and(|sql| sql.starts_with("INSERT INTO")))
+                .count();
+            assert_eq!(
+                inserts,
+                if failure == "target" { 3 } else { 2 },
+                "failed writes must not be replayed or followed by more writes"
+            );
+        }
+    }
+}
@@ -43,8 +43,8 @@ pub use identifiers::{
    transfer_column_identifier,
 };
 pub use table_select::{
-    build_count_table_sql, build_table_data_select_sql, build_table_data_select_sql_with_database,
-    build_table_select_sql, DBX_LARGE_VALUE_BYTES_COLUMN_PREFIX,
+    build_count_table_sql, build_iris_table_select_sql, build_table_data_select_sql,
+    build_table_data_select_sql_with_database, build_table_select_sql, DBX_LARGE_VALUE_BYTES_COLUMN_PREFIX,
 };
 pub use table_select::{
    database_qualified_table_name, neo4j_element_id_function, quote_table_data_identifier,
@@ -339,7 +339,7 @@ pub fn build_table_data_select_sql_with_database(
            if options.use_driver_row_offset {
                format!("SELECT {select_columns} FROM {table_alias}{where_clause}{order}")
            } else {
-                build_iris_table_select_sql(&select_columns, &table_alias, &where_clause, &order, limit, offset)
+                build_iris_table_select_sql(&select_columns, &table_alias, &where_clause, &order, limit, offset as u64)
            }
        }
        TablePaginationStrategy::InformixFirst => {
@@ -679,24 +679,23 @@ fn quoted_table_columns_or_star(database_type: Option<DatabaseType>, columns: &[

 /// Builds one page of a Caché/IRIS table read.
 ///
-/// InterSystems SQL has `TOP` but no `OFFSET` clause — both Caché 2016 and IRIS
-/// reject `SELECT TOP n ... OFFSET m` — so later pages bound a derived table
-/// with `TOP(offset + limit)` and drop the leading rows with `%VID`, the row
+/// Caché and older IRIS SQL have `TOP` but no `OFFSET` clause, so later pages
+/// bound a derived table with `TOP(offset + limit)` and drop the leading rows with `%VID`, the row
 /// number InterSystems assigns to the rows a query produces. Running the same
 /// `TOP limit` statement for every page was the reason the grid kept showing
 /// the first page (#8929).
-fn build_iris_table_select_sql(
+pub fn build_iris_table_select_sql(
    select_columns: &str,
    table_alias: &str,
    where_clause: &str,
    order: &str,
    limit: usize,
-    offset: usize,
+    offset: u64,
 ) -> String {
    if offset == 0 {
        return format!("SELECT TOP {limit} {select_columns} FROM {table_alias}{where_clause}{order}");
    }
-    let window = offset.saturating_add(limit);
+    let window = offset.saturating_add(limit as u64);
    format!(
        "SELECT * FROM (SELECT TOP {window} {select_columns} FROM {table_alias}{where_clause}{order}) WHERE %VID > {offset}"
    )
@@ -33,7 +33,7 @@ capabilities:
  tableDataEdit: false
  tableStructureEdit: true
  tableImport: false
-  dataTransfer: false
+  dataTransfer: true
  sqlFileExecution: true
  databaseCreate: false
  fieldLineage: false