Fix #218 - could not open relation with OID

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

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); <waiting ...>
+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)
+

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 <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

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