}
/*
- * Make sure that we started local transaction.
+ * Begin one step (one INSERT, UPDATE, etc) of a replication transaction.
*
- * Also switches to ApplyMessageContext as necessary.
+ * Start a transaction, if this is the first step (else we keep using the
+ * existing transaction).
+ * Also provide a global snapshot and ensure we run in ApplyMessageContext.
*/
-static bool
-ensure_transaction(void)
+static void
+begin_replication_step(void)
{
- if (IsTransactionState())
- {
- SetCurrentStatementStartTimestamp();
-
- if (CurrentMemoryContext != ApplyMessageContext)
- MemoryContextSwitchTo(ApplyMessageContext);
+ SetCurrentStatementStartTimestamp();
- return false;
+ if (!IsTransactionState())
+ {
+ StartTransactionCommand();
+ maybe_reread_subscription();
}
- SetCurrentStatementStartTimestamp();
- StartTransactionCommand();
-
- maybe_reread_subscription();
+ PushActiveSnapshot(GetTransactionSnapshot());
MemoryContextSwitchTo(ApplyMessageContext);
- return true;
+}
+
+/*
+ * Finish up one step of a replication transaction.
+ * Callers of begin_replication_step() must also call this.
+ *
+ * We don't close out the transaction here, but we should increment
+ * the command counter to make the effects of this step visible.
+ */
+static void
+end_replication_step(void)
+{
+ PopActiveSnapshot();
+
+ CommandCounterIncrement();
}
ResultRelInfo *resultRelInfo;
RangeTblEntry *rte;
- /*
- * Input functions may need an active snapshot, as may AFTER triggers
- * invoked during finish_estate. For safety, ensure an active snapshot
- * exists throughout all our usage of the executor.
- */
- PushActiveSnapshot(GetTransactionSnapshot());
-
estate = CreateExecutorState();
rte = makeNode(RangeTblEntry);
/* Cleanup. */
ExecResetTupleTable(estate->es_tupleTable, false);
FreeExecutorState(estate);
- PopActiveSnapshot();
}
/*
TupleTableSlot *remoteslot;
MemoryContext oldctx;
- ensure_transaction();
+ begin_replication_step();
relid = logicalrep_read_insert(s, &newtup);
rel = logicalrep_rel_open(relid, RowExclusiveLock);
* transaction so it's safe to unlock it.
*/
logicalrep_rel_close(rel, RowExclusiveLock);
+ end_replication_step();
return;
}
logicalrep_rel_close(rel, NoLock);
- CommandCounterIncrement();
+ end_replication_step();
}
/*
bool found;
MemoryContext oldctx;
- ensure_transaction();
+ begin_replication_step();
relid = logicalrep_read_update(s, &has_oldtup, &oldtup,
&newtup);
* transaction so it's safe to unlock it.
*/
logicalrep_rel_close(rel, RowExclusiveLock);
+ end_replication_step();
return;
}
logicalrep_rel_close(rel, NoLock);
- CommandCounterIncrement();
+ end_replication_step();
}
/*
bool found;
MemoryContext oldctx;
- ensure_transaction();
+ begin_replication_step();
relid = logicalrep_read_delete(s, &oldtup);
rel = logicalrep_rel_open(relid, RowExclusiveLock);
* transaction so it's safe to unlock it.
*/
logicalrep_rel_close(rel, RowExclusiveLock);
+ end_replication_step();
return;
}
logicalrep_rel_close(rel, NoLock);
- CommandCounterIncrement();
+ end_replication_step();
}
/*
ListCell *lc;
LOCKMODE lockmode = AccessExclusiveLock;
- ensure_transaction();
+ begin_replication_step();
remote_relids = logicalrep_read_truncate(s, &cascade, &restart_seqs);
logicalrep_rel_close(rel, NoLock);
}
- CommandCounterIncrement();
+ end_replication_step();
}