From 463ce8e09ca87e6c8c8e1088872e4297d3fc1623 Mon Sep 17 00:00:00 2001 From: Jimmy Angelakos Date: Fri, 25 Sep 2026 20:26:27 +0100 Subject: [PATCH 1/3] fix: adopt the partitioner's tables at CREATE EXTENSION --- ci/journey.sh | 11 ++++++ docs/changelog.md | 3 ++ docs/usage.md | 4 +++ extension/coldfront/coldfront--1.0.sql | 22 ++++++++++++ internal/partcfg/mirror_test.go | 50 ++++++++++++++++++++++++++ internal/partcfg/partcfg.go | 11 +++--- internal/watermark/watermark.go | 21 +++++++---- 7 files changed, 110 insertions(+), 12 deletions(-) create mode 100644 internal/partcfg/mirror_test.go diff --git a/ci/journey.sh b/ci/journey.sh index f38537f..f2c386f 100755 --- a/ci/journey.sh +++ b/ci/journey.sh @@ -4112,6 +4112,17 @@ story_partitioner_stock_pg() { fail "TC-151: reconcile failed on stock PG"; tail -5 $TMPD/stock-run.log fi + # The cold tier can be added to this database later: CREATE EXTENSION adopts + # the partition_config the partitioner made, rows and all, instead of + # refusing to skip a table it does not own. + if qdb $db "CREATE EXTENSION pg_duckdb; CREATE EXTENSION coldfront;" >$TMPD/stock-ext.log 2>&1; then + pass "TC-151: the extension installs on a partitioner-first database" + else + fail "TC-151: CREATE EXTENSION failed on a partitioner-first database"; tail -3 $TMPD/stock-ext.log + fi + assert_eq "TC-151: partition_config is a member of the extension and kept its row" "1" \ + "$(qdb $db "SELECT count(*) FROM coldfront.partition_config WHERE table_name = 'stockev' AND EXISTS (SELECT 1 FROM pg_depend d JOIN pg_extension e ON e.oid = d.refobjid WHERE d.objid = 'coldfront.partition_config'::regclass AND e.extname = 'coldfront' AND d.deptype = 'e');")" + "$PARTITIONER" remove --dsn "$dsn" --table stockev >/dev/null 2>&1 assert_eq "TC-151: remove unregistered it" "0" \ "$(qdb $db "SELECT count(*) FROM coldfront.partition_config WHERE table_name='stockev';")" diff --git a/docs/changelog.md b/docs/changelog.md index 782d261..752197f 100644 --- a/docs/changelog.md +++ b/docs/changelog.md @@ -45,6 +45,9 @@ and this project adheres to ### Fixed +- `CREATE EXTENSION coldfront` failed on a database the standalone partitioner + had already set up ("table partition_config is not a member of extension"): + the extension now adopts that table, registrations included. - After one cold write on a mesh, an app role could run any SQL as the loopback connection's user through the `coldfront_self` dblink connection the claim left open in its session, and by setting diff --git a/docs/usage.md b/docs/usage.md index e93e1de..94b1efa 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -307,6 +307,10 @@ go build -o bin/partitioner ./cmd/partitioner # or: make build ./bin/partitioner --config config.yaml ``` +A database set up this way can gain the cold tier later: `CREATE EXTENSION +coldfront` adopts the `coldfront.partition_config` the partitioner created, +registrations included. + A partition-only config omits the `iceberg:` and `s3:` sections entirely (supply either all of them or none - a half-filled cold config is rejected): diff --git a/extension/coldfront/coldfront--1.0.sql b/extension/coldfront/coldfront--1.0.sql index 6fa5f98..f162cee 100644 --- a/extension/coldfront/coldfront--1.0.sql +++ b/extension/coldfront/coldfront--1.0.sql @@ -60,6 +60,28 @@ CREATE TABLE coldfront.tiered_views ( -- 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. +-- +-- A database the standalone partitioner set up before the extension holds +-- coldfront.partition_config as a plain table, and one the archiver ran on +-- holds archive_watermark the same way, both from the Go copies of this DDL +-- (internal/partcfg, internal/watermark). An extension script may skip +-- CREATE ... IF NOT EXISTS only for an object the extension owns, so such a +-- table is adopted first: its rows stay, and the pg_extension_config_dump +-- calls below then cover it like a table this script created. +DO $$ +DECLARE t text; +BEGIN + FOREACH t IN ARRAY ARRAY['archive_watermark', 'partition_config'] LOOP + IF to_regclass('coldfront.' || t) IS NOT NULL AND NOT EXISTS ( + SELECT 1 FROM pg_depend + WHERE classid = 'pg_class'::regclass + AND objid = to_regclass('coldfront.' || t) + AND refclassid = 'pg_extension'::regclass + AND deptype = 'e') THEN + EXECUTE format('ALTER EXTENSION coldfront ADD TABLE coldfront.%I', t); + END IF; + END LOOP; +END $$; CREATE TABLE IF NOT EXISTS coldfront.archive_watermark ( schema_name text NOT NULL, table_name text NOT NULL, diff --git a/internal/partcfg/mirror_test.go b/internal/partcfg/mirror_test.go new file mode 100644 index 0000000..ab10b10 --- /dev/null +++ b/internal/partcfg/mirror_test.go @@ -0,0 +1,50 @@ +package partcfg + +import ( + "os" + "regexp" + "strings" + "testing" + + "github.com/pgedge/coldfront/internal/watermark" +) + +// extensionTableDDL returns the CREATE TABLE block for one table from the +// extension's install script, without its trailing comments and semicolon. +func extensionTableDDL(t *testing.T, table string) string { + t.Helper() + script, err := os.ReadFile("../../extension/coldfront/coldfront--1.0.sql") + if err != nil { + t.Fatal(err) + } + head := "CREATE TABLE IF NOT EXISTS coldfront." + table + " (" + start := strings.Index(string(script), head) + if start < 0 { + t.Fatalf("%s not found in the extension script", head) + } + end := strings.Index(string(script)[start:], "\n);") + if end < 0 { + t.Fatalf("%s: no closing line in the extension script", head) + } + return regexp.MustCompile(`--[^\n]*`).ReplaceAllString(string(script)[start:start+end+2], "") +} + +// tokens collapses whitespace so two spellings of one DDL compare equal. +func tokens(ddl string) string { + return strings.Join(strings.Fields(ddl), " ") +} + +// The Go copies of the two tables the extension owns are what a database gets +// when the partitioner or the archiver runs before CREATE EXTENSION, and the +// extension adopts those tables as its own at install. So each copy has to be +// the extension's DDL, token for token. +func TestMirrorDDL_MatchesExtension(t *testing.T) { + for table, mirror := range map[string]string{ + "partition_config": CreateTableSQL, + "archive_watermark": watermark.CreateTableSQL, + } { + if got, want := tokens(mirror), tokens(extensionTableDDL(t, table)); got != want { + t.Errorf("%s: the Go copy differs from the extension script\n got: %s\nwant: %s", table, got, want) + } + } +} diff --git a/internal/partcfg/partcfg.go b/internal/partcfg/partcfg.go index 86f8ede..bc03c77 100644 --- a/internal/partcfg/partcfg.go +++ b/internal/partcfg/partcfg.go @@ -25,11 +25,12 @@ type DBTX interface { Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error) } -// createTableSQL mirrors the coldfront.partition_config DDL in the C extension +// CreateTableSQL mirrors the coldfront.partition_config DDL in the C extension // (coldfront--1.0.sql) so the vanilla partitioner — stock PG, no extension — -// can self-materialize the same table. Keep the two in sync (same pattern as -// archive_watermark / watermark.EnsureTable). -const createTableSQL = ` +// can self-materialize the same table. The extension adopts that table when it +// is installed later, so the two are kept token-identical (TestMirrorDDL_MatchesExtension; +// same pattern as archive_watermark / watermark.EnsureTable). +const CreateTableSQL = ` CREATE TABLE IF NOT EXISTS coldfront.partition_config ( schema_name text NOT NULL DEFAULT 'public', table_name text NOT NULL, @@ -92,7 +93,7 @@ func EnsureTable(ctx context.Context, db DBTX) error { if _, err := db.Exec(ctx, `CREATE SCHEMA IF NOT EXISTS coldfront`); err != nil { return fmt.Errorf("create schema: %w", err) } - if _, err := db.Exec(ctx, createTableSQL); err != nil { + if _, err := db.Exec(ctx, CreateTableSQL); err != nil { return fmt.Errorf("create partition_config: %w", err) } if _, err := db.Exec(ctx, addColumnsSQL); err != nil { diff --git a/internal/watermark/watermark.go b/internal/watermark/watermark.go index 4dfa728..ac67e64 100644 --- a/internal/watermark/watermark.go +++ b/internal/watermark/watermark.go @@ -26,18 +26,25 @@ func NewStore(db DBTX) *Store { return &Store{db: db} } +// CreateTableSQL mirrors the coldfront.archive_watermark DDL in the C extension +// (coldfront--1.0.sql), so an archiver run can materialize the table before +// CREATE EXTENSION. The extension adopts that table when it is installed +// later, so the two are kept token-identical (partcfg's +// TestMirrorDDL_MatchesExtension). +const CreateTableSQL = ` +CREATE TABLE IF NOT EXISTS coldfront.archive_watermark ( + schema_name text NOT NULL, + table_name text NOT NULL, + cutoff_time timestamptz NOT NULL, + PRIMARY KEY (schema_name, table_name) +)` + // EnsureTable creates the coldfront schema and archive_watermark table if they don't exist. func (s *Store) EnsureTable(ctx context.Context) error { if _, err := s.db.Exec(ctx, `CREATE SCHEMA IF NOT EXISTS coldfront`); err != nil { return fmt.Errorf("create schema: %w", err) } - if _, err := s.db.Exec(ctx, ` - CREATE TABLE IF NOT EXISTS coldfront.archive_watermark ( - schema_name text NOT NULL, - table_name text NOT NULL, - cutoff_time timestamptz NOT NULL, - PRIMARY KEY (schema_name, table_name) - )`); err != nil { + if _, err := s.db.Exec(ctx, CreateTableSQL); err != nil { return fmt.Errorf("create watermark table: %w", err) } return nil From 77e03f8ed3f391513ffd3aae77d16d3765916efb Mon Sep 17 00:00:00 2001 From: Jimmy Angelakos Date: Fri, 25 Sep 2026 20:26:27 +0100 Subject: [PATCH 2/3] test: reap a stale same-node claim by the epoch rule --- ci/journey.sh | 31 +++++++++++++++++++++++++++++++ 1 file changed, 31 insertions(+) diff --git a/ci/journey.sh b/ci/journey.sh index f2c386f..a7b405e 100755 --- a/ci/journey.sh +++ b/ci/journey.sh @@ -2360,6 +2360,37 @@ SQL wait "$holder" 2>/dev/null assert_eq "TC-163 peer deferred behind an orphan is unwedged by its own poke once the lock drops" "0" "$rc" assert_eq "TC-163 orphan claim reaped on both nodes" "0" "$(orphan_gone_everywhere)" + + # TC-191: epoch path. A same-node claim on another table whose ticket predates + # this postmaster is a leftover of a previous owner: a restart, or a + # reassigned snowflake.node. The other-table rule cannot reap it while a live + # session holds that table's lock, so only the epoch rule does. The lock is + # held for the whole write; a write that outlived it would prove nothing. + local stale_tbl="${ref}_tc191" stale + stale=$(q "$HOST" "WITH ins AS (INSERT INTO coldfront.claims (iceberg_table, ticket) SELECT '$stale_tbl', snowflake.nextval() - (((extract(epoch FROM now() - pg_postmaster_start_time())::bigint + 3600) * 1000) << 22) RETURNING ticket) SELECT ticket FROM ins;") + assert_eq "TC-191 the planted ticket carries this node and predates the postmaster" "t" \ + "$(q "$HOST" "SELECT snowflake.get_node($stale) = coldfront.node_id() AND snowflake.get_epoch($stale) < extract(epoch FROM pg_postmaster_start_time());")" + docker exec -i -e PGUSER="$CF_DBUSER" -e PGDATABASE="$CF_DBNAME" "$HOST" "$CF_PSQL" -tA -v ON_ERROR_STOP=1 >/dev/null 2>&1 <"$TMPD/reap.d" 2>&1; rc=$? + w_end=$(date +%s) + wait "$holder" 2>/dev/null + assert_eq "TC-191 the cold write to another table lands" "0" "$rc" + assert_gt "TC-191 the write finished while the stale claim's table stayed locked" "$w_end" "$(date +%s)" + assert_eq "TC-191 the stale claim was reaped by the epoch rule" "0" \ + "$(q "$HOST" "SELECT count(*) FROM coldfront.claims WHERE ticket = $stale;")" assert_eq "TC-160..163 all three reaper-path writes landed" "3" \ "$(q "$HOST" "SELECT count(*) FROM $tbl WHERE status='reap';")" rm -f $TMPD/reap.* 2>/dev/null From fc7d68f04e4b27ead6cc202b26a8f5ab9830b963 Mon Sep 17 00:00:00 2001 From: Jimmy Angelakos Date: Fri, 25 Sep 2026 21:22:16 +0100 Subject: [PATCH 3/3] test: probe the stale claim's table lock around the TC-191 write --- ci/journey.sh | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/ci/journey.sh b/ci/journey.sh index a7b405e..ad51723 100755 --- a/ci/journey.sh +++ b/ci/journey.sh @@ -2377,18 +2377,18 @@ SELECT pg_sleep(8); COMMIT; SQL holder=$! - local w_d w_end + local w_d probe="SELECT pg_try_advisory_xact_lock(hashtext('coldfront_iceberg:' || '$stale_tbl'));" if [ "$MODE" = tiered ]; then w_d="INSERT INTO events (ts,status,data) VALUES (date_trunc('month',now()) - interval '4 months' + interval '23 days','reap191','{}');" else w_d="INSERT INTO iceonly VALUES (9104,date_trunc('month',now()) + interval '3 months' + interval '4 days','reap191','{}');" fi sleep 1 + assert_eq "TC-191 the stale claim's table lock is held before the write" "f" "$(q "$HOST" "$probe")" tq 90 "$HOST" "$w_d" >"$TMPD/reap.d" 2>&1; rc=$? - w_end=$(date +%s) - wait "$holder" 2>/dev/null assert_eq "TC-191 the cold write to another table lands" "0" "$rc" - assert_gt "TC-191 the write finished while the stale claim's table stayed locked" "$w_end" "$(date +%s)" + assert_eq "TC-191 the stale claim's table lock is still held after the write" "f" "$(q "$HOST" "$probe")" + wait "$holder" 2>/dev/null assert_eq "TC-191 the stale claim was reaped by the epoch rule" "0" \ "$(q "$HOST" "SELECT count(*) FROM coldfront.claims WHERE ticket = $stale;")" assert_eq "TC-160..163 all three reaper-path writes landed" "3" \