kylesoda/go-migrate
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Remove obsolete MSSQL and PostgreSQL scripts and related types

Browse Source 
- Deleted `seed-manzana.go`, `site-holder-attach.go`, `types.go`, `utils.go`, `main.go` files from both `mssql-info-test` and `mssql-test` directories.
- Removed `pg-info-test/main.go` and `test-etl/main.go` files.
- Eliminated `wkb-to-ewkb/main.go` for unused WKB to EWKB conversion functionality.
- Cleaned up associated functions and types that are no longer needed.
This commit is contained in:
DiegoAlessandroMotta
2026-07-16 20:23:58 -05:00
parent 86258718d8
commit bfe335c933
18 changed files with 99 additions and 1701 deletions
Download Patch File Download Diff File Expand all files Collapse all files

72
README.md
View File

@@ -1,60 +1,60 @@
# go-migrate # 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 ```bash
go build -o go-migrate ./cmd/go_migrate go build -o go-migrate ./cmd/go_migrate
``` ```
## Uso ## Usage
```bash ```bash
./go-migrate [opciones] [<ruta-config>] ./go-migrate [options] [<config-path>]
``` ```
### Opciones ### Options
| Flag | Descripción | | Flag | Description |
|------|-------------| |------|-------------|
| `-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`. | | `-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` | Compara la cantidad de filas entre origen y destino por cada job. No migra datos. | | `-validate` | Compares the row count between source and target for each job. Does not migrate data. |
| `-dry-run` | Valida conexiones, acceso a storage (si aplica) y cuenta filas en origen sin migrar. | | `-dry-run` | Validates connections, storage access (if applicable), and counts source rows without migrating. |
### Ejemplos ### Examples
```bash ```bash
# Migrar con config.yaml por defecto # Migrate using the default config.yaml
./go-migrate ./go-migrate
# Usar un archivo de configuración específico # Use a specific configuration file
./go-migrate -config produccion.yaml ./go-migrate -config production.yaml
# Validar que origen y destino tengan la misma cantidad de filas # Validate that source and target have the same row count
./go-migrate -validate -config produccion.yaml ./go-migrate -validate -config production.yaml
# Verificar conectividad sin migrar # Check connectivity without migrating
./go-migrate -dry-run -config produccion.yaml ./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`). | | `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` | URL de conexión a la base de datos destino (o `TARGET_DB_HOST`, `TARGET_DB_NAME`, `TARGET_DB_USER`, `TARGET_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` | Nivel de log: `DEBUG`, `INFO`, `WARN`, `ERROR` (por defecto: `INFO`). | | `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 ```yaml
source_db_type: sqlserver source_db_type: sqlserver
@@ -72,21 +72,21 @@ defaults:
max_delay_ms: 5000 max_delay_ms: 5000
jobs: jobs:
- name: migrar_usuarios - name: demo_users
enabled: true enabled: true
source: source:
schema: dbo schema: dbo
table: Usuarios table: users
primary_key: Id primary_key: id
target: target:
schema: public 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. - **Migrate** (default): extracts, transforms, and loads data in parallel.
- **Validación** (`-validate`): cuenta y compara filas entre origen y destino. - **Validate** (`-validate`): counts and compares rows between source and target.
- **Dry run** (`-dry-run`): verifica conexiones y muestra la cantidad de filas en origen. - **Dry run** (`-dry-run`): validates connections and reports the source row count without migrating.

2
benchmark-results.md
View File

@@ -1,6 +1,6 @@
# Benchmark go-migrate — 2,000,000 filas # Benchmark go-migrate — 2,000,000 filas
**Tabla**: `Cartografia.MANZANA` **Tabla**: `civic.parcels`
**Fecha**: 2026-05-29 **Fecha**: 2026-05-29
**Entorno**: Docker local (MSSQL 2022 Developer / PostgreSQL 16 + PostGIS) **Entorno**: Docker local (MSSQL 2022 Developer / PostgreSQL 16 + PostGIS)

12
config-reverse-original.yaml
View File

@@ -24,12 +24,12 @@ defaults:
max_failed_batches_load: 5 max_failed_batches_load: 5
jobs: jobs:
- name: cartografia_manzana_reverse - name: demo_users_reverse
enabled: true enabled: true
source: source:
schema: Cartografia schema: demo
table: MANZANA table: users
primary_key: GDB_ARCHIVE_OID primary_key: id
target: target:
schema: Cartografia schema: demo
table: MANZANA table: users

12
config-reverse.yaml
View File

@@ -24,12 +24,12 @@ defaults:
max_failed_batches_load: 5 max_failed_batches_load: 5
jobs: jobs:
- name: cartografia_manzana_reverse - name: demo_users_reverse
enabled: true enabled: true
source: source:
schema: Cartografia schema: demo
table: MANZANA table: users
primary_key: GDB_ARCHIVE_OID primary_key: id
target: target:
schema: Cartografia schema: demo
table: MANZANA table: users

70
config.yaml
View File

@@ -24,43 +24,43 @@ defaults:
max_failed_batches_load: 5 max_failed_batches_load: 5
jobs: jobs:
- name: cartografia_manzana - name: demo_users
enabled: true enabled: true
source: source:
schema: Cartografia schema: demo
table: MANZANA table: users
primary_key: GDB_ARCHIVE_OID primary_key: id
target: target:
schema: Cartografia schema: demo
table: MANZANA table: users
# - name: red_puerto # - name: analytics_events
# enabled: true # enabled: true
# source: # source:
# schema: Red # schema: analytics
# table: PUERTO # table: events
# primary_key: ID_PUERTO # primary_key: id
# from_json: # from_json:
# - column: $node_id* # - column: payload
# field: id # field: id
# target: # target:
# schema: Red # schema: analytics
# table: PUERTO # table: events
# - name: infraestructura_site_holder__attach # - name: storage_attachments
# source: # source:
# schema: Infraestructura # schema: storage
# table: SITE_HOLDER__ATTACH # table: attachments
# primary_key: GDB_ARCHIVE_OID # primary_key: id
# target: # target:
# schema: Infraestructura # schema: storage
# table: SITE_HOLDER__ATTACH # table: attachments
# to_storage: # to_storage:
# columns: # columns:
# - source: DATA # - source: DATA
# target: FILE_URL # target: FILE_URL
# mode: REFERENCE_ONLY # mode: REFERENCE_ONLY
# prefix: Infraestructura/SITE_HOLDER__ATTACH # prefix: storage/attachments
# batches_per_partition: 20 # batches_per_partition: 20
# max_extractors: 32 # max_extractors: 32
# extractor_batch_size: 1 # extractor_batch_size: 1
@@ -75,3 +75,33 @@ jobs:
# base_delay_ms: 1000 # base_delay_ms: 1000
# max_delay_ms: 15000 # max_delay_ms: 15000
# max_jitter_ms: 500 # 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

44
scripts/az-blob/main.go
View File

@@ -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()
}

30
scripts/config-parser/main.go
View File

@@ -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)
}

114
scripts/mssql-copy-in/main.go
View File

@@ -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
}

225
scripts/mssql-copy-in/puerto.go
View File

@@ -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],
)
}

226
scripts/mssql-copy-in/seed-manzana.go
View File

@@ -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
}

227
scripts/mssql-copy-in/site-holder-attach.go
View File

@@ -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],
)
}

52
scripts/mssql-copy-in/types.go
View File

@@ -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

81
scripts/mssql-copy-in/utils.go
View File

@@ -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)
}

107
scripts/mssql-info-test/main.go
View File

@@ -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
}

97
scripts/mssql-test/main.go
View File

@@ -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)
}

232
scripts/pg-info-test/main.go
View File

@@ -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
}

138
scripts/test-etl/main.go
View File

@@ -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))
}

59
scripts/wkb-to-ewkb/main.go
View File

@@ -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])
}