feat: enhance logging and performance metrics for migration processes

cmd/go_migrate/extractor.go

@@ -3,6 +3,7 @@ package main
 import (
 	"context"
 	"database/sql"
+	"time"
 
 	"github.com/jackc/pgx/v5/pgxpool"
 	_ "github.com/microsoft/go-mssqldb"
@@ -15,13 +16,18 @@ func extractFromMssql(ctx context.Context, job MigrationJob, columns []ColumnTyp
 	query := buildExtractQueryMssql(job, columns)
 	log.Debug("Query used to extract data from mssql: ", query)
 
+	queryStartTime := time.Now()
 	rows, err := db.QueryContext(ctx, query)
 	if err != nil {
 		return err
 	}
 	defer rows.Close()
+	log.Debugf("Query executed in %v", time.Since(queryStartTime))
 
 	rowsChunk := make([]UnknownRowValues, 0, chunkSize)
+	totalRowsExtracted := 0
+	chunkCount := 0
+	chunkStartTime := time.Now()
 
 	for rows.Next() {
 		values := make([]any, len(columns))
@@ -36,20 +42,29 @@ func extractFromMssql(ctx context.Context, job MigrationJob, columns []ColumnTyp
 		}
 
 		rowsChunk = append(rowsChunk, values)
+		totalRowsExtracted++
 
 		if len(rowsChunk) >= chunkSize {
+			chunkCount++
+			chunkDuration := time.Since(chunkStartTime)
+			rowsPerSec := float64(chunkSize) / chunkDuration.Seconds()
+			log.Infof("Extracted chunk #%d: %d rows in %v (%.0f rows/sec) - Total: %d rows", chunkCount, len(rowsChunk), chunkDuration, rowsPerSec, totalRowsExtracted)
 			out <- rowsChunk
 			rowsChunk = make([]UnknownRowValues, 0, chunkSize)
-			log.Debugf("Chunk send... %+v", job)
+			chunkStartTime = time.Now()
 		}
 	}
 
 	if len(rowsChunk) > 0 {
+		chunkCount++
+		chunkDuration := time.Since(chunkStartTime)
+		rowsPerSec := float64(len(rowsChunk)) / chunkDuration.Seconds()
+		log.Infof("Extracted final chunk #%d: %d rows in %v (%.0f rows/sec) - Total: %d rows",
+			chunkCount, len(rowsChunk), chunkDuration, rowsPerSec, totalRowsExtracted)
 		out <- rowsChunk
-		log.Debugf("Chunk send... %+v", job)
 	}
 
-	return nil
+	return rows.Err()
 }
 
 func extractFromPostgres(ctx context.Context, job MigrationJob, columns []ColumnType, chunkSize int, db *pgxpool.Pool, out chan<- []UnknownRowValues) error {