diff --git a/apps/desktop/src/components/import/TableImportDialog.vue b/apps/desktop/src/components/import/TableImportDialog.vue index e44b4eb04c..18eeb65570 100644 --- a/apps/desktop/src/components/import/TableImportDialog.vue +++ b/apps/desktop/src/components/import/TableImportDialog.vue @@ -32,7 +32,7 @@ import { import { importPreviewInput, importSourceDisplayName, uploadedImportSourceFromPreview } from "@/lib/import/importSource"; import { getDataTypeOptions } from "@/lib/table/tableStructureEditorState"; import { metadataSchemaForConnection, tableStructureDatabaseTypeForConnection } from "@/lib/database/jdbcDialect"; -import type { ColumnInfo } from "@/types/database"; +import type { ColumnInfo, DatabaseType } from "@/types/database"; import * as api from "@/lib/backend/api"; const { t } = useI18n(); @@ -140,6 +140,7 @@ const titleRow = ref(1); const dataStartRow = ref(2); const lastDataRow = ref(0); const trimValues = ref(false); +const skipDuplicateRows = ref(false); const emptyStringAsNull = ref(defaultTableImportEmptyStringAsNull(sourceFormat.value)); const selectedSheet = ref(""); const jsonShape = ref("auto"); @@ -175,6 +176,10 @@ const wizardSteps: Array<{ value: TableImportWizardStep; labelKey: string }> = [ const selectedConnection = computed(() => (props.prefillConnectionId ? store.getConfig(props.prefillConnectionId) : undefined)); const structureDatabaseType = computed(() => tableStructureDatabaseTypeForConnection(selectedConnection.value)); +// Mirrors the dialect list in skip-generator dispatch (transfer.rs): only these +// engines turn the skip-duplicates option into conflict-handling INSERTs. +const SKIP_DUPLICATE_ROWS_DATABASE_TYPES = new Set(["postgres", "kingbase", "opengauss", "sqlite", "cloudflare-d1", "duckdb", "mysql", "doris", "starrocks"]); +const supportsSkipDuplicateRows = computed(() => structureDatabaseType.value !== undefined && SKIP_DUPLICATE_ROWS_DATABASE_TYPES.has(structureDatabaseType.value)); const targetSchema = computed(() => metadataSchemaForConnection(selectedConnection.value, props.prefillDatabase || "", props.prefillSchema)); const dataTypeOptions = computed(() => mergeDataTypeOptions(dynamicDataTypeOptions.value, getDataTypeOptions(structureDatabaseType.value), Object.values(columnDataTypes.value))); const hasExistingTarget = computed(() => !!props.prefillTable || loadingExistingTables.value || existingTableNames.value.length > 0); @@ -379,6 +384,7 @@ function resetState() { dataStartRow.value = 2; lastDataRow.value = 0; trimValues.value = false; + skipDuplicateRows.value = false; emptyStringAsNull.value = defaultTableImportEmptyStringAsNull(sourceFormat.value); selectedSheet.value = ""; jsonShape.value = "auto"; @@ -935,6 +941,7 @@ async function startImport() { mode: targetMode.value === "create" ? "append" : importMode.value, createTable: targetMode.value === "create", batchSize: Math.max(1, Number(batchSize.value) || 500), + skipDuplicateRows: supportsSkipDuplicateRows.value && skipDuplicateRows.value, dateTimeFormat: settingsStore.editorSettings.globalDateTimeImportFormat || undefined, preparedSource: preparedImportSource(currentPreview), }, @@ -1024,6 +1031,7 @@ async function startBatchImport() { mode: "append", createTable: true, batchSize: Math.max(1, Number(batchSize.value) || 500), + skipDuplicateRows: supportsSkipDuplicateRows.value && skipDuplicateRows.value, dateTimeFormat: settingsStore.editorSettings.globalDateTimeImportFormat || undefined, preparedSource: preparedImportSource(task.preview), retainSource: true, @@ -1414,6 +1422,10 @@ watch(rawProgressPercent, (percent) => { {{ t("tableImport.emptyStringAsNull") }} +
diff --git a/apps/desktop/src/components/import/__tests__/TableImportDialog.batchSelection.spec.ts b/apps/desktop/src/components/import/__tests__/TableImportDialog.batchSelection.spec.ts index bf4121bb43..3a4bef3523 100644 --- a/apps/desktop/src/components/import/__tests__/TableImportDialog.batchSelection.spec.ts +++ b/apps/desktop/src/components/import/__tests__/TableImportDialog.batchSelection.spec.ts @@ -26,6 +26,7 @@ vi.mock("@/stores/connectionStore", () => ({ getConfig: (id: string) => { if (id === "connection-1") return { id, name: "SQLite", db_type: "sqlite" }; if (id === "postgres-1") return { id, name: "PostgreSQL", db_type: "postgres" }; + if (id === "sqlserver-1") return { id, name: "SQL Server", db_type: "sqlserver" }; return undefined; }, ensureConnected: mocks.ensureConnected, @@ -158,6 +159,21 @@ import TableImportDialog from "@/components/import/TableImportDialog.vue"; const mountedApps: App[] = []; const sheets = ["First", "Second", "Third"]; +function delimitedPreview(): TableImportPreview { + return { + fileName: "rows.csv", + filePath: "/tmp/rows.csv", + sourceRef: "delimited-source", + fileType: "csv", + sizeBytes: 10, + columns: ["id", "name"], + rows: [[1, "a"]], + totalRows: 2, + totalRowsExact: true, + sourceFingerprint: "rows-csv", + }; +} + function workbookPreview(sheetName: string): TableImportPreview { const index = sheets.indexOf(sheetName); return { @@ -182,10 +198,10 @@ async function flushAsyncUpdates() { } } -async function mountDialog(files = [new File(["workbook"], "rows.xlsx")]) { +async function mountDialog(files = [new File(["workbook"], "rows.xlsx")], connectionId = "connection-1") { const container = document.createElement("div"); document.body.append(container); - const app = createApp(defineComponent({ setup: () => () => h(TableImportDialog, { open: true, prefillConnectionId: "connection-1", prefillDatabase: "main" }) })); + const app = createApp(defineComponent({ setup: () => () => h(TableImportDialog, { open: true, prefillConnectionId: connectionId, prefillDatabase: "main" }) })); mountedApps.push(app); app.use(i18n); app.mount(container); @@ -272,6 +288,22 @@ describe("TableImportDialog batch selection", () => { expect(mocks.importTableFile).not.toHaveBeenCalled(); }); + it("hides the skip-duplicates option for dialects without conflict handling", async () => { + mocks.previewTableImportFile.mockImplementation(() => Promise.resolve(delimitedPreview())); + await mountDialog([new File(["id,name\n1,a"], "rows.csv")], "sqlserver-1"); + await flushAsyncUpdates(); + expect(document.body.textContent).toContain("Trim values"); + expect(document.body.textContent).not.toContain("Skip duplicate rows"); + }); + + it("offers the skip-duplicates option for dialects with conflict handling", async () => { + mocks.previewTableImportFile.mockImplementation(() => Promise.resolve(delimitedPreview())); + await mountDialog([new File(["id,name\n1,a"], "rows.csv")]); + await flushAsyncUpdates(); + expect(document.body.textContent).toContain("Trim values"); + expect(document.body.textContent).toContain("Skip duplicate rows"); + }); + it("keeps every worksheet selected by default and imports them in workbook order", async () => { await mountDialog(); for (const name of ["rows_First", "rows_Second", "rows_Third"]) expect(taskCheckbox(name).checked).toBe(true); diff --git a/apps/desktop/src/i18n/locales/en.ts b/apps/desktop/src/i18n/locales/en.ts index 99b16246f8..69c6d0323b 100644 --- a/apps/desktop/src/i18n/locales/en.ts +++ b/apps/desktop/src/i18n/locales/en.ts @@ -6852,6 +6852,7 @@ export default { lastDataRow: "Last data row (0 for end)", trimValues: "Trim values", emptyStringAsNull: "Empty string as NULL", + skipDuplicateRows: "Skip duplicate rows", jsonShape: "JSON rows", jsonShapeAuto: "Auto-detect", jsonShapeObjects: "Objects", diff --git a/apps/desktop/src/lib/backend/tauri.ts b/apps/desktop/src/lib/backend/tauri.ts index ac580da510..c3bc1e814d 100644 --- a/apps/desktop/src/lib/backend/tauri.ts +++ b/apps/desktop/src/lib/backend/tauri.ts @@ -5443,6 +5443,7 @@ export interface TableImportRequest { dateTimeFormat?: string; preparedSource?: TableImportPreparedSource | null; retainSource?: boolean; + skipDuplicateRows?: boolean; } export interface TableImportSummary { diff --git a/crates/dbx-core/examples/table_import_live_bench.rs b/crates/dbx-core/examples/table_import_live_bench.rs index 6f4021270d..797e91f24a 100644 --- a/crates/dbx-core/examples/table_import_live_bench.rs +++ b/crates/dbx-core/examples/table_import_live_bench.rs @@ -406,6 +406,7 @@ fn import_request( date_time_format: None, prepared_source: None, retain_source: false, + skip_duplicate_rows: false, } } diff --git a/crates/dbx-core/src/data/table_import.rs b/crates/dbx-core/src/data/table_import.rs index 530003412e..745003c9eb 100644 --- a/crates/dbx-core/src/data/table_import.rs +++ b/crates/dbx-core/src/data/table_import.rs @@ -28,9 +28,8 @@ use crate::models::connection::DatabaseType; use crate::sql::SqlParsingOptions; use crate::sql_file_import::{SqlFileStreamDecoder, StreamingSqlFileSplitter}; use crate::transfer::{ - escape_value_typed, execute_on_pool, generate_insert_typed_from_value_rows, - generate_insert_typed_sql_batches_from_value_rows, get_columns_for_transfer, normalize_integer_literal, - normalize_thousands_numeric_literal, qualified_table, quote_identifier, SqlBatchLimits, + escape_value_typed, execute_on_pool, generate_insert_typed_from_value_rows, get_columns_for_transfer, + normalize_integer_literal, normalize_thousands_numeric_literal, qualified_table, quote_identifier, SqlBatchLimits, }; pub const DEFAULT_PREVIEW_LIMIT: usize = 50; @@ -272,6 +271,8 @@ pub struct TableImportRequest { pub prepared_source: Option, #[serde(default)] pub retain_source: bool, + #[serde(default)] + pub skip_duplicate_rows: bool, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -4297,6 +4298,7 @@ fn build_import_insert_batches_with_plan( schema: &str, db_type: &DatabaseType, kingbase_oracle_mode: bool, + skip_duplicate_rows: bool, date_time_format: Option<&str>, hard_sql_bytes: Option, ) -> Result, String> { @@ -4304,7 +4306,7 @@ fn build_import_insert_batches_with_plan( return Ok(Vec::new()); } let value_rows = import_value_rows_sql(rows, plan, db_type, kingbase_oracle_mode, date_time_format); - let batches = generate_insert_typed_sql_batches_from_value_rows( + let batches = crate::data::transfer::generate_insert_typed_sql_batches_from_value_rows_with_options( &plan.target_columns, &value_rows, table, @@ -4312,6 +4314,7 @@ fn build_import_insert_batches_with_plan( db_type, None, SqlBatchLimits::for_database(db_type, rows.len()).with_hard_sql_bytes(hard_sql_bytes), + skip_duplicate_rows, )?; Ok(batches.into_iter().map(|(sql, row_count)| ImportSqlBatch { sql, row_count }).collect()) } @@ -4377,6 +4380,7 @@ fn build_import_execution_batches( schema: &str, db_type: &DatabaseType, kingbase_oracle_mode: bool, + skip_duplicate_rows: bool, date_time_format: Option<&str>, hard_sql_bytes: Option, ) -> Result, String> { @@ -4388,6 +4392,7 @@ fn build_import_execution_batches( schema, db_type, kingbase_oracle_mode, + skip_duplicate_rows, date_time_format, hard_sql_bytes, ); @@ -4411,6 +4416,7 @@ fn build_import_execution_batches( schema, db_type, kingbase_oracle_mode, + skip_duplicate_rows, date_time_format, hard_sql_bytes, ) @@ -4609,6 +4615,7 @@ fn build_import_insert_batches_with_format( schema, db_type, kingbase_oracle_mode, + false, date_time_format, None, )?); @@ -5583,31 +5590,36 @@ async fn execute_import_rows_batch( postgres_copy_accumulator: &mut Option, sqlite_append_transaction: &mut Option, kingbase_oracle_mode: bool, + skip_duplicate_rows: bool, date_time_format: Option<&str>, hard_sql_bytes: Option, db_write_ms: &mut u128, statement_count: &mut usize, ) -> Result { let execution_policy = import_batch_execution_policy(mode, pending_truncate, db_type); - if let Some((import_plan, bulk_plan)) = sqlserver_bulk_plans_for_rows(db_type, plan, sqlserver_bulk_plan, rows) { - return execute_sqlserver_bulk_rows_batch( - state, - pool_key, - import_id, - is_cancelled, - rows, - import_plan, - bulk_plan, - execution_policy.include_truncate, - date_time_format, - db_write_ms, - statement_count, - ) - .await; + if !skip_duplicate_rows { + if let Some((import_plan, bulk_plan)) = sqlserver_bulk_plans_for_rows(db_type, plan, sqlserver_bulk_plan, rows) + { + return execute_sqlserver_bulk_rows_batch( + state, + pool_key, + import_id, + is_cancelled, + rows, + import_plan, + bulk_plan, + execution_policy.include_truncate, + date_time_format, + db_write_ms, + statement_count, + ) + .await; + } } // COPY is used only for plain scalar PostgreSQL rows and ordinary tables. Any unsupported // value or table feature falls through to the portable INSERT generator below. - if execution_policy.allow_postgres_copy + if !skip_duplicate_rows + && execution_policy.allow_postgres_copy && *db_type == DatabaseType::Postgres && !rows .iter() @@ -5650,6 +5662,7 @@ async fn execute_import_rows_batch( schema, db_type, kingbase_oracle_mode, + skip_duplicate_rows, date_time_format, hard_sql_bytes, ) @@ -5700,7 +5713,7 @@ async fn execute_import_rows_batch( statements.push(truncate_sql(table, schema, db_type)); } statements.extend(batches.into_iter().map(|batch| batch.sql)); - execute_import_transaction( + let result = execute_import_transaction( state, pool_key, connection_id, @@ -5712,14 +5725,22 @@ async fn execute_import_rows_batch( ) .await .map_err(|message| ImportRowsBatchError::with_rows_imported(rows_imported, message))?; - return Ok(rows_imported.saturating_add(rows.len())); + return Ok(if skip_duplicate_rows { + rows_imported.saturating_add(result.affected_rows as usize) + } else { + rows_imported.saturating_add(rows.len()) + }); } for batch in batches { ensure_import_write_allowed(import_id, is_cancelled, rows_imported).await?; - if let Err(error) = execute_import_statement(state, pool_key, &batch.sql, db_write_ms, statement_count).await { - return Err(ImportRowsBatchError::with_rows_imported(rows_imported, error)); - } - rows_imported = rows_imported.saturating_add(batch.row_count); + let result = execute_import_statement(state, pool_key, &batch.sql, db_write_ms, statement_count) + .await + .map_err(|error| ImportRowsBatchError::with_rows_imported(rows_imported, error))?; + rows_imported = if skip_duplicate_rows { + rows_imported.saturating_add(result.affected_rows as usize) + } else { + rows_imported.saturating_add(batch.row_count) + }; } Ok(rows_imported) } @@ -6879,6 +6900,7 @@ where &mut postgres_copy_accumulator, &mut sqlite_append_transaction, kingbase_oracle_mode, + request.skip_duplicate_rows, request.date_time_format.as_deref(), import_sql_hard_limit, &mut db_write_ms, @@ -7381,6 +7403,7 @@ where &mut postgres_copy_accumulator, &mut sqlite_append_transaction, kingbase_oracle_mode, + request.skip_duplicate_rows, request.date_time_format.as_deref(), import_sql_hard_limit, &mut db_write_ms, @@ -7729,6 +7752,7 @@ where &mut postgres_copy_accumulator, &mut sqlite_append_transaction, kingbase_oracle_mode, + request.skip_duplicate_rows, request.date_time_format.as_deref(), import_sql_hard_limit, &mut db_write_ms, @@ -7973,6 +7997,7 @@ mod tests { total_rows_exact: true, effective_encoding: Some(TableImportTextEncoding::Utf8), }), + skip_duplicate_rows: false, retain_source: false, }; @@ -9619,6 +9644,7 @@ mod tests { batch_size: 500, date_time_format: None, prepared_source: None, + skip_duplicate_rows: false, retain_source: false, }; let started_at = Instant::now(); @@ -10029,6 +10055,7 @@ mod tests { batch_size: 1, date_time_format: None, prepared_source: None, + skip_duplicate_rows: false, retain_source: false, }; @@ -10118,6 +10145,7 @@ mod tests { batch_size: 1, date_time_format: None, prepared_source: None, + skip_duplicate_rows: false, retain_source: false, }; @@ -10198,6 +10226,7 @@ mod tests { batch_size: 1, date_time_format: None, prepared_source: None, + skip_duplicate_rows: false, retain_source: false, }; @@ -11553,6 +11582,7 @@ mod tests { "dbo", &DatabaseType::SqlServer, false, + false, None, None, ) @@ -11875,8 +11905,9 @@ mod tests { &DatabaseType::Sqlite, &TableImportMode::Append, false, - &mut postgres_copy_accumulator, &mut sqlite_append_transaction, + &mut postgres_copy_accumulator, + false, false, None, None, @@ -12095,6 +12126,7 @@ mod tests { &mut self.postgres_copy_accumulator, &mut self.transaction, false, + false, None, None, &mut self.db_write_ms, @@ -12219,6 +12251,7 @@ mod tests { batch_size: 2, date_time_format: None, prepared_source: None, + skip_duplicate_rows: false, retain_source: false, }; diff --git a/crates/dbx-core/src/data/transfer.rs b/crates/dbx-core/src/data/transfer.rs index 73f9dda548..dbdcd04e28 100644 --- a/crates/dbx-core/src/data/transfer.rs +++ b/crates/dbx-core/src/data/transfer.rs @@ -3979,6 +3979,52 @@ pub fn generate_upsert_typed( ) } +fn generate_insert_ignore_duplicates_from_value_rows( + columns: &[String], + value_rows: &[String], + table: &str, + schema: &str, + db_type: &DatabaseType, + catalog: Option<&str>, + overrides_postgres_system_values: bool, + quote_target_column_names: bool, +) -> String { + if value_rows.is_empty() || columns.is_empty() { + return String::new(); + } + + let full_table = qualified_table(table, schema, db_type, catalog); + let col_list = columns + .iter() + .map(|column| transfer_column_identifier(column, db_type, quote_target_column_names)) + .collect::>() + .join(", "); + + let overriding = if overrides_postgres_system_values && matches!(db_type, DatabaseType::Postgres) { + " OVERRIDING SYSTEM VALUE" + } else { + "" + }; + + let mut sql = format!("INSERT INTO {full_table} ({col_list}){overriding} VALUES\n{}", value_rows.join(",\n")); + + match db_type { + db_type + if (is_postgres_transfer_dialect(db_type) && !matches!(db_type, DatabaseType::OpenGauss)) + || matches!(db_type, DatabaseType::Sqlite | DatabaseType::CloudflareD1 | DatabaseType::DuckDb) => + { + sql.push_str("\nON CONFLICT DO NOTHING"); + } + db_type if uses_mysql_style_upsert(db_type) => { + let first_column = transfer_column_identifier(&columns[0], db_type, quote_target_column_names); + sql.push_str(&format!("\nON DUPLICATE KEY UPDATE {first_column} = {first_column}")); + } + _ => {} + } + + sql +} + /// Upsert targets that take the MySQL-style `INSERT ... ON DUPLICATE KEY UPDATE /// ... VALUES(col)` arm. openGauss belongs here instead of the PostgreSQL /// `ON CONFLICT` arm: its INSERT grammar has no `ON CONFLICT` clause, but it @@ -4591,11 +4637,116 @@ pub(crate) fn generate_insert_typed_sql_batches_from_value_rows( catalog: Option<&str>, limits: SqlBatchLimits, ) -> Result, String> { - generate_insert_sql_batches_from_value_rows( - columns, value_rows, table, schema, db_type, catalog, limits, false, true, + generate_insert_typed_sql_batches_from_value_rows_with_options( + columns, value_rows, table, schema, db_type, catalog, limits, false, ) } +#[allow(clippy::too_many_arguments)] +pub(crate) fn generate_insert_typed_sql_batches_from_value_rows_with_options( + columns: &[String], + value_rows: &[String], + table: &str, + schema: &str, + db_type: &DatabaseType, + catalog: Option<&str>, + limits: SqlBatchLimits, + skip_duplicate_rows: bool, +) -> Result, String> { + if !skip_duplicate_rows { + return generate_insert_sql_batches_from_value_rows( + columns, value_rows, table, schema, db_type, catalog, limits, false, true, + ); + } + + if value_rows.is_empty() { + return Ok(Vec::new()); + } + + let max_rows = limits.max_rows.max(1).min(match db_type { + DatabaseType::SqlServer => MAX_SQLSERVER_INSERT_ROWS, + DatabaseType::Oracle => MAX_ORACLE_INSERT_ALL_ROWS, + _ => usize::MAX, + }); + let target_sql_bytes = limits.target_sql_bytes.max(1); + let batch_sql_bytes = limits.hard_sql_bytes.map_or(target_sql_bytes, |hard| target_sql_bytes.min(hard)); + + let template = InsertSqlTemplate::new_with_column_quoting(columns, table, schema, db_type, catalog, false, true); + + let value_row_bytes = value_rows.iter().map(|row| sql_text_bytes(row, db_type)).collect::>(); + + let mut statements = Vec::new(); + let mut start = 0usize; + + while start < value_rows.len() { + let mut end = start; + let mut rows_bytes = 0usize; + + while end < value_rows.len() && end - start < max_rows { + let single_row_bytes = template.statement_bytes(value_row_bytes[end], 1, db_type); + + if let Some(hard_sql_bytes) = limits.hard_sql_bytes { + if single_row_bytes > hard_sql_bytes { + return Err(format!( + "SQL batch row {} requires {} bytes and exceeds the {} byte hard limit", + end + 1, + single_row_bytes, + hard_sql_bytes + )); + } + } + + let candidate_rows_bytes = rows_bytes.saturating_add(value_row_bytes[end]); + let candidate_row_count = end - start + 1; + let candidate_bytes = template.statement_bytes(candidate_rows_bytes, candidate_row_count, db_type); + + if candidate_row_count > 1 && candidate_bytes > batch_sql_bytes { + break; + } + + rows_bytes = candidate_rows_bytes; + end += 1; + } + + let value_rows_batch = &value_rows[start..end]; + + let mut sql = if matches!( + db_type, + DatabaseType::Postgres + | DatabaseType::Kingbase + | DatabaseType::Sqlite + | DatabaseType::CloudflareD1 + | DatabaseType::DuckDb + | DatabaseType::Mysql + | DatabaseType::Doris + | DatabaseType::StarRocks + | DatabaseType::OpenGauss + ) { + generate_insert_ignore_duplicates_from_value_rows( + columns, + value_rows_batch, + table, + schema, + db_type, + catalog, + false, + true, + ) + } else { + template.build(value_rows_batch) + }; + + if sql.is_empty() { + return Err("Generated empty SQL batch".to_string()); + } + + statements.push((std::mem::take(&mut sql), end - start)); + start = end; + } + + Ok(statements) +} + #[allow(clippy::too_many_arguments)] fn generate_insert_typed_sql_batches_for_transfer( columns: &[String], @@ -16407,6 +16558,70 @@ SELECT 1 FROM dual"# assert_eq!(sql, "INSERT INTO \"public\".\"users\" (\"id\", \"name\") VALUES\n(42, 'Ada')"); } + #[test] + fn postgres_insert_can_skip_duplicate_rows() { + let batches = generate_insert_typed_sql_batches_from_value_rows_with_options( + &[String::from("id"), String::from("name")], + &[String::from("(42, 'Ada')"), String::from("(43, 'Bob')")], + "users", + "public", + &DatabaseType::Postgres, + None, + SqlBatchLimits { max_rows: 100, target_sql_bytes: 1024, hard_sql_bytes: None }, + true, + ) + .unwrap(); + + assert_eq!(batches.len(), 1); + assert_eq!(batches[0].1, 2); + assert!(batches[0].0.contains("ON CONFLICT DO NOTHING")); + assert!(batches[0].0.contains("(42, 'Ada')")); + assert!(batches[0].0.contains("(43, 'Bob')")); + } + + #[test] + fn mysql_insert_can_skip_duplicate_rows() { + let batches = generate_insert_typed_sql_batches_from_value_rows_with_options( + &[String::from("id"), String::from("name")], + &[String::from("(42, 'Ada')"), String::from("(43, 'Bob')")], + "users", + "public", + &DatabaseType::Mysql, + None, + SqlBatchLimits { max_rows: 100, target_sql_bytes: 1024, hard_sql_bytes: None }, + true, + ) + .unwrap(); + + assert_eq!(batches.len(), 1); + assert_eq!(batches[0].1, 2); + assert!(batches[0].0.contains("ON DUPLICATE KEY UPDATE")); + assert!(batches[0].0.contains("`id` = `id`")); + assert!(batches[0].0.contains("(42, 'Ada')")); + assert!(batches[0].0.contains("(43, 'Bob')")); + } + + #[test] + fn sqlite_insert_can_skip_duplicate_rows() { + let batches = generate_insert_typed_sql_batches_from_value_rows_with_options( + &[String::from("id"), String::from("name")], + &[String::from("(42, 'Ada')"), String::from("(43, 'Bob')")], + "users", + "", + &DatabaseType::Sqlite, + None, + SqlBatchLimits { max_rows: 100, target_sql_bytes: 1024, hard_sql_bytes: None }, + true, + ) + .unwrap(); + + assert_eq!(batches.len(), 1); + assert_eq!(batches[0].1, 2); + assert!(batches[0].0.contains("ON CONFLICT DO NOTHING")); + assert!(batches[0].0.contains("(42, 'Ada')")); + assert!(batches[0].0.contains("(43, 'Bob')")); + } + #[test] fn database_from_pool_key_handles_session_scoped_keys() { assert_eq!(database_from_pool_key("conn:analytics"), Some("analytics")); diff --git a/crates/dbx-core/tests/live_sqlserver_completion.rs b/crates/dbx-core/tests/live_sqlserver_completion.rs index ac3abd1811..7f41a335bc 100644 --- a/crates/dbx-core/tests/live_sqlserver_completion.rs +++ b/crates/dbx-core/tests/live_sqlserver_completion.rs @@ -133,6 +133,7 @@ fn live_sqlserver_import_request( date_time_format: None, prepared_source: None, retain_source: false, + skip_duplicate_rows: false, } }