refactor: add opencode.jsonc to .gitignore for improved file management

refactor: enhance SQL query formatting for PostgreSQL compatibility in dry run and validation processes
2026-07-06 00:39:30 -05:00 · 2026-05-29 16:01:25 -05:00
21 changed files with 1776 additions and 119 deletions
--- a/.gitignore
+++ b/.gitignore
@@ -30,3 +30,5 @@ go.work.sum
 .vscode/
 .temp
 .atl
+
+opencode.jsonc
--- a/README.md
+++ b/README.md
@@ -1,60 +1,60 @@
 # go-migrate

-Data migrator between SQL Server and PostgreSQL with parallel ETL processing.
+Migrador de datos entre SQL Server y PostgreSQL con procesamiento en paralelo.

-## Build
+## Compilar

 ```bash
 go build -o go-migrate ./cmd/go_migrate
 ```

-## Usage
+## Uso

 ```bash
-./go-migrate [options] [<config-path>]
+./go-migrate [opciones] [<ruta-config>]
 ```

-### Options
+### Opciones

-| Flag | Description |
+| Flag | Descripción |
 |------|-------------|
-| `-config <path>` | Path to the YAML configuration file. Can also be passed as a positional argument. Defaults to `config.yaml` in the current directory. |
-| `-validate` | Compares the row count between source and target for each job. Does not migrate data. |
-| `-dry-run` | Validates connections, storage access (if applicable), and counts source rows without migrating. |
+| `-config <path>` | Ruta al archivo de configuración YAML. También se puede pasar como argumento posicional. Si no se indica, se busca `config.yaml`. |
+| `-validate` | Compara la cantidad de filas entre origen y destino por cada job. No migra datos. |
+| `-dry-run` | Valida conexiones, acceso a storage (si aplica) y cuenta filas en origen sin migrar. |

-### Examples
+### Ejemplos

 ```bash
-# Migrate using the default config.yaml
+# Migrar con config.yaml por defecto
 ./go-migrate

-# Use a specific configuration file
-./go-migrate -config production.yaml
+# Usar un archivo de configuración específico
+./go-migrate -config produccion.yaml

-# Validate that source and target have the same row count
-./go-migrate -validate -config production.yaml
+# Validar que origen y destino tengan la misma cantidad de filas
+./go-migrate -validate -config produccion.yaml

-# Check connectivity without migrating
-./go-migrate -dry-run -config production.yaml
+# Verificar conectividad sin migrar
+./go-migrate -dry-run -config produccion.yaml
 ```

-## Configuration
+## Configuración

-The tool reads credentials and parameters from environment variables or a `.env` file.
+La herramienta lee credenciales y parámetros desde variables de entorno o un archivo `.env`.

-### Key environment variables
+### Variables clave

-| Variable | Description |
+| Variable | Descripción |
 |----------|-------------|
-| `SOURCE_DB_URL` | Source database connection URL. Alternatively, set `SOURCE_DB_HOST`, `SOURCE_DB_NAME`, `SOURCE_DB_USER`, and `SOURCE_DB_PWD`. |
-| `TARGET_DB_URL` | Target database connection URL. Alternatively, set `TARGET_DB_HOST`, `TARGET_DB_NAME`, `TARGET_DB_USER`, and `TARGET_DB_PWD`. |
-| `LOG_LEVEL` | Log level: `DEBUG`, `INFO`, `WARN`, `ERROR` (default: `INFO`). |
+| `SOURCE_DB_URL` | URL de conexión a la base de datos origen (o `SOURCE_DB_HOST`, `SOURCE_DB_NAME`, `SOURCE_DB_USER`, `SOURCE_DB_PWD`). |
+| `TARGET_DB_URL` | URL de conexión a la base de datos destino (o `TARGET_DB_HOST`, `TARGET_DB_NAME`, `TARGET_DB_USER`, `TARGET_DB_PWD`). |
+| `LOG_LEVEL` | Nivel de log: `DEBUG`, `INFO`, `WARN`, `ERROR` (por defecto: `INFO`). |

-To migrate binary data to Azure Blob Storage, also set `AZ_STORAGE_ENABLED`, `AZ_ACCOUNT_NAME`, `AZ_CONTAINER`, and `AZ_ACCOUNT_KEY`.
+Para migrar datos binarios a Azure Blob, también se requieren `AZ_STORAGE_ENABLED`, `AZ_ACCOUNT_NAME`, `AZ_CONTAINER`, `AZ_ACCOUNT_KEY`.

-### Migration config file (YAML)
+### Archivo de migración (YAML)

-Defines the migration jobs. Minimal example:
+Define los jobs de migración. Ejemplo mínimo:

 ```yaml
 source_db_type: sqlserver
@@ -72,21 +72,21 @@ defaults:
    max_delay_ms: 5000

 jobs:
-  - name: demo_users
+  - name: migrar_usuarios
    enabled: true
    source:
      schema: dbo
-      table: users
-      primary_key: id
+      table: Usuarios
+      primary_key: Id
    target:
      schema: public
-      table: users
+      table: usuarios
 ```

-See the `config.yaml` in this repository for the full set of supported options (`from_json`, `to_storage`, partition slicing via `range`, per-job overrides, `pre_sql`/`post_sql`, etc.).
+Consulta el archivo `config.yaml` de tu entorno para ver los jobs disponibles y sus parámetros específicos.

-## Execution modes
+## Modos de ejecución

- **Migrate** (default): extracts, transforms, and loads data in parallel.
- **Validate** (`-validate`): counts and compares rows between source and target.
- **Dry run** (`-dry-run`): validates connections and reports the source row count without migrating.
+- **Migración** (por defecto): extrae, transforma y carga datos en paralelo.
+- **Validación** (`-validate`): cuenta y compara filas entre origen y destino.
+- **Dry run** (`-dry-run`): verifica conexiones y muestra la cantidad de filas en origen.
--- a/benchmark-results.md
+++ b/benchmark-results.md
@@ -1,6 +1,6 @@
 # Benchmark go-migrate — 2,000,000 filas

-**Tabla**: `civic.parcels`  
+**Tabla**: `Cartografia.MANZANA`  
 **Fecha**: 2026-05-29  
 **Entorno**: Docker local (MSSQL 2022 Developer / PostgreSQL 16 + PostGIS)

--- a/cmd/go_migrate/dryrun.go
+++ b/cmd/go_migrate/dryrun.go
@@ -8,6 +8,7 @@ import (
 	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/azure"
 	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/config"
 	dbwrapper "git.ksdemosapps.com/kylesoda/go-migrate/internal/app/db-wrapper"
+	db_dialects "git.ksdemosapps.com/kylesoda/go-migrate/internal/app/db-wrapper/db_dialects"
 	log "github.com/sirupsen/logrus"
 )

@@ -67,9 +68,15 @@ func dryRunCountSourceRows(
 	for range maxParallelWorkers {
 		wg.Go(func() {
 			for job := range chJobs {
+				var sourceTableDisplay string
+				if sourceDb.GetDialect() == db_dialects.Postgres {
+					sourceTableDisplay = fmt.Sprintf(`"%s"."%s"`, job.SourceTable.Schema, job.SourceTable.Table)
+				} else {
+					sourceTableDisplay = fmt.Sprintf("[%s].[%s]", job.SourceTable.Schema, job.SourceTable.Table)
+				}
 				res := DryRunResult{
 					JobName:     job.Name,
-					SourceTable: fmt.Sprintf("[%s].[%s]", job.SourceTable.Schema, job.SourceTable.Table),
+					SourceTable: sourceTableDisplay,
 				}
 				count, err := countSourceRows(ctx, sourceDb, job)
 				if err != nil {
--- a/cmd/go_migrate/validate.go
+++ b/cmd/go_migrate/validate.go
@@ -8,6 +8,7 @@ import (

 	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/config"
 	dbwrapper "git.ksdemosapps.com/kylesoda/go-migrate/internal/app/db-wrapper"
+	db_dialects "git.ksdemosapps.com/kylesoda/go-migrate/internal/app/db-wrapper/db_dialects"
 	log "github.com/sirupsen/logrus"
 )

@@ -32,26 +33,53 @@ func countSourceRows(ctx context.Context, db dbwrapper.DbWrapper, job config.Job
 		args  []any
 	)

-	if hasRange && job.SourceTable.PrimaryKey != "" {
-		query = fmt.Sprintf("SELECT COUNT_BIG(*) FROM [%s].[%s] WHERE 1=1", schema, table)
-		if job.Range.Min != nil {
-			op := ">"
-			if job.Range.IsMinInclusive {
-				op = ">="
+	if db.GetDialect() == db_dialects.Postgres {
+		if hasRange && job.SourceTable.PrimaryKey != "" {
+			query = fmt.Sprintf(`SELECT COUNT(*) FROM "%s"."%s" WHERE 1=1`, schema, table)
+			paramIdx := 0
+			if job.Range.Min != nil {
+				paramIdx++
+				op := ">"
+				if job.Range.IsMinInclusive {
+					op = ">="
+				}
+				query += fmt.Sprintf(` AND "%s" %s $%d`, job.SourceTable.PrimaryKey, op, paramIdx)
+				args = append(args, *job.Range.Min)
 			}
-			query += fmt.Sprintf(" AND [%s] %s @min", job.SourceTable.PrimaryKey, op)
-			args = append(args, sql.Named("min", *job.Range.Min))
-		}
-		if job.Range.Max != nil {
-			op := "<"
-			if job.Range.IsMaxInclusive {
-				op = "<="
+			if job.Range.Max != nil {
+				paramIdx++
+				op := "<"
+				if job.Range.IsMaxInclusive {
+					op = "<="
+				}
+				query += fmt.Sprintf(` AND "%s" %s $%d`, job.SourceTable.PrimaryKey, op, paramIdx)
+				args = append(args, *job.Range.Max)
 			}
-			query += fmt.Sprintf(" AND [%s] %s @max", job.SourceTable.PrimaryKey, op)
-			args = append(args, sql.Named("max", *job.Range.Max))
+		} else {
+			query = fmt.Sprintf(`SELECT COUNT(*) FROM "%s"."%s"`, schema, table)
 		}
 	} else {
-		query = fmt.Sprintf("SELECT COUNT_BIG(*) FROM [%s].[%s]", schema, table)
+		if hasRange && job.SourceTable.PrimaryKey != "" {
+			query = fmt.Sprintf("SELECT COUNT_BIG(*) FROM [%s].[%s] WHERE 1=1", schema, table)
+			if job.Range.Min != nil {
+				op := ">"
+				if job.Range.IsMinInclusive {
+					op = ">="
+				}
+				query += fmt.Sprintf(" AND [%s] %s @min", job.SourceTable.PrimaryKey, op)
+				args = append(args, sql.Named("min", *job.Range.Min))
+			}
+			if job.Range.Max != nil {
+				op := "<"
+				if job.Range.IsMaxInclusive {
+					op = "<="
+				}
+				query += fmt.Sprintf(" AND [%s] %s @max", job.SourceTable.PrimaryKey, op)
+				args = append(args, sql.Named("max", *job.Range.Max))
+			}
+		} else {
+			query = fmt.Sprintf("SELECT COUNT_BIG(*) FROM [%s].[%s]", schema, table)
+		}
 	}

 	var count int64
@@ -64,7 +92,13 @@ func countSourceRows(ctx context.Context, db dbwrapper.DbWrapper, job config.Job
 func countTargetRows(ctx context.Context, db dbwrapper.DbWrapper, job config.Job) (int64, error) {
 	schema := job.TargetTable.Schema
 	table := job.TargetTable.Table
-	query := fmt.Sprintf(`SELECT COUNT(*) FROM "%s"."%s"`, schema, table)
+
+	var query string
+	if db.GetDialect() == db_dialects.Postgres {
+		query = fmt.Sprintf(`SELECT COUNT(*) FROM "%s"."%s"`, schema, table)
+	} else {
+		query = fmt.Sprintf("SELECT COUNT_BIG(*) FROM [%s].[%s]", schema, table)
+	}

 	var count int64
 	if err := db.QueryRow(ctx, query).Scan(&count); err != nil {
@@ -74,10 +108,22 @@ func countTargetRows(ctx context.Context, db dbwrapper.DbWrapper, job config.Job
 }

 func validateJob(ctx context.Context, sourceDb, targetDb dbwrapper.DbWrapper, job config.Job) ValidationResult {
+	var sourceTable, targetTable string
+	if sourceDb.GetDialect() == db_dialects.Postgres {
+		sourceTable = fmt.Sprintf(`"%s"."%s"`, job.SourceTable.Schema, job.SourceTable.Table)
+	} else {
+		sourceTable = fmt.Sprintf("[%s].[%s]", job.SourceTable.Schema, job.SourceTable.Table)
+	}
+	if targetDb.GetDialect() == db_dialects.Postgres {
+		targetTable = fmt.Sprintf(`"%s"."%s"`, job.TargetTable.Schema, job.TargetTable.Table)
+	} else {
+		targetTable = fmt.Sprintf("[%s].[%s]", job.TargetTable.Schema, job.TargetTable.Table)
+	}
+
 	result := ValidationResult{
 		JobName:     job.Name,
-		SourceTable: fmt.Sprintf("[%s].[%s]", job.SourceTable.Schema, job.SourceTable.Table),
-		TargetTable: fmt.Sprintf(`"%s"."%s"`, job.TargetTable.Schema, job.TargetTable.Table),
+		SourceTable: sourceTable,
+		TargetTable: targetTable,
 	}

 	var (
--- a/config-reverse-original.yaml
+++ b/config-reverse-original.yaml
@@ -24,12 +24,12 @@ defaults:
    max_failed_batches_load: 5

 jobs:
-  - name: demo_users_reverse
+  - name: cartografia_manzana_reverse
    enabled: true
    source:
-      schema: demo
-      table: users
-      primary_key: id
+      schema: Cartografia
+      table: MANZANA
+      primary_key: GDB_ARCHIVE_OID
    target:
-      schema: demo
-      table: users
+      schema: Cartografia
+      table: MANZANA
--- a/config-reverse.yaml
+++ b/config-reverse.yaml
@@ -24,12 +24,12 @@ defaults:
    max_failed_batches_load: 5

 jobs:
-  - name: demo_users_reverse
+  - name: cartografia_manzana_reverse
    enabled: true
    source:
-      schema: demo
-      table: users
-      primary_key: id
+      schema: Cartografia
+      table: MANZANA
+      primary_key: GDB_ARCHIVE_OID
    target:
-      schema: demo
-      table: users
+      schema: Cartografia
+      table: MANZANA
--- a/config.yaml
+++ b/config.yaml
@@ -24,43 +24,43 @@ defaults:
    max_failed_batches_load: 5

 jobs:
-  - name: demo_users
+  - name: cartografia_manzana
    enabled: true
    source:
-      schema: demo
-      table: users
-      primary_key: id
+      schema: Cartografia
+      table: MANZANA
+      primary_key: GDB_ARCHIVE_OID
    target:
-      schema: demo
-      table: users
+      schema: Cartografia
+      table: MANZANA

-  # - name: analytics_events
+  # - name: red_puerto
  #   enabled: true
  #   source:
-  #     schema: analytics
-  #     table: events
-  #     primary_key: id
+  #     schema: Red
+  #     table: PUERTO
+  #     primary_key: ID_PUERTO
  #     from_json:
-  #       - column: payload
+  #       - column: $node_id*
  #         field: id
  #   target:
-  #     schema: analytics
-  #     table: events
+  #     schema: Red
+  #     table: PUERTO

-  # - name: storage_attachments
+  # - name: infraestructura_site_holder__attach
  #   source:
-  #     schema: storage
-  #     table: attachments
-  #     primary_key: id
+  #     schema: Infraestructura
+  #     table: SITE_HOLDER__ATTACH
+  #     primary_key: GDB_ARCHIVE_OID
  #   target:
-  #     schema: storage
-  #     table: attachments
+  #     schema: Infraestructura
+  #     table: SITE_HOLDER__ATTACH
  #   to_storage:
  #     columns:
  #       - source: DATA
  #         target: FILE_URL
  #         mode: REFERENCE_ONLY
-  #         prefix: storage/attachments
+  #         prefix: Infraestructura/SITE_HOLDER__ATTACH
  #   batches_per_partition: 20
  #   max_extractors: 32
  #   extractor_batch_size: 1
@@ -75,33 +75,3 @@ jobs:
  #     base_delay_ms: 1000
  #     max_delay_ms: 15000
  #     max_jitter_ms: 500
-
-  # - name: analytics_audit_log
-  #   source:
-  #     schema: analytics
-  #     table: audit_log
-  #     primary_key: id
-  #   target:
-  #     schema: analytics
-  #     table: audit_log
-  #     pre_sql:
-  #       - "DROP INDEX IF EXISTS audit_log_created_at_idx"
-  #       - "DROP INDEX IF EXISTS audit_log_user_id_idx"
-  #     post_sql:
-  #       - "CREATE INDEX audit_log_created_at_idx ON analytics.audit_log (created_at)"
-  #       - "CREATE INDEX audit_log_user_id_idx ON analytics.audit_log (user_id)"
-  #       - "ANALYZE analytics.audit_log"
-  #   # Partition slicing: only migrate the slice [2024-01-01, 2024-12-31]
-  #   # in this re-run. is_min_inclusive=false resumes AFTER any failed boundary.
-  #   range:
-  #     min: 1704067200000   # 2024-01-01 UTC, as epoch millis
-  #     max: 1735689599999   # 2024-12-31 UTC, as epoch millis
-  #     is_min_inclusive: false
-  #     is_max_inclusive: true
-  #   partition_calculation_strategy: ESTIMATION  # faster than EXACT on large tables
-  #   truncate_method: DELETE                    # use DELETE instead of TRUNCATE
-  #   batches_per_partition: 8
-  #   retry:
-  #     attempts: 5
-  #     max_failed_partitions: 3
-  #     max_failed_batches_load: 10
--- a/scripts/az-blob/main.go
+++ b/scripts/az-blob/main.go
@@ -0,0 +1,44 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"math/rand"
+	"sync"
+
+	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/azure"
+	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/config"
+)
+
+func main() {
+	cfg := config.App.AzureStorage
+	containerName := cfg.Container
+
+	client, err := azure.NewClient(cfg)
+	if err != nil {
+		log.Fatalf("Error creando cliente: %v", err)
+	}
+	ctx := context.Background()
+
+	var wg sync.WaitGroup
+
+	for i := 1; i <= 10; i++ {
+		wg.Add(1)
+		go func(id int) {
+			defer wg.Done()
+
+			blobName := fmt.Sprintf("%sarchivo-%d.txt", cfg.Prefix, id)
+			content := fmt.Sprintf("Contenido aleatorio: %d", rand.Intn(100000))
+
+			err := client.UploadBuffer(ctx, containerName, blobName, []byte(content))
+			if err != nil {
+				log.Printf("Fallo al subir %s: %v", blobName, err)
+			} else {
+				fmt.Printf("Subido exitosamente: %s\n", blobName)
+			}
+		}(i)
+	}
+
+	wg.Wait()
+}
--- a/scripts/config-parser/main.go
+++ b/scripts/config-parser/main.go
@@ -0,0 +1,30 @@
+package main
+
+import (
+	"flag"
+
+	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/config"
+	log "github.com/sirupsen/logrus"
+)
+
+func main() {
+	log.SetLevel(log.DebugLevel)
+
+	configPath := flag.String("config", "", "path to migration config file")
+	flag.Parse()
+
+	if flag.NArg() > 1 {
+		log.Fatalf("only one config file path is allowed")
+	}
+
+	if *configPath == "" && flag.NArg() == 1 {
+		*configPath = flag.Arg(0)
+	}
+
+	migrationConfig, err := config.ReadMigrationConfig(*configPath)
+	if err != nil {
+		log.Fatalf("error leyendo configuracion: %v", err)
+	}
+
+	log.Debugf("Config: %+v", migrationConfig)
+}
--- a/scripts/mssql-copy-in/main.go
+++ b/scripts/mssql-copy-in/main.go
@@ -0,0 +1,114 @@
+package main
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"sync"
+	"time"
+
+	mssql "github.com/microsoft/go-mssqldb"
+	log "github.com/sirupsen/logrus"
+)
+
+const (
+	totalRows int = 2_000_000
+	chunkSize int = 5000
+	queueSize int = 8
+)
+
+func main() {
+	log.SetFormatter(&log.TextFormatter{
+		FullTimestamp:   true,
+		TimestampFormat: time.StampMilli,
+		DisableSorting:  false,
+		PadLevelText:    true,
+	})
+	log.SetLevel(log.DebugLevel)
+
+	db, connError := connectToSqlServer()
+	if connError != nil {
+		log.Fatal("Connection error: ", connError)
+	}
+	defer db.Close()
+
+	ctx := context.Background()
+
+	var wgSeed sync.WaitGroup
+
+	wgSeed.Go(func() {
+		seedManzanas(ctx, db)
+	})
+
+	// wgSeed.Go(func() {
+	// 	seedPuertos(ctx, db)
+	// })
+
+	// wgSeed.Go(func() {
+	// 	seedSiteHolderAttach(ctx, db)
+	// })
+
+	wgSeed.Wait()
+}
+
+func loadRowsMssql(ctx context.Context, job MigrationJob, colNames []string, db *sql.DB, in <-chan []UnknownRowValues) error {
+	chunkCount := 0
+	totalRowsLoaded := 0
+
+	for rows := range in {
+		chunkStartTime := time.Now()
+
+		tx, err := db.BeginTx(ctx, nil)
+		if err != nil {
+			return fmt.Errorf("error starting transaction: %w", err)
+		}
+
+		fullTableName := fmt.Sprintf("[%s].[%s]", job.Schema, job.Table)
+
+		stmt, err := tx.PrepareContext(ctx, mssql.CopyIn(fullTableName, mssql.BulkOptions{}, colNames...))
+		if err != nil {
+			tx.Rollback()
+			return fmt.Errorf("error preparing bulk copy statement: %w", err)
+		}
+
+		copyStartTime := time.Now()
+
+		for _, row := range rows {
+			_, err = stmt.ExecContext(ctx, row...)
+			if err != nil {
+				stmt.Close()
+				tx.Rollback()
+				return fmt.Errorf("error executing row insert: %w", err)
+			}
+		}
+
+		result, err := stmt.ExecContext(ctx)
+		if err != nil {
+			stmt.Close()
+			tx.Rollback()
+			return fmt.Errorf("error flushing bulk data: %w", err)
+		}
+
+		err = stmt.Close()
+		if err != nil {
+			tx.Rollback()
+			return fmt.Errorf("error closing statement: %w", err)
+		}
+
+		if err := tx.Commit(); err != nil {
+			return fmt.Errorf("error committing transaction: %w", err)
+		}
+
+		rowsAffected, _ := result.RowsAffected()
+		chunkCount++
+		totalRowsLoaded += int(rowsAffected)
+
+		copyDuration := time.Since(copyStartTime)
+		chunkDuration := time.Since(chunkStartTime)
+		rowsPerSec := float64(len(rows)) / chunkDuration.Seconds()
+
+		log.Infof("Loaded chunk #%d (MSSQL): %d rows in %v (copy: %v, %.0f rows/sec) - Total: %d rows", chunkCount, len(rows), chunkDuration, copyDuration, rowsPerSec, totalRowsLoaded)
+	}
+
+	return nil
+}
--- a/scripts/mssql-copy-in/puerto.go
+++ b/scripts/mssql-copy-in/puerto.go
@@ -0,0 +1,225 @@
+package main
+
+import (
+	"context"
+	"database/sql"
+	"math/rand"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+)
+
+func seedPuertos(ctx context.Context, db *sql.DB) {
+	rowsChan := make(chan []UnknownRowValues, queueSize)
+
+	// Column names for PUERTO table (excluding ID_PUERTO which is IDENTITY)
+	colNames := []string{
+		"ID_EQUIPO",
+		"ID_TERMINAL",
+		"ID_TIPO_EQUIPO",
+		"ID_PROYECTO_RESERVA",
+		"ID_TIPO_PUERTO",
+		"NUMERO",
+		"CODIGO",
+		"ESTADO",
+		"FECHA_ALTA",
+		"FECHA_ACT",
+		"ID_SITE_HOLDER",
+		"ID_PROYECTO_RESERVA_INICIAL",
+		"ID_DIRECCION",
+		"ID_TIPO_PUERTO_MEMORY",
+	}
+
+	// Start the data generator goroutine
+	go generatePuertoRows(ctx, totalRows, chunkSize, rowsChan)
+
+	// Load rows into MSSQL
+	job := MigrationJob{
+		Schema: "Red",
+		Table:  "PUERTO",
+	}
+
+	if err := loadRowsMssql(ctx, job, colNames, db, rowsChan); err != nil {
+		log.Fatal("Error loading PUERTO rows: ", err)
+	}
+
+	log.Info("PUERTO data generation and loading completed successfully")
+}
+
+// generatePuertoRows creates random row data for the PUERTO table and sends it through a channel
+func generatePuertoRows(
+	ctx context.Context,
+	totalRows int,
+	chunkSize int,
+	out chan<- []UnknownRowValues,
+) {
+	defer close(out)
+
+	rowsGenerated := 0
+	currentChunk := make([]UnknownRowValues, 0, chunkSize)
+
+	for range totalRows {
+		row := generatePuertoRow()
+		currentChunk = append(currentChunk, row)
+		rowsGenerated++
+
+		// Send chunk when it reaches the desired size
+		if len(currentChunk) == chunkSize {
+			select {
+			case out <- currentChunk:
+				log.Debugf("Sent PUERTO chunk with %d rows", len(currentChunk))
+			case <-ctx.Done():
+				log.Info("Context cancelled, stopping PUERTO row generation")
+				return
+			}
+			currentChunk = make([]UnknownRowValues, 0, chunkSize)
+		}
+
+		if rowsGenerated%100_000 == 0 {
+			logPuertoSampleRow(rowsGenerated, row)
+		}
+	}
+
+	// Send remaining rows
+	if len(currentChunk) > 0 {
+		select {
+		case out <- currentChunk:
+			log.Debugf("Sent final PUERTO chunk with %d rows", len(currentChunk))
+		case <-ctx.Done():
+			log.Info("Context cancelled, stopping PUERTO row generation")
+		}
+	}
+
+	log.Infof("Finished generating %d PUERTO rows", rowsGenerated)
+}
+
+// generatePuertoRow creates a single random row for the PUERTO table
+func generatePuertoRow() UnknownRowValues {
+	dateLowerLimit, _ := time.Parse(time.RFC3339, "2020-12-31T23:59:59Z")
+	dateUpperLimit, _ := time.Parse(time.RFC3339, "2025-12-31T23:59:59Z")
+
+	// Required columns
+	idEquipo := rand.Intn(10000) + 1    // ID_EQUIPO (1-10000)
+	idTipoEquipo := rand.Intn(100) + 1  // ID_TIPO_EQUIPO (1-100)
+	idTipoPuerto := rand.Intn(50) + 1   // ID_TIPO_PUERTO (1-50)
+	numero := rand.Intn(1000) + 1       // NUMERO (1-1000)
+	codigo := generateRandomString(100) // CODIGO: Random alphanumeric (up to 100 chars)
+
+	// Optional columns - randomly decide whether to include NULL or a value
+	var idTerminal any
+	if rand.Intn(2) == 0 {
+		idTerminal = rand.Intn(5000) + 1
+	} else {
+		idTerminal = nil
+	}
+
+	var idProyectoReserva any
+	if rand.Intn(2) == 0 {
+		idProyectoReserva = rand.Intn(1000) + 1
+	} else {
+		idProyectoReserva = nil
+	}
+
+	var estado any
+	if rand.Intn(2) == 0 {
+		estados := []string{"ACTIVO", "LIBRE", "DISPONIBLE", "MANTENIMIENTO", "RESERVADO"}
+		estado = estados[rand.Intn(len(estados))]
+	} else {
+		estado = nil
+	}
+
+	var fechaAlta any
+	if rand.Intn(2) == 0 {
+		fechaAlta = generateRandomTimestamp(dateLowerLimit, dateUpperLimit)
+	} else {
+		fechaAlta = nil
+	}
+
+	var fechaAct any
+	if rand.Intn(2) == 0 {
+		fechaAct = generateRandomTimestamp(dateLowerLimit, dateUpperLimit)
+	} else {
+		fechaAct = nil
+	}
+
+	var idSiteHolder any
+	if rand.Intn(2) == 0 {
+		idSiteHolder = rand.Intn(500) + 1
+	} else {
+		idSiteHolder = nil
+	}
+
+	var idProyectoReservaInicial any
+	if rand.Intn(2) == 0 {
+		idProyectoReservaInicial = rand.Intn(1000) + 1
+	} else {
+		idProyectoReservaInicial = nil
+	}
+
+	var idDireccion any
+	if rand.Intn(2) == 0 {
+		idDireccion = rand.Intn(100) + 1
+	} else {
+		idDireccion = nil
+	}
+
+	var idTipoPuertoMemory any
+	if rand.Intn(2) == 0 {
+		idTipoPuertoMemory = rand.Intn(50) + 1
+	} else {
+		idTipoPuertoMemory = nil
+	}
+
+	return UnknownRowValues{
+		idEquipo,
+		idTerminal,
+		idTipoEquipo,
+		idProyectoReserva,
+		idTipoPuerto,
+		numero,
+		codigo,
+		estado,
+		fechaAlta,
+		fechaAct,
+		idSiteHolder,
+		idProyectoReservaInicial,
+		idDireccion,
+		idTipoPuertoMemory,
+	}
+}
+
+func logPuertoSampleRow(id int, rowValues UnknownRowValues) {
+	log.Infof(`
+Sample row #%d:
+ID_EQUIPO (%T): %v
+ID_TERMINAL (%T): %v
+ID_TIPO_EQUIPO (%T): %v
+ID_PROYECTO_RESERVA (%T): %v
+ID_TIPO_PUERTO (%T): %v
+NUMERO (%T): %v
+CODIGO (%T): %v
+ESTADO (%T): %v
+FECHA_ALTA (%T): %v
+FECHA_ACT (%T): %v
+ID_SITE_HOLDER (%T): %v
+ID_PROYECTO_RESERVA_INICIAL (%T): %v
+ID_DIRECCION (%T): %v
+ID_TIPO_PUERTO_MEMORY (%T): %v
+`,
+		id,
+		rowValues[0], rowValues[0],
+		rowValues[1], rowValues[1],
+		rowValues[2], rowValues[2],
+		rowValues[3], rowValues[3],
+		rowValues[4], rowValues[4],
+		rowValues[5], rowValues[5],
+		rowValues[6], rowValues[6],
+		rowValues[7], rowValues[7],
+		rowValues[8], rowValues[8],
+		rowValues[9], rowValues[9],
+		rowValues[10], rowValues[10],
+		rowValues[11], rowValues[11],
+		rowValues[12], rowValues[12],
+		rowValues[13], rowValues[13],
+	)
+}
--- a/scripts/mssql-copy-in/seed-manzana.go
+++ b/scripts/mssql-copy-in/seed-manzana.go
@@ -0,0 +1,226 @@
+package main
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"math/rand"
+	"sync"
+	"time"
+
+	"github.com/gaspardle/go-mssqlclrgeo"
+	"github.com/google/uuid"
+	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]
+	`, manzanaJob.Schema, manzanaJob.Table)
+
+	err := db.QueryRowContext(ctx, query).Scan(&maxOid)
+	if err != nil && err != sql.ErrNoRows {
+		return 0, err
+	}
+
+	if !maxOid.Valid {
+		return 0, nil
+	}
+
+	return int(maxOid.Int64), nil
+}
+
+func generateManzanaRows(
+	ctx context.Context,
+	startOid int,
+	totalRows int,
+	chunkSize int,
+	out chan<- []UnknownRowValues,
+) {
+	defer close(out)
+
+	rowsGenerated := 0
+	currentChunk := make([]UnknownRowValues, 0, chunkSize)
+
+	for i := range totalRows {
+		gdbArchiveOid := startOid + i + 1
+		row := generateManzanaRow(gdbArchiveOid)
+		currentChunk = append(currentChunk, row)
+		rowsGenerated++
+
+		if len(currentChunk) == chunkSize {
+			select {
+			case out <- currentChunk:
+				log.Debugf("Sent chunk with %d rows", len(currentChunk))
+			case <-ctx.Done():
+				log.Info("Context cancelled, stopping row generation")
+				return
+			}
+			currentChunk = make([]UnknownRowValues, 0, chunkSize)
+		}
+
+		if rowsGenerated%100_000 == 0 {
+			logManzanaSampleRow(rowsGenerated, row)
+		}
+	}
+
+	if len(currentChunk) > 0 {
+		select {
+		case out <- currentChunk:
+			log.Debugf("Sent final chunk with %d rows", len(currentChunk))
+		case <-ctx.Done():
+			log.Info("Context cancelled, stopping row generation")
+		}
+	}
+
+	log.Infof("Finished generating %d rows", rowsGenerated)
+}
+
+func generateManzanaRow(gdbArchiveOid int) UnknownRowValues {
+	dateLowerLimit, _ := time.Parse(time.RFC3339, "2020-12-31T23:59:59Z")
+	dateUpperLimit, _ := time.Parse(time.RFC3339, "2025-12-31T23:59:59Z")
+
+	rowID := gdbArchiveOid
+	distrito := fmt.Sprintf("D%d", rand.Intn(100))
+	nombre := generateRandomString(15)
+	codigo := generateRandomString(15)
+	cantidadTotal := rand.Intn(1000)
+	ocupacionResidencial := rand.Intn(1000)
+	ocupacionNegocio := rand.Intn(1000)
+	ocupacionDepartamento := rand.Intn(1000)
+	indicador := rand.Intn(10000)
+	fechaAlta := generateRandomTimestamp(dateLowerLimit, dateUpperLimit)
+	fechaAct := generateRandomTimestamp(dateLowerLimit, dateUpperLimit)
+	shapeWKB := generateRandomPolygonWKB()
+	geoData := []byte{}
+	globalID, _ := uuid.New().MarshalBinary()
+	gdbFromDate := fechaAct
+	gdbToDate, _ := time.Parse(time.RFC3339, "9999-12-31T23:59:59Z")
+	objectID := gdbArchiveOid
+
+	shapeMssql, err := mssqlclrgeo.WkbToUdtGeo(shapeWKB, false)
+	if err != nil {
+		log.Errorf("Error convirtiendo WKB a formato MSSQL: %v", err)
+		shapeMssql = []byte{}
+	}
+
+	return UnknownRowValues{
+		gdbArchiveOid,
+		rowID,
+		distrito,
+		nombre,
+		codigo,
+		cantidadTotal,
+		ocupacionResidencial,
+		ocupacionNegocio,
+		ocupacionDepartamento,
+		indicador,
+		fechaAlta,
+		fechaAct,
+		shapeMssql,
+		geoData,
+		globalID,
+		gdbFromDate,
+		gdbToDate,
+		objectID,
+	}
+}
+
+func logManzanaSampleRow(id int, rowValues UnknownRowValues) {
+	log.Infof(`
+Sample row #%d:
+GDB_ARCHIVE_OID (%T): %v
+ID_MANZANA (%T): %v
+ID_DISTRITO (%T): %v
+NOMBRE (%T): %v
+CODIGO (%T): %v
+CANTIDAD_TOTAL (%T): %v
+OCUPACION_RESIDENCIAL (%T): %v
+OCUPACION_NEGOCIO (%T): %v
+OCUPACION_DEPARTAMENTO (%T): %v
+INDICADOR (%T): %v
+FECHA_ALTA (%T): %v
+FECHA_ACT (%T): %v
+Shape (%T): %v
+GDB_GEOMATTR_DATA (%T): %v
+GlobalID (%T): %v
+GDB_FROM_DATE (%T): %v
+GDB_TO_DATE (%T): %v
+OBJECTID (%T): %v
+`,
+		id,
+		rowValues[0], rowValues[0],
+		rowValues[1], rowValues[1],
+		rowValues[2], rowValues[2],
+		rowValues[3], rowValues[3],
+		rowValues[4], rowValues[4],
+		rowValues[5], rowValues[5],
+		rowValues[6], rowValues[6],
+		rowValues[7], rowValues[7],
+		rowValues[8], rowValues[8],
+		rowValues[9], rowValues[9],
+		rowValues[10], rowValues[10],
+		rowValues[11], rowValues[11],
+		rowValues[12], rowValues[12],
+		rowValues[13], rowValues[13],
+		rowValues[14], rowValues[14],
+		rowValues[15], rowValues[15],
+		rowValues[16], rowValues[16],
+		rowValues[17], rowValues[17],
+	)
+}
+
+func seedManzanas(ctx context.Context, db *sql.DB) error {
+	maxOid, err := getMaxGDBArchiveOid(ctx, db)
+	if err != nil {
+		log.Fatal("Error getting max GDB_ARCHIVE_OID: ", err)
+	}
+
+	log.Infof("Starting data generation from GDB_ARCHIVE_OID: %d", maxOid+1)
+
+	rowsChan := make(chan []UnknownRowValues, queueSize)
+
+	var wgRowGenerator sync.WaitGroup
+
+	wgRowGenerator.Go(func() {
+		generateManzanaRows(ctx, maxOid, totalRows, chunkSize, rowsChan)
+	})
+
+	columns := []string{
+		"GDB_ARCHIVE_OID",
+		"ID_MANZANA",
+		"ID_DISTRITO",
+		"NOMBRE",
+		"CODIGO",
+		"CANTIDAD_TOTAL",
+		"OCUPACION_RESIDENCIAL",
+		"OCUPACION_NEGOCIO",
+		"OCUPACION_DEPARTAMENTO",
+		"INDICADOR",
+		"FECHA_ALTA",
+		"FECHA_ACT",
+		"Shape",
+		"GDB_GEOMATTR_DATA",
+		"GlobalID",
+		"GDB_FROM_DATE",
+		"GDB_TO_DATE",
+		"OBJECTID",
+	}
+
+	if err := loadRowsMssql(ctx, manzanaJob, columns, db, rowsChan); err != nil {
+		return fmt.Errorf("Error loading rows (MANZANA): %w", err)
+	}
+
+	log.Info("Data generation and loading completed successfully (MANZANA)")
+	wgRowGenerator.Wait()
+
+	return nil
+}
--- a/scripts/mssql-copy-in/site-holder-attach.go
+++ b/scripts/mssql-copy-in/site-holder-attach.go
@@ -0,0 +1,227 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"database/sql"
+	"fmt"
+	"math/rand"
+	"sync"
+	"time"
+
+	"github.com/google/uuid"
+	log "github.com/sirupsen/logrus"
+)
+
+var siteHolderAttachJob = MigrationJob{
+	Schema: "Infraestructura",
+	Table:  "SITE_HOLDER__ATTACH",
+}
+
+func seedSiteHolderAttach(ctx context.Context, db *sql.DB) error {
+	maxOid, err := getMaxGDBArchiveOidForAttach(ctx, db)
+	if err != nil {
+		log.Fatal("Error getting max GDB_ARCHIVE_OID: ", err)
+	}
+
+	log.Infof("Starting SITE_HOLDER__ATTACH data generation from GDB_ARCHIVE_OID: %d", maxOid+1)
+
+	rowsChan := make(chan []UnknownRowValues, queueSize)
+
+	var wgRowGenerator sync.WaitGroup
+
+	wgRowGenerator.Go(func() {
+		generateSiteHolderAttachRows(ctx, maxOid, totalRows, chunkSize, rowsChan)
+	})
+
+	columns := []string{
+		"GDB_ARCHIVE_OID",
+		"REL_GLOBALID",
+		"CONTENT_TYPE",
+		"ATT_NAME",
+		"DATA_SIZE",
+		"DATA",
+		"GLOBALID",
+		"GDB_FROM_DATE",
+		"GDB_TO_DATE",
+		"ATTACHMENTID",
+	}
+
+	if err := loadRowsMssql(ctx, siteHolderAttachJob, columns, db, rowsChan); err != nil {
+		return fmt.Errorf("Error loading rows (SITE_HOLDER__ATTACH): %w", err)
+	}
+
+	log.Info("Data generation and loading completed successfully (SITE_HOLDER__ATTACH)")
+	wgRowGenerator.Wait()
+
+	return nil
+}
+
+func getMaxGDBArchiveOidForAttach(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]
+	`, siteHolderAttachJob.Schema, siteHolderAttachJob.Table)
+
+	err := db.QueryRowContext(ctx, query).Scan(&maxOid)
+	if err != nil && err != sql.ErrNoRows {
+		return 0, err
+	}
+
+	if !maxOid.Valid {
+		return 0, nil
+	}
+
+	return int(maxOid.Int64), nil
+}
+
+func generateSiteHolderAttachRows(
+	ctx context.Context,
+	startOid int,
+	totalRows int,
+	chunkSize int,
+	out chan<- []UnknownRowValues,
+) {
+	defer close(out)
+
+	rowsGenerated := 0
+	currentChunk := make([]UnknownRowValues, 0, chunkSize)
+
+	for i := range totalRows {
+		gdbArchiveOid := startOid + i + 1
+		row := generateSiteHolderAttachRow(gdbArchiveOid)
+		currentChunk = append(currentChunk, row)
+		rowsGenerated++
+
+		if len(currentChunk) == chunkSize {
+			select {
+			case out <- currentChunk:
+				log.Debugf("Sent SITE_HOLDER__ATTACH chunk with %d rows", len(currentChunk))
+			case <-ctx.Done():
+				log.Info("Context cancelled, stopping SITE_HOLDER__ATTACH row generation")
+				return
+			}
+			currentChunk = make([]UnknownRowValues, 0, chunkSize)
+		}
+
+		if rowsGenerated%100_000 == 0 {
+			logSiteHolderAttachSampleRow(rowsGenerated, row)
+		}
+	}
+
+	if len(currentChunk) > 0 {
+		select {
+		case out <- currentChunk:
+			log.Debugf("Sent final SITE_HOLDER__ATTACH chunk with %d rows", len(currentChunk))
+		case <-ctx.Done():
+			log.Info("Context cancelled, stopping SITE_HOLDER__ATTACH row generation")
+		}
+	}
+
+	log.Infof("Finished generating %d SITE_HOLDER__ATTACH rows", rowsGenerated)
+}
+
+func generateSiteHolderAttachRow(gdbArchiveOid int) UnknownRowValues {
+	dateLowerLimit, _ := time.Parse(time.RFC3339, "2020-12-31T23:59:59Z")
+	dateUpperLimit, _ := time.Parse(time.RFC3339, "2025-12-31T23:59:59Z")
+
+	relGlobalID, _ := uuid.New().MarshalBinary()
+	contentType := generateRandomContentType()
+	attName := generateRandomAttachmentName()
+	binaryData := generateRandomBinaryContent()
+	dataSize := len(binaryData)
+	globalID, _ := uuid.New().MarshalBinary()
+	gdbFromDate := generateRandomTimestamp(dateLowerLimit, dateUpperLimit)
+	gdbToDate, _ := time.Parse(time.RFC3339, "9999-12-31T23:59:59Z")
+	attachmentID := rand.Intn(10000) + 1
+
+	return UnknownRowValues{
+		gdbArchiveOid,
+		relGlobalID,
+		contentType,
+		attName,
+		dataSize,
+		binaryData,
+		globalID,
+		gdbFromDate,
+		gdbToDate,
+		attachmentID,
+	}
+}
+
+func generateRandomContentType() string {
+	contentTypes := []string{
+		"text/plain",
+		"application/pdf",
+		"image/jpeg",
+		"image/png",
+		"application/msword",
+		"application/vnd.openxmlformats-officedocument.wordprocessingml.document",
+		"text/csv",
+		"application/json",
+	}
+	return contentTypes[rand.Intn(len(contentTypes))]
+}
+
+func generateRandomAttachmentName() string {
+	extensions := []string{".txt", ".pdf", ".jpg", ".png", ".doc", ".docx", ".csv", ".json"}
+	baseName := generateRandomString(20)
+	extension := extensions[rand.Intn(len(extensions))]
+	return baseName + extension
+}
+
+func generateRandomBinaryContent() []byte {
+	sizeOptions := []int{100, 500, 1000, 5000, 10000, 50000, 100000}
+	size := sizeOptions[rand.Intn(len(sizeOptions))]
+
+	var buf bytes.Buffer
+
+	lineCount := rand.Intn(size/50) + 1
+	for range lineCount {
+		line := generateRandomString(rand.Intn(80) + 20)
+		buf.WriteString(line)
+		buf.WriteString("\n")
+	}
+
+	for buf.Len() < size {
+		randomText := generateRandomString(rand.Intn(100) + 50)
+		buf.WriteString(randomText)
+		buf.WriteString("\n")
+	}
+
+	result := buf.Bytes()
+	if len(result) > size {
+		result = result[:size]
+	}
+
+	return result
+}
+
+func logSiteHolderAttachSampleRow(id int, rowValues UnknownRowValues) {
+	dataBytes := rowValues[5].([]byte)
+	log.Infof(`
+Sample SITE_HOLDER__ATTACH row #%d:
+GDB_ARCHIVE_OID: %v
+REL_GLOBALID: [binary UUID]
+CONTENT_TYPE: %v
+ATT_NAME: %v
+DATA_SIZE: %v
+DATA: [%d bytes of binary content]
+GLOBALID: [binary UUID]
+GDB_FROM_DATE: %v
+GDB_TO_DATE: %v
+ATTACHMENTID: %v
+`,
+		id,
+		rowValues[0],
+		rowValues[2],
+		rowValues[3],
+		rowValues[4],
+		len(dataBytes),
+		rowValues[7],
+		rowValues[8],
+		rowValues[9],
+	)
+}
--- a/scripts/mssql-copy-in/types.go
+++ b/scripts/mssql-copy-in/types.go
@@ -0,0 +1,52 @@
+package main
+
+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
+}
+
+type MigrationJob struct {
+	Schema     string
+	Table      string
+	PrimaryKey string
+}
+
+type UnknownRowValues = []any
--- a/scripts/mssql-copy-in/utils.go
+++ b/scripts/mssql-copy-in/utils.go
@@ -0,0 +1,81 @@
+package main
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"math/rand"
+	"time"
+
+	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/config"
+	"github.com/twpayne/go-geom"
+	"github.com/twpayne/go-geom/encoding/wkb"
+)
+
+func connectToSqlServer() (*sql.DB, error) {
+	db, err := sql.Open("sqlserver", config.App.SourceDbUrl)
+	if err != nil {
+		return nil, fmt.Errorf("Unable to connect to sqlserver: %w", err)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+	defer cancel()
+
+	if err := db.PingContext(ctx); err != nil {
+		return nil, fmt.Errorf("Unable to ping sqlserver: %w", err)
+	}
+
+	return db, nil
+}
+
+func Map[T any, V any](input []T, mapper func(T) V) []V {
+	result := make([]V, len(input))
+
+	for i, v := range input {
+		result[i] = mapper(v)
+	}
+
+	return result
+}
+
+func generateRandomPolygonWKB() []byte {
+	minX := rand.Float64()*180 - 90
+	minY := rand.Float64()*180 - 90
+
+	size := 0.01
+
+	coords := []geom.Coord{
+		{minX, minY},
+		{minX + size, minY},
+		{minX + size, minY + size},
+		{minX, minY + size},
+		{minX, minY},
+	}
+
+	polygon := geom.NewPolygon(geom.XY).MustSetCoords([][]geom.Coord{coords})
+
+	polygonWkb, _ := wkb.Marshal(polygon, wkb.NDR)
+
+	return polygonWkb
+}
+
+func generateRandomTimestamp(min, max time.Time) time.Time {
+	minUnix := min.Unix()
+	maxUnix := max.Unix()
+
+	delta := maxUnix - minUnix
+	secAleatorios := rand.Int63n(delta)
+
+	return time.Unix(minUnix+secAleatorios, 0)
+}
+
+func generateRandomString(maxLength int) string {
+	const charset = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
+	length := min(rand.Intn(maxLength)+1, maxLength)
+
+	b := make([]byte, length)
+	for i := range b {
+		b[i] = charset[rand.Intn(len(charset))]
+	}
+	return string(b)
+}
--- a/scripts/mssql-info-test/main.go
+++ b/scripts/mssql-info-test/main.go
@@ -0,0 +1,107 @@
+package main
+
+import (
+	"database/sql"
+	"time"
+
+	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/config"
+	_ "github.com/microsoft/go-mssqldb"
+	log "github.com/sirupsen/logrus"
+)
+
+func main() {
+	log.SetFormatter(&log.TextFormatter{
+		FullTimestamp:   true,
+		TimestampFormat: time.StampMilli,
+	})
+	log.SetLevel(log.DebugLevel)
+
+	db, err := sql.Open("sqlserver", config.App.SourceDbUrl)
+	if err != nil {
+		log.Fatal("Unexpected error connecting to mssql", err)
+	}
+
+	log.Debug(config.App.SourceDbUrl)
+
+	if err := db.Ping(); err != nil {
+		log.Fatal("Couldn't ping mssql db", err)
+	}
+
+	schema := "Cartografia"
+	table := "MANZANA"
+
+	columns, err := getTableMetadata(db, schema, table)
+	if err != nil {
+		log.Fatal("Unexpected error extracting table info", err)
+	}
+
+	log.Info("Table info:")
+
+	for _, c := range columns {
+		log.Infof("%+v", c)
+	}
+}
+
+type ColumnInfo struct {
+	Name           string
+	UserDataType   string
+	SystemDataType string
+	MaxLength      int16
+	Precision      uint8
+	Scale          uint8
+	Nullable       bool
+	IsIdentity     bool
+}
+
+func getTableMetadata(db *sql.DB, schema, table string) ([]ColumnInfo, error) {
+	query := `
+SELECT 
+	c.name AS column_name,
+	t.name AS user_type_name,
+	CASE WHEN t.is_user_defined = 0 THEN t.name ELSE bt.name END AS system_type_name,
+	c.max_length AS max_length,
+	c.precision AS precision,
+	c.scale AS scale,
+	c.is_nullable AS is_nullable,
+	c.is_identity AS is_identity
+FROM sys.columns c
+JOIN sys.types t ON c.user_type_id = t.user_type_id
+LEFT JOIN sys.types bt ON t.is_user_defined = 1 AND bt.user_type_id = t.system_type_id
+JOIN sys.tables st ON c.object_id = st.object_id
+JOIN sys.schemas s ON st.schema_id = s.schema_id
+WHERE s.name = @schema AND st.name = @table
+ORDER BY c.column_id;`
+
+	rows, err := db.Query(query, sql.Named("schema", schema), sql.Named("table", table))
+	if err != nil {
+		log.Warn(err)
+
+		return nil, err
+	}
+	defer rows.Close()
+
+	var columns []ColumnInfo
+	for rows.Next() {
+		var c ColumnInfo
+		err := rows.Scan(
+			&c.Name,
+			&c.UserDataType,
+			&c.SystemDataType,
+			&c.MaxLength,
+			&c.Precision,
+			&c.Scale,
+			&c.Nullable,
+			&c.IsIdentity,
+		)
+
+		if err != nil {
+			log.Warn(err)
+
+			return nil, err
+		}
+
+		columns = append(columns, c)
+	}
+
+	return columns, nil
+}
--- a/scripts/mssql-test/main.go
+++ b/scripts/mssql-test/main.go
@@ -0,0 +1,97 @@
+package main
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"strings"
+	"time"
+
+	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/config"
+	_ "github.com/microsoft/go-mssqldb"
+	log "github.com/sirupsen/logrus"
+)
+
+func main() {
+	log.SetFormatter(&log.TextFormatter{
+		FullTimestamp:   true,
+		TimestampFormat: time.StampMilli,
+	})
+	log.SetLevel(log.DebugLevel)
+
+	db, err := sql.Open("sqlserver", config.App.SourceDbUrl)
+	if err != nil {
+		log.Fatal("Unexpected error connecting to mssql", err)
+	}
+
+	ctx := context.Background()
+
+	schema := "Cartografia"
+	table := "MANZANA"
+
+	query := buildExtractSqlSentence(schema, table, []string{})
+
+	rows, err := db.QueryContext(ctx, query)
+	if err != nil {
+		log.Fatal("Unexpected error extracting data", err)
+	}
+	defer rows.Close()
+
+	cols, _ := rows.Columns()
+
+	values := make([]any, len(cols))
+	scanArgs := make([]any, len(cols))
+	for i := range values {
+		scanArgs[i] = &values[i]
+	}
+
+	colTypes, err := rows.ColumnTypes()
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	for _, col := range colTypes {
+		log.Debugf("%+v", col)
+	}
+
+	count := 0
+	for rows.Next() {
+		if err := rows.Scan(scanArgs...); err != nil {
+			log.Fatal("Error scan", err)
+		}
+
+		count++
+		if count%100000 == 0 {
+			log.Infof("Procesadas %d filas...\n", count)
+		}
+
+		if count < 2 {
+			log.Debugf("Processed row (%d):", count)
+
+			for i, col := range cols {
+				log.Debugf("%s: %v", col, values[i])
+			}
+		}
+	}
+
+	log.Infof("Rows processed %d", count)
+}
+
+func buildExtractSqlSentence(schema, table string, colNames []string) string {
+	var sbColumns strings.Builder
+
+	if len(colNames) == 0 {
+		sbColumns.WriteString("*")
+	} else {
+		for i, col := range colNames {
+			sbColumns.WriteString(`[`)
+			sbColumns.WriteString(col)
+			sbColumns.WriteString(`]`)
+			if i < len(colNames)-1 {
+				sbColumns.WriteString(", ")
+			}
+		}
+	}
+
+	return fmt.Sprintf(`SELECT %s FROM [%s].[%s] WITH (NOLOCK)`, sbColumns.String(), schema, table)
+}
--- a/scripts/pg-info-test/main.go
+++ b/scripts/pg-info-test/main.go
@@ -0,0 +1,232 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/config"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgtype"
+	"github.com/jackc/pgx/v5/pgxpool"
+	log "github.com/sirupsen/logrus"
+)
+
+func Connect(ctx context.Context, dbURL string) (*pgxpool.Pool, error) {
+	pool, err := pgxpool.New(ctx, dbURL)
+	if err != nil {
+		return nil, fmt.Errorf("unable to connect to database: %w", err)
+	}
+
+	if err := pool.Ping(ctx); err != nil {
+		pool.Close()
+		return nil, fmt.Errorf("unable to ping database: %w", err)
+	}
+
+	return pool, nil
+}
+
+func Close(pool *pgxpool.Pool) {
+	if pool != nil {
+		pool.Close()
+	}
+}
+
+func main() {
+	log.SetFormatter(&log.TextFormatter{
+		FullTimestamp:   true,
+		TimestampFormat: time.StampMilli,
+	})
+	log.SetLevel(log.DebugLevel)
+
+	log.Info("Starting migration...")
+
+	ctxSource, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+	defer cancel()
+
+	sourcePool, err := Connect(ctxSource, config.App.SourceDbUrl)
+	defer Close(sourcePool)
+	if err != nil {
+		log.Fatal(err)
+	}
+	log.Info("Successfully connected to from_db")
+
+	ctxTarget, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+	defer cancel()
+
+	targetPool, err := Connect(ctxTarget, config.App.TargetDbUrl)
+	defer Close(targetPool)
+	if err != nil {
+		log.Fatal(err)
+	}
+	log.Info("Successfully connected to to_db")
+
+	schema := "test"
+	table := "migration_test"
+	colNames := []string{"id", "nombre_producto", "descripcion", "stock", "precio", "es_activo", "fecha_creacion", "ultima_actualizacion", "configuracion_json", "etiquetas", "binario_test", "ip_servidor", "rango_prueba"}
+
+	rowValues, err := extractData(ctxSource, sourcePool, schema, table, colNames, 10000)
+
+	if err != nil {
+		log.Fatal("Unexpected error when extracting data", err)
+	}
+
+	// for index, row := range rowValues {
+	// 	log.Debugf("Values for row %d", index+1)
+	// 	for i, v := range row {
+	// 		log.Debugf("%s: %v", colNames[i], v)
+	// 	}
+	// }
+
+	insertedRows, err := insertData(ctxTarget, targetPool, schema, table, colNames, rowValues)
+	if err != nil {
+		log.Fatal("Unexpected error when inserting rows: ", err)
+	}
+
+	log.Infof("Inserted rows: %d", insertedRows)
+
+	log.Info("Migration completed successfully!")
+}
+
+func buildExtractSqlSentence(schema, table string, colNames []string) string {
+	var sbColumns strings.Builder
+
+	for i, col := range colNames {
+		sbColumns.WriteString(`"`)
+		sbColumns.WriteString(col)
+		sbColumns.WriteString(`"`)
+		if i < len(colNames)-1 {
+			sbColumns.WriteString(", ")
+		}
+	}
+
+	return fmt.Sprintf(`SELECT %s FROM "%s"."%s" LIMIT $1`, sbColumns.String(), schema, table)
+}
+
+func extractData(ctx context.Context, sourcePool *pgxpool.Pool, schema string, table string, colNames []string, limit int) ([][]any, error) {
+	if len(colNames) == 0 {
+		return nil, errors.New("Empty column names received")
+	}
+
+	sql := buildExtractSqlSentence(schema, table, colNames)
+
+	log.Debug("Executing query: ", sql)
+
+	rows, err := sourcePool.Query(ctx, sql, limit)
+	if err != nil {
+		if !errors.Is(err, pgx.ErrNoRows) {
+			return nil, fmt.Errorf("Unexpected error: %w", err)
+		}
+
+		log.Warn("Unexpected error", err)
+		return [][]any{}, nil
+	}
+	defer rows.Close()
+
+	cols := rows.FieldDescriptions()
+	oids := make([]uint32, len(cols))
+	for i, c := range cols {
+		oids[i] = c.DataTypeOID
+	}
+
+	rowValues := make([][]any, 0, limit)
+
+	for rows.Next() {
+		values, _ := rows.Values()
+
+		for i, v := range values {
+			values[i] = castValueByOID(v, oids[i])
+		}
+
+		rowValues = append(rowValues, values)
+	}
+
+	return rowValues, nil
+}
+
+func castValueByOID(val any, oid uint32) any {
+	if val == nil {
+		return nil
+	}
+
+	switch oid {
+	case 3904:
+		if r, ok := val.(pgtype.Range[any]); ok {
+			newRange := pgtype.Range[int32]{
+				LowerType: r.LowerType,
+				UpperType: r.UpperType,
+				Valid:     r.Valid,
+			}
+			if r.Lower != nil {
+				newRange.Lower = anyToInt32(r.Lower)
+			}
+			if r.Upper != nil {
+				newRange.Upper = anyToInt32(r.Upper)
+			}
+			return newRange
+		}
+
+	case 3906:
+		if r, ok := val.(pgtype.Range[any]); ok {
+			newRange := pgtype.Range[pgtype.Numeric]{
+				LowerType: r.LowerType,
+				UpperType: r.UpperType,
+				Valid:     r.Valid,
+			}
+			if r.Lower != nil {
+				newRange.Lower = r.Lower.(pgtype.Numeric)
+			}
+			if r.Upper != nil {
+				newRange.Upper = r.Upper.(pgtype.Numeric)
+			}
+			return newRange
+		}
+	}
+
+	return val
+}
+
+func anyToInt32(v any) int32 {
+	switch t := v.(type) {
+	case int32:
+		return t
+	case int64:
+		return int32(t)
+	case int:
+		return int32(t)
+	case float64:
+		return int32(t)
+	default:
+		log.Warnf("Valor inesperado en rango: %T", v)
+		return 0
+	}
+}
+
+func insertData(ctx context.Context, targetPool *pgxpool.Pool, schema string, table string, colNames []string, rowValues [][]any) (int64, error) {
+	identifier := pgx.Identifier{schema, table}
+
+	count, err := targetPool.CopyFrom(
+		ctx,
+		identifier,
+		colNames,
+		pgx.CopyFromRows(rowValues),
+	)
+
+	if err != nil {
+		return 0, fmt.Errorf("error en CopyFrom: %w", err)
+	}
+
+	return count, nil
+}
+
+func Map[T any, V any](input []T, mapper func(T) V) []V {
+	result := make([]V, len(input))
+
+	for i, v := range input {
+		result[i] = mapper(v)
+	}
+
+	return result
+}
--- a/scripts/test-etl/main.go
+++ b/scripts/test-etl/main.go
@@ -0,0 +1,138 @@
+package main
+
+import (
+	"fmt"
+	"iter"
+	"math/rand"
+	"sync"
+	"time"
+)
+
+const (
+	numExtractors         int = 2
+	numTransformers       int = numExtractors
+	numLoaders            int = 4
+	chunkSize             int = 20
+	totalRecords          int = 1000
+	extractorsQueueSize   int = 10
+	transformersQueueSize int = 10
+	recordsPerExtractor   int = totalRecords / numExtractors
+)
+
+type Record struct {
+	Id   int
+	Data string
+}
+
+var idCounter int = 0
+
+func generateData(max int) iter.Seq[Record] {
+	time.Sleep(randomDurationMs(500, 1500))
+
+	return func(yield func(Record) bool) {
+		for i := range max {
+			record := Record{
+				Id:   idCounter,
+				Data: fmt.Sprintf("Data-%d", i),
+			}
+			idCounter++
+
+			if !yield(record) {
+				return
+			}
+		}
+	}
+}
+
+func randomDurationMs(min int, max int) time.Duration {
+	if min < 0 {
+		panic(fmt.Sprintf(`Invalid negative value for "min" argument: (min=%d)`, min))
+	}
+
+	if max <= min {
+		panic(fmt.Sprintf(`Invalid value for "max" argument: (max=%d), "max" should be greater than "min"`, max))
+	}
+
+	return time.Duration(min+rand.Intn(max-min)) * time.Millisecond
+}
+
+func Extractor(id int, chunkSize int, out chan<- []Record) {
+	chunk := make([]Record, 0, chunkSize)
+
+	for record := range generateData(recordsPerExtractor) {
+		chunk = append(chunk, record)
+
+		if len(chunk) == chunkSize {
+			out <- chunk
+			chunk = make([]Record, 0, chunkSize)
+			fmt.Printf("[Extractor %d] Lote enviado\n", id)
+		}
+	}
+
+	if len(chunk) > 0 {
+		out <- chunk
+		fmt.Printf("[Extractor %d] Lote enviado (residuos=%d)\n", id, len(chunk))
+	}
+}
+
+func Transformer(id int, in <-chan []Record, out chan<- []Record) {
+	for records := range in {
+		fmt.Printf("[Transformer %d] Transformando lote de %d registros...\n", id, len(records))
+		time.Sleep(randomDurationMs(50, 200))
+
+		for i, record := range records {
+			records[i].Data = record.Data + "-transformed"
+		}
+
+		out <- records
+	}
+}
+
+func Loader(id int, in <-chan []Record) {
+	for chunk := range in {
+		fmt.Printf("[Loader %d] Procesando lote de %d registros...\n", id, len(chunk))
+		time.Sleep(randomDurationMs(100, 2000))
+	}
+}
+
+func main() {
+	chChunksExtract := make(chan []Record, extractorsQueueSize)
+	chChunksTransform := make(chan []Record, transformersQueueSize)
+
+	var wgExtractors sync.WaitGroup
+	for i := 1; i <= numExtractors; i++ {
+		wgExtractors.Go(func() {
+			Extractor(i, chunkSize, chChunksExtract)
+		})
+	}
+
+	go func() {
+		wgExtractors.Wait()
+		close(chChunksExtract)
+		fmt.Println("--- Todos los extractores terminaron. Canal cerrado (chChunksExtract). ---")
+	}()
+
+	var wgTransformers sync.WaitGroup
+	for i := 1; i <= numTransformers; i++ {
+		wgTransformers.Go(func() {
+			Transformer(i, chChunksExtract, chChunksTransform)
+		})
+	}
+
+	go func() {
+		wgTransformers.Wait()
+		close(chChunksTransform)
+		fmt.Println("--- Todos los transformadores terminaron. Canal cerrado (chChunksTransform). ---")
+	}()
+
+	var wgLoaders sync.WaitGroup
+	for i := 1; i <= numLoaders; i++ {
+		wgLoaders.Go(func() {
+			Loader(i, chChunksTransform)
+		})
+	}
+
+	start := time.Now()
+	wgLoaders.Wait()
+	fmt.Printf("ETL Finalizado en %v\n", time.Since(start))
+}
--- a/scripts/wkb-to-ewkb/main.go
+++ b/scripts/wkb-to-ewkb/main.go
@@ -0,0 +1,59 @@
+package main
+
+import (
+	"encoding/binary"
+	"fmt"
+)
+
+const sridFlag = 0x20000000
+
+func wkbToEwkbWithSrid(geometry []byte, srid int) []byte {
+	if len(geometry) < 5 {
+		return geometry
+	}
+
+	var byteOrder binary.ByteOrder
+	if geometry[0] == 0 {
+		byteOrder = binary.BigEndian
+	} else {
+		byteOrder = binary.LittleEndian
+	}
+
+	wkbType := byteOrder.Uint32(geometry[1:5])
+	if wkbType&sridFlag != 0 {
+		return geometry
+	}
+
+	ewkbType := wkbType | sridFlag
+
+	result := make([]byte, len(geometry)+4)
+
+	result[0] = geometry[0]
+
+	byteOrder.PutUint32(result[1:5], ewkbType)
+
+	byteOrder.PutUint32(result[5:9], uint32(srid))
+
+	copy(result[9:], geometry[5:])
+
+	return result
+}
+
+func main() {
+	shape := []byte{
+		1, 3, 0, 0, 0, 1, 0, 0, 0, 5, 0, 0, 0, 217, 61, 121, 88, 168, 57, 83, 192,
+		60, 78, 209, 145, 92, 222, 39, 192, 232, 106, 43, 246, 151, 57, 83, 192,
+		60, 78, 209, 145, 92, 222, 39, 192, 232, 106, 43, 246, 151, 57, 83, 192,
+		174, 182, 98, 127, 217, 221, 39, 192, 217, 61, 121, 88, 168, 57, 83, 192,
+		174, 182, 98, 127, 217, 221, 39, 192, 217, 61, 121, 88, 168, 57, 83, 192,
+		60, 78, 209, 145, 92, 222, 39, 192,
+	}
+
+	srid := 4326
+	result := wkbToEwkbWithSrid(shape, srid)
+
+	fmt.Printf("WKB Original (len): %d\n", len(shape))
+	fmt.Printf("EWKB Result   (len): %d\n", len(result))
+	fmt.Printf("Primeros bytes (original): %v\n", shape[:10])
+	fmt.Printf("Primeros bytes (resultado): %v\n", result[:10])
+}
Author	SHA1	Message	Date
Kylesoda	0ecfa0a9e9	refactor: add opencode.jsonc to .gitignore for improved file management	2026-07-06 00:39:30 -05:00
Kylesoda	aa34f66e0b	refactor: enhance SQL query formatting for PostgreSQL compatibility in dry run and validation processes	2026-05-29 16:01:25 -05:00