fix: address code review comments from Copilot

## Performance Improvements
- Optimize readCSVFile to skip rows before startRow instead of reading all rows
- This significantly improves performance for parallel processing with large files

## Thread Safety
- Add sync.Once to GetSourceReadRowsCount for thread-safe caching
- Prevents race conditions during concurrent initialization

## Error Handling
- Add support for > operator in parseRowCondition (in addition to >=)
- Add explicit error messages for missing operators in conditions
- Improve error handling for invalid condition formats

## Data Validation
- Add column header validation for multiple CSV files
- Ensure all files have matching schemas before processing
- Prevent data corruption from mismatched columns

## Feature Enhancements
- Add support for custom CSV delimiters via sourceCSVDelimiter config
- Support tab-separated files and other delimiter formats
- Document empty string handling in convertCSVValue

## Changes
- source/csv.go: All improvements listed above
- config/config.go: Add sourceCSVDelimiter field

All unit tests pass successfully.

Generated with [Claude Code](https://claude.ai/code)
via [Happy](https://happy.engineering)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
Co-Authored-By: Happy <yesreply@happy.engineering>

Author: 3 people

config/config.go

@@ -52,7 +52,8 @@ type Config struct {
 	TimeSplitUnit      string `json:"TimeSplitUnit" default:"hour"` // time split unit, default is hour, option is: minute, hour, day
 
 	// CSV specific configuration
-	SourceCSVPath string `json:"sourceCSVPath"` // path to CSV file or directory containing CSV files
+	SourceCSVPath      string `json:"sourceCSVPath"`      // path to CSV file or directory containing CSV files
+	SourceCSVDelimiter string `json:"sourceCSVDelimiter"` // CSV delimiter (default: comma). Use "tab" for tab-separated files
 
 	// Databend configuration
 	DatabendDSN      string `json:"databendDSN" default:"localhost:8000"`

source/csv.go

@@ -9,6 +9,7 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/sirupsen/logrus"
@@ -23,6 +24,8 @@ type CSVSource struct {
 	currentFile   string   // Current file being processed
 	columns       []string // Column names from CSV header
 	totalRows     int      // Total rows across all files
+	totalRowsOnce sync.Once // Ensures totalRows is computed only once
+	totalRowsErr  error    // Error from computing totalRows
 }
 
 func NewCSVSource(cfg *config.Config) (*CSVSource, error) {
@@ -92,33 +95,47 @@ func (s *CSVSource) AdjustBatchSizeAccordingToSourceDbTable() uint64 {
 }
 
 // GetSourceReadRowsCount returns the total number of rows in all CSV files
+// This method is thread-safe and caches the result after first call
 func (s *CSVSource) GetSourceReadRowsCount() (int, error) {
-	if s.totalRows > 0 {
-		return s.totalRows, nil
-	}
-
-	totalRows := 0
-	for _, file := range s.files {
-		count, err := countCSVRows(file)
-		if err != nil {
-			return 0, fmt.Errorf("failed to count rows in %s: %w", file, err)
+	s.totalRowsOnce.Do(func() {
+		totalRows := 0
+		for _, file := range s.files {
+			count, err := s.countCSVRows(file)
+			if err != nil {
+				s.totalRowsErr = fmt.Errorf("failed to count rows in %s: %w", file, err)
+				return
+			}
+			totalRows += count
 		}
-		totalRows += count
-	}
+		s.totalRows = totalRows
+	})
 
-	s.totalRows = totalRows
-	return totalRows, nil
+	if s.totalRowsErr != nil {
+		return 0, s.totalRowsErr
+	}
+	return s.totalRows, nil
 }
 
 // countCSVRows counts the number of data rows in a CSV file (excluding header)
-func countCSVRows(filename string) (int, error) {
+func (s *CSVSource) countCSVRows(filename string) (int, error) {
 	file, err := os.Open(filename)
 	if err != nil {
 		return 0, err
 	}
 	defer file.Close()
 
 	reader := csv.NewReader(file)
+
+	// Configure CSV delimiter if specified
+	if s.cfg.SourceCSVDelimiter != "" {
+		delimiter := s.cfg.SourceCSVDelimiter
+		if delimiter == "tab" || delimiter == "\\t" {
+			reader.Comma = '\t'
+		} else if len(delimiter) > 0 {
+			reader.Comma = rune(delimiter[0])
+		}
+	}
+
 	count := 0
 
 	// Skip header
@@ -209,7 +226,18 @@ func (s *CSVSource) QueryTableData(threadNum int, conditionSql string) ([][]inte
 		}
 
 		if len(columns) == 0 {
+			// Use the first non-empty file's columns as the reference schema
 			columns = cols
+		} else if len(cols) > 0 {
+			// Validate that subsequent files have the same columns
+			if len(cols) != len(columns) {
+				return nil, nil, fmt.Errorf("header mismatch in file %s: expected %d columns, got %d columns", file, len(columns), len(cols))
+			}
+			for i := range columns {
+				if columns[i] != cols[i] {
+					return nil, nil, fmt.Errorf("header mismatch in file %s at column %d: expected %q, got %q", file, i, columns[i], cols[i])
+				}
+			}
 		}
 
 		allData = append(allData, data...)
@@ -229,6 +257,7 @@ func (s *CSVSource) QueryTableData(threadNum int, conditionSql string) ([][]inte
 }
 
 // readCSVFile reads a specific range of rows from a CSV file
+// Optimized to skip rows before startRow to improve performance for parallel processing
 func (s *CSVSource) readCSVFile(filename string, startRow, endRow, currentRow uint64) ([][]interface{}, []string, uint64, error) {
 	file, err := os.Open(filename)
 	if err != nil {
@@ -238,6 +267,16 @@ func (s *CSVSource) readCSVFile(filename string, startRow, endRow, currentRow ui
 
 	reader := csv.NewReader(file)
 
+	// Configure CSV delimiter if specified
+	if s.cfg.SourceCSVDelimiter != "" {
+		delimiter := s.cfg.SourceCSVDelimiter
+		if delimiter == "tab" || delimiter == "\\t" {
+			reader.Comma = '\t'
+		} else if len(delimiter) > 0 {
+			reader.Comma = rune(delimiter[0])
+		}
+	}
+
 	// Read header
 	header, err := reader.Read()
 	if err != nil {
@@ -247,8 +286,21 @@ func (s *CSVSource) readCSVFile(filename string, startRow, endRow, currentRow ui
 	var data [][]interface{}
 	rowNum := currentRow
 
-	// Read all rows
-	for {
+	// Skip rows before startRow for better performance
+	for rowNum < startRow {
+		_, err := reader.Read()
+		if err == io.EOF {
+			// Reached end of file before startRow
+			return data, header, rowNum - 1, nil
+		}
+		if err != nil {
+			return nil, nil, 0, fmt.Errorf("failed to skip row: %w", err)
+		}
+		rowNum++
+	}
+
+	// Read rows in the desired range
+	for rowNum < endRow {
 		record, err := reader.Read()
 		if err == io.EOF {
 			break
@@ -257,29 +309,27 @@ func (s *CSVSource) readCSVFile(filename string, startRow, endRow, currentRow ui
 			return nil, nil, 0, fmt.Errorf("failed to read row: %w", err)
 		}
 
-		// Check if this row is in the desired range
-		if rowNum >= startRow && rowNum < endRow {
-			// Convert string values to interface{}
-			row := make([]interface{}, len(record))
-			for i, val := range record {
-				row[i] = convertCSVValue(val)
-			}
-			data = append(data, row)
+		// Convert string values to interface{}
+		row := make([]interface{}, len(record))
+		for i, val := range record {
+			row[i] = convertCSVValue(val)
 		}
-
+		data = append(data, row)
 		rowNum++
-
-		// If we've passed the end row, stop reading this file
-		if rowNum >= endRow {
-			break
-		}
 	}
 
 	return data, header, rowNum - 1, nil
 }
 
 // convertCSVValue attempts to convert CSV string values to appropriate types
+// Empty strings are returned as empty strings (not nil) to maintain consistency
 func convertCSVValue(val string) interface{} {
+	// Handle empty strings - return as-is
+	// Note: Empty CSV cells will be imported as empty strings in Databend
+	if val == "" {
+		return val
+	}
+
 	// Try to parse as integer
 	if intVal, err := strconv.ParseInt(val, 10, 64); err == nil {
 		return intVal
@@ -303,7 +353,8 @@ func convertCSVValue(val string) interface{} {
 }
 
 // parseRowCondition parses a condition like "(row_num >= 1 and row_num < 1001)"
-// and returns the start and end row numbers
+// Supports both >= and > for start condition, and both < and <= for end condition
+// Returns the start and end row numbers
 func parseRowCondition(condition string) (uint64, uint64, error) {
 	// Remove parentheses and split by "and"
 	condition = strings.Trim(condition, "()")
@@ -316,7 +367,7 @@ func parseRowCondition(condition string) (uint64, uint64, error) {
 	var startRow, endRow uint64
 	var err error
 
-	// Parse first part (e.g., "row_num >= 1")
+	// Parse first part (e.g., "row_num >= 1" or "row_num > 0")
 	if strings.Contains(parts[0], ">=") {
 		fields := strings.Split(parts[0], ">=")
 		if len(fields) != 2 {
@@ -326,9 +377,21 @@ func parseRowCondition(condition string) (uint64, uint64, error) {
 		if err != nil {
 			return 0, 0, fmt.Errorf("failed to parse start row: %w", err)
 		}
+	} else if strings.Contains(parts[0], ">") {
+		fields := strings.Split(parts[0], ">")
+		if len(fields) != 2 {
+			return 0, 0, fmt.Errorf("invalid start condition: %s", parts[0])
+		}
+		startRow, err = strconv.ParseUint(strings.TrimSpace(fields[1]), 10, 64)
+		if err != nil {
+			return 0, 0, fmt.Errorf("failed to parse start row: %w", err)
+		}
+		startRow++ // Convert > to >=
+	} else {
+		return 0, 0, fmt.Errorf("invalid start condition (missing >= or >): %s", parts[0])
 	}
 
-	// Parse second part (e.g., "row_num < 1001")
+	// Parse second part (e.g., "row_num < 1001" or "row_num <= 1000")
 	if strings.Contains(parts[1], "<=") {
 		fields := strings.Split(parts[1], "<=")
 		if len(fields) != 2 {
@@ -348,6 +411,8 @@ func parseRowCondition(condition string) (uint64, uint64, error) {
 		if err != nil {
 			return 0, 0, fmt.Errorf("failed to parse end row: %w", err)
 		}
+	} else {
+		return 0, 0, fmt.Errorf("invalid end condition (missing < or <=): %s", parts[1])
 	}
 
 	return startRow, endRow, nil