feat: accept an org's Trino catalog name as a logical catalog alias (#1177)

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

A Duckgres session may now connect with `database=org_<database_name>` —
the same catalog name the org has on Trino — and get the physical DuckLake
catalog under that name. current_database(), pg_database,
information_schema, three-part references, and `USE` all answer to it.

The point is SQLMesh: it sees ONE catalog name on both the Duckgres and
Trino engines, so moving a project between engines needs no state rewrite.

PR #651's invariant is preserved. The startup `database` is still never
used to find, select, or route to an org. Identity stays SNI-only, and the
alias is validated AGAINST the org SNI has already resolved — it is
compared to that org's own catalog name, never used as a lookup key. A
sibling tenant's catalog name is just another unrecognized string and
fails closed, exactly as today. Sessions connecting with "ducklake" or
nothing are unchanged; the alias is opt-in per connection.

The name derivation moves to configstore (untagged) so every build has it;
provisioner.TrinoCatalogName now delegates to that one definition.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WCY5Jf2BQPCVKJTZU1TpEe

* test(e2e): assert the logical catalog alias against mw-dev

logical_catalog_alias connects the cnpg tenant with its own Trino catalog
name and asserts the alias renames without redirecting: current_database()
and pg_database report it, a three-part reference and `USE <alias>` reach
the real catalog, and a session connected the ordinary way sees the same
row. It also asserts the security half — a sibling tenant's catalog name
and an arbitrary name both get 3D000.

Docs: CLAUDE.md gains the alias contract and the PR #651 invariant that
governs it; the harness path references are corrected to the directory
that exists (tests/mw-dev/e2e/, not tests/e2e-mw-dev/).

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WCY5Jf2BQPCVKJTZU1TpEe

* refactor: drop two unreachable guards in the alias checks

`requestedCatalog != ""` cannot be false where it sits — an empty startup
database already set CatalogValid — and `c.database != ""` is subsumed by
the EqualFold against a non-empty USE target.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WCY5Jf2BQPCVKJTZU1TpEe

* fix: restore the empty-database guard in namesDuckLakeCatalog

dcb043d dropped `c.database != ""` on the claim that it was "subsumed by
the EqualFold against a non-empty USE target". That claim was wrong: the
USE target is only checked for emptiness BEFORE quote-stripping, so `USE ""`
passes that check as the two-character `""` and then unquotes to "". A
session whose database is unset then matched its own empty name, and invalid
SQL was silently rewritten into `USE ducklake.main` instead of passing
through to DuckDB and erroring. No cross-tenant exposure — it resolves to
the catalog the session already executes against — but an invalid statement
must not silently succeed.

The other guard that commit dropped (`requestedCatalog != ""` in store.go)
is genuinely unreachable and stays dropped.

TestRewriteDirectQueryEmptyQuotedUseIsNotAnAlias pins it at the rewrite
entry point, plus an `USE ""` case in the alias table.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WCY5Jf2BQPCVKJTZU1TpEe

* fix(e2e): send USE as its own message in the logical-alias assertion

The e2e assertion batched `USE <alias>; SELECT ...` into one psql -c, which
duckgres cannot split. handleQuery splits a multi-statement simple query only
when pg_query parses it (conn.go: `parseErr == nil && len(tree.Stmts) > 1`),
and `USE` is not PostgreSQL syntax — so the batch reached rewriteDirectQuery
whole, its USE target was `<alias>; SELECT ...` (matching no catalog name, so
correctly left alone), and DuckDB split it and failed the bare USE.

The product code was right; the assertion was written in a shape the simple
query protocol does not support here. `USE ducklake; SELECT ...` fails the
same way on main, so this is pre-existing and not alias-specific.

pg_script feeds a script on stdin, where psql sends each statement as its own
simple-query message on ONE session. Verified against a local standalone
server: state set by the first statement is visible to the third, on one pid.

Tests: TestLogicalCatalogAliasThroughConnectionSetup drives the alias through
the REAL setup path — NewClientConn plus the exported setters control.go
calls, in order, including the post-worker-switch replay — for both the simple
and the extended query composition. That closes the gap that let this reach
CI: the old tests hand-built a clientConn and never exercised how the session
fields get populated. TestUseStatementIsNeverSplitOutOfASimpleQueryBatch pins
the batching limitation so the next harness author does not rediscover it, and
fails loudly if pg_query ever learns to parse USE.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WCY5Jf2BQPCVKJTZU1TpEe

* docs: record the catalog-qualified SET search_path gap

`SET search_path = '<alias>.main'` is not rewritten and fails on the worker,
while the physical `'ducklake.main'` works. The catalog name sits in a string
literal rather than a RangeVar, so LogicalCatalogTransform has nothing to
match and the USE pass does not look at SET. Deliberately left alone: SQLMesh
selects a catalog with `USE <catalog>` as its own statement, and touches
search_path only in dbt code it marks unsupported.

Also record in the harness that project_reader_isolation already issues its
USE as a separate message (psql -c/-c) and is not a latent failure, and that
the retry case list now has four copies worth folding together later.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WCY5Jf2BQPCVKJTZU1TpEe

---------

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

Author: fuziontech

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