diff --git a/ci/journey.sh b/ci/journey.sh index d64f799..a1e428a 100755 --- a/ci/journey.sh +++ b/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 < /tmp/journey-race.yaml </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 < /tmp/journey-fk.yaml </tmp/journey-fk-arch.log 2>&1; then @@ -1774,7 +1772,7 @@ EOSQL cat > /tmp/journey-rb.yaml < /tmp/journey-conn.yaml </tmp/journey-dbrun.log 2>&1; then diff --git a/ci/lib.sh b/ci/lib.sh index 9dd0558..4408189 100755 --- a/ci/lib.sh +++ b/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 } diff --git a/ci/probe-standby.sh b/ci/probe-standby.sh index 879a1ad..e840436 100755 --- a/ci/probe-standby.sh +++ b/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" diff --git a/cmd/archiver/main.go b/cmd/archiver/main.go index cae8ddb..addbc78 100644 --- a/cmd/archiver/main.go +++ b/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.. +// 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) } diff --git a/cmd/archiver/main_test.go b/cmd/archiver/main_test.go index c3cb503..cc314ac 100644 --- a/cmd/archiver/main_test.go +++ b/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..
). +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") +} diff --git a/cmd/compactor/config.go b/cmd/compactor/config.go index 7ee697c..74bdcd7 100644 --- a/cmd/compactor/config.go +++ b/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, diff --git a/cmd/compactor/config_test.go b/cmd/compactor/config_test.go index f124dc8..d0092b5 100644 --- a/cmd/compactor/config_test.go +++ b/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{} diff --git a/cmd/compactor/main.go b/cmd/compactor/main.go index ad146e2..a9af68c 100644 --- a/cmd/compactor/main.go +++ b/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 } } diff --git a/config.example.yaml b/config.example.yaml index f832f74..63ed30d 100644 --- a/config.example.yaml +++ b/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 diff --git a/docs/architecture.md b/docs/architecture.md index 84f95d2..36942e3 100644 --- a/docs/architecture.md +++ b/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 diff --git a/docs/architecture_decoupled.md b/docs/architecture_decoupled.md index 987d636..21dad82 100644 --- a/docs/architecture_decoupled.md +++ b/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. (col1 +2. `duckdb.raw_query('CREATE TABLE ice.public. (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 . AS SELECT r['col']::pg_type - AS col, … FROM duckdb.query('SELECT * FROM ice.default.') AS + AS col, … FROM duckdb.query('SELECT * FROM ice.public.') 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. + `SELECT duckdb.raw_query('INSERT/UPDATE/DELETE ice.public. …')`. No INSTEAD OF INSERT trigger is created - the hook is the dispatch path. diff --git a/docs/architecture_tiered.md b/docs/architecture_tiered.md index 0ace624..0fa03b9 100644 --- a/docs/architecture_tiered.md +++ b/docs/architecture_tiered.md @@ -112,7 +112,7 @@ CREATE OR REPLACE VIEW events AS WHERE "ts" >= '2026-03-01'::timestamptz UNION ALL SELECT r['id']::bigint, r['ts']::timestamptz, r['status']::text, r['data']::text - FROM iceberg_scan('ice.default.events') r + FROM iceberg_scan('ice.public.events') r WHERE r['ts'] < '2026-03-01'::timestamptz; ``` @@ -211,7 +211,7 @@ WITH hot_ins AS MATERIALIZED ( ), cold_call AS MATERIALIZED ( SELECT duckdb.raw_query( - 'INSERT INTO ice.default.events + 'INSERT INTO ice.public.events SELECT id, ts, status, data FROM () ... WHERE ts < ''''' ) @@ -246,7 +246,7 @@ tier and rewrite: | Predicate shape | Tier | Rewrite | |---|---|---| | WHERE proves all matching rows have `ts >= cutoff` (equality, `>=`, `>`, BETWEEN, IN, OR all in hot range) | HOT | `UPDATE _events SET ... WHERE ...` - plain PG DML, preserves RETURNING | -| WHERE proves all matching rows have `ts < cutoff` | COLD | `SELECT duckdb.raw_query('UPDATE ice.default.events SET ... WHERE ...')` - DuckDB DML wrapped as a standard SQL literal (via `quote_literal_cstr`); the SELECT envelope keeps it off PG's command-ID counter so there's no mixed-write tripwire | +| WHERE proves all matching rows have `ts < cutoff` | COLD | `SELECT duckdb.raw_query('UPDATE ice.public.events SET ... WHERE ...')` - DuckDB DML wrapped as a standard SQL literal (via `quote_literal_cstr`); the SELECT envelope keeps it off PG's command-ID counter so there's no mixed-write tripwire | | WHERE cannot be proven to target one tier | AMBIGUOUS | depends on `coldfront.allow_mixed_writes` - see next section | The classifier understands `Var Const` (both operand orders), @@ -290,7 +290,7 @@ from the `coldfront.allow_mixed_writes` GUC (USERSET, default `on`). ```sql WITH hot AS (UPDATE _events SET ... WHERE ... RETURNING *) - , cold AS (SELECT duckdb.raw_query('UPDATE ice.default.events SET ... WHERE ...')) + , cold AS (SELECT duckdb.raw_query('UPDATE ice.public.events SET ... WHERE ...')) SELECT h.* FROM hot h CROSS JOIN cold c; ``` diff --git a/docs/compaction.md b/docs/compaction.md index d2fced2..70e69ad 100644 --- a/docs/compaction.md +++ b/docs/compaction.md @@ -49,7 +49,7 @@ compactor --config deploy.yaml --table events --expire-snapshots --orphans The config is the same deployment YAML the archiver reads - `postgres.dsn` (for the bakery claim), -`iceberg.{warehouse, lakekeeper_endpoint, namespace}`, and exactly one +`iceberg.{warehouse, lakekeeper_endpoint}`, and exactly one cold-store stanza. ## Backends diff --git a/docs/installation.md b/docs/installation.md index a767d72..2af7f6c 100644 --- a/docs/installation.md +++ b/docs/installation.md @@ -112,7 +112,7 @@ docker compose --profile local-store up -d --build Bootstrap Lakekeeper, create the `wh` warehouse against the SeaweedFS credentials in [`docker/seaweedfs-s3.json`](https://github.com/pgEdge/ColdFront/blob/main/docker/seaweedfs-s3.json), -and seed the `default` namespace: +and seed the `public` namespace: ```bash curl -sf -X POST http://localhost:8181/management/v1/bootstrap \ @@ -131,7 +131,7 @@ curl -s -X POST http://localhost:8181/management/v1/warehouse \ WID=$(curl -s http://localhost:8181/management/v1/warehouse \ | grep -oE '"warehouse-id":"[^"]+"' | head -1 | cut -d'"' -f4) curl -s -X POST "http://localhost:8181/catalog/v1/$WID/namespaces" \ - -H 'Content-Type: application/json' -d '{"namespace":["default"]}' + -H 'Content-Type: application/json' -d '{"namespace":["public"]}' ``` Create the extensions, set the cold-store secret, create a decoupled diff --git a/docs/object_store.md b/docs/object_store.md index 68a6415..7c1b66f 100644 --- a/docs/object_store.md +++ b/docs/object_store.md @@ -200,9 +200,9 @@ in the warehouse and in the role's trust condition must match. On the database side, replace the `set_storage_secret(...)` call in Section 4 with `SELECT coldfront.set_storage_secret_vended();`. -### 3c. Pre-create the `default` namespace +### 3c. Pre-create the `public` namespace -Resolve the warehouse id, then create the `default` namespace under it: +Resolve the warehouse id, then create the `public` namespace under it: ```bash WID=$(curl -s http://localhost:8181/management/v1/warehouse \ @@ -211,14 +211,14 @@ WID=$(curl -s http://localhost:8181/management/v1/warehouse \ curl -X POST "http://localhost:8181/catalog/v1/$WID/namespaces" \ -H "Content-Type: application/json" \ - -d '{"namespace": ["default"]}' + -d '{"namespace":["public"]}' ``` > **Why this step is required (decoupled mode).** > `coldfront.create_iceberg_table()` (Section 5) runs `CREATE SCHEMA` > and `CREATE TABLE` in one transaction. The schema create is deferred > to COMMIT but the table create is POSTed eagerly, so against a -> namespace-less warehouse it 404s. Pre-creating `default` makes the +> namespace-less warehouse it 404s. Pre-creating `public` makes the > in-transaction `CREATE SCHEMA IF NOT EXISTS` a no-op. (Tiered mode's > archiver creates the namespace itself, so this is only needed for the > decoupled demo below.) @@ -332,6 +332,6 @@ Work through this checklist if something failed: 2. **`set_storage_secret(..., NULL, 'eu-west-1')`** - 3rd arg `NULL` (native vhost+HTTPS), 4th arg your real region? A non-NULL endpoint forces path-style and breaks modern Regions (HTTP 400). -3. **Namespace `default` pre-created** in Lakekeeper before +3. **Namespace `public` pre-created** in Lakekeeper before `create_iceberg_table`? Without it the decoupled create 404s. 4. **Long-term key** - not an SSO / temporary session-token credential. diff --git a/docs/usage.md b/docs/usage.md index c2191bd..b2262cb 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -68,7 +68,7 @@ curl -X POST http://localhost:8181/management/v1/warehouse \ WID=$(curl -s http://localhost:8181/management/v1/warehouse \ | grep -oE '"warehouse-id":"[^"]+"' | head -1 | cut -d'"' -f4) curl -X POST "http://localhost:8181/catalog/v1/$WID/namespaces" \ - -H "Content-Type: application/json" -d '{"namespace": ["default"]}' + -H "Content-Type: application/json" -d '{"namespace":["public"]}' ``` Then install the extensions and set the cold-tier credentials, once per @@ -198,11 +198,11 @@ SELECT coldfront.create_iceberg_table( That single statement provisions: -- `ice.default.events` on the attached Iceberg catalog +- `ice.public.events` on the attached Iceberg catalog - a PG-side wrapper view `public.events` with proper PG-typed columns - a `coldfront.tiered_views` registry row - every INSERT, UPDATE, and DELETE on the view is intercepted by the coldfront C hook and rewritten - to a single `duckdb.raw_query(...)` against `ice.default.events` + to a single `duckdb.raw_query(...)` against `ice.public.events` Spock's `ddl_sql` repset replicates the `CREATE VIEW`, but the registry row does not propagate with it: run `create_iceberg_table()` (idempotent, @@ -630,7 +630,7 @@ Keep the following caveats in mind when running either mode: The throughput ceiling is Lakekeeper's commit rate, not the writer count. - **Direct table access**: `_events` is the hot heap (tiered mode only). - `ice.default.` is the Iceberg table - only addressable via + `ice.public.` is the Iceberg table - only addressable via `iceberg_scan(...)` or `duckdb.raw_query('… ice.… …')`, never via PG-native 3-part names. - **Tiered INSERT with omitted IDENTITY column** (e.g. `INSERT INTO diff --git a/docs/walkthrough.md b/docs/walkthrough.md index 86ea353..70bf1f3 100644 --- a/docs/walkthrough.md +++ b/docs/walkthrough.md @@ -79,7 +79,7 @@ If port 5432, 8181, or 8333 is already in use on your host, set (and if you remap the PG port, match it in the `psql` commands below). Bootstrap Lakekeeper, create the `wh` warehouse backed by SeaweedFS, -and seed the `default` namespace. The warehouse POST retries until +and seed the `public` namespace. The warehouse POST retries until SeaweedFS is ready, and the namespace creation is idempotent: ```bash @@ -111,14 +111,14 @@ curl -sf -X POST http://localhost:8181/management/v1/warehouse \ } }' -# Seed the default namespace +# Seed the public namespace WID=$(curl -s http://localhost:8181/management/v1/warehouse \ | grep -oE '"warehouse-id":"[^"]+"' \ | head -1 | cut -d'"' -f4) curl -sf -X POST \ "http://localhost:8181/catalog/v1/${WID}/namespaces" \ -H 'Content-Type: application/json' \ - -d '{"namespace":["default"]}' + -d '{"namespace":["public"]}' ``` ### Using a cloud object store @@ -369,7 +369,6 @@ postgres: iceberg: warehouse: "wh" lakekeeper_endpoint: "http://lakekeeper:8181/catalog" - namespace: "default" s3: endpoint: "seaweedfs:8333" region: "us-east-1" @@ -429,7 +428,7 @@ WH_ID=$(curl -s http://localhost:8181/management/v1/warehouse \ | head -1 | cut -d'"' -f4) META_LOC=$(curl -s \ - "http://localhost:8181/catalog/v1/${WH_ID}/namespaces/default/tables/events" \ + "http://localhost:8181/catalog/v1/${WH_ID}/namespaces/public/tables/events" \ -H 'accept: application/json' \ | grep -o '"metadata-location":"[^"]*"' \ | head -1 | cut -d'"' -f4) @@ -683,7 +682,7 @@ modes are independent. ### Create an Iceberg-only table One SQL call provisions the Iceberg table and the PostgreSQL view. The -`default` namespace was seeded during setup, and the call wraps both +`public` namespace was seeded during setup, and the call wraps both steps in a single transaction. A short retry loop guards against a timing edge where the warehouse is still warming up: diff --git a/examples/walkthrough/config/archiver.yaml b/examples/walkthrough/config/archiver.yaml index cc8f7f3..b924f48 100644 --- a/examples/walkthrough/config/archiver.yaml +++ b/examples/walkthrough/config/archiver.yaml @@ -9,7 +9,6 @@ postgres: iceberg: warehouse: "wh" lakekeeper_endpoint: "http://lakekeeper:8181/catalog" - namespace: "default" s3: endpoint: "seaweedfs:8333" region: "us-east-1" diff --git a/extension/coldfront/Makefile b/extension/coldfront/Makefile index 6167c82..da935e1 100644 --- a/extension/coldfront/Makefile +++ b/extension/coldfront/Makefile @@ -23,7 +23,8 @@ REGRESS = load_order update_unregistered_view update_heap_table \ ddl_partition_passthrough bakery_wraps_cold_writes \ param_cold_via_plpgsql async_requires_patch \ storage_secret_azure storage_secret_vended privilege_model \ - partition_config_interval self_join_rejected returning_cold_rejected + partition_config_interval self_join_rejected returning_cold_rejected \ + schema_collision REGRESS_OPTS = --inputdir=test --outputdir=test PG_CONFIG ?= pg_config diff --git a/extension/coldfront/coldfront--1.0.sql b/extension/coldfront/coldfront--1.0.sql index 00f0e83..6c7dba7 100644 --- a/extension/coldfront/coldfront--1.0.sql +++ b/extension/coldfront/coldfront--1.0.sql @@ -21,27 +21,29 @@ CREATE TABLE coldfront.tiered_views ( schema_name text NOT NULL, -- namespace of the transparent view relname text NOT NULL, -- name of the transparent view hot_table text, -- 'public._events' (tiered) or NULL (iceberg-only) - iceberg_table text NOT NULL, -- DuckDB ref: 'ice.default.events' + iceberg_table text NOT NULL, -- DuckDB ref, e.g. 'ice.myapp.events' partition_col text, -- 'ts' (tiered) or NULL (iceberg-only) is_iceberg_only boolean NOT NULL DEFAULT false, PRIMARY KEY (schema_name, relname) ); --- Archive watermark: one row per managed table, recording the cutoff time +-- Archive watermark: one row per managed (schema, table), recording the cutoff time -- that divides the hot tier (rows with partition_col >= cutoff_time, living -- in the PG heap as _
) from the cold tier (rows older than cutoff, --- living in Iceberg as ice.default.
). Written by the archiver at the +-- living in Iceberg as ice..
). Written by the archiver at the -- end of each archive cycle; read by the generated tiered view's hot/cold -- UNION (internal/view/view.go) and by the coldfront rewriter's tier -- classifier when deciding whether a predicate proves hot-only or cold-only. -- --- PK on table_name because the archiver upserts this row every cycle via --- ON CONFLICT (table_name) DO UPDATE (internal/watermark/watermark.go:Set). +-- Composite PK on (schema_name, table_name) because the archiver upserts this row every cycle via +-- ON CONFLICT (schema_name, table_name) DO UPDATE (internal/watermark/watermark.go:Set). -- Created with IF NOT EXISTS because the archiver can also materialize it -- on first run before CREATE EXTENSION runs against the DB. CREATE TABLE IF NOT EXISTS coldfront.archive_watermark ( - table_name text PRIMARY KEY, - cutoff_time timestamptz NOT NULL + schema_name text NOT NULL, + table_name text NOT NULL, + cutoff_time timestamptz NOT NULL, + PRIMARY KEY (schema_name, table_name) ); -- coldfront._dummy_dml_target — a permanent, single-row DUMMY table whose data is @@ -790,7 +792,7 @@ BEGIN COALESCE(aw.cutoff_time, '-infinity'::timestamptz) INTO v_hot_table, v_iceberg, v_partcol, v_cutoff FROM coldfront.tiered_views tv - LEFT JOIN coldfront.archive_watermark aw ON aw.table_name = p_view_name + LEFT JOIN coldfront.archive_watermark aw ON aw.schema_name = p_view_schema AND aw.table_name = p_view_name WHERE tv.schema_name = p_view_schema AND tv.relname = p_view_name; IF v_hot_table IS NULL THEN @@ -1008,7 +1010,7 @@ BEGIN SELECT tv.hot_table, tv.iceberg_table, tv.partition_col, aw.cutoff_time INTO v_hot_table, v_iceberg, v_partcol, v_cutoff FROM coldfront.tiered_views tv - LEFT JOIN coldfront.archive_watermark aw ON aw.table_name = p_view_name + LEFT JOIN coldfront.archive_watermark aw ON aw.schema_name = p_view_schema AND aw.table_name = p_view_name WHERE tv.schema_name = p_view_schema AND tv.relname = p_view_name; IF v_hot_table IS NULL OR v_cutoff IS NULL THEN RAISE EXCEPTION 'coldfront: cross-tier move on %.% requires a tiered view with a cutoff', @@ -1417,10 +1419,10 @@ BEGIN UPDATE coldfront.archive_watermark SET cutoff_time = p_new_cutoff - WHERE table_name = p_source; + WHERE schema_name = p_schema AND table_name = p_source; IF NOT FOUND THEN - INSERT INTO coldfront.archive_watermark (table_name, cutoff_time) - VALUES (p_source, p_new_cutoff); + INSERT INTO coldfront.archive_watermark (schema_name, table_name, cutoff_time) + VALUES (p_schema, p_source, p_new_cutoff); END IF; -- GLOBAL LOCK ORDER, step 1 — take the bakery (resource B) BEFORE any @@ -1572,7 +1574,7 @@ $$; -- -- p_schema PG schema for the wrapper view (e.g. 'public'). -- p_table relation/view name (e.g. 'events'). Iceberg table is --- created at ice.default.. +-- created at ice... -- p_columns jsonb array of {name, type} entries. Type is a PG type -- name from the supported set; see _iceberg_storage_type. -- p_partition_cols array of column names for Iceberg partitioning, or NULL. @@ -1592,7 +1594,7 @@ CREATE OR REPLACE FUNCTION coldfront.create_iceberg_table( ) RETURNS void LANGUAGE plpgsql AS $$ DECLARE - ice_ref text := format('ice."default".%I', p_table); + ice_ref text := format('ice.%I.%I', p_schema, p_table); iceberg_cols text := ''; view_proj text := ''; placeholders text := ''; @@ -1679,7 +1681,7 @@ BEGIN -- against an existing table — useful for distributed setups where each -- node registers the same shared Iceberg table independently. PERFORM coldfront.ensure_attached(); - PERFORM duckdb.raw_query('CREATE SCHEMA IF NOT EXISTS ice."default"'); + PERFORM duckdb.raw_query(format('CREATE SCHEMA IF NOT EXISTS ice.%I', p_schema)); PERFORM duckdb.raw_query(format( 'CREATE TABLE IF NOT EXISTS %s (%s)', ice_ref, iceberg_cols @@ -1699,7 +1701,7 @@ BEGIN EXECUTE format( 'CREATE OR REPLACE VIEW %I.%I AS SELECT %s FROM duckdb.query(%L) AS t(r)', p_schema, p_table, view_proj, - format('SELECT * FROM ice.default.%s', p_table) + format('SELECT * FROM ice.%I.%I', p_schema, p_table) ); -- 3. The C post_parse_analyze hook intercepts INSERT INTO the iceberg-only @@ -1715,7 +1717,7 @@ BEGIN -- 4. Registry row — is_iceberg_only=true tells the C hook to short-circuit -- classify_tier to TIER_COLD for any INSERT/UPDATE/DELETE on this view. INSERT INTO coldfront.tiered_views (schema_name, relname, hot_table, iceberg_table, partition_col, is_iceberg_only) - VALUES (p_schema, p_table, NULL, format('ice.default.%s', p_table), NULL, true) + VALUES (p_schema, p_table, NULL, ice_ref, NULL, true) ON CONFLICT (schema_name, relname) DO UPDATE SET hot_table = NULL, iceberg_table = EXCLUDED.iceberg_table, @@ -2475,7 +2477,7 @@ $$; -- Migrate the name-keyed registry + watermark rows when the transparent VIEW is -- renamed. coldfront.tiered_views (keyed on schema+relname) and archive_watermark --- (keyed on the bare view name == archiver SourceTable) both follow the new name. +-- (keyed on schema + bare view name == archiver SourceTable) both follow the new name. -- _rebuild_tiered_view + the regenerated INSERT trigger look the registry/cutoff -- up by the NEW name; without this migration the rebuild would not find the row, -- v_has_cutoff would be false, and the rebuilt view would drop its cold (Iceberg) @@ -2490,7 +2492,7 @@ CREATE FUNCTION coldfront._rename_tiered_view( WHERE schema_name = p_schema AND relname = p_old_view_name; UPDATE coldfront.archive_watermark SET table_name = p_new_view_name - WHERE table_name = p_old_view_name; + WHERE schema_name = p_schema AND table_name = p_old_view_name; $$; -- coldfront._rebuild_tiered_view: regenerate the transparent UNION-ALL view @@ -2522,9 +2524,9 @@ LANGUAGE plpgsql SET client_min_messages = warning AS $$ DECLARE v_schema text; - v_view_name text; -- bare relname == archiver SourceTable == watermark key + v_view_name text; -- bare relname == archiver SourceTable == watermark table_name v_hot_table text; -- stored quoted, e.g. "public"."_events" - v_iceberg text; -- DuckDB ref, e.g. ice.default.events + v_iceberg text; -- DuckDB ref, e.g. ice.myapp.events v_partcol text; v_is_ice_only boolean; v_hot_schema text; @@ -2577,10 +2579,10 @@ BEGIN v_hot_schema := (parse_ident(v_hot_table))[1]; v_hot_relname := (parse_ident(v_hot_table))[2]; - -- 2. Watermark cutoff, keyed on the BARE view name (== SourceTable). + -- 2. Watermark cutoff, keyed on (schema, bare view name == SourceTable). SELECT cutoff_time INTO v_cutoff FROM coldfront.archive_watermark - WHERE table_name = v_view_name; + WHERE schema_name = v_schema AND table_name = v_view_name; v_has_cutoff := (v_cutoff IS NOT NULL); IF v_has_cutoff THEN -- UTC text literal, matching internal/view/view.go cutoffLiteral(). @@ -2712,7 +2714,7 @@ $fn$CREATE OR REPLACE FUNCTION %s() RETURNS trigger AS $body$ DECLARE cutoff timestamptz; BEGIN - SELECT cutoff_time INTO cutoff FROM coldfront.archive_watermark WHERE table_name = %L; + SELECT cutoff_time INTO cutoff FROM coldfront.archive_watermark WHERE schema_name = %L AND table_name = %L; IF cutoff IS NULL THEN cutoff := %s; END IF; @@ -2733,7 +2735,7 @@ BEGIN END; $body$ LANGUAGE plpgsql$fn$, v_funcname, - v_view_name, -- watermark key literal + v_schema, v_view_name, -- watermark key literals (schema, name) CASE WHEN v_has_cutoff THEN quote_literal(v_cutoff_lit) || '::timestamptz' ELSE '''-infinity''::timestamptz' END, -- default cutoff @@ -2854,7 +2856,7 @@ $$; -- these run cleanly at CREATE EXTENSION. COMMENT ON SCHEMA coldfront IS 'pgEdge ColdFront: transparent PostgreSQL to Apache Iceberg tiering, plus decoupled iceberg-only tables.'; COMMENT ON TABLE coldfront.tiered_views IS 'Registry (keyed by schema, relname) of views the coldfront DML hook handles — tiered (hot+cold) and decoupled (iceberg-only).'; -COMMENT ON TABLE coldfront.archive_watermark IS 'Per-tiered-table hot/cold cutoff: ts >= cutoff is hot (PG), ts < cutoff is cold (Iceberg).'; +COMMENT ON TABLE coldfront.archive_watermark IS 'Per-tiered-table (schema, table) hot/cold cutoff: ts >= cutoff is hot (PG), ts < cutoff is cold (Iceberg).'; COMMENT ON TABLE coldfront.storage_secret IS 'Cold-store credential; materialized as a DuckDB PERSISTENT SECRET, replicated by value across a Spock mesh, excluded from pg_dump. A vended row stores no credential and materializes nothing: Lakekeeper mints per-table creds and ensure_attached uses ACCESS_DELEGATION_MODE VENDED_CREDENTIALS.'; COMMENT ON TABLE coldfront.partition_config IS 'Name-keyed per-table partition/tiering lifecycle config (period, hot_period, retention); replicates by value so every mesh node reads identical config.'; COMMENT ON TABLE coldfront.claims IS 'Ricart-Agrawala bakery: a writer''s outstanding iceberg-commit claim (iceberg_table, snowflake ticket); deleted on release.'; diff --git a/extension/coldfront/test/expected/allow_mixed_writes.out b/extension/coldfront/test/expected/allow_mixed_writes.out index 5f934ca..f077f1a 100644 --- a/extension/coldfront/test/expected/allow_mixed_writes.out +++ b/extension/coldfront/test/expected/allow_mixed_writes.out @@ -13,8 +13,8 @@ CREATE TABLE public._events (id int, ts timestamptz, status text); CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- Default GUC value is permissive (on) SHOW coldfront.allow_mixed_writes; coldfront.allow_mixed_writes diff --git a/extension/coldfront/test/expected/bakery_wraps_cold_writes.out b/extension/coldfront/test/expected/bakery_wraps_cold_writes.out index dd42cb2..46edb0e 100644 --- a/extension/coldfront/test/expected/bakery_wraps_cold_writes.out +++ b/extension/coldfront/test/expected/bakery_wraps_cold_writes.out @@ -15,8 +15,8 @@ CREATE TABLE public._events (id int, ts timestamptz, status text); CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- (1) Tiered cold UPDATE (ts < cutoff): plan must call -- coldfront._exec_iceberg_with_claim, NOT bare duckdb.raw_query. EXPLAIN (COSTS OFF, VERBOSE) diff --git a/extension/coldfront/test/expected/cast_in_dquoted_identifier.out b/extension/coldfront/test/expected/cast_in_dquoted_identifier.out index b39fd2c..dda9dbb 100644 --- a/extension/coldfront/test/expected/cast_in_dquoted_identifier.out +++ b/extension/coldfront/test/expected/cast_in_dquoted_identifier.out @@ -17,8 +17,8 @@ CREATE TABLE public._events ( CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- Cold path: SET targets the column whose name contains the cast spelling. -- The column name must survive verbatim in the raw_query argument; only the -- real ts cast should be normalised. diff --git a/extension/coldfront/test/expected/cast_literal_in_value.out b/extension/coldfront/test/expected/cast_literal_in_value.out index e18b8c6..4044be5 100644 --- a/extension/coldfront/test/expected/cast_literal_in_value.out +++ b/extension/coldfront/test/expected/cast_literal_in_value.out @@ -19,8 +19,8 @@ CREATE TABLE public._events (id int, ts timestamptz, status text); CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- Cold path. SET value contains an embedded "::timestamp with time zone" inside -- a text literal; that must stay intact in the raw_query argument. The real -- timestamptz cast in the WHERE clause must still be normalised to ::timestamptz. diff --git a/extension/coldfront/test/expected/cast_normalize.out b/extension/coldfront/test/expected/cast_normalize.out index eb7a37a..da37aec 100644 --- a/extension/coldfront/test/expected/cast_normalize.out +++ b/extension/coldfront/test/expected/cast_normalize.out @@ -21,8 +21,8 @@ CREATE TABLE public._events (id int, ts timestamptz, c_tsn timestamp, CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- (A) Cold UPDATE (ts < cutoff): the deparsed cold SQL must carry DuckDB spellings. -- Casts: timestamp without time zone → timestamp, character varying → varchar, -- double precision → double; the timestamptz bound stays timestamptz. jsonb → json diff --git a/extension/coldfront/test/expected/classify_between_in_or.out b/extension/coldfront/test/expected/classify_between_in_or.out index 48fc50e..dc75a17 100644 --- a/extension/coldfront/test/expected/classify_between_in_or.out +++ b/extension/coldfront/test/expected/classify_between_in_or.out @@ -21,8 +21,8 @@ CREATE TABLE public._events (id int, ts timestamptz, status text); CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- BETWEEN entirely below the cutoff → cold EXPLAIN (COSTS OFF) UPDATE public.events SET status = 'x' diff --git a/extension/coldfront/test/expected/classify_now.out b/extension/coldfront/test/expected/classify_now.out index a745bec..bf9a031 100644 --- a/extension/coldfront/test/expected/classify_now.out +++ b/extension/coldfront/test/expected/classify_now.out @@ -15,8 +15,8 @@ CREATE TABLE public._events (id int, ts timestamptz, status text); CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- Stable, far in the past (< cutoff) → COLD: rewritten to a single cold _exec, -- NOT a dual write. The original now()-expression is preserved for execution. EXPLAIN (COSTS OFF, VERBOSE) diff --git a/extension/coldfront/test/expected/cte_on_dml.out b/extension/coldfront/test/expected/cte_on_dml.out index e70d89e..f1993ac 100644 --- a/extension/coldfront/test/expected/cte_on_dml.out +++ b/extension/coldfront/test/expected/cte_on_dml.out @@ -14,8 +14,8 @@ CREATE TABLE public._events (id int, ts timestamptz, status text); CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- Cold UPDATE (ts < cutoff) with a CTE referenced in SET: the WITH must be -- carried into the cold DML with the relation swapped to ice.default.events. EXPLAIN (COSTS OFF, VERBOSE) diff --git a/extension/coldfront/test/expected/cte_on_insert.out b/extension/coldfront/test/expected/cte_on_insert.out index 9e6a576..2b1b269 100644 --- a/extension/coldfront/test/expected/cte_on_insert.out +++ b/extension/coldfront/test/expected/cte_on_insert.out @@ -15,8 +15,8 @@ CREATE TABLE public._events (id int, ts timestamptz, status text); CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- INSERT ... SELECT FROM a CTE into the tiered view: the WITH must reach both the -- hot (PG) and cold (DuckDB) halves, folded into the source derived table. EXPLAIN (COSTS OFF, VERBOSE) diff --git a/extension/coldfront/test/expected/cte_on_insert_1.out b/extension/coldfront/test/expected/cte_on_insert_1.out index a358bf3..651b710 100644 --- a/extension/coldfront/test/expected/cte_on_insert_1.out +++ b/extension/coldfront/test/expected/cte_on_insert_1.out @@ -15,8 +15,8 @@ CREATE TABLE public._events (id int, ts timestamptz, status text); CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- INSERT ... SELECT FROM a CTE into the tiered view: the WITH must reach both the -- hot (PG) and cold (DuckDB) halves, folded into the source derived table. EXPLAIN (COSTS OFF, VERBOSE) diff --git a/extension/coldfront/test/expected/ddl_rename_view.out b/extension/coldfront/test/expected/ddl_rename_view.out index b463cd3..8c46c1d 100644 --- a/extension/coldfront/test/expected/ddl_rename_view.out +++ b/extension/coldfront/test/expected/ddl_rename_view.out @@ -1,5 +1,5 @@ -- RENAME of the transparent VIEW must migrate every name-keyed reference: --- * coldfront.archive_watermark.table_name (keyed on the bare view name) +-- * coldfront.archive_watermark.table_name (the bare view name in its (schema, table) key) -- * the regenerated INSERT trigger function/trigger names -- Without the watermark migration the rebuilt view would silently lose its -- cold (Iceberg) UNION branch — the watermark lookup would miss, v_has_cutoff @@ -18,8 +18,8 @@ CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -- A watermark exists, so the rebuilt view must keep its cold UNION branch. -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- Before: view definition has the cold iceberg_scan branch. SELECT pg_get_viewdef('public.events'::regclass) LIKE '%iceberg_scan%' AS has_cold_branch; has_cold_branch @@ -29,7 +29,7 @@ SELECT pg_get_viewdef('public.events'::regclass) LIKE '%iceberg_scan%' AS has_co -- Rename the view. ALTER VIEW public.events RENAME TO events_v2; --- The watermark row followed the rename (keyed on the new bare view name). +-- The watermark row followed the rename (its table_name is the new bare view name). SELECT table_name FROM coldfront.archive_watermark ORDER BY table_name; table_name ------------ diff --git a/extension/coldfront/test/expected/dollar_quote_in_value.out b/extension/coldfront/test/expected/dollar_quote_in_value.out index fb0fe99..3d2b50d 100644 --- a/extension/coldfront/test/expected/dollar_quote_in_value.out +++ b/extension/coldfront/test/expected/dollar_quote_in_value.out @@ -17,8 +17,8 @@ CREATE TABLE public._events (id int, ts timestamptz, status text); CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- Cold path: SET value contains the exact dollar-quote tag the old code used. -- Without the fix, pg_parse_query would error on the rewritten SQL. EXPLAIN (COSTS OFF, VERBOSE) diff --git a/extension/coldfront/test/expected/mixed_case_identifier.out b/extension/coldfront/test/expected/mixed_case_identifier.out index 69b8362..d6df6fe 100644 --- a/extension/coldfront/test/expected/mixed_case_identifier.out +++ b/extension/coldfront/test/expected/mixed_case_identifier.out @@ -20,8 +20,8 @@ VALUES ('public', 'MixedEvents', 'public."_MixedEvents"', 'ice.default."MixedEvents"', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('MixedEvents', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'MixedEvents', '2026-03-01'::timestamptz); -- Cold UPDATE through the mixed-case view: must rewrite, not error. EXPLAIN (COSTS OFF, VERBOSE) UPDATE public."MixedEvents" SET status = 'cold_upd' diff --git a/extension/coldfront/test/expected/param_cold_via_plpgsql.out b/extension/coldfront/test/expected/param_cold_via_plpgsql.out index d01f102..e48b811 100644 --- a/extension/coldfront/test/expected/param_cold_via_plpgsql.out +++ b/extension/coldfront/test/expected/param_cold_via_plpgsql.out @@ -28,8 +28,8 @@ CREATE TABLE public._events (id int, ts timestamptz, status text, data bytea); CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- (1) pure-cold UPDATE (literal cold predicate), param in SET: the cold call's -- SQL arg must be a format(...) call with a %N$L spec + a LIVE $1 arg, NOT a -- literal with a baked $1. @@ -128,8 +128,8 @@ CREATE TABLE public._eid (id bigint GENERATED ALWAYS AS IDENTITY, ts timestamptz CREATE VIEW public.eid AS SELECT * FROM public._eid; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'eid', 'public._eid', 'ice.default.eid', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('eid', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'eid', '2026-03-01'::timestamptz); PREPARE eid_ins(timestamptz, bytea) AS INSERT INTO public.eid (ts, data) VALUES ($1, $2); EXPLAIN (COSTS OFF, VERBOSE) EXECUTE eid_ins('2026-01-10 00:00+00', '\xcafe'::bytea); QUERY PLAN diff --git a/extension/coldfront/test/expected/param_cold_via_plpgsql_1.out b/extension/coldfront/test/expected/param_cold_via_plpgsql_1.out index 391570e..7e0ef05 100644 --- a/extension/coldfront/test/expected/param_cold_via_plpgsql_1.out +++ b/extension/coldfront/test/expected/param_cold_via_plpgsql_1.out @@ -28,8 +28,8 @@ CREATE TABLE public._events (id int, ts timestamptz, status text, data bytea); CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- (1) pure-cold UPDATE (literal cold predicate), param in SET: the cold call's -- SQL arg must be a format(...) call with a %N$L spec + a LIVE $1 arg, NOT a -- literal with a baked $1. @@ -128,8 +128,8 @@ CREATE TABLE public._eid (id bigint GENERATED ALWAYS AS IDENTITY, ts timestamptz CREATE VIEW public.eid AS SELECT * FROM public._eid; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'eid', 'public._eid', 'ice.default.eid', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('eid', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'eid', '2026-03-01'::timestamptz); PREPARE eid_ins(timestamptz, bytea) AS INSERT INTO public.eid (ts, data) VALUES ($1, $2); EXPLAIN (COSTS OFF, VERBOSE) EXECUTE eid_ins('2026-01-10 00:00+00', '\xcafe'::bytea); QUERY PLAN diff --git a/extension/coldfront/test/expected/returning_cold_rejected.out b/extension/coldfront/test/expected/returning_cold_rejected.out index c25a47f..3092996 100644 --- a/extension/coldfront/test/expected/returning_cold_rejected.out +++ b/extension/coldfront/test/expected/returning_cold_rejected.out @@ -16,8 +16,8 @@ CREATE TABLE public._events (id int, ts timestamptz, status text, data jsonb); CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- Cold UPDATE … RETURNING (ts < cutoff) — rejected. UPDATE public.events SET status = 'x' WHERE ts = '2026-01-15 01:00:00+00' RETURNING id; ERROR: RETURNING is not supported for writes to the cold tier of "events" diff --git a/extension/coldfront/test/expected/returning_literal_in_where.out b/extension/coldfront/test/expected/returning_literal_in_where.out index c13bc53..98a465f 100644 --- a/extension/coldfront/test/expected/returning_literal_in_where.out +++ b/extension/coldfront/test/expected/returning_literal_in_where.out @@ -17,8 +17,8 @@ CREATE TABLE public._events (id int, ts timestamptz, status text); CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- Cold-path UPDATE whose WHERE clause contains " RETURNING " inside a text -- literal. The raw_query argument must carry the full literal verbatim. EXPLAIN (COSTS OFF, VERBOSE) diff --git a/extension/coldfront/test/expected/schema_collision.out b/extension/coldfront/test/expected/schema_collision.out new file mode 100644 index 0000000..a4e11ce --- /dev/null +++ b/extension/coldfront/test/expected/schema_collision.out @@ -0,0 +1,68 @@ +-- Two tables sharing a name in DIFFERENT PG schemas map to DISTINCT Iceberg +-- tables (ice..
) and keep DISTINCT watermarks, so archiving one +-- does not clobber the other's cold data. White-box: EXPLAIN VERBOSE shows each +-- cold DML is rewritten against its own per-schema iceberg ref; real cold I/O is +-- ci/journey.sh (pg_regress has no Iceberg attached). Exercises both the iceberg +-- ref (naming) and the (schema, table) archive_watermark key. +CREATE EXTENSION IF NOT EXISTS pg_duckdb; +NOTICE: extension "pg_duckdb" already exists, skipping +CREATE EXTENSION IF NOT EXISTS coldfront; +NOTICE: extension "coldfront" already exists, skipping +SET TIME ZONE 'UTC'; +-- White-box: checks the hook's SQL, not Iceberg I/O. Real cold I/O is ci/journey.sh. +SET coldfront.warehouse = ''; +SET coldfront.lakekeeper_endpoint = ''; +CREATE SCHEMA myapp; +CREATE SCHEMA analytics; +CREATE TABLE myapp._events (id int, ts timestamptz, status text); +CREATE TABLE analytics._events (id int, ts timestamptz, status text); +CREATE VIEW myapp.events AS SELECT * FROM myapp._events; +CREATE VIEW analytics.events AS SELECT * FROM analytics._events; +-- Same relname 'events', different schema -> DISTINCT iceberg tables. +INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) +VALUES ('myapp', 'events', 'myapp._events', 'ice.myapp.events', 'ts'), + ('analytics', 'events', 'analytics._events', 'ice.analytics.events', 'ts'); +-- Two watermark rows: same bare name, different schema + cutoff. The composite +-- (schema, table) key keeps both distinct. +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('myapp', 'events', '2026-03-01'::timestamptz), + ('analytics', 'events', '2026-06-01'::timestamptz); +-- Cold UPDATE on analytics.events (ts < 2026-06-01) -> rewritten against +-- ice.analytics.events. +EXPLAIN (COSTS OFF, VERBOSE) + UPDATE analytics.events SET status = 'x' WHERE ts = '2026-01-15 00:00:00+00'; + QUERY PLAN +--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- + Result + Output: _exec_iceberg_with_claim('ice.analytics.events'::text, 'UPDATE ice.analytics.events SET status = ''x''::text WHERE (ts = ''Thu Jan 15 00:00:00 2026 UTC''::timestamptz)'::text) +(2 rows) + +-- Cold UPDATE on myapp.events (ts < 2026-03-01) -> rewritten against +-- ice.myapp.events, its OWN table, never analytics'. +EXPLAIN (COSTS OFF, VERBOSE) + UPDATE myapp.events SET status = 'x' WHERE ts = '2026-01-15 00:00:00+00'; + QUERY PLAN +------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- + Result + Output: _exec_iceberg_with_claim('ice.myapp.events'::text, 'UPDATE ice.myapp.events SET status = ''x''::text WHERE (ts = ''Thu Jan 15 00:00:00 2026 UTC''::timestamptz)'::text) +(2 rows) + +-- Both watermarks coexist, one row per (schema, table). +SELECT schema_name, table_name, cutoff_time +FROM coldfront.archive_watermark ORDER BY schema_name; + schema_name | table_name | cutoff_time +-------------+------------+------------------------------ + analytics | events | Mon Jun 01 00:00:00 2026 UTC + myapp | events | Sun Mar 01 00:00:00 2026 UTC +(2 rows) + +-- Cleanup. Unregister before dropping: the DDL hook blocks DROP of a registered +-- tiered relation. +DELETE FROM coldfront.tiered_views; +DELETE FROM coldfront.archive_watermark; +DROP VIEW myapp.events; +DROP VIEW analytics.events; +DROP TABLE myapp._events; +DROP TABLE analytics._events; +DROP SCHEMA myapp; +DROP SCHEMA analytics; diff --git a/extension/coldfront/test/expected/self_join_rejected.out b/extension/coldfront/test/expected/self_join_rejected.out index b00945a..ee808d6 100644 --- a/extension/coldfront/test/expected/self_join_rejected.out +++ b/extension/coldfront/test/expected/self_join_rejected.out @@ -17,8 +17,8 @@ CREATE TABLE public._events (id int, ts timestamptz, status text); CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- Self-join via FROM — rejected. UPDATE public.events SET status = 'x' FROM public.events e2 WHERE public.events.id = e2.id; ERROR: UPDATE/DELETE on tiered view "events" cannot reference it more than once diff --git a/extension/coldfront/test/expected/update_ambiguous_rejected.out b/extension/coldfront/test/expected/update_ambiguous_rejected.out index d4ace95..e911d81 100644 --- a/extension/coldfront/test/expected/update_ambiguous_rejected.out +++ b/extension/coldfront/test/expected/update_ambiguous_rejected.out @@ -18,8 +18,8 @@ INSERT INTO public._events VALUES (1, '2026-04-01 12:00:00+00', 'hot_orig'); CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- WHERE predicate does not reference ts → TIER_AMBIGUOUS → ERROR UPDATE public.events SET status = 'x' WHERE id = 1; ERROR: UPDATE/DELETE on tiered view "events" must include a WHERE condition on "ts" that targets one tier diff --git a/extension/coldfront/test/expected/update_cold_via_view.out b/extension/coldfront/test/expected/update_cold_via_view.out index 6acefb0..d1e0f91 100644 --- a/extension/coldfront/test/expected/update_cold_via_view.out +++ b/extension/coldfront/test/expected/update_cold_via_view.out @@ -15,8 +15,8 @@ INSERT INTO public._events VALUES (1, '2026-04-01 12:00:00+00', 'hot_orig'); CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- Cold-only (ts < cutoff): plan must call duckdb.raw_query with the cold DML. -- VERBOSE exposes the function argument so we can assert the generated SQL. EXPLAIN (COSTS OFF, VERBOSE) diff --git a/extension/coldfront/test/expected/update_hot_via_view.out b/extension/coldfront/test/expected/update_hot_via_view.out index ac3387b..1f4e896 100644 --- a/extension/coldfront/test/expected/update_hot_via_view.out +++ b/extension/coldfront/test/expected/update_hot_via_view.out @@ -16,8 +16,8 @@ INSERT INTO public._events VALUES CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- Hot-only (ts >= cutoff): plan must target _events, not duckdb.raw_query EXPLAIN (COSTS OFF) UPDATE public.events SET status = 'hot_upd' WHERE ts = '2026-04-01 12:00:00+00'; diff --git a/extension/coldfront/test/expected/update_partition_key_blocked.out b/extension/coldfront/test/expected/update_partition_key_blocked.out index f5e7638..4b635c0 100644 --- a/extension/coldfront/test/expected/update_partition_key_blocked.out +++ b/extension/coldfront/test/expected/update_partition_key_blocked.out @@ -18,8 +18,8 @@ INSERT INTO public._events VALUES (1, '2026-04-01 12:00:00+00', 'hot_orig'); CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- Strict mode: no cross-tier move, so every partition-column SET is rejected. SET coldfront.allow_mixed_writes = off; -- Cold→hot crossing: rejected. diff --git a/extension/coldfront/test/expected/update_partition_key_move.out b/extension/coldfront/test/expected/update_partition_key_move.out index 4c6cfe0..459bf76 100644 --- a/extension/coldfront/test/expected/update_partition_key_move.out +++ b/extension/coldfront/test/expected/update_partition_key_move.out @@ -18,8 +18,8 @@ CREATE TABLE public._events (id int, ts timestamptz, status text); CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); SET coldfront.allow_mixed_writes = on; -- (1) Constant new ts, WHERE on a non-partition column: the rewrite is one -- _cross_tier_move call carrying the deparsed WHERE and new-ts expression. diff --git a/extension/coldfront/test/sql/allow_mixed_writes.sql b/extension/coldfront/test/sql/allow_mixed_writes.sql index a2b516c..12c25de 100644 --- a/extension/coldfront/test/sql/allow_mixed_writes.sql +++ b/extension/coldfront/test/sql/allow_mixed_writes.sql @@ -15,8 +15,8 @@ CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- Default GUC value is permissive (on) SHOW coldfront.allow_mixed_writes; diff --git a/extension/coldfront/test/sql/bakery_wraps_cold_writes.sql b/extension/coldfront/test/sql/bakery_wraps_cold_writes.sql index 4a8994d..6f4a570 100644 --- a/extension/coldfront/test/sql/bakery_wraps_cold_writes.sql +++ b/extension/coldfront/test/sql/bakery_wraps_cold_writes.sql @@ -17,8 +17,8 @@ CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- (1) Tiered cold UPDATE (ts < cutoff): plan must call -- coldfront._exec_iceberg_with_claim, NOT bare duckdb.raw_query. diff --git a/extension/coldfront/test/sql/cast_in_dquoted_identifier.sql b/extension/coldfront/test/sql/cast_in_dquoted_identifier.sql index c9527ac..8a4b4b4 100644 --- a/extension/coldfront/test/sql/cast_in_dquoted_identifier.sql +++ b/extension/coldfront/test/sql/cast_in_dquoted_identifier.sql @@ -19,8 +19,8 @@ CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- Cold path: SET targets the column whose name contains the cast spelling. -- The column name must survive verbatim in the raw_query argument; only the diff --git a/extension/coldfront/test/sql/cast_literal_in_value.sql b/extension/coldfront/test/sql/cast_literal_in_value.sql index 1f4fd17..571e3e8 100644 --- a/extension/coldfront/test/sql/cast_literal_in_value.sql +++ b/extension/coldfront/test/sql/cast_literal_in_value.sql @@ -21,8 +21,8 @@ CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- Cold path. SET value contains an embedded "::timestamp with time zone" inside -- a text literal; that must stay intact in the raw_query argument. The real diff --git a/extension/coldfront/test/sql/cast_normalize.sql b/extension/coldfront/test/sql/cast_normalize.sql index d41a381..24980bd 100644 --- a/extension/coldfront/test/sql/cast_normalize.sql +++ b/extension/coldfront/test/sql/cast_normalize.sql @@ -24,8 +24,8 @@ CREATE TABLE public._events (id int, ts timestamptz, c_tsn timestamp, CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- (A) Cold UPDATE (ts < cutoff): the deparsed cold SQL must carry DuckDB spellings. -- Casts: timestamp without time zone → timestamp, character varying → varchar, diff --git a/extension/coldfront/test/sql/classify_between_in_or.sql b/extension/coldfront/test/sql/classify_between_in_or.sql index 1b27b93..9a269d8 100644 --- a/extension/coldfront/test/sql/classify_between_in_or.sql +++ b/extension/coldfront/test/sql/classify_between_in_or.sql @@ -23,8 +23,8 @@ CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- BETWEEN entirely below the cutoff → cold EXPLAIN (COSTS OFF) diff --git a/extension/coldfront/test/sql/classify_now.sql b/extension/coldfront/test/sql/classify_now.sql index a894f8e..6a573e4 100644 --- a/extension/coldfront/test/sql/classify_now.sql +++ b/extension/coldfront/test/sql/classify_now.sql @@ -16,8 +16,8 @@ CREATE TABLE public._events (id int, ts timestamptz, status text); CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- Stable, far in the past (< cutoff) → COLD: rewritten to a single cold _exec, -- NOT a dual write. The original now()-expression is preserved for execution. diff --git a/extension/coldfront/test/sql/cte_on_dml.sql b/extension/coldfront/test/sql/cte_on_dml.sql index 1a72f68..520e4a3 100644 --- a/extension/coldfront/test/sql/cte_on_dml.sql +++ b/extension/coldfront/test/sql/cte_on_dml.sql @@ -15,8 +15,8 @@ CREATE TABLE public._events (id int, ts timestamptz, status text); CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- Cold UPDATE (ts < cutoff) with a CTE referenced in SET: the WITH must be -- carried into the cold DML with the relation swapped to ice.default.events. diff --git a/extension/coldfront/test/sql/cte_on_insert.sql b/extension/coldfront/test/sql/cte_on_insert.sql index 29ba85c..5fbf972 100644 --- a/extension/coldfront/test/sql/cte_on_insert.sql +++ b/extension/coldfront/test/sql/cte_on_insert.sql @@ -16,8 +16,8 @@ CREATE TABLE public._events (id int, ts timestamptz, status text); CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- INSERT ... SELECT FROM a CTE into the tiered view: the WITH must reach both the -- hot (PG) and cold (DuckDB) halves, folded into the source derived table. diff --git a/extension/coldfront/test/sql/ddl_rename_view.sql b/extension/coldfront/test/sql/ddl_rename_view.sql index 8df134b..f773f6e 100644 --- a/extension/coldfront/test/sql/ddl_rename_view.sql +++ b/extension/coldfront/test/sql/ddl_rename_view.sql @@ -1,5 +1,5 @@ -- RENAME of the transparent VIEW must migrate every name-keyed reference: --- * coldfront.archive_watermark.table_name (keyed on the bare view name) +-- * coldfront.archive_watermark.table_name (the bare view name in its (schema, table) key) -- * the regenerated INSERT trigger function/trigger names -- Without the watermark migration the rebuilt view would silently lose its -- cold (Iceberg) UNION branch — the watermark lookup would miss, v_has_cutoff @@ -20,8 +20,8 @@ CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -- A watermark exists, so the rebuilt view must keep its cold UNION branch. -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- Before: view definition has the cold iceberg_scan branch. SELECT pg_get_viewdef('public.events'::regclass) LIKE '%iceberg_scan%' AS has_cold_branch; @@ -29,7 +29,7 @@ SELECT pg_get_viewdef('public.events'::regclass) LIKE '%iceberg_scan%' AS has_co -- Rename the view. ALTER VIEW public.events RENAME TO events_v2; --- The watermark row followed the rename (keyed on the new bare view name). +-- The watermark row followed the rename (its table_name is the new bare view name). SELECT table_name FROM coldfront.archive_watermark ORDER BY table_name; -- The rebuilt view under the new name STILL has the cold UNION branch diff --git a/extension/coldfront/test/sql/dollar_quote_in_value.sql b/extension/coldfront/test/sql/dollar_quote_in_value.sql index 37dc3c1..0ada5b6 100644 --- a/extension/coldfront/test/sql/dollar_quote_in_value.sql +++ b/extension/coldfront/test/sql/dollar_quote_in_value.sql @@ -19,8 +19,8 @@ CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- Cold path: SET value contains the exact dollar-quote tag the old code used. -- Without the fix, pg_parse_query would error on the rewritten SQL. diff --git a/extension/coldfront/test/sql/mixed_case_identifier.sql b/extension/coldfront/test/sql/mixed_case_identifier.sql index 7d5cb7b..2807928 100644 --- a/extension/coldfront/test/sql/mixed_case_identifier.sql +++ b/extension/coldfront/test/sql/mixed_case_identifier.sql @@ -22,8 +22,8 @@ VALUES ('public', 'MixedEvents', 'public."_MixedEvents"', 'ice.default."MixedEvents"', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('MixedEvents', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'MixedEvents', '2026-03-01'::timestamptz); -- Cold UPDATE through the mixed-case view: must rewrite, not error. EXPLAIN (COSTS OFF, VERBOSE) diff --git a/extension/coldfront/test/sql/param_cold_via_plpgsql.sql b/extension/coldfront/test/sql/param_cold_via_plpgsql.sql index 4bb15f7..d99a5ac 100644 --- a/extension/coldfront/test/sql/param_cold_via_plpgsql.sql +++ b/extension/coldfront/test/sql/param_cold_via_plpgsql.sql @@ -32,8 +32,8 @@ CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- (1) pure-cold UPDATE (literal cold predicate), param in SET: the cold call's -- SQL arg must be a format(...) call with a %N$L spec + a LIVE $1 arg, NOT a @@ -85,8 +85,8 @@ CREATE TABLE public._eid (id bigint GENERATED ALWAYS AS IDENTITY, ts timestamptz CREATE VIEW public.eid AS SELECT * FROM public._eid; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'eid', 'public._eid', 'ice.default.eid', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('eid', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'eid', '2026-03-01'::timestamptz); PREPARE eid_ins(timestamptz, bytea) AS INSERT INTO public.eid (ts, data) VALUES ($1, $2); EXPLAIN (COSTS OFF, VERBOSE) EXECUTE eid_ins('2026-01-10 00:00+00', '\xcafe'::bytea); diff --git a/extension/coldfront/test/sql/returning_cold_rejected.sql b/extension/coldfront/test/sql/returning_cold_rejected.sql index b59e081..7b64aea 100644 --- a/extension/coldfront/test/sql/returning_cold_rejected.sql +++ b/extension/coldfront/test/sql/returning_cold_rejected.sql @@ -17,8 +17,8 @@ CREATE TABLE public._events (id int, ts timestamptz, status text, data jsonb); CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- Cold UPDATE … RETURNING (ts < cutoff) — rejected. UPDATE public.events SET status = 'x' WHERE ts = '2026-01-15 01:00:00+00' RETURNING id; diff --git a/extension/coldfront/test/sql/returning_literal_in_where.sql b/extension/coldfront/test/sql/returning_literal_in_where.sql index c2a0ac1..77e2391 100644 --- a/extension/coldfront/test/sql/returning_literal_in_where.sql +++ b/extension/coldfront/test/sql/returning_literal_in_where.sql @@ -19,8 +19,8 @@ CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- Cold-path UPDATE whose WHERE clause contains " RETURNING " inside a text -- literal. The raw_query argument must carry the full literal verbatim. diff --git a/extension/coldfront/test/sql/schema_collision.sql b/extension/coldfront/test/sql/schema_collision.sql new file mode 100644 index 0000000..920bde0 --- /dev/null +++ b/extension/coldfront/test/sql/schema_collision.sql @@ -0,0 +1,57 @@ +-- Two tables sharing a name in DIFFERENT PG schemas map to DISTINCT Iceberg +-- tables (ice..
) and keep DISTINCT watermarks, so archiving one +-- does not clobber the other's cold data. White-box: EXPLAIN VERBOSE shows each +-- cold DML is rewritten against its own per-schema iceberg ref; real cold I/O is +-- ci/journey.sh (pg_regress has no Iceberg attached). Exercises both the iceberg +-- ref (naming) and the (schema, table) archive_watermark key. + +CREATE EXTENSION IF NOT EXISTS pg_duckdb; +CREATE EXTENSION IF NOT EXISTS coldfront; + +SET TIME ZONE 'UTC'; +-- White-box: checks the hook's SQL, not Iceberg I/O. Real cold I/O is ci/journey.sh. +SET coldfront.warehouse = ''; +SET coldfront.lakekeeper_endpoint = ''; + +CREATE SCHEMA myapp; +CREATE SCHEMA analytics; +CREATE TABLE myapp._events (id int, ts timestamptz, status text); +CREATE TABLE analytics._events (id int, ts timestamptz, status text); +CREATE VIEW myapp.events AS SELECT * FROM myapp._events; +CREATE VIEW analytics.events AS SELECT * FROM analytics._events; + +-- Same relname 'events', different schema -> DISTINCT iceberg tables. +INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) +VALUES ('myapp', 'events', 'myapp._events', 'ice.myapp.events', 'ts'), + ('analytics', 'events', 'analytics._events', 'ice.analytics.events', 'ts'); + +-- Two watermark rows: same bare name, different schema + cutoff. The composite +-- (schema, table) key keeps both distinct. +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('myapp', 'events', '2026-03-01'::timestamptz), + ('analytics', 'events', '2026-06-01'::timestamptz); + +-- Cold UPDATE on analytics.events (ts < 2026-06-01) -> rewritten against +-- ice.analytics.events. +EXPLAIN (COSTS OFF, VERBOSE) + UPDATE analytics.events SET status = 'x' WHERE ts = '2026-01-15 00:00:00+00'; + +-- Cold UPDATE on myapp.events (ts < 2026-03-01) -> rewritten against +-- ice.myapp.events, its OWN table, never analytics'. +EXPLAIN (COSTS OFF, VERBOSE) + UPDATE myapp.events SET status = 'x' WHERE ts = '2026-01-15 00:00:00+00'; + +-- Both watermarks coexist, one row per (schema, table). +SELECT schema_name, table_name, cutoff_time +FROM coldfront.archive_watermark ORDER BY schema_name; + +-- Cleanup. Unregister before dropping: the DDL hook blocks DROP of a registered +-- tiered relation. +DELETE FROM coldfront.tiered_views; +DELETE FROM coldfront.archive_watermark; +DROP VIEW myapp.events; +DROP VIEW analytics.events; +DROP TABLE myapp._events; +DROP TABLE analytics._events; +DROP SCHEMA myapp; +DROP SCHEMA analytics; diff --git a/extension/coldfront/test/sql/self_join_rejected.sql b/extension/coldfront/test/sql/self_join_rejected.sql index 2b72fd9..d6f9448 100644 --- a/extension/coldfront/test/sql/self_join_rejected.sql +++ b/extension/coldfront/test/sql/self_join_rejected.sql @@ -18,8 +18,8 @@ CREATE TABLE public._events (id int, ts timestamptz, status text); CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- Self-join via FROM — rejected. UPDATE public.events SET status = 'x' FROM public.events e2 WHERE public.events.id = e2.id; diff --git a/extension/coldfront/test/sql/update_ambiguous_rejected.sql b/extension/coldfront/test/sql/update_ambiguous_rejected.sql index 2ae51d9..b33aee9 100644 --- a/extension/coldfront/test/sql/update_ambiguous_rejected.sql +++ b/extension/coldfront/test/sql/update_ambiguous_rejected.sql @@ -20,8 +20,8 @@ CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- WHERE predicate does not reference ts → TIER_AMBIGUOUS → ERROR UPDATE public.events SET status = 'x' WHERE id = 1; diff --git a/extension/coldfront/test/sql/update_cold_via_view.sql b/extension/coldfront/test/sql/update_cold_via_view.sql index 06bc691..1f537f1 100644 --- a/extension/coldfront/test/sql/update_cold_via_view.sql +++ b/extension/coldfront/test/sql/update_cold_via_view.sql @@ -17,8 +17,8 @@ CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- Cold-only (ts < cutoff): plan must call duckdb.raw_query with the cold DML. -- VERBOSE exposes the function argument so we can assert the generated SQL. diff --git a/extension/coldfront/test/sql/update_hot_via_view.sql b/extension/coldfront/test/sql/update_hot_via_view.sql index 3665da0..2996d1e 100644 --- a/extension/coldfront/test/sql/update_hot_via_view.sql +++ b/extension/coldfront/test/sql/update_hot_via_view.sql @@ -18,8 +18,8 @@ CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- Hot-only (ts >= cutoff): plan must target _events, not duckdb.raw_query EXPLAIN (COSTS OFF) diff --git a/extension/coldfront/test/sql/update_partition_key_blocked.sql b/extension/coldfront/test/sql/update_partition_key_blocked.sql index 335c384..95e5edc 100644 --- a/extension/coldfront/test/sql/update_partition_key_blocked.sql +++ b/extension/coldfront/test/sql/update_partition_key_blocked.sql @@ -20,8 +20,8 @@ CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); -- Strict mode: no cross-tier move, so every partition-column SET is rejected. SET coldfront.allow_mixed_writes = off; diff --git a/extension/coldfront/test/sql/update_partition_key_move.sql b/extension/coldfront/test/sql/update_partition_key_move.sql index d4edf1d..cd70f9f 100644 --- a/extension/coldfront/test/sql/update_partition_key_move.sql +++ b/extension/coldfront/test/sql/update_partition_key_move.sql @@ -20,8 +20,8 @@ CREATE VIEW public.events AS SELECT * FROM public._events; INSERT INTO coldfront.tiered_views(schema_name, relname, hot_table, iceberg_table, partition_col) VALUES ('public', 'events', 'public._events', 'ice.default.events', 'ts'); -INSERT INTO coldfront.archive_watermark(table_name, cutoff_time) -VALUES ('events', '2026-03-01'::timestamptz); +INSERT INTO coldfront.archive_watermark(schema_name, table_name, cutoff_time) +VALUES ('public', 'events', '2026-03-01'::timestamptz); SET coldfront.allow_mixed_writes = on; diff --git a/internal/config/config.go b/internal/config/config.go index b2b6f4b..5d951a9 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -29,7 +29,6 @@ type PostgresConfig struct { type IcebergConfig struct { Warehouse string `yaml:"warehouse"` // Lakekeeper warehouse name LakekeeperEndpoint string `yaml:"lakekeeper_endpoint"` // e.g. http://lakekeeper:8181/catalog - Namespace string `yaml:"namespace"` } // S3Config holds credentials and endpoint for the object store backing the @@ -139,9 +138,6 @@ func Load(path string) (*Config, error) { // any `schema.table` syntax in SourceTable into separate SourceSchema and // SourceTable components. func applyDefaults(cfg *Config) { - if cfg.Iceberg.Namespace == "" { - cfg.Iceberg.Namespace = "default" - } if cfg.S3.Region == "" { cfg.S3.Region = "us-east-1" } diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 4bd44c1..f556943 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -52,7 +52,6 @@ func TestLoad_ValidConfig(t *testing.T) { func TestLoad_Defaults(t *testing.T) { cfg, err := Load(writeConfig(t, validConfig)) require.NoError(t, err) - assert.Equal(t, "default", cfg.Iceberg.Namespace) assert.Equal(t, "us-east-1", cfg.S3.Region) assert.Equal(t, "public", cfg.Archiver.Tables[0].SourceSchema) assert.Equal(t, 3, cfg.Archiver.Tables[0].FuturePartitions) diff --git a/internal/partcfg/commands.go b/internal/partcfg/commands.go index 51aed5a..4a05ecf 100644 --- a/internal/partcfg/commands.go +++ b/internal/partcfg/commands.go @@ -363,7 +363,13 @@ func (r configRow) insertSQL() string { return fmt.Sprintf(`INSERT INTO coldfront.partition_config (schema_name, table_name, partition_period, partition_column, future_partitions, part_mode, id_scheme, hot_period, retention_period, sub_part_values_source, expiration_strategy) -VALUES (%s, %s, %s, %s, %d, %s, %s, %s, %s, %s, %s);`, +VALUES (%s, %s, %s, %s, %d, %s, %s, %s, %s, %s, %s) +ON CONFLICT (schema_name, table_name) DO UPDATE SET + partition_period = EXCLUDED.partition_period, partition_column = EXCLUDED.partition_column, + future_partitions = EXCLUDED.future_partitions, part_mode = EXCLUDED.part_mode, + id_scheme = EXCLUDED.id_scheme, hot_period = EXCLUDED.hot_period, + retention_period = EXCLUDED.retention_period, sub_part_values_source = EXCLUDED.sub_part_values_source, + expiration_strategy = EXCLUDED.expiration_strategy;`, lit(r.schema), lit(r.table), lit(r.period), lit(r.column), r.premake, lit(r.partMode), lit(r.idScheme), lit(r.hot), lit(r.retention), lit(r.subValues), lit(r.strategy)) } diff --git a/internal/partcfg/commands_test.go b/internal/partcfg/commands_test.go index 9a9b580..1237ebf 100644 --- a/internal/partcfg/commands_test.go +++ b/internal/partcfg/commands_test.go @@ -94,6 +94,25 @@ func TestInsertSQL_EscapesQuotes(t *testing.T) { } } +// Re-registering an existing (schema, table) upserts: the INSERT carries an +// ON CONFLICT (schema_name, table_name) DO UPDATE clause. +func TestInsertSQL_Upsert(t *testing.T) { + got := configRow{ + schema: "myapp", table: "events", period: "monthly", + premake: 3, partMode: "timestamp", hot: "30 days", + }.insertSQL() + for _, want := range []string{ + "ON CONFLICT (schema_name, table_name) DO UPDATE SET", + "partition_period = EXCLUDED.partition_period", + "hot_period = EXCLUDED.hot_period", + "retention_period = EXCLUDED.retention_period", + } { + if !strings.Contains(got, want) { + t.Errorf("insertSQL missing %q:\n%s", want, got) + } + } +} + func TestSetClauses(t *testing.T) { vals := setVals{ period: "monthly", column: "ts", premake: 6, diff --git a/internal/view/view.go b/internal/view/view.go index 80bf7e0..ce63be7 100644 --- a/internal/view/view.go +++ b/internal/view/view.go @@ -22,7 +22,7 @@ type DBTX interface { type ViewConfig struct { SourceSchema string SourceTable string // original table name — becomes the view name after swap - IcebergTable string // attached catalog ref, e.g. ice.default.events + IcebergTable string // attached catalog ref, e.g. ice.myapp.events CutoffTime time.Time PartitionColumn string Columns []Column @@ -286,7 +286,7 @@ func GenerateTriggerFuncSQL(cfg ViewConfig) string { DECLARE cutoff timestamptz; BEGIN - SELECT cutoff_time INTO cutoff FROM coldfront.archive_watermark WHERE table_name = %s; + SELECT cutoff_time INTO cutoff FROM coldfront.archive_watermark WHERE schema_name = %s AND table_name = %s; IF cutoff IS NULL THEN cutoff := %s; END IF; @@ -307,7 +307,7 @@ BEGIN END; $fn$ LANGUAGE plpgsql`, funcName, - sqlutil.Literal(cfg.SourceTable), + sqlutil.Literal(cfg.SourceSchema), sqlutil.Literal(cfg.SourceTable), cutoff, col, iceRef, coldPlaceholders, coldVals, diff --git a/internal/view/view_test.go b/internal/view/view_test.go index b45ebea..dc0b917 100644 --- a/internal/view/view_test.go +++ b/internal/view/view_test.go @@ -161,6 +161,7 @@ func TestGenerateViewSQL_BootstrapMatchesCutoverHotCasts(t *testing.T) { func TestGenerateTriggerFuncSQL_ColdInsertWatermarkCheck(t *testing.T) { sql := GenerateTriggerFuncSQL(testCfg) assert.Contains(t, sql, "archive_watermark") + assert.Contains(t, sql, "schema_name = 'public'") // watermark lookup is schema-scoped assert.Contains(t, sql, `NEW."ts" < cutoff`) } diff --git a/internal/watermark/watermark.go b/internal/watermark/watermark.go index c22f7a8..4dfa728 100644 --- a/internal/watermark/watermark.go +++ b/internal/watermark/watermark.go @@ -33,19 +33,21 @@ func (s *Store) EnsureTable(ctx context.Context) error { } if _, err := s.db.Exec(ctx, ` CREATE TABLE IF NOT EXISTS coldfront.archive_watermark ( - table_name text PRIMARY KEY, - cutoff_time timestamptz NOT NULL + schema_name text NOT NULL, + table_name text NOT NULL, + cutoff_time timestamptz NOT NULL, + PRIMARY KEY (schema_name, table_name) )`); err != nil { return fmt.Errorf("create watermark table: %w", err) } return nil } -// Get returns the cutoff time for the given table. If no watermark exists, found is false. -func (s *Store) Get(ctx context.Context, tableName string) (cutoff time.Time, found bool, err error) { +// Get returns the cutoff time for the given (schema, table). If no watermark exists, found is false. +func (s *Store) Get(ctx context.Context, schemaName, tableName string) (cutoff time.Time, found bool, err error) { err = s.db.QueryRow(ctx, - `SELECT cutoff_time FROM coldfront.archive_watermark WHERE table_name = $1`, - tableName, + `SELECT cutoff_time FROM coldfront.archive_watermark WHERE schema_name = $1 AND table_name = $2`, + schemaName, tableName, ).Scan(&cutoff) if errors.Is(err, pgx.ErrNoRows) { return time.Time{}, false, nil @@ -56,13 +58,13 @@ func (s *Store) Get(ctx context.Context, tableName string) (cutoff time.Time, fo return cutoff, true, nil } -// Set upserts the cutoff time for the given table. -func (s *Store) Set(ctx context.Context, tableName string, cutoff time.Time) error { +// Set upserts the cutoff time for the given (schema, table). +func (s *Store) Set(ctx context.Context, schemaName, tableName string, cutoff time.Time) error { if _, err := s.db.Exec(ctx, ` - INSERT INTO coldfront.archive_watermark (table_name, cutoff_time) - VALUES ($1, $2) - ON CONFLICT (table_name) DO UPDATE SET cutoff_time = EXCLUDED.cutoff_time`, - tableName, cutoff, + INSERT INTO coldfront.archive_watermark (schema_name, table_name, cutoff_time) + VALUES ($1, $2, $3) + ON CONFLICT (schema_name, table_name) DO UPDATE SET cutoff_time = EXCLUDED.cutoff_time`, + schemaName, tableName, cutoff, ); err != nil { return fmt.Errorf("set watermark: %w", err) } diff --git a/internal/watermark/watermark_test.go b/internal/watermark/watermark_test.go index de546c0..3a9b7c9 100644 --- a/internal/watermark/watermark_test.go +++ b/internal/watermark/watermark_test.go @@ -64,7 +64,7 @@ func TestGet_Found(t *testing.T) { }, } s := NewStore(db) - got, found, err := s.Get(context.Background(), "events") + got, found, err := s.Get(context.Background(), "public", "events") require.NoError(t, err) assert.True(t, found) assert.Equal(t, cutoff, got) @@ -73,7 +73,7 @@ func TestGet_Found(t *testing.T) { func TestGet_NotFound(t *testing.T) { db := &mockDB{} // default rowFunc returns ErrNoRows s := NewStore(db) - _, found, err := s.Get(context.Background(), "events") + _, found, err := s.Get(context.Background(), "public", "events") require.NoError(t, err) assert.False(t, found) } @@ -87,7 +87,7 @@ func TestGet_Error(t *testing.T) { }, } s := NewStore(db) - _, _, err := s.Get(context.Background(), "events") + _, _, err := s.Get(context.Background(), "public", "events") require.Error(t, err) assert.Contains(t, err.Error(), "get watermark") } @@ -96,7 +96,7 @@ func TestSet(t *testing.T) { cutoff := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) db := &mockDB{} s := NewStore(db) - err := s.Set(context.Background(), "events", cutoff) + err := s.Set(context.Background(), "public", "events", cutoff) require.NoError(t, err) require.Len(t, db.execSQL, 1) assert.Contains(t, db.execSQL[0], "INSERT INTO") @@ -110,7 +110,7 @@ func TestSet_Error(t *testing.T) { }, } s := NewStore(db) - err := s.Set(context.Background(), "events", time.Now()) + err := s.Set(context.Background(), "public", "events", time.Now()) require.Error(t, err) assert.Contains(t, err.Error(), "set watermark") }