feat: accept an org's Trino catalog name as a logical catalog alias

CLAUDE.md

@@ -180,7 +180,7 @@ The project uses [just](https://github.com/casey/just) as a command runner. Run
 
 **Every feature, behavior change, bugfix, AND refactor that affects runtime or
 cluster behavior MUST ship with a solid end-to-end test case in
-`tests/e2e-mw-dev/` (`harness.sh`).** This is not just for new features — any
+`tests/mw-dev/e2e/` (`harness.sh`).** This is not just for new features — any
 change to how the system behaves at runtime (new capability, changed semantics,
 a fixed bug, a new config knob, an activation/routing/teardown tweak) extends or
 adds a harness assertion in the same PR. Refactors count too: when you move or
@@ -202,11 +202,11 @@ Three test lanes worth knowing about, in increasing order of blast radius:
 
 - **Unit / package tests** (`go test ./...`): in-process, no external deps. Where most coverage lives. Includes `tests/manifests/` (static-manifest artifact asserts for `k8s/rbac.yaml` + `k8s/networkpolicy.yaml`).
 - **`tests/integration/`** (`just test-integration`): spins up the standalone server binary against a real MinIO + Postgres metadata store via docker compose. Covers wire protocol, DuckLake on real S3-compatible storage, transpilation against a live server.
-- **`tests/e2e-mw-dev/`** (per-PR GitHub workflow `e2e-mw-dev.yml`): the full multi-tenant activation pipeline against the **real posthog-mw-dev EKS cluster** — real Cilium, real Crossplane ducklings, real cnpg-shard + external-RDS metadata, real AWS S3. A shell harness (`harness.sh`) runs as an in-cluster Job per PR; `run.sh` orchestrates deploy/test/teardown/e2e-cleanup. **Replaces the retired kind suite** (`tests/k8s/`) — that suite's `k8s-integration-tests` CI job and its Go tests are gone; the supporting `k8s/` scripts/manifests + Dockerfiles are kept for now. See `tests/e2e-mw-dev/README.md`.
+- **`tests/mw-dev/e2e/`** (per-PR GitHub workflow `e2e-mw-dev.yml`): the full multi-tenant activation pipeline against the **real posthog-mw-dev EKS cluster** — real Cilium, real Crossplane ducklings, real cnpg-shard + external-RDS metadata, real AWS S3. A shell harness (`harness.sh`) runs as an in-cluster Job per PR; `run.sh` orchestrates deploy/test/teardown/e2e-cleanup. **Replaces the retired kind suite** (`tests/k8s/`) — that suite's `k8s-integration-tests` CI job and its Go tests are gone; the supporting `k8s/` scripts/manifests + Dockerfiles are kept for now. See `tests/mw-dev/README.md`.
 
 ### When code changes obligate test changes
 
-`tests/e2e-mw-dev/` is the only place we exercise the full activation pipeline (control plane → STS broker → worker pod → DuckDB → ATTACH against real cloud storage). If your change touches any of the following, treat updating the harness as part of the change, not a follow-up:
+`tests/mw-dev/e2e/` is the only place we exercise the full activation pipeline (control plane → STS broker → worker pod → DuckDB → ATTACH against real cloud storage). If your change touches any of the following, treat updating the harness as part of the change, not a follow-up:
 
 - `controlplane/shared_worker_activator.go`, `controlplane/sts_broker.go`, anything in the activation payload shape (`TenantActivationPayload`, `server.DuckLakeConfig`)
 - `server/server.go::AttachDeltaCatalog`, `server.attachDuckLake*`, `server.refresh*Secret`
@@ -1716,6 +1716,39 @@ password/tenant/catalog changes never propagate.
   the `ui/src/lib/trino.test.ts` derivations and
   `tests/mw-dev/e2e/trino.sh`.
 
+## Logical Catalog Alias (`org_<database_name>` as the startup `database`)
+
+A pgwire session may select its catalog by the name the org has on Trino
+(`configstore.TrinoCatalogName`, `org_<sanitized database_name>`) instead of
+`ducklake`. The catalog is the same one either way: the alias only renames it
+on the wire. This exists so SQLMesh sees ONE catalog name across the Duckgres
+and Trino engines and migrating between them needs no state rewrite.
+
+- **What the startup `database` may be**: `""`, `ducklake`, or the
+  SNI-resolved org's own catalog name. Everything else is 3D000, as before.
+- **Identity is still SNI-only, and this must stay true.** The alias is
+  compared against the catalog name of the org the managed hostname ALREADY
+  resolved — it is never a key into `DatabaseOrg`, `Orgs`, or any other map,
+  so it can neither discover nor select an org. A sibling tenant's catalog
+  name is just an unrecognized string and fails closed. If a change here ever
+  looks up an org BY the database name, it has reintroduced exactly what PR
+  #651 removed. See the comment in `ResolvePostgresConnection`.
+- **Physical vs. visible**: `EffectiveCatalog` stays `ducklake` and is what
+  every statement executes against; `LogicalCatalog` (and
+  `sessionMetadataResult.visibleCatalog`) is the name reported by
+  `current_database()`, `pg_database`, `information_schema`, and the logs.
+  `visibleCatalogName` is the one place that chooses between them.
+- **SQL written against the alias**: the transpiler's `LogicalCatalogTransform`
+  rewrites `<alias>.public.t` → `ducklake.main.t` (fed by `clientConn.database`
+  in `newTranspiler`), and `rewriteDirectQuery` expands `USE <alias>` to
+  `ducklake.main`.
+- Opt-in per connection: a session that connects with `ducklake` or nothing is
+  byte-for-byte unaffected.
+- Touching any of this → update `controlplane/configstore/store_test.go`,
+  `controlplane/session_search_path_test.go`, `server/direct_query_rewrite_test.go`,
+  `server/logical_catalog_alias_test.go`, `server/session_database_metadata_test.go`,
+  and `logical_catalog_alias` in `tests/mw-dev/e2e/harness.sh`.
+
 ## TODO Reference
 
 `TODO.md` is a lightweight backlog for ideas that do not yet have a better

cmd/cache-proxy/README.md

@@ -278,7 +278,7 @@ observed peer outcome (`present`, `in_flight`, `negative`, `timeout`,
 span for every probe. `org_id` is intentionally absent — the proxy has no
 per-request tenant identity.
 
-> The cache proxy is not deployed in the `tests/e2e-mw-dev` environment
+> The cache proxy is not deployed in the `tests/mw-dev/e2e` environment
 > (`DUCKGRES_CACHE_ENABLED` is off there). Unit tests in
 > `cmd/cache-proxy/tracing_test.go` cover propagation behavior; validate the
 > complete trace in a cache-enabled dev deployment.

controlplane/admin/README.md

@@ -337,7 +337,7 @@ region.
 `dashboard_test.go` (TokenSet / break-glass login / cookie), `api_test.go` +
 `api_postgres_test.go` (CRUD), `models_api_test.go` (redaction). e2e: the
 `admin_*` / `impersonation_*` / `models_explorer_api` assertions in
-`tests/e2e-mw-dev/harness.sh`.
+`tests/mw-dev/e2e/harness.sh`.
 
 **Frontend** (`ui/`, Vitest + Testing Library — `just ui-test`, CI job
 `ui-tests`): the dashboard's data-derivation logic has shipped wrong more than

controlplane/configstore/store.go

@@ -110,9 +110,10 @@ type Snapshot struct {
 	OrgUserAccess      map[OrgUserKey]OrgUserAccessConfig
 }
 
-// Selectable catalog names. The startup `database` param now names the catalog
-// a session defaults to rather than identifying the org — these are the only
-// non-empty values a client may request.
+// The physical catalog name. The startup `database` param now names the
+// catalog a session defaults to rather than identifying the org; a client may
+// request this, "", or its own org's logical alias (TrinoCatalogName of the
+// org's database_name) and nothing else.
 const (
 	catalogDuckLake = "ducklake"
 )
@@ -133,11 +134,21 @@ type PostgresConnectionResolution struct {
 	SNIAliasUsed bool
 	// SNIResolved is true when the managed hostname resolved to a known org.
 	SNIResolved bool
-	// EffectiveCatalog is the catalog the session should default to, selected by
-	// the startup `database` param: "" (use the attached default) or "ducklake".
+	// EffectiveCatalog is the REAL catalog the session should default to,
+	// selected by the startup `database` param: "" (use the attached default)
+	// or "ducklake". A logical alias still resolves to "ducklake" here —
+	// execution always targets the physical catalog.
 	EffectiveCatalog string
+	// LogicalCatalog is the client-visible name for that same catalog, set only
+	// when the startup `database` matched the SNI-resolved org's own Trino
+	// catalog name (`org_<database_name>`). Empty for "" and "ducklake", which
+	// report the physical name. It renames the catalog on the PG wire
+	// (current_database(), pg_database, information_schema) and in three-part
+	// references; it never changes what the session executes against.
+	LogicalCatalog string
 	// CatalogValid is false when the requested `database` is not a selectable
-	// catalog name (anything other than "" or "ducklake").
+	// catalog name: anything other than "", "ducklake", or the SNI-resolved
+	// org's own catalog name.
 	CatalogValid bool
 	// Valid is true when (OrgID, username, password) authenticated.
 	Valid bool
@@ -565,10 +576,12 @@ func (cs *ConfigStore) ResolvePostgresConnection(startupDatabase, sniPrefix stri
 	result := PostgresConnectionResolution{}
 
 	// The startup `database` param is now pure catalog selection, not identity.
-	// Valid values: "" (use the attached default) or "ducklake". Anything else
-	// fails closed — there is no logical-name masking, so an arbitrary name no
-	// longer routes anywhere.
-	switch strings.ToLower(strings.TrimSpace(startupDatabase)) {
+	// Valid values: "" (use the attached default), "ducklake", or — resolved
+	// further down, once SNI has named an org — that org's own Trino catalog
+	// name. Anything else fails closed: there is no logical-name masking, so an
+	// arbitrary name no longer routes anywhere.
+	requestedCatalog := strings.ToLower(strings.TrimSpace(startupDatabase))
+	switch requestedCatalog {
 	case "":
 		result.CatalogValid = true
 	case catalogDuckLake:
@@ -588,7 +601,7 @@ func (cs *ConfigStore) ResolvePostgresConnection(startupDatabase, sniPrefix stri
 	if !useManagedSNI {
 		return result
 	}
-	orgID, _, aliasUsed := resolveSNIPrefixFromSnapshot(cs.snapshot, sniPrefix)
+	orgID, databaseName, aliasUsed := resolveSNIPrefixFromSnapshot(cs.snapshot, sniPrefix)
 	if orgID == "" {
 		return result
 	}
@@ -597,6 +610,24 @@ func (cs *ConfigStore) ResolvePostgresConnection(startupDatabase, sniPrefix stri
 	result.SNIOrgID = orgID
 	result.OrgID = orgID
 
+	// Logical catalog alias. A session may also name the catalog THIS org
+	// already has on Trino (`org_<database_name>`) and get the same physical
+	// DuckLake catalog under that name, so one engine-agnostic catalog name
+	// works on both engines.
+	//
+	// Direction matters, and it is the whole of PR #651's invariant: the name
+	// is compared against the catalog name derived from the org SNI has ALREADY
+	// resolved. It is never a key into DatabaseOrg, Orgs, or any other map, so
+	// it can neither discover nor select an org — a sibling tenant's catalog
+	// name is just another unrecognized string here, and fails closed. Never
+	// rewrite this as a lookup from name to org.
+	if !result.CatalogValid && databaseName != "" &&
+		requestedCatalog == TrinoCatalogName(databaseName) {
+		result.EffectiveCatalog = catalogDuckLake
+		result.LogicalCatalog = requestedCatalog
+		result.CatalogValid = true
+	}
+
 	// Authenticate the user within the resolved org. Minted service
 	// credentials (svc_-prefixed usernames) resolve against the grants
 	// snapshot map ONLY — the service plane shares no storage with

controlplane/configstore/store_test.go

@@ -830,3 +830,98 @@ func TestWithSnapshotHoldsPublicationReadLock(t *testing.T) {
 		t.Fatal("WithSnapshot did not invoke callback")
 	}
 }
+
+// TestResolvePostgresConnectionLogicalCatalog covers the logical catalog alias:
+// a session may name its org's Trino catalog (`org_<database_name>`) as the
+// startup `database` and get the SAME physical DuckLake catalog under that
+// name. The alias is validated AGAINST the org the managed hostname already
+// resolved — it is never a lookup key, so it cannot route anywhere (PR #651).
+func TestResolvePostgresConnectionLogicalCatalog(t *testing.T) {
+	cs := &ConfigStore{
+		snapshot: &Snapshot{
+			Orgs: map[string]*OrgConfig{
+				"acme":    {Name: "acme", DatabaseName: "acme-analytics"},
+				"billing": {Name: "billing", DatabaseName: "billing_db"},
+			},
+			DatabaseOrg: map[string]string{
+				"acme-analytics": "acme",
+				"billing_db":     "billing",
+			},
+			OrgUserPassword: map[OrgUserKey]string{
+				{OrgID: "acme", Username: "root"}:    mustHash(t, "secret"),
+				{OrgID: "billing", Username: "root"}: mustHash(t, "secret"),
+			},
+		},
+	}
+
+	t.Run("org catalog name selects the physical catalog under the logical name", func(t *testing.T) {
+		got := cs.ResolvePostgresConnection("org_acme_analytics", "acme-analytics", true, "root", "secret")
+		if !got.CatalogValid {
+			t.Fatalf("org catalog name must be selectable: %+v", got)
+		}
+		if got.EffectiveCatalog != "ducklake" {
+			t.Fatalf("EffectiveCatalog = %q, want ducklake (execution stays physical): %+v", got.EffectiveCatalog, got)
+		}
+		if got.LogicalCatalog != "org_acme_analytics" {
+			t.Fatalf("LogicalCatalog = %q, want org_acme_analytics: %+v", got.LogicalCatalog, got)
+		}
+		if !got.Valid || got.OrgID != "acme" {
+			t.Fatalf("unexpected auth result: %+v", got)
+		}
+	})
+
+	t.Run("mixed case and surrounding space normalize to the canonical name", func(t *testing.T) {
+		got := cs.ResolvePostgresConnection("  ORG_Acme_Analytics ", "acme-analytics", true, "root", "secret")
+		if !got.CatalogValid || got.LogicalCatalog != "org_acme_analytics" {
+			t.Fatalf("catalog = (valid=%v, logical=%q), want the canonical lowercase name: %+v",
+				got.CatalogValid, got.LogicalCatalog, got)
+		}
+	})
+
+	t.Run("ducklake and empty carry no logical name", func(t *testing.T) {
+		for _, db := range []string{"", "ducklake"} {
+			got := cs.ResolvePostgresConnection(db, "acme-analytics", true, "root", "secret")
+			if !got.CatalogValid || got.LogicalCatalog != "" {
+				t.Fatalf("database %q: catalog = (valid=%v, logical=%q), want valid with no logical name: %+v",
+					db, got.CatalogValid, got.LogicalCatalog, got)
+			}
+		}
+	})
+
+	t.Run("an arbitrary name still fails closed", func(t *testing.T) {
+		for _, db := range []string{"postgres", "org_", "org_nope", "acme-analytics", "acme"} {
+			got := cs.ResolvePostgresConnection(db, "acme-analytics", true, "root", "secret")
+			if got.CatalogValid || got.LogicalCatalog != "" {
+				t.Fatalf("database %q must fail closed: %+v", db, got)
+			}
+		}
+	})
+
+	t.Run("another org's catalog name is refused", func(t *testing.T) {
+		// The security case: SNI authenticates acme, so billing's catalog name
+		// must not be selectable — not even with acme's valid credentials.
+		got := cs.ResolvePostgresConnection("org_billing_db", "acme-analytics", true, "root", "secret")
+		if got.CatalogValid || got.LogicalCatalog != "" {
+			t.Fatalf("a sibling org's catalog name must fail closed: %+v", got)
+		}
+		if got.OrgID != "acme" {
+			t.Fatalf("OrgID = %q, want acme — identity still comes from SNI alone", got.OrgID)
+		}
+	})
+
+	t.Run("without managed SNI there is no org to validate against", func(t *testing.T) {
+		// No SNI-resolved org means no catalog name to compare to, so the alias
+		// cannot be accepted. The startup database must never resolve an org.
+		got := cs.ResolvePostgresConnection("org_acme_analytics", "acme-analytics", false, "root", "secret")
+		if got.CatalogValid || got.SNIResolved {
+			t.Fatalf("logical alias must not be accepted without managed SNI: %+v", got)
+		}
+	})
+
+	t.Run("unknown managed hostname refuses the alias", func(t *testing.T) {
+		got := cs.ResolvePostgresConnection("org_acme_analytics", "ghostorg", true, "root", "secret")
+		if got.CatalogValid || got.SNIResolved || got.OrgID != "" {
+			t.Fatalf("unknown SNI must not admit a logical alias: %+v", got)
+		}
+	})
+}

controlplane/configstore/trinoname.go

@@ -0,0 +1,48 @@
+package configstore
+
+import (
+	"regexp"
+	"strings"
+)
+
+// trinoCatalogIdentifier is the Trino catalog identifier grammar
+// ([a-z0-9_]+). Anything outside this set in the principal is replaced with
+// `_` before forming the catalog name.
+var trinoCatalogIdentifier = regexp.MustCompile(`[^a-z0-9_]`)
+
+// TrinoSanitize lowercases and replaces non-[a-z0-9_] runs with `_`.
+// Pure function so callers can recover the sanitized name without holding a
+// provisioner.
+func TrinoSanitize(principal string) string {
+	return trinoCatalogIdentifier.ReplaceAllString(strings.ToLower(principal), "_")
+}
+
+// TrinoCatalogName returns the catalog identifier for an org.
+// Format: org_<sanitized>. The sanitization maps the org's TrinoPrincipal
+// (its database_name) to Trino identifier rules ([a-z0-9_]); any other
+// characters collapse to underscores.
+//
+// For principals that satisfy ValidateDatabaseName the mapping is injective
+// — that grammar allows only lowercase alphanumerics and hyphens, so the
+// hyphen is the only character rewritten and no valid principal contains the
+// underscore it becomes — which, with database_name's global unique index,
+// makes distinct orgs' catalog names distinct by construction. Grandfathered
+// rows predate the validation and can still converge; the Trino provisioner's
+// rejectPrincipalCollisions holds those orgs back rather than letting one read
+// the other's catalog.
+//
+// The name carried an `_iceberg` suffix while the backing table format was
+// Iceberg behind Lakekeeper. Warehouses are DuckLake now (migration 000014
+// dropped every iceberg_* column), so the suffix went with it. The shape is
+// pinned from three sides — this function, opa.ManagedCatalogPattern, and
+// the regex literal inside policy.rego — and the pair of tests named in
+// ManagedCatalogPattern's doc comment fails if any one of them moves alone.
+//
+// It lives here, not in the (kubernetes-tagged) Trino provisioner, because
+// ResolvePostgresConnection needs it in every build: the same name is the
+// logical catalog alias a pgwire session may connect with, so SQLMesh and
+// friends see ONE catalog name across the Duckgres and Trino engines.
+// provisioner.TrinoCatalogName delegates here.
+func TrinoCatalogName(principal string) string {
+	return "org_" + TrinoSanitize(principal)
+}