Reject lake ON CONFLICT queries

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 |

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

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) {

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",

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

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

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