refactor: rename Batch to Partition across pipeline

Rename internal types and channels from Batch to Partition for consistency with the data model.

cmd/go_migrate/process.go

@@ -4,10 +4,14 @@ import (
 	"context"
 	"database/sql"
 	"sync"
+	"sync/atomic"
 	"time"
 
+	"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/etl"
+	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/models"
 	"github.com/jackc/pgx/v5/pgxpool"
-
 	_ "github.com/microsoft/go-mssqldb"
 	log "github.com/sirupsen/logrus"
 )
@@ -16,14 +20,22 @@ func processMigrationJob(
 	ctx context.Context,
 	sourceDb *sql.DB,
 	targetDb *pgxpool.Pool,
-	job MigrationJob,
-) {
-	jobStartTime := time.Now()
-	log.Infof("Starting migration job: %s.%s [PK: %s]", job.Schema, job.Table, job.PrimaryKey)
+	extractor etl.Extractor,
+	transformer etl.Transformer,
+	loader etl.Loader,
+	job config.Job,
+) JobResult {
+	result := JobResult{
+		JobName:   job.Name,
+		StartTime: time.Now(),
+	}
 
-	sourceColTypes, targetColTypes, err := GetColumnTypes(sourceDb, targetDb, job)
+	var rowsRead, rowsLoaded, rowsFailed int64
+
+	sourceColTypes, targetColTypes, err := GetColumnTypes(sourceDb, targetDb, job.SourceTable, job.TargetTable)
 	if err != nil {
-		log.Fatal("Unexpected error: ", err)
+		result.Error = err
+		return result
 	}
 
 	logColumnTypes(sourceColTypes, "Source col types")
@@ -32,17 +44,17 @@ func processMigrationJob(
 	jobCtx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
-	batches, err := batchGeneratorMssql(jobCtx, sourceDb, job)
+	batches, err := batchGeneratorMssql(jobCtx, sourceDb, job.SourceTable, job.RowsPerBatch)
 	if err != nil {
 		log.Error("Unexpected error calculating batch ranges: ", err)
 	}
 
-	chJobErrors := make(chan JobError, 50)
-	chBatches := make(chan Batch, QueueSize)
-	chExtractorErrors := make(chan ExtractorError, QueueSize)
-	chChunksRaw := make(chan Chunk, QueueSize)
-	chChunksTransformed := make(chan Chunk, QueueSize)
-	chLoadersErrors := make(chan LoaderError, QueueSize)
+	chJobErrors := make(chan custom_errors.JobError, job.QueueSize)
+	chBatches := make(chan models.Partition, job.QueueSize)
+	chExtractorErrors := make(chan custom_errors.ExtractorError, job.QueueSize)
+	chChunksRaw := make(chan models.Batch, job.QueueSize)
+	chChunksTransformed := make(chan models.Batch, job.QueueSize)
+	chLoadersErrors := make(chan custom_errors.LoaderError, job.QueueSize)
 
 	var wgActiveBatches sync.WaitGroup
 	var wgActiveChunks sync.WaitGroup
@@ -51,21 +63,32 @@ func processMigrationJob(
 	var wgLoaders sync.WaitGroup
 
 	go func() {
-		if err := jobErrorHandler(jobCtx, chJobErrors); err != nil {
+		if err := custom_errors.JobErrorHandler(jobCtx, chJobErrors); err != nil {
 			cancel()
+			result.Error = err
 		}
 	}()
 
-	go extractorErrorHandler(jobCtx, chExtractorErrors, chBatches, chJobErrors, &wgActiveBatches)
-	go loaderErrorHandler(jobCtx, chLoadersErrors, chChunksTransformed, chJobErrors, &wgActiveChunks)
+	go custom_errors.ExtractorErrorHandler(jobCtx, job.Retry.Attempts, chExtractorErrors, chBatches, chJobErrors, &wgActiveBatches)
+	go custom_errors.LoaderErrorHandler(jobCtx, job.Retry.Attempts, chLoadersErrors, chChunksTransformed, chJobErrors, &wgActiveChunks)
 
-	maxExtractors := min(NumExtractors, len(batches))
-	log.Infof("Starting %d extractors...", maxExtractors)
-	extractStartTime := time.Now()
+	maxExtractors := min(job.MaxExtractors, len(batches))
+	log.Infof("Starting %d extractor(s)...", maxExtractors)
 
 	for range maxExtractors {
 		wgExtractors.Go(func() {
-			extractFromMssql(jobCtx, sourceDb, job, sourceColTypes, ChunkSize, chBatches, chChunksRaw, chExtractorErrors, chJobErrors, &wgActiveBatches)
+			extractor.Exec(
+				jobCtx,
+				job.SourceTable,
+				sourceColTypes,
+				job.ChunkSize,
+				chBatches,
+				chChunksRaw,
+				chExtractorErrors,
+				chJobErrors,
+				&wgActiveBatches,
+				&rowsRead,
+			)
 		})
 	}
 
@@ -76,21 +99,35 @@ func processMigrationJob(
 		}
 	}()
 
-	log.Infof("Starting %d transformers...", maxExtractors)
-	transformStartTime := time.Now()
+	log.Infof("Starting %d transformer(s)...", maxExtractors)
 
 	for range maxExtractors {
 		wgTransformers.Go(func() {
-			transformRowsMssql(jobCtx, sourceColTypes, chChunksRaw, chChunksTransformed, chJobErrors, &wgActiveChunks)
+			transformer.Exec(
+				jobCtx,
+				sourceColTypes,
+				chChunksRaw,
+				chChunksTransformed,
+				chJobErrors,
+				&wgActiveChunks,
+			)
 		})
 	}
 
-	log.Infof("Starting %d PostgreSQL loader(s)...", NumLoaders)
-	loadStartTime := time.Now()
+	log.Infof("Starting %d loader(s)...", job.MaxLoaders)
 
-	for range NumLoaders {
+	for range job.MaxLoaders {
 		wgLoaders.Go(func() {
-			loadRowsPostgres(jobCtx, targetDb, job, targetColTypes, chChunksTransformed, chLoadersErrors, chJobErrors, &wgActiveChunks)
+			loader.Exec(
+				jobCtx,
+				job.TargetTable,
+				targetColTypes,
+				chChunksTransformed,
+				chLoadersErrors,
+				chJobErrors,
+				&wgActiveChunks,
+				&rowsLoaded,
+			)
 		})
 	}
 
@@ -100,37 +137,37 @@ 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) {
+func logColumnTypes(columnTypes []models.ColumnType, label string) {
 	log.Debug(label)
 
 	for _, col := range columnTypes {
 		log.Debugf("%+v", col)
 	}
 }
-
-func logSampleRow(job MigrationJob, columns []ColumnType, rowValues UnknownRowValues, tag string) {
-	log.Infof("[%s.%s] Sample row: (%s)", job.Schema, job.Table, tag)
-	for i, col := range columns {
-		log.Infof("%s (%T): %v", col.Name(), rowValues[i], rowValues[i])
-	}
-}