diff --git a/README.md b/README.md index a9e1088..ca331e1 100644 --- a/README.md +++ b/README.md @@ -1,60 +1,60 @@ # go-migrate -Migrador de datos entre SQL Server y PostgreSQL con procesamiento en paralelo. +Data migrator between SQL Server and PostgreSQL with parallel ETL processing. -## Compilar +## Build ```bash go build -o go-migrate ./cmd/go_migrate ``` -## Uso +## Usage ```bash -./go-migrate [opciones] [] +./go-migrate [options] [] ``` -### Opciones +### Options -| Flag | Descripción | +| Flag | Description | |------|-------------| -| `-config ` | 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. | +| `-config ` | 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. | -### Ejemplos +### Examples ```bash -# Migrar con config.yaml por defecto +# Migrate using the default config.yaml ./go-migrate -# Usar un archivo de configuración específico -./go-migrate -config produccion.yaml +# Use a specific configuration file +./go-migrate -config production.yaml -# Validar que origen y destino tengan la misma cantidad de filas -./go-migrate -validate -config produccion.yaml +# Validate that source and target have the same row count +./go-migrate -validate -config production.yaml -# Verificar conectividad sin migrar -./go-migrate -dry-run -config produccion.yaml +# Check connectivity without migrating +./go-migrate -dry-run -config production.yaml ``` -## Configuración +## Configuration -La herramienta lee credenciales y parámetros desde variables de entorno o un archivo `.env`. +The tool reads credentials and parameters from environment variables or a `.env` file. -### Variables clave +### Key environment variables -| Variable | Descripción | +| Variable | Description | |----------|-------------| -| `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`). | +| `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`). | -Para migrar datos binarios a Azure Blob, también se requieren `AZ_STORAGE_ENABLED`, `AZ_ACCOUNT_NAME`, `AZ_CONTAINER`, `AZ_ACCOUNT_KEY`. +To migrate binary data to Azure Blob Storage, also set `AZ_STORAGE_ENABLED`, `AZ_ACCOUNT_NAME`, `AZ_CONTAINER`, and `AZ_ACCOUNT_KEY`. -### Archivo de migración (YAML) +### Migration config file (YAML) -Define los jobs de migración. Ejemplo mínimo: +Defines the migration jobs. Minimal example: ```yaml source_db_type: sqlserver @@ -72,21 +72,21 @@ defaults: max_delay_ms: 5000 jobs: - - name: migrar_usuarios + - name: demo_users enabled: true source: schema: dbo - table: Usuarios - primary_key: Id + table: users + primary_key: id target: schema: public - table: usuarios + table: users ``` -Consulta el archivo `config.yaml` de tu entorno para ver los jobs disponibles y sus parámetros específicos. +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.). -## Modos de ejecución +## Execution modes -- **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. +- **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. diff --git a/benchmark-results.md b/benchmark-results.md index 7c282e1..624d52e 100644 --- a/benchmark-results.md +++ b/benchmark-results.md @@ -1,6 +1,6 @@ # Benchmark go-migrate — 2,000,000 filas -**Tabla**: `Cartografia.MANZANA` +**Tabla**: `civic.parcels` **Fecha**: 2026-05-29 **Entorno**: Docker local (MSSQL 2022 Developer / PostgreSQL 16 + PostGIS) diff --git a/config-reverse-original.yaml b/config-reverse-original.yaml index bcb73a2..181748e 100644 --- a/config-reverse-original.yaml +++ b/config-reverse-original.yaml @@ -24,12 +24,12 @@ defaults: max_failed_batches_load: 5 jobs: - - name: cartografia_manzana_reverse + - name: demo_users_reverse enabled: true source: - schema: Cartografia - table: MANZANA - primary_key: GDB_ARCHIVE_OID + schema: demo + table: users + primary_key: id target: - schema: Cartografia - table: MANZANA + schema: demo + table: users diff --git a/config-reverse.yaml b/config-reverse.yaml index 82d5fb5..f1c8723 100644 --- a/config-reverse.yaml +++ b/config-reverse.yaml @@ -24,12 +24,12 @@ defaults: max_failed_batches_load: 5 jobs: - - name: cartografia_manzana_reverse + - name: demo_users_reverse enabled: true source: - schema: Cartografia - table: MANZANA - primary_key: GDB_ARCHIVE_OID + schema: demo + table: users + primary_key: id target: - schema: Cartografia - table: MANZANA + schema: demo + table: users diff --git a/config.yaml b/config.yaml index c8ea2d0..218d6d7 100644 --- a/config.yaml +++ b/config.yaml @@ -24,43 +24,43 @@ defaults: max_failed_batches_load: 5 jobs: - - name: cartografia_manzana + - name: demo_users enabled: true source: - schema: Cartografia - table: MANZANA - primary_key: GDB_ARCHIVE_OID + schema: demo + table: users + primary_key: id target: - schema: Cartografia - table: MANZANA + schema: demo + table: users - # - name: red_puerto + # - name: analytics_events # enabled: true # source: - # schema: Red - # table: PUERTO - # primary_key: ID_PUERTO + # schema: analytics + # table: events + # primary_key: id # from_json: - # - column: $node_id* + # - column: payload # field: id # target: - # schema: Red - # table: PUERTO + # schema: analytics + # table: events - # - name: infraestructura_site_holder__attach + # - name: storage_attachments # source: - # schema: Infraestructura - # table: SITE_HOLDER__ATTACH - # primary_key: GDB_ARCHIVE_OID + # schema: storage + # table: attachments + # primary_key: id # target: - # schema: Infraestructura - # table: SITE_HOLDER__ATTACH + # schema: storage + # table: attachments # to_storage: # columns: # - source: DATA # target: FILE_URL # mode: REFERENCE_ONLY - # prefix: Infraestructura/SITE_HOLDER__ATTACH + # prefix: storage/attachments # batches_per_partition: 20 # max_extractors: 32 # extractor_batch_size: 1 @@ -75,3 +75,33 @@ 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 diff --git a/scripts/az-blob/main.go b/scripts/az-blob/main.go deleted file mode 100644 index 3a4173f..0000000 --- a/scripts/az-blob/main.go +++ /dev/null @@ -1,44 +0,0 @@ -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() -} diff --git a/scripts/config-parser/main.go b/scripts/config-parser/main.go deleted file mode 100644 index 99e22c0..0000000 --- a/scripts/config-parser/main.go +++ /dev/null @@ -1,30 +0,0 @@ -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) -} diff --git a/scripts/mssql-copy-in/main.go b/scripts/mssql-copy-in/main.go deleted file mode 100644 index ec88755..0000000 --- a/scripts/mssql-copy-in/main.go +++ /dev/null @@ -1,114 +0,0 @@ -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 -} diff --git a/scripts/mssql-copy-in/puerto.go b/scripts/mssql-copy-in/puerto.go deleted file mode 100644 index e791084..0000000 --- a/scripts/mssql-copy-in/puerto.go +++ /dev/null @@ -1,225 +0,0 @@ -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], - ) -} diff --git a/scripts/mssql-copy-in/seed-manzana.go b/scripts/mssql-copy-in/seed-manzana.go deleted file mode 100644 index 734844b..0000000 --- a/scripts/mssql-copy-in/seed-manzana.go +++ /dev/null @@ -1,226 +0,0 @@ -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 -} diff --git a/scripts/mssql-copy-in/site-holder-attach.go b/scripts/mssql-copy-in/site-holder-attach.go deleted file mode 100644 index 1283f64..0000000 --- a/scripts/mssql-copy-in/site-holder-attach.go +++ /dev/null @@ -1,227 +0,0 @@ -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], - ) -} diff --git a/scripts/mssql-copy-in/types.go b/scripts/mssql-copy-in/types.go deleted file mode 100644 index 31d6ff2..0000000 --- a/scripts/mssql-copy-in/types.go +++ /dev/null @@ -1,52 +0,0 @@ -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 diff --git a/scripts/mssql-copy-in/utils.go b/scripts/mssql-copy-in/utils.go deleted file mode 100644 index 52b6844..0000000 --- a/scripts/mssql-copy-in/utils.go +++ /dev/null @@ -1,81 +0,0 @@ -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) -} diff --git a/scripts/mssql-info-test/main.go b/scripts/mssql-info-test/main.go deleted file mode 100644 index 6c56520..0000000 --- a/scripts/mssql-info-test/main.go +++ /dev/null @@ -1,107 +0,0 @@ -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 -} diff --git a/scripts/mssql-test/main.go b/scripts/mssql-test/main.go deleted file mode 100644 index 8165a25..0000000 --- a/scripts/mssql-test/main.go +++ /dev/null @@ -1,97 +0,0 @@ -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) -} diff --git a/scripts/pg-info-test/main.go b/scripts/pg-info-test/main.go deleted file mode 100644 index 0e34a02..0000000 --- a/scripts/pg-info-test/main.go +++ /dev/null @@ -1,232 +0,0 @@ -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 -} diff --git a/scripts/test-etl/main.go b/scripts/test-etl/main.go deleted file mode 100644 index 66561f7..0000000 --- a/scripts/test-etl/main.go +++ /dev/null @@ -1,138 +0,0 @@ -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)) -} diff --git a/scripts/wkb-to-ewkb/main.go b/scripts/wkb-to-ewkb/main.go deleted file mode 100644 index 81491f3..0000000 --- a/scripts/wkb-to-ewkb/main.go +++ /dev/null @@ -1,59 +0,0 @@ -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]) -}