feat: add configuration parsing and job settings structure for YAML migration configuration

cmd/go_migrate/batch-generator.go

@@ -6,7 +6,6 @@ import (
 	"fmt"
 	"time"
 
-	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/config"
 	"github.com/google/uuid"
 )
 
@@ -20,7 +19,7 @@ type Batch struct {
 	RetryCounter          int
 }
 
-func estimateTotalRowsMssql(ctx context.Context, db *sql.DB, tableInfo config.SourceTableInfo) (int64, error) {
+func estimateTotalRowsMssql(ctx context.Context, db *sql.DB, job MigrationJob) (int64, error) {
 	query := `
 SELECT
 	SUM(p.rows) AS count
@@ -34,7 +33,7 @@ GROUP BY t.name`
 	defer cancel()
 
 	var rowsCount int64
-	err := db.QueryRowContext(ctxTimeout, query, sql.Named("schema", tableInfo.Schema), sql.Named("table", tableInfo.Table)).Scan(&rowsCount)
+	err := db.QueryRowContext(ctxTimeout, query, sql.Named("schema", job.Schema), sql.Named("table", job.Table)).Scan(&rowsCount)
 	if err != nil {
 		return 0, err
 	}
@@ -42,7 +41,7 @@ GROUP BY t.name`
 	return rowsCount, nil
 }
 
-func calculateBatchesMssql(ctx context.Context, db *sql.DB, tableInfo config.SourceTableInfo, batchCount int64) ([]Batch, error) {
+func calculateBatchesMssql(ctx context.Context, db *sql.DB, job MigrationJob, batchCount int64) ([]Batch, error) {
 	query := fmt.Sprintf(`
 SELECT
 	MIN([%s]) AS lower_limit,
@@ -50,13 +49,7 @@ SELECT
 FROM
 	(SELECT [%s], NTILE(@batchCount) OVER (ORDER BY [%s]) AS batch_id FROM [%s].[%s]) AS T
 GROUP BY batch_id
-ORDER BY batch_id`,
+ORDER BY batch_id`, job.PrimaryKey, job.PrimaryKey, job.PrimaryKey, job.PrimaryKey, job.Schema, job.Table)
-		tableInfo.PrimaryKey,
-		tableInfo.PrimaryKey,
-		tableInfo.PrimaryKey,
-		tableInfo.PrimaryKey,
-		tableInfo.Schema,
-		tableInfo.Table)
 
 	ctxTimeout, cancel := context.WithTimeout(ctx, time.Second*20)
 	defer cancel()
@@ -91,15 +84,15 @@ ORDER BY batch_id`,
 	return batches, nil
 }
 
-func batchGeneratorMssql(ctx context.Context, db *sql.DB, tableInfo config.SourceTableInfo, rowsPerBatch int64) ([]Batch, error) {
+func batchGeneratorMssql(ctx context.Context, db *sql.DB, job MigrationJob) ([]Batch, error) {
-	rowsCount, err := estimateTotalRowsMssql(ctx, db, tableInfo)
+	rowsCount, err := estimateTotalRowsMssql(ctx, db, job)
 	if err != nil {
 		return nil, err
 	}
 
 	var batchCount int64 = 1
-	if rowsCount > rowsPerBatch {
+	if rowsCount > RowsPerBatch {
-		batchCount = rowsCount / rowsPerBatch
+		batchCount = rowsCount / RowsPerBatch
 	} else {
 		return []Batch{{
 			Id:             uuid.New(),
@@ -108,7 +101,7 @@ func batchGeneratorMssql(ctx context.Context, db *sql.DB, tableInfo config.Sourc
 		}}, nil
 	}
 
-	batches, err := calculateBatchesMssql(ctx, db, tableInfo, batchCount)
+	batches, err := calculateBatchesMssql(ctx, db, job, batchCount)
 	if err != nil {
 		return nil, err
 	}

cmd/go_migrate/build-extract-query.go

@@ -3,11 +3,9 @@ package main
 import (
 	"fmt"
 	"strings"
-
-	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/config"
 )
 
-func buildExtractQueryMssql(sourceDbInfo config.SourceTableInfo, columns []ColumnType, includeRange bool, isMinInclusive bool) string {
+func buildExtractQueryMssql(job MigrationJob, columns []ColumnType, includeRange bool, isMinInclusive bool) string {
 	var sbQuery strings.Builder
 
 	sbQuery.WriteString("SELECT ")
@@ -28,25 +26,25 @@ func buildExtractQueryMssql(sourceDbInfo config.SourceTableInfo, columns []Colum
 		}
 	}
 
-	fmt.Fprintf(&sbQuery, " FROM [%s].[%s]", sourceDbInfo.Schema, sourceDbInfo.Table)
+	fmt.Fprintf(&sbQuery, " FROM [%s].[%s]", job.Schema, job.Table)
 
 	if includeRange {
-		fmt.Fprintf(&sbQuery, " WHERE [%s]", sourceDbInfo.PrimaryKey)
+		fmt.Fprintf(&sbQuery, " WHERE [%s]", job.PrimaryKey)
 		if isMinInclusive {
 			sbQuery.WriteString(" >=")
 		} else {
 			sbQuery.WriteString(" >")
 		}
 
-		fmt.Fprintf(&sbQuery, " @min AND [%s] <= @max", sourceDbInfo.PrimaryKey)
+		fmt.Fprintf(&sbQuery, " @min AND [%s] <= @max", job.PrimaryKey)
 	}
 
-	fmt.Fprintf(&sbQuery, " ORDER BY [%s] ASC", sourceDbInfo.PrimaryKey)
+	fmt.Fprintf(&sbQuery, " ORDER BY [%s] ASC", job.PrimaryKey)
 
 	return sbQuery.String()
 }
 
-func buildExtractQueryPostgres(sourceDbInfo config.SourceTableInfo, columns []ColumnType) string {
+func buildExtractQueryPostgres(job MigrationJob, columns []ColumnType) string {
 	var sbColumns strings.Builder
 
 	if len(columns) == 0 {
@@ -71,5 +69,5 @@ func buildExtractQueryPostgres(sourceDbInfo config.SourceTableInfo, columns []Co
 		}
 	}
 
-	return fmt.Sprintf(`SELECT %s FROM "%s"."%s"  ORDER BY "%s" ASC`, sbColumns.String(), sourceDbInfo.Schema, sourceDbInfo.Table, sourceDbInfo.PrimaryKey)
+	return fmt.Sprintf(`SELECT %s FROM "%s"."%s"  ORDER BY "%s" ASC`, sbColumns.String(), job.Schema, job.Table, job.PrimaryKey)
 }

cmd/go_migrate/extractor.go

@@ -7,10 +7,8 @@ import (
 	"slices"
 	"strings"
 	"sync"
-	"sync/atomic"
 	"time"
 
-	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/config"
 	"github.com/google/uuid"
 	"github.com/jackc/pgx/v5/pgxpool"
 	_ "github.com/microsoft/go-mssqldb"
@@ -29,7 +27,7 @@ type Chunk struct {
 func extractFromMssql(
 	ctx context.Context,
 	db *sql.DB,
-	tableInfo config.SourceTableInfo,
+	job MigrationJob,
 	columns []ColumnType,
 	chunkSize int,
 	chBatchesIn <-chan Batch,
@@ -37,10 +35,9 @@ func extractFromMssql(
 	chErrorsOut chan<- ExtractorError,
 	chJobErrorsOut chan<- JobError,
 	wgActiveBatches *sync.WaitGroup,
-	rowsRead *int64,
 ) {
 	indexPrimaryKey := slices.IndexFunc(columns, func(col ColumnType) bool {
-		return strings.EqualFold(col.name, tableInfo.PrimaryKey)
+		return strings.EqualFold(col.name, job.PrimaryKey)
 	})
 
 	if indexPrimaryKey == -1 {
@@ -71,7 +68,7 @@ func extractFromMssql(
 				return
 			}
 
-			if abort := processBatch(ctx, db, tableInfo, columns, chunkSize, batch, indexPrimaryKey, chChunksOut, chErrorsOut, wgActiveBatches, rowsRead); abort {
+			if abort := processBatch(ctx, db, job, columns, chunkSize, batch, indexPrimaryKey, chChunksOut, chErrorsOut, wgActiveBatches); abort {
 				return
 			}
 		}
@@ -81,7 +78,7 @@ func extractFromMssql(
 func processBatch(
 	ctx context.Context,
 	db *sql.DB,
-	tableInfo config.SourceTableInfo,
+	job MigrationJob,
 	columns []ColumnType,
 	chunkSize int,
 	batch Batch,
@@ -89,9 +86,8 @@ func processBatch(
 	chChunksOut chan<- Chunk,
 	chErrorsOut chan<- ExtractorError,
 	wgActiveBatches *sync.WaitGroup,
-	rowsRead *int64,
 ) (abort bool) {
-	query := buildExtractQueryMssql(tableInfo, columns, batch.ShouldUseRange, batch.IsLowerLimitInclusive)
+	query := buildExtractQueryMssql(job, columns, batch.ShouldUseRange, batch.IsLowerLimitInclusive)
 	log.Debug("Query used to extract data from mssql: ", query)
 
 	var queryArgs []any
@@ -150,8 +146,6 @@ func processBatch(
 				return true
 			}
 
-			atomic.AddInt64(rowsRead, int64(len(rowsChunk)))
-
 			return false
 		}
 
@@ -169,7 +163,6 @@ func processBatch(
 				return true
 			}
 
-			atomic.AddInt64(rowsRead, int64(len(rowsChunk)))
 			rowsChunk = make([]UnknownRowValues, 0, chunkSize)
 			chunkStartTime = time.Now()
 		}
@@ -207,16 +200,14 @@ func processBatch(
 		case <-ctx.Done():
 			return true
 		}
-
-		atomic.AddInt64(rowsRead, int64(len(rowsChunk)))
 	}
 
 	wgActiveBatches.Done()
 	return false
 }
 
-func extractFromPostgres(ctx context.Context, tableInfo config.SourceTableInfo, columns []ColumnType, chunkSize int, db *pgxpool.Pool, out chan<- []UnknownRowValues) error {
+func extractFromPostgres(ctx context.Context, job MigrationJob, columns []ColumnType, chunkSize int, db *pgxpool.Pool, out chan<- []UnknownRowValues) error {
-	query := buildExtractQueryPostgres(tableInfo, columns)
+	query := buildExtractQueryPostgres(job, columns)
 	log.Debug("Query used to extract data from postgres: ", query)
 
 	rows, err := db.Query(ctx, query)
@@ -238,13 +229,13 @@ func extractFromPostgres(ctx context.Context, tableInfo config.SourceTableInfo,
 		if len(rowsChunk) >= chunkSize {
 			out <- rowsChunk
 			rowsChunk = make([]UnknownRowValues, 0, chunkSize)
-			log.Infof("Chunk send... %+v", tableInfo)
+			log.Infof("Chunk send... %+v", job)
 		}
 	}
 
 	if len(rowsChunk) > 0 {
 		out <- rowsChunk
-		log.Infof("Chunk send... %+v", tableInfo)
+		log.Infof("Chunk send... %+v", job)
 	}
 
 	return nil

cmd/go_migrate/inspect-columns.go

@@ -9,7 +9,6 @@ import (
 	"sync"
 	"time"
 
-	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/config"
 	"github.com/jackc/pgx/v5/pgxpool"
 	_ "github.com/microsoft/go-mssqldb"
 	log "github.com/sirupsen/logrus"
@@ -112,7 +111,7 @@ func MapPostgresColumn(column ColumnType, maxLength *int64, precision *int64, sc
 	return column
 }
 
-func GetColumnTypesPostgres(db *pgxpool.Pool, tableInfo config.TargetTableInfo) ([]ColumnType, error) {
+func GetColumnTypesPostgres(db *pgxpool.Pool, migrationJob MigrationJob) ([]ColumnType, error) {
 	query := `
 SELECT 
   c.column_name AS name,
@@ -130,7 +129,7 @@ ORDER BY c.ordinal_position;
 	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
 	defer cancel()
 
-	rows, err := db.Query(ctx, query, tableInfo.Schema, tableInfo.Table)
+	rows, err := db.Query(ctx, query, migrationJob.Schema, migrationJob.Table)
 	if err != nil {
 		return nil, fmt.Errorf("Error querying column types: %w", err)
 	}
@@ -198,7 +197,7 @@ func MapMssqlColumn(column ColumnType) ColumnType {
 	return column
 }
 
-func GetColumnTypesMssql(db *sql.DB, tableInfo config.SourceTableInfo) ([]ColumnType, error) {
+func GetColumnTypesMssql(db *sql.DB, migrationJob MigrationJob) ([]ColumnType, error) {
 	query := `
 SELECT 
 	c.name AS name,
@@ -220,7 +219,7 @@ ORDER BY c.column_id;
 	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
 	defer cancel()
 
-	rows, err := db.QueryContext(ctx, query, sql.Named("schema", tableInfo.Schema), sql.Named("table", tableInfo.Table))
+	rows, err := db.QueryContext(ctx, query, sql.Named("schema", migrationJob.Schema), sql.Named("table", migrationJob.Table))
 	if err != nil {
 		return nil, fmt.Errorf("Error querying column types: %w", err)
 	}
@@ -253,12 +252,7 @@ ORDER BY c.column_id;
 	return colTypes, nil
 }
 
-func GetColumnTypes(
+func GetColumnTypes(sourceDb *sql.DB, targetDb *pgxpool.Pool, migrationJob MigrationJob) ([]ColumnType, []ColumnType, error) {
-	sourceDb *sql.DB,
-	targetDb *pgxpool.Pool,
-	sourceTable config.SourceTableInfo,
-	targetTable config.TargetTableInfo,
-) ([]ColumnType, []ColumnType, error) {
 	var sourceDbErr error
 	var targetDbErr error
 	var sourceColTypes []ColumnType
@@ -266,14 +260,14 @@ func GetColumnTypes(
 	var wg sync.WaitGroup
 
 	wg.Go(func() {
-		sourceColTypes, sourceDbErr = GetColumnTypesMssql(sourceDb, sourceTable)
+		sourceColTypes, sourceDbErr = GetColumnTypesMssql(sourceDb, migrationJob)
 		if sourceDbErr != nil {
 			log.Error("Error (sourceDb): ", sourceDbErr)
 		}
 	})
 
 	wg.Go(func() {
-		targetColTypes, targetDbErr = GetColumnTypesPostgres(targetDb, targetTable)
+		targetColTypes, targetDbErr = GetColumnTypesPostgres(targetDb, migrationJob)
 		if targetDbErr != nil {
 			log.Error("Error (targetDb): ", targetDbErr)
 		}

cmd/go_migrate/loader.go

@@ -6,10 +6,8 @@ import (
 	"errors"
 	"fmt"
 	"sync"
-	"sync/atomic"
 	"time"
 
-	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/config"
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgconn"
 	"github.com/jackc/pgx/v5/pgxpool"
@@ -20,15 +18,14 @@ import (
 func loadRowsPostgres(
 	ctx context.Context,
 	db *pgxpool.Pool,
-	tableInfo config.TargetTableInfo,
+	job MigrationJob,
 	columns []ColumnType,
 	chChunksIn <-chan Chunk,
 	chErrorsOut chan<- LoaderError,
 	chJobErrorsOut chan<- JobError,
 	wgActiveChunks *sync.WaitGroup,
-	rowsLoaded *int64,
 ) {
-	tableId := pgx.Identifier{tableInfo.Schema, tableInfo.Table}
+	tableId := pgx.Identifier{job.Schema, job.Table}
 	colNames := Map(columns, func(col ColumnType) string {
 		return col.name
 	})
@@ -46,7 +43,7 @@ func loadRowsPostgres(
 				return
 			}
 
-			if abort := loadChunkPostgres(ctx, db, tableId, colNames, chunk, chErrorsOut, chJobErrorsOut, wgActiveChunks, rowsLoaded); abort {
+			if abort := loadChunkPostgres(ctx, db, tableId, colNames, chunk, chErrorsOut, chJobErrorsOut, wgActiveChunks); abort {
 				return
 			}
 		}
@@ -62,7 +59,6 @@ func loadChunkPostgres(
 	chErrorsOut chan<- LoaderError,
 	chJobErrorsOut chan<- JobError,
 	wgActiveChunks *sync.WaitGroup,
-	rowsLoaded *int64,
 ) (abort bool) {
 	chunkStartTime := time.Now()
 	_, err := db.CopyFrom(
@@ -79,7 +75,7 @@ func loadChunkPostgres(
 				select {
 				case chJobErrorsOut <- JobError{
 					ShouldCancelJob: true,
-					Msg:             fmt.Sprintf("Fatal error in table %s", identifier.Sanitize()),
+					Msg:             fmt.Sprintf("Fatal data integrity error in table %s", identifier.Sanitize()),
 					Prev:            err,
 				}:
 				case <-ctx.Done():
@@ -102,12 +98,11 @@ func loadChunkPostgres(
 
 	log.Infof("Loaded chunk: %d rows in %v (%.0f rows/sec)", len(chunk.Data), chunkDuration, rowsPerSec)
 
-	atomic.AddInt64(rowsLoaded, int64(len(chunk.Data)))
 	wgActiveChunks.Done()
 	return false
 }
 
-func loadRowsMssql(ctx context.Context, tableInfo config.TargetTableInfo, columns []ColumnType, db *sql.DB, in <-chan []UnknownRowValues) error {
+func loadRowsMssql(ctx context.Context, job MigrationJob, columns []ColumnType, db *sql.DB, in <-chan []UnknownRowValues) error {
 	chunkCount := 0
 	totalRowsLoaded := 0
 
@@ -119,7 +114,7 @@ func loadRowsMssql(ctx context.Context, tableInfo config.TargetTableInfo, column
 			return fmt.Errorf("error starting transaction: %w", err)
 		}
 
-		fullTableName := fmt.Sprintf("[%s].[%s]", tableInfo.Schema, tableInfo.Table)
+		fullTableName := fmt.Sprintf("[%s].[%s]", job.Schema, job.Table)
 		colNames := Map(columns, func(col ColumnType) string {
 			return col.name
 		})
@@ -182,13 +177,14 @@ func Map[T any, V any](input []T, mapper func(T) V) []V {
 	return result
 }
 
-func fakeLoader(tableInfo config.TargetTableInfo, columns []ColumnType, in <-chan [][]any) {
+func fakeLoader(job MigrationJob, columns []ColumnType, in <-chan [][]any) {
+
 	for rows := range in {
 		log.Debugf("Chunk received, loading data into...")
 
 		for i, rowValues := range rows {
 			if i%100 == 0 {
-				logSampleRow(tableInfo.Schema, tableInfo.Table, columns, rowValues, fmt.Sprintf("row %d", i))
+				logSampleRow(job, columns, rowValues, fmt.Sprintf("row %d", i))
 			}
 		}
 	}

cmd/go_migrate/main.go

@@ -2,31 +2,48 @@ package main
 
 import (
 	"context"
-	"database/sql"
-	"sync"
 	"time"
 
-	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/config"
-	"github.com/jackc/pgx/v5/pgxpool"
 	log "github.com/sirupsen/logrus"
 )
 
+type MigrationJob struct {
+	Schema     string
+	Table      string
+	PrimaryKey string
+}
+
+var migrationJobs []MigrationJob = []MigrationJob{
+	{
+		Schema:     "Cartografia",
+		Table:      "MANZANA",
+		PrimaryKey: "GDB_ARCHIVE_OID",
+	},
+	{
+		Schema:     "Red",
+		Table:      "PUERTO",
+		PrimaryKey: "ID_PUERTO",
+	},
+}
+
+const (
+	NumExtractors  int   = 4
+	NumLoaders     int   = 8
+	ChunkSize      int   = 25000
+	QueueSize      int   = 8
+	ChunksPerBatch int   = 16
+	RowsPerBatch   int64 = int64(ChunkSize * ChunksPerBatch)
+)
+
 func main() {
 	configureLog()
-
-	migrationConfig, err := config.ReadMigrationConfig()
-	if err != nil {
-		log.Fatalf("error leyendo configuracion: %v", err)
-	}
-
-	log.Debugf("Config: %+v", migrationConfig)
-
 	startTime := time.Now()
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
 	log.Info("=== Starting migration ===")
+	log.Infof("Number of loaders: %d, Chunk size: %d", NumLoaders, ChunkSize)
 
 	sourceDb, targetDb, connError := connectToDatabases()
 	if connError != nil {
@@ -36,81 +53,12 @@ func main() {
 	defer sourceDb.Close()
 	defer targetDb.Close()
 
-	results := processMigrationJobs(ctx, sourceDb, targetDb, migrationConfig.Jobs, migrationConfig.MaxParallelWorkers)
+	for _, job := range migrationJobs {
-
+		log.Infof(">>> Processing job: %s.%s <<<", job.Schema, job.Table)
-	log.Info("=== RESUMEN DE MIGRACIÓN ===")
+		processMigrationJob(ctx, sourceDb, targetDb, job)
-	var totalProcessed, totalErrors int64
-
-	for _, res := range results {
-		status := "OK"
-		if res.Error != nil {
-			status = "FAILED"
 	}
-		log.Infof("[%s] Status: %s | Read: %d | Loaded: %d | Errors: %d | Time: %v", res.JobName, status, res.RowsRead, res.RowsLoaded, res.RowsFailed, res.Duration)
-
-		totalProcessed += res.RowsLoaded
-		if res.Error != nil {
-			totalErrors++
-		}
-	}
-
-	log.Infof("Migración terminada. Tablas: %d, Errores: %d, Filas totales: %d", len(results), totalErrors, totalProcessed)
 
 	totalDuration := time.Since(startTime)
 	log.Infof("=== Migration completed successfully! ===")
 	log.Infof("Total migration time: %v", totalDuration)
 }
-
-func processMigrationJobs(
-	ctx context.Context,
-	sourceDb *sql.DB,
-	targetDb *pgxpool.Pool,
-	jobs []config.Job,
-	maxParallelWorkers int,
-) []JobResult {
-	if len(jobs) == 0 {
-		log.Info("No migration jobs configured")
-		return []JobResult{}
-	}
-
-	if maxParallelWorkers <= 0 {
-		maxParallelWorkers = 1
-	}
-
-	if maxParallelWorkers > len(jobs) {
-		maxParallelWorkers = len(jobs)
-	}
-
-	log.Infof("Starting migration with %d parallel worker(s)", maxParallelWorkers)
-
-	chJobResults := make(chan JobResult, len(jobs))
-	chJobs := make(chan config.Job, len(jobs))
-	var wgJobs sync.WaitGroup
-
-	for i := range maxParallelWorkers {
-		wgJobs.Go(func() {
-			for job := range chJobs {
-				log.Infof("[worker %d] >>> Processing job: %s.%s <<<", i, job.SourceTable.Schema, job.SourceTable.Table)
-				res := processMigrationJob(ctx, sourceDb, targetDb, job)
-				chJobResults <- res
-			}
-		})
-	}
-
-	for _, job := range jobs {
-		chJobs <- job
-	}
-	close(chJobs)
-
-	go func() {
-		wgJobs.Wait()
-		close(chJobResults)
-	}()
-
-	var finalResults []JobResult
-	for res := range chJobResults {
-		finalResults = append(finalResults, res)
-	}
-
-	return finalResults
-}

cmd/go_migrate/metrics.go

@@ -1,13 +0,0 @@
-package main
-
-import "time"
-
-type JobResult struct {
-	JobName    string
-	StartTime  time.Time
-	Duration   time.Duration
-	RowsRead   int64
-	RowsLoaded int64
-	RowsFailed int64
-	Error      error
-}

cmd/go_migrate/process.go

@@ -4,10 +4,8 @@ import (
 	"context"
 	"database/sql"
 	"sync"
-	"sync/atomic"
 	"time"
 
-	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/config"
 	"github.com/jackc/pgx/v5/pgxpool"
 
 	_ "github.com/microsoft/go-mssqldb"
@@ -18,19 +16,14 @@ func processMigrationJob(
 	ctx context.Context,
 	sourceDb *sql.DB,
 	targetDb *pgxpool.Pool,
-	job config.Job,
+	job MigrationJob,
-) JobResult {
+) {
-	result := JobResult{
+	jobStartTime := time.Now()
-		JobName:   job.Name,
+	log.Infof("Starting migration job: %s.%s [PK: %s]", job.Schema, job.Table, job.PrimaryKey)
-		StartTime: time.Now(),
-	}
 
-	var rowsRead, rowsLoaded, rowsFailed int64
+	sourceColTypes, targetColTypes, err := GetColumnTypes(sourceDb, targetDb, job)
-
-	sourceColTypes, targetColTypes, err := GetColumnTypes(sourceDb, targetDb, job.SourceTable, job.TargetTable)
 	if err != nil {
-		result.Error = err
+		log.Fatal("Unexpected error: ", err)
-		return result
 	}
 
 	logColumnTypes(sourceColTypes, "Source col types")
@@ -39,17 +32,17 @@ func processMigrationJob(
 	jobCtx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
-	batches, err := batchGeneratorMssql(jobCtx, sourceDb, job.SourceTable, job.RowsPerBatch)
+	batches, err := batchGeneratorMssql(jobCtx, sourceDb, job)
 	if err != nil {
 		log.Error("Unexpected error calculating batch ranges: ", err)
 	}
 
-	chJobErrors := make(chan JobError, job.QueueSize)
+	chJobErrors := make(chan JobError, 50)
-	chBatches := make(chan Batch, job.QueueSize)
+	chBatches := make(chan Batch, QueueSize)
-	chExtractorErrors := make(chan ExtractorError, job.QueueSize)
+	chExtractorErrors := make(chan ExtractorError, QueueSize)
-	chChunksRaw := make(chan Chunk, job.QueueSize)
+	chChunksRaw := make(chan Chunk, QueueSize)
-	chChunksTransformed := make(chan Chunk, job.QueueSize)
+	chChunksTransformed := make(chan Chunk, QueueSize)
-	chLoadersErrors := make(chan LoaderError, job.QueueSize)
+	chLoadersErrors := make(chan LoaderError, QueueSize)
 
 	var wgActiveBatches sync.WaitGroup
 	var wgActiveChunks sync.WaitGroup
@@ -60,19 +53,19 @@ func processMigrationJob(
 	go func() {
 		if err := jobErrorHandler(jobCtx, chJobErrors); err != nil {
 			cancel()
-			result.Error = err
 		}
 	}()
 
 	go extractorErrorHandler(jobCtx, chExtractorErrors, chBatches, chJobErrors, &wgActiveBatches)
 	go loaderErrorHandler(jobCtx, chLoadersErrors, chChunksTransformed, chJobErrors, &wgActiveChunks)
 
-	maxExtractors := min(job.MaxExtractors, len(batches))
+	maxExtractors := min(NumExtractors, len(batches))
-	log.Infof("Starting %d extractor(s)...", maxExtractors)
+	log.Infof("Starting %d extractors...", maxExtractors)
+	extractStartTime := time.Now()
 
 	for range maxExtractors {
 		wgExtractors.Go(func() {
-			extractFromMssql(jobCtx, sourceDb, job.SourceTable, sourceColTypes, job.ChunkSize, chBatches, chChunksRaw, chExtractorErrors, chJobErrors, &wgActiveBatches, &rowsRead)
+			extractFromMssql(jobCtx, sourceDb, job, sourceColTypes, ChunkSize, chBatches, chChunksRaw, chExtractorErrors, chJobErrors, &wgActiveBatches)
 		})
 	}
 
@@ -83,7 +76,8 @@ func processMigrationJob(
 		}
 	}()
 
-	log.Infof("Starting %d transformer(s)...", maxExtractors)
+	log.Infof("Starting %d transformers...", maxExtractors)
+	transformStartTime := time.Now()
 
 	for range maxExtractors {
 		wgTransformers.Go(func() {
@@ -91,11 +85,12 @@ func processMigrationJob(
 		})
 	}
 
-	log.Infof("Starting %d loader(s)...", job.MaxLoaders)
+	log.Infof("Starting %d PostgreSQL loader(s)...", NumLoaders)
+	loadStartTime := time.Now()
 
-	for range job.MaxLoaders {
+	for range NumLoaders {
 		wgLoaders.Go(func() {
-			loadRowsPostgres(jobCtx, targetDb, job.TargetTable, targetColTypes, chChunksTransformed, chLoadersErrors, chJobErrors, &wgActiveChunks, &rowsLoaded)
+			loadRowsPostgres(jobCtx, targetDb, job, targetColTypes, chChunksTransformed, chLoadersErrors, chJobErrors, &wgActiveChunks)
 		})
 	}
 
@@ -105,31 +100,24 @@ func processMigrationJob(
 		close(chExtractorErrors)
 
 		wgExtractors.Wait()
+		log.Infof("Extraction completed in %v", time.Since(extractStartTime))
 		close(chChunksRaw)
 
 		wgTransformers.Wait()
+		log.Infof("Transformation completed in %v", time.Since(transformStartTime))
 
 		wgActiveChunks.Wait()
 		close(chChunksTransformed)
 		close(chLoadersErrors)
 
 		wgLoaders.Wait()
+		log.Infof("Loading completed in %v", time.Since(loadStartTime))
 
 		cancel()
 	}()
 
 	<-jobCtx.Done()
-
+	log.Infof("Migration job completed. Total time: %v", time.Since(jobStartTime))
-	if ctx.Err() != nil {
-		result.Error = ctx.Err()
-	}
-
-	result.Duration = time.Since(result.StartTime)
-	result.RowsRead = atomic.LoadInt64(&rowsRead)
-	result.RowsLoaded = atomic.LoadInt64(&rowsLoaded)
-	result.RowsFailed = atomic.LoadInt64(&rowsFailed)
-
-	return result
 }
 
 func logColumnTypes(columnTypes []ColumnType, label string) {
@@ -140,14 +128,8 @@ func logColumnTypes(columnTypes []ColumnType, label string) {
 	}
 }
 
-func logSampleRow(
+func logSampleRow(job MigrationJob, columns []ColumnType, rowValues UnknownRowValues, tag string) {
-	schema string,
+	log.Infof("[%s.%s] Sample row: (%s)", job.Schema, job.Table, tag)
-	table string,
-	columns []ColumnType,
-	rowValues UnknownRowValues,
-	tag string,
-) {
-	log.Infof("[%s.%s] Sample row: (%s)", schema, table, tag)
 	for i, col := range columns {
 		log.Infof("%s (%T): %v", col.Name(), rowValues[i], rowValues[i])
 	}

config.yaml

@@ -1,11 +1,11 @@
-max_parallel_workers: 4
+max_parallel_workers: 2
 
 defaults:
-  max_extractors: 2
+  max_extractors: 4
-  max_loaders: 4
+  max_loaders: 8
   queue_size: 8
-  chunk_size: 25000
+  chunk_size: 50000
-  chunks_per_batch: 8
+  chunks_per_batch: 10
   truncate_target: true
   truncate_method: TRUNCATE # TRUNCATE | DELETE
   retry:
@@ -21,6 +21,19 @@ jobs:
     target:
       schema: Cartografia
       table: MANZANA
+    max_extractors: 2 # overrides default config
+    max_loaders: 4 # overrides default config
+    queue_size: 4 # overrides default config
+    chunk_size: 25000 # overrides default config
+    chunks_per_batch: 8 # overrides default config
+    truncate_target: false # overrides default config
+    truncate_method: DELETE # overrides default config
+    retry:
+      attempts: 5 # overrides default config
+    pre_sql:
+      - "SELECT 1"
+    post_sql:
+      - "SELECT 2"
 
   - name: red_puerto
     enabled: true

go.mod

@@ -4,13 +4,13 @@ go 1.25.7
 
 require (
 	github.com/gaspardle/go-mssqlclrgeo v0.0.0-20160129143314-97ceabf987a4
+	github.com/goccy/go-yaml v1.19.2
 	github.com/google/uuid v1.6.0
 	github.com/jackc/pgx/v5 v5.9.1
 	github.com/joho/godotenv v1.5.1
 	github.com/microsoft/go-mssqldb v1.9.8
 	github.com/sirupsen/logrus v1.9.4
 	github.com/twpayne/go-geom v1.6.1
-	gopkg.in/yaml.v3 v3.0.1
 )
 
 require (
@@ -19,8 +19,6 @@ require (
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
 	github.com/jackc/puddle/v2 v2.2.2 // indirect
-	github.com/kr/text v0.2.0 // indirect
-	github.com/rogpeppe/go-internal v1.14.1 // indirect
 	github.com/shopspring/decimal v1.4.0 // indirect
 	golang.org/x/crypto v0.48.0 // indirect
 	golang.org/x/sync v0.19.0 // indirect

go.sum

@@ -16,12 +16,13 @@ github.com/alecthomas/assert/v2 v2.10.0 h1:jjRCHsj6hBJhkmhznrCzoNpbA3zqy0fYiUcYZ
 github.com/alecthomas/assert/v2 v2.10.0/go.mod h1:Bze95FyfUr7x34QZrjL+XP+0qgp/zg8yS+TtBj1WA3k=
 github.com/alecthomas/repr v0.4.0 h1:GhI2A8MACjfegCPVq9f1FLvIBS+DrQ2KQBFZP1iFzXc=
 github.com/alecthomas/repr v0.4.0/go.mod h1:Fr0507jx4eOXV7AlPV6AVZLYrLIuIeSOWtW57eE/O/4=
-github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/gaspardle/go-mssqlclrgeo v0.0.0-20160129143314-97ceabf987a4 h1:4vH4+3zfwZTqoJEFw7DsTaH1V8jgVwnyeDvNi2TxzAc=
 github.com/gaspardle/go-mssqlclrgeo v0.0.0-20160129143314-97ceabf987a4/go.mod h1:jlB0I5BIfcJBGdV6rRGPthSBfeY86RGkSAwcsldbHJc=
+github.com/goccy/go-yaml v1.19.2 h1:PmFC1S6h8ljIz6gMRBopkjP1TVT7xuwrButHID66PoM=
+github.com/goccy/go-yaml v1.19.2/go.mod h1:XBurs7gK8ATbW4ZPGKgcbrY1Br56PdM69F7LkFRi1kA=
 github.com/golang-jwt/jwt/v5 v5.3.1 h1:kYf81DTWFe7t+1VvL7eS+jKFVWaUnK9cB1qbwn63YCY=
 github.com/golang-jwt/jwt/v5 v5.3.1/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArsqaEUEa5bE=
 github.com/golang-sql/civil v0.0.0-20220223132316-b832511892a9 h1:au07oEsX2xN0ktxqI+Sida1w446QrXBRJ0nee3SNZlA=
@@ -42,10 +43,6 @@ github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo
 github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
 github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0=
 github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4=
-github.com/kr/pretty v0.3.0 h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0=
-github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk=
-github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
-github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
 github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
 github.com/microsoft/go-mssqldb v1.9.8 h1:d4IFMvF/o+HdpXUqbBfzHvn/NlFA75YGcfHUUvDFJEM=
@@ -54,8 +51,6 @@ github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c h1:+mdjkGKdHQG3305AYmd
 github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c/go.mod h1:7rwL4CYBLnjLxUqIJNnCWiEdr3bn6IUYi15bNlnbCCU=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
-github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
 github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp81k=
 github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME=
 github.com/sirupsen/logrus v1.9.4 h1:TsZE7l11zFCLZnZ+teH4Umoq5BhEIfIzfRDZ1Uzql2w=
@@ -78,8 +73,6 @@ golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
 golang.org/x/text v0.34.0 h1:oL/Qq0Kdaqxa1KbNeMKwQq0reLCCaFtqu2eNuSeNHbk=
 golang.org/x/text v0.34.0/go.mod h1:homfLqTYRFyVYemLBFl5GgL/DWEiH5wcsQ5gSh1yziA=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
-gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

internal/app/config/migration.go

@@ -1,118 +0,0 @@
-package config
-
-import (
-	"fmt"
-	"os"
-
-	"gopkg.in/yaml.v3"
-)
-
-type RetryConfig struct {
-	Attempts int `yaml:"attempts"`
-}
-
-type JobConfig struct {
-	MaxExtractors  int `yaml:"max_extractors"`
-	MaxLoaders     int `yaml:"max_loaders"`
-	QueueSize      int `yaml:"queue_size"`
-	ChunkSize      int `yaml:"chunk_size"`
-	ChunksPerBatch int `yaml:"chunks_per_batch"`
-	RowsPerBatch   int64
-	TruncateTarget bool        `yaml:"truncate_target"`
-	TruncateMethod string      `yaml:"truncate_method"`
-	Retry          RetryConfig `yaml:"retry"`
-}
-
-type TargetTableInfo struct {
-	Schema string `yaml:"schema"`
-	Table  string `yaml:"table"`
-}
-
-type SourceTableInfo struct {
-	Schema     string `yaml:"schema"`
-	Table      string `yaml:"table"`
-	PrimaryKey string `yaml:"primary_key"`
-}
-
-type Job struct {
-	Name        string          `yaml:"name"`
-	Enabled     bool            `yaml:"enabled"`
-	SourceTable SourceTableInfo `yaml:"source"`
-	TargetTable TargetTableInfo `yaml:"target"`
-	PreSQL      []string        `yaml:"pre_sql"`
-	PostSQL     []string        `yaml:"post_sql"`
-	JobConfig   `yaml:",inline"`
-}
-
-type MigrationConfig struct {
-	MaxParallelWorkers int       `yaml:"max_parallel_workers"`
-	Defaults           JobConfig `yaml:"defaults"`
-	Jobs               []Job     `yaml:"jobs"`
-}
-
-type rawConfig struct {
-	MaxParallelWorkers int         `yaml:"max_parallel_workers"`
-	Defaults           JobConfig   `yaml:"defaults"`
-	Jobs               []yaml.Node `yaml:"jobs"`
-}
-
-func (c *MigrationConfig) UnmarshalYAML(value *yaml.Node) error {
-	var raw rawConfig
-	if err := value.Decode(&raw); err != nil {
-		return err
-	}
-
-	c.MaxParallelWorkers = raw.MaxParallelWorkers
-	c.Defaults = raw.Defaults
-	c.Defaults.RowsPerBatch = int64(raw.Defaults.ChunkSize * raw.Defaults.ChunksPerBatch)
-
-	for _, node := range raw.Jobs {
-		job := Job{
-			JobConfig: raw.Defaults,
-		}
-
-		if err := node.Decode(&job); err != nil {
-			return err
-		}
-
-		job.RowsPerBatch = int64(job.ChunkSize * job.ChunksPerBatch)
-
-		c.Jobs = append(c.Jobs, job)
-	}
-
-	return nil
-}
-
-const defaultConfigFileName string = "config.yaml"
-
-func filenamesOrDefault(filenames []string) []string {
-	if len(filenames) == 0 {
-		return []string{defaultConfigFileName}
-	}
-	return filenames
-}
-
-func ReadMigrationConfig(filenames ...string) (MigrationConfig, error) {
-	filenames = filenamesOrDefault(filenames)
-	var data []byte
-	var err error
-
-	for _, filename := range filenames {
-		data, err = os.ReadFile(filename)
-		if err != nil {
-			continue
-		}
-		break
-	}
-
-	if err != nil {
-		return MigrationConfig{}, fmt.Errorf("Error reading config file: %v", err)
-	}
-
-	var config MigrationConfig
-	if err := yaml.Unmarshal(data, &config); err != nil {
-		return MigrationConfig{}, fmt.Errorf("Error parsing config file: %v", err)
-	}
-
-	return config, nil
-}

internal/app/convert/main.go

@@ -1,18 +0,0 @@
-package convert
-
-func ToInt64(v any) (int64, bool) {
-	switch t := v.(type) {
-	case int:
-		return int64(t), true
-	case int8:
-		return int64(t), true
-	case int16:
-		return int64(t), true
-	case int32:
-		return int64(t), true
-	case int64:
-		return int64(t), true
-	default:
-		return 0, false
-	}
-}

internal/app/custom_errors/extractor.error.go

@@ -1,80 +0,0 @@
-package custom_errors
-
-import (
-	"context"
-	"fmt"
-	"sync"
-
-	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/models"
-	"github.com/google/uuid"
-)
-
-type ExtractorError struct {
-	Batch     models.Batch
-	LastId    int64
-	HasLastId bool
-	Msg       string
-}
-
-func (e *ExtractorError) Error() string {
-	return e.Msg
-}
-
-const maxRetryAttempts = 3
-
-func ExtractorErrorHandler(
-	ctx context.Context,
-	chErrorsIn <-chan ExtractorError,
-	chBatchesOut chan<- models.Batch,
-	chJobErrorsOut chan<- JobError,
-	wgActiveBatches *sync.WaitGroup,
-) {
-	for {
-		if ctx.Err() != nil {
-			return
-		}
-
-		select {
-		case <-ctx.Done():
-			return
-
-		case err, ok := <-chErrorsIn:
-			if !ok {
-				return
-			}
-
-			if err.Batch.RetryCounter >= maxRetryAttempts {
-				jobError := JobError{
-					ShouldCancelJob: false,
-					Msg:             fmt.Sprintf("batch %v reached max retries (%d)", err.Batch.Id, maxRetryAttempts),
-					Prev:            &err,
-				}
-
-				select {
-				case chJobErrorsOut <- jobError:
-				case <-ctx.Done():
-					return
-				}
-
-				wgActiveBatches.Done()
-				continue
-			}
-
-			newBatch := err.Batch
-			newBatch.RetryCounter++
-
-			if err.HasLastId {
-				newBatch.ParentId = err.Batch.Id
-				newBatch.Id = uuid.New()
-				newBatch.LowerLimit = err.LastId
-				newBatch.IsLowerLimitInclusive = false
-			}
-
-			select {
-			case chBatchesOut <- newBatch:
-			case <-ctx.Done():
-				return
-			}
-		}
-	}
-}

internal/app/custom_errors/job.error.go

@@ -1,47 +0,0 @@
-package custom_errors
-
-import (
-	"context"
-	"fmt"
-
-	log "github.com/sirupsen/logrus"
-)
-
-type JobError struct {
-	ShouldCancelJob bool
-	Msg             string
-	Prev            error
-}
-
-func (e *JobError) Error() string {
-	if e.Prev != nil {
-		return fmt.Sprintf("%s: %v", e.Msg, e.Prev)
-	}
-
-	return e.Msg
-}
-
-func JobErrorHandler(ctx context.Context, chErrorsIn <-chan JobError) error {
-	for {
-		if ctx.Err() != nil {
-			return nil
-		}
-
-		select {
-		case <-ctx.Done():
-			return nil
-
-		case err, ok := <-chErrorsIn:
-			if !ok {
-				return nil
-			}
-
-			if err.ShouldCancelJob {
-				log.Error(err.Msg, " - ", err.Prev)
-				return &err
-			}
-
-			log.Error(err.Msg, " - ", err.Prev)
-		}
-	}
-}

internal/app/etl/extractor/main.go

@@ -1,36 +0,0 @@
-package extractor
-
-import (
-	"context"
-	"sync"
-
-	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/config"
-	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/custom_errors"
-	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/models"
-)
-
-type Extractor interface {
-	ProcessBatch(
-		ctx context.Context,
-		tableInfo config.SourceTableInfo,
-		columns []models.ColumnType,
-		chunkSize int,
-		batch models.Batch,
-		indexPrimaryKey int,
-		chChunksOut chan<- models.Chunk,
-		rowsRead *int64,
-	) error
-
-	Exec(
-		ctx context.Context,
-		tableInfo config.SourceTableInfo,
-		columns []models.ColumnType,
-		chunkSize int,
-		chBatchesIn <-chan models.Batch,
-		chChunksOut chan<- models.Chunk,
-		chErrorsOut chan<- custom_errors.ExtractorError,
-		chJobErrorsOut chan<- custom_errors.JobError,
-		wgActiveBatches *sync.WaitGroup,
-		rowsRead *int64,
-	)
-}

internal/app/etl/extractor/mssql.go

@@ -1,269 +0,0 @@
-package extractor
-
-import (
-	"context"
-	"database/sql"
-	"errors"
-	"fmt"
-	"slices"
-	"strings"
-	"sync"
-	"sync/atomic"
-
-	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/config"
-	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/convert"
-	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/custom_errors"
-	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/models"
-	"github.com/google/uuid"
-)
-
-type MssqlExtractor struct {
-	db *sql.DB
-}
-
-func NewMssqlExtractor(db *sql.DB) *MssqlExtractor {
-	return &MssqlExtractor{db: db}
-}
-
-func buildExtractQueryMssql(
-	tableInfo config.SourceTableInfo,
-	columns []models.ColumnType,
-	includeRange bool,
-	isMinInclusive bool,
-) string {
-	var sbQuery strings.Builder
-
-	sbQuery.WriteString("SELECT ")
-
-	if len(columns) == 0 {
-		sbQuery.WriteString("*")
-	} else {
-		for i, col := range columns {
-			fmt.Fprintf(&sbQuery, "[%s]", col.Name())
-
-			if col.Type() == "GEOMETRY" {
-				fmt.Fprintf(&sbQuery, ".STAsBinary() AS [%s]", col.Name())
-			}
-
-			if i < len(columns)-1 {
-				sbQuery.WriteString(", ")
-			}
-		}
-	}
-
-	fmt.Fprintf(&sbQuery, " FROM [%s].[%s]", tableInfo.Schema, tableInfo.Table)
-
-	if includeRange {
-		fmt.Fprintf(&sbQuery, " WHERE [%s]", tableInfo.PrimaryKey)
-		if isMinInclusive {
-			sbQuery.WriteString(" >=")
-		} else {
-			sbQuery.WriteString(" >")
-		}
-
-		fmt.Fprintf(&sbQuery, " @min AND [%s] <= @max", tableInfo.PrimaryKey)
-	}
-
-	fmt.Fprintf(&sbQuery, " ORDER BY [%s] ASC", tableInfo.PrimaryKey)
-
-	return sbQuery.String()
-}
-
-func extractorErrorFromLastRowMssql(
-	lastRow models.UnknownRowValues,
-	indexPrimaryKey int,
-	batch *models.Batch,
-	previousError error,
-) *custom_errors.ExtractorError {
-	lastIdRawValue := lastRow[indexPrimaryKey]
-
-	lastId, ok := convert.ToInt64(lastIdRawValue)
-	if !ok {
-		currentBatch := *batch
-		currentBatch.RetryCounter = 3
-		return &custom_errors.ExtractorError{
-			Batch:     currentBatch,
-			HasLastId: true,
-			Msg:       fmt.Sprintf("Couldn't cast last id value as int: %s", previousError.Error()),
-		}
-
-	}
-
-	return &custom_errors.ExtractorError{
-		Batch:     *batch,
-		HasLastId: true,
-		LastId:    lastId,
-		Msg:       previousError.Error(),
-	}
-}
-
-func (mssqlEx *MssqlExtractor) ProcessBatch(
-	ctx context.Context,
-	tableInfo config.SourceTableInfo,
-	columns []models.ColumnType,
-	chunkSize int,
-	batch models.Batch,
-	indexPrimaryKey int,
-	chChunksOut chan<- models.Chunk,
-	rowsRead *int64,
-) error {
-	query := buildExtractQueryMssql(tableInfo, columns, batch.ShouldUseRange, batch.IsLowerLimitInclusive)
-
-	var queryArgs []any
-	if batch.ShouldUseRange {
-		queryArgs = append(queryArgs,
-			sql.Named("min", batch.LowerLimit),
-			sql.Named("max", batch.UpperLimit),
-		)
-	}
-
-	rows, err := mssqlEx.db.QueryContext(ctx, query, queryArgs...)
-	if err != nil {
-		return &custom_errors.ExtractorError{Batch: batch, HasLastId: false, Msg: err.Error()}
-	}
-	defer rows.Close()
-
-	rowsChunk := make([]models.UnknownRowValues, 0, chunkSize)
-
-	for rows.Next() {
-		values := make([]any, len(columns))
-		scanArgs := make([]any, len(columns))
-
-		for i := range values {
-			scanArgs[i] = &values[i]
-		}
-
-		if err := rows.Scan(scanArgs...); err != nil {
-			if len(rowsChunk) == 0 {
-				return &custom_errors.ExtractorError{Batch: batch, HasLastId: false, Msg: err.Error()}
-			}
-
-			lastRow := rowsChunk[len(rowsChunk)-1]
-
-			select {
-			case chChunksOut <- models.Chunk{Id: uuid.New(), BatchId: batch.Id, Data: rowsChunk, RetryCounter: 0}:
-			case <-ctx.Done():
-				return nil
-			}
-
-			atomic.AddInt64(rowsRead, int64(len(rowsChunk)))
-
-			return extractorErrorFromLastRowMssql(lastRow, indexPrimaryKey, &batch, err)
-		}
-
-		rowsChunk = append(rowsChunk, values)
-
-		if len(rowsChunk) >= chunkSize {
-			select {
-			case chChunksOut <- models.Chunk{Id: uuid.New(), BatchId: batch.Id, Data: rowsChunk, RetryCounter: 0}:
-			case <-ctx.Done():
-				return nil
-			}
-
-			atomic.AddInt64(rowsRead, int64(len(rowsChunk)))
-			rowsChunk = make([]models.UnknownRowValues, 0, chunkSize)
-		}
-	}
-
-	if err := rows.Err(); err != nil {
-		if errors.Is(err, ctx.Err()) {
-			return ctx.Err()
-		}
-
-		if len(rowsChunk) == 0 {
-			return &custom_errors.ExtractorError{Batch: batch, HasLastId: false, Msg: err.Error()}
-		}
-
-		lastRow := rowsChunk[len(rowsChunk)-1]
-		return extractorErrorFromLastRowMssql(lastRow, indexPrimaryKey, &batch, err)
-	}
-
-	if len(rowsChunk) > 0 {
-		select {
-		case chChunksOut <- models.Chunk{Id: uuid.New(), BatchId: batch.Id, Data: rowsChunk, RetryCounter: 0}:
-		case <-ctx.Done():
-			return nil
-		}
-
-		atomic.AddInt64(rowsRead, int64(len(rowsChunk)))
-	}
-
-	return nil
-}
-
-func (mssqlEx *MssqlExtractor) Exec(
-	ctx context.Context,
-	tableInfo config.SourceTableInfo,
-	columns []models.ColumnType,
-	chunkSize int,
-	chBatchesIn <-chan models.Batch,
-	chChunksOut chan<- models.Chunk,
-	chErrorsOut chan<- custom_errors.ExtractorError,
-	chJobErrorsOut chan<- custom_errors.JobError,
-	wgActiveBatches *sync.WaitGroup,
-	rowsRead *int64,
-) {
-	indexPrimaryKey := slices.IndexFunc(columns, func(col models.ColumnType) bool {
-		return strings.EqualFold(col.Name(), tableInfo.PrimaryKey)
-	})
-
-	if indexPrimaryKey == -1 {
-		select {
-		case <-ctx.Done():
-			return
-		case chJobErrorsOut <- custom_errors.JobError{
-			ShouldCancelJob: true,
-			Msg:             "Primary key not found in provided columns",
-		}:
-		}
-
-		return
-	}
-
-	for {
-		if ctx.Err() != nil {
-			return
-		}
-
-		select {
-		case <-ctx.Done():
-			return
-		case batch, ok := <-chBatchesIn:
-			if !ok {
-				return
-			}
-
-			err := mssqlEx.ProcessBatch(
-				ctx,
-				tableInfo,
-				columns,
-				chunkSize,
-				batch,
-				indexPrimaryKey,
-				chChunksOut,
-				rowsRead,
-			)
-
-			if err != nil {
-				var exError *custom_errors.ExtractorError
-				if errors.As(err, &exError) {
-					select {
-					case <-ctx.Done():
-						return
-					case chErrorsOut <- *exError:
-					}
-				}
-
-				select {
-				case <-ctx.Done():
-					return
-				case chJobErrorsOut <- custom_errors.JobError{ShouldCancelJob: false, Prev: err}:
-				}
-
-				return
-			}
-
-			wgActiveBatches.Done()
-		}
-	}
-}

internal/app/etl/extractor/postgres.go

@@ -1,127 +0,0 @@
-package extractor
-
-import (
-	"context"
-	"errors"
-	"fmt"
-	"strings"
-	"sync"
-	"sync/atomic"
-
-	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/config"
-	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/custom_errors"
-	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/models"
-	"github.com/google/uuid"
-	"github.com/jackc/pgx/v5/pgxpool"
-)
-
-type PostgresExtractor struct {
-	db *pgxpool.Pool
-}
-
-func NewPostgresExtractor(pool *pgxpool.Pool) *PostgresExtractor {
-	return &PostgresExtractor{db: pool}
-}
-
-func buildExtractQueryPostgres(sourceDbInfo config.SourceTableInfo, columns []models.ColumnType) string {
-	var sbColumns strings.Builder
-
-	if len(columns) == 0 {
-		sbColumns.WriteString("*")
-	} else {
-		for i, col := range columns {
-			if col.Type() == "GEOMETRY" {
-				sbColumns.WriteString(`ST_AsEWKB("`)
-				sbColumns.WriteString(col.Name())
-				sbColumns.WriteString(`") AS "`)
-				sbColumns.WriteString(col.Name())
-				sbColumns.WriteString(`"`)
-			} else {
-				sbColumns.WriteString(`"`)
-				sbColumns.WriteString(col.Name())
-				sbColumns.WriteString(`"`)
-			}
-
-			if i < len(columns)-1 {
-				sbColumns.WriteString(", ")
-			}
-		}
-	}
-
-	return fmt.Sprintf(`SELECT %s FROM "%s"."%s"  ORDER BY "%s" ASC`, sbColumns.String(), sourceDbInfo.Schema, sourceDbInfo.Table, sourceDbInfo.PrimaryKey)
-}
-
-func (postgresEx *PostgresExtractor) ProcessBatch(
-	ctx context.Context,
-	tableInfo config.SourceTableInfo,
-	columns []models.ColumnType,
-	chunkSize int,
-	batch models.Batch,
-	indexPrimaryKey int,
-	chChunksOut chan<- models.Chunk,
-	rowsRead *int64,
-) error {
-	query := buildExtractQueryPostgres(tableInfo, columns)
-
-	if batch.ShouldUseRange {
-		return errors.New("Batch config not yet supported")
-	}
-
-	rows, err := postgresEx.db.Query(ctx, query)
-	if err != nil {
-		return &custom_errors.ExtractorError{Batch: batch, HasLastId: false, Msg: err.Error()}
-	}
-	defer rows.Close()
-
-	rowsChunk := make([]models.UnknownRowValues, 0, chunkSize)
-
-	for rows.Next() {
-		values, err := rows.Values()
-		if err != nil {
-			return errors.New("Unexpected error reading rows from source")
-		}
-
-		rowsChunk = append(rowsChunk, values)
-
-		if len(rowsChunk) >= chunkSize {
-			select {
-			case chChunksOut <- models.Chunk{Id: uuid.New(), BatchId: batch.Id, Data: rowsChunk, RetryCounter: 0}:
-			case <-ctx.Done():
-				return nil
-			}
-
-			atomic.AddInt64(rowsRead, int64(len(rowsChunk)))
-			rowsChunk = make([]models.UnknownRowValues, 0, chunkSize)
-		}
-	}
-
-	if err := rows.Err(); err != nil {
-		return errors.New("Unexpected error reading rows from source")
-	}
-
-	if len(rowsChunk) > 0 {
-		select {
-		case chChunksOut <- models.Chunk{Id: uuid.New(), BatchId: batch.Id, Data: rowsChunk, RetryCounter: 0}:
-		case <-ctx.Done():
-			return nil
-		}
-
-		atomic.AddInt64(rowsRead, int64(len(rowsChunk)))
-	}
-
-	return nil
-}
-
-func (postgresEx *PostgresExtractor) Exec(
-	ctx context.Context,
-	tableInfo config.SourceTableInfo,
-	columns []models.ColumnType,
-	chunkSize int,
-	chBatchesIn <-chan models.Batch,
-	chChunksOut chan<- models.Chunk,
-	chErrorsOut chan<- custom_errors.ExtractorError,
-	chJobErrorsOut chan<- custom_errors.JobError,
-	wgActiveBatches *sync.WaitGroup,
-	rowsRead *int64,
-) {
-}

internal/app/models/colum-type.go

@@ -1,44 +0,0 @@
-package models
-
-type ColumnType struct {
-	name string
-
-	hasMaxLength      bool
-	hasPrecisionScale bool
-
-	userType    string
-	systemType  string
-	unifiedType string
-	nullable    bool
-	maxLength   int64
-	precision   int64
-	scale       int64
-}
-
-func (c *ColumnType) Name() string {
-	return c.name
-}
-
-func (c *ColumnType) UserType() string {
-	return c.userType
-}
-
-func (c *ColumnType) SystemType() string {
-	return c.systemType
-}
-
-func (c *ColumnType) Length() (length int64, ok bool) {
-	return c.maxLength, c.hasMaxLength
-}
-
-func (c *ColumnType) DecimalSize() (precision, scale int64, ok bool) {
-	return c.precision, c.scale, c.hasPrecisionScale
-}
-
-func (c *ColumnType) Nullable() bool {
-	return c.nullable
-}
-
-func (c *ColumnType) Type() string {
-	return c.unifiedType
-}

internal/app/models/main.go

@@ -1,22 +0,0 @@
-package models
-
-import "github.com/google/uuid"
-
-type UnknownRowValues = []any
-
-type Chunk struct {
-	Id           uuid.UUID
-	BatchId      uuid.UUID
-	Data         []UnknownRowValues
-	RetryCounter int
-}
-
-type Batch struct {
-	Id                    uuid.UUID
-	ParentId              uuid.UUID
-	LowerLimit            int64
-	UpperLimit            int64
-	IsLowerLimitInclusive bool
-	ShouldUseRange        bool
-	RetryCounter          int
-}

scripts/config-parser/main.go

@@ -1,17 +1,131 @@
 package main
 
 import (
-	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/config"
+	"fmt"
-	log "github.com/sirupsen/logrus"
+	"log"
+	"os"
+
+	"github.com/goccy/go-yaml"
 )
 
-func main() {
+// Estructuras para mapear el YAML
-	log.SetLevel(log.DebugLevel)
+type RetryConfig struct {
+	Attempts int `yaml:"attempts"`
+}
 
-	migrationConfig, err := config.ReadMigrationConfig()
+type DBInfo struct {
+	Schema     string `yaml:"schema"`
+	Table      string `yaml:"table"`
+	PrimaryKey string `yaml:"primary_key,omitempty"` // omitempty si no siempre existe
+}
+
+// JobSettings contiene los campos que se comparten entre 'defaults' y cada 'job'
+type JobSettings struct {
+	MaxExtractors  *int         `yaml:"max_extractors"`
+	MaxLoaders     *int         `yaml:"max_loaders"`
+	QueueSize      *int         `yaml:"queue_size"`
+	ChunkSize      *int         `yaml:"chunk_size"`
+	ChunksPerBatch *int         `yaml:"chunks_per_batch"`
+	TruncateTarget *bool        `yaml:"truncate_target"`
+	TruncateMethod *string      `yaml:"truncate_method"`
+	Retry          *RetryConfig `yaml:"retry"`
+}
+
+type Job struct {
+	Name    string   `yaml:"name"`
+	Enabled bool     `yaml:"enabled"`
+	Source  DBInfo   `yaml:"source"`
+	Target  DBInfo   `yaml:"target"`
+	PreSQL  []string `yaml:"pre_sql"`
+	PostSQL []string `yaml:"post_sql"`
+	// Incrustamos los settings para permitir los overrides
+	JobSettings `yaml:",inline"`
+}
+
+type Config struct {
+	MaxParallelWorkers int         `yaml:"max_parallel_workers"`
+	Defaults           JobSettings `yaml:"defaults"`
+	Jobs               []Job       `yaml:"jobs"`
+}
+
+func main() {
+	yamlFile, err := os.ReadFile("config.yaml")
 	if err != nil {
-		log.Fatalf("error leyendo configuracion: %v", err)
+		log.Fatalf("Error leyendo archivo: %v", err)
 	}
 
-	log.Debugf("Config: %+v", migrationConfig)
+	var config Config
+	err = yaml.Unmarshal(yamlFile, &config)
+	if err != nil {
+		log.Fatalf("Error parseando YAML: %v", err)
+	}
+
+	fmt.Printf("Configuración cargada. Trabajos: %d\n", len(config.Jobs))
+
+	for i, job := range config.Jobs {
+		jobPtr := &config.Jobs[i]
+
+		if job.MaxExtractors == nil {
+			jobPtr.MaxExtractors = config.Defaults.MaxExtractors
+		}
+		if job.MaxLoaders == nil {
+			jobPtr.MaxLoaders = config.Defaults.MaxLoaders
+		}
+		if job.QueueSize == nil {
+			jobPtr.QueueSize = config.Defaults.QueueSize
+		}
+		if job.ChunkSize == nil {
+			jobPtr.ChunkSize = config.Defaults.ChunkSize
+		}
+		if job.ChunksPerBatch == nil {
+			jobPtr.ChunksPerBatch = config.Defaults.ChunksPerBatch
+		}
+		if job.TruncateTarget == nil {
+			jobPtr.TruncateTarget = config.Defaults.TruncateTarget
+		}
+		if job.TruncateMethod == nil {
+			jobPtr.TruncateMethod = config.Defaults.TruncateMethod
+		}
+		if job.Retry == nil {
+			jobPtr.Retry = config.Defaults.Retry
+		}
+	}
+
+	printConfig(config)
+}
+
+func printConfig(config Config) {
+	fmt.Println("Max parallel workers: ", config.MaxParallelWorkers)
+
+	fmt.Println("Defaults:")
+	fmt.Printf("\tMaxExtractors: %v\n", *config.Defaults.MaxExtractors)
+	fmt.Printf("\tMaxLoaders: %v\n", *config.Defaults.MaxLoaders)
+	fmt.Printf("\tQueueSize: %v\n", *config.Defaults.QueueSize)
+	fmt.Printf("\tChunkSize: %v\n", *config.Defaults.ChunkSize)
+	fmt.Printf("\tChunksPerBatch: %v\n", *config.Defaults.ChunksPerBatch)
+	fmt.Printf("\tTruncateTarget: %v\n", *config.Defaults.TruncateTarget)
+	fmt.Printf("\tTruncateMethod: %v\n", *config.Defaults.TruncateMethod)
+	fmt.Printf("\tRetry: %v\n", *config.Defaults.Retry)
+
+	fmt.Println("Jobs:")
+	for i, job := range config.Jobs {
+		fmt.Printf("Job Name: %v\n", job.Name)
+		fmt.Printf("\tEnabled: %v\n", job.Enabled)
+		fmt.Printf("\tSource: %v\n", job.Source)
+		fmt.Printf("\tTarget: %v\n", job.Target)
+		fmt.Printf("\tMaxExtractors: %v\n", *job.MaxExtractors)
+		fmt.Printf("\tMaxLoaders: %v\n", *job.MaxLoaders)
+		fmt.Printf("\tQueueSize: %v\n", *job.QueueSize)
+		fmt.Printf("\tChunkSize: %v\n", *job.ChunkSize)
+		fmt.Printf("\tChunksPerBatch: %v\n", *job.ChunksPerBatch)
+		fmt.Printf("\tTruncateTarget: %v\n", *job.TruncateTarget)
+		fmt.Printf("\tTruncateMethod: %v\n", *job.TruncateMethod)
+		fmt.Printf("\tRetry: %v\n", *job.Retry)
+		fmt.Printf("\tPreSQL: %v\n", job.PreSQL)
+		fmt.Printf("\tPostSQL: %v\n", job.PostSQL)
+
+		if i >= 2 {
+			fmt.Println("Skipping remaining jobs...")
+		}
+	}
 }