diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c index 0602398a545..0b21076dbc4 100644 --- a/src/backend/catalog/pg_publication.c +++ b/src/backend/catalog/pg_publication.c @@ -1052,14 +1052,27 @@ Datum pg_get_publication_tables(PG_FUNCTION_ARGS) { #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; ArrayType *arr; Datum *elems; int nelems, @@ -1158,26 +1171,47 @@ pg_get_publication_tables(PG_FUNCTION_ARGS) PG_NODE_TREEOID, -1, 0); funcctx->tuple_desc = BlessTupleDesc(tupdesc); - funcctx->user_fctx = (void *) 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. */ @@ -1191,6 +1225,7 @@ pg_get_publication_tables(PG_FUNCTION_ARGS) * 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), @@ -1220,7 +1255,6 @@ pg_get_publication_tables(PG_FUNCTION_ARGS) /* 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); @@ -1243,10 +1277,10 @@ pg_get_publication_tables(PG_FUNCTION_ARGS) 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 7a4ab2ce805..ac1b5e6de5f 100644 --- a/src/test/isolation/isolation_schedule +++ b/src/test/isolation/isolation_schedule @@ -117,3 +117,4 @@ test: serializable-parallel-3 test: matview-write-skew test: lock-nowait test: ddl-dependency-locking +test: pub-concurrent-drop 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 e5634ae2969..86b194da8e3 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -3805,6 +3805,7 @@ pthread_mutex_t pthread_once_t pthread_t ptrdiff_t +publication_tables_state published_rel pull_var_clause_context pull_varattnos_context