From 9338e69fb78b8d10885f96f99116d171ee0aee7f Mon Sep 17 00:00:00 2001 From: t8y2 <1156263951@qq.com> Date: Fri, 2 Oct 2026 22:31:43 +0000 Subject: [PATCH] fix(transfer): enable IRIS and Cache source and target data transfers Closes #10660 --- .../__tests__/transferConnections.spec.ts | 21 +- .../components/transfer/transferStrategy.ts | 2 +- .../assets/database-drivers.manifest.json | 2 +- crates/dbx-core/src/data/transfer.rs | 54 ++- .../dbx-core/src/data/transfer/iris_tests.rs | 357 ++++++++++++++++++ crates/dbx-sql-dialect/src/sql_dialect.rs | 4 +- .../src/sql_dialect/table_select.rs | 13 +- plugins/connection-types/iris.yaml | 2 +- 8 files changed, 422 insertions(+), 33 deletions(-) create mode 100644 crates/dbx-core/src/data/transfer/iris_tests.rs diff --git a/apps/desktop/src/components/transfer/__tests__/transferConnections.spec.ts b/apps/desktop/src/components/transfer/__tests__/transferConnections.spec.ts index b9c39c89f6..1eac92528b 100644 --- a/apps/desktop/src/components/transfer/__tests__/transferConnections.spec.ts +++ b/apps/desktop/src/components/transfer/__tests__/transferConnections.spec.ts @@ -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); diff --git a/apps/desktop/src/components/transfer/transferStrategy.ts b/apps/desktop/src/components/transfer/transferStrategy.ts index dacfaae6da..55c4b28ed1 100644 --- a/apps/desktop/src/components/transfer/transferStrategy.ts +++ b/apps/desktop/src/components/transfer/transferStrategy.ts @@ -16,7 +16,7 @@ export function transferStrategyOptions(strategy: TransferStrategy): Pick(["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 { diff --git a/crates/dbx-core/assets/database-drivers.manifest.json b/crates/dbx-core/assets/database-drivers.manifest.json index d4ba91a8b8..c32b2a50e2 100644 --- a/crates/dbx-core/assets/database-drivers.manifest.json +++ b/crates/dbx-core/assets/database-drivers.manifest.json @@ -2803,7 +2803,7 @@ "tableDataEdit": false, "tableStructureEdit": true, "tableImport": false, - "dataTransfer": false, + "dataTransfer": true, "sqlFileExecution": true, "databaseCreate": false, "fieldLineage": false, diff --git a/crates/dbx-core/src/data/transfer.rs b/crates/dbx-core/src/data/transfer.rs index 3b5fd5ba18..32d1ed754d 100644 --- a/crates/dbx-core/src/data/transfer.rs +++ b/crates/dbx-core/src/data/transfer.rs @@ -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>> = @@ -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::>().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 { diff --git a/crates/dbx-core/src/data/transfer/iris_tests.rs b/crates/dbx-core/src/data/transfer/iris_tests.rs new file mode 100644 index 0000000000..08117748de --- /dev/null +++ b/crates/dbx-core/src/data/transfer/iris_tests.rs @@ -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, 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 { + 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" + ); + } + } +} diff --git a/crates/dbx-sql-dialect/src/sql_dialect.rs b/crates/dbx-sql-dialect/src/sql_dialect.rs index 20331f57f7..595fc3a19b 100644 --- a/crates/dbx-sql-dialect/src/sql_dialect.rs +++ b/crates/dbx-sql-dialect/src/sql_dialect.rs @@ -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, diff --git a/crates/dbx-sql-dialect/src/sql_dialect/table_select.rs b/crates/dbx-sql-dialect/src/sql_dialect/table_select.rs index cd2ed03013..5335962450 100644 --- a/crates/dbx-sql-dialect/src/sql_dialect/table_select.rs +++ b/crates/dbx-sql-dialect/src/sql_dialect/table_select.rs @@ -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, 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}" ) diff --git a/plugins/connection-types/iris.yaml b/plugins/connection-types/iris.yaml index 55d236fc13..ed563e6b17 100644 --- a/plugins/connection-types/iris.yaml +++ b/plugins/connection-types/iris.yaml @@ -33,7 +33,7 @@ capabilities: tableDataEdit: false tableStructureEdit: true tableImport: false - dataTransfer: false + dataTransfer: true sqlFileExecution: true databaseCreate: false fieldLineage: false