Fix updating of pg_subscription_rel from workers

A logical replication worker should not insert new rows into
pg_subscription_rel, only update existing rows, so that there are no
races if a concurrent refresh removes rows.  Adjust the API to be able
to choose that behavior.

Author: Masahiko Sawada <sawada.mshk@gmail.com>
Reported-by: tushar <tushar.ahuja@enterprisedb.com>

diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c
index ab5f3719fc397fdaba5f4f7fefc7dc7fd54b2a35..c5b2541319ee8fa13e01cd1d368e409316c64c74 100644 (file)
--- a/src/backend/catalog/pg_subscription.c
+++ b/src/backend/catalog/pg_subscription.c
@@ -227,17 +227,22 @@ textarray_to_stringlist(ArrayType *textarray)
 /*
  * Set the state of a subscription table.
  *
+ * If update_only is true and the record for given table doesn't exist, do
+ * nothing.  This can be used to avoid inserting a new record that was deleted
+ * by someone else.  Generally, subscription DDL commands should use false,
+ * workers should use true.
+ *
  * The insert-or-update logic in this function is not concurrency safe so it
  * might raise an error in rare circumstances.  But if we took a stronger lock
  * such as ShareRowExclusiveLock, we would risk more deadlocks.
  */
 Oid
 SetSubscriptionRelState(Oid subid, Oid relid, char state,
-                       XLogRecPtr sublsn)
+                       XLogRecPtr sublsn, bool update_only)
 {
    Relation    rel;
    HeapTuple   tup;
-   Oid         subrelid;
+   Oid         subrelid = InvalidOid;
    bool        nulls[Natts_pg_subscription_rel];
    Datum       values[Natts_pg_subscription_rel];
 
@@ -252,7 +257,7 @@ SetSubscriptionRelState(Oid subid, Oid relid, char state,
     * If the record for given table does not exist yet create new record,
     * otherwise update the existing one.
     */
-   if (!HeapTupleIsValid(tup))
+   if (!HeapTupleIsValid(tup) && !update_only)
    {
        /* Form the tuple. */
        memset(values, 0, sizeof(values));
@@ -272,7 +277,7 @@ SetSubscriptionRelState(Oid subid, Oid relid, char state,
 
        heap_freetuple(tup);
    }
-   else
+   else if (HeapTupleIsValid(tup))
    {
        bool        replaces[Natts_pg_subscription_rel];
 

diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index ad98b38efe8adc6d899f319dc171fd0eb1ac7959..49737a904207ed8c000e092ee9d95ee0b76b78bb 100644 (file)
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -451,7 +451,7 @@ CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel)
                                         rv->schemaname, rv->relname);
 
                SetSubscriptionRelState(subid, relid, table_state,
-                                       InvalidXLogRecPtr);
+                                       InvalidXLogRecPtr, false);
            }
 
            ereport(NOTICE,
@@ -574,7 +574,7 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data)
        {
            SetSubscriptionRelState(sub->oid, relid,
                          copy_data ? SUBREL_STATE_INIT : SUBREL_STATE_READY,
-                                   InvalidXLogRecPtr);
+                                   InvalidXLogRecPtr, false);
            ereport(NOTICE,
                    (errmsg("added subscription for table %s.%s",
                            quote_identifier(rv->schemaname),

diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 6fe39d20237d3956819b3da64ffc997df21cd5bc..f57ae6ee2d560abc048abad2b7b345aff600c955 100644 (file)
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -287,7 +287,8 @@ process_syncing_tables_for_sync(XLogRecPtr current_lsn)
        SetSubscriptionRelState(MyLogicalRepWorker->subid,
                                MyLogicalRepWorker->relid,
                                MyLogicalRepWorker->relstate,
-                               MyLogicalRepWorker->relstate_lsn);
+                               MyLogicalRepWorker->relstate_lsn,
+                               true);
 
        walrcv_endstreaming(wrconn, &tli);
        finish_sync_worker();
@@ -414,7 +415,7 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
                }
                SetSubscriptionRelState(MyLogicalRepWorker->subid,
                                        rstate->relid, rstate->state,
-                                       rstate->lsn);
+                                       rstate->lsn, true);
            }
        }
        else
@@ -845,7 +846,8 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
                SetSubscriptionRelState(MyLogicalRepWorker->subid,
                                        MyLogicalRepWorker->relid,
                                        MyLogicalRepWorker->relstate,
-                                       MyLogicalRepWorker->relstate_lsn);
+                                       MyLogicalRepWorker->relstate_lsn,
+                                       true);
                CommitTransactionCommand();
                pgstat_report_stat(false);
 
@@ -932,7 +934,8 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
                    SetSubscriptionRelState(MyLogicalRepWorker->subid,
                                            MyLogicalRepWorker->relid,
                                            SUBREL_STATE_SYNCDONE,
-                                           *origin_startpos);
+                                           *origin_startpos,
+                                           true);
                    finish_sync_worker();
                }
                break;

diff --git a/src/include/catalog/pg_subscription_rel.h b/src/include/catalog/pg_subscription_rel.h
index 391f96b76e4f4110e90336a6572d261fdab6d352..f5f6191676884af40a21070741e4120a8517455c 100644 (file)
--- a/src/include/catalog/pg_subscription_rel.h
+++ b/src/include/catalog/pg_subscription_rel.h
@@ -71,7 +71,7 @@ typedef struct SubscriptionRelState
 } SubscriptionRelState;
 
 extern Oid SetSubscriptionRelState(Oid subid, Oid relid, char state,
-                       XLogRecPtr sublsn);
+                       XLogRecPtr sublsn, bool update_only);
 extern char GetSubscriptionRelState(Oid subid, Oid relid,
                        XLogRecPtr *sublsn, bool missing_ok);
 extern void RemoveSubscriptionRel(Oid subid, Oid relid);