Split the SetSubscriptionRelState function into two

We don't actually need the insert-or-update logic, so it's clearer to
have separate functions for the inserting and updating.

Author: Petr Jelinek <petr.jelinek@2ndquadrant.com>
Reviewed-by: Masahiko Sawada <sawada.mshk@gmail.com>

diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c
index 8e16d3b7bcec3239fb151f0038e23966a1c92a87..8705d8b1d36d536e3ee33266b504a890f3208896 100644 (file)
--- a/src/backend/catalog/pg_subscription.c
+++ b/src/backend/catalog/pg_subscription.c
@@ -227,24 +227,15 @@ 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.
+ * Add new state record for a subscription table.
  */
 Oid
-SetSubscriptionRelState(Oid subid, Oid relid, char state,
-                       XLogRecPtr sublsn, bool update_only)
+AddSubscriptionRelState(Oid subid, Oid relid, char state,
+                       XLogRecPtr sublsn)
 {
    Relation    rel;
    HeapTuple   tup;
-   Oid         subrelid = InvalidOid;
+   Oid         subrelid;
    bool        nulls[Natts_pg_subscription_rel];
    Datum       values[Natts_pg_subscription_rel];
 
@@ -256,57 +247,81 @@ SetSubscriptionRelState(Oid subid, Oid relid, char state,
    tup = SearchSysCacheCopy2(SUBSCRIPTIONRELMAP,
                              ObjectIdGetDatum(relid),
                              ObjectIdGetDatum(subid));
+   if (HeapTupleIsValid(tup))
+       elog(ERROR, "subscription table %u in subscription %u already exists",
+            relid, subid);
 
-   /*
-    * If the record for given table does not exist yet create new record,
-    * otherwise update the existing one.
-    */
-   if (!HeapTupleIsValid(tup) && !update_only)
-   {
-       /* Form the tuple. */
-       memset(values, 0, sizeof(values));
-       memset(nulls, false, sizeof(nulls));
-       values[Anum_pg_subscription_rel_srsubid - 1] = ObjectIdGetDatum(subid);
-       values[Anum_pg_subscription_rel_srrelid - 1] = ObjectIdGetDatum(relid);
-       values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
-       if (sublsn != InvalidXLogRecPtr)
-           values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
-       else
-           nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
-
-       tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
-
-       /* Insert tuple into catalog. */
-       subrelid = CatalogTupleInsert(rel, tup);
-
-       heap_freetuple(tup);
-   }
-   else if (HeapTupleIsValid(tup))
-   {
-       bool        replaces[Natts_pg_subscription_rel];
+   /* Form the tuple. */
+   memset(values, 0, sizeof(values));
+   memset(nulls, false, sizeof(nulls));
+   values[Anum_pg_subscription_rel_srsubid - 1] = ObjectIdGetDatum(subid);
+   values[Anum_pg_subscription_rel_srrelid - 1] = ObjectIdGetDatum(relid);
+   values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
+   if (sublsn != InvalidXLogRecPtr)
+       values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
+   else
+       nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
 
-       /* Update the tuple. */
-       memset(values, 0, sizeof(values));
-       memset(nulls, false, sizeof(nulls));
-       memset(replaces, false, sizeof(replaces));
+   tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
 
-       replaces[Anum_pg_subscription_rel_srsubstate - 1] = true;
-       values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
+   /* Insert tuple into catalog. */
+   subrelid = CatalogTupleInsert(rel, tup);
 
-       replaces[Anum_pg_subscription_rel_srsublsn - 1] = true;
-       if (sublsn != InvalidXLogRecPtr)
-           values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
-       else
-           nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
+   heap_freetuple(tup);
 
-       tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
-                               replaces);
+   /* Cleanup. */
+   heap_close(rel, NoLock);
 
-       /* Update the catalog. */
-       CatalogTupleUpdate(rel, &tup->t_self, tup);
+   return subrelid;
+}
 
-       subrelid = HeapTupleGetOid(tup);
-   }
+/*
+ * Update the state of a subscription table.
+ */
+Oid
+UpdateSubscriptionRelState(Oid subid, Oid relid, char state,
+                          XLogRecPtr sublsn)
+{
+   Relation    rel;
+   HeapTuple   tup;
+   Oid         subrelid;
+   bool        nulls[Natts_pg_subscription_rel];
+   Datum       values[Natts_pg_subscription_rel];
+   bool        replaces[Natts_pg_subscription_rel];
+
+   LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
+
+   rel = heap_open(SubscriptionRelRelationId, RowExclusiveLock);
+
+   /* Try finding existing mapping. */
+   tup = SearchSysCacheCopy2(SUBSCRIPTIONRELMAP,
+                             ObjectIdGetDatum(relid),
+                             ObjectIdGetDatum(subid));
+   if (!HeapTupleIsValid(tup))
+       elog(ERROR, "subscription table %u in subscription %u does not exist",
+            relid, subid);
+
+   /* Update the tuple. */
+   memset(values, 0, sizeof(values));
+   memset(nulls, false, sizeof(nulls));
+   memset(replaces, false, sizeof(replaces));
+
+   replaces[Anum_pg_subscription_rel_srsubstate - 1] = true;
+   values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
+
+   replaces[Anum_pg_subscription_rel_srsublsn - 1] = true;
+   if (sublsn != InvalidXLogRecPtr)
+       values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
+   else
+       nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
+
+   tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
+                           replaces);
+
+   /* Update the catalog. */
+   CatalogTupleUpdate(rel, &tup->t_self, tup);
+
+   subrelid = HeapTupleGetOid(tup);
 
    /* Cleanup. */
    heap_close(rel, NoLock);

diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 2694e1b2d746ff714b25fdcbd8bc38fd3717b31c..f138e61a8d3d9d111e52b5fe59742df1967e9042 100644 (file)
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -450,8 +450,8 @@ CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel)
                CheckSubscriptionRelkind(get_rel_relkind(relid),
                                         rv->schemaname, rv->relname);
 
-               SetSubscriptionRelState(subid, relid, table_state,
-                                       InvalidXLogRecPtr, false);
+               AddSubscriptionRelState(subid, relid, table_state,
+                                       InvalidXLogRecPtr);
            }
 
            /*
@@ -569,9 +569,9 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data)
        if (!bsearch(&relid, subrel_local_oids,
                     list_length(subrel_states), sizeof(Oid), oid_cmp))
        {
-           SetSubscriptionRelState(sub->oid, relid,
+           AddSubscriptionRelState(sub->oid, relid,
                                    copy_data ? SUBREL_STATE_INIT : SUBREL_STATE_READY,
-                                   InvalidXLogRecPtr, false);
+                                   InvalidXLogRecPtr);
            ereport(DEBUG1,
                    (errmsg("table \"%s.%s\" added to subscription \"%s\"",
                            rv->schemaname, rv->relname, sub->name)));

diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index e50b9f790565b912594995b1189fe0573f2fb575..acc6498567d07c3e93d0adf5039a10c693e38f64 100644 (file)
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -298,11 +298,10 @@ process_syncing_tables_for_sync(XLogRecPtr current_lsn)
 
        SpinLockRelease(&MyLogicalRepWorker->relmutex);
 
-       SetSubscriptionRelState(MyLogicalRepWorker->subid,
-                               MyLogicalRepWorker->relid,
-                               MyLogicalRepWorker->relstate,
-                               MyLogicalRepWorker->relstate_lsn,
-                               true);
+       UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
+                                  MyLogicalRepWorker->relid,
+                                  MyLogicalRepWorker->relstate,
+                                  MyLogicalRepWorker->relstate_lsn);
 
        walrcv_endstreaming(wrconn, &tli);
        finish_sync_worker();
@@ -427,9 +426,10 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
                    StartTransactionCommand();
                    started_tx = true;
                }
-               SetSubscriptionRelState(MyLogicalRepWorker->subid,
-                                       rstate->relid, rstate->state,
-                                       rstate->lsn, true);
+
+               UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
+                                          rstate->relid, rstate->state,
+                                          rstate->lsn);
            }
        }
        else
@@ -870,11 +870,10 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
 
                /* Update the state and make it visible to others. */
                StartTransactionCommand();
-               SetSubscriptionRelState(MyLogicalRepWorker->subid,
-                                       MyLogicalRepWorker->relid,
-                                       MyLogicalRepWorker->relstate,
-                                       MyLogicalRepWorker->relstate_lsn,
-                                       true);
+               UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
+                                          MyLogicalRepWorker->relid,
+                                          MyLogicalRepWorker->relstate,
+                                          MyLogicalRepWorker->relstate_lsn);
                CommitTransactionCommand();
                pgstat_report_stat(false);
 
@@ -961,11 +960,10 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
                     * Update the new state in catalog.  No need to bother
                     * with the shmem state as we are exiting for good.
                     */
-                   SetSubscriptionRelState(MyLogicalRepWorker->subid,
-                                           MyLogicalRepWorker->relid,
-                                           SUBREL_STATE_SYNCDONE,
-                                           *origin_startpos,
-                                           true);
+                   UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
+                                              MyLogicalRepWorker->relid,
+                                              SUBREL_STATE_SYNCDONE,
+                                              *origin_startpos);
                    finish_sync_worker();
                }
                break;

diff --git a/src/include/catalog/pg_subscription_rel.h b/src/include/catalog/pg_subscription_rel.h
index d936973a9da091f768037fa1f4e543daf500356f..5cf268f181451e894be293e1feeaeb7f699db430 100644 (file)
--- a/src/include/catalog/pg_subscription_rel.h
+++ b/src/include/catalog/pg_subscription_rel.h
@@ -67,8 +67,10 @@ typedef struct SubscriptionRelState
    char        state;
 } SubscriptionRelState;
 
-extern Oid SetSubscriptionRelState(Oid subid, Oid relid, char state,
-                       XLogRecPtr sublsn, bool update_only);
+extern Oid AddSubscriptionRelState(Oid subid, Oid relid, char state,
+                       XLogRecPtr sublsn);
+extern Oid UpdateSubscriptionRelState(Oid subid, Oid relid, char state,
+                          XLogRecPtr sublsn);
 extern char GetSubscriptionRelState(Oid subid, Oid relid,
                        XLogRecPtr *sublsn, bool missing_ok);
 extern void RemoveSubscriptionRel(Oid subid, Oid relid);