From aecb66cb599f0089738851be29367226bc1bc148 Mon Sep 17 00:00:00 2001 From: "Ahmad N. F." Date: Wed, 11 Feb 2026 16:51:19 +0700 Subject: [PATCH 1/8] feat: support dry run execution with EXPLAIN --- mc2mc/internal/client/odps.go | 3 +-- mc2mc/internal/query/builder.go | 33 +++++++++++++++++++++++++++------ mc2mc/internal/query/option.go | 6 ++++++ mc2mc/mc2mc.go | 3 +++ 4 files changed, 37 insertions(+), 8 deletions(-) diff --git a/mc2mc/internal/client/odps.go b/mc2mc/internal/client/odps.go index 788ee00..698b040 100644 --- a/mc2mc/internal/client/odps.go +++ b/mc2mc/internal/client/odps.go @@ -39,8 +39,7 @@ func NewODPSClient(logger *slog.Logger, client *odps.Odps) *odpsClient { // when context is cancelled. func (c *odpsClient) ExecSQL(ctx context.Context, query string, additionalHints map[string]string) error { if c.isDryRun { - c.logger.Info("dry run mode, skipping execution") - return nil + c.logger.Info("[DRY-RUN] running query in dry-run mode with EXPLAIN.") } hints := addHints(additionalHints, query) diff --git a/mc2mc/internal/query/builder.go b/mc2mc/internal/query/builder.go index 1e95df1..2a298f7 100644 --- a/mc2mc/internal/query/builder.go +++ b/mc2mc/internal/query/builder.go @@ -30,6 +30,7 @@ type Builder struct { enableAutoPartition bool enablePartitionValue bool + enableDryRun bool } // NewBuilder creates a new query builder with the given options @@ -68,12 +69,6 @@ func (b *Builder) Build() (string, error) { if b.method == MERGE { // split query components hrs, vars, queries := SplitQueryComponents(b.query) - if len(queries) <= 1 { - if b.costAttributionTeam != "" { - return fmt.Sprintf("%s\n%s", b.query, getCostAttributionComment(b.costAttributionTeam)), nil - } - return b.query, nil - } query := b.constructMergeQuery(hrs, vars, queries) return query, nil } @@ -141,11 +136,25 @@ func (b *Builder) Build() (string, error) { } } + if b.enableDryRun { + // append explain to the query for dry run + query = fmt.Sprintf("EXPLAIN \n%s", query) + } + // construct final query with headers, drops, variables and udfs if hr != "" { hr += "\n" } if drops != "" { + if b.enableDryRun { + dropLines := strings.Split(drops, "\n") + var explainedDrops []string + for _, line := range dropLines { + line = strings.TrimSpace(line) + explainedDrops = append(explainedDrops, fmt.Sprintf("EXPLAIN %s", line)) + } + drops = strings.Join(explainedDrops, "\n") + } drops += "\n" } if varsAndUDFs != "" { @@ -201,6 +210,14 @@ func (b *Builder) constructOverridedValues(query string) (string, error) { // constructMergeQueries constructs merge queries with headers and variables func (b *Builder) constructMergeQuery(hrs, vars, queries []string) string { + if !b.enableDryRun && len(queries) == 1 { + // maintaining existing logic for single query + if b.costAttributionTeam != "" { + return fmt.Sprintf("%s\n%s", b.query, getCostAttributionComment(b.costAttributionTeam)) + } + return b.query + } + builder := strings.Builder{} for i, q := range queries { q = strings.TrimSpace(q) @@ -215,6 +232,10 @@ func (b *Builder) constructMergeQuery(hrs, vars, queries []string) string { if variables != "" && !IsDDL(q) { // skip variables if it's ddl builder.WriteString(fmt.Sprintf("%s\n", variables)) } + if b.enableDryRun { + // append explain to the DDL query part + builder.WriteString("EXPLAIN\n") + } builder.WriteString(fmt.Sprintf("%s\n;", q)) if b.costAttributionTeam != "" { builder.WriteString(fmt.Sprintf("\n%s\n", getCostAttributionComment(b.costAttributionTeam))) diff --git a/mc2mc/internal/query/option.go b/mc2mc/internal/query/option.go index 3c03d4c..f1f5ab8 100644 --- a/mc2mc/internal/query/option.go +++ b/mc2mc/internal/query/option.go @@ -52,3 +52,9 @@ func WithCostAttributionLabel(teamName string) Option { b.costAttributionTeam = teamName } } + +func WithDryRun(enable bool) Option { + return func(b *Builder) { + b.enableDryRun = enable + } +} diff --git a/mc2mc/mc2mc.go b/mc2mc/mc2mc.go index ca9cf78..8437c69 100644 --- a/mc2mc/mc2mc.go +++ b/mc2mc/mc2mc.go @@ -84,6 +84,7 @@ func mc2mc(envs []string) error { query.WithPartitionValue(cfg.DevEnablePartitionValue == "true"), query.WithCostAttributionLabel(cfg.CostAttributionTeam), query.WithColumnOrder(), + query.WithDryRun(cfg.DryRun), ).Build() if err != nil { return errors.WithStack(err) @@ -100,6 +101,7 @@ func mc2mc(envs []string) error { query.WithPartitionValue(cfg.DevEnablePartitionValue == "true"), query.WithCostAttributionLabel(cfg.CostAttributionTeam), query.WithColumnOrder(), + query.WithDryRun(cfg.DryRun), ) // -- TODO(START): refactor this part -- @@ -155,6 +157,7 @@ func mc2mc(envs []string) error { query.WithQuery(string(raw)), query.WithCostAttributionLabel(cfg.CostAttributionTeam), query.WithMethod(query.MERGE), + query.WithDryRun(cfg.DryRun), ).Build() if err != nil { return errors.WithStack(err) From fcbdcdefeecc0e85f6bf5415d41f57f4f8d44525 Mon Sep 17 00:00:00 2001 From: "Ahmad N. F." Date: Wed, 11 Feb 2026 22:16:50 +0700 Subject: [PATCH 2/8] migrate query package to ext folder --- mc2mc/{internal => ext}/query/builder.go | 0 mc2mc/{internal => ext}/query/builder_test.go | 0 mc2mc/{internal => ext}/query/helper.go | 0 mc2mc/{internal => ext}/query/helper_test.go | 0 mc2mc/{internal => ext}/query/method.go | 0 mc2mc/{internal => ext}/query/option.go | 0 mc2mc/mc2mc.go | 2 +- 7 files changed, 1 insertion(+), 1 deletion(-) rename mc2mc/{internal => ext}/query/builder.go (100%) rename mc2mc/{internal => ext}/query/builder_test.go (100%) rename mc2mc/{internal => ext}/query/helper.go (100%) rename mc2mc/{internal => ext}/query/helper_test.go (100%) rename mc2mc/{internal => ext}/query/method.go (100%) rename mc2mc/{internal => ext}/query/option.go (100%) diff --git a/mc2mc/internal/query/builder.go b/mc2mc/ext/query/builder.go similarity index 100% rename from mc2mc/internal/query/builder.go rename to mc2mc/ext/query/builder.go diff --git a/mc2mc/internal/query/builder_test.go b/mc2mc/ext/query/builder_test.go similarity index 100% rename from mc2mc/internal/query/builder_test.go rename to mc2mc/ext/query/builder_test.go diff --git a/mc2mc/internal/query/helper.go b/mc2mc/ext/query/helper.go similarity index 100% rename from mc2mc/internal/query/helper.go rename to mc2mc/ext/query/helper.go diff --git a/mc2mc/internal/query/helper_test.go b/mc2mc/ext/query/helper_test.go similarity index 100% rename from mc2mc/internal/query/helper_test.go rename to mc2mc/ext/query/helper_test.go diff --git a/mc2mc/internal/query/method.go b/mc2mc/ext/query/method.go similarity index 100% rename from mc2mc/internal/query/method.go rename to mc2mc/ext/query/method.go diff --git a/mc2mc/internal/query/option.go b/mc2mc/ext/query/option.go similarity index 100% rename from mc2mc/internal/query/option.go rename to mc2mc/ext/query/option.go diff --git a/mc2mc/mc2mc.go b/mc2mc/mc2mc.go index 8437c69..08e6e85 100644 --- a/mc2mc/mc2mc.go +++ b/mc2mc/mc2mc.go @@ -14,10 +14,10 @@ import ( "github.com/pkg/errors" + "github.com/goto/transformers/mc2mc/ext/query" "github.com/goto/transformers/mc2mc/internal/client" "github.com/goto/transformers/mc2mc/internal/config" "github.com/goto/transformers/mc2mc/internal/logger" - "github.com/goto/transformers/mc2mc/internal/query" ) func mc2mc(envs []string) error { From 1e90be29a0a3c3e5715d8a527c1362bd1c592d82 Mon Sep 17 00:00:00 2001 From: "Ahmad N. F." Date: Tue, 17 Feb 2026 16:55:14 +0700 Subject: [PATCH 3/8] migrate test to ext also --- mc2mc/ext/query/builder_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/mc2mc/ext/query/builder_test.go b/mc2mc/ext/query/builder_test.go index 6dc8a94..4c9d77f 100644 --- a/mc2mc/ext/query/builder_test.go +++ b/mc2mc/ext/query/builder_test.go @@ -4,8 +4,8 @@ import ( "context" "testing" + "github.com/goto/transformers/mc2mc/ext/query" "github.com/goto/transformers/mc2mc/internal/logger" - "github.com/goto/transformers/mc2mc/internal/query" "github.com/stretchr/testify/assert" ) From 185a89df7082aa4c5e76202d89f6b5d06ab9e40d Mon Sep 17 00:00:00 2001 From: "Ahmad N. F." Date: Tue, 17 Feb 2026 17:18:35 +0700 Subject: [PATCH 4/8] migrate test to ext also --- mc2mc/ext/query/helper_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/mc2mc/ext/query/helper_test.go b/mc2mc/ext/query/helper_test.go index 655bf8c..57464f0 100644 --- a/mc2mc/ext/query/helper_test.go +++ b/mc2mc/ext/query/helper_test.go @@ -5,7 +5,7 @@ import ( "github.com/stretchr/testify/assert" - "github.com/goto/transformers/mc2mc/internal/query" + "github.com/goto/transformers/mc2mc/ext/query" ) func TestSplitQueryComponents(t *testing.T) { From c6e4473b5d334231880c1968555659421b1125dc Mon Sep 17 00:00:00 2001 From: "Ahmad N. F." Date: Wed, 18 Feb 2026 10:40:59 +0700 Subject: [PATCH 5/8] attach explain on drop query separation, update unit test --- mc2mc/ext/query/builder.go | 11 +- mc2mc/ext/query/builder_test.go | 303 ++++++++++++++++++++++++++++++++ mc2mc/ext/query/helper.go | 8 +- mc2mc/ext/query/helper_test.go | 21 ++- 4 files changed, 329 insertions(+), 14 deletions(-) diff --git a/mc2mc/ext/query/builder.go b/mc2mc/ext/query/builder.go index 2a298f7..4888386 100644 --- a/mc2mc/ext/query/builder.go +++ b/mc2mc/ext/query/builder.go @@ -76,7 +76,7 @@ func (b *Builder) Build() (string, error) { // separate headers, variables and udfs from the query hr, query := SeparateHeadersAndQuery(b.query) varsAndUDFs, query := SeparateVariablesUDFsAndQuery(query) - drops, query := SeparateDropsAndQuery(query) + drops, query := SeparateDropsAndQuery(query, b.enableDryRun) // destination table is required for append and replace method if b.destinationTableID == "" { @@ -146,15 +146,6 @@ func (b *Builder) Build() (string, error) { hr += "\n" } if drops != "" { - if b.enableDryRun { - dropLines := strings.Split(drops, "\n") - var explainedDrops []string - for _, line := range dropLines { - line = strings.TrimSpace(line) - explainedDrops = append(explainedDrops, fmt.Sprintf("EXPLAIN %s", line)) - } - drops = strings.Join(explainedDrops, "\n") - } drops += "\n" } if varsAndUDFs != "" { diff --git a/mc2mc/ext/query/builder_test.go b/mc2mc/ext/query/builder_test.go index 4c9d77f..c9905e9 100644 --- a/mc2mc/ext/query/builder_test.go +++ b/mc2mc/ext/query/builder_test.go @@ -994,6 +994,309 @@ SET append_test.id = 2 }) } +func TestBuilder_BuildWithDryRun(t *testing.T) { + t.Run("returns dry run query for merge load method", func(t *testing.T) { + queryToExecute := `DELETE FROM project.playground.table WHERE id = 1; +INSERT INTO project.playground.table SELECT * FROM project.playground.table_source;` + odspClient := &mockOdpsClient{} + + query, err := query.NewBuilder( + logger.NewDefaultLogger(), + odspClient, + query.WithQuery(queryToExecute), + query.WithMethod(query.MERGE), + query.WithDryRun(true), + ).Build() + + assert.NoError(t, err) + assert.Equal(t, `EXPLAIN +DELETE FROM project.playground.table WHERE id = 1 +; +--*--optimus-break-marker--*-- +EXPLAIN +INSERT INTO project.playground.table SELECT * FROM project.playground.table_source +;`, query) + }) + + t.Run("returns dry run query for APPEND load method", func(t *testing.T) { + queryToExecute := `select * from project.playground.table;` + odspClient := &mockOdpsClient{ + orderedColumns: func() ([]string, error) { + return []string{"col1", "col2", "_partitiontime"}, nil + }, + partitionResult: func() ([]string, error) { + return []string{}, nil + }, + } + destinationTableID := "project.playground.table_destination" + + query, err := query.NewBuilder( + logger.NewDefaultLogger(), + odspClient, + query.WithQuery(queryToExecute), + query.WithMethod(query.APPEND), + query.WithDestination(destinationTableID), + query.WithColumnOrder(), + query.WithDryRun(true), + ).Build() + + assert.NoError(t, err) + assert.Equal(t, `EXPLAIN +INSERT INTO TABLE project.playground.table_destination +SELECT col1, col2, _partitiontime FROM ( +select * from project.playground.table +) +;`, query) + }) + + t.Run("returns dry run query for replace load method", func(t *testing.T) { + queryToExecute := `select * from project.playground.table;` + odspClient := &mockOdpsClient{ + orderedColumns: func() ([]string, error) { + return []string{"col1", "col2", "_partitiontime"}, nil + }, + partitionResult: func() ([]string, error) { + return []string{}, nil + }, + } + destinationTableID := "project.playground.table_destination" + + query, err := query.NewBuilder( + logger.NewDefaultLogger(), + odspClient, + query.WithQuery(queryToExecute), + query.WithMethod(query.REPLACE), + query.WithDestination(destinationTableID), + query.WithColumnOrder(), + query.WithDryRun(true), + ).Build() + + assert.NoError(t, err) + assert.Equal(t, `EXPLAIN +INSERT OVERWRITE TABLE project.playground.table_destination +SELECT col1, col2, _partitiontime FROM ( +select * from project.playground.table +) +;`, query) + }) + + t.Run("returns dry run query for append load method with partition", func(t *testing.T) { + queryToExecute := `select * from project.playground.table;` + odspClient := &mockOdpsClient{ + orderedColumns: func() ([]string, error) { + return []string{"col1", "col2", "_partitiontime"}, nil + }, + partitionResult: func() ([]string, error) { + return []string{"col3"}, nil + }, + } + destinationTableID := "project.playground.table_destination" + + query, err := query.NewBuilder( + logger.NewDefaultLogger(), + odspClient, + query.WithQuery(queryToExecute), + query.WithMethod(query.APPEND), + query.WithDestination(destinationTableID), + query.WithColumnOrder(), + query.WithDryRun(true), + ).Build() + + assert.NoError(t, err) + assert.Equal(t, `EXPLAIN +INSERT INTO TABLE project.playground.table_destination PARTITION (col3) +SELECT col1, col2, _partitiontime FROM ( +select * from project.playground.table +) +;`, query) + }) + + t.Run("returns dry run query with drops and variables", func(t *testing.T) { + queryToExecute := `SET odps.sql.decimal.odps2=true; +DROP TABLE IF EXISTS project.playground.temp_table; +DROP TABLE IF EXISTS project.playground.temp_table_2; +@variable := SELECT 1 as value; +SELECT * FROM project.playground.table;` + odspClient := &mockOdpsClient{ + orderedColumns: func() ([]string, error) { + return []string{"col1", "col2"}, nil + }, + partitionResult: func() ([]string, error) { + return []string{}, nil + }, + } + destinationTableID := "project.playground.table_destination" + + query, err := query.NewBuilder( + logger.NewDefaultLogger(), + odspClient, + query.WithQuery(queryToExecute), + query.WithMethod(query.APPEND), + query.WithDestination(destinationTableID), + query.WithColumnOrder(), + query.WithDryRun(true), + ).Build() + + assert.NoError(t, err) + assert.Equal(t, `SET odps.sql.decimal.odps2=true +; +EXPLAIN +DROP TABLE IF EXISTS project.playground.temp_table +; +EXPLAIN +DROP TABLE IF EXISTS project.playground.temp_table_2 +; +@variable := SELECT 1 as value +; +EXPLAIN +INSERT INTO TABLE project.playground.table_destination +SELECT col1, col2 FROM ( +SELECT * FROM project.playground.table +) +;`, query) + }) + + t.Run("returns dry run query with overridden values", func(t *testing.T) { + queryToExecute := `select * from project.playground.table;` + odspClient := &mockOdpsClient{ + orderedColumns: func() ([]string, error) { + return []string{"col1", "col2", "_partitiontime"}, nil + }, + partitionResult: func() ([]string, error) { + return []string{}, nil + }, + } + destinationTableID := "project.playground.table_destination" + + query, err := query.NewBuilder( + logger.NewDefaultLogger(), + odspClient, + query.WithQuery(queryToExecute), + query.WithMethod(query.APPEND), + query.WithDestination(destinationTableID), + query.WithOverridedValue("_partitiontime", "TIMESTAMP('2021-01-01')"), + query.WithColumnOrder(), + query.WithDryRun(true), + ).Build() + + assert.NoError(t, err) + assert.Equal(t, `EXPLAIN +INSERT INTO TABLE project.playground.table_destination +SELECT col1, col2, _partitiontime FROM ( +SELECT col1, col2, TIMESTAMP('2021-01-01') as _partitiontime FROM ( +select * from project.playground.table +) +) +;`, query) + }) + + t.Run("returns dry run query with cost attribution", func(t *testing.T) { + queryToExecute := `select * from project.playground.table;` + odspClient := &mockOdpsClient{ + orderedColumns: func() ([]string, error) { + return []string{"col1", "col2"}, nil + }, + partitionResult: func() ([]string, error) { + return []string{}, nil + }, + } + destinationTableID := "project.playground.table_destination" + + query, err := query.NewBuilder( + logger.NewDefaultLogger(), + odspClient, + query.WithQuery(queryToExecute), + query.WithMethod(query.APPEND), + query.WithDestination(destinationTableID), + query.WithCostAttributionLabel("test-team"), + query.WithColumnOrder(), + query.WithDryRun(true), + ).Build() + + assert.NoError(t, err) + assert.Contains(t, query, "EXPLAIN") + assert.Contains(t, query, "--cost_attribution_team=test-team") + }) + + t.Run("returns error for empty query in dry run mode", func(t *testing.T) { + odspClient := &mockOdpsClient{} + + queryToExecute, err := query.NewBuilder( + logger.NewDefaultLogger(), + odspClient, + query.WithDryRun(true), + ).Build() + + assert.Error(t, err) + assert.Empty(t, queryToExecute) + assert.Contains(t, err.Error(), "query is required") + }) + + t.Run("returns dry run query for merge method with complex query components", func(t *testing.T) { + queryToExecute := `SET odps.table.append2.enable=true; +@src := SELECT 1 id; +@src2 := SELECT id FROM append_tmp; +CREATE TABLE append_tmp AS SELECT * FROM @src; +CREATE TABLE append_tmp2(id bigint); +MERGE INTO append_test +USING (SELECT * FROM @src2) source +on append_test.id = source.id +WHEN MATCHED THEN UPDATE +SET append_test.id = 2;` + odspClient := &mockOdpsClient{} + + query, err := query.NewBuilder( + logger.NewDefaultLogger(), + odspClient, + query.WithQuery(queryToExecute), + query.WithMethod(query.MERGE), + query.WithCostAttributionLabel("test-team"), + query.WithDryRun(true), + ).Build() + + assert.NoError(t, err) + assert.NoError(t, err) + assert.Equal(t, `SET odps.table.append2.enable=true +; +@src := SELECT 1 id +; +@src2 := SELECT id FROM append_tmp +; +EXPLAIN +CREATE TABLE append_tmp AS SELECT * FROM @src +; +--cost_attribution_team=test-team + + +--*--optimus-break-marker--*-- +SET odps.table.append2.enable=true +; +EXPLAIN +CREATE TABLE append_tmp2(id bigint) +; +--cost_attribution_team=test-team + + +--*--optimus-break-marker--*-- +SET odps.table.append2.enable=true +; +@src := SELECT 1 id +; +@src2 := SELECT id FROM append_tmp +; +EXPLAIN +MERGE INTO append_test +USING (SELECT * FROM @src2) source +on append_test.id = source.id +WHEN MATCHED THEN UPDATE +SET append_test.id = 2 +; +--cost_attribution_team=test-team + +`, query) + }) +} + type mockOdpsClient struct { partitionResult func() ([]string, error) execSQLResult func() error diff --git a/mc2mc/ext/query/helper.go b/mc2mc/ext/query/helper.go index 050ecc2..b4ed9e6 100644 --- a/mc2mc/ext/query/helper.go +++ b/mc2mc/ext/query/helper.go @@ -165,7 +165,7 @@ func SeparateVariablesUDFsAndQuery(query string) (string, string) { return variableUDFStr, queryStr } -func SeparateDropsAndQuery(query string) (string, string) { +func SeparateDropsAndQuery(query string, useExplainInDrops bool) (string, string) { drops := []string{} query = strings.TrimSpace(query) remainingQueries := []string{} @@ -191,7 +191,11 @@ func SeparateDropsAndQuery(query string) (string, string) { dropStr := "" if len(drops) > 0 { for i, drop := range drops { - drops[i] = strings.TrimSpace(drop) + dropStmt := strings.TrimSpace(drop) + if useExplainInDrops { + dropStmt = fmt.Sprintf("EXPLAIN\n%s", dropStmt) + } + drops[i] = dropStmt } dropStr = strings.Join(drops, "\n;\n") dropStr += "\n;" diff --git a/mc2mc/ext/query/helper_test.go b/mc2mc/ext/query/helper_test.go index 57464f0..366abc9 100644 --- a/mc2mc/ext/query/helper_test.go +++ b/mc2mc/ext/query/helper_test.go @@ -359,7 +359,7 @@ USING (SELECT * FROM @src) source on append_test.id = source.id WHEN MATCHED THEN UPDATE SET append_test.id = 2;` - drops, query := query.SeparateDropsAndQuery(q1) + drops, query := query.SeparateDropsAndQuery(q1, false) assert.Empty(t, drops) assert.Equal(t, `MERGE INTO append_test USING (SELECT * FROM @src) source @@ -374,12 +374,29 @@ USING (SELECT * FROM @src) source on append_test.id = source.id WHEN MATCHED THEN UPDATE SET append_test.id = 2;` - drops, query := query.SeparateDropsAndQuery(q1) + drops, query := query.SeparateDropsAndQuery(q1, false) assert.Equal(t, "DROP TABLE IF EXISTS append_test\n;", drops) assert.Equal(t, `MERGE INTO append_test USING (SELECT * FROM @src) source on append_test.id = source.id WHEN MATCHED THEN UPDATE +SET append_test.id = 2`, query) + }) + + t.Run("split multiple drops and query with explain", func(t *testing.T) { + q1 := `DROP TABLE IF EXISTS append_test; +DROP TABLE IF EXISTS append_test_2; +MERGE INTO append_test +USING (SELECT * FROM @src) source +on append_test.id = source.id +WHEN MATCHED THEN UPDATE +SET append_test.id = 2;` + drops, query := query.SeparateDropsAndQuery(q1, true) + assert.Equal(t, "EXPLAIN\nDROP TABLE IF EXISTS append_test\n;\nEXPLAIN\nDROP TABLE IF EXISTS append_test_2\n;", drops) + assert.Equal(t, `MERGE INTO append_test +USING (SELECT * FROM @src) source +on append_test.id = source.id +WHEN MATCHED THEN UPDATE SET append_test.id = 2`, query) }) } From 9aef4a99d8a2fa3ae45816b21e827d1515ed224e Mon Sep 17 00:00:00 2001 From: "Ahmad N. F." Date: Fri, 20 Feb 2026 09:56:39 +0700 Subject: [PATCH 6/8] rename ext to pkg --- mc2mc/mc2mc.go | 2 +- mc2mc/{ext => pkg}/query/builder.go | 0 mc2mc/{ext => pkg}/query/builder_test.go | 2 +- mc2mc/{ext => pkg}/query/helper.go | 0 mc2mc/{ext => pkg}/query/helper_test.go | 2 +- mc2mc/{ext => pkg}/query/method.go | 0 mc2mc/{ext => pkg}/query/option.go | 0 7 files changed, 3 insertions(+), 3 deletions(-) rename mc2mc/{ext => pkg}/query/builder.go (100%) rename mc2mc/{ext => pkg}/query/builder_test.go (99%) rename mc2mc/{ext => pkg}/query/helper.go (100%) rename mc2mc/{ext => pkg}/query/helper_test.go (99%) rename mc2mc/{ext => pkg}/query/method.go (100%) rename mc2mc/{ext => pkg}/query/option.go (100%) diff --git a/mc2mc/mc2mc.go b/mc2mc/mc2mc.go index 08e6e85..79b134e 100644 --- a/mc2mc/mc2mc.go +++ b/mc2mc/mc2mc.go @@ -14,10 +14,10 @@ import ( "github.com/pkg/errors" - "github.com/goto/transformers/mc2mc/ext/query" "github.com/goto/transformers/mc2mc/internal/client" "github.com/goto/transformers/mc2mc/internal/config" "github.com/goto/transformers/mc2mc/internal/logger" + "github.com/goto/transformers/mc2mc/pkg/query" ) func mc2mc(envs []string) error { diff --git a/mc2mc/ext/query/builder.go b/mc2mc/pkg/query/builder.go similarity index 100% rename from mc2mc/ext/query/builder.go rename to mc2mc/pkg/query/builder.go diff --git a/mc2mc/ext/query/builder_test.go b/mc2mc/pkg/query/builder_test.go similarity index 99% rename from mc2mc/ext/query/builder_test.go rename to mc2mc/pkg/query/builder_test.go index c9905e9..8404c5c 100644 --- a/mc2mc/ext/query/builder_test.go +++ b/mc2mc/pkg/query/builder_test.go @@ -4,8 +4,8 @@ import ( "context" "testing" - "github.com/goto/transformers/mc2mc/ext/query" "github.com/goto/transformers/mc2mc/internal/logger" + "github.com/goto/transformers/mc2mc/pkg/query" "github.com/stretchr/testify/assert" ) diff --git a/mc2mc/ext/query/helper.go b/mc2mc/pkg/query/helper.go similarity index 100% rename from mc2mc/ext/query/helper.go rename to mc2mc/pkg/query/helper.go diff --git a/mc2mc/ext/query/helper_test.go b/mc2mc/pkg/query/helper_test.go similarity index 99% rename from mc2mc/ext/query/helper_test.go rename to mc2mc/pkg/query/helper_test.go index 366abc9..954d915 100644 --- a/mc2mc/ext/query/helper_test.go +++ b/mc2mc/pkg/query/helper_test.go @@ -5,7 +5,7 @@ import ( "github.com/stretchr/testify/assert" - "github.com/goto/transformers/mc2mc/ext/query" + "github.com/goto/transformers/mc2mc/pkg/query" ) func TestSplitQueryComponents(t *testing.T) { diff --git a/mc2mc/ext/query/method.go b/mc2mc/pkg/query/method.go similarity index 100% rename from mc2mc/ext/query/method.go rename to mc2mc/pkg/query/method.go diff --git a/mc2mc/ext/query/option.go b/mc2mc/pkg/query/option.go similarity index 100% rename from mc2mc/ext/query/option.go rename to mc2mc/pkg/query/option.go From 00d3a39f18e86ad5ee637890796fd496ea935d05 Mon Sep 17 00:00:00 2001 From: "Ahmad N. F." Date: Fri, 20 Feb 2026 10:53:20 +0700 Subject: [PATCH 7/8] explode drop statements again to attach explain --- mc2mc/pkg/query/builder.go | 13 +++++++++++-- mc2mc/pkg/query/builder_test.go | 10 +++++----- mc2mc/pkg/query/helper.go | 8 ++------ mc2mc/pkg/query/helper_test.go | 21 ++------------------- 4 files changed, 20 insertions(+), 32 deletions(-) diff --git a/mc2mc/pkg/query/builder.go b/mc2mc/pkg/query/builder.go index 4888386..f94e913 100644 --- a/mc2mc/pkg/query/builder.go +++ b/mc2mc/pkg/query/builder.go @@ -76,7 +76,7 @@ func (b *Builder) Build() (string, error) { // separate headers, variables and udfs from the query hr, query := SeparateHeadersAndQuery(b.query) varsAndUDFs, query := SeparateVariablesUDFsAndQuery(query) - drops, query := SeparateDropsAndQuery(query, b.enableDryRun) + drops, query := SeparateDropsAndQuery(query) // destination table is required for append and replace method if b.destinationTableID == "" { @@ -138,7 +138,7 @@ func (b *Builder) Build() (string, error) { if b.enableDryRun { // append explain to the query for dry run - query = fmt.Sprintf("EXPLAIN \n%s", query) + query = fmt.Sprintf("EXPLAIN\n%s", query) } // construct final query with headers, drops, variables and udfs @@ -146,6 +146,15 @@ func (b *Builder) Build() (string, error) { hr += "\n" } if drops != "" { + if b.enableDryRun { + // explode drops into separate statements for dry run to ensure they are included in the explain plan + dropStatements := strings.Split(strings.TrimSuffix(drops, "\n;"), "\n;\n") + for i, drop := range dropStatements { + dropStatements[i] = fmt.Sprintf("EXPLAIN\n%s\n;", drop) + } + // rejoin the drop statements back together + drops = strings.Join(dropStatements, "\n") + } drops += "\n" } if varsAndUDFs != "" { diff --git a/mc2mc/pkg/query/builder_test.go b/mc2mc/pkg/query/builder_test.go index 8404c5c..64292df 100644 --- a/mc2mc/pkg/query/builder_test.go +++ b/mc2mc/pkg/query/builder_test.go @@ -1041,7 +1041,7 @@ INSERT INTO project.playground.table SELECT * FROM project.playground.table_sour ).Build() assert.NoError(t, err) - assert.Equal(t, `EXPLAIN + assert.Equal(t, `EXPLAIN INSERT INTO TABLE project.playground.table_destination SELECT col1, col2, _partitiontime FROM ( select * from project.playground.table @@ -1072,7 +1072,7 @@ select * from project.playground.table ).Build() assert.NoError(t, err) - assert.Equal(t, `EXPLAIN + assert.Equal(t, `EXPLAIN INSERT OVERWRITE TABLE project.playground.table_destination SELECT col1, col2, _partitiontime FROM ( select * from project.playground.table @@ -1103,7 +1103,7 @@ select * from project.playground.table ).Build() assert.NoError(t, err) - assert.Equal(t, `EXPLAIN + assert.Equal(t, `EXPLAIN INSERT INTO TABLE project.playground.table_destination PARTITION (col3) SELECT col1, col2, _partitiontime FROM ( select * from project.playground.table @@ -1148,7 +1148,7 @@ DROP TABLE IF EXISTS project.playground.temp_table_2 ; @variable := SELECT 1 as value ; -EXPLAIN +EXPLAIN INSERT INTO TABLE project.playground.table_destination SELECT col1, col2 FROM ( SELECT * FROM project.playground.table @@ -1180,7 +1180,7 @@ SELECT * FROM project.playground.table ).Build() assert.NoError(t, err) - assert.Equal(t, `EXPLAIN + assert.Equal(t, `EXPLAIN INSERT INTO TABLE project.playground.table_destination SELECT col1, col2, _partitiontime FROM ( SELECT col1, col2, TIMESTAMP('2021-01-01') as _partitiontime FROM ( diff --git a/mc2mc/pkg/query/helper.go b/mc2mc/pkg/query/helper.go index b4ed9e6..050ecc2 100644 --- a/mc2mc/pkg/query/helper.go +++ b/mc2mc/pkg/query/helper.go @@ -165,7 +165,7 @@ func SeparateVariablesUDFsAndQuery(query string) (string, string) { return variableUDFStr, queryStr } -func SeparateDropsAndQuery(query string, useExplainInDrops bool) (string, string) { +func SeparateDropsAndQuery(query string) (string, string) { drops := []string{} query = strings.TrimSpace(query) remainingQueries := []string{} @@ -191,11 +191,7 @@ func SeparateDropsAndQuery(query string, useExplainInDrops bool) (string, string dropStr := "" if len(drops) > 0 { for i, drop := range drops { - dropStmt := strings.TrimSpace(drop) - if useExplainInDrops { - dropStmt = fmt.Sprintf("EXPLAIN\n%s", dropStmt) - } - drops[i] = dropStmt + drops[i] = strings.TrimSpace(drop) } dropStr = strings.Join(drops, "\n;\n") dropStr += "\n;" diff --git a/mc2mc/pkg/query/helper_test.go b/mc2mc/pkg/query/helper_test.go index 954d915..ebed786 100644 --- a/mc2mc/pkg/query/helper_test.go +++ b/mc2mc/pkg/query/helper_test.go @@ -359,7 +359,7 @@ USING (SELECT * FROM @src) source on append_test.id = source.id WHEN MATCHED THEN UPDATE SET append_test.id = 2;` - drops, query := query.SeparateDropsAndQuery(q1, false) + drops, query := query.SeparateDropsAndQuery(q1) assert.Empty(t, drops) assert.Equal(t, `MERGE INTO append_test USING (SELECT * FROM @src) source @@ -374,29 +374,12 @@ USING (SELECT * FROM @src) source on append_test.id = source.id WHEN MATCHED THEN UPDATE SET append_test.id = 2;` - drops, query := query.SeparateDropsAndQuery(q1, false) + drops, query := query.SeparateDropsAndQuery(q1) assert.Equal(t, "DROP TABLE IF EXISTS append_test\n;", drops) assert.Equal(t, `MERGE INTO append_test USING (SELECT * FROM @src) source on append_test.id = source.id WHEN MATCHED THEN UPDATE -SET append_test.id = 2`, query) - }) - - t.Run("split multiple drops and query with explain", func(t *testing.T) { - q1 := `DROP TABLE IF EXISTS append_test; -DROP TABLE IF EXISTS append_test_2; -MERGE INTO append_test -USING (SELECT * FROM @src) source -on append_test.id = source.id -WHEN MATCHED THEN UPDATE -SET append_test.id = 2;` - drops, query := query.SeparateDropsAndQuery(q1, true) - assert.Equal(t, "EXPLAIN\nDROP TABLE IF EXISTS append_test\n;\nEXPLAIN\nDROP TABLE IF EXISTS append_test_2\n;", drops) - assert.Equal(t, `MERGE INTO append_test -USING (SELECT * FROM @src) source -on append_test.id = source.id -WHEN MATCHED THEN UPDATE SET append_test.id = 2`, query) }) } From 50437947f5d4f90b133949be9cd2ff64d69f6f56 Mon Sep 17 00:00:00 2001 From: "Ahmad N. F." Date: Fri, 20 Feb 2026 11:33:59 +0700 Subject: [PATCH 8/8] return list of drop statement instead of string of drops --- mc2mc/pkg/query/builder.go | 16 +++++++--------- mc2mc/pkg/query/helper.go | 15 +++------------ mc2mc/pkg/query/helper_test.go | 2 +- 3 files changed, 11 insertions(+), 22 deletions(-) diff --git a/mc2mc/pkg/query/builder.go b/mc2mc/pkg/query/builder.go index f94e913..06710e9 100644 --- a/mc2mc/pkg/query/builder.go +++ b/mc2mc/pkg/query/builder.go @@ -145,23 +145,21 @@ func (b *Builder) Build() (string, error) { if hr != "" { hr += "\n" } - if drops != "" { + dropsStr := "" + if len(drops) > 0 { if b.enableDryRun { - // explode drops into separate statements for dry run to ensure they are included in the explain plan - dropStatements := strings.Split(strings.TrimSuffix(drops, "\n;"), "\n;\n") - for i, drop := range dropStatements { - dropStatements[i] = fmt.Sprintf("EXPLAIN\n%s\n;", drop) + for i, drop := range drops { + drops[i] = fmt.Sprintf("EXPLAIN\n%s", drop) } - // rejoin the drop statements back together - drops = strings.Join(dropStatements, "\n") } - drops += "\n" + dropsStr = strings.Join(drops, "\n;\n") + dropsStr += "\n;\n" } if varsAndUDFs != "" { varsAndUDFs += "\n" } - query = fmt.Sprintf("%s%s%s%s", hr, drops, varsAndUDFs, query) + query = fmt.Sprintf("%s%s%s%s", hr, dropsStr, varsAndUDFs, query) if b.costAttributionTeam != "" { query = fmt.Sprintf("%s\n%s\n", query, getCostAttributionComment(b.costAttributionTeam)) } diff --git a/mc2mc/pkg/query/helper.go b/mc2mc/pkg/query/helper.go index 050ecc2..bf6cceb 100644 --- a/mc2mc/pkg/query/helper.go +++ b/mc2mc/pkg/query/helper.go @@ -165,7 +165,7 @@ func SeparateVariablesUDFsAndQuery(query string) (string, string) { return variableUDFStr, queryStr } -func SeparateDropsAndQuery(query string) (string, string) { +func SeparateDropsAndQuery(query string) ([]string, string) { drops := []string{} query = strings.TrimSpace(query) remainingQueries := []string{} @@ -179,7 +179,7 @@ func SeparateDropsAndQuery(query string) (string, string) { } stmtWithoutComment := RemoveComments(stmt) if dropPattern.MatchString(strings.TrimSpace(stmtWithoutComment)) { - drops = append(drops, stmt) + drops = append(drops, strings.TrimSpace(stmt)) } else if strings.TrimSpace(stmtWithoutComment) == "" { // if the statement is empty, it's a comment, then omit it // since it doesn't make sense to execute this statement @@ -188,19 +188,10 @@ func SeparateDropsAndQuery(query string) (string, string) { } } - dropStr := "" - if len(drops) > 0 { - for i, drop := range drops { - drops[i] = strings.TrimSpace(drop) - } - dropStr = strings.Join(drops, "\n;\n") - dropStr += "\n;" - } - // join the remaining queries back together queryStr := strings.Join(remainingQueries, "\n;\n") - return dropStr, queryStr + return drops, queryStr } func RemoveComments(query string) string { diff --git a/mc2mc/pkg/query/helper_test.go b/mc2mc/pkg/query/helper_test.go index ebed786..ec6e16d 100644 --- a/mc2mc/pkg/query/helper_test.go +++ b/mc2mc/pkg/query/helper_test.go @@ -375,7 +375,7 @@ on append_test.id = source.id WHEN MATCHED THEN UPDATE SET append_test.id = 2;` drops, query := query.SeparateDropsAndQuery(q1) - assert.Equal(t, "DROP TABLE IF EXISTS append_test\n;", drops) + assert.Equal(t, []string{"DROP TABLE IF EXISTS append_test"}, drops) assert.Equal(t, `MERGE INTO append_test USING (SELECT * FROM @src) source on append_test.id = source.id