feat(transfer): add uppercase column names option

Closes #12741

apps/desktop/src/components/transfer/DataTransferDialog.vue

@@ -30,7 +30,7 @@ import { useConnectionStore } from "@/stores/connectionStore";
 import { ensureReadOnlyWriteAccess } from "@/lib/database/readOnlyWriteAccess";
 import { useProductionSafetyStore } from "@/stores/productionSafetyStore";
 import * as api from "@/lib/backend/api";
-import type { TransferContent, TransferObjectKind, TransferTableNameCase } from "@/lib/backend/api";
+import type { TransferColumnNameCase, TransferContent, TransferObjectKind, TransferTableNameCase } from "@/lib/backend/api";
 import { crossFamilyTransferableKinds, isSameTransferFamily, isTransferPairSupported, transferObjectFamily, transferObjectKindsForDatabase } from "@/lib/database/transferObjectKinds";
 import ObjectSelectionTree from "@/components/transfer/ObjectSelectionTree.vue";
 import TransferTaskTree from "@/components/transfer/TransferTaskTree.vue";
@@ -208,6 +208,7 @@ const targetTableStrategy = computed<TransferStrategy>({
   },
 });
 const targetTableNameCase = ref<TransferTableNameCase>("preserve");
+const targetColumnNameCase = ref<TransferColumnNameCase>("preserve");
 const quoteTargetColumnNames = ref(true);
 const batchSize = ref(1000);
 const operationTimeoutMode = ref<"inherit" | "custom">("inherit");
@@ -297,7 +298,10 @@ function structureOverwriteSupported(): boolean {
 // The table-name-case selector is hidden for Xugu pairs; reset any earlier choice so the
 // blocking xuguPreserveTableNames hint cannot strand the user without a visible control.
 watch(isXuguTransferPair, (isXuguPair) => {
-  if (isXuguPair) targetTableNameCase.value = "preserve";
+  if (isXuguPair) {
+    targetTableNameCase.value = "preserve";
+    targetColumnNameCase.value = "preserve";
+  }
 });
 const targetStrategyType = computed(() => (targetTransferType.value === "xugu" ? "xugu" : connectionType(targetConnectionId.value)));
 const xuguTransferHint = computed(() => {
@@ -823,6 +827,7 @@ function resetState(cancelTaskLoad = true) {
   transferContent.value = "structureAndData";
   selectedStrategy.value = "append";
   targetTableNameCase.value = "preserve";
+  targetColumnNameCase.value = "preserve";
   quoteTargetColumnNames.value = true;
   batchSize.value = 1000;
   operationTimeoutMode.value = "inherit";
@@ -945,6 +950,7 @@ async function requestStartTransfer() {
     ...buildTransferObjectSelectionField(selectedObjects.value, treeDisabledGroups.value),
     ...transferStrategyOptions(targetTableStrategy.value),
     targetTableNameCase: targetTableNameCase.value,
+    targetColumnNameCase: targetColumnNameCase.value,
     quoteTargetColumnNames: quoteTargetColumnNames.value,
     ownershipPolicy: "preserve",
     batchSize: batchSize.value,
@@ -1029,6 +1035,7 @@ function currentConfig(): TransferTaskConfig {
     content: transferContent.value,
     ...transferStrategyOptions(targetTableStrategy.value),
     targetTableNameCase: targetTableNameCase.value,
+    targetColumnNameCase: targetColumnNameCase.value,
     quoteTargetColumnNames: quoteTargetColumnNames.value,
     batchSize: batchSize.value,
     operationTimeoutSecs: operationTimeoutMode.value === "custom" && operationTimeoutValid.value ? operationTimeoutSecs.value : undefined,
@@ -1083,6 +1090,7 @@ async function loadTaskIntoForm(task: TransferTask) {
   transferContent.value = config.content;
   selectedStrategy.value = resolveTransferStrategy(config);
   targetTableNameCase.value = config.targetTableNameCase;
+  targetColumnNameCase.value = config.targetColumnNameCase ?? "preserve";
   quoteTargetColumnNames.value = config.quoteTargetColumnNames;
   batchSize.value = config.batchSize;
   operationTimeoutMode.value = config.operationTimeoutSecs === undefined ? "inherit" : "custom";
@@ -1632,6 +1640,19 @@ async function saveConfigTask() {
                   </SelectContent>
                 </Select>
               </div>
+              <div v-if="!isXuguTransferPair" class="flex items-center gap-3">
+                <Label class="text-xs shrink-0">{{ t("transfer.targetColumnNameCase") }}</Label>
+                <Select v-model="targetColumnNameCase">
+                  <SelectTrigger class="h-7 text-xs">
+                    <SelectValue />
+                  </SelectTrigger>
+                  <SelectContent>
+                    <SelectItem value="preserve">{{ t("transfer.tableNameCasePreserve") }}</SelectItem>
+                    <SelectItem value="lower">{{ t("transfer.tableNameCaseLower") }}</SelectItem>
+                    <SelectItem value="upper">{{ t("transfer.tableNameCaseUpper") }}</SelectItem>
+                  </SelectContent>
+                </Select>
+              </div>
               <div v-if="showTargetColumnQuoteOption" class="flex items-center gap-3">
                 <Label for="transfer-quote-target-column-names" class="text-xs shrink-0">{{ t("transfer.quoteTargetColumnNames") }}</Label>
                 <Switch id="transfer-quote-target-column-names" v-model="quoteTargetColumnNames" size="sm" />

apps/desktop/src/i18n/locales/az.ts

@@ -6306,6 +6306,7 @@ export default withEnglishFallback({
     modeOverwrite: "Üzərinə yaz (TRUNCATE + INSERT)",
     modeUpsert: "Əlavə et və ya yenilə (PK üzrə INSERT və ya UPDATE)",
     targetTableNameCase: "Hədəf cədvəl adının hərf registri",
+    targetColumnNameCase: "Hədəf sütun adının hərf registri",
     quoteTargetColumnNames: "Hədəf sütun adlarını dırnağa al",
     tableNameCasePreserve: "Qoru",
     tableNameCaseLower: "Kiçik hərflər",

apps/desktop/src/i18n/locales/en.ts

@@ -7632,6 +7632,7 @@ export default {
     modeReplaceExisting: "Replace existing objects",
     modeUpsert: "Upsert (INSERT or UPDATE by PK)",
     targetTableNameCase: "Target table name case",
+    targetColumnNameCase: "Target column name case",
     quoteTargetColumnNames: "Quote target column names",
     tableNameCasePreserve: "Preserve",
     tableNameCaseLower: "Lowercase",

apps/desktop/src/i18n/locales/es.ts

@@ -6848,6 +6848,7 @@ export default withEnglishFallback({
     modeReplaceExisting: "Reemplazar objetos existentes",
     modeUpsert: "Upsert (INSERT o UPDATE por PK)",
     targetTableNameCase: "Mayúsculas/minúsculas de tabla destino",
+    targetColumnNameCase: "Mayúsculas/minúsculas de columna destino",
     quoteTargetColumnNames: "Citar columnas de destino",
     tableNameCasePreserve: "Conservar",
     tableNameCaseLower: "Convertir a minúsculas",

apps/desktop/src/i18n/locales/id.ts

@@ -6969,6 +6969,7 @@ export default withEnglishFallback({
     modeOverwrite: "Timpa (TRUNCATE + INSERT)",
     modeUpsert: "Upsert (INSERT atau UPDATE berdasarkan PK)",
     targetTableNameCase: "Kapitalisasi nama tabel target",
+    targetColumnNameCase: "Kapitalisasi nama kolom target",
     quoteTargetColumnNames: "Beri tanda kutip pada nama kolom target",
     tableNameCasePreserve: "Pertahankan",
     tableNameCaseLower: "Huruf kecil",

apps/desktop/src/i18n/locales/it.ts

@@ -6665,6 +6665,7 @@ export default withEnglishFallback({
     modeReplaceExisting: "Sostituisci oggetti esistenti",
     modeUpsert: "Aggiorna/Inserisci (INSERT o UPDATE per chiave primaria)",
     targetTableNameCase: "Maiuscole/minuscole tabella destinazione",
+    targetColumnNameCase: "Maiuscole/minuscole colonna destinazione",
     quoteTargetColumnNames: "Quota colonne di destinazione",
     tableNameCasePreserve: "Mantieni",
     tableNameCaseLower: "Converti in minuscolo",

apps/desktop/src/i18n/locales/ja.ts

@@ -6685,6 +6685,7 @@ export default withEnglishFallback({
     modeReplaceExisting: "既存のオブジェクトを置き換え",
     modeUpsert: "アップサート (主キーでINSERT or UPDATE)",
     targetTableNameCase: "対象テーブル名の大小文字",
+    targetColumnNameCase: "対象列名の大小文字",
     quoteTargetColumnNames: "対象列名を引用",
     tableNameCasePreserve: "そのまま",
     tableNameCaseLower: "小文字に変換",

apps/desktop/src/i18n/locales/ko.ts

@@ -6463,6 +6463,7 @@ export default withEnglishFallback({
     modeReplaceExisting: "기존 객체 교체",
     modeUpsert: "Upsert (PK 기준 INSERT 또는 UPDATE)",
     targetTableNameCase: "대상 테이블 이름 대소문자",
+    targetColumnNameCase: "대상 컬럼 이름 대소문자",
     quoteTargetColumnNames: "대상 열 이름 인용",
     tableNameCasePreserve: "유지",
     tableNameCaseLower: "소문자",

apps/desktop/src/i18n/locales/pl.ts

@@ -7013,6 +7013,7 @@ export default withEnglishFallback({
     modeOverwrite: "Nadpisz (TRUNCATE + INSERT)",
     modeUpsert: "Wstaw lub zaktualizuj (INSERT lub UPDATE wg PK)",
     targetTableNameCase: "Wielkość liter nazwy tabeli docelowej",
+    targetColumnNameCase: "Wielkość liter nazwy kolumny docelowej",
     quoteTargetColumnNames: "Ujmuj nazwy kolumn docelowych w cudzysłowy",
     tableNameCasePreserve: "Zachowaj",
     tableNameCaseLower: "Małe litery",

apps/desktop/src/i18n/locales/pt-BR.ts

@@ -6665,6 +6665,7 @@ export default withEnglishFallback({
     modeReplaceExisting: "Substituir objetos existentes",
     modeUpsert: "Upsert (INSERT ou UPDATE por PK)",
     targetTableNameCase: "Maiúsculas/minúsculas da tabela de destino",
+    targetColumnNameCase: "Maiúsculas/minúsculas da coluna de destino",
     quoteTargetColumnNames: "Citar colunas de destino",
     tableNameCasePreserve: "Preservar",
     tableNameCaseLower: "Converter para minúsculas",

apps/desktop/src/i18n/locales/ru.ts

@@ -7898,6 +7898,7 @@ export default withEnglishFallback({
     modeReplaceExisting: "Заменить существующие объекты",
     modeUpsert: "Upsert (INSERT или UPDATE по PK)",
     targetTableNameCase: "Регистр имён целевых таблиц",
+    targetColumnNameCase: "Регистр имён целевых столбцов",
     quoteTargetColumnNames: "Заключать имена целевых столбцов в кавычки",
     tableNameCasePreserve: "Сохранять",
     tableNameCaseLower: "Нижний регистр",

apps/desktop/src/i18n/locales/th.ts

@@ -8287,6 +8287,7 @@ export default withEnglishFallback({
     modeReplaceExisting: "แทนที่ออบเจ็กต์ที่มีอยู่",
     modeUpsert: "Upsert (INSERT หรือ UPDATE ด้วย PK)",
     targetTableNameCase: "ตัวพิมพ์เล็กใหญ่ของชื่อ Table เป้าหมาย",
+    targetColumnNameCase: "ตัวพิมพ์เล็กใหญ่ของชื่อคอลัมน์เป้าหมาย",
     quoteTargetColumnNames: "ใส่ Quote ชื่อ Column เป้าหมาย",
     tableNameCasePreserve: "คงเดิม",
     tableNameCaseLower: "ตัวพิมพ์เล็ก",

apps/desktop/src/i18n/locales/tr.ts

@@ -6197,6 +6197,7 @@ export default withEnglishFallback({
     modeOverwrite: "Üzerine yaz (TRUNCATE + INSERT)",
     modeUpsert: "Upsert (PK'ye göre INSERT veya UPDATE)",
     targetTableNameCase: "Hedef tablo adı harf durumu",
+    targetColumnNameCase: "Hedef sütun adı harf durumu",
     quoteTargetColumnNames: "Hedef sütun adlarını tırnak içine al",
     tableNameCasePreserve: "Koru",
     tableNameCaseLower: "Küçük harf",

apps/desktop/src/i18n/locales/zh-CN.ts

@@ -7613,6 +7613,7 @@ export default withEnglishFallback({
     modeReplaceExisting: "存在则替换对象",
     modeUpsert: "更新插入 (按主键 INSERT 或 UPDATE)",
     targetTableNameCase: "目标表名大小写",
+    targetColumnNameCase: "目标字段名大小写",
     quoteTargetColumnNames: "引用目标字段名",
     tableNameCasePreserve: "保持原样",
     tableNameCaseLower: "转为小写",

apps/desktop/src/i18n/locales/zh-TW.ts

@@ -6064,6 +6064,7 @@ export default withEnglishFallback({
     overallProgress: "整體進度",
     dataTransfer: "資料傳輸",
     targetTableNameCase: "目標資料表名稱大小寫",
+    targetColumnNameCase: "目標欄位名稱大小寫",
     quoteTargetColumnNames: "引用目標欄位名稱",
     tableNameCasePreserve: "保留",
     tableNameCaseLower: "小寫",

apps/desktop/src/lib/backend/api.ts

@@ -1161,6 +1161,7 @@ export type {
   TransferObjectKind,
   TransferObjectSelection,
   TransferTableNameCase,
+  TransferColumnNameCase,
   TransferOwnershipPolicy,
   TransferStructureOperation,
   TransferStructureOperationKind,

apps/desktop/src/lib/backend/tauri.ts

@@ -5840,6 +5840,7 @@ export async function listenSqlFileProgress(handler: (progress: SqlFileProgress)
 // --- Data Transfer ---
 export type TransferMode = "append" | "overwrite" | "upsert";
 export type TransferTableNameCase = "preserve" | "lower" | "upper";
+export type TransferColumnNameCase = "preserve" | "lower" | "upper";
 export type TransferOwnershipPolicy = "preserve" | "skip" | "reassignMissing";
 export type TransferContent = "structureAndData" | "structureOnly" | "dataOnly";
 export type TransferObjectKind = "TABLE" | "VIEW" | "MATERIALIZED_VIEW" | "PROCEDURE" | "FUNCTION" | "TRIGGER" | "SEQUENCE" | "EVENT";
@@ -5865,6 +5866,7 @@ export interface TransferRequest {
   objects: TransferObjectSelection[];
   mode: TransferMode;
   targetTableNameCase: TransferTableNameCase;
+  targetColumnNameCase?: TransferColumnNameCase;
   quoteTargetColumnNames: boolean;
   ownershipPolicy?: TransferOwnershipPolicy;
   batchSize: number;

apps/desktop/src/stores/__tests__/transferTaskStore.spec.ts

@@ -115,6 +115,27 @@ describe("transferTaskStore", () => {
     expect(store.tasks[0]?.config.quoteTargetColumnNames).toBe(true);
   });
 
+  it("round-trips the target column name case and defaults legacy tasks to preserve", async () => {
+    const legacyConfig = makeConfig() as Partial<TransferTaskConfig>;
+    const store = useTransferTaskStore();
+    const saved = await store.saveTask({ name: "upper", config: makeConfig({ targetColumnNameCase: "upper" }) });
+    vi.mocked(api.loadTransferTaskLibrary).mockResolvedValue({
+      version: 1,
+      folders: [],
+      tasks: [
+        saved,
+        { id: "legacy", name: "legacy", config: legacyConfig, createdAt: "2026-01-01T00:00:00.000Z", updatedAt: "2026-01-01T00:00:00.000Z" },
+      ],
+    } as unknown as TransferTaskLibrary);
+
+    setActivePinia(createPinia());
+    const reloaded = useTransferTaskStore();
+    await reloaded.initFromStorage();
+
+    expect(reloaded.getTask(saved.id)?.config.targetColumnNameCase).toBe("upper");
+    expect(reloaded.getTask("legacy")?.config.targetColumnNameCase).toBe("preserve");
+  });
+
   it("keeps legacy tasks on inherited timeouts and round-trips a custom timeout", async () => {
     const store = useTransferTaskStore();
     const inherited = await store.saveTask({ name: "inherited", config: makeConfig() });

apps/desktop/src/stores/transferTaskStore.ts

@@ -106,6 +106,7 @@ function normalizeTask(raw: unknown): TransferTask | null {
       content: config.content ?? "structureAndData",
       ...transferStrategyOptions(resolveTransferStrategy(config)),
       targetTableNameCase: config.targetTableNameCase ?? "preserve",
+      targetColumnNameCase: config.targetColumnNameCase ?? "preserve",
       quoteTargetColumnNames: config.quoteTargetColumnNames ?? true,
       batchSize: typeof config.batchSize === "number" && config.batchSize > 0 ? config.batchSize : 1000,
       operationTimeoutSecs: normalizeTransferOperationTimeoutSecs(config.operationTimeoutSecs),

apps/desktop/src/types/database.ts

@@ -1,5 +1,5 @@
 import type { BackendError } from "@/lib/backend/errorUtils";
-import type { TransferContent, TransferMode, TransferObjectKind, TransferTableNameCase } from "@/lib/backend/tauri";
+import type { TransferColumnNameCase, TransferContent, TransferMode, TransferObjectKind, TransferTableNameCase } from "@/lib/backend/tauri";
 import type { SqlFormatDialect } from "@/lib/sql/sqlFormatter";
 import type { MultiDbExecutionTarget, MultiDbResultRunExecution } from "@/types/sqlExecution";
 import type { DatabaseType } from "@/types/generated/databaseTypes";
@@ -2621,6 +2621,7 @@ export interface TransferTaskConfig {
   content: TransferContent;
   mode: TransferMode;
   targetTableNameCase: TransferTableNameCase;
+  targetColumnNameCase?: TransferColumnNameCase;
   quoteTargetColumnNames: boolean;
   batchSize: number;
   /** Per-operation timeout for this task. Undefined inherits each connection; 0 disables the limit. */

crates/dbx-core/src/data/transfer.rs

@@ -217,6 +217,27 @@ pub enum TransferTableNameCase {
     Upper,
 }
 
+/// Case folding applied to every target column name (DDL, keys, comments and write SQL).
+/// Source reads keep the original names.
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, Default, PartialEq, Eq)]
+#[serde(rename_all = "camelCase")]
+pub enum TransferColumnNameCase {
+    #[default]
+    Preserve,
+    Lower,
+    Upper,
+}
+
+impl TransferColumnNameCase {
+    pub fn apply(self, name: &str) -> String {
+        match self {
+            Self::Preserve => name.to_string(),
+            Self::Lower => name.to_lowercase(),
+            Self::Upper => name.to_uppercase(),
+        }
+    }
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq)]
 #[serde(rename_all = "camelCase")]
 pub enum TransferOwnershipPolicy {
@@ -313,6 +334,8 @@ pub struct TransferRequest {
     pub mode: TransferMode,
     #[serde(default)]
     pub target_table_name_case: TransferTableNameCase,
+    #[serde(default)]
+    pub target_column_name_case: TransferColumnNameCase,
     #[serde(default = "default_quote_target_column_names")]
     pub quote_target_column_names: bool,
     #[serde(default)]
@@ -517,6 +540,24 @@ impl TransferRequest {
             TransferTableNameCase::Upper => source_table.to_uppercase(),
         }
     }
+
+    pub fn target_column_name(&self, source_column: &str) -> String {
+        self.target_column_name_case.apply(source_column)
+    }
+
+    /// True when target column names equal the source ones, so source-derived DDL and
+    /// index/foreign-key metadata can be reused verbatim.
+    pub fn preserves_column_names(&self) -> bool {
+        self.target_column_name_case == TransferColumnNameCase::Preserve
+    }
+
+    /// Source column metadata renamed to the names the target table will declare.
+    pub fn target_columns(&self, columns: &[db::ColumnInfo]) -> Vec<db::ColumnInfo> {
+        columns
+            .iter()
+            .map(|column| db::ColumnInfo { name: self.target_column_name(&column.name), ..column.clone() })
+            .collect()
+    }
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -1109,6 +1150,9 @@ pub fn validate_transfer_database_pair(
     if request.mode == TransferMode::Upsert || request.drop_target_before_create {
         return Err("虚谷数据传输当前仅支持追加或覆盖，暂不支持更新插入或重建".to_string());
     }
+    if !request.preserves_column_names() {
+        return Err("虚谷数据传输当前要求保留源字段名大小写".to_string());
+    }
     if request.create_table
         && !matches!(request.content, TransferContent::DataOnly)
         && request.target_table_name_case != TransferTableNameCase::Preserve
@@ -1124,6 +1168,28 @@ struct ResolvedTransferTargetTable {
     preexisting: bool,
 }
 
+/// Case folding must not merge two distinct source columns (`a` and `A`) into one target
+/// column, which would otherwise fail later with an opaque duplicate-column DDL error.
+fn ensure_target_column_names_unique(
+    request: &TransferRequest,
+    table: &str,
+    columns: &[db::ColumnInfo],
+) -> Result<(), String> {
+    if request.preserves_column_names() {
+        return Ok(());
+    }
+    let mut seen: HashMap<String, &str> = HashMap::new();
+    for column in columns {
+        if let Some(previous) = seen.insert(request.target_column_name(&column.name), column.name.as_str()) {
+            return Err(format!(
+                "Columns '{previous}' and '{}' of table '{table}' collide after the target column name case is applied",
+                column.name
+            ));
+        }
+    }
+    Ok(())
+}
+
 fn json_scalar_to_string(value: &serde_json::Value) -> Option<String> {
     match value {
         serde_json::Value::String(value) => Some(value.clone()),
@@ -2225,15 +2291,26 @@ fn transfer_column_names_match(
 /// column that really exists; an exact match still wins so case-sensitive
 /// targets that do have the source-cased column keep addressing it.
 fn resolve_transfer_target_column_names(col_names: &[String], target_columns: &[db::ColumnInfo]) -> Vec<String> {
+    resolve_transfer_target_column_names_or(col_names, target_columns, col_names)
+}
+
+/// Like [`resolve_transfer_target_column_names`], but a name the target does not declare
+/// falls back to the matching entry of `fallbacks` (the requested target column case).
+fn resolve_transfer_target_column_names_or(
+    col_names: &[String],
+    target_columns: &[db::ColumnInfo],
+    fallbacks: &[String],
+) -> Vec<String> {
     col_names
         .iter()
-        .map(|name| {
+        .zip(fallbacks)
+        .map(|(name, fallback)| {
             target_columns
                 .iter()
                 .find(|column| column.name == *name)
                 .or_else(|| target_columns.iter().find(|column| column.name.eq_ignore_ascii_case(name)))
                 .map(|column| column.name.clone())
-                .unwrap_or_else(|| name.clone())
+                .unwrap_or_else(|| fallback.clone())
         })
         .collect()
 }
@@ -3056,12 +3133,12 @@ fn generate_mysql_foreign_key_alter_statements(
     for (name, group) in group_foreign_keys_by_constraint_name(foreign_keys) {
         let columns = group
             .iter()
-            .map(|foreign_key| quote_identifier(&foreign_key.column, target_db_type))
+            .map(|foreign_key| quote_identifier(&request.target_column_name(&foreign_key.column), target_db_type))
             .collect::<Vec<_>>()
             .join(", ");
         let ref_columns = group
             .iter()
-            .map(|foreign_key| quote_identifier(&foreign_key.ref_column, target_db_type))
+            .map(|foreign_key| quote_identifier(&request.target_column_name(&foreign_key.ref_column), target_db_type))
             .collect::<Vec<_>>()
             .join(", ");
         let referenced_table = match group[0].ref_schema.as_deref() {
@@ -12277,7 +12354,7 @@ async fn create_transfer_target_table(
     if target_table_created {
         pending_fk_alters.extend(deferred_fk_alters.into_iter().map(|statement| (target_table.to_string(), statement)));
         let comment_stmts = generate_comment_ddl_with_column_quoting(
-            columns,
+            &request.target_columns(columns),
             target_table,
             &request.target_schema,
             target_db_type,
@@ -12370,7 +12447,8 @@ where
             request.target_catalog.as_deref(),
         )
         .await;
-    let preserves_target_table_name = target_table == table;
+    // Column renaming also breaks source-derived index/foreign-key/sequence metadata.
+    let preserves_target_table_name = target_table == table && request.preserves_column_names();
 
     // Did the rename pre-pass move this table's target aside? False when the option is off,
     // or when the target table did not exist, so nothing was renamed. Removing the backup is
@@ -12476,6 +12554,7 @@ where
     if columns.is_empty() {
         return Err(format!("No columns found for table {table}"));
     }
+    ensure_target_column_names_unique(request, table, &columns)?;
 
     let is_doris_source = {
         let configs = state.configs.read().await;
@@ -12503,6 +12582,9 @@ where
     }
 
     let mut col_names: Vec<String> = writable_columns.iter().map(|c| c.name.clone()).collect();
+    // Names the target declares (== `col_names` unless a column name case is requested);
+    // preexisting-target validation and write SQL compare/emit these, reads keep `col_names`.
+    let mut target_col_names: Vec<String> = col_names.iter().map(|name| request.target_column_name(name)).collect();
     let mut col_types: Vec<Option<String>> = writable_columns.iter().map(|c| Some(c.data_type.clone())).collect();
     let primary_key_columns = transfer_key_columns(&writable_columns, source_db_type);
     if should_copy_data(&request.content) {
@@ -12630,7 +12712,7 @@ where
             })?;
             validate_preexisting_target_columns(
                 &target_columns,
-                &col_names,
+                &target_col_names,
                 target_db_type,
                 request.quote_target_column_names,
                 &target_table,
@@ -12708,7 +12790,7 @@ where
     if request.create_table && target_table_preexisting && !request.drop_target_before_create {
         validate_preexisting_target_columns(
             &target_columns,
-            &col_names,
+            &target_col_names,
             target_db_type,
             request.quote_target_column_names,
             &target_table,
@@ -12718,6 +12800,7 @@ where
     if *target_db_type == DatabaseType::H2 {
         let columns = h2_writable_transfer_columns(&writable_columns, &target_columns, &request.mode, &target_table)?;
         col_names = columns.iter().map(|column| column.name.clone()).collect();
+        target_col_names = col_names.iter().map(|name| request.target_column_name(name)).collect();
         col_types = columns
             .iter()
             .map(|column| {
@@ -12730,9 +12813,12 @@ where
     // Read SQL keeps the source names (the source table is untouched); write SQL
     // has to use the names the target table actually declares.
     let write_col_names = if target_table_preexisting && !target_columns.is_empty() {
-        resolve_transfer_target_column_names(&col_names, &target_columns)
+        // A preexisting target keeps its own declared spelling (exact, then
+        // case-insensitive match against the source name); only columns it does not
+        // declare fall back to the requested case.
+        resolve_transfer_target_column_names_or(&col_names, &target_columns, &target_col_names)
     } else {
-        col_names.clone()
+        target_col_names
     };
 
     if *target_db_type == DatabaseType::Db2 {
@@ -13209,11 +13295,12 @@ where
         .await?;
     }
 
+    let target_ddl_columns = request.target_columns(&columns);
     let syncs_postgres_identity = !pg_compat_transfer
         && postgres_identity_target(source_db_type, target_db_type)
         && columns.iter().any(|column| column.is_primary_key && column_is_auto_increment(column));
     if pg_compat_transfer || syncs_postgres_identity {
-        for statement in generate_postgres_sequence_sync_sql(&columns, &target_table, &request.target_schema) {
+        for statement in generate_postgres_sequence_sync_sql(&target_ddl_columns, &target_table, &request.target_schema) {
             execute_on_pool(state, target_pool_key, &statement)
                 .await
                 .map_err(|e| format!("Failed to sync PostgreSQL sequence for {target_table}: {e}"))?;
@@ -13231,7 +13318,7 @@ where
         && total_transferred > 0
     {
         for statement in oracle_identity_reseed_statements(
-            &columns,
+            &target_ddl_columns,
             &target_table,
             &request.target_schema,
             &request.source_schema,
@@ -15221,6 +15308,7 @@ CREATE TABLE "Other"."prefix""Source"."NAME" ("ID" INT);"#;
             }]),
             mode: TransferMode::Append,
             target_table_name_case: TransferTableNameCase::Preserve,
+            target_column_name_case: TransferColumnNameCase::Preserve,
             quote_target_column_names: true,
             ownership_policy: TransferOwnershipPolicy::Preserve,
             batch_size: 1000,
@@ -15308,6 +15396,7 @@ CREATE TABLE "Other"."prefix""Source"."NAME" ("ID" INT);"#;
                 objects: None,
                 mode: TransferMode::Append,
                 target_table_name_case: TransferTableNameCase::Preserve,
+                target_column_name_case: TransferColumnNameCase::Preserve,
                 quote_target_column_names: true,
                 ownership_policy: TransferOwnershipPolicy::Preserve,
                 batch_size: 1000,
@@ -15410,6 +15499,7 @@ CREATE TABLE "Other"."prefix""Source"."NAME" ("ID" INT);"#;
                 create_table: true,
                 mode: TransferMode::Append,
                 target_table_name_case: TransferTableNameCase::Preserve,
+                target_column_name_case: TransferColumnNameCase::Preserve,
                 quote_target_column_names: true,
                 ownership_policy: TransferOwnershipPolicy::Preserve,
                 batch_size: 1000,
@@ -15451,6 +15541,7 @@ CREATE TABLE "Other"."prefix""Source"."NAME" ("ID" INT);"#;
                 create_table: true,
                 mode: TransferMode::Append,
                 target_table_name_case: TransferTableNameCase::Preserve,
+                target_column_name_case: TransferColumnNameCase::Preserve,
                 quote_target_column_names: true,
                 ownership_policy: TransferOwnershipPolicy::Preserve,
                 batch_size: 1000,
@@ -16779,6 +16870,7 @@ CREATE TABLE "Other"."prefix""Source"."NAME" ("ID" INT);"#;
             objects: Some(Vec::new()),
             mode: TransferMode::Append,
             target_table_name_case: TransferTableNameCase::Preserve,
+            target_column_name_case: TransferColumnNameCase::Preserve,
             quote_target_column_names: true,
             ownership_policy: TransferOwnershipPolicy::Preserve,
             batch_size: 1000,
@@ -16881,6 +16973,119 @@ CREATE TABLE "Other"."prefix""Source"."NAME" ("ID" INT);"#;
         assert_eq!(request.target_table_name("orders"), "ORDERS");
     }
 
+    #[test]
+    fn transfer_request_defaults_preserve_column_name_case() {
+        let mut value = serde_json::to_value(test_transfer_request(vec!["ORDERS"])).unwrap();
+        value.as_object_mut().unwrap().remove("targetColumnNameCase");
+        let request: TransferRequest = serde_json::from_value(value).unwrap();
+        assert_eq!(request.target_column_name_case, TransferColumnNameCase::Preserve);
+        assert!(request.preserves_column_names());
+
+        let mut value = serde_json::to_value(test_transfer_request(vec!["ORDERS"])).unwrap();
+        value["targetColumnNameCase"] = serde_json::json!("upper");
+        let request: TransferRequest = serde_json::from_value(value).unwrap();
+        assert_eq!(request.target_column_name("updatedBy"), "UPDATEDBY");
+    }
+
+    #[test]
+    fn transfer_column_name_case_applies_to_ddl_and_write_sql_consistently() {
+        let mut request = test_transfer_request(vec!["t"]);
+        let columns = vec![
+            db::ColumnInfo { is_primary_key: true, is_nullable: false, ..test_column("id", "bigint") },
+            test_column("updatedby", "varchar(300)"),
+            test_column("updatedtime", "datetime(6)"),
+        ];
+
+        // Disabled: byte-identical to generating straight from the source columns.
+        assert!(request.preserves_column_names());
+        let unchanged: Vec<String> = request.target_columns(&columns).into_iter().map(|c| c.name).collect();
+        assert_eq!(unchanged, vec!["id", "updatedby", "updatedtime"]);
+
+        request.target_column_name_case = TransferColumnNameCase::Upper;
+        let target_columns = request.target_columns(&columns);
+        for target in [DatabaseType::Oracle, DatabaseType::OceanbaseOracle] {
+            let ddl = generate_create_table_ddl(
+                &target_columns,
+                "T",
+                "",
+                "APP",
+                &target,
+                &DatabaseType::Mysql,
+                None,
+                None,
+            );
+            assert!(ddl.contains("\"UPDATEDTIME\" TIMESTAMP"), "{ddl}");
+            assert!(ddl.contains("\"UPDATEDBY\" VARCHAR(300"), "{ddl}");
+            assert!(ddl.contains("PRIMARY KEY (\"ID\")"), "{ddl}");
+            assert!(!ddl.contains("\"updatedby\"") && !ddl.contains("DATETIME"), "{ddl}");
+
+            let write_names: Vec<String> = columns.iter().map(|c| request.target_column_name(&c.name)).collect();
+            let sql = generate_transfer_write_sql_batches_with_column_quoting(
+                &TransferMode::Append,
+                &write_names,
+                &[None, None, None],
+                &[vec![serde_json::json!(1), serde_json::json!("a"), serde_json::json!("2024-01-02 03:04:05")]],
+                "T",
+                "APP",
+                &target,
+                &[],
+                None,
+                false,
+                false,
+                true,
+                None,
+            )
+            .unwrap()
+            .join("\n");
+            assert!(sql.contains("(\"ID\", \"UPDATEDBY\", \"UPDATEDTIME\")"), "{sql}");
+        }
+
+        request.target_column_name_case = TransferColumnNameCase::Lower;
+        assert_eq!(request.target_column_name("UpdatedBy"), "updatedby");
+    }
+
+    #[test]
+    fn transfer_column_name_case_resolves_preexisting_and_missing_target_columns() {
+        let target = vec![test_column("ID", "NUMBER"), test_column("Name", "VARCHAR2(10)")];
+        let names = vec!["id".to_string(), "name".to_string(), "extra".to_string()];
+        let upper: Vec<String> = names.iter().map(|n| n.to_uppercase()).collect();
+        // Declared columns keep their own spelling; unknown ones fall back to the requested case.
+        assert_eq!(
+            resolve_transfer_target_column_names_or(&names, &target, &upper),
+            vec!["ID".to_string(), "Name".to_string(), "EXTRA".to_string()]
+        );
+        assert_eq!(resolve_transfer_target_column_names(&names, &target)[2], "extra");
+    }
+
+    #[test]
+    fn transfer_column_name_case_rejects_colliding_columns_and_xugu_pairs() {
+        let mut request = test_transfer_request(vec!["t"]);
+        let columns = vec![test_column("a", "int"), test_column("A", "int")];
+        assert!(ensure_target_column_names_unique(&request, "t", &columns).is_ok());
+        request.target_column_name_case = TransferColumnNameCase::Upper;
+        let error = ensure_target_column_names_unique(&request, "t", &columns).unwrap_err();
+        assert!(error.contains("collide"), "{error}");
+        assert!(ensure_target_column_names_unique(&request, "t", &columns[..1]).is_ok());
+    }
+
+    #[test]
+    fn transfer_column_name_case_applies_to_deferred_foreign_keys() {
+        let mut request = test_transfer_request(vec!["child"]);
+        request.target_column_name_case = TransferColumnNameCase::Upper;
+        let foreign_keys = vec![db::ForeignKeyInfo {
+            name: "fk_parent".into(),
+            column: "parent_id".into(),
+            ref_table: "parent".into(),
+            ref_column: "id".into(),
+            ref_schema: None,
+            on_update: None,
+            on_delete: None,
+        }];
+        let statements =
+            generate_mysql_foreign_key_alter_statements(&foreign_keys, &request, "child", &DatabaseType::Mysql);
+        assert!(statements[0].contains("FOREIGN KEY (`PARENT_ID`) REFERENCES `parent` (`ID`)"), "{}", statements[0]);
+    }
+
     #[test]
     fn transfer_table_name_case_detects_target_collisions() {
         let mut request = test_transfer_request(vec!["ORDERS", "orders"]);

crates/dbx-core/src/data/transfer/ddl_plan.rs

@@ -34,6 +34,9 @@ pub(super) async fn prepare_table_ddl(
     known_foreign_keys: &HashMap<String, Vec<db::ForeignKeyInfo>>,
 ) -> Result<PreparedTableDdl, String> {
     let rebuild = request.drop_target_before_create;
+    // Every column reference below (CREATE TABLE, keys) uses the target-side names.
+    let target_columns = request.target_columns(columns);
+    let columns = target_columns.as_slice();
 
     let foreign_keys = if let Some(foreign_keys) = known_foreign_keys.get(table) {
         foreign_keys.clone()
@@ -84,7 +87,7 @@ pub(super) async fn prepare_table_ddl(
         target_db_type,
         source_driver_profile.as_deref(),
         target_driver_profile.as_deref(),
-        target_table == table,
+        target_table == table && request.preserves_column_names(),
     ) && (request.quote_target_column_names
         || !matches!(target_db_type, DatabaseType::Gaussdb | DatabaseType::OpenGauss));
 

crates/dbx-core/src/data/transfer/overwrite_clear.rs

@@ -186,6 +186,7 @@ mod tests {
             objects: None,
             mode: TransferMode::Overwrite,
             target_table_name_case: TransferTableNameCase::Preserve,
+            target_column_name_case: TransferColumnNameCase::Preserve,
             quote_target_column_names: true,
             ownership_policy: TransferOwnershipPolicy::Preserve,
             batch_size: 1000,

crates/dbx-core/src/data/transfer/structure_plan.rs

@@ -367,7 +367,7 @@ async fn plan_table_structure(
     pg_compat_transfer: bool,
     index_if_not_exists: bool,
 ) -> Result<PlannedTableStructure, String> {
-    let preserves_target_table_name = target_table == table;
+    let preserves_target_table_name = target_table == table && request.preserves_column_names();
     let columns = source_transfer_columns(state, request, table, source_pool_key).await?;
     let table_comment = source_table_comment(state, request, table, source_db_type).await;
 
@@ -426,15 +426,16 @@ async fn plan_table_structure(
     // metadata below. A statement the create pass filters out must never appear here.
     statements.extend(transfer_ddl_statements(&ddl, target_db_type));
 
+    let target_columns = request.target_columns(&columns);
     let comment_statements = generate_comment_ddl_with_column_quoting(
-        &columns,
+        &target_columns,
         target_table,
         &request.target_schema,
         target_db_type,
         table_comment.as_deref(),
         request.quote_target_column_names,
     );
-    operations.extend(comment_operations(&columns, table, target_table, target_db_type, table_comment.as_deref()));
+    operations.extend(comment_operations(&target_columns, table, target_table, target_db_type, table_comment.as_deref()));
     statements.extend(comment_statements);
 
     for sequence in &owned_sequences.owned_sequences {

crates/dbx-core/src/persistence/task_history.rs

@@ -932,7 +932,7 @@ fn to_sqlite_count(value: u64) -> Option<i64> {
 mod tests {
     use super::*;
     use crate::data::transfer::{
-        TransferMode, TransferObjectSelection, TransferOwnershipPolicy, TransferTableNameCase,
+        TransferMode, TransferObjectSelection, TransferOwnershipPolicy, TransferColumnNameCase, TransferTableNameCase,
     };
 
     #[tokio::test]
@@ -959,6 +959,7 @@ mod tests {
             objects: Some(vec![TransferObjectSelection::default()]),
             mode: TransferMode::Append,
             target_table_name_case: TransferTableNameCase::Preserve,
+            target_column_name_case: TransferColumnNameCase::Preserve,
             quote_target_column_names: true,
             ownership_policy: TransferOwnershipPolicy::Preserve,
             batch_size: 500,

crates/dbx-core/tests/live_mysql_transfer.rs

@@ -3,7 +3,7 @@ use dbx_core::db::mysql;
 use dbx_core::models::connection::{ConnectionConfig, DatabaseType};
 use dbx_core::transfer::{
     drop_backup_tables, execute_on_pool, rename_tables_to_backup, sort_tables_by_fk_dependency_with_foreign_keys,
-    transfer_table, TransferContent, TransferMode, TransferOwnershipPolicy, TransferRequest, TransferTableNameCase,
+    transfer_table, TransferContent, TransferMode, TransferOwnershipPolicy, TransferRequest, TransferColumnNameCase, TransferTableNameCase,
 };
 use serde_json::json;
 use std::sync::Arc;
@@ -92,6 +92,7 @@ fn transfer_request(
         objects: Some(Vec::new()),
         mode,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 2,
@@ -199,6 +200,7 @@ async fn run_live_mysql_cross_version_transfer_completes_on_small_stack() {
         objects: Some(Vec::new()),
         mode: TransferMode::Append,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 100,
@@ -355,6 +357,7 @@ async fn live_mysql_transfer_keeps_columns_whose_comment_mentions_foreign_key()
         objects: Some(Vec::new()),
         mode: TransferMode::Append,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 10,
@@ -515,6 +518,7 @@ async fn live_mysql_transfer_downgrades_unsupported_source_collations() {
         objects: Some(Vec::new()),
         mode: TransferMode::Append,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 10,
@@ -580,6 +584,7 @@ async fn live_mysql_transfer_downgrades_unsupported_source_collations() {
             objects: Some(Vec::new()),
             mode: TransferMode::Append,
             target_table_name_case: TransferTableNameCase::Preserve,
+            target_column_name_case: TransferColumnNameCase::Preserve,
             quote_target_column_names: true,
             ownership_policy: TransferOwnershipPolicy::Preserve,
             batch_size: 10,
@@ -888,6 +893,7 @@ async fn live_mysql_transfer_structure_overwrite_rejects_incompatible_target_col
         objects: Some(Vec::new()),
         mode: TransferMode::Overwrite,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 10,
@@ -945,6 +951,7 @@ async fn live_mysql_transfer_structure_overwrite_rejects_incompatible_target_col
             objects: Some(Vec::new()),
             mode: TransferMode::Overwrite,
             target_table_name_case: TransferTableNameCase::Preserve,
+            target_column_name_case: TransferColumnNameCase::Preserve,
             quote_target_column_names: true,
             ownership_policy: TransferOwnershipPolicy::Preserve,
             batch_size: 10,
@@ -1043,6 +1050,7 @@ async fn live_mysql_transfer_structure_only_rejects_incompatible_target_columns(
         objects: Some(Vec::new()),
         mode: TransferMode::Append,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 10,
@@ -1143,6 +1151,7 @@ async fn live_mysql_transfer_drop_target_parent_child_foreign_key() {
         objects: Some(Vec::new()),
         mode: TransferMode::Append,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 10,
@@ -1337,6 +1346,7 @@ async fn live_mysql_transfer_drop_target_rebuilds_incompatible_structure() {
         objects: Some(Vec::new()),
         mode: TransferMode::Append,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 10,
@@ -1526,6 +1536,7 @@ async fn live_mysql_transfer_drop_target_rejects_external_incoming_fk() {
         objects: Some(Vec::new()),
         mode: TransferMode::Append,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 10,
@@ -1618,6 +1629,7 @@ async fn live_mysql_transfer_drop_target_circular_foreign_keys() {
         objects: Some(Vec::new()),
         mode: TransferMode::Append,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 10,
@@ -1848,6 +1860,7 @@ async fn live_mysql_transfer_drop_target_retains_backup_on_failure() {
         objects: Some(Vec::new()),
         mode: TransferMode::Append,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 10,
@@ -2027,6 +2040,7 @@ async fn live_mysql_keyset_pagination_copies_every_row() {
         objects: Some(Vec::new()),
         mode: TransferMode::Append,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 3,
@@ -2152,6 +2166,7 @@ async fn live_mysql_progress_read_survives_total_duration_beyond_timeout() {
         objects: Some(Vec::new()),
         mode: TransferMode::Append,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 10000,
@@ -2253,6 +2268,7 @@ async fn live_mysql_transfer_drop_target_inspects_dependent_views_without_the_us
         objects: Some(Vec::new()),
         mode: TransferMode::Append,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 10,

crates/dbx-core/tests/live_oracle_transfer.rs

@@ -3,7 +3,7 @@ use dbx_core::models::connection::{ConnectionConfig, DatabaseType};
 use dbx_core::query::execute_sql_statement;
 use dbx_core::transfer::{
     ensure_transfer_pool, transfer_schema_objects, transfer_table, TransferContent, TransferMode, TransferObjectKind,
-    TransferObjectSelection, TransferOwnershipPolicy, TransferRequest, TransferTableNameCase,
+    TransferObjectSelection, TransferOwnershipPolicy, TransferRequest, TransferColumnNameCase, TransferTableNameCase,
 };
 use std::sync::Arc;
 
@@ -113,6 +113,7 @@ async fn live_oracle_transfer_creates_table_with_index_comment_and_trigger() {
         ]),
         mode: TransferMode::Append,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 100,

crates/dbx-core/tests/live_postgres_transfer.rs

@@ -5,7 +5,7 @@ use dbx_core::transfer::{
     drop_backup_tables, get_db_type, preview_transfer_ownership, rename_tables_to_backup,
     transfer_postgres_schema_dependencies, transfer_postgres_schema_objects, transfer_table, TransferContent,
     TransferMode, TransferObjectKind, TransferObjectSelection, TransferOwnershipPolicy, TransferRequest,
-    TransferTableNameCase,
+    TransferColumnNameCase, TransferTableNameCase,
 };
 use serde_json::json;
 use std::sync::Arc;
@@ -266,6 +266,7 @@ async fn live_postgres_transfer_upserts_generated_always_identity_values() {
         objects: Some(Vec::new()),
         mode: TransferMode::Upsert,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 100,
@@ -444,6 +445,7 @@ async fn live_postgres_structure_only_preserves_table_indexes() {
         objects: Some(Vec::new()),
         mode: TransferMode::Append,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 100,
@@ -729,6 +731,7 @@ async fn live_postgres_transfer_preserves_data_and_schema_objects() {
         objects: None,
         mode: TransferMode::Append,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 100,
@@ -1032,6 +1035,7 @@ async fn live_postgres_transfer_skips_create_ddl_for_existing_target_table() {
         objects: Some(Vec::new()),
         mode: TransferMode::Append,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 100,
@@ -1163,6 +1167,7 @@ async fn live_postgres_transfer_creates_selected_sequence_before_referencing_tab
         ]),
         mode: TransferMode::Append,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 100,
@@ -1349,6 +1354,7 @@ async fn live_postgres_transfer_drop_target_rebuilds_structure_and_indexes() {
         objects: Some(Vec::new()),
         mode: TransferMode::Append,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 1000,
@@ -1551,6 +1557,7 @@ impl PostgresRebuildFixture {
                 objects: Some(Vec::new()),
                 mode: TransferMode::Append,
                 target_table_name_case: TransferTableNameCase::Preserve,
+                target_column_name_case: TransferColumnNameCase::Preserve,
                 quote_target_column_names: true,
                 ownership_policy: TransferOwnershipPolicy::Preserve,
                 batch_size: 100,
@@ -1875,6 +1882,7 @@ async fn live_postgres_keyset_pagination_copies_every_row() {
         objects: Some(Vec::new()),
         mode: TransferMode::Append,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 3,
@@ -1988,6 +1996,7 @@ async fn live_postgres_progress_read_survives_total_duration_beyond_timeout() {
         objects: Some(Vec::new()),
         mode: TransferMode::Append,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 10000,
@@ -2095,6 +2104,7 @@ async fn live_postgres_keyset_large_batch_copies_every_row() {
         objects: Some(Vec::new()),
         mode: TransferMode::Append,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 50000,
@@ -2212,6 +2222,7 @@ async fn live_postgres_structure_only_preview_renders_ddl_without_touching_the_t
         }]),
         mode: TransferMode::Append,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 1000,

crates/dbx-core/tests/live_sqlite_to_sqlserver_transfer.rs

@@ -8,7 +8,7 @@
 use dbx_core::connection::AppState;
 use dbx_core::models::connection::{ConnectionConfig, DatabaseType};
 use dbx_core::transfer::{
-    transfer_table, TransferContent, TransferMode, TransferOwnershipPolicy, TransferRequest, TransferTableNameCase,
+    transfer_table, TransferContent, TransferMode, TransferOwnershipPolicy, TransferRequest, TransferColumnNameCase, TransferTableNameCase,
 };
 use std::collections::HashMap;
 use std::sync::Arc;
@@ -177,6 +177,7 @@ async fn live_sqlite_to_sqlserver_keeps_sixty_four_bit_integer_values() {
         objects: Some(Vec::new()),
         mode: TransferMode::Append,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 100,

crates/dbx-core/tests/live_sqlserver_completion.rs

@@ -1425,6 +1425,7 @@ async fn live_sqlserver_transfer_table_skips_rowversion_insert_column() {
         objects: Some(Vec::new()),
         mode: dbx_core::transfer::TransferMode::Append,
         target_table_name_case: dbx_core::transfer::TransferTableNameCase::Upper,
+        target_column_name_case: dbx_core::transfer::TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: dbx_core::transfer::TransferOwnershipPolicy::Preserve,
         batch_size: 100,

crates/dbx-core/tests/live_sqlserver_transfer.rs

@@ -3,7 +3,7 @@ use dbx_core::models::connection::{ConnectionConfig, DatabaseType};
 use dbx_core::transfer::{
     clear_foreign_key_linked_overwrite_targets, drop_backup_tables, rename_tables_to_backup,
     sort_tables_by_fk_dependency_with_foreign_keys, transfer_table, transfer_table_with_result, TransferContent,
-    TransferMode, TransferOwnershipPolicy, TransferRequest, TransferTableNameCase,
+    TransferMode, TransferOwnershipPolicy, TransferRequest, TransferColumnNameCase, TransferTableNameCase,
 };
 use std::collections::HashMap;
 use std::sync::Arc;
@@ -190,6 +190,7 @@ async fn live_sqlserver_transfer_rebuild_releases_constraint_and_index_names() {
         objects: Some(Vec::new()),
         mode: TransferMode::Append,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 10,
@@ -352,6 +353,7 @@ async fn live_sqlserver_transfer_overwrite_handles_existing_identity_target() {
         objects: Some(Vec::new()),
         mode: TransferMode::Overwrite,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 1,
@@ -470,6 +472,7 @@ async fn live_sqlserver_keyset_pagination_copies_every_row() {
         objects: Some(Vec::new()),
         mode: TransferMode::Append,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 3,
@@ -592,6 +595,7 @@ async fn live_sqlserver_progress_read_survives_total_duration_beyond_timeout() {
         objects: Some(Vec::new()),
         mode: TransferMode::Append,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 10000,
@@ -712,6 +716,7 @@ async fn live_sqlserver_keyset_uniqueidentifier_datetime2_composite_key() {
         objects: Some(Vec::new()),
         mode: TransferMode::Append,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 3,
@@ -825,6 +830,7 @@ async fn live_sqlserver_transfer_new_identity_target_keeps_explicit_identity_val
         objects: Some(Vec::new()),
         mode: TransferMode::Append,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 1,
@@ -1070,6 +1076,7 @@ async fn live_sqlserver_transfer_overwrite_replaces_tables_linked_by_foreign_key
         objects: Some(Vec::new()),
         mode: TransferMode::Overwrite,
         target_table_name_case: TransferTableNameCase::Preserve,
+        target_column_name_case: TransferColumnNameCase::Preserve,
         quote_target_column_names: true,
         ownership_policy: TransferOwnershipPolicy::Preserve,
         batch_size: 1000,

crates/dbx-web/src/routes/transfer.rs

@@ -320,7 +320,7 @@ mod tests {
     };
     use dbx_core::persistence::task_history::{TaskRunItemsQuery, TaskRunStatus, TransferObjectSelectionMode};
     use dbx_core::transfer::{
-        TransferContent, TransferMode, TransferOwnershipPolicy, TransferRequest, TransferTableNameCase,
+        TransferContent, TransferMode, TransferOwnershipPolicy, TransferRequest, TransferColumnNameCase, TransferTableNameCase,
     };
 
     fn sqlite_config(id: &str, path: &str) -> ConnectionConfig {
@@ -428,6 +428,7 @@ mod tests {
             objects: Some(Vec::new()),
             mode: TransferMode::Append,
             target_table_name_case: TransferTableNameCase::Preserve,
+            target_column_name_case: TransferColumnNameCase::Preserve,
             quote_target_column_names: true,
             ownership_policy: TransferOwnershipPolicy::Preserve,
             batch_size: 1000,