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.

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 <bharath.rupireddyforpostgres@gmail.com>
Reviewed-by: Bertrand Drouvot <bertranddrouvot.pg@gmail.com>
Reviewed-by: shveta malik <shveta.malik@gmail.com>
Reviewed-by: Ajin Cherian <itsajin@gmail.com>
Reviewed-by: Masahiko Sawada <sawada.mshk@gmail.com>
Reviewed-by: Chao Li <li.evan.chao@gmail.com>
Discussion: https://www.postgresql.org/message-id/CALj2ACVYYooWH-5tJ6cPKkU%2BmutVxwb_z4S%2BqAi-zdrFqxXE2Q%40mail.gmail.com
Backpatch-through: 16

diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c
index 1ec94c851b2f3a06e41ef44fed7252d1050e9f56..ea28ec319c56811abee1da8bf6624f9e02150a31 100644 (file)
--- a/src/backend/catalog/pg_publication.c
+++ b/src/backend/catalog/pg_publication.c
@@ -1424,14 +1424,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;
@@ -1554,26 +1567,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.
                 */
@@ -1587,6 +1621,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),
@@ -1616,7 +1651,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);
@@ -1653,10 +1687,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 (file)
index 0000000..8360af0
--- /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;
+ <waiting ...>
+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 b8ebe92553c54d56bc6faf72bba96b4aa3eceb4b..26abed9f9f072b982ead03590c1915b8d79554ec 100644 (file)
--- a/src/test/isolation/isolation_schedule
+++ b/src/test/isolation/isolation_schedule
@@ -128,3 +128,4 @@ test: matview-write-skew
 test: lock-nowait
 test: for-portion-of
 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 (file)
index 0000000..4f7d701
--- /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 56c1f997f88b1002fd03a1bcdd74bb42314d4182..85d989f395d417c7e33168d45d05a1dbecd54c88 100644 (file)
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -4209,6 +4209,7 @@ pthread_mutex_t
 pthread_once_t
 pthread_t
 ptrdiff_t
+publication_tables_state
 published_rel
 pull_var_clause_context
 pull_varattnos_context