diff --git a/docs/postgres-compatibility.md b/docs/postgres-compatibility.md index fce6b90c..5f0c1b82 100644 --- a/docs/postgres-compatibility.md +++ b/docs/postgres-compatibility.md @@ -74,9 +74,9 @@ why. | DELETE (incl. `DELETE…USING`, subquery) | ✅ | `dml_test.go::TestDMLDelete`, `::TestDMLDeleteUsing`; transform `wiring_ops_test.go::TestOperatorTransform_DeleteUsing` | | | RETURNING (simple query protocol) | 🟡 | transform `wiring_ops_test.go::TestOperatorTransform_{Insert,Update,Delete}Returning`; `conn_test.go::TestContainsReturning`/`::TestIsDMLReturning` | Differential tests `TestDML{Insert,Update,Delete}Returning` are `skipIfKnown` (stale — passes at unit/client level) | | RETURNING (extended query protocol) | ⛔ | `conn_test.go::TestIsDMLReturning` | Rejected at Describe time with `0A000` by design — Describe would execute the mutation. See CLAUDE.md "DML RETURNING Detection" | -| ON CONFLICT / UPSERT (DO UPDATE) | ✅ | `dml_test.go::TestDMLInsertOnConflict`; transpiler `transform/onconflict_test.go` | Also exercised via DuckLake ON CONFLICT→MERGE rewrite: `ducklake_concurrency_test.go::TestDuckLakeConcurrentTransactions` | +| ON CONFLICT / UPSERT (DO UPDATE) | ✅ | `dml_test.go::TestDMLInsertOnConflict`; transpiler `transform/onconflict_test.go` | Constraint-less lake catalogs reject ON CONFLICT with `0A000` because they cannot enforce unique constraints | | ON CONFLICT DO NOTHING | 🟡 | `dml_test.go::TestDMLInsertOnConflict` (subtest skipped) | | -| MERGE (user-facing) | ⛔ | — | DuckDB has no MERGE; only internal ON CONFLICT→MERGE rewrite exists | +| MERGE (user-facing) | ⛔ | — | Not a PostgreSQL compatibility target; Duckgres does not use MERGE to emulate ON CONFLICT | | TRUNCATE | ✅ | `ddl_test.go::TestDDLTruncate` | | | COPY … FROM STDIN (text/CSV) | ✅ | `copy_test.go::TestCopyFromStdin`, `::TestCopyFromStdinWithSpecialChars`, `::TestCopyFromStdinMultilineJSON` | Escape sequences stored literally (documented DuckDB CSV-parser limitation) | | COPY … TO STDOUT | 🟡 | `copy_test.go::TestCopyToStdout` (skipped under lib/pq); `conn_test.go::TestCopyToStdoutRegex`; client-compat `psycopg` COPY suite | Integration skip is a lib/pq driver limitation, not a Duckgres gap | diff --git a/tests/integration/dml_test.go b/tests/integration/dml_test.go index 38e44b2a..5c22bc1b 100644 --- a/tests/integration/dml_test.go +++ b/tests/integration/dml_test.go @@ -1,6 +1,7 @@ package integration import ( + "strings" "testing" ) @@ -132,6 +133,30 @@ func TestDMLInsertReturning(t *testing.T) { // TestDMLInsertOnConflict tests INSERT ... ON CONFLICT (UPSERT) func TestDMLInsertOnConflict(t *testing.T) { + assertOnConflictRejected := func(t *testing.T, query string) { + t.Helper() + _, err := testHarness.DuckgresDB.Exec(query) + if err == nil { + t.Fatal("expected ON CONFLICT to be rejected") + } + if !strings.Contains(err.Error(), "ON CONFLICT is not supported") { + t.Fatalf("expected ON CONFLICT unsupported error, got %v", err) + } + } + + if testHarness.useDuckLake { + t.Run("on_conflict_do_nothing", func(t *testing.T) { + assertOnConflictRejected(t, "INSERT INTO dml_upsert_test (id, name, counter) VALUES (1, 'Duplicate', 99) ON CONFLICT DO NOTHING") + }) + t.Run("on_conflict_do_update", func(t *testing.T) { + assertOnConflictRejected(t, ` + INSERT INTO dml_upsert_test (id, name, counter) VALUES (1, 'Updated', 20) + ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, counter = dml_upsert_test.counter + EXCLUDED.counter + `) + }) + return + } + mustExec(t, testHarness.DuckgresDB, "CREATE TABLE dml_upsert_test (id INTEGER PRIMARY KEY, name TEXT, counter INTEGER)") mustExec(t, testHarness.DuckgresDB, "INSERT INTO dml_upsert_test VALUES (1, 'Alice', 10)") @@ -149,10 +174,11 @@ func TestDMLInsertOnConflict(t *testing.T) { }) t.Run("on_conflict_do_update", func(t *testing.T) { - _, err := testHarness.DuckgresDB.Exec(` + query := ` INSERT INTO dml_upsert_test (id, name, counter) VALUES (1, 'Updated', 20) ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, counter = dml_upsert_test.counter + EXCLUDED.counter - `) + ` + _, err := testHarness.DuckgresDB.Exec(query) if err != nil { t.Fatalf("ON CONFLICT DO UPDATE failed: %v", err) } diff --git a/tests/integration/ducklake_concurrency_test.go b/tests/integration/ducklake_concurrency_test.go index d9b41232..6fb48914 100644 --- a/tests/integration/ducklake_concurrency_test.go +++ b/tests/integration/ducklake_concurrency_test.go @@ -721,70 +721,31 @@ func runConcurrencyBenchmarks(t *testing.T, latencyMs int, openConn func(*testin t.Logf("read/write isolation: %d write conflicts, %d read errors", writeConflicts.Load(), readErrors.Load()) }) - benchSub(t, "concurrent_upsert_storm", latencyMs, func(t *testing.T, m *metric) { + benchSub(t, "on_conflict_rejected", latencyMs, func(t *testing.T, m *metric) { conn := openConn(t) defer func() { _ = conn.Close() }() - // DuckLake rewrites ON CONFLICT to MERGE — test it under concurrency mustExec(t, conn, "CREATE TABLE dl_conc_upsert (id INTEGER, counter INTEGER, last_ts BIGINT)") - // Seed some rows so upserts hit the update path - for i := range 20 { - mustExec(t, conn, fmt.Sprintf("INSERT INTO dl_conc_upsert VALUES (%d, 0, 0)", i)) - } defer func() { _, _ = conn.Exec("DROP TABLE IF EXISTS dl_conc_upsert") }() - const numWorkers = 8 - const opsPerWorker = 30 - var wg sync.WaitGroup - var conflicts atomic.Int64 - var successes atomic.Int64 - errs := make(chan error, numWorkers*opsPerWorker) - - for w := range numWorkers { - wg.Add(1) - go func(workerID int) { - defer wg.Done() - wconn := openConn(t) - defer func() { _ = wconn.Close() }() - - for i := range opsPerWorker { - // Target overlapping rows across workers - targetID := (workerID + i*3) % 30 // some IDs don't exist yet -> insert path - ts := time.Now().UnixMicro() - // Use literal values: ON CONFLICT → MERGE rewriting breaks $N placeholders - _, err := wconn.Exec(fmt.Sprintf( - `INSERT INTO dl_conc_upsert (id, counter, last_ts) VALUES (%d, 1, %d) - ON CONFLICT (id) DO UPDATE SET counter = dl_conc_upsert.counter + 1, last_ts = %d`, - targetID, ts, ts, - )) - if err != nil { - if isTransactionConflict(err) || isAbortedTransaction(err) { - conflicts.Add(1) - recoverConnection(wconn) - continue - } - // ON CONFLICT may not be supported in DuckLake — log and skip - if strings.Contains(err.Error(), "Not implemented") || - strings.Contains(err.Error(), "not supported") { - t.Logf("upsert not supported: %v", err) - return - } - errs <- fmt.Errorf("worker %d op %d: %w", workerID, i, err) - return - } - successes.Add(1) - } - }(w) + queries := []string{ + `INSERT INTO dl_conc_upsert (id, counter, last_ts) VALUES (1, 1, 1) + ON CONFLICT (id) DO UPDATE SET counter = dl_conc_upsert.counter + 1, last_ts = 1`, + `INSERT INTO dl_conc_upsert (id, counter, last_ts) VALUES (1, 1, 1) + ON CONFLICT (id) DO NOTHING`, } - - wg.Wait() - close(errs) - for err := range errs { - t.Error(err) + for _, query := range queries { + _, err := conn.Exec(query) + if err == nil { + t.Fatalf("expected ON CONFLICT to be rejected") + } + if !strings.Contains(err.Error(), "ON CONFLICT is not supported") { + t.Fatalf("expected ON CONFLICT unsupported error, got %v", err) + } } - m.Successes, m.Conflicts = successes.Load(), conflicts.Load() - t.Logf("upsert storm: %d succeeded, %d conflicts", m.Successes, m.Conflicts) + m.Successes = int64(len(queries)) + t.Logf("ON CONFLICT rejection: %d forms rejected", m.Successes) }) benchSub(t, "concurrent_ddl_while_writing", latencyMs, func(t *testing.T, m *metric) { diff --git a/tests/integration/setup_test.go b/tests/integration/setup_test.go index 6182a4fa..284037a2 100644 --- a/tests/integration/setup_test.go +++ b/tests/integration/setup_test.go @@ -25,7 +25,7 @@ var knownFailures = map[string]string{ "insert_returning_columns": "DuckLake: RETURNING clause not yet supported", "update_returning_star": "DuckLake: RETURNING clause not yet supported", "delete_returning_star": "DuckLake: RETURNING clause not yet supported", - "on_conflict_do_nothing": "DuckLake: constraint detection requires explicit columns", + "on_conflict_do_nothing": "DuckDB: targetless DO NOTHING behavior differs", // DuckDB missing functions "where_regex_match": "DuckDB: ~* operator not available", diff --git a/transpiler/backend/profile.go b/transpiler/backend/profile.go index 662cdfe5..8caeea5b 100644 --- a/transpiler/backend/profile.go +++ b/transpiler/backend/profile.go @@ -20,7 +20,7 @@ const ( NoOpUnsupportedDDL UnsupportedDDLHandling = "noop" UseInsertOnConflict ConflictHandling = "insert_on_conflict" - RewriteToMerge ConflictHandling = "rewrite_to_merge" + RejectOnConflict ConflictHandling = "reject_on_conflict" ) type CatalogPolicy struct { @@ -138,7 +138,7 @@ func lakeProfile(name Name, physical string, mapPublicToMain, hybridDDLGuards bo WarnOnStrippedConstraints: hybridDDLGuards, ErrorOnSilentNullDefaults: hybridDDLGuards, }, - dml: DMLPolicy{ConflictHandling: RewriteToMerge}, + dml: DMLPolicy{ConflictHandling: RejectOnConflict}, metadata: MetadataPolicy{InterceptShowCreate: true}, } } diff --git a/transpiler/backend/profile_test.go b/transpiler/backend/profile_test.go index efa9208b..6bcb8076 100644 --- a/transpiler/backend/profile_test.go +++ b/transpiler/backend/profile_test.go @@ -13,8 +13,8 @@ func TestIcebergProfilePolicies(t *testing.T) { if profile.DDL().ConstraintHandling != StripConstraints { t.Fatalf("ConstraintHandling = %v, want StripConstraints", profile.DDL().ConstraintHandling) } - if profile.DML().ConflictHandling != RewriteToMerge { - t.Fatalf("ConflictHandling = %v, want RewriteToMerge", profile.DML().ConflictHandling) + if profile.DML().ConflictHandling != RejectOnConflict { + t.Fatalf("ConflictHandling = %v, want RejectOnConflict", profile.DML().ConflictHandling) } } diff --git a/transpiler/classify_wiring_test.go b/transpiler/classify_wiring_test.go index 3da41e97..3e9805ee 100644 --- a/transpiler/classify_wiring_test.go +++ b/transpiler/classify_wiring_test.go @@ -271,14 +271,17 @@ func TestClassifyWiring_OnConflictWhitespace(t *testing.T) { wantFlags(t, cfg, "INSERT INTO t (a, b) VALUES (1, 2) ON\tCONFLICT (a) DO NOTHING", FlagOnConflict) dlTr := New(Config{DuckLakeMode: true}) - out := transpileSQL(t, dlTr, "INSERT INTO t (a, b) VALUES (1, 2) ON\nCONFLICT (a) DO NOTHING") - if !strings.Contains(strings.ToUpper(out), "MERGE INTO") { - t.Errorf("DuckLake Transpile(ON\\nCONFLICT) = %q, want MERGE INTO rewrite", out) + res, err := dlTr.Transpile("INSERT INTO t (a, b) VALUES (1, 2) ON\nCONFLICT (a) DO NOTHING") + if err != nil { + t.Fatalf("DuckLake Transpile(ON\\nCONFLICT) error: %v", err) + } + if res.Error == nil { + t.Fatalf("DuckLake Transpile(ON\\nCONFLICT) should reject ON CONFLICT, got SQL=%q", res.SQL) } // False positive (CONFLICT in a string literal) must be a harmless no-op. tr := New(cfg) - out = transpileSQL(t, tr, "SELECT note FROM audit WHERE note = 'ON CONFLICT'") + out := transpileSQL(t, tr, "SELECT note FROM audit WHERE note = 'ON CONFLICT'") if strings.Contains(strings.ToUpper(out), "MERGE") { t.Errorf("Transpile(literal 'ON CONFLICT') = %q, must not rewrite", out) } diff --git a/transpiler/transform/onconflict.go b/transpiler/transform/onconflict.go index 5735bbb3..f8d72627 100644 --- a/transpiler/transform/onconflict.go +++ b/transpiler/transform/onconflict.go @@ -1,35 +1,22 @@ package transform -import ( - "strings" - - pg_query "github.com/pganalyze/pg_query_go/v6" -) +import pg_query "github.com/pganalyze/pg_query_go/v6" // OnConflictTransform handles PostgreSQL ON CONFLICT (upsert) syntax. -// PostgreSQL: INSERT ... ON CONFLICT (columns) DO UPDATE SET ... -// PostgreSQL: INSERT ... ON CONFLICT (columns) DO NOTHING -// DuckDB: INSERT OR REPLACE INTO ... (for simple cases) -// DuckDB: INSERT OR IGNORE INTO ... (for DO NOTHING) -// -// Note: DuckDB's ON CONFLICT support has evolved. As of DuckDB 0.9+, -// it supports ON CONFLICT DO NOTHING and ON CONFLICT DO UPDATE. -// This transform handles cases where the syntax might differ. // -// In DuckLake mode, PRIMARY KEY and UNIQUE constraints are stripped, -// so ON CONFLICT clauses will fail with "columns not referenced by constraint". -// Instead of stripping ON CONFLICT, we convert the INSERT to a MERGE statement -// which DuckLake supports for upserts. +// DuckDB supports ON CONFLICT for catalogs that enforce unique constraints. +// Constraint-less lake catalogs do not, so they reject ON CONFLICT instead of +// trying to emulate PostgreSQL uniqueness semantics. type OnConflictTransform struct { - DuckLakeMode bool + RejectOnConflict bool } func NewOnConflictTransform() *OnConflictTransform { - return &OnConflictTransform{DuckLakeMode: false} + return &OnConflictTransform{RejectOnConflict: false} } -func NewOnConflictTransformWithConfig(duckLakeMode bool) *OnConflictTransform { - return &OnConflictTransform{DuckLakeMode: duckLakeMode} +func NewOnConflictTransformWithConfig(rejectOnConflict bool) *OnConflictTransform { + return &OnConflictTransform{RejectOnConflict: rejectOnConflict} } func (t *OnConflictTransform) Name() string { @@ -37,408 +24,18 @@ func (t *OnConflictTransform) Name() string { } func (t *OnConflictTransform) Transform(tree *pg_query.ParseResult, result *Result) (bool, error) { - changed := false - - for _, stmt := range tree.Stmts { - if stmt.Stmt == nil { - continue - } - - if insert := stmt.Stmt.GetInsertStmt(); insert != nil { - // On a constraint-less backend (MERGE rewrite enabled), ON CONFLICT ON - // CONSTRAINT cannot be honored: there is no named constraint to - // infer the conflict target from. Reject it with a clean PostgreSQL - // error rather than letting it fail opaquely at DuckDB. - if t.DuckLakeMode && insert.OnConflictClause != nil && - insert.OnConflictClause.Infer != nil && - strings.TrimSpace(insert.OnConflictClause.Infer.Conname) != "" { - result.Error = NewFeatureNotSupported( - "ON CONFLICT ON CONSTRAINT is not supported: this catalog does not enforce named constraints") - return false, nil - } - - if mergeStmt := t.transformInsertToMerge(insert); mergeStmt != nil { - // Replace the INSERT statement with MERGE - stmt.Stmt = &pg_query.Node{ - Node: &pg_query.Node_MergeStmt{MergeStmt: mergeStmt}, - } - changed = true - } else if t.transformInsert(insert) { - changed = true - } - } + if !t.RejectOnConflict { + return false, nil } - return changed, nil -} - -// transformInsertToMerge converts INSERT ... ON CONFLICT to MERGE for DuckLake mode -func (t *OnConflictTransform) transformInsertToMerge(insert *pg_query.InsertStmt) *pg_query.MergeStmt { - if !t.DuckLakeMode { - return nil - } - - if insert == nil || insert.OnConflictClause == nil { - return nil - } - - occ := insert.OnConflictClause - - // Need conflict columns to build the join condition - if occ.Infer == nil || len(occ.Infer.IndexElems) == 0 { - return nil - } - - // Get column names from INSERT - colNames := make([]string, len(insert.Cols)) - for i, col := range insert.Cols { - if rt := col.GetResTarget(); rt != nil { - colNames[i] = rt.Name + WalkFunc(tree, func(node *pg_query.Node) bool { + if insert := node.GetInsertStmt(); insert != nil && insert.OnConflictClause != nil { + result.Error = NewFeatureNotSupported( + "ON CONFLICT is not supported: this catalog does not enforce unique constraints") + return false } - } - - // Get the SelectStmt from INSERT - selectStmt := insert.SelectStmt.GetSelectStmt() - if selectStmt == nil { - return nil - } - - // Build source SELECT - handle both VALUES and SELECT ... FROM cases - var sourceSelect *pg_query.SelectStmt - if len(selectStmt.ValuesLists) > 0 { - // INSERT ... VALUES (...) ON CONFLICT - build SELECT from values - sourceSelect = t.buildSourceSelect(colNames, selectStmt.ValuesLists) - } else if len(selectStmt.FromClause) > 0 || len(selectStmt.TargetList) > 0 { - // INSERT ... SELECT ... FROM ... ON CONFLICT - use SELECT directly - // The SELECT already has the right columns, just use it as the source - sourceSelect = selectStmt - } - - if sourceSelect == nil { - return nil - } - - // Build source relation as a subquery with alias "excluded" - sourceRelation := &pg_query.Node{ - Node: &pg_query.Node_RangeSubselect{ - RangeSubselect: &pg_query.RangeSubselect{ - Subquery: &pg_query.Node{ - Node: &pg_query.Node_SelectStmt{SelectStmt: sourceSelect}, - }, - Alias: &pg_query.Alias{Aliasname: "excluded"}, - }, - }, - } - - // Build join condition from conflict columns - joinCondition := t.buildJoinCondition(occ.Infer.IndexElems, insert.Relation.Relname) - - // Build MERGE WHEN clauses - var whenClauses []*pg_query.Node - - // If DO UPDATE, add WHEN MATCHED THEN UPDATE - if occ.Action == pg_query.OnConflictAction_ONCONFLICT_UPDATE { - updateClause := t.buildUpdateClause(occ.TargetList, occ.WhereClause) - whenClauses = append(whenClauses, &pg_query.Node{ - Node: &pg_query.Node_MergeWhenClause{MergeWhenClause: updateClause}, - }) - } - // For DO NOTHING, we skip WHEN MATCHED (no action on match) - - // Always add WHEN NOT MATCHED THEN INSERT - insertClause := t.buildInsertClause(colNames) - whenClauses = append(whenClauses, &pg_query.Node{ - Node: &pg_query.Node_MergeWhenClause{MergeWhenClause: insertClause}, + return true }) - return &pg_query.MergeStmt{ - Relation: insert.Relation, - SourceRelation: sourceRelation, - JoinCondition: joinCondition, - MergeWhenClauses: whenClauses, - } -} - -// buildSourceSelect creates a SELECT statement from VALUES for use as MERGE source -func (t *OnConflictTransform) buildSourceSelect(colNames []string, valuesLists []*pg_query.Node) *pg_query.SelectStmt { - if len(valuesLists) == 0 { - return nil - } - - if len(valuesLists) == 1 { - // Single row: SELECT val1 AS col1, val2 AS col2, ... - valueList := valuesLists[0].GetList() - if valueList == nil || len(valueList.Items) != len(colNames) { - return nil - } - - targetList := make([]*pg_query.Node, len(colNames)) - for i, colName := range colNames { - targetList[i] = &pg_query.Node{ - Node: &pg_query.Node_ResTarget{ - ResTarget: &pg_query.ResTarget{ - Name: colName, - Val: valueList.Items[i], - }, - }, - } - } - - return &pg_query.SelectStmt{ - TargetList: targetList, - LimitOption: pg_query.LimitOption_LIMIT_OPTION_DEFAULT, - Op: pg_query.SetOperation_SETOP_NONE, - } - } - - // Multiple rows: Use UNION ALL of SELECT statements - // First row - firstList := valuesLists[0].GetList() - if firstList == nil || len(firstList.Items) != len(colNames) { - return nil - } - - targetList := make([]*pg_query.Node, len(colNames)) - for i, colName := range colNames { - targetList[i] = &pg_query.Node{ - Node: &pg_query.Node_ResTarget{ - ResTarget: &pg_query.ResTarget{ - Name: colName, - Val: firstList.Items[i], - }, - }, - } - } - - result := &pg_query.SelectStmt{ - TargetList: targetList, - LimitOption: pg_query.LimitOption_LIMIT_OPTION_DEFAULT, - Op: pg_query.SetOperation_SETOP_NONE, - } - - // Add remaining rows as UNION ALL - for i := 1; i < len(valuesLists); i++ { - valueList := valuesLists[i].GetList() - if valueList == nil || len(valueList.Items) != len(colNames) { - continue - } - - rightTargetList := make([]*pg_query.Node, len(colNames)) - for j, colName := range colNames { - rightTargetList[j] = &pg_query.Node{ - Node: &pg_query.Node_ResTarget{ - ResTarget: &pg_query.ResTarget{ - Name: colName, - Val: valueList.Items[j], - }, - }, - } - } - - rightSelect := &pg_query.SelectStmt{ - TargetList: rightTargetList, - LimitOption: pg_query.LimitOption_LIMIT_OPTION_DEFAULT, - Op: pg_query.SetOperation_SETOP_NONE, - } - - result = &pg_query.SelectStmt{ - Op: pg_query.SetOperation_SETOP_UNION, - All: true, - Larg: result, - Rarg: rightSelect, - LimitOption: pg_query.LimitOption_LIMIT_OPTION_DEFAULT, - } - } - - return result -} - -// buildJoinCondition creates the ON condition for MERGE -func (t *OnConflictTransform) buildJoinCondition(indexElems []*pg_query.Node, tableName string) *pg_query.Node { - if len(indexElems) == 0 { - return nil - } - - // Build equality conditions for each conflict column - var conditions []*pg_query.Node - for _, elem := range indexElems { - indexElem := elem.GetIndexElem() - if indexElem == nil { - continue - } - colName := indexElem.Name - - // excluded.col = table.col - condition := &pg_query.Node{ - Node: &pg_query.Node_AExpr{ - AExpr: &pg_query.A_Expr{ - Kind: pg_query.A_Expr_Kind_AEXPR_OP, - Name: []*pg_query.Node{ - {Node: &pg_query.Node_String_{String_: &pg_query.String{Sval: "="}}}, - }, - Lexpr: &pg_query.Node{ - Node: &pg_query.Node_ColumnRef{ - ColumnRef: &pg_query.ColumnRef{ - Fields: []*pg_query.Node{ - {Node: &pg_query.Node_String_{String_: &pg_query.String{Sval: "excluded"}}}, - {Node: &pg_query.Node_String_{String_: &pg_query.String{Sval: colName}}}, - }, - }, - }, - }, - Rexpr: &pg_query.Node{ - Node: &pg_query.Node_ColumnRef{ - ColumnRef: &pg_query.ColumnRef{ - Fields: []*pg_query.Node{ - {Node: &pg_query.Node_String_{String_: &pg_query.String{Sval: tableName}}}, - {Node: &pg_query.Node_String_{String_: &pg_query.String{Sval: colName}}}, - }, - }, - }, - }, - }, - }, - } - conditions = append(conditions, condition) - } - - if len(conditions) == 1 { - return conditions[0] - } - - // Multiple columns: combine with AND - result := conditions[0] - for i := 1; i < len(conditions); i++ { - result = &pg_query.Node{ - Node: &pg_query.Node_BoolExpr{ - BoolExpr: &pg_query.BoolExpr{ - Boolop: pg_query.BoolExprType_AND_EXPR, - Args: []*pg_query.Node{result, conditions[i]}, - }, - }, - } - } - - return result -} - -// buildUpdateClause creates WHEN MATCHED THEN UPDATE clause -func (t *OnConflictTransform) buildUpdateClause(targetList []*pg_query.Node, whereClause *pg_query.Node) *pg_query.MergeWhenClause { - clause := &pg_query.MergeWhenClause{ - MatchKind: pg_query.MergeMatchKind_MERGE_WHEN_MATCHED, - CommandType: pg_query.CmdType_CMD_UPDATE, - } - - if len(targetList) > 0 { - // Specific columns to update - SET col = val, ... - clause.TargetList = targetList - } - // If targetList is empty, it's a full row update (UPDATE without SET) - - if whereClause != nil { - clause.Condition = whereClause - } - - return clause + return false, nil } - -// buildInsertClause creates WHEN NOT MATCHED THEN INSERT clause -func (t *OnConflictTransform) buildInsertClause(colNames []string) *pg_query.MergeWhenClause { - // Build target list (column names) - targetList := make([]*pg_query.Node, len(colNames)) - for i, colName := range colNames { - targetList[i] = &pg_query.Node{ - Node: &pg_query.Node_ResTarget{ - ResTarget: &pg_query.ResTarget{ - Name: colName, - }, - }, - } - } - - // Build values list (references to excluded.col) - values := make([]*pg_query.Node, len(colNames)) - for i, colName := range colNames { - values[i] = &pg_query.Node{ - Node: &pg_query.Node_ColumnRef{ - ColumnRef: &pg_query.ColumnRef{ - Fields: []*pg_query.Node{ - {Node: &pg_query.Node_String_{String_: &pg_query.String{Sval: "excluded"}}}, - {Node: &pg_query.Node_String_{String_: &pg_query.String{Sval: colName}}}, - }, - }, - }, - } - } - - return &pg_query.MergeWhenClause{ - MatchKind: pg_query.MergeMatchKind_MERGE_WHEN_NOT_MATCHED_BY_TARGET, - CommandType: pg_query.CmdType_CMD_INSERT, - TargetList: targetList, - Values: values, - } -} - -func (t *OnConflictTransform) transformInsert(insert *pg_query.InsertStmt) bool { - if insert == nil || insert.OnConflictClause == nil { - return false - } - - // DuckDB now supports ON CONFLICT syntax similar to PostgreSQL - // However, there are some differences: - // - // 1. PostgreSQL allows ON CONFLICT ON CONSTRAINT constraint_name - // DuckDB only supports ON CONFLICT (columns) - // - // 2. PostgreSQL has EXCLUDED pseudo-table - // DuckDB also supports EXCLUDED - // - // 3. PostgreSQL allows WHERE clause in ON CONFLICT - // DuckDB supports this too (as of recent versions) - // - // For most common cases, the syntax is compatible. - // We mainly need to handle edge cases. - - occ := insert.OnConflictClause - - // Check for ON CONFLICT ON CONSTRAINT (not supported in DuckDB) - // The constraint name is in the Infer clause's Conname field - if occ.Infer != nil && occ.Infer.Conname != "" { - // This is ON CONFLICT ON CONSTRAINT constraint_name - // DuckDB doesn't support this syntax - // We could try to look up the constraint columns, but that's complex - // For now, we leave it as-is and let it error with a clear message - return false - } - - // ON CONFLICT (columns) DO NOTHING - supported in DuckDB - // ON CONFLICT (columns) DO UPDATE - supported in DuckDB - // These should work as-is - - return false -} - -// OnConflictNote documents the ON CONFLICT syntax differences: -// -// PostgreSQL: -// INSERT INTO t (a, b) VALUES (1, 2) -// ON CONFLICT (a) DO UPDATE SET b = EXCLUDED.b -// -// INSERT INTO t (a, b) VALUES (1, 2) -// ON CONFLICT (a) DO NOTHING -// -// INSERT INTO t (a, b) VALUES (1, 2) -// ON CONFLICT ON CONSTRAINT pk_name DO UPDATE SET b = EXCLUDED.b -// -// DuckDB (supported): -// INSERT INTO t (a, b) VALUES (1, 2) -// ON CONFLICT (a) DO UPDATE SET b = EXCLUDED.b -// -// INSERT INTO t (a, b) VALUES (1, 2) -// ON CONFLICT (a) DO NOTHING -// -// INSERT OR REPLACE INTO t (a, b) VALUES (1, 2) -- Alternative syntax -// INSERT OR IGNORE INTO t (a, b) VALUES (1, 2) -- Alternative syntax -// -// DuckDB (NOT supported): -// ON CONFLICT ON CONSTRAINT constraint_name -- Use column list instead diff --git a/transpiler/transform/onconflict_test.go b/transpiler/transform/onconflict_test.go index 164b1305..26f7f078 100644 --- a/transpiler/transform/onconflict_test.go +++ b/transpiler/transform/onconflict_test.go @@ -14,8 +14,7 @@ func TestOnConflictTransform_Name(t *testing.T) { } } -func TestOnConflictTransform_NonDuckLakeMode(t *testing.T) { - // In non-DuckLake mode, ON CONFLICT should pass through unchanged +func TestOnConflictTransform_NonRejectingMode(t *testing.T) { tr := NewOnConflictTransformWithConfig(false) tests := []struct { @@ -30,6 +29,14 @@ func TestOnConflictTransform_NonDuckLakeMode(t *testing.T) { name: "DO NOTHING passes through", input: "INSERT INTO users (id, name) VALUES (1, 'test') ON CONFLICT (id) DO NOTHING", }, + { + name: "ON CONSTRAINT passes through", + input: "INSERT INTO users (id, name) VALUES (1, 'test') ON CONFLICT ON CONSTRAINT users_pkey DO NOTHING", + }, + { + name: "targetless DO NOTHING passes through", + input: "INSERT INTO users (id, name) VALUES (1, 'test') ON CONFLICT DO NOTHING", + }, } for _, tt := range tests { @@ -45,12 +52,12 @@ func TestOnConflictTransform_NonDuckLakeMode(t *testing.T) { t.Fatalf("Transform error: %v", err) } - // Should not change in non-DuckLake mode if changed { - t.Error("Transform should not change SQL in non-DuckLake mode") + t.Error("Transform should not change SQL when ON CONFLICT rejection is disabled") + } + if result.Error != nil { + t.Fatalf("Transform returned unexpected error: %v", result.Error) } - - // Should still be an INSERT statement if tree.Stmts[0].Stmt.GetInsertStmt() == nil { t.Error("Statement should remain an INSERT") } @@ -58,87 +65,48 @@ func TestOnConflictTransform_NonDuckLakeMode(t *testing.T) { } } -func TestOnConflictTransform_DuckLakeMode_DoUpdate(t *testing.T) { +func TestOnConflictTransform_RejectingModeRejectsAllOnConflict(t *testing.T) { tr := NewOnConflictTransformWithConfig(true) tests := []struct { - name string - input string - wantContains []string - wantNotContains []string + name string + input string }{ { - name: "single row DO UPDATE", + name: "DO UPDATE", input: "INSERT INTO users (id, name) VALUES (1, 'test') ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name", - wantContains: []string{ - "MERGE INTO users", - "USING (SELECT 1 AS id, 'test' AS name)", - "excluded", - "ON excluded.id = users.id", - "WHEN MATCHED THEN UPDATE SET name = excluded.name", - "WHEN NOT MATCHED THEN INSERT", - }, - wantNotContains: []string{ - "ON CONFLICT", - "EXCLUDED", // should be lowercase "excluded" - }, }, { - name: "multiple columns in SET", - input: "INSERT INTO users (id, name, email) VALUES (1, 'test', 'test@example.com') ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, email = EXCLUDED.email", - wantContains: []string{ - "MERGE INTO users", - "SELECT 1 AS id, 'test' AS name, 'test@example.com' AS email", - "WHEN MATCHED THEN UPDATE SET name = excluded.name, email = excluded.email", - }, + name: "DO NOTHING", + input: "INSERT INTO users (id, name) VALUES (1, 'test') ON CONFLICT (id) DO NOTHING", + }, + { + name: "multiple values", + input: "INSERT INTO users (id, name) VALUES (1, 'first'), (2, 'second') ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name", + }, + { + name: "INSERT SELECT", + input: "INSERT INTO users (id, name) SELECT id, name FROM staging_users ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name", }, { - name: "multiple values become UNION ALL", - input: "INSERT INTO users (id, name) VALUES (1, 'first'), (2, 'second'), (3, 'third') ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name", - wantContains: []string{ - "MERGE INTO users", - "UNION ALL", - "WHEN MATCHED THEN UPDATE", - "WHEN NOT MATCHED THEN INSERT", - }, + name: "ON CONSTRAINT", + input: "INSERT INTO users (id, name) VALUES (1, 'test') ON CONFLICT ON CONSTRAINT users_pkey DO NOTHING", }, { - name: "composite key conflict", - input: "INSERT INTO events (org_id, event_id, data) VALUES (1, 100, 'payload') ON CONFLICT (org_id, event_id) DO UPDATE SET data = EXCLUDED.data", - wantContains: []string{ - "MERGE INTO events", - "excluded.org_id = events.org_id", - "excluded.event_id = events.event_id", - "AND", - }, + name: "expression target", + input: "INSERT INTO users (id, email) VALUES (1, 'a@example.com') ON CONFLICT ((lower(email))) DO NOTHING", }, { - name: "INSERT SELECT FROM staging table (Fivetran pattern)", - input: `INSERT INTO users (id, name, email) SELECT id, name, email FROM staging_table ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, email = EXCLUDED.email`, - wantContains: []string{ - "MERGE INTO users", - "USING (SELECT id, name, email FROM staging_table)", - "excluded", - "ON excluded.id = users.id", - "WHEN MATCHED THEN UPDATE SET name = excluded.name, email = excluded.email", - "WHEN NOT MATCHED THEN INSERT", - }, - wantNotContains: []string{ - "ON CONFLICT", - }, + name: "partial target", + input: "INSERT INTO users (id, email, active) VALUES (1, 'a@example.com', true) ON CONFLICT (email) WHERE active DO NOTHING", }, { - name: "INSERT SELECT with schema-qualified staging table", - input: `INSERT INTO "myschema"."users" (id, name) SELECT "id", "name" FROM "myschema_staging"."temp_users" ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name`, - wantContains: []string{ - "MERGE INTO myschema.users", - "FROM myschema_staging.temp_users", - "WHEN MATCHED THEN UPDATE", - "WHEN NOT MATCHED THEN INSERT", - }, - wantNotContains: []string{ - "ON CONFLICT", - }, + name: "targetless DO NOTHING", + input: "INSERT INTO users (id, name) VALUES (1, 'test') ON CONFLICT DO NOTHING", + }, + { + name: "DEFAULT VALUES", + input: "INSERT INTO users DEFAULT VALUES ON CONFLICT DO NOTHING", }, } @@ -155,71 +123,60 @@ func TestOnConflictTransform_DuckLakeMode_DoUpdate(t *testing.T) { t.Fatalf("Transform error: %v", err) } - if !changed { - t.Error("Transform should change SQL in DuckLake mode") + if changed { + t.Error("Transform should reject without rewriting SQL") } - - // Should now be a MERGE statement - if tree.Stmts[0].Stmt.GetMergeStmt() == nil { - t.Error("Statement should be converted to MERGE") + if result.Error == nil { + t.Fatal("expected ON CONFLICT to be rejected") } - - // Deparse and check contents - sql, err := pg_query.Deparse(tree) - if err != nil { - t.Fatalf("Deparse error: %v", err) + if got := transformErrSQLState(result.Error); got != "0A000" { + t.Fatalf("SQLSTATE = %q, want 0A000", got) } - - for _, want := range tt.wantContains { - if !strings.Contains(sql, want) { - t.Errorf("SQL should contain %q\nGot: %s", want, sql) - } - } - - for _, notWant := range tt.wantNotContains { - if strings.Contains(sql, notWant) { - t.Errorf("SQL should NOT contain %q\nGot: %s", notWant, sql) - } + if !strings.Contains(result.Error.Error(), "ON CONFLICT is not supported") { + t.Fatalf("error = %q, want ON CONFLICT unsupported message", result.Error.Error()) } }) } } -func TestOnConflictTransform_DuckLakeMode_DoNothing(t *testing.T) { +func TestOnConflictTransform_RejectingModeRejectsNestedOnConflict(t *testing.T) { tr := NewOnConflictTransformWithConfig(true) tests := []struct { - name string - input string - wantContains []string - wantNotContains []string + name string + input string }{ { - name: "single row DO NOTHING", - input: "INSERT INTO users (id, name) VALUES (1, 'test') ON CONFLICT (id) DO NOTHING", - wantContains: []string{ - "MERGE INTO users", - "USING (SELECT 1 AS id, 'test' AS name)", - "ON excluded.id = users.id", - "WHEN NOT MATCHED THEN INSERT", - }, - wantNotContains: []string{ - "ON CONFLICT", - "WHEN MATCHED THEN UPDATE", // DO NOTHING should not have UPDATE clause - }, + name: "writable CTE under SELECT", + input: `WITH up AS ( + INSERT INTO users (id, name) VALUES (1, 'test') + ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name + RETURNING id + ) SELECT * FROM up`, + }, + { + name: "writable CTE under INSERT", + input: `WITH up AS ( + INSERT INTO users (id, name) VALUES (1, 'test') + ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name + RETURNING id + ) INSERT INTO audit_users (id) SELECT id FROM up`, + }, + { + name: "writable CTE under UPDATE", + input: `WITH up AS ( + INSERT INTO users (id, name) VALUES (1, 'test') + ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name + RETURNING id + ) UPDATE audit_users SET touched = true FROM up WHERE audit_users.id = up.id`, }, { - name: "multiple values DO NOTHING", - input: "INSERT INTO users (id, name) VALUES (1, 'first'), (2, 'second') ON CONFLICT (id) DO NOTHING", - wantContains: []string{ - "MERGE INTO users", - "UNION ALL", - "WHEN NOT MATCHED THEN INSERT", - }, - wantNotContains: []string{ - "ON CONFLICT", - "WHEN MATCHED", - }, + name: "writable CTE under DELETE", + input: `WITH up AS ( + INSERT INTO users (id, name) VALUES (1, 'test') + ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name + RETURNING id + ) DELETE FROM audit_users USING up WHERE audit_users.id = up.id`, }, } @@ -236,25 +193,14 @@ func TestOnConflictTransform_DuckLakeMode_DoNothing(t *testing.T) { t.Fatalf("Transform error: %v", err) } - if !changed { - t.Error("Transform should change SQL in DuckLake mode") + if changed { + t.Error("Transform should reject without rewriting SQL") } - - sql, err := pg_query.Deparse(tree) - if err != nil { - t.Fatalf("Deparse error: %v", err) + if result.Error == nil { + t.Fatal("expected nested ON CONFLICT to be rejected") } - - for _, want := range tt.wantContains { - if !strings.Contains(sql, want) { - t.Errorf("SQL should contain %q\nGot: %s", want, sql) - } - } - - for _, notWant := range tt.wantNotContains { - if strings.Contains(sql, notWant) { - t.Errorf("SQL should NOT contain %q\nGot: %s", notWant, sql) - } + if got := transformErrSQLState(result.Error); got != "0A000" { + t.Fatalf("SQLSTATE = %q, want 0A000", got) } }) } @@ -301,500 +247,9 @@ func TestOnConflictTransform_NoTransformCases(t *testing.T) { if changed { t.Error("Transform should not change SQL without ON CONFLICT") } - }) - } -} - -func TestOnConflictTransform_PreservesTableName(t *testing.T) { - tr := NewOnConflictTransformWithConfig(true) - - tests := []struct { - name string - input string - tableName string - }{ - { - name: "simple table name", - input: "INSERT INTO users (id, name) VALUES (1, 'test') ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name", - tableName: "users", - }, - { - name: "different table name", - input: "INSERT INTO events (id, data) VALUES (1, 'payload') ON CONFLICT (id) DO UPDATE SET data = EXCLUDED.data", - tableName: "events", - }, - { - name: "table name with underscore", - input: "INSERT INTO user_events (id, event_type) VALUES (1, 'click') ON CONFLICT (id) DO UPDATE SET event_type = EXCLUDED.event_type", - tableName: "user_events", - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - tree, err := pg_query.Parse(tt.input) - if err != nil { - t.Fatalf("Parse error: %v", err) - } - - result := &Result{} - _, err = tr.Transform(tree, result) - if err != nil { - t.Fatalf("Transform error: %v", err) - } - - mergeStmt := tree.Stmts[0].Stmt.GetMergeStmt() - if mergeStmt == nil { - t.Fatal("Expected MERGE statement") - return - } - - if mergeStmt.Relation.Relname != tt.tableName { - t.Errorf("Table name = %q, want %q", mergeStmt.Relation.Relname, tt.tableName) + if result.Error != nil { + t.Fatalf("Transform returned unexpected error: %v", result.Error) } }) } } - -func TestOnConflictTransform_MergeWhenClauses(t *testing.T) { - tr := NewOnConflictTransformWithConfig(true) - - t.Run("DO UPDATE has both WHEN clauses", func(t *testing.T) { - input := "INSERT INTO users (id, name) VALUES (1, 'test') ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name" - tree, err := pg_query.Parse(input) - if err != nil { - t.Fatalf("Parse error: %v", err) - } - - result := &Result{} - _, err = tr.Transform(tree, result) - if err != nil { - t.Fatalf("Transform error: %v", err) - } - - mergeStmt := tree.Stmts[0].Stmt.GetMergeStmt() - if mergeStmt == nil { - t.Fatal("Expected MERGE statement") - return - } - - // Should have 2 WHEN clauses: MATCHED (UPDATE) and NOT MATCHED (INSERT) - if len(mergeStmt.MergeWhenClauses) != 2 { - t.Errorf("Expected 2 WHEN clauses, got %d", len(mergeStmt.MergeWhenClauses)) - } - - // First should be WHEN MATCHED - firstClause := mergeStmt.MergeWhenClauses[0].GetMergeWhenClause() - if firstClause.MatchKind != pg_query.MergeMatchKind_MERGE_WHEN_MATCHED { - t.Errorf("First clause should be WHEN MATCHED, got %v", firstClause.MatchKind) - } - if firstClause.CommandType != pg_query.CmdType_CMD_UPDATE { - t.Errorf("First clause command should be UPDATE, got %v", firstClause.CommandType) - } - - // Second should be WHEN NOT MATCHED - secondClause := mergeStmt.MergeWhenClauses[1].GetMergeWhenClause() - if secondClause.MatchKind != pg_query.MergeMatchKind_MERGE_WHEN_NOT_MATCHED_BY_TARGET { - t.Errorf("Second clause should be WHEN NOT MATCHED, got %v", secondClause.MatchKind) - } - if secondClause.CommandType != pg_query.CmdType_CMD_INSERT { - t.Errorf("Second clause command should be INSERT, got %v", secondClause.CommandType) - } - }) - - t.Run("DO NOTHING has only NOT MATCHED clause", func(t *testing.T) { - input := "INSERT INTO users (id, name) VALUES (1, 'test') ON CONFLICT (id) DO NOTHING" - tree, err := pg_query.Parse(input) - if err != nil { - t.Fatalf("Parse error: %v", err) - } - - result := &Result{} - _, err = tr.Transform(tree, result) - if err != nil { - t.Fatalf("Transform error: %v", err) - } - - mergeStmt := tree.Stmts[0].Stmt.GetMergeStmt() - if mergeStmt == nil { - t.Fatal("Expected MERGE statement") - return - } - - // Should have only 1 WHEN clause: NOT MATCHED (INSERT) - if len(mergeStmt.MergeWhenClauses) != 1 { - t.Errorf("Expected 1 WHEN clause, got %d", len(mergeStmt.MergeWhenClauses)) - } - - // Should be WHEN NOT MATCHED - clause := mergeStmt.MergeWhenClauses[0].GetMergeWhenClause() - if clause.MatchKind != pg_query.MergeMatchKind_MERGE_WHEN_NOT_MATCHED_BY_TARGET { - t.Errorf("Clause should be WHEN NOT MATCHED, got %v", clause.MatchKind) - } - if clause.CommandType != pg_query.CmdType_CMD_INSERT { - t.Errorf("Clause command should be INSERT, got %v", clause.CommandType) - } - }) -} - -func TestOnConflictTransform_SourceSubquery(t *testing.T) { - tr := NewOnConflictTransformWithConfig(true) - - t.Run("single row creates simple SELECT", func(t *testing.T) { - input := "INSERT INTO users (id, name) VALUES (1, 'test') ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name" - tree, err := pg_query.Parse(input) - if err != nil { - t.Fatalf("Parse error: %v", err) - } - - result := &Result{} - _, err = tr.Transform(tree, result) - if err != nil { - t.Fatalf("Transform error: %v", err) - } - - sql, _ := pg_query.Deparse(tree) - - // Should have a simple SELECT, not UNION ALL - if strings.Contains(sql, "UNION ALL") { - t.Errorf("Single row should not use UNION ALL: %s", sql) - } - if !strings.Contains(sql, "SELECT 1 AS id, 'test' AS name") { - t.Errorf("Should have SELECT with aliased columns: %s", sql) - } - }) - - t.Run("multiple rows creates UNION ALL", func(t *testing.T) { - input := "INSERT INTO users (id, name) VALUES (1, 'first'), (2, 'second') ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name" - tree, err := pg_query.Parse(input) - if err != nil { - t.Fatalf("Parse error: %v", err) - } - - result := &Result{} - _, err = tr.Transform(tree, result) - if err != nil { - t.Fatalf("Transform error: %v", err) - } - - sql, _ := pg_query.Deparse(tree) - - if !strings.Contains(sql, "UNION ALL") { - t.Errorf("Multiple rows should use UNION ALL: %s", sql) - } - }) - - t.Run("three rows creates nested UNION ALL", func(t *testing.T) { - input := "INSERT INTO users (id, name) VALUES (1, 'a'), (2, 'b'), (3, 'c') ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name" - tree, err := pg_query.Parse(input) - if err != nil { - t.Fatalf("Parse error: %v", err) - } - - result := &Result{} - _, err = tr.Transform(tree, result) - if err != nil { - t.Fatalf("Transform error: %v", err) - } - - sql, _ := pg_query.Deparse(tree) - - // Should have two UNION ALLs for three rows - count := strings.Count(sql, "UNION ALL") - if count != 2 { - t.Errorf("Three rows should have 2 UNION ALLs, got %d: %s", count, sql) - } - }) -} - -func TestOnConflictTransform_JoinCondition(t *testing.T) { - tr := NewOnConflictTransformWithConfig(true) - - t.Run("single column join", func(t *testing.T) { - input := "INSERT INTO users (id, name) VALUES (1, 'test') ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name" - tree, err := pg_query.Parse(input) - if err != nil { - t.Fatalf("Parse error: %v", err) - } - - result := &Result{} - _, err = tr.Transform(tree, result) - if err != nil { - t.Fatalf("Transform error: %v", err) - } - - sql, _ := pg_query.Deparse(tree) - - if !strings.Contains(sql, "ON excluded.id = users.id") { - t.Errorf("Should have single column join condition: %s", sql) - } - // Should NOT have AND for single column - if strings.Contains(sql, " AND ") { - t.Errorf("Single column should not have AND: %s", sql) - } - }) - - t.Run("two column join", func(t *testing.T) { - input := "INSERT INTO events (org_id, event_id, data) VALUES (1, 100, 'x') ON CONFLICT (org_id, event_id) DO UPDATE SET data = EXCLUDED.data" - tree, err := pg_query.Parse(input) - if err != nil { - t.Fatalf("Parse error: %v", err) - } - - result := &Result{} - _, err = tr.Transform(tree, result) - if err != nil { - t.Fatalf("Transform error: %v", err) - } - - sql, _ := pg_query.Deparse(tree) - - if !strings.Contains(sql, "excluded.org_id = events.org_id") { - t.Errorf("Should have org_id condition: %s", sql) - } - if !strings.Contains(sql, "excluded.event_id = events.event_id") { - t.Errorf("Should have event_id condition: %s", sql) - } - if !strings.Contains(sql, " AND ") { - t.Errorf("Two columns should be joined with AND: %s", sql) - } - }) - - t.Run("three column join", func(t *testing.T) { - input := "INSERT INTO data (a, b, c, val) VALUES (1, 2, 3, 'x') ON CONFLICT (a, b, c) DO UPDATE SET val = EXCLUDED.val" - tree, err := pg_query.Parse(input) - if err != nil { - t.Fatalf("Parse error: %v", err) - } - - result := &Result{} - _, err = tr.Transform(tree, result) - if err != nil { - t.Fatalf("Transform error: %v", err) - } - - sql, _ := pg_query.Deparse(tree) - - // Should have 2 ANDs for 3 columns - count := strings.Count(sql, " AND ") - if count != 2 { - t.Errorf("Three columns should have 2 ANDs, got %d: %s", count, sql) - } - }) -} - -func TestOnConflictTransform_ExcludedAlias(t *testing.T) { - tr := NewOnConflictTransformWithConfig(true) - - input := "INSERT INTO users (id, name, email) VALUES (1, 'test', 'test@example.com') ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, email = EXCLUDED.email" - tree, err := pg_query.Parse(input) - if err != nil { - t.Fatalf("Parse error: %v", err) - } - - result := &Result{} - _, err = tr.Transform(tree, result) - if err != nil { - t.Fatalf("Transform error: %v", err) - } - - sql, _ := pg_query.Deparse(tree) - - // EXCLUDED should be converted to lowercase "excluded" (the alias) - if strings.Contains(sql, "EXCLUDED.") { - t.Errorf("EXCLUDED should be converted to excluded: %s", sql) - } - - // Should use excluded alias in source - if !strings.Contains(sql, ") excluded ON") { - t.Errorf("Source should have 'excluded' alias: %s", sql) - } - - // Should use excluded in UPDATE SET - if !strings.Contains(sql, "name = excluded.name") { - t.Errorf("UPDATE should reference excluded.name: %s", sql) - } - if !strings.Contains(sql, "email = excluded.email") { - t.Errorf("UPDATE should reference excluded.email: %s", sql) - } -} - -func TestOnConflictTransform_InsertClauseValues(t *testing.T) { - tr := NewOnConflictTransformWithConfig(true) - - input := "INSERT INTO users (id, name, email) VALUES (1, 'test', 'test@example.com') ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name" - tree, err := pg_query.Parse(input) - if err != nil { - t.Fatalf("Parse error: %v", err) - } - - result := &Result{} - _, err = tr.Transform(tree, result) - if err != nil { - t.Fatalf("Transform error: %v", err) - } - - sql, _ := pg_query.Deparse(tree) - - // WHEN NOT MATCHED THEN INSERT should reference all columns from excluded - if !strings.Contains(sql, "INSERT (id, name, email) VALUES (excluded.id, excluded.name, excluded.email)") { - t.Errorf("INSERT clause should reference all columns from excluded: %s", sql) - } -} - -func TestOnConflictTransform_DataTypes(t *testing.T) { - tr := NewOnConflictTransformWithConfig(true) - - tests := []struct { - name string - input string - check string - }{ - { - name: "integer values", - input: "INSERT INTO t (id, count) VALUES (1, 100) ON CONFLICT (id) DO UPDATE SET count = EXCLUDED.count", - check: "SELECT 1 AS id, 100 AS count", - }, - { - name: "string values", - input: "INSERT INTO t (id, name) VALUES (1, 'hello world') ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name", - check: "'hello world' AS name", - }, - { - name: "null values", - input: "INSERT INTO t (id, val) VALUES (1, NULL) ON CONFLICT (id) DO UPDATE SET val = EXCLUDED.val", - check: "NULL AS val", - }, - { - name: "negative numbers", - input: "INSERT INTO t (id, val) VALUES (1, -42) ON CONFLICT (id) DO UPDATE SET val = EXCLUDED.val", - check: "-42 AS val", - }, - { - name: "decimal numbers", - input: "INSERT INTO t (id, price) VALUES (1, 19.99) ON CONFLICT (id) DO UPDATE SET price = EXCLUDED.price", - check: "19.99 AS price", - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - tree, err := pg_query.Parse(tt.input) - if err != nil { - t.Fatalf("Parse error: %v", err) - } - - result := &Result{} - _, err = tr.Transform(tree, result) - if err != nil { - t.Fatalf("Transform error: %v", err) - } - - sql, _ := pg_query.Deparse(tree) - - if !strings.Contains(sql, tt.check) { - t.Errorf("SQL should contain %q\nGot: %s", tt.check, sql) - } - }) - } -} - -func TestOnConflictTransform_FivetranPattern(t *testing.T) { - // Test the exact pattern Fivetran uses: INSERT INTO target SELECT FROM staging ON CONFLICT DO UPDATE - tr := NewOnConflictTransformWithConfig(true) - - // Simplified version of the Fivetran query - input := `INSERT INTO "stripe_test"."invoice" ("id", "status", "amount_due", "currency", "_fivetran_synced") - SELECT "id", "status", "amount_due", "currency", "_fivetran_synced" - FROM "stripe_test_staging"."temp_invoice" - ON CONFLICT ("id") DO UPDATE SET - "status" = "excluded"."status", - "amount_due" = "excluded"."amount_due", - "currency" = "excluded"."currency", - "_fivetran_synced" = "excluded"."_fivetran_synced"` - - tree, err := pg_query.Parse(input) - if err != nil { - t.Fatalf("Parse error: %v", err) - } - - result := &Result{} - changed, err := tr.Transform(tree, result) - if err != nil { - t.Fatalf("Transform error: %v", err) - } - - if !changed { - t.Error("Transform should change SQL in DuckLake mode") - } - - // Should be converted to MERGE - if tree.Stmts[0].Stmt.GetMergeStmt() == nil { - t.Fatal("Statement should be converted to MERGE") - } - - sql, err := pg_query.Deparse(tree) - if err != nil { - t.Fatalf("Deparse error: %v", err) - } - - // Verify key parts of the MERGE statement - checks := []string{ - "MERGE INTO stripe_test.invoice", - "USING (SELECT", - "FROM stripe_test_staging.temp_invoice", - "excluded", - "WHEN MATCHED THEN UPDATE SET", - "WHEN NOT MATCHED THEN INSERT", - } - - for _, check := range checks { - if !strings.Contains(sql, check) { - t.Errorf("SQL should contain %q\nGot: %s", check, sql) - } - } - - // Should NOT contain ON CONFLICT - if strings.Contains(sql, "ON CONFLICT") { - t.Errorf("SQL should NOT contain ON CONFLICT\nGot: %s", sql) - } - - t.Logf("Transformed SQL:\n%s", sql) -} - -func TestOnConflictTransform_InsertSelectDoNothing(t *testing.T) { - tr := NewOnConflictTransformWithConfig(true) - - input := `INSERT INTO target (id, data) SELECT id, data FROM source ON CONFLICT (id) DO NOTHING` - - tree, err := pg_query.Parse(input) - if err != nil { - t.Fatalf("Parse error: %v", err) - } - - result := &Result{} - changed, err := tr.Transform(tree, result) - if err != nil { - t.Fatalf("Transform error: %v", err) - } - - if !changed { - t.Error("Transform should change SQL in DuckLake mode") - } - - sql, err := pg_query.Deparse(tree) - if err != nil { - t.Fatalf("Deparse error: %v", err) - } - - // Should have MERGE with only WHEN NOT MATCHED (no UPDATE for DO NOTHING) - if !strings.Contains(sql, "MERGE INTO target") { - t.Errorf("Should have MERGE INTO target: %s", sql) - } - if !strings.Contains(sql, "WHEN NOT MATCHED THEN INSERT") { - t.Errorf("Should have WHEN NOT MATCHED: %s", sql) - } - if strings.Contains(sql, "WHEN MATCHED") { - t.Errorf("Should NOT have WHEN MATCHED for DO NOTHING: %s", sql) - } -} diff --git a/transpiler/transform/transform.go b/transpiler/transform/transform.go index e8ddfbad..f4c77779 100644 --- a/transpiler/transform/transform.go +++ b/transpiler/transform/transform.go @@ -357,6 +357,9 @@ func walkInsertStmt(stmt *pg_query.InsertStmt, fn func(*pg_query.Node) bool) { walkNode(col, fn) } walkNode(stmt.SelectStmt, fn) + if stmt.WithClause != nil { + walkNode(&pg_query.Node{Node: &pg_query.Node_WithClause{WithClause: stmt.WithClause}}, fn) + } for _, target := range stmt.ReturningList { walkNode(target, fn) } @@ -377,6 +380,9 @@ func walkUpdateStmt(stmt *pg_query.UpdateStmt, fn func(*pg_query.Node) bool) { for _, from := range stmt.FromClause { walkNode(from, fn) } + if stmt.WithClause != nil { + walkNode(&pg_query.Node{Node: &pg_query.Node_WithClause{WithClause: stmt.WithClause}}, fn) + } for _, target := range stmt.ReturningList { walkNode(target, fn) } @@ -394,6 +400,9 @@ func walkDeleteStmt(stmt *pg_query.DeleteStmt, fn func(*pg_query.Node) bool) { for _, target := range stmt.ReturningList { walkNode(target, fn) } + if stmt.WithClause != nil { + walkNode(&pg_query.Node{Node: &pg_query.Node_WithClause{WithClause: stmt.WithClause}}, fn) + } } func walkCreateStmt(stmt *pg_query.CreateStmt, fn func(*pg_query.Node) bool) { diff --git a/transpiler/transpiler.go b/transpiler/transpiler.go index 45aaafcd..496f8410 100644 --- a/transpiler/transpiler.go +++ b/transpiler/transpiler.go @@ -85,57 +85,57 @@ func New(cfg Config) *Transpiler { // Core transforms - always registered // Order matters: more specific transforms should come first - // 0. Writable CTE transform - MUST BE FIRST + // 0. ON CONFLICT rejection must run before rewrites that can emit multiple + // statements and skip the rest of the transform pipeline. + t.transforms = append(t.transforms, taggedTransform{FlagOnConflict, transform.NewOnConflictTransformWithConfig(dmlPolicy.ConflictHandling == backend.RejectOnConflict)}) + + // 1. Writable CTE transform - first mutating rewrite. t.transforms = append(t.transforms, taggedTransform{FlagWritableCTE, transform.NewWritableCTETransform()}) - // 1. version() replacement - MUST run before PgCatalogTransform + // 2. version() replacement - MUST run before PgCatalogTransform t.transforms = append(t.transforms, taggedTransform{FlagVersion, transform.NewVersionTransform()}) - // 2. pg_catalog schema and view mappings + // 3. pg_catalog schema and view mappings t.transforms = append(t.transforms, taggedTransform{FlagPgCatalog, transform.NewPgCatalogTransformWithConfig(catalogPolicy.QualifyMacros)}) - // 3. information_schema mappings to compat views + // 4. information_schema mappings to compat views t.transforms = append(t.transforms, taggedTransform{FlagInfoSchema, transform.NewInformationSchemaTransform()}) - // 3.1 Map PostgreSQL "public" schema to DuckDB "main" (backends whose default + // 4.1 Map PostgreSQL "public" schema to DuckDB "main" (backends whose default // schema is "main"; disabled for Iceberg whose schema is literally "public") t.transforms = append(t.transforms, taggedTransform{FlagPublicSchema, transform.NewPublicSchemaTransform(catalogPolicy.MapPublicToMain)}) - // 3.2 Map logical database catalog references to the physical backend catalog + // 4.2 Map logical database catalog references to the physical backend catalog t.transforms = append(t.transforms, taggedTransform{FlagLogicalCatalog, transform.NewLogicalCatalogTransform(cfg.LogicalDatabaseName, catalogPolicy.PhysicalName, catalogPolicy.MapPublicToMain)}) - // 3.3 Literal rewrites (bytea \x hex, bit-string B'..') — MUST run before + // 4.3 Literal rewrites (bytea \x hex, bit-string B'..') — MUST run before // TypeMapping so it still sees the `bytea` type name. t.transforms = append(t.transforms, taggedTransform{FlagLiterals, transform.NewLiteralTransform()}) - // 4. Type mappings (JSONB->JSON, CHAR->TEXT, etc.) + // 5. Type mappings (JSONB->JSON, CHAR->TEXT, etc.) t.transforms = append(t.transforms, taggedTransform{FlagTypeMapping, transform.NewTypeMappingTransform()}) - // 5. Type casts (::regtype -> ::varchar) + // 6. Type casts (::regtype -> ::varchar) t.transforms = append(t.transforms, taggedTransform{FlagTypeCast, transform.NewTypeCastTransform()}) - // 6. Function mappings (array_agg->list, string_to_array->string_split, etc.) + // 7. Function mappings (array_agg->list, string_to_array->string_split, etc.) t.transforms = append(t.transforms, taggedTransform{FlagFunctions, transform.NewFunctionTransform()}) - // 7. Function alias normalization (current_database() -> AS current_database) + // 8. Function alias normalization (current_database() -> AS current_database) t.transforms = append(t.transforms, taggedTransform{FlagFuncAlias, transform.NewFuncAliasTransform()}) - // 8. Boolean predicate normalization for DuckLake/worker execution. + // 9. Boolean predicate normalization for DuckLake/worker execution. t.transforms = append(t.transforms, taggedTransform{FlagBooleanPredicates, transform.NewBooleanPredicateTransform()}) - // 9. Operator mappings (regex operators, etc.) + // 10. Operator mappings (regex operators, etc.) t.transforms = append(t.transforms, taggedTransform{FlagOperators, transform.NewOperatorTransform()}) - // 10. SET/SHOW command handling + // 11. SET/SHOW command handling t.transforms = append(t.transforms, taggedTransform{FlagSetShow, transform.NewSetShowTransform()}) - // 11. _pg_expandarray handling (PostgreSQL array expansion function used by JDBC) + // 12. _pg_expandarray handling (PostgreSQL array expansion function used by JDBC) t.transforms = append(t.transforms, taggedTransform{FlagExpandArray, transform.NewExpandArrayTransform()}) - // 12. ON CONFLICT handling (rewrites ON CONFLICT to MERGE when the backend has - // no enforced unique constraints to infer against) - t.transforms = append(t.transforms, taggedTransform{FlagOnConflict, transform.NewOnConflictTransformWithConfig(dmlPolicy.ConflictHandling == backend.RewriteToMerge)}) - // 13. Locking clause removal (FOR UPDATE, FOR SHARE, etc.) - DuckDB doesn't support these t.transforms = append(t.transforms, taggedTransform{FlagLocking, transform.NewLockingTransform()}) diff --git a/transpiler/transpiler_test.go b/transpiler/transpiler_test.go index b2c652b2..4ecc70db 100644 --- a/transpiler/transpiler_test.go +++ b/transpiler/transpiler_test.go @@ -426,17 +426,60 @@ func TestTranspile_FeatureNotSupported_Rejected(t *testing.T) { } } + onConflictQueries := []struct { + name string + input string + }{ + { + name: "ON CONFLICT DO UPDATE", + input: "INSERT INTO t (id, v) VALUES (1, 2) ON CONFLICT (id) DO UPDATE SET v = EXCLUDED.v", + }, + { + name: "ON CONFLICT DO NOTHING", + input: "INSERT INTO t (id, v) VALUES (1, 2) ON CONFLICT (id) DO NOTHING", + }, + { + name: "ON CONFLICT INSERT SELECT", + input: "INSERT INTO t (id, v) SELECT id, v FROM staging_t ON CONFLICT (id) DO UPDATE SET v = EXCLUDED.v", + }, + { + name: "ON CONFLICT ON CONSTRAINT", + input: "INSERT INTO t (id, v) VALUES (1, 2) ON CONFLICT ON CONSTRAINT t_pkey DO NOTHING", + }, + { + name: "ON CONFLICT expression target", + input: "INSERT INTO t (id, email) VALUES (1, 'a@example.com') ON CONFLICT ((lower(email))) DO NOTHING", + }, + { + name: "ON CONFLICT partial target", + input: "INSERT INTO t (id, email, active) VALUES (1, 'a@example.com', true) ON CONFLICT (email) WHERE active DO NOTHING", + }, + { + name: "ON CONFLICT targetless DO NOTHING", + input: "INSERT INTO t (id, v) VALUES (1, 2) ON CONFLICT DO NOTHING", + }, + { + name: "writable CTE containing ON CONFLICT", + input: `WITH up AS ( + INSERT INTO t (id, v) VALUES (1, 2) + ON CONFLICT (id) DO UPDATE SET v = EXCLUDED.v + RETURNING id + ) SELECT * FROM up`, + }, + } + for _, backend := range []StorageBackend{BackendDuckLake, BackendIceberg} { tr := New(Config{Backend: backend}) - t.Run(string(backend)+"/ON CONFLICT ON CONSTRAINT", func(t *testing.T) { - result, err := tr.Transpile( - "INSERT INTO t (id, v) VALUES (1, 2) ON CONFLICT ON CONSTRAINT t_pkey DO NOTHING") - if err != nil { - t.Fatalf("Transpile error: %v", err) - } - assertFeatureNotSupported(t, result) - }) + for _, tt := range onConflictQueries { + t.Run(string(backend)+"/"+tt.name, func(t *testing.T) { + result, err := tr.Transpile(tt.input) + if err != nil { + t.Fatalf("Transpile error: %v", err) + } + assertFeatureNotSupported(t, result) + }) + } t.Run(string(backend)+"/ALTER COLUMN TYPE USING", func(t *testing.T) { result, err := tr.Transpile( @@ -2164,72 +2207,25 @@ func TestTranspile_OnConflict(t *testing.T) { } func TestTranspile_OnConflict_DuckLakeMode(t *testing.T) { - // In DuckLake mode, ON CONFLICT is converted to MERGE statement - // because PRIMARY KEY/UNIQUE constraints don't exist in DuckLake tests := []struct { - name string - input string - wantMerge bool - contains []string - notContains []string + name string + input string }{ { - name: "ON CONFLICT DO NOTHING becomes MERGE with only INSERT", - input: "INSERT INTO users (id, name) VALUES (1, 'test') ON CONFLICT (id) DO NOTHING", - wantMerge: true, - contains: []string{ - "MERGE INTO users", - "USING (SELECT 1 AS id, 'test' AS name) excluded", - "ON excluded.id = users.id", - "WHEN NOT MATCHED THEN INSERT", - }, - notContains: []string{ - "ON CONFLICT", - "WHEN MATCHED THEN UPDATE", // DO NOTHING means no update action - }, + name: "ON CONFLICT DO NOTHING", + input: "INSERT INTO users (id, name) VALUES (1, 'test') ON CONFLICT (id) DO NOTHING", }, { - name: "ON CONFLICT DO UPDATE becomes MERGE with UPDATE and INSERT", - input: "INSERT INTO users (id, name) VALUES (1, 'test') ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name", - wantMerge: true, - contains: []string{ - "MERGE INTO users", - "USING (SELECT 1 AS id, 'test' AS name) excluded", - "ON excluded.id = users.id", - "WHEN MATCHED THEN UPDATE SET name = excluded.name", - "WHEN NOT MATCHED THEN INSERT", - }, - notContains: []string{ - "ON CONFLICT", - }, + name: "ON CONFLICT DO UPDATE", + input: "INSERT INTO users (id, name) VALUES (1, 'test') ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name", }, { - name: "Multiple values become UNION ALL", - input: "INSERT INTO users (id, name) VALUES (1, 'test'), (2, 'test2') ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name", - wantMerge: true, - contains: []string{ - "MERGE INTO users", - "UNION ALL", - "WHEN MATCHED THEN UPDATE", - "WHEN NOT MATCHED THEN INSERT", - }, - notContains: []string{ - "ON CONFLICT", - }, + name: "multiple values", + input: "INSERT INTO users (id, name) VALUES (1, 'test'), (2, 'test2') ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name", }, { - name: "Multiple conflict columns", - input: "INSERT INTO users (id, org_id, name) VALUES (1, 100, 'test') ON CONFLICT (id, org_id) DO UPDATE SET name = EXCLUDED.name", - wantMerge: true, - contains: []string{ - "MERGE INTO users", - "excluded.id = users.id", - "excluded.org_id = users.org_id", - "AND", // Multiple conditions joined with AND - }, - notContains: []string{ - "ON CONFLICT", - }, + name: "multiple conflict columns", + input: "INSERT INTO users (id, org_id, name) VALUES (1, 100, 'test') ON CONFLICT (id, org_id) DO UPDATE SET name = EXCLUDED.name", }, } @@ -2242,27 +2238,15 @@ func TestTranspile_OnConflict_DuckLakeMode(t *testing.T) { t.Fatalf("Transpile(%q) error: %v", tt.input, err) } - sql := result.SQL - - // Check MERGE statement is generated - if tt.wantMerge { - if !strings.Contains(strings.ToUpper(sql), "MERGE INTO") { - t.Errorf("Transpile(%q) = %q, should contain MERGE INTO", tt.input, sql) - } + if result.Error == nil { + t.Fatalf("expected ON CONFLICT to be rejected, got SQL=%q", result.SQL) } - - // Check expected contents - for _, want := range tt.contains { - if !strings.Contains(sql, want) { - t.Errorf("Transpile(%q) = %q, should contain %q", tt.input, sql, want) - } + coded, ok := result.Error.(interface{ SQLState() string }) + if !ok { + t.Fatalf("error does not expose SQLState: %T", result.Error) } - - // Check contents that should not be present - for _, notWant := range tt.notContains { - if strings.Contains(strings.ToUpper(sql), strings.ToUpper(notWant)) { - t.Errorf("Transpile(%q) = %q, should NOT contain %q", tt.input, sql, notWant) - } + if coded.SQLState() != "0A000" { + t.Fatalf("SQLState = %q, want 0A000", coded.SQLState()) } }) }