feat(import): add option to skip duplicate rows

2026-10-19 16:16:45 +00:00 · 2026-09-26 23:14:03 +08:00
parent daebae9b50
commit 3c46f8d440
8 changed files with 328 additions and 32 deletions
@@ -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<api.TableImportJsonShape>("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<DatabaseType>(["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) => {
              <input v-model="emptyStringAsNull" type="checkbox" class="h-3.5 w-3.5 accent-primary" />
              {{ t("tableImport.emptyStringAsNull") }}
            </label>
+            <label v-if="supportsSkipDuplicateRows" class="flex items-center gap-2 text-xs">
+              <input v-model="skipDuplicateRows" type="checkbox" class="h-3.5 w-3.5 accent-primary" />
+              {{ t("tableImport.skipDuplicateRows") }}
+            </label>
          </div>

          <div v-else-if="sourceFormat === 'sql'" class="grid grid-cols-5 gap-3 rounded-md border p-3">
@@ -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);
@@ -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",
@@ -5443,6 +5443,7 @@ export interface TableImportRequest {
  dateTimeFormat?: string;
  preparedSource?: TableImportPreparedSource | null;
  retainSource?: boolean;
+  skipDuplicateRows?: boolean;
 }

 export interface TableImportSummary {
@@ -406,6 +406,7 @@ fn import_request(
        date_time_format: None,
        prepared_source: None,
        retain_source: false,
+        skip_duplicate_rows: false,
    }
 }

@@ -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<TableImportPreparedSource>,
    #[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<usize>,
 ) -> Result<Vec<ImportSqlBatch>, 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<usize>,
 ) -> Result<Vec<ImportSqlBatch>, 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<PostgresCopyAccumulator>,
    sqlite_append_transaction: &mut Option<SqliteAppendTransaction>,
    kingbase_oracle_mode: bool,
+    skip_duplicate_rows: bool,
    date_time_format: Option<&str>,
    hard_sql_bytes: Option<usize>,
    db_write_ms: &mut u128,
    statement_count: &mut usize,
 ) -> Result<usize, ImportRowsBatchError> {
    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,
        };

@@ -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::<Vec<_>>()
+        .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<Vec<(String, usize)>, 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<Vec<(String, usize)>, 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::<Vec<_>>();
+
+    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"));
@@ -133,6 +133,7 @@ fn live_sqlserver_import_request(
        date_time_format: None,
        prepared_source: None,
        retain_source: false,
+        skip_duplicate_rows: false,
    }
 }