From ee545a7e09906b891c1e81fbcab6015efccae7b9 Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy Date: Sat, 4 Jul 2026 02:47:29 +0000 Subject: [PATCH v11] Fix pg_get_publication_tables() failure with concurrent DROP TABLE. pg_get_publication_tables() collects the OIDs of the published tables on its first call, without locking them, and then reopens each table later, once per result row, to compute its column list and fetch its row filter. The reopen used table_open(), which errors out with "could not open relation with OID" if the table has been dropped in the meantime. This could happen for any published table without an explicit column list, which is every table in FOR ALL TABLES and FOR TABLES IN SCHEMA publications, but also FOR TABLE entries without a column list. The failure is common in environments where many tables are created and dropped while publication tables are being queried, e.g. by table synchronization on a subscriber. Fix by opening every table with try_table_open(), which returns NULL if the relation no longer exists, and skipping the table in that case. Concurrently dropped tables are thus simply absent from the result set, which is the expected point-in-time behavior. To skip dropped tables without emitting a row, track the current index into the table list in the SRF state ourselves instead of relying on funcctx->call_cntr. As a side effect, tables with an explicit column list, which were previously returned without being opened, are now also locked with AccessShareLock, so the function can block behind concurrent DDL on such tables where it previously did not. Backpatch to v16, where we added the table_open() call in pg_get_publication_tables(). Author: Bharath Rupireddy Reviewed-by: Bertrand Drouvot Reviewed-by: shveta malik Reviewed-by: Ajin Cherian Reviewed-by: Masahiko Sawada Discussion: https://www.postgresql.org/message-id/CALj2ACVYYooWH-5tJ6cPKkU%2BmutVxwb_z4S%2BqAi-zdrFqxXE2Q%40mail.gmail.com Backpatch-through: 16 --- src/backend/catalog/pg_publication.c | 52 +++++++++++++++---- .../expected/pub-concurrent-drop.out | 16 ++++++ src/test/isolation/isolation_schedule | 1 + .../isolation/specs/pub-concurrent-drop.spec | 36 +++++++++++++ src/tools/pgindent/typedefs.list | 1 + 5 files changed, 97 insertions(+), 9 deletions(-) create mode 100644 src/test/isolation/expected/pub-concurrent-drop.out create mode 100644 src/test/isolation/specs/pub-concurrent-drop.spec diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c index 5c457d9aca8..e6ebc1e2627 100644 --- a/src/backend/catalog/pg_publication.c +++ b/src/backend/catalog/pg_publication.c @@ -1414,14 +1414,27 @@ pg_get_publication_tables(FunctionCallInfo fcinfo, ArrayType *pubnames, bool pub_missing_ok) { #define NUM_PUBLICATION_TABLES_ELEM 4 + + /* + * State carried across SRF calls. We track the index ourselves instead of + * using funcctx->call_cntr, so that concurrently dropped tables can be + * skipped without emitting a row. + */ + typedef struct + { + List *table_infos; /* list of published_rel */ + int curr_idx; /* current index into table_infos */ + } publication_tables_state; + FuncCallContext *funcctx; - List *table_infos = NIL; + publication_tables_state *ptstate = NULL; /* stuff done only on the first call of the function */ if (SRF_IS_FIRSTCALL()) { TupleDesc tupdesc; MemoryContext oldcontext; + List *table_infos = NIL; Datum *elems; int nelems, i; @@ -1544,26 +1557,47 @@ pg_get_publication_tables(FunctionCallInfo fcinfo, ArrayType *pubnames, TupleDescFinalize(tupdesc); funcctx->tuple_desc = BlessTupleDesc(tupdesc); - funcctx->user_fctx = table_infos; + + /* Store the state to be used across SRF calls. */ + ptstate = palloc_object(publication_tables_state); + ptstate->table_infos = table_infos; + ptstate->curr_idx = 0; + funcctx->user_fctx = ptstate; MemoryContextSwitchTo(oldcontext); } /* stuff done on every call of the function */ funcctx = SRF_PERCALL_SETUP(); - table_infos = (List *) funcctx->user_fctx; + ptstate = (publication_tables_state *) funcctx->user_fctx; - if (funcctx->call_cntr < list_length(table_infos)) + while (ptstate->curr_idx < list_length(ptstate->table_infos)) { HeapTuple pubtuple = NULL; HeapTuple rettuple; Publication *pub; - published_rel *table_info = (published_rel *) list_nth(table_infos, funcctx->call_cntr); + published_rel *table_info = (published_rel *) list_nth(ptstate->table_infos, + ptstate->curr_idx); Oid relid = table_info->relid; - Oid schemaid = get_rel_namespace(relid); + Relation rel; + Oid schemaid; Datum values[NUM_PUBLICATION_TABLES_ELEM] = {0}; bool nulls[NUM_PUBLICATION_TABLES_ELEM] = {0}; + /* Advance the index for the next call. */ + ptstate->curr_idx++; + + /* + * The table OIDs were collected earlier, so a table may have been + * dropped before we get here. try_table_open() returns NULL if it is + * already gone, in which case we skip it; such tables are simply + * absent from the result set, which is the expected point-in-time + * behavior. + */ + rel = try_table_open(relid, AccessShareLock); + if (rel == NULL) + continue; + /* * Form tuple with appropriate data. */ @@ -1577,6 +1611,7 @@ pg_get_publication_tables(FunctionCallInfo fcinfo, ArrayType *pubnames, * We don't consider row filters or column lists for FOR ALL TABLES or * FOR TABLES IN SCHEMA publications. */ + schemaid = RelationGetNamespace(rel); if (!pub->alltables && !SearchSysCacheExists2(PUBLICATIONNAMESPACEMAP, ObjectIdGetDatum(schemaid), @@ -1606,7 +1641,6 @@ pg_get_publication_tables(FunctionCallInfo fcinfo, ArrayType *pubnames, /* Show all columns when the column list is not specified. */ if (nulls[2]) { - Relation rel = table_open(relid, AccessShareLock); int nattnums = 0; int16 *attnums; TupleDesc desc = RelationGetDescr(rel); @@ -1643,10 +1677,10 @@ pg_get_publication_tables(FunctionCallInfo fcinfo, ArrayType *pubnames, values[2] = PointerGetDatum(buildint2vector(attnums, nattnums)); nulls[2] = false; } - - table_close(rel, AccessShareLock); } + table_close(rel, AccessShareLock); + rettuple = heap_form_tuple(funcctx->tuple_desc, values, nulls); SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(rettuple)); diff --git a/src/test/isolation/expected/pub-concurrent-drop.out b/src/test/isolation/expected/pub-concurrent-drop.out new file mode 100644 index 00000000000..8360af0ec9c --- /dev/null +++ b/src/test/isolation/expected/pub-concurrent-drop.out @@ -0,0 +1,16 @@ +Parsed test spec with 2 sessions + +starting permutation: lock list_pub_tables drop_and_commit +step lock: BEGIN; LOCK pubdrop.dropme IN ACCESS EXCLUSIVE MODE; +step list_pub_tables: + SELECT relid::regclass AS tablename + FROM pg_get_publication_tables('pub_schema') + ORDER BY tablename; + +step drop_and_commit: DROP TABLE pubdrop.dropme; COMMIT; +step list_pub_tables: <... completed> +tablename +-------------- +pubdrop.keepme +(1 row) + diff --git a/src/test/isolation/isolation_schedule b/src/test/isolation/isolation_schedule index b8ebe92553c..28a33cc0104 100644 --- a/src/test/isolation/isolation_schedule +++ b/src/test/isolation/isolation_schedule @@ -101,6 +101,7 @@ test: async-notify test: vacuum-no-cleanup-lock test: timeouts test: vacuum-concurrent-drop +test: pub-concurrent-drop test: vacuum-conflict test: vacuum-skip-locked test: stats diff --git a/src/test/isolation/specs/pub-concurrent-drop.spec b/src/test/isolation/specs/pub-concurrent-drop.spec new file mode 100644 index 00000000000..4f7d701d60c --- /dev/null +++ b/src/test/isolation/specs/pub-concurrent-drop.spec @@ -0,0 +1,36 @@ +# Tests for concurrently dropping a relation while a publication's tables are +# being listed. + +setup +{ + CREATE SCHEMA pubdrop; + CREATE PUBLICATION pub_schema FOR TABLES IN SCHEMA pubdrop; + CREATE TABLE pubdrop.dropme (id int); + CREATE TABLE pubdrop.keepme (id int); +} + +teardown +{ + DROP SCHEMA pubdrop CASCADE; + DROP PUBLICATION pub_schema; +} + +session s1 +step lock { BEGIN; LOCK pubdrop.dropme IN ACCESS EXCLUSIVE MODE; } +step drop_and_commit { DROP TABLE pubdrop.dropme; COMMIT; } + +session s2 +step list_pub_tables +{ + SELECT relid::regclass AS tablename + FROM pg_get_publication_tables('pub_schema') + ORDER BY tablename; +} + +# Hold an ACCESS EXCLUSIVE lock on the table in one session, so that the query +# listing a publication's tables in another session blocks when it tries to +# open the locked table. Then drop the table in the same lock-holding session +# and commit, releasing the lock, so the query in another session resumes and +# skips the now-dropped table instead of erroring with "could not open relation +# with OID". +permutation lock list_pub_tables drop_and_commit diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 3442c4b9ec9..5fcdc7a131c 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -4202,6 +4202,7 @@ pthread_mutex_t pthread_once_t pthread_t ptrdiff_t +publication_tables_state published_rel pull_var_clause_context pull_varattnos_context -- 2.54.0