Compare commits
2 Commits
main
...
refactor/e
| Author | SHA1 | Date | |
|---|---|---|---|
|
0ecfa0a9e9
|
|||
|
aa34f66e0b
|
2
.gitignore
vendored
2
.gitignore
vendored
@@ -30,3 +30,5 @@ go.work.sum
|
|||||||
.vscode/
|
.vscode/
|
||||||
.temp
|
.temp
|
||||||
.atl
|
.atl
|
||||||
|
|
||||||
|
opencode.jsonc
|
||||||
|
|||||||
@@ -8,6 +8,7 @@ import (
|
|||||||
"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/azure"
|
"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/azure"
|
||||||
"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/config"
|
"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/config"
|
||||||
dbwrapper "git.ksdemosapps.com/kylesoda/go-migrate/internal/app/db-wrapper"
|
dbwrapper "git.ksdemosapps.com/kylesoda/go-migrate/internal/app/db-wrapper"
|
||||||
|
db_dialects "git.ksdemosapps.com/kylesoda/go-migrate/internal/app/db-wrapper/db_dialects"
|
||||||
log "github.com/sirupsen/logrus"
|
log "github.com/sirupsen/logrus"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -67,9 +68,15 @@ func dryRunCountSourceRows(
|
|||||||
for range maxParallelWorkers {
|
for range maxParallelWorkers {
|
||||||
wg.Go(func() {
|
wg.Go(func() {
|
||||||
for job := range chJobs {
|
for job := range chJobs {
|
||||||
|
var sourceTableDisplay string
|
||||||
|
if sourceDb.GetDialect() == db_dialects.Postgres {
|
||||||
|
sourceTableDisplay = fmt.Sprintf(`"%s"."%s"`, job.SourceTable.Schema, job.SourceTable.Table)
|
||||||
|
} else {
|
||||||
|
sourceTableDisplay = fmt.Sprintf("[%s].[%s]", job.SourceTable.Schema, job.SourceTable.Table)
|
||||||
|
}
|
||||||
res := DryRunResult{
|
res := DryRunResult{
|
||||||
JobName: job.Name,
|
JobName: job.Name,
|
||||||
SourceTable: fmt.Sprintf("[%s].[%s]", job.SourceTable.Schema, job.SourceTable.Table),
|
SourceTable: sourceTableDisplay,
|
||||||
}
|
}
|
||||||
count, err := countSourceRows(ctx, sourceDb, job)
|
count, err := countSourceRows(ctx, sourceDb, job)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -8,6 +8,7 @@ import (
|
|||||||
|
|
||||||
"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/config"
|
"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/config"
|
||||||
dbwrapper "git.ksdemosapps.com/kylesoda/go-migrate/internal/app/db-wrapper"
|
dbwrapper "git.ksdemosapps.com/kylesoda/go-migrate/internal/app/db-wrapper"
|
||||||
|
db_dialects "git.ksdemosapps.com/kylesoda/go-migrate/internal/app/db-wrapper/db_dialects"
|
||||||
log "github.com/sirupsen/logrus"
|
log "github.com/sirupsen/logrus"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -32,6 +33,32 @@ func countSourceRows(ctx context.Context, db dbwrapper.DbWrapper, job config.Job
|
|||||||
args []any
|
args []any
|
||||||
)
|
)
|
||||||
|
|
||||||
|
if db.GetDialect() == db_dialects.Postgres {
|
||||||
|
if hasRange && job.SourceTable.PrimaryKey != "" {
|
||||||
|
query = fmt.Sprintf(`SELECT COUNT(*) FROM "%s"."%s" WHERE 1=1`, schema, table)
|
||||||
|
paramIdx := 0
|
||||||
|
if job.Range.Min != nil {
|
||||||
|
paramIdx++
|
||||||
|
op := ">"
|
||||||
|
if job.Range.IsMinInclusive {
|
||||||
|
op = ">="
|
||||||
|
}
|
||||||
|
query += fmt.Sprintf(` AND "%s" %s $%d`, job.SourceTable.PrimaryKey, op, paramIdx)
|
||||||
|
args = append(args, *job.Range.Min)
|
||||||
|
}
|
||||||
|
if job.Range.Max != nil {
|
||||||
|
paramIdx++
|
||||||
|
op := "<"
|
||||||
|
if job.Range.IsMaxInclusive {
|
||||||
|
op = "<="
|
||||||
|
}
|
||||||
|
query += fmt.Sprintf(` AND "%s" %s $%d`, job.SourceTable.PrimaryKey, op, paramIdx)
|
||||||
|
args = append(args, *job.Range.Max)
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
query = fmt.Sprintf(`SELECT COUNT(*) FROM "%s"."%s"`, schema, table)
|
||||||
|
}
|
||||||
|
} else {
|
||||||
if hasRange && job.SourceTable.PrimaryKey != "" {
|
if hasRange && job.SourceTable.PrimaryKey != "" {
|
||||||
query = fmt.Sprintf("SELECT COUNT_BIG(*) FROM [%s].[%s] WHERE 1=1", schema, table)
|
query = fmt.Sprintf("SELECT COUNT_BIG(*) FROM [%s].[%s] WHERE 1=1", schema, table)
|
||||||
if job.Range.Min != nil {
|
if job.Range.Min != nil {
|
||||||
@@ -53,6 +80,7 @@ func countSourceRows(ctx context.Context, db dbwrapper.DbWrapper, job config.Job
|
|||||||
} else {
|
} else {
|
||||||
query = fmt.Sprintf("SELECT COUNT_BIG(*) FROM [%s].[%s]", schema, table)
|
query = fmt.Sprintf("SELECT COUNT_BIG(*) FROM [%s].[%s]", schema, table)
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
var count int64
|
var count int64
|
||||||
if err := db.QueryRow(ctx, query, args...).Scan(&count); err != nil {
|
if err := db.QueryRow(ctx, query, args...).Scan(&count); err != nil {
|
||||||
@@ -64,7 +92,13 @@ func countSourceRows(ctx context.Context, db dbwrapper.DbWrapper, job config.Job
|
|||||||
func countTargetRows(ctx context.Context, db dbwrapper.DbWrapper, job config.Job) (int64, error) {
|
func countTargetRows(ctx context.Context, db dbwrapper.DbWrapper, job config.Job) (int64, error) {
|
||||||
schema := job.TargetTable.Schema
|
schema := job.TargetTable.Schema
|
||||||
table := job.TargetTable.Table
|
table := job.TargetTable.Table
|
||||||
query := fmt.Sprintf(`SELECT COUNT(*) FROM "%s"."%s"`, schema, table)
|
|
||||||
|
var query string
|
||||||
|
if db.GetDialect() == db_dialects.Postgres {
|
||||||
|
query = fmt.Sprintf(`SELECT COUNT(*) FROM "%s"."%s"`, schema, table)
|
||||||
|
} else {
|
||||||
|
query = fmt.Sprintf("SELECT COUNT_BIG(*) FROM [%s].[%s]", schema, table)
|
||||||
|
}
|
||||||
|
|
||||||
var count int64
|
var count int64
|
||||||
if err := db.QueryRow(ctx, query).Scan(&count); err != nil {
|
if err := db.QueryRow(ctx, query).Scan(&count); err != nil {
|
||||||
@@ -74,10 +108,22 @@ func countTargetRows(ctx context.Context, db dbwrapper.DbWrapper, job config.Job
|
|||||||
}
|
}
|
||||||
|
|
||||||
func validateJob(ctx context.Context, sourceDb, targetDb dbwrapper.DbWrapper, job config.Job) ValidationResult {
|
func validateJob(ctx context.Context, sourceDb, targetDb dbwrapper.DbWrapper, job config.Job) ValidationResult {
|
||||||
|
var sourceTable, targetTable string
|
||||||
|
if sourceDb.GetDialect() == db_dialects.Postgres {
|
||||||
|
sourceTable = fmt.Sprintf(`"%s"."%s"`, job.SourceTable.Schema, job.SourceTable.Table)
|
||||||
|
} else {
|
||||||
|
sourceTable = fmt.Sprintf("[%s].[%s]", job.SourceTable.Schema, job.SourceTable.Table)
|
||||||
|
}
|
||||||
|
if targetDb.GetDialect() == db_dialects.Postgres {
|
||||||
|
targetTable = fmt.Sprintf(`"%s"."%s"`, job.TargetTable.Schema, job.TargetTable.Table)
|
||||||
|
} else {
|
||||||
|
targetTable = fmt.Sprintf("[%s].[%s]", job.TargetTable.Schema, job.TargetTable.Table)
|
||||||
|
}
|
||||||
|
|
||||||
result := ValidationResult{
|
result := ValidationResult{
|
||||||
JobName: job.Name,
|
JobName: job.Name,
|
||||||
SourceTable: fmt.Sprintf("[%s].[%s]", job.SourceTable.Schema, job.SourceTable.Table),
|
SourceTable: sourceTable,
|
||||||
TargetTable: fmt.Sprintf(`"%s"."%s"`, job.TargetTable.Schema, job.TargetTable.Table),
|
TargetTable: targetTable,
|
||||||
}
|
}
|
||||||
|
|
||||||
var (
|
var (
|
||||||
|
|||||||
Reference in New Issue
Block a user