fix(controlplane): default passthrough sessions into the tenant catalog

On the remote-worker (MTCP) path, passthrough sessions skipped the entire
session-init block, leaving the worker's DuckDB connection in its empty
in-memory catalog. As a result `current_database()` returned `memory` and
any unqualified DDL/DML executed against the ephemeral in-memory catalog
instead of the tenant's DuckLake warehouse — verified live (6/6 fresh
passthrough connections reported `current_database()=memory`).

Standalone passthrough already lands in the right place via
server.setDuckLakeDefault / setIcebergDefault in CreatePassthroughDBConnection;
the remote-worker path had no equivalent. Add a passthrough branch that detects
the attached `ducklake` catalog and issues the matching `USE` (preferring an
iceberg-default user's configured catalog), so passthrough sessions default
into the tenant catalog just like standalone and non-passthrough sessions.

This does NOT enable logical-catalog mapping for passthrough: those sessions
still talk raw DuckDB against the physical catalog name (`ducklake`), so
SetLogicalCatalogMapping stays disabled for them.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: fuziontech

controlplane/control.go

@@ -1126,9 +1126,12 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) {
 		}
 	}()
 
-	// Passthrough users skip pg_catalog initialization and DuckLake catalog
-	// detection — they bypass the PG compatibility layer entirely, so the
-	// metadata setup that drives logical catalog mapping is unused for them.
+	// Passthrough users skip pg_catalog initialization and logical-catalog
+	// mapping — they bypass the PG compatibility layer entirely. They still
+	// need a default catalog, though: without one the worker session stays in
+	// DuckDB's empty in-memory catalog, so current_database() reports "memory"
+	// and unqualified DDL/DML never reaches the warehouse (see the passthrough
+	// branch below).
 	var duckLakeAttached bool
 	if !passthroughUser {
 		initCtx, initCancel := context.WithTimeout(context.Background(), cp.cfg.SessionInitTimeout)
@@ -1170,6 +1173,31 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) {
 				slog.Warn("Failed to apply client connect-time search_path; using default.", "user", username, "org", orgID, "search_path", clientSearchPath, "error", err)
 			}
 		}
+	} else {
+		// Passthrough: no pg_catalog views and no logical-catalog rewriting, but
+		// the session must still land in the tenant's catalog instead of the
+		// empty in-memory one. Standalone passthrough does this via
+		// server.setDuckLakeDefault/setIcebergDefault; the remote-worker path
+		// has to issue the equivalent explicitly here.
+		initCtx, initCancel := context.WithTimeout(context.Background(), cp.cfg.SessionInitTimeout)
+		duckLakeAttached, err = sessionmeta.HasAttachedCatalog(initCtx, executor, "ducklake")
+		if err != nil {
+			initCancel()
+			slog.Error("Failed to detect ducklake catalog attachment.", "user", username, "org", orgID, "database", database, "remote_addr", remoteAddr, "error", err, "worker", workerID, "worker_pod", workerPod)
+			_ = server.WriteErrorResponse(writer, "FATAL", "XX000", "failed to detect ducklake catalog attachment")
+			_ = writer.Flush()
+			return
+		}
+		if cmd := passthroughSessionDefaultCatalogCommand(defaultCatalog, duckLakeAttached); cmd != "" {
+			if _, err := executor.ExecContext(initCtx, cmd); err != nil {
+				initCancel()
+				slog.Error("Failed to apply passthrough session default catalog.", "user", username, "org", orgID, "command", cmd, "error", err, "worker", workerID, "worker_pod", workerPod)
+				_ = server.WriteErrorResponse(writer, "FATAL", "XX000", "failed to apply default catalog")
+				_ = writer.Flush()
+				return
+			}
+		}
+		initCancel()
 	}
 
 	// Register the TCP connection so OnWorkerCrash can close it to unblock
@@ -1183,7 +1211,11 @@ func (cp *ControlPlane) handleConnection(conn net.Conn) {
 			server.SetConnectionIcebergConfig(cc, icebergCfg)
 		}
 	}
-	server.SetLogicalCatalogMapping(cc, duckLakeAttached)
+	// Logical-catalog mapping (current_database masking + USE/qualified-name
+	// rewriting) is a non-passthrough feature; passthrough sessions talk raw
+	// DuckDB against the physical catalog name, so keep it disabled for them
+	// even though duckLakeAttached is now populated on the passthrough path.
+	server.SetLogicalCatalogMapping(cc, duckLakeAttached && !passthroughUser)
 	server.SetPassthrough(cc, passthroughUser)
 	if orgID != "" {
 		observeOrgPgSessionAccepted(orgID, passthroughUser)

controlplane/session_search_path.go

@@ -25,6 +25,24 @@ func effectiveSessionDefaultCommand(clientSearchPath, defaultCatalog string) (st
 	}
 }
 
+// passthroughSessionDefaultCatalogCommand returns the connect-time command that
+// points a passthrough session at the tenant's catalog. Passthrough users skip
+// InitSessionDatabaseMetadata (whose defer issues `USE ducklake` for the
+// standard path), so without this the session stays in DuckDB's empty in-memory
+// catalog — current_database() reports "memory" and unqualified DDL/DML never
+// reaches the warehouse. Mirrors server.setIcebergDefault / setDuckLakeDefault
+// used by the standalone passthrough path.
+func passthroughSessionDefaultCatalogCommand(defaultCatalog string, duckLakeAttached bool) string {
+	switch {
+	case defaultCatalog == iceberg.CatalogName:
+		return fmt.Sprintf("USE %s.%s", iceberg.CatalogName, iceberg.DefaultSchema)
+	case duckLakeAttached:
+		return "USE ducklake"
+	default:
+		return ""
+	}
+}
+
 func ensureMemoryMainInSearchPath(searchPath string) string {
 	if strings.Contains(strings.ToLower(searchPath), "memory.main") {
 		return searchPath

controlplane/session_search_path_test.go

@@ -31,3 +31,24 @@ func TestEffectiveSessionDefaultCommandReturnsEmptyWhenUnset(t *testing.T) {
 		t.Fatalf("source = %q, want empty", source)
 	}
 }
+
+func TestPassthroughSessionDefaultCatalogCommand(t *testing.T) {
+	tests := []struct {
+		name             string
+		defaultCatalog   string
+		duckLakeAttached bool
+		want             string
+	}{
+		{name: "ducklake attached defaults to ducklake", defaultCatalog: "", duckLakeAttached: true, want: "USE ducklake"},
+		{name: "iceberg-default user prefers iceberg over ducklake", defaultCatalog: "iceberg", duckLakeAttached: true, want: "USE iceberg.public"},
+		{name: "iceberg-default user with no ducklake", defaultCatalog: "iceberg", duckLakeAttached: false, want: "USE iceberg.public"},
+		{name: "no ducklake and no configured catalog leaves session as-is", defaultCatalog: "", duckLakeAttached: false, want: ""},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := passthroughSessionDefaultCatalogCommand(tt.defaultCatalog, tt.duckLakeAttached); got != tt.want {
+				t.Fatalf("command = %q, want %q", got, tt.want)
+			}
+		})
+	}
+}