feat: add MANZANA migration job and update related processing logic

cmd/go_migrate/main.go

@@ -13,6 +13,11 @@ type MigrationJob struct {
 }
 
 var migrationJobs []MigrationJob = []MigrationJob{
+	{
+		Schema:     "Cartografia",
+		Table:      "MANZANA",
+		PrimaryKey: "GDB_ARCHIVE_OID",
+	},
 	{
 		Schema:     "Red",
 		Table:      "PUERTO",

cmd/go_migrate/process.go

@@ -79,17 +79,17 @@ func processMigrationJob(sourceDb *sql.DB, targetDb *pgxpool.Pool, job Migration
 	}()
 
 	var wgPostgresLoaders sync.WaitGroup
-	// postgresLoaderCtx := context.Background()
+	postgresLoaderCtx := context.Background()
 
 	log.Infof("Starting %d PostgreSQL loader(s)...", NumLoaders)
 	loaderStartTime := time.Now()
 
 	for range NumLoaders {
 		wgPostgresLoaders.Go(func() {
-			// if err := loadRowsPostgres(postgresLoaderCtx, job, sourceColTypes, targetDb, chRowsTransform); err != nil {
-			// 	log.Error("Unexpected error loading data into postgres: ", err)
-			// }
-			fakeLoader(job, sourceColTypes, chRowsTransform)
+			if err := loadRowsPostgres(postgresLoaderCtx, job, targetColTypes, targetDb, chRowsTransform); err != nil {
+				log.Error("Unexpected error loading data into postgres: ", err)
+			}
+			// fakeLoader(job, sourceColTypes, chRowsTransform)
 		})
 	}
 

scripts/mssql-copy-in/main.go

@@ -14,8 +14,6 @@ import (
 const (
 	totalRows int    = 1_000_000
 	chunkSize int    = 50_000
-	schema    string = "Red"
-	table     string = "PUERTO"
 	queueSize int    = 4
 )
 
@@ -39,7 +37,7 @@ func main() {
 	var wgSeed sync.WaitGroup
 
 	wgSeed.Go(func() {
-		seedPuertos(ctx, db)
+		seedManzanas(ctx, db)
 	})
 
 	wgSeed.Wait()

scripts/mssql-copy-in/puerto.go

@@ -55,12 +55,10 @@ func generatePuertoRows(
 ) {
 	defer close(out)
 
-	rand.Seed(time.Now().UnixNano())
-
 	rowsGenerated := 0
 	currentChunk := make([]UnknownRowValues, 0, chunkSize)
 
-	for i := 0; i < totalRows; i++ {
+	for range totalRows {
 		row := generatePuertoRow()
 		currentChunk = append(currentChunk, row)
 		rowsGenerated++

scripts/mssql-copy-in/seed-manzana.go

@@ -13,13 +13,18 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
+var manzanaJob = MigrationJob{
+	Schema: "Cartografia",
+	Table:  "MANZANA",
+}
+
 func getMaxGDBArchiveOid(ctx context.Context, db *sql.DB) (int, error) {
 	var maxOid sql.NullInt64
 
 	query := fmt.Sprintf(`
 		SELECT ISNULL(MAX(GDB_ARCHIVE_OID), 0)
 		FROM [%s].[%s]
-	`, schema, table)
+	`, manzanaJob.Schema, manzanaJob.Table)
 
 	err := db.QueryRowContext(ctx, query).Scan(&maxOid)
 	if err != nil && err != sql.ErrNoRows {
@@ -211,12 +216,7 @@ func seedManzanas(ctx context.Context, db *sql.DB) error {
 		"OBJECTID",
 	}
 
-	job := MigrationJob{
-		Schema: schema,
-		Table:  table,
-	}
-
-	if err := loadRowsMssql(ctx, job, columns, db, rowsChan); err != nil {
+	if err := loadRowsMssql(ctx, manzanaJob, columns, db, rowsChan); err != nil {
 		return fmt.Errorf("Error loading rows (MANZANA): %w", err)
 	}