Merge branch 'master' into default-values

integration_tests/mysql/main.go

@@ -66,7 +66,7 @@ func readTable(db *sql.DB, dbName, tableName string, batchSize int) ([]kafkalib.
 		BatchSize: uint(batchSize),
 	}
 
-	dbzAdapter, err := adapter.NewMySQLAdapter(db, dbName, tableCfg)
+	dbzAdapter, err := adapter.NewMySQLAdapter(db, dbName, tableCfg, nil)
 	if err != nil {
 		return nil, err
 	}

lib/mysql/schema/schema.go

@@ -107,7 +107,7 @@ func GetCreateTableDDL(db *sql.DB, table string) (string, error) {
 	return createTableDDL, nil
 }
 
-func ParseColumnDataType(originalS string, defaultValue *string, optionalSQLMode []string) (DataType, *Opts, error) {
+func ParseColumnDataType(originalS string, defaultValue *string, sqlMode []string) (DataType, *Opts, error) {
 	// Preserve the original value, so we can return the error message without the actual value being mutated.
 	s := strings.ToLower(originalS)
 	var metadata string
@@ -190,7 +190,7 @@ func ParseColumnDataType(originalS string, defaultValue *string, optionalSQLMode
 		return Float, nil, nil
 	case "real":
 		// https://dev.mysql.com/doc/refman/8.4/en/sql-mode.html#sqlmode_real_as_float
-		if slices.Contains(optionalSQLMode, "REAL_AS_FLOAT") {
+		if slices.Contains(sqlMode, "REAL_AS_FLOAT") {
 			return Float, nil, nil
 		}
 		return Double, nil, nil

lib/mysql/table.go

@@ -17,7 +17,7 @@ type Table struct {
 	PrimaryKeys []string
 }
 
-func LoadTable(db *sql.DB, name string) (*Table, error) {
+func LoadTable(db *sql.DB, name string, sqlMode []string) (*Table, error) {
 	tbl := &Table{
 		Name: name,
 	}
@@ -43,8 +43,13 @@ func LoadTable(db *sql.DB, name string) (*Table, error) {
 	}
 
 	for _, col := range createTableResult.GetColumns() {
+<<<<<<< HEAD
 		// TODO: Pass in sql mode
 		dataType, opts, err := schema.ParseColumnDataType(col.DataType, col.DefaultValue, nil)
+=======
+		dataType, opts, err := schema.ParseColumnDataType(col.DataType, sqlMode)
+
+>>>>>>> master
 		if err != nil {
 			return nil, fmt.Errorf("failed to parse column data type: %w", err)
 		}

sources/mysql/adapter/adapter.go

@@ -26,9 +26,9 @@ type MySQLAdapter struct {
 	scannerCfg      scan.ScannerConfig
 }
 
-func NewMySQLAdapter(db *sql.DB, dbName string, tableCfg config.MySQLTable) (MySQLAdapter, error) {
+func NewMySQLAdapter(db *sql.DB, dbName string, tableCfg config.MySQLTable, sqlMode []string) (MySQLAdapter, error) {
 	slog.Info("Loading metadata for table")
-	table, err := mysql.LoadTable(db, tableCfg.Name)
+	table, err := mysql.LoadTable(db, tableCfg.Name, sqlMode)
 	if err != nil {
 		return MySQLAdapter{}, fmt.Errorf("failed to load metadata for table %q: %w", tableCfg.Name, err)
 	}

sources/mysql/mysql.go

@@ -36,5 +36,9 @@ func Load(ctx context.Context, cfg config.MySQL) (sources.Source, bool, error) {
 		return stream, true, nil
 	}
 
-	return &Snapshot{cfg: cfg, db: db}, false, nil
+	return &Snapshot{
+		cfg:     cfg,
+		db:      db,
+		sqlMode: settings.SQLMode,
+	}, false, nil
 }

sources/mysql/snapshot.go

@@ -19,8 +19,9 @@ import (
 )
 
 type Snapshot struct {
-	cfg config.MySQL
-	db  *sql.DB
+	cfg     config.MySQL
+	db      *sql.DB
+	sqlMode []string
 }
 
 func (s Snapshot) Close() error {
@@ -40,7 +41,7 @@ func (s Snapshot) snapshotTable(ctx context.Context, writer writers.Writer, tabl
 	logger := slog.With(slog.String("table", tableCfg.Name), slog.String("database", s.cfg.Database))
 	snapshotStartTime := time.Now()
 
-	dbzAdapter, err := adapter.NewMySQLAdapter(s.db, s.cfg.Database, tableCfg)
+	dbzAdapter, err := adapter.NewMySQLAdapter(s.db, s.cfg.Database, tableCfg, s.sqlMode)
 	if err != nil {
 		return fmt.Errorf("failed to create MySQL adapter: %w", err)
 	}