diff --git a/Makefile b/Makefile index c0effef..1cf5dee 100644 --- a/Makefile +++ b/Makefile @@ -6,10 +6,17 @@ REGRESS = cmdline insert1 update1 update2 update3 update4 delete1 delete2 \ include_domain_data_type truncate type_oid actions position default \ pk rename_column numeric_data_types_as_string +ISOLATION = concurrent_index + PG_CONFIG = pg_config PGXS := $(shell $(PG_CONFIG) --pgxs) include $(PGXS) +# PGXS isolation-test support is available in 12+ +ifneq (,$(findstring $(MAJORVERSION),9.4 9.5 9.6 10 11)) +undefine ISOLATION +endif + # message API is available in 9.6+ ifneq (,$(findstring $(MAJORVERSION),9.4 9.5)) REGRESS := $(filter-out message, $(REGRESS)) diff --git a/expected/concurrent_index.out b/expected/concurrent_index.out new file mode 100644 index 0000000..683800d --- /dev/null +++ b/expected/concurrent_index.out @@ -0,0 +1,21 @@ +Parsed test spec with 4 sessions + +starting permutation: s1_begin s1_update s2_cic s3_insert s1_commit s4_get +?column? +-------- +init +(1 row) + +step s1_begin: BEGIN; +step s1_update: UPDATE cic_t SET v = 'x2' WHERE id = 1; +step s2_cic: CREATE INDEX CONCURRENTLY cic_i ON cic_t (v); +step s3_insert: INSERT INTO cic_t VALUES (2, 'y'); +step s1_commit: COMMIT; +step s2_cic: <... completed> +step s4_get: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'format-version', '2') WHERE data LIKE '%cic_t%'; +data +--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- +{"action":"I","schema":"public","table":"cic_t","columns":[{"name":"id","type":"integer","value":2},{"name":"v","type":"text","value":"y"}]} +{"action":"U","schema":"public","table":"cic_t","columns":[{"name":"id","type":"integer","value":1},{"name":"v","type":"text","value":"x2"}],"identity":[{"name":"id","type":"integer","value":1}]} +(2 rows) + diff --git a/specs/concurrent_index.spec b/specs/concurrent_index.spec new file mode 100644 index 0000000..6d92ed4 --- /dev/null +++ b/specs/concurrent_index.spec @@ -0,0 +1,51 @@ +# CREATE INDEX CONCURRENTLY vs logical decoding. +# +# An index created by CREATE INDEX CONCURRENTLY while an older transaction +# was still in progress used to make wal2json fail with +# ERROR: could not open relation with OID +# when decoding that transaction's changes, permanently wedging the slot: +# decoding a transaction that committed after CIC's first phase caches the +# new (invalid) index in the table's rd_indexlist, and decoding the older +# transaction's UPDATE afterwards opened every listed index under an older +# historic snapshot to which the new index's pg_class row is not visible. +# +# The WHERE clause keeps the output independent of incidental empty +# transactions (the CIC phases themselves, autovacuum). + +setup +{ + CREATE TABLE cic_t (id int PRIMARY KEY, v text); + INSERT INTO cic_t VALUES (1, 'seed'); +} + +setup +{ + SELECT 'init' FROM pg_create_logical_replication_slot('isolation_slot', 'wal2json'); +} + +teardown +{ + SELECT 'stop' FROM pg_drop_replication_slot('isolation_slot'); + DROP TABLE cic_t; +} + +session s1 +step s1_begin { BEGIN; } +step s1_update { UPDATE cic_t SET v = 'x2' WHERE id = 1; } +step s1_commit { COMMIT; } + +session s2 +step s2_cic { CREATE INDEX CONCURRENTLY cic_i ON cic_t (v); } + +session s3 +step s3_insert { INSERT INTO cic_t VALUES (2, 'y'); } + +session s4 +step s4_get { SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'format-version', '2') WHERE data LIKE '%cic_t%'; } + +# s2_cic commits its first phase (catalog entries for the new index), then +# blocks in WaitForLockers on s1's open transaction; the isolation tester +# proceeds once the session is detected as waiting. s3's insert-only +# transaction is decoded first (caching the new index in rd_indexlist), +# then s1's older update trips the bug on unpatched wal2json. +permutation s1_begin s1_update s2_cic s3_insert s1_commit s4_get diff --git a/wal2json.c b/wal2json.c index 50e6503..096720d 100644 --- a/wal2json.c +++ b/wal2json.c @@ -1468,6 +1468,63 @@ tuple_to_stringinfo(LogicalDecodingContext *ctx, TupleDesc tupdesc, HeapTuple tu pfree(colvalues.data); } +/* + * Historic-snapshot-safe substitute for RelationGetIndexAttrBitmap(). + * + * RelationGetIndexAttrBitmap() opens every index returned by + * RelationGetIndexList() -- including invalid/in-progress ones created by + * CREATE INDEX CONCURRENTLY. During logical decoding, rd_indexlist may have + * been cached while decoding a transaction that used a newer historic + * snapshot than the one installed for the change currently being decoded, so + * such an index's pg_class tuple can be invisible to the current snapshot and + * relation_open() fails with "could not open relation with OID ...". That + * error permanently blocks the slot because the same WAL is replayed on every + * retry. + * + * Instead, open only the one index we actually care about (primary key or + * replica identity), without taking a lock, the same way pgoutput's + * RelationGetIdentityKeyBitmap() does. If that index cannot be opened under + * the current historic snapshot, return NULL so the caller degrades + * gracefully instead of erroring out. + * + * The caller must have called RelationGetIndexList() on this relation so that + * rd_pkindex/rd_replidindex are set. + */ +static Bitmapset * +get_index_column_bitmap(Relation relation, Oid indexoid) +{ + Bitmapset *bs = NULL; + Relation indexrel; + int nkeyatts; + int i; + + if (!OidIsValid(indexoid)) + return NULL; + + indexrel = RelationIdGetRelation(indexoid); + if (indexrel == NULL) + return NULL; + +#if PG_VERSION_NUM >= 110000 + nkeyatts = IndexRelationGetNumberOfKeyAttributes(indexrel); +#else + nkeyatts = indexrel->rd_index->indnatts; +#endif + + for (i = 0; i < nkeyatts; i++) + { + int attnum = indexrel->rd_index->indkey.values[i]; + + /* primary key/replica identity indexes cannot contain expressions */ + if (attnum > 0) + bs = bms_add_member(bs, attnum - FirstLowInvalidHeapAttributeNumber); + } + + RelationClose(indexrel); + + return bs; +} + /* Print columns information */ static void columns_to_stringinfo(LogicalDecodingContext *ctx, TupleDesc tupdesc, HeapTuple tuple, bool addcomma, Relation relation) @@ -1879,7 +1936,7 @@ pg_decode_change_v1(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, if (data->include_pk) #if PG_VERSION_NUM >= 100000 - pkbs = RelationGetIndexAttrBitmap(relation, INDEX_ATTR_BITMAP_PRIMARY_KEY); + pkbs = get_index_column_bitmap(relation, relation->rd_pkindex); #else pkbs = RelationGetIndexAttrBitmap(relation, INDEX_ATTR_BITMAP_KEY); #endif @@ -1947,7 +2004,7 @@ pg_decode_change_v1(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, { elog(DEBUG1, "old tuple is null"); - ribs = RelationGetIndexAttrBitmap(relation, INDEX_ATTR_BITMAP_IDENTITY_KEY); + ribs = get_index_column_bitmap(relation, relation->rd_replidindex); #if PG_VERSION_NUM >= 170000 identity_to_stringinfo(ctx, tupdesc, change->data.tp.newtuple, ribs); #else @@ -1979,7 +2036,7 @@ pg_decode_change_v1(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, #endif } - ribs = RelationGetIndexAttrBitmap(relation, INDEX_ATTR_BITMAP_IDENTITY_KEY); + ribs = get_index_column_bitmap(relation, relation->rd_replidindex); #if PG_VERSION_NUM >= 170000 identity_to_stringinfo(ctx, tupdesc, change->data.tp.oldtuple, ribs); #else @@ -2130,12 +2187,12 @@ pg_decode_write_tuple(LogicalDecodingContext *ctx, Relation relation, HeapTuple /* figure out replica identity columns */ if (kind == PGOUTPUTJSON_IDENTITY) { - bs = RelationGetIndexAttrBitmap(relation, INDEX_ATTR_BITMAP_IDENTITY_KEY); + bs = get_index_column_bitmap(relation, relation->rd_replidindex); } else if (kind == PGOUTPUTJSON_PK) { #if PG_VERSION_NUM >= 100000 - bs = RelationGetIndexAttrBitmap(relation, INDEX_ATTR_BITMAP_PRIMARY_KEY); + bs = get_index_column_bitmap(relation, relation->rd_pkindex); #else bs = RelationGetIndexAttrBitmap(relation, INDEX_ATTR_BITMAP_KEY); #endif