fix: scope cold Iceberg tables and watermark by PG schema

ci/journey.sh

@@ -184,7 +184,6 @@ postgres:
 iceberg:
   warehouse: "${WAREHOUSE}"
   lakekeeper_endpoint: "http://${LK_IP}:8181/catalog"
-  namespace: "default"
 $(storage_yaml)
 archiver:
   tables:
@@ -234,7 +233,6 @@ postgres:
 iceberg:
   warehouse: "${WAREHOUSE}"
   lakekeeper_endpoint: "http://${LK_IP}:8181/catalog"
-  namespace: "default"
 $(storage_yaml)
 archiver:
   tables:
@@ -281,8 +279,8 @@ story_provision_decoupled() {
         sleep 2
     done
     assert_eq "iceonly wrapper view created" "v" "$(q "$HOST" "SELECT relkind FROM pg_class WHERE relname='iceonly' AND relnamespace='public'::regnamespace;")"
-    assert_eq "iceberg-only registry row present" "1" "$(q "$HOST" "SELECT count(*) FROM coldfront.tiered_views WHERE is_iceberg_only AND iceberg_table='ice.default.iceonly';")"
-    assert_eq "no hot table for iceberg-only view" "" "$(q "$HOST" "SELECT hot_table FROM coldfront.tiered_views WHERE iceberg_table='ice.default.iceonly';")"
+    assert_eq "iceberg-only registry row present" "1" "$(q "$HOST" "SELECT count(*) FROM coldfront.tiered_views WHERE is_iceberg_only AND iceberg_table='ice.public.iceonly';")"
+    assert_eq "no hot table for iceberg-only view" "" "$(q "$HOST" "SELECT hot_table FROM coldfront.tiered_views WHERE iceberg_table='ice.public.iceonly';")"
 }
 
 # ───────────────────────────────────────────────────────────────────────────
@@ -478,7 +476,7 @@ EOSQL
     local ret_days=$(( ( $(date -u +%s) - $(date -u -d "$(date -u +%Y-%m-01) -1 month" +%s) ) / 86400 ))
     cat > /tmp/journey-typed.yaml <<EOF
 postgres: { dsn: "host=${DB_IP} port=5432 dbname=coldfront user=coldfront password=coldfront sslmode=disable" }
-iceberg:  { warehouse: "${WAREHOUSE}", lakekeeper_endpoint: "http://${LK_IP}:8181/catalog", namespace: "default" }
+iceberg:  { warehouse: "${WAREHOUSE}", lakekeeper_endpoint: "http://${LK_IP}:8181/catalog" }
 $(storage_yaml)
 archiver: { tables: [ { source_table: typed, partition_period: monthly, hot_period: "${ret_days} days" } ] }
 EOF
@@ -1023,7 +1021,7 @@ EOSQL
     local ret_race=$(( ( $(date -u +%s) - $(date -u -d "$(date -u +%Y-%m-01) +14 days" +%s) ) / 86400 ))
     cat > /tmp/journey-race.yaml <<EOF
 postgres: { dsn: "host=${DB_IP} port=5432 dbname=coldfront user=coldfront password=coldfront sslmode=disable" }
-iceberg:  { warehouse: "${WAREHOUSE}", lakekeeper_endpoint: "http://${LK_IP}:8181/catalog", namespace: "default" }
+iceberg:  { warehouse: "${WAREHOUSE}", lakekeeper_endpoint: "http://${LK_IP}:8181/catalog" }
 $(storage_yaml)
 archiver: { tables: [ { source_table: events, partition_period: monthly, hot_period: "${ret_race} days" } ] }
 EOF
@@ -1143,7 +1141,7 @@ story_mesh() {
     local pc
     for pc in "${PARR[@]}"; do
         q_may "$pc" "SELECT coldfront.create_iceberg_table('public','iceonly','[{\"name\":\"id\",\"type\":\"bigint\"},{\"name\":\"ts\",\"type\":\"timestamptz\"},{\"name\":\"status\",\"type\":\"text\"},{\"name\":\"data\",\"type\":\"jsonb\"}]'::jsonb);" >/dev/null 2>&1
-        assert_eq "iceberg-only registered on peer $pc" "1" "$(q "$pc" "SELECT count(*) FROM coldfront.tiered_views WHERE is_iceberg_only AND iceberg_table='ice.default.iceonly';")"
+        assert_eq "iceberg-only registered on peer $pc" "1" "$(q "$pc" "SELECT count(*) FROM coldfront.tiered_views WHERE is_iceberg_only AND iceberg_table='ice.public.iceonly';")"
     done
 
     # Cross-node READ: every row db1 wrote to Iceberg is visible on each peer via
@@ -1517,7 +1515,7 @@ EOSQL
     local ret_days; ret_days=$(( ( $(date -u +%s) - $(date -u -d "$(date -u +%Y-%m-01) -1 month" +%s) ) / 86400 ))  # cutoff = start of now-1mo
     cat > /tmp/journey-tl.yaml <<EOF
 postgres: { dsn: "host=${DB_IP} port=5432 dbname=coldfront user=coldfront password=coldfront sslmode=disable" }
-iceberg:  { warehouse: "${WAREHOUSE}", lakekeeper_endpoint: "http://${LK_IP}:8181/catalog", namespace: "default" }
+iceberg:  { warehouse: "${WAREHOUSE}", lakekeeper_endpoint: "http://${LK_IP}:8181/catalog" }
 $(storage_yaml)
 archiver:
   tables:
@@ -1650,7 +1648,7 @@ EOSQL
 
     cat > /tmp/journey-fk.yaml <<EOF
 postgres: { dsn: "${dsn}" }
-iceberg:  { warehouse: "${WAREHOUSE}", lakekeeper_endpoint: "http://${LK_IP}:8181/catalog", namespace: "default" }
+iceberg:  { warehouse: "${WAREHOUSE}", lakekeeper_endpoint: "http://${LK_IP}:8181/catalog" }
 $(storage_yaml)
 EOF
     if "$ARCHIVER" --config /tmp/journey-fk.yaml >/tmp/journey-fk-arch.log 2>&1; then
@@ -1774,7 +1772,7 @@ EOSQL
 
     cat > /tmp/journey-rb.yaml <<EOF
 postgres: { dsn: "${dsn}" }
-iceberg:  { warehouse: "${WAREHOUSE}", lakekeeper_endpoint: "http://${LK_IP}:8181/catalog", namespace: "default" }
+iceberg:  { warehouse: "${WAREHOUSE}", lakekeeper_endpoint: "http://${LK_IP}:8181/catalog" }
 $(storage_yaml)
 EOF
 
@@ -2011,7 +2009,7 @@ EOF
     # drive entirely off coldfront.partition_config.
     cat > /tmp/journey-conn.yaml <<EOF
 postgres: { dsn: "host=${DB_IP} port=5432 dbname=coldfront user=coldfront password=coldfront sslmode=disable" }
-iceberg:  { warehouse: "${WAREHOUSE}", lakekeeper_endpoint: "http://${LK_IP}:8181/catalog", namespace: "default" }
+iceberg:  { warehouse: "${WAREHOUSE}", lakekeeper_endpoint: "http://${LK_IP}:8181/catalog" }
 $(storage_yaml)
 EOF
     if "$ARCHIVER" --config /tmp/journey-conn.yaml >/tmp/journey-dbrun.log 2>&1; then

ci/lib.sh

@@ -115,6 +115,6 @@ create_warehouse_and_seed() {
     [ -z "$wid" ] && wid=$(curl -s "http://$lk_ip:8181/management/v1/warehouse" | grep -oE '"warehouse-id":"[^"]+"' | head -1 | cut -d'"' -f4)
     if [ -n "$wid" ]; then
         curl -s -X POST "http://$lk_ip:8181/catalog/v1/$wid/namespaces" \
-            -H "Content-Type: application/json" -d '{"namespace":["default"]}' >/dev/null 2>&1 || true
+            -H "Content-Type: application/json" -d '{"namespace":["public"]}' >/dev/null 2>&1 || true
     fi
 }

ci/probe-standby.sh

@@ -101,7 +101,6 @@ postgres:
 iceberg:
   warehouse: "${WAREHOUSE}"
   lakekeeper_endpoint: "http://${LK_IP}:8181/catalog"
-  namespace: "default"
 s3:
   endpoint: "${SW_IP}:8333"
   region: "us-east-1"

cmd/archiver/main.go

@@ -412,7 +412,7 @@ func runCycle(ctx context.Context, cfg *config.Config, t *config.TableConfig, co
 		cfg: cfg, t: t, conn: conn, wmStore: wmStore,
 		partMgr:  partition.NewManager(conn),
 		viewGen:  view.NewGenerator(conn),
-		iceTable: pgx.Identifier{"ice", cfg.Iceberg.Namespace, t.SourceTable}.Sanitize(),
+		iceTable: icebergRef(t.SourceSchema, t.SourceTable),
 		now:      now, debugExportDelay: debugExportDelay,
 	}
 
@@ -479,7 +479,7 @@ func (ac *archiveCycle) attachAndEnsureTable(ctx context.Context) error {
 	if err := attachIceberg(ctx, ac.conn, ac.cfg); err != nil {
 		return err
 	}
-	if err := ensureIcebergTable(ctx, ac.conn, ac.cfg, ac.t, ac.iceTable); err != nil {
+	if err := ensureIcebergTable(ctx, ac.conn, ac.t, ac.iceTable); err != nil {
 		return fmt.Errorf("ensure iceberg table: %w", err)
 	}
 	return nil
@@ -582,7 +582,7 @@ func (ac *archiveCycle) tierExpiredPartitions(ctx context.Context, hotExpired []
 // (idempotent cleanup, no race); otherwise run the full archive pipeline.
 func (ac *archiveCycle) archiveOnePartition(ctx context.Context, part partition.Info, columns []view.Column) error {
 	t := ac.t
-	wmCutoff, found, err := ac.wmStore.Get(ctx, t.SourceTable)
+	wmCutoff, found, err := ac.wmStore.Get(ctx, t.SourceSchema, t.SourceTable)
 	if err != nil {
 		return fmt.Errorf("get watermark: %w", err)
 	}
@@ -604,7 +604,7 @@ func (ac *archiveCycle) archiveOnePartition(ctx context.Context, part partition.
 // per-partition / per-period loop, never inside it.
 func (ac *archiveCycle) bootstrapTieredView(ctx context.Context, columns []view.Column) error {
 	t, iceTable := ac.t, ac.iceTable
-	wmCutoff, _, err := ac.wmStore.Get(ctx, t.SourceTable)
+	wmCutoff, _, err := ac.wmStore.Get(ctx, t.SourceSchema, t.SourceTable)
 	if err != nil {
 		return fmt.Errorf("get watermark: %w", err)
 	}
@@ -691,7 +691,7 @@ func runCycleTwoLevel(ctx context.Context, cfg *config.Config, t *config.TableCo
 		cfg: cfg, t: t, conn: conn, wmStore: wmStore,
 		partMgr:  partition.NewManager(conn),
 		viewGen:  view.NewGenerator(conn),
-		iceTable: pgx.Identifier{"ice", cfg.Iceberg.Namespace, t.SourceTable}.Sanitize(),
+		iceTable: icebergRef(t.SourceSchema, t.SourceTable),
 		now:      now, debugExportDelay: debugExportDelay,
 	}
 
@@ -1261,11 +1261,18 @@ func bulkExportWithSnapshot(ctx context.Context, conn *pgx.Conn, t *config.Table
 // bulk export is one autocommit DuckDB transaction; if Iceberg rejects it,
 // the whole archive cycle errors out and cron retries.)
 
+// icebergRef is the DuckDB reference for a source table's cold Iceberg table.
+// The PG schema is the Iceberg namespace, so the ref is ice.<schema>.<table>
+// and same-named tables in different PG schemas resolve to distinct tables.
+func icebergRef(schema, table string) string {
+	return pgx.Identifier{"ice", schema, table}.Sanitize()
+}
+
 // ensureIcebergTable creates the Iceberg namespace and table (matching the
 // PG source schema) if they don't already exist. Safe to call every run.
-func ensureIcebergTable(ctx context.Context, conn *pgx.Conn, cfg *config.Config, t *config.TableConfig, iceTable string) error {
+func ensureIcebergTable(ctx context.Context, conn *pgx.Conn, t *config.TableConfig, iceTable string) error {
 	if err := execDuckDB(ctx, conn, fmt.Sprintf("CREATE SCHEMA IF NOT EXISTS %s",
-		pgx.Identifier{"ice", cfg.Iceberg.Namespace}.Sanitize())); err != nil {
+		pgx.Identifier{"ice", t.SourceSchema}.Sanitize())); err != nil {
 		return fmt.Errorf("create namespace: %w", err)
 	}
 

cmd/archiver/main_test.go

@@ -495,3 +495,14 @@ func TestCutoverFailHint(t *testing.T) {
 	assert.Empty(t, cutoverFailHint(&pgconn.PgError{Code: "55P03"}), "no hint for the transient lock timeout")
 	assert.Empty(t, cutoverFailHint(errors.New("boom")), "no hint for non-Postgres errors")
 }
+
+// Two tables sharing a name in different PG schemas map to distinct Iceberg
+// tables, since the PG schema is the Iceberg namespace (ice.<schema>.<table>).
+func TestIcebergRef_SchemaScoped(t *testing.T) {
+	a := icebergRef("myapp", "events")
+	b := icebergRef("analytics", "events")
+	assert.NotEqual(t, a, b, "distinct schemas yield distinct iceberg refs")
+	assert.Contains(t, a, "myapp")
+	assert.Contains(t, a, "events")
+	assert.Contains(t, b, "analytics")
+}

cmd/compactor/config.go

@@ -23,7 +23,6 @@ type Config struct {
 	Iceberg struct {
 		Warehouse          string `yaml:"warehouse"`
 		LakekeeperEndpoint string `yaml:"lakekeeper_endpoint"`
-		Namespace          string `yaml:"namespace"`
 	} `yaml:"iceberg"`
 	S3 struct {
 		Endpoint  string `yaml:"endpoint"`
@@ -48,12 +47,20 @@ func LoadConfig(path string) (*Config, error) {
 	if err := yaml.Unmarshal(data, &c); err != nil {
 		return nil, fmt.Errorf("parse config: %w", err)
 	}
-	if c.Iceberg.Namespace == "" {
-		c.Iceberg.Namespace = "default"
-	}
 	return &c, nil
 }
 
+// splitSchemaTable parses a "schema.table" CLI argument into its parts; a bare
+// name defaults to the "public" schema, matching the archiver's default source
+// schema. The PG schema is the Iceberg namespace, so same-named tables in
+// different schemas resolve to distinct Iceberg tables.
+func splitSchemaTable(arg string) (schema, table string) {
+	if s, t, found := strings.Cut(arg, "."); found {
+		return s, t
+	}
+	return "public", arg
+}
+
 // storageProps builds the iceberg-go fileio credential properties for whichever
 // cold-store backend the deployment configures — exactly one of S3 or Azure,
 // mirroring ColdFront's set_storage_secret. The S3 path serves SeaweedFS/MinIO,

cmd/compactor/config_test.go

@@ -8,6 +8,20 @@ import (
 	"github.com/apache/iceberg-go/utils"
 )
 
+func TestSplitSchemaTable(t *testing.T) {
+	cases := []struct{ arg, schema, table string }{
+		{"events", "public", "events"},
+		{"myapp.events", "myapp", "events"},
+		{"analytics.events", "analytics", "events"},
+	}
+	for _, c := range cases {
+		s, tbl := splitSchemaTable(c.arg)
+		if s != c.schema || tbl != c.table {
+			t.Errorf("splitSchemaTable(%q) = (%q, %q); want (%q, %q)", c.arg, s, tbl, c.schema, c.table)
+		}
+	}
+}
+
 func TestStorageProps_S3Compat(t *testing.T) {
 	// SeaweedFS/MinIO: explicit http endpoint, path-style addressing.
 	c := &Config{}

cmd/compactor/main.go

@@ -43,7 +43,7 @@ type runOpts struct {
 
 func main() {
 	cfgPath := flag.String("config", "", "deployment YAML: DSN + iceberg/S3/azure storage creds")
-	tableName := flag.String("table", "", "table in the configured iceberg namespace (required)")
+	tableName := flag.String("table", "", "[schema.]table to maintain; schema (default public) maps to the Iceberg namespace (required)")
 	targetMB := flag.Int64("target-size-mb", 128, "compaction: target output Parquet file size in MiB")
 	dryRun := flag.Bool("dry-run", false, "plan only — report what would change, change nothing")
 	expire := flag.Bool("expire-snapshots", false, "also expire old snapshots and reclaim the files they alone pinned")
@@ -85,7 +85,6 @@ func run(cfgPath, tableName string, o runOpts) error {
 	if err != nil {
 		return err
 	}
-	ns := cfg.Iceberg.Namespace
 
 	// Keep Accept-Encoding out of SigV4 signing for a TLS S3-compatible cold
 	// store (GCS rejects it); a no-op otherwise. Rides the context into every
@@ -102,9 +101,11 @@ func run(cfgPath, tableName string, o runOpts) error {
 
 	// The bakery claim key MUST be byte-identical to the cold-write path's so every
 	// mutating step mutually-excludes with concurrent cold writers. The archiver and
-	// hook build it as pgx.Identifier{"ice", namespace, table}.Sanitize()
-	// (cmd/archiver/main.go iceTable; coldfront--1.0.sql tiered_views.iceberg_table).
-	icebergRef := pgx.Identifier{"ice", ns, tableName}.Sanitize()
+	// hook build it as pgx.Identifier{"ice", schema, table}.Sanitize() with the PG
+	// schema as the Iceberg namespace (cmd/archiver/main.go icebergRef;
+	// coldfront--1.0.sql tiered_views.iceberg_table).
+	schema, table := splitSchemaTable(tableName)
+	icebergRef := pgx.Identifier{"ice", schema, table}.Sanitize()
 
 	// One PG connection, shared across each step's bakery claim, opened lazily — a
 	// pure dry-run never needs it.
@@ -125,16 +126,16 @@ func run(cfgPath, tableName string, o runOpts) error {
 		return withBakeryClaim(ctx, conn, icebergRef, fn)
 	}
 
-	if err := doCompaction(ctx, cat, ns, tableName, o, claim); err != nil {
+	if err := doCompaction(ctx, cat, schema, table, o, claim); err != nil {
 		return err
 	}
 	if o.expire {
-		if err := doExpire(ctx, cat, ns, tableName, o, claim); err != nil {
+		if err := doExpire(ctx, cat, schema, table, o, claim); err != nil {
 			return err
 		}
 	}
 	if o.orphans {
-		if err := doOrphans(ctx, cat, ns, tableName, o, claim); err != nil {
+		if err := doOrphans(ctx, cat, schema, table, o, claim); err != nil {
 			return err
 		}
 	}

config.example.yaml

@@ -4,7 +4,6 @@ postgres:
 iceberg:
   warehouse: "wh"                                        # Lakekeeper warehouse name
   lakekeeper_endpoint: "http://lakekeeper:8181/catalog"  # Lakekeeper REST catalog
-  namespace: "default"                                   # optional, default: "default"
 
 # Cold-store backend: configure EXACTLY ONE of s3 or azure. Any S3-compatible
 # store works under s3 — SeaweedFS/MinIO, real AWS S3, OR Google Cloud Storage

docs/architecture.md

@@ -232,7 +232,7 @@ DuckDB temp table:
 ```sql
 CREATE TEMP TABLE duck_stage USING duckdb AS
   SELECT * FROM public.p_2026_01;
-SELECT duckdb.raw_query($$INSERT INTO ice.default.events
+SELECT duckdb.raw_query($$INSERT INTO ice.public.events
   SELECT * FROM pg_temp.duck_stage$$);
 DROP TABLE duck_stage;
 ```
@@ -243,7 +243,7 @@ DROP TABLE duck_stage;
 
 ```sql
 SELECT r['id']::bigint, r['ts']::timestamptz, r['status']::text
-FROM iceberg_scan('ice.default.events') r
+FROM iceberg_scan('ice.public.events') r
 WHERE r['ts'] < '2026-03-01'::timestamptz;
 ```
 
@@ -641,7 +641,7 @@ but dwarfed by the Iceberg commit work for any realistic batch.
 straight into the Iceberg writer - e.g. a `COPY` form:
 
 ```sql
-COPY (SELECT * FROM public.events_partition) TO ICEBERG 'ice.default.events';
+COPY (SELECT * FROM public.events_partition) TO ICEBERG 'ice.public.events';
 ```
 
 That would make pg_duckdb the only place in the data path that touches

docs/architecture_decoupled.md

@@ -70,7 +70,7 @@ materializes a DuckDB PERSISTENT SECRET, which DuckDB loads at instance
 init. It is set once; no per-session arming is needed.
 
 After that, the first query touching a tiered view in any session
-lazily attaches the catalog and `ice.default.*` becomes available.
+lazily attaches the catalog and `ice.public.*` becomes available.
 
 ## Interface
 
@@ -100,8 +100,8 @@ The following attempts fail, with the reason for each:
 
 | Attempt | Failure |
 |---|---|
-| `SELECT * FROM ice.default.events` | PG parser rejects: `cross-database references are not implemented`. PG sees the 3-part name as `database.schema.table` and refuses. There is no "ice is an attached duckdb catalog" handling at the PG parser level. |
-| `INSERT INTO ice.default.events VALUES (...)` (PG-native DML on the 3-part name) | Same parser rejection. |
+| `SELECT * FROM ice.public.events` | PG parser rejects: `cross-database references are not implemented`. PG sees the 3-part name as `database.schema.table` and refuses. There is no "ice is an attached duckdb catalog" handling at the PG parser level. |
+| `INSERT INTO ice.public.events VALUES (...)` (PG-native DML on the 3-part name) | Same parser rejection. |
 | Bare-column predicates on `iceberg_scan(...)` | `iceberg_scan` returns a single-column row of struct; columns must be accessed via `r['col']`. Bare `WHERE col = …` fails with "column does not exist". |
 
 The net effect: every read or write of an Iceberg-only table either
@@ -183,16 +183,16 @@ DELETE FROM events WHERE id = 1;
 
 What the helper does:
 
-1. `duckdb.raw_query('CREATE SCHEMA IF NOT EXISTS ice."default"')` -
+1. `duckdb.raw_query('CREATE SCHEMA IF NOT EXISTS ice."public"')` -
    idempotent namespace creation against Lakekeeper.
-2. `duckdb.raw_query('CREATE TABLE ice.default.<name> (col1
+2. `duckdb.raw_query('CREATE TABLE ice.public.<name> (col1
    STORAGE_TYPE, …)')` - column types are validated by
    `coldfront._iceberg_storage_type()`, which mirrors the canonical
    map in `cmd/archiver/main.go pgFormatTypeToDuckDB`. Anything outside
    the supported set (see "Supported column types" above) raises before
    any DDL is issued.
 3. `CREATE OR REPLACE VIEW <schema>.<name> AS SELECT r['col']::pg_type
-   AS col, … FROM duckdb.query('SELECT * FROM ice.default.<name>') AS
+   AS col, … FROM duckdb.query('SELECT * FROM ice.public.<name>') AS
    t(r)` - projection wraps the struct accessor so applications see
    flat columns. View-cast types (`jsonb` → `json`, `interval`) are
    surfaced via the appropriate cast. The view reads via
@@ -205,7 +205,7 @@ What the helper does:
    this flag and short-circuits `classify_tier()` to `TIER_COLD` for
    any INSERT/UPDATE/DELETE on the wrapper view, regardless of WHERE
    clause or watermark - so every write rewrites cleanly into a single
-   `SELECT duckdb.raw_query('INSERT/UPDATE/DELETE ice.default.<name>
+   `SELECT duckdb.raw_query('INSERT/UPDATE/DELETE ice.public.<name>
    …')`. No INSTEAD OF INSERT trigger is created - the hook is the
    dispatch path.