✨ feat(import-export): 优化数据表导入导出能力

- 修复表格格式导出时空值显示为 null 的问题
- 支持按列导出并保持字段顺序
- 支持导入列与数据库字段映射及冲突校验
- 隔离并发导入任务进度并完善元数据回退

Fixes #646

internal/app/export_options.go

@@ -1,6 +1,9 @@
 package app
 
-import "strings"
+import (
+	"errors"
+	"strings"
+)
 
 const (
 	maxXLSXRowsPerSheet     = 1048575
@@ -9,6 +12,7 @@ const (
 
 type ExportFileOptions struct {
 	Format                         string            `json:"format"`
+	Columns                        []string          `json:"columns,omitempty"`
 	XLSXMaxRowsPerSheet            int               `json:"xlsxMaxRowsPerSheet,omitempty"`
 	JobID                          string            `json:"jobId,omitempty"`
 	TotalRowsHint                  int64             `json:"totalRowsHint,omitempty"`
@@ -27,6 +31,7 @@ func normalizeExportFileOptions(format string, options ExportFileOptions) Export
 	}
 	return ExportFileOptions{
 		Format:                         resolvedFormat,
+		Columns:                        normalizeExportColumns(options.Columns),
 		XLSXMaxRowsPerSheet:            normalizeXLSXRowsPerSheet(options.XLSXMaxRowsPerSheet),
 		JobID:                          strings.TrimSpace(options.JobID),
 		TotalRowsHint:                  normalizeExportTotalRowsHint(options.TotalRowsHint, options.TotalRowsKnown),
@@ -39,6 +44,32 @@ func normalizeExportFileOptions(format string, options ExportFileOptions) Export
 	}
 }
 
+func normalizeExportColumns(columns []string) []string {
+	if columns == nil {
+		return nil
+	}
+	result := make([]string, 0, len(columns))
+	seen := make(map[string]struct{}, len(columns))
+	for _, column := range columns {
+		if strings.TrimSpace(column) == "" {
+			continue
+		}
+		if _, exists := seen[column]; exists {
+			continue
+		}
+		seen[column] = struct{}{}
+		result = append(result, column)
+	}
+	return result
+}
+
+func validateExportColumnsSelection(options ExportFileOptions) error {
+	if options.Columns != nil && len(options.Columns) == 0 {
+		return errors.New("at least one export column must be selected")
+	}
+	return nil
+}
+
 func normalizeXLSXRowsPerSheet(value int) int {
 	if value <= 0 {
 		return defaultXLSXRowsPerSheet

internal/app/import_pipeline.go

@@ -30,12 +30,22 @@ type importPreviewData struct {
 	PreviewRows []map[string]interface{}
 }
 
+// ImportFileOptions controls how a selected import file is applied to the target table.
+// A nil ColumnMappings value preserves the legacy behavior where file headers are used
+// directly as database column names. A non-nil map enables explicit source-to-target
+// mapping; entries with an empty target are skipped.
+type ImportFileOptions struct {
+	ColumnMappings map[string]string `json:"columnMappings,omitempty"`
+	JobID          string            `json:"jobId,omitempty"`
+}
+
 type importProgressState struct {
-	Current        int  `json:"current"`
-	Total          int  `json:"total,omitempty"`
-	Success        int  `json:"success"`
-	Errors         int  `json:"errors"`
-	TotalRowsKnown bool `json:"totalRowsKnown,omitempty"`
+	JobID          string `json:"jobId,omitempty"`
+	Current        int    `json:"current"`
+	Total          int    `json:"total,omitempty"`
+	Success        int    `json:"success"`
+	Errors         int    `json:"errors"`
+	TotalRowsKnown bool   `json:"totalRowsKnown,omitempty"`
 }
 
 type importExecutionResult struct {
@@ -95,6 +105,135 @@ func (c *importCollectConsumer) ConsumeRow(row map[string]interface{}) error {
 	return nil
 }
 
+type importResolvedColumnMapping struct {
+	source string
+	target string
+}
+
+type importColumnMappingConsumer struct {
+	downstream       importFileConsumer
+	targetBySource   map[string]string
+	selectedSources  []string
+	resolvedMappings []importResolvedColumnMapping
+}
+
+func newImportColumnMappingConsumer(
+	downstream importFileConsumer,
+	columnMappings map[string]string,
+	targetColumns []connection.ColumnDefinition,
+) (importFileConsumer, error) {
+	if columnMappings == nil {
+		return downstream, nil
+	}
+	if downstream == nil {
+		return nil, fmt.Errorf("导入字段映射缺少下游处理器")
+	}
+
+	targetColumnsByExactName := make(map[string]string, len(targetColumns))
+	targetColumnsByFoldedName := make(map[string][]string, len(targetColumns))
+	for _, column := range targetColumns {
+		name := column.Name
+		if strings.TrimSpace(name) == "" {
+			continue
+		}
+		targetColumnsByExactName[name] = name
+		foldedName := normalizeColumnName(name)
+		targetColumnsByFoldedName[foldedName] = append(targetColumnsByFoldedName[foldedName], name)
+	}
+
+	sources := make([]string, 0, len(columnMappings))
+	for source := range columnMappings {
+		sources = append(sources, source)
+	}
+	sort.Strings(sources)
+
+	targetBySource := make(map[string]string, len(columnMappings))
+	selectedSources := make([]string, 0, len(columnMappings))
+	usedTargets := make(map[string]string, len(columnMappings))
+	for _, source := range sources {
+		if strings.TrimSpace(source) == "" {
+			return nil, fmt.Errorf("导入字段映射源字段不能为空")
+		}
+		requestedTarget := columnMappings[source]
+		if strings.TrimSpace(requestedTarget) == "" {
+			continue
+		}
+
+		actualTarget, exactMatch := targetColumnsByExactName[requestedTarget]
+		if !exactMatch {
+			foldedMatches := targetColumnsByFoldedName[normalizeColumnName(requestedTarget)]
+			switch len(foldedMatches) {
+			case 0:
+				return nil, fmt.Errorf("导入字段映射目标字段 %q 不存在", requestedTarget)
+			case 1:
+				actualTarget = foldedMatches[0]
+			default:
+				return nil, fmt.Errorf("导入字段映射目标字段 %q 的大小写匹配不明确", requestedTarget)
+			}
+		}
+		if previousSource, exists := usedTargets[actualTarget]; exists {
+			return nil, fmt.Errorf("导入字段映射目标字段 %q 被源字段 %q 和 %q 重复使用", actualTarget, previousSource, source)
+		}
+		usedTargets[actualTarget] = source
+		targetBySource[source] = actualTarget
+		selectedSources = append(selectedSources, source)
+	}
+	if len(selectedSources) == 0 {
+		return nil, fmt.Errorf("导入字段映射至少需要选择一个目标字段")
+	}
+
+	return &importColumnMappingConsumer{
+		downstream:      downstream,
+		targetBySource:  targetBySource,
+		selectedSources: selectedSources,
+	}, nil
+}
+
+func (c *importColumnMappingConsumer) SetColumns(columns []string) error {
+	if c == nil || c.downstream == nil {
+		return fmt.Errorf("导入字段映射缺少下游处理器")
+	}
+
+	foundSources := make(map[string]struct{}, len(c.selectedSources))
+	resolved := make([]importResolvedColumnMapping, 0, len(c.selectedSources))
+	targets := make([]string, 0, len(c.selectedSources))
+	for _, source := range columns {
+		target, selected := c.targetBySource[source]
+		if !selected {
+			continue
+		}
+		if _, duplicate := foundSources[source]; duplicate {
+			return fmt.Errorf("导入字段映射源字段 %q 在文件表头中重复", source)
+		}
+		foundSources[source] = struct{}{}
+		resolved = append(resolved, importResolvedColumnMapping{source: source, target: target})
+		targets = append(targets, target)
+	}
+	for _, source := range c.selectedSources {
+		if _, ok := foundSources[source]; !ok {
+			return fmt.Errorf("导入字段映射源字段 %q 不存在", source)
+		}
+	}
+
+	c.resolvedMappings = resolved
+	return c.downstream.SetColumns(targets)
+}
+
+func (c *importColumnMappingConsumer) ConsumeRow(row map[string]interface{}) error {
+	if c == nil || c.downstream == nil {
+		return fmt.Errorf("导入字段映射缺少下游处理器")
+	}
+	if len(c.resolvedMappings) == 0 {
+		return fmt.Errorf("导入字段映射尚未解析文件表头")
+	}
+
+	mappedRow := make(map[string]interface{}, len(c.resolvedMappings))
+	for _, mapping := range c.resolvedMappings {
+		mappedRow[mapping.target] = row[mapping.source]
+	}
+	return c.downstream.ConsumeRow(mappedRow)
+}
+
 type importRowWriter interface {
 	SetColumns(columns []string)
 	ApplyBatch(rows []map[string]interface{}) error
@@ -102,21 +241,56 @@ type importRowWriter interface {
 	BatchEnabled() bool
 }
 
-type importDatabaseRowWriter struct {
-	dbInst        db.Database
-	applier       db.BatchApplier
-	dbType        string
-	tableName     string
-	columns       []string
-	columnTypeMap map[string]string
+type importColumnTypeLookup struct {
+	byExactName  map[string]string
+	byFoldedName map[string][]string
 }
 
-func newImportDatabaseRowWriter(dbInst db.Database, dbType, tableName string, columnTypeMap map[string]string) *importDatabaseRowWriter {
+func newImportColumnTypeLookup(columns []connection.ColumnDefinition) importColumnTypeLookup {
+	lookup := importColumnTypeLookup{
+		byExactName:  make(map[string]string, len(columns)),
+		byFoldedName: make(map[string][]string, len(columns)),
+	}
+	for _, column := range columns {
+		name := column.Name
+		if strings.TrimSpace(name) == "" {
+			continue
+		}
+		if _, exists := lookup.byExactName[name]; !exists {
+			foldedName := normalizeColumnName(name)
+			lookup.byFoldedName[foldedName] = append(lookup.byFoldedName[foldedName], name)
+		}
+		lookup.byExactName[name] = strings.TrimSpace(column.Type)
+	}
+	return lookup
+}
+
+func (l importColumnTypeLookup) Resolve(columnName string) string {
+	if columnType, ok := l.byExactName[columnName]; ok {
+		return columnType
+	}
+	foldedMatches := l.byFoldedName[normalizeColumnName(columnName)]
+	if len(foldedMatches) != 1 {
+		return ""
+	}
+	return l.byExactName[foldedMatches[0]]
+}
+
+type importDatabaseRowWriter struct {
+	dbInst      db.Database
+	applier     db.BatchApplier
+	dbType      string
+	tableName   string
+	columns     []string
+	columnTypes importColumnTypeLookup
+}
+
+func newImportDatabaseRowWriter(dbInst db.Database, dbType, tableName string, columnTypes importColumnTypeLookup) *importDatabaseRowWriter {
 	writer := &importDatabaseRowWriter{
-		dbInst:        dbInst,
-		dbType:        dbType,
-		tableName:     tableName,
-		columnTypeMap: columnTypeMap,
+		dbInst:      dbInst,
+		dbType:      dbType,
+		tableName:   tableName,
+		columnTypes: columnTypes,
 	}
 	if applier, ok := dbInst.(db.BatchApplier); ok {
 		writer.applier = applier
@@ -143,7 +317,7 @@ func (w *importDatabaseRowWriter) ApplyOne(row map[string]interface{}) error {
 	if w.applier != nil {
 		return w.applier.ApplyChanges(w.tableName, connection.ChangeSet{Inserts: []map[string]interface{}{cloneImportRow(row)}})
 	}
-	query, err := buildImportInsertQuery(w.dbType, w.tableName, w.columns, row, w.columnTypeMap)
+	query, err := buildImportInsertQuery(w.dbType, w.tableName, w.columns, row, w.columnTypes)
 	if err != nil {
 		return err
 	}
@@ -153,6 +327,7 @@ func (w *importDatabaseRowWriter) ApplyOne(row map[string]interface{}) error {
 
 type importBatchConsumer struct {
 	writer         importRowWriter
+	jobID          string
 	batchSize      int
 	totalRows      int
 	totalRowsKnown bool
@@ -244,6 +419,7 @@ func (c *importBatchConsumer) emitProgress(current int) {
 		return
 	}
 	c.report(importProgressState{
+		JobID:          c.jobID,
 		Current:        current,
 		Total:          c.totalRows,
 		Success:        c.successCount,
@@ -362,7 +538,7 @@ func streamCSVImportFile(filePath string, consumer importFileConsumer) error {
 	}
 }
 
-func buildImportInsertQuery(dbType, tableName string, columns []string, row map[string]interface{}, columnTypeMap map[string]string) (string, error) {
+func buildImportInsertQuery(dbType, tableName string, columns []string, row map[string]interface{}, columnTypes importColumnTypeLookup) (string, error) {
 	quotedCols := make([]string, 0, len(columns))
 	values := make([]string, 0, len(columns))
 	for _, column := range columns {
@@ -370,7 +546,7 @@ func buildImportInsertQuery(dbType, tableName string, columns []string, row map[
 			continue
 		}
 		quotedCols = append(quotedCols, quoteIdentByType(dbType, column))
-		colType := columnTypeMap[normalizeColumnName(column)]
+		colType := columnTypes.Resolve(column)
 		values = append(values, formatImportSQLValue(dbType, colType, row[column]))
 	}
 	if len(quotedCols) == 0 {

internal/app/methods_db.go

@@ -2463,6 +2463,30 @@ func buildFallbackColumnCommentStatement(dbType string, schemaName string, table
 	return fmt.Sprintf("COMMENT ON COLUMN %s IS '%s';", columnRef, strings.ReplaceAll(commentText, "'", "''"))
 }
 
+func getColumnsWithMetadataFallback(
+	dbInst db.Database,
+	config connection.ConnectionConfig,
+	schemaName string,
+	tableName string,
+	text func(string, map[string]any) string,
+) ([]connection.ColumnDefinition, error) {
+	columns, err := dbInst.GetColumns(schemaName, tableName)
+	if err != nil {
+		return nil, err
+	}
+	if len(columns) > 0 || resolveDDLDBType(config) != "oracle" {
+		return columns, nil
+	}
+
+	if inferred, inferErr := inferOracleColumnsFromDictionary(dbInst, schemaName, tableName, text); inferErr == nil && len(inferred) > 0 {
+		return inferred, nil
+	}
+	if inferred, inferErr := inferOracleColumnsFromEmptySelect(dbInst, schemaName, tableName, text); inferErr == nil && len(inferred) > 0 {
+		return inferred, nil
+	}
+	return columns, nil
+}
+
 func (a *App) DBGetColumns(config connection.ConnectionConfig, dbName string, tableName string) connection.QueryResult {
 	runConfig := normalizeRunConfig(config, dbName)
 	text := a.appText
@@ -2474,7 +2498,7 @@ func (a *App) DBGetColumns(config connection.ConnectionConfig, dbName string, ta
 	}
 
 	schemaName, pureTableName := normalizeMetadataSchemaAndTable(config, dbName, tableName)
-	columns, err := dbInst.GetColumns(schemaName, pureTableName)
+	columns, err := getColumnsWithMetadataFallback(dbInst, config, schemaName, pureTableName, text)
 	if err != nil && shouldRefreshCachedConnection(err) {
 		if a.invalidateCachedDatabase(runConfig, err) {
 			retryInst, retryErr := a.getDatabaseForcePing(runConfig)
@@ -2482,24 +2506,13 @@ func (a *App) DBGetColumns(config connection.ConnectionConfig, dbName string, ta
 				logger.Error(retryErr, "DBGetColumns 重建连接失败：%s 表=%s.%s", formatConnSummary(runConfig), dbName, tableName)
 				return connection.QueryResult{Success: false, Message: retryErr.Error()}
 			}
-			columns, err = retryInst.GetColumns(schemaName, pureTableName)
+			columns, err = getColumnsWithMetadataFallback(retryInst, config, schemaName, pureTableName, text)
 		}
 	}
 	if err != nil {
 		logger.Error(err, "DBGetColumns 获取列定义失败：%s 表=%s.%s schema=%s pureTable=%s", formatConnSummary(runConfig), dbName, tableName, schemaName, pureTableName)
 		return connection.QueryResult{Success: false, Message: err.Error()}
 	}
-	if len(columns) == 0 && resolveDDLDBType(config) == "oracle" {
-		if inferred, inferErr := inferOracleColumnsFromDictionary(dbInst, schemaName, pureTableName, text); inferErr == nil && len(inferred) > 0 {
-			columns = inferred
-		}
-		if len(columns) == 0 {
-			if inferred, inferErr := inferOracleColumnsFromEmptySelect(dbInst, schemaName, pureTableName, text); inferErr == nil && len(inferred) > 0 {
-				columns = inferred
-			}
-		}
-	}
-
 	return connection.QueryResult{Success: true, Data: ensureNonNilSlice(columns)}
 }
 

internal/app/methods_file.go

@@ -2468,8 +2468,18 @@ func formatImportSQLValue(dbType, columnType string, value interface{}) string {
 
 // ImportDataWithProgress 执行导入并发送进度事件
 func (a *App) ImportDataWithProgress(config connection.ConnectionConfig, dbName, tableName, filePath string) (result connection.QueryResult) {
+	return a.ImportDataWithProgressOptions(config, dbName, tableName, filePath, ImportFileOptions{})
+}
+
+// ImportDataWithProgressOptions executes a streamed import with optional source-header
+// to database-column mappings. ImportDataWithProgress remains the compatibility entrypoint.
+func (a *App) ImportDataWithProgressOptions(config connection.ConnectionConfig, dbName, tableName, filePath string, options ImportFileOptions) (result connection.QueryResult) {
+	if strings.TrimSpace(filePath) == "" {
+		return connection.QueryResult{Success: false, Message: a.appText("file.backend.error.import_file_empty", nil)}
+	}
 	dbType := resolveDDLDBType(config)
 	schemaName, pureTableName := normalizeSchemaAndTable(config, dbName, tableName)
+	metadataSchemaName, metadataTableName := normalizeMetadataSchemaAndTable(config, dbName, tableName)
 	auditTarget := strings.TrimSpace(tableName)
 	if pureTableName != "" {
 		auditTarget = quoteTableIdentByType(dbType, schemaName, pureTableName)
@@ -2491,27 +2501,32 @@ func (a *App) ImportDataWithProgress(config connection.ConnectionConfig, dbName,
 		return connection.QueryResult{Success: false, Message: err.Error()}
 	}
 
-	columnTypeMap := map[string]string{}
-	if defs, colErr := dbInst.GetColumns(schemaName, pureTableName); colErr == nil {
-		columnTypeMap = buildImportColumnTypeMap(defs)
+	targetColumns, colErr := getColumnsWithMetadataFallback(dbInst, config, metadataSchemaName, metadataTableName, a.appText)
+	if colErr != nil && options.ColumnMappings != nil {
+		return connection.QueryResult{Success: false, Message: colErr.Error()}
 	}
 
-	writer := newImportDatabaseRowWriter(dbInst, dbType, tableName, columnTypeMap)
-	consumer := newImportBatchConsumer(writer, defaultImportApplyBatchSize, 0, false, func(state importProgressState) {
+	writer := newImportDatabaseRowWriter(dbInst, dbType, tableName, newImportColumnTypeLookup(targetColumns))
+	batchConsumer := newImportBatchConsumer(writer, defaultImportApplyBatchSize, 0, false, func(state importProgressState) {
 		uievents.Emit(a.ctx, "import:progress", state)
 	})
+	batchConsumer.jobID = strings.TrimSpace(options.JobID)
+	consumer, err := newImportColumnMappingConsumer(batchConsumer, options.ColumnMappings, targetColumns)
+	if err != nil {
+		return connection.QueryResult{Success: false, Message: err.Error()}
+	}
 	if err := streamImportFile(filePath, consumer); err != nil {
-		resultData := consumer.Result()
+		resultData := batchConsumer.Result()
 		maybeReleaseFileTransferMemory("import-stream-error", int64(resultData.Total), filePath)
 		return connection.QueryResult{Success: false, Message: err.Error()}
 	}
-	if err := consumer.Flush(); err != nil {
-		resultData := consumer.Result()
+	if err := batchConsumer.Flush(); err != nil {
+		resultData := batchConsumer.Result()
 		maybeReleaseFileTransferMemory("import-flush-error", int64(resultData.Total), filePath)
 		return connection.QueryResult{Success: false, Message: err.Error()}
 	}
 
-	resultData := consumer.Result()
+	resultData := batchConsumer.Result()
 	if resultData.Total == 0 {
 		maybeReleaseFileTransferMemory("import-empty", 0, filePath)
 		return connection.QueryResult{Success: true, Message: a.appText("file.backend.message.import_no_data", nil)}
@@ -2612,8 +2627,23 @@ func (a *App) ExportTable(config connection.ConnectionConfig, dbName string, tab
 	return a.ExportTableWithOptions(config, dbName, tableName, ExportFileOptions{Format: format})
 }
 
+func buildExportTableSelectQuery(dbType string, tableName string, columns []string) string {
+	selectList := "*"
+	if len(columns) > 0 {
+		quotedColumns := make([]string, len(columns))
+		for index, column := range columns {
+			quotedColumns[index] = quoteIdentByType(dbType, column)
+		}
+		selectList = strings.Join(quotedColumns, ", ")
+	}
+	return fmt.Sprintf("SELECT %s FROM %s", selectList, quoteQualifiedIdentByType(dbType, tableName))
+}
+
 func (a *App) ExportTableWithOptions(config connection.ConnectionConfig, dbName string, tableName string, options ExportFileOptions) connection.QueryResult {
 	options = normalizeExportFileOptions("", options)
+	if err := validateExportColumnsSelection(options); err != nil {
+		return connection.QueryResult{Success: false, Message: err.Error()}
+	}
 	format := options.Format
 	if format != "sql" {
 		if err := verifyOptionalDriverAgentReadyForExport(config); err != nil {
@@ -2684,7 +2714,7 @@ func (a *App) ExportTableWithOptions(config connection.ConnectionConfig, dbName
 	}
 
 	dbType := resolveDDLDBType(config)
-	query := fmt.Sprintf("SELECT * FROM %s", quoteQualifiedIdentByType(dbType, tableName))
+	query := buildExportTableSelectQuery(dbType, tableName, options.Columns)
 
 	f, err := os.Create(filename)
 	if err != nil {
@@ -4060,6 +4090,9 @@ func (a *App) ExportDataWithOptions(data []map[string]interface{}, columns []str
 		defaultName = "export"
 	}
 	options = normalizeExportFileOptions("", options)
+	if err := validateExportColumnsSelection(options); err != nil {
+		return connection.QueryResult{Success: false, Message: err.Error()}
+	}
 	if !options.TotalRowsKnown {
 		options.TotalRowsKnown = true
 		options.TotalRowsHint = int64(len(data))
@@ -4116,6 +4149,9 @@ func (a *App) ExportQueryWithOptions(config connection.ConnectionConfig, dbName
 		defaultName = "export"
 	}
 	options = normalizeExportFileOptions("", options)
+	if err := validateExportColumnsSelection(options); err != nil {
+		return connection.QueryResult{Success: false, Message: err.Error()}
+	}
 	format := options.Format
 	if format != "sql" {
 		if err := verifyOptionalDriverAgentReadyForExport(config); err != nil {
@@ -4234,6 +4270,66 @@ type exportValueStreamConsumer interface {
 	ConsumeRowValues(values []interface{}) error
 }
 
+type exportColumnProjectionConsumer struct {
+	delegate         db.QueryStreamConsumer
+	requestedColumns []string
+	columns          []string
+	columnIndexes    []int
+	values           []interface{}
+}
+
+func (c *exportColumnProjectionConsumer) SetColumns(columns []string) error {
+	selectedColumns, err := resolveRequestedExportColumns(columns, c.requestedColumns)
+	if err != nil {
+		return err
+	}
+	indexByColumn := make(map[string]int, len(columns))
+	for index, column := range columns {
+		if _, exists := indexByColumn[column]; !exists {
+			indexByColumn[column] = index
+		}
+	}
+
+	c.columns = selectedColumns
+	c.columnIndexes = make([]int, len(selectedColumns))
+	for index, column := range selectedColumns {
+		c.columnIndexes[index] = indexByColumn[column]
+	}
+	c.values = make([]interface{}, len(c.columns))
+	if c.delegate == nil {
+		return nil
+	}
+	return c.delegate.SetColumns(c.columns)
+}
+
+func (c *exportColumnProjectionConsumer) ConsumeRow(row map[string]interface{}) error {
+	if c.delegate == nil {
+		return nil
+	}
+	return c.delegate.ConsumeRow(row)
+}
+
+func (c *exportColumnProjectionConsumer) ConsumeRowValues(values []interface{}) error {
+	for selectedIndex, sourceIndex := range c.columnIndexes {
+		if sourceIndex < len(values) {
+			c.values[selectedIndex] = values[sourceIndex]
+		} else {
+			c.values[selectedIndex] = nil
+		}
+	}
+	if c.delegate == nil {
+		return nil
+	}
+	if valueConsumer, ok := c.delegate.(exportValueStreamConsumer); ok {
+		return valueConsumer.ConsumeRowValues(c.values)
+	}
+	row := make(map[string]interface{}, len(c.columns))
+	for index, column := range c.columns {
+		row[column] = c.values[index]
+	}
+	return c.delegate.ConsumeRow(row)
+}
+
 type countingExportConsumer struct {
 	delegate db.QueryStreamConsumer
 	columns  []string
@@ -4861,6 +4957,24 @@ func resolveExportColumns(columns []string, data []map[string]interface{}) []str
 	return derived
 }
 
+func resolveRequestedExportColumns(columns []string, requested []string) ([]string, error) {
+	if len(requested) == 0 {
+		return columns, nil
+	}
+	available := make(map[string]struct{}, len(columns))
+	for _, column := range columns {
+		available[column] = struct{}{}
+	}
+	selected := make([]string, 0, len(requested))
+	for _, column := range requested {
+		if _, exists := available[column]; !exists {
+			return nil, fmt.Errorf("requested export column %q was not found in query result", column)
+		}
+		selected = append(selected, column)
+	}
+	return selected, nil
+}
+
 func newExportFileWriter(f *os.File, options ExportFileOptions) (exportFileWriter, error) {
 	options = normalizeExportFileOptions("", options)
 	switch options.Format {
@@ -4924,6 +5038,10 @@ func streamQueryDataForExport(dbInst db.Database, config connection.ConnectionCo
 }
 
 func exportQueryResultToFile(f *os.File, dbInst db.Database, config connection.ConnectionConfig, query string, options ExportFileOptions, reporter *exportProgressReporter) (int64, []string, error) {
+	options = normalizeExportFileOptions("", options)
+	if err := validateExportColumnsSelection(options); err != nil {
+		return 0, nil, err
+	}
 	writer, err := newExportFileWriter(f, options)
 	if err != nil {
 		return 0, nil, err
@@ -4932,19 +5050,32 @@ func exportQueryResultToFile(f *os.File, dbInst db.Database, config connection.C
 	if reporter != nil {
 		reporter.Start(reporter.text("data_export.progress.stage.querying_data", nil))
 	}
-	consumer := &countingExportConsumer{delegate: writer, reporter: reporter}
+	var projection *exportColumnProjectionConsumer
+	delegate := db.QueryStreamConsumer(writer)
+	if len(options.Columns) > 0 {
+		projection = &exportColumnProjectionConsumer{
+			delegate:         writer,
+			requestedColumns: options.Columns,
+		}
+		delegate = projection
+	}
+	consumer := &countingExportConsumer{delegate: delegate, reporter: reporter}
 	streamErr := streamQueryDataForExport(dbInst, config, query, consumer)
 	if reporter != nil && streamErr == nil {
 		reporter.Finalizing(consumer.rowCount)
 	}
 	closeErr := writer.Close()
+	exportedColumns := consumer.columns
+	if projection != nil {
+		exportedColumns = projection.columns
+	}
 	if streamErr != nil {
-		return consumer.rowCount, consumer.columns, streamErr
+		return consumer.rowCount, exportedColumns, streamErr
 	}
 	if closeErr != nil {
-		return consumer.rowCount, consumer.columns, closeErr
+		return consumer.rowCount, exportedColumns, closeErr
 	}
-	return consumer.rowCount, consumer.columns, nil
+	return consumer.rowCount, exportedColumns, nil
 }
 
 func fillExportRecordFromValues(record []string, values []interface{}, markdown bool) []string {
@@ -4969,7 +5100,7 @@ func fillExportRecordFromRow(record []string, row map[string]interface{}, column
 
 func formatExportRecordValue(val interface{}, markdown bool) string {
 	if val == nil {
-		return "NULL"
+		return ""
 	}
 	text := formatExportCellText(val)
 	if markdown {
@@ -4988,7 +5119,15 @@ func writeRowsToFileWithReporter(f *os.File, data []map[string]interface{}, colu
 	if f == nil {
 		return 0, fmt.Errorf("file required")
 	}
+	options = normalizeExportFileOptions("", options)
+	if err := validateExportColumnsSelection(options); err != nil {
+		return 0, err
+	}
 	columns = resolveExportColumns(columns, data)
+	columns, err := resolveRequestedExportColumns(columns, options.Columns)
+	if err != nil {
+		return 0, err
+	}
 	writer, err := newExportFileWriter(f, options)
 	if err != nil {
 		return 0, err
@@ -5202,7 +5341,7 @@ func writeRowsToHTML(f *os.File, data []map[string]interface{}, columns []string
 
 func formatExportCellText(val interface{}) string {
 	if val == nil {
-		return "NULL"
+		return ""
 	}
 
 	switch v := val.(type) {
@@ -5210,7 +5349,7 @@ func formatExportCellText(val interface{}) string {
 		return v.Format("2006-01-02 15:04:05")
 	case *time.Time:
 		if v == nil {
-			return "NULL"
+			return ""
 		}
 		return v.Format("2006-01-02 15:04:05")
 	case float32:

internal/app/methods_file_export_test.go

@@ -249,6 +249,139 @@ func TestFormatExportCellText_FloatNoScientificNotation(t *testing.T) {
 	}
 }
 
+func TestBuildExportTableSelectQuery_QuotesRequestedColumnsInOrder(t *testing.T) {
+	got := buildExportTableSelectQuery(
+		"mysql",
+		"audit.users",
+		[]string{"display name", " id "},
+	)
+	want := "SELECT `display name`, ` id ` FROM `audit`.`users`"
+	if got != want {
+		t.Fatalf("整表选列查询异常，want=%q got=%q", want, got)
+	}
+
+	got = buildExportTableSelectQuery("postgres", "public.users", nil)
+	want = `SELECT * FROM "public"."users"`
+	if got != want {
+		t.Fatalf("未指定列时应保持 SELECT * 兼容行为，want=%q got=%q", want, got)
+	}
+}
+
+func TestWriteRowsToFile_TabularFormatsExportNilAsEmptyCell(t *testing.T) {
+	var nilTime *time.Time
+	data := []map[string]interface{}{
+		{"id": 1, "nullable": nil, "nullable_time": nilTime, "tail": "end"},
+	}
+	columns := []string{"id", "nullable", "nullable_time", "tail"}
+
+	for _, format := range []string{"csv", "md", "html", "xlsx"} {
+		t.Run(format, func(t *testing.T) {
+			f, err := os.CreateTemp("", fmt.Sprintf("gonavi-export-null-*.%s", format))
+			if err != nil {
+				t.Fatalf("创建临时文件失败: %v", err)
+			}
+			defer os.Remove(f.Name())
+			defer f.Close()
+
+			if err := writeRowsToFile(f, data, columns, ExportFileOptions{Format: format}); err != nil {
+				t.Fatalf("写入 %s 失败: %v", format, err)
+			}
+
+			if format == "xlsx" {
+				workbook, err := excelize.OpenFile(f.Name())
+				if err != nil {
+					t.Fatalf("打开 xlsx 失败: %v", err)
+				}
+				defer workbook.Close()
+				rows, err := workbook.GetRows("Sheet1")
+				if err != nil {
+					t.Fatalf("读取 xlsx 失败: %v", err)
+				}
+				if len(rows) < 2 || len(rows[1]) < 4 || rows[1][1] != "" || rows[1][2] != "" {
+					t.Fatalf("xlsx 实际 nil 应导出为空单元格，rows=%v", rows)
+				}
+				return
+			}
+
+			contentBytes, err := os.ReadFile(f.Name())
+			if err != nil {
+				t.Fatalf("读取 %s 失败: %v", format, err)
+			}
+			content := string(contentBytes)
+			switch format {
+			case "csv":
+				if !strings.Contains(content, "1,,,end") {
+					t.Fatalf("csv 实际 nil 应导出为空单元格: %q", content)
+				}
+			case "md":
+				if !strings.Contains(content, "| 1 |  |  | end |") {
+					t.Fatalf("markdown 实际 nil 应导出为空单元格: %q", content)
+				}
+			case "html":
+				if !strings.Contains(content, "<td>1</td><td></td><td></td><td>end</td>") {
+					t.Fatalf("html 实际 nil 应导出为空单元格: %q", content)
+				}
+			}
+		})
+	}
+}
+
+func TestWriteRowsToFile_ProjectsColumnsFromExportOptions(t *testing.T) {
+	f, err := os.CreateTemp("", "gonavi-export-buffered-selected-columns-*.csv")
+	if err != nil {
+		t.Fatalf("创建临时文件失败: %v", err)
+	}
+	defer os.Remove(f.Name())
+	defer f.Close()
+
+	data := []map[string]interface{}{
+		{"id": 1, " name ": "alice", "note": "internal"},
+	}
+	columns := []string{"id", " name ", "note"}
+	if err := writeRowsToFile(f, data, columns, ExportFileOptions{
+		Format:  "csv",
+		Columns: []string{" name ", "id", " name ", "   "},
+	}); err != nil {
+		t.Fatalf("写入 csv 失败: %v", err)
+	}
+
+	contentBytes, err := os.ReadFile(f.Name())
+	if err != nil {
+		t.Fatalf("读取导出文件失败: %v", err)
+	}
+	content := strings.TrimPrefix(string(contentBytes), "\uFEFF")
+	want := "\" name \",id\nalice,1\n"
+	if content != want {
+		t.Fatalf("缓冲导出未按 options.Columns 投影，want=%q got=%q", want, content)
+	}
+}
+
+func TestWriteRowsToFile_RejectsExplicitEmptyColumnSelection(t *testing.T) {
+	data := []map[string]interface{}{{"id": 1}}
+	columns := []string{"id"}
+	for name, selectedColumns := range map[string][]string{
+		"empty":      {},
+		"blank-only": {"", "   "},
+	} {
+		t.Run(name, func(t *testing.T) {
+			f, err := os.CreateTemp("", "gonavi-export-empty-columns-*.csv")
+			if err != nil {
+				t.Fatalf("创建临时文件失败: %v", err)
+			}
+			defer os.Remove(f.Name())
+			defer f.Close()
+
+			err = writeRowsToFile(f, data, columns, ExportFileOptions{
+				Format:  "csv",
+				Columns: selectedColumns,
+			})
+			if err == nil || !strings.Contains(err.Error(), "at least one export column must be selected") {
+				t.Fatalf("显式空选列应拒绝导出，err=%v", err)
+			}
+		})
+	}
+}
+
 func TestWriteRowsToFile_Markdown_NumberKeepPlainText(t *testing.T) {
 	f, err := os.CreateTemp("", "gonavi-export-*.md")
 	if err != nil {
@@ -319,6 +452,37 @@ func TestWriteRowsToFile_JSON_NumberKeepPlainText(t *testing.T) {
 	}
 }
 
+func TestWriteRowsToFile_JSONKeepsNilAsJSONNull(t *testing.T) {
+	f, err := os.CreateTemp("", "gonavi-export-null-*.json")
+	if err != nil {
+		t.Fatalf("创建临时文件失败: %v", err)
+	}
+	defer os.Remove(f.Name())
+	defer f.Close()
+
+	if err := writeRowsToFile(
+		f,
+		[]map[string]interface{}{{"nullable": nil}},
+		[]string{"nullable"},
+		ExportFileOptions{Format: "json"},
+	); err != nil {
+		t.Fatalf("写入 json 失败: %v", err)
+	}
+
+	contentBytes, err := os.ReadFile(f.Name())
+	if err != nil {
+		t.Fatalf("读取 json 失败: %v", err)
+	}
+	var decoded []map[string]interface{}
+	if err := json.Unmarshal(contentBytes, &decoded); err != nil {
+		t.Fatalf("解析 json 失败: %v", err)
+	}
+	value, exists := decoded[0]["nullable"]
+	if !exists || value != nil {
+		t.Fatalf("JSON 导出应保留 null 语义，decoded=%v", decoded)
+	}
+}
+
 func TestNormalizeExportJSONValue_LocalDateTimeString_NoTimezoneShift(t *testing.T) {
 	originalLocal := time.Local
 	time.Local = time.FixedZone("UTC+8", 8*60*60)
@@ -830,6 +994,150 @@ func TestExportQueryResultToFile_UsesValueStreamPathWhenAvailable(t *testing.T)
 	}
 }
 
+func TestExportQueryResultToFile_ProjectsRequestedColumnsInOrderForValueStream(t *testing.T) {
+	f, err := os.CreateTemp("", "gonavi-export-selected-columns-*.csv")
+	if err != nil {
+		t.Fatalf("创建临时文件失败: %v", err)
+	}
+	defer os.Remove(f.Name())
+	defer f.Close()
+
+	fake := &fakeValueStreamExportDB{
+		streamCols: []string{"id", "name", "note"},
+		streamValues: [][]interface{}{
+			{1, "alice", "internal"},
+			{2, "bob", "private"},
+		},
+	}
+
+	rowCount, columns, err := exportQueryResultToFile(
+		f,
+		fake,
+		connection.ConnectionConfig{Type: "mysql", Timeout: 10},
+		"SELECT id, name, note FROM users",
+		ExportFileOptions{Format: "csv", Columns: []string{"name", "id"}},
+		nil,
+	)
+	if err != nil {
+		t.Fatalf("exportQueryResultToFile 返回错误: %v", err)
+	}
+	if rowCount != 2 {
+		t.Fatalf("导出行数异常，want=2 got=%d", rowCount)
+	}
+	if len(columns) != 2 || columns[0] != "name" || columns[1] != "id" {
+		t.Fatalf("导出列未按请求顺序投影，got=%v", columns)
+	}
+
+	contentBytes, err := os.ReadFile(f.Name())
+	if err != nil {
+		t.Fatalf("读取导出文件失败: %v", err)
+	}
+	content := strings.TrimPrefix(string(contentBytes), "\uFEFF")
+	want := "name,id\nalice,1\nbob,2\n"
+	if content != want {
+		t.Fatalf("选列导出内容异常，want=%q got=%q", want, content)
+	}
+}
+
+func TestExportQueryResultToFile_ProjectsRequestedColumnsInOrderForMapStream(t *testing.T) {
+	f, err := os.CreateTemp("", "gonavi-export-selected-map-columns-*.csv")
+	if err != nil {
+		t.Fatalf("创建临时文件失败: %v", err)
+	}
+	defer os.Remove(f.Name())
+	defer f.Close()
+
+	fake := &fakeStreamExportDB{
+		streamCols: []string{"id", "name", "note"},
+		streamData: []map[string]interface{}{
+			{"id": 1, "name": "alice", "note": "internal"},
+		},
+	}
+
+	_, columns, err := exportQueryResultToFile(
+		f,
+		fake,
+		connection.ConnectionConfig{Type: "mysql", Timeout: 10},
+		"SELECT id, name, note FROM users",
+		ExportFileOptions{Format: "csv", Columns: []string{"note", "id"}},
+		nil,
+	)
+	if err != nil {
+		t.Fatalf("exportQueryResultToFile 返回错误: %v", err)
+	}
+	if len(columns) != 2 || columns[0] != "note" || columns[1] != "id" {
+		t.Fatalf("导出列未按请求顺序投影，got=%v", columns)
+	}
+
+	contentBytes, err := os.ReadFile(f.Name())
+	if err != nil {
+		t.Fatalf("读取导出文件失败: %v", err)
+	}
+	content := strings.TrimPrefix(string(contentBytes), "\uFEFF")
+	want := "note,id\ninternal,1\n"
+	if content != want {
+		t.Fatalf("选列 map 流导出内容异常，want=%q got=%q", want, content)
+	}
+}
+
+func TestExportQueryResultToFile_RejectsRequestedColumnMissingFromResult(t *testing.T) {
+	f, err := os.CreateTemp("", "gonavi-export-missing-column-*.csv")
+	if err != nil {
+		t.Fatalf("创建临时文件失败: %v", err)
+	}
+	defer os.Remove(f.Name())
+	defer f.Close()
+
+	fake := &fakeValueStreamExportDB{
+		streamCols:   []string{"id", "name"},
+		streamValues: [][]interface{}{{1, "alice"}},
+	}
+
+	_, _, err = exportQueryResultToFile(
+		f,
+		fake,
+		connection.ConnectionConfig{Type: "mysql", Timeout: 10},
+		"SELECT id, name FROM users",
+		ExportFileOptions{Format: "csv", Columns: []string{"name", "missing"}},
+		nil,
+	)
+	if err == nil || !strings.Contains(err.Error(), `requested export column "missing" was not found`) {
+		t.Fatalf("查询结果不包含请求列时应拒绝导出，err=%v", err)
+	}
+}
+
+func TestExportQueryResultToFile_RejectsExplicitEmptyColumnSelection(t *testing.T) {
+	fake := &fakeValueStreamExportDB{
+		streamCols:   []string{"id"},
+		streamValues: [][]interface{}{{1}},
+	}
+	for name, selectedColumns := range map[string][]string{
+		"empty":      {},
+		"blank-only": {"", "   "},
+	} {
+		t.Run(name, func(t *testing.T) {
+			f, err := os.CreateTemp("", "gonavi-export-empty-query-columns-*.csv")
+			if err != nil {
+				t.Fatalf("创建临时文件失败: %v", err)
+			}
+			defer os.Remove(f.Name())
+			defer f.Close()
+
+			_, _, err = exportQueryResultToFile(
+				f,
+				fake,
+				connection.ConnectionConfig{Type: "mysql", Timeout: 10},
+				"SELECT id FROM users",
+				ExportFileOptions{Format: "csv", Columns: selectedColumns},
+				nil,
+			)
+			if err == nil || !strings.Contains(err.Error(), "at least one export column must be selected") {
+				t.Fatalf("显式空选列应拒绝查询导出，err=%v", err)
+			}
+		})
+	}
+}
+
 func TestGetExportQueryTimeout_ClickHouseUsesLongerMinimum(t *testing.T) {
 	timeout := getExportQueryTimeout(connection.ConnectionConfig{
 		Type:    "clickhouse",
@@ -912,7 +1220,7 @@ func TestWriteRowsToFile_HTML_EscapeAndStyle(t *testing.T) {
 	if !strings.Contains(content, "line1<br>line2") {
 		t.Fatalf("html 导出换行未转为 <br>: %s", content)
 	}
-	if !strings.Contains(content, "<td>NULL</td>") {
+	if !strings.Contains(content, "<td></td>") {
 		t.Fatalf("html 导出空值显示异常: %s", content)
 	}
 }

internal/app/methods_file_import_test.go

@@ -1,6 +1,7 @@
 package app
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"os"
@@ -9,6 +10,11 @@ import (
 	"strings"
 	"testing"
 
+	"GoNavi-Wails/internal/connection"
+	"GoNavi-Wails/internal/db"
+	"GoNavi-Wails/internal/secretstore"
+	"GoNavi-Wails/internal/uievents"
+
 	"github.com/xuri/excelize/v2"
 )
 
@@ -177,10 +183,15 @@ type fakeImportRowWriter struct {
 	batchCalls       int
 	singleCalls      int
 	batchSizes       []int
+	batchRows        []map[string]interface{}
 	batchErr         error
 	singleErrByRowID map[interface{}]error
 }
 
+type noopImportEventEmitter struct{}
+
+func (noopImportEventEmitter) Emit(string, ...any) {}
+
 func (w *fakeImportRowWriter) SetColumns(columns []string) {
 	w.columns = append([]string(nil), columns...)
 }
@@ -188,6 +199,7 @@ func (w *fakeImportRowWriter) SetColumns(columns []string) {
 func (w *fakeImportRowWriter) ApplyBatch(rows []map[string]interface{}) error {
 	w.batchCalls++
 	w.batchSizes = append(w.batchSizes, len(rows))
+	w.batchRows = append(w.batchRows, cloneImportRows(rows)...)
 	return w.batchErr
 }
 
@@ -203,6 +215,252 @@ func (w *fakeImportRowWriter) BatchEnabled() bool {
 	return true
 }
 
+func TestImportColumnMappingConsumerStreamsMappedColumnsAndRows(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "users.csv")
+	if err := os.WriteFile(path, []byte("User ID,Display Name,Ignored\n1,Alice,skip me\n"), 0o600); err != nil {
+		t.Fatalf("write csv: %v", err)
+	}
+
+	writer := &fakeImportRowWriter{}
+	batchConsumer := newImportBatchConsumer(writer, 1000, 0, false, nil)
+	consumer, err := newImportColumnMappingConsumer(batchConsumer, map[string]string{
+		"User ID":      "ID",
+		"Display Name": "display_name",
+	}, []connection.ColumnDefinition{
+		{Name: "id", Type: "bigint"},
+		{Name: "display_name", Type: "varchar(255)"},
+	})
+	if err != nil {
+		t.Fatalf("newImportColumnMappingConsumer returned error: %v", err)
+	}
+
+	if err := streamImportFile(path, consumer); err != nil {
+		t.Fatalf("streamImportFile returned error: %v", err)
+	}
+	if err := batchConsumer.Flush(); err != nil {
+		t.Fatalf("Flush returned error: %v", err)
+	}
+
+	if !reflect.DeepEqual(writer.columns, []string{"id", "display_name"}) {
+		t.Fatalf("unexpected mapped columns: %#v", writer.columns)
+	}
+	wantRows := []map[string]interface{}{{"id": "1", "display_name": "Alice"}}
+	if !reflect.DeepEqual(writer.batchRows, wantRows) {
+		t.Fatalf("unexpected mapped rows: %#v", writer.batchRows)
+	}
+}
+
+func TestImportColumnMappingConsumerNilMappingsPreserveLegacyHeaders(t *testing.T) {
+	collector := newImportPreviewCollector(5)
+	consumer, err := newImportColumnMappingConsumer(collector, nil, nil)
+	if err != nil {
+		t.Fatalf("newImportColumnMappingConsumer returned error: %v", err)
+	}
+	if err := consumer.SetColumns([]string{"Raw Header"}); err != nil {
+		t.Fatalf("SetColumns returned error: %v", err)
+	}
+	if err := consumer.ConsumeRow(map[string]interface{}{"Raw Header": "value"}); err != nil {
+		t.Fatalf("ConsumeRow returned error: %v", err)
+	}
+	result := collector.Result()
+	if !reflect.DeepEqual(result.Columns, []string{"Raw Header"}) {
+		t.Fatalf("legacy columns changed: %#v", result.Columns)
+	}
+	if got := result.PreviewRows[0]["Raw Header"]; got != "value" {
+		t.Fatalf("legacy row changed: %#v", result.PreviewRows)
+	}
+}
+
+func TestImportColumnMappingConsumerPrefersExactTargetWhenCaseDistinct(t *testing.T) {
+	collector := newImportPreviewCollector(5)
+	consumer, err := newImportColumnMappingConsumer(collector, map[string]string{
+		"Source Value": "Foo",
+	}, []connection.ColumnDefinition{
+		{Name: "Foo", Type: "text"},
+		{Name: "foo", Type: "integer"},
+	})
+	if err != nil {
+		t.Fatalf("newImportColumnMappingConsumer returned error: %v", err)
+	}
+	if err := consumer.SetColumns([]string{"Source Value"}); err != nil {
+		t.Fatalf("SetColumns returned error: %v", err)
+	}
+	if !reflect.DeepEqual(collector.columns, []string{"Foo"}) {
+		t.Fatalf("unexpected exact mapped target: %#v", collector.columns)
+	}
+}
+
+func TestImportColumnTypeLookupKeepsCaseDistinctTypes(t *testing.T) {
+	lookup := newImportColumnTypeLookup([]connection.ColumnDefinition{
+		{Name: "Foo", Type: "text"},
+		{Name: "foo", Type: "boolean"},
+		{Name: "event_id", Type: "bigint"},
+	})
+	if got := lookup.Resolve("Foo"); got != "text" {
+		t.Fatalf("exact Foo type = %q, want text", got)
+	}
+	if got := lookup.Resolve("foo"); got != "boolean" {
+		t.Fatalf("exact foo type = %q, want boolean", got)
+	}
+	if got := lookup.Resolve("FOO"); got != "" {
+		t.Fatalf("ambiguous folded FOO type = %q, want empty", got)
+	}
+	if got := lookup.Resolve("EVENT_ID"); got != "bigint" {
+		t.Fatalf("unique folded EVENT_ID type = %q, want bigint", got)
+	}
+	query, err := buildImportInsertQuery(
+		"postgres",
+		"events",
+		[]string{"Foo", "foo"},
+		map[string]interface{}{"Foo": "false", "foo": "false"},
+		lookup,
+	)
+	if err != nil {
+		t.Fatalf("buildImportInsertQuery returned error: %v", err)
+	}
+	if !strings.Contains(query, `("Foo", "foo") VALUES ('false', false)`) {
+		t.Fatalf("case-distinct target types produced wrong SQL: %s", query)
+	}
+}
+
+func TestImportColumnMappingConsumerRejectsInvalidMappings(t *testing.T) {
+	targetColumns := []connection.ColumnDefinition{
+		{Name: "id", Type: "bigint"},
+		{Name: "display_name", Type: "varchar(255)"},
+	}
+	tests := []struct {
+		name          string
+		mappings      map[string]string
+		headers       []string
+		targetColumns []connection.ColumnDefinition
+		wantInError   string
+	}{
+		{
+			name:        "requires at least one selected target",
+			mappings:    map[string]string{"User ID": ""},
+			headers:     []string{"User ID"},
+			wantInError: "至少",
+		},
+		{
+			name:        "rejects unknown target",
+			mappings:    map[string]string{"User ID": "missing"},
+			headers:     []string{"User ID"},
+			wantInError: "目标字段",
+		},
+		{
+			name: "rejects duplicate targets",
+			mappings: map[string]string{
+				"User ID":      "id",
+				"Display Name": "ID",
+			},
+			headers:     []string{"User ID", "Display Name"},
+			wantInError: "重复",
+		},
+		{
+			name:        "rejects unknown source",
+			mappings:    map[string]string{"Missing Header": "id"},
+			headers:     []string{"User ID"},
+			wantInError: "源字段",
+		},
+		{
+			name:     "rejects ambiguous case insensitive target",
+			mappings: map[string]string{"Value": "FOO"},
+			headers:  []string{"Value"},
+			targetColumns: []connection.ColumnDefinition{
+				{Name: "Foo", Type: "text"},
+				{Name: "foo", Type: "integer"},
+			},
+			wantInError: "不明确",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			baseConsumer := newImportPreviewCollector(5)
+			columns := targetColumns
+			if tt.targetColumns != nil {
+				columns = tt.targetColumns
+			}
+			consumer, err := newImportColumnMappingConsumer(baseConsumer, tt.mappings, columns)
+			if err == nil {
+				err = consumer.SetColumns(tt.headers)
+			}
+			if err == nil || !strings.Contains(err.Error(), tt.wantInError) {
+				t.Fatalf("error = %v, want substring %q", err, tt.wantInError)
+			}
+		})
+	}
+}
+
+func TestImportDataWithProgressOptionsRejectsEmptyFilePathBeforeDatabaseAccess(t *testing.T) {
+	app := &App{}
+	wantMessage := app.appText("file.backend.error.import_file_empty", nil)
+	result := app.ImportDataWithProgressOptions(connection.ConnectionConfig{}, "", "users", "  ", ImportFileOptions{})
+	if result.Success {
+		t.Fatal("empty file path should fail")
+	}
+	if result.Message != wantMessage {
+		t.Fatalf("message = %q, want %q", result.Message, wantMessage)
+	}
+}
+
+func TestImportDataWithProgressOptionsUsesOracleColumnMetadataFallback(t *testing.T) {
+	originalNewDatabaseFunc := newDatabaseFunc
+	originalResolveDialConfigWithProxyFunc := resolveDialConfigWithProxyFunc
+	originalDriverRuntimeSupportStatusFunc := driverRuntimeSupportStatusFunc
+	originalVerifyDriverAgentRevisionFunc := verifyDriverAgentRevisionFunc
+	t.Cleanup(func() {
+		newDatabaseFunc = originalNewDatabaseFunc
+		resolveDialConfigWithProxyFunc = originalResolveDialConfigWithProxyFunc
+		driverRuntimeSupportStatusFunc = originalDriverRuntimeSupportStatusFunc
+		verifyDriverAgentRevisionFunc = originalVerifyDriverAgentRevisionFunc
+	})
+
+	fakeDB := &fakeMetadataRetryDB{
+		queryResults: []fakeMetadataQueryResult{{
+			match: "all_tab_columns",
+			rows: []map[string]interface{}{
+				{"COLUMN_NAME": "ID", "DATA_TYPE": "NUMBER", "DATA_PRECISION": 19, "NULLABLE": "N"},
+				{"COLUMN_NAME": "DISPLAY_NAME", "DATA_TYPE": "VARCHAR2", "CHAR_LENGTH": 255, "NULLABLE": "Y"},
+			},
+			fields: []string{"COLUMN_NAME", "DATA_TYPE", "DATA_PRECISION", "CHAR_LENGTH", "NULLABLE"},
+		}},
+	}
+	newDatabaseFunc = func(string) (db.Database, error) { return fakeDB, nil }
+	resolveDialConfigWithProxyFunc = func(config connection.ConnectionConfig) (connection.ConnectionConfig, error) {
+		return config, nil
+	}
+	driverRuntimeSupportStatusFunc = func(string) (bool, string) { return true, "" }
+	verifyDriverAgentRevisionFunc = func(connection.ConnectionConfig) error { return nil }
+
+	path := filepath.Join(t.TempDir(), "users.csv")
+	if err := os.WriteFile(path, []byte("User ID,Display Name\n1,Alice\n"), 0o600); err != nil {
+		t.Fatalf("write csv: %v", err)
+	}
+
+	app := NewAppWithSecretStore(secretstore.NewUnavailableStore("test"))
+	app.ctx = uievents.WithEmitter(context.Background(), noopImportEventEmitter{})
+	result := app.ImportDataWithProgressOptions(
+		connection.ConnectionConfig{Type: "oracle", Host: "127.0.0.1", Port: 1521, Database: "ORCL"},
+		"APP",
+		"USERS",
+		path,
+		ImportFileOptions{ColumnMappings: map[string]string{
+			"User ID":      "ID",
+			"Display Name": "DISPLAY_NAME",
+		}},
+	)
+	if !result.Success {
+		t.Fatalf("Oracle fallback columns should allow mapped import, got: %s", result.Message)
+	}
+	if fakeDB.columnSchema != "APP" || fakeDB.columnTable != "USERS" {
+		t.Fatalf("GetColumns target = %q.%q, want APP.USERS", fakeDB.columnSchema, fakeDB.columnTable)
+	}
+	if len(fakeDB.queries) == 0 || !strings.Contains(fakeDB.queries[0], "all_tab_columns") {
+		t.Fatalf("expected Oracle dictionary metadata fallback, queries=%v", fakeDB.queries)
+	}
+}
+
 func TestImportBatchConsumerUsesBatchWriterInConfiguredBatches(t *testing.T) {
 	writer := &fakeImportRowWriter{}
 	consumer := newImportBatchConsumer(writer, 1000, 1201, true, nil)
@@ -267,3 +525,22 @@ func TestImportBatchConsumerFallsBackToSingleRowsWhenBatchFails(t *testing.T) {
 		t.Fatalf("unexpected error logs: %#v", result.ErrorLogs)
 	}
 }
+
+func TestImportBatchConsumerProgressIncludesJobID(t *testing.T) {
+	writer := &fakeImportRowWriter{}
+	var progress []importProgressState
+	consumer := newImportBatchConsumer(writer, 1, 1, true, func(state importProgressState) {
+		progress = append(progress, state)
+	})
+	consumer.jobID = "import-job-1"
+
+	if err := consumer.ConsumeRow(map[string]interface{}{"id": 1}); err != nil {
+		t.Fatalf("ConsumeRow returned error: %v", err)
+	}
+	if len(progress) != 1 {
+		t.Fatalf("progress event count = %d, want 1", len(progress))
+	}
+	if progress[0].JobID != "import-job-1" {
+		t.Fatalf("progress job id = %q, want import-job-1", progress[0].JobID)
+	}
+}

internal/app/xlsx_stream_writer.go

@@ -127,7 +127,7 @@ func (w *xlsxExportFileWriter) ConsumeRow(row map[string]interface{}) error {
 	for i, col := range w.columns {
 		val := row[col]
 		if val == nil {
-			values[i] = "NULL"
+			values[i] = ""
 			continue
 		}
 		values[i] = formatExportCellText(val)
@@ -154,7 +154,7 @@ func (w *xlsxExportFileWriter) ConsumeRowValues(values []interface{}) error {
 			value = values[i]
 		}
 		if value == nil {
-			record[i] = "NULL"
+			record[i] = ""
 			continue
 		}
 		record[i] = formatExportCellText(value)

internal/webserver/server.go

@@ -68,6 +68,7 @@ var desktopOnlyAppMethods = map[string]struct{}{
 	"ImportData":                    {},
 	"PreviewImportFile":             {},
 	"ImportDataWithProgress":        {},
+	"ImportDataWithProgressOptions": {},
 	"ExportTable":                   {},
 	"ExportTableWithOptions":        {},
 	"ExportTablesSQL":               {},

internal/webserver/server_test.go

@@ -104,7 +104,7 @@ func TestMethodInvokerRejectsDesktopOnlyAppMethodsBeforeReflection(t *testing.T)
 
 	for _, method := range []string{
 		"Shutdown", "ExportSQLAuditFile", "OpenSQLFile", "ExecuteSQLFile", "ReadSQLFile",
-		"PreviewImportFile", "ImportDataWithProgress", "GetDataRootDirectoryInfo",
+		"PreviewImportFile", "ImportDataWithProgress", "ImportDataWithProgressOptions", "GetDataRootDirectoryInfo",
 		"ApplyDataRootDirectory", "OpenDataRootDirectory", "SetApplicationBrandIcon",
 	} {
 		_, err := invoker.Invoke(invokeRequest{Namespace: "app", Receiver: "app", Method: method})