Subject: RE: Parallel Apply

From: "Zhijie Hou (Fujitsu)" <[email-masked]>
Date: Thu, 4 Jun 2026 06:58:31 +0000

On Wednesday, May 13, 2026 5:32 PM shveta malik <[email-masked]> wrote:
> 
> On Wed, May 13, 2026 at 10:54 AM shveta malik <[email-masked]>
> wrote:
> >
> >
> > Please find a few comments for patch003 mostly:

Thanks for the comments!

> >
> > 1)
> >   * depended on by other transactions. Entries are of type
> ParallelizedTxnEntry.
> >   *
> >   * dshash is used to enable dynamic shared memory allocation based on
> > the number
> > - * of transactions being applied in parallel.
> > + * of transactions being applied in parallel. Entries are of type
> > ParallelizedTxnEntry.
> >   */
> >  static dsa_area *parallel_apply_dsa_area = NULL;  static dshash_table
> > *parallelized_txns = NULL;
> >
> > 'Entries are of type ParallelizedTxnEntry' repeated twice in this comment.

Removed.

> >
> >
> > 2)
> > cleanup_committed_replica_identity_entries:
> >
> > + if (!skipped_write && !XLogRecPtrIsValid(pos->local_end))
> > + continue;
> >
> > Perhaps a comment will help to indicate above checks means a txn not
> > yet finished.

Added.

> >
> > 3)
> > Can you please clarify the scope, life-span of entries in
> > parallelized_txns vs ParallelApplyTxnHash. Both have remote-xid field.
> > So at any point of time, do both tables will have same number of
> > entries or if entries in one has bigger life-span/scope as compared to
> > other? It will be good to briefly mention these atop the hash-tables.

Added comments for it atop of these hash tables.

> >
> > 4)
> > +/*
> > + * Hash table storing replica identity values for changes being
> > +applied in
> > + * parallel, along with the last transaction that modified each row.
> > ...
> > +static replica_identity_hash *replica_identity_table = NULL;
> >
> > Regarding 'along with the last transaction that modified each row', is
> > 'remote_xid' in ReplicaIdentityEntry is the last transaction that
> > modified this row or the one which is currently modifying it?

It could be either, depending on whether the transaction is still running or
has already finished. In this version, I've changed it to be the "latest"
transaction.

> >
> > 5)
> > Since we have added comments for rest for the fields for below
> > existing structure, do you think we can update comment for 'xid' as
> > well to say it is remote-one. It does not mention it anywhere in
> > comment.
> >
> > typedef struct ParallelApplyWorkerEntry { TransactionId xid; /* Hash
> > key -- must be first */
> >

Added.

> >
> > 6)
> > 003' commit message says about RI table entry:
> >
> > Entries are deleted when transactions committed by parallel workers
> > are gathered, or the number of entries exceeds the limit.
> > --
> > I don't understand what do we mean by "when transactions committed by
> > parallel workers are gathered". Can we please make it more
> > clear/elaborate.
> >

I modified this to be consistent with some detailed comments in the patch.

> > ~~
> >
> > Reviewing further.
> >
> 
> Few more on 003:
> 
> 
> 7)
> I find check_and_record_ri_dependency() difficult to understand.
> 
> a)
> As an example, this part:
> 
> + /*
> + * Return if no entry exists, or if the current transaction was the
> + last one
> + * to modify the key.
> + */
> + if (!rientry || TransactionIdEquals(rientry->remote_xid,
> + new_depended_xid)) return;
> 
> IMO, the second check makes sense if new_depended_xid is valid. If so, why
> don't we make it part of previous 'if
> (TransactionIdIsValid(new_depended_xid))' logic. If 'found' was true, we can
> check it inside that if-block and 'return' from there instead of proceeding
> further.
> 
> Once we make above change, we can even move below logic inside previous
> 'if (TransactionIdIsValid(new_depended_xid))' block, as it looks strange that in
> previous if-block we are assigning InvalidTransactionId to 'rientry-
> >remote_xid' while we have valid new_depended_xid available there.
> 
> + /*
> + * Update the new depended xid into the entry if valid, the new xid
> + could
> + * be invalid if the transaction will be applied by the leader itself
> + * which means all the changes will be committed before processing next
> + * transaction, so no need to be depended on.
> + */
> + if (TransactionIdIsValid(new_depended_xid))
> + rientry->remote_xid = new_depended_xid;

Right. The original code was written that way to save a few lines,
but I agree it's harder to understand. I've refactored this part to
make each section clearer.

> 
> 
> b)
> Also this part is not clear:
> 
> + /*
> + * Return if RI key is NULL or is explicitly marked unchanged. The key
> + * value could be NULL in the new tuple of a update opertaion which
> + * means the RI key is not updated.
> + */
> + if (original_data->colstatus[i] == LOGICALREP_COLUMN_NULL ||
> + original_data->colstatus[i] == LOGICALREP_COLUMN_UNCHANGED)
> return;
> 
> Why do have we 'return' here when one of the columns is NULL or
> unchanged? What happens to rest of the RI columns? Which scenario may hit
> this? It needs more comments to explain the scenario.

After testing, I found that the handling for unchanged toasted columns was
incorrect. When the RI key includes two columns and only the non-toasted column
is changed, the toasted column is not included in the new tuple of an UPDATE
change. This means we cannot obtain the complete RI key value to check for
dependencies when handling the new change. I fixed this by copying the toasted
value from the old tuple to the new one before dependency checking.

> 
> 
> 8)
> check_and_record_ri_dependency() has this comment and logic around
> invalid remote_xid:
> 
> + /*
> + * Remove the entry if the transaction has been committed and no new
> + * dependency needs to be added.
> + */
> + else if (!TransactionIdIsValid(rientry->remote_xid))
> + {
> + free_replica_identity_key(rientry->keydata);
> + replica_identity_delete_item(replica_identity_table, rientry); }
> 
> While find_all_dependencies_on_rel() has this assert:
> + Assert(TransactionIdIsValid(rientry->remote_xid));
> 
> The first logic says that we may have Invalid remote_xid in existing entry in
> replica_identity_table while second logic has a sanity check while iterating
> the same hash-table that all entries must have valid remote_xid. Is the Assert
> correct? We might have Invalid remote_xid if txn is committed (done in
> check_and_append_xid_dependency).

It's correct. The remote_xid could be invalid only before being stored
in the table. The check_and_append_xid_dependency function wasn't very clear
about this, but after refactoring, I hope it's easier to understand now.

Here is the V19 patch set which addressed all the comments above.

Best Regards,
Hou zj

---

From: Peter Smith <[email-masked]>
Date: Tue, 9 Jun 2026 14:14:27 +1000

Hi Hou-San.

Some review comments for v19-0001 and v19-0002

//////////
v19-0001

======
.../replication/logical/applyparallelworker.c

1.
+/* An entry in the parallelized_txns shared hash table */
+typedef struct ParallelizedTxnEntry
+{
+ TransactionId xid; /* Hash key, remote transaction ID */
+} ParallelizedTxnEntry;
+
 /*
  * A hash table used to cache the state of streaming transactions being applied
- * by the parallel apply workers.
+ * by the parallel apply workers. Entries are of type ParallelApplyWorkerEntry.
  */
 static HTAB *ParallelApplyTxnHash = NULL;

+/*
+ * A hash table used to track the parallelized remote transactions
that could be
+ * depended on by other transactions. Entries are of type ParallelizedTxnEntry.
+ *
+ * dshash is used to enable dynamic shared memory allocation based on
the number
+ * of transactions being applied in parallel.
+ */
+static dsa_area *parallel_apply_dsa_area = NULL;
+static dshash_table *parallelized_txns = NULL;
+
+/* parameters for the parallelized_txns shared hash table */
+static const dshash_parameters dsh_params = {
+ sizeof(TransactionId),
+ sizeof(ParallelizedTxnEntry),
+ dshash_memcmp,
+ dshash_memhash,
+ dshash_memcpy,
+ LWTRANCHE_PARALLEL_APPLY_DSA
+};
+

1a.
Maybe that ParallelizedTxnEntry should be moved to just immediately
above 'dshash_parameters' because it seems to belong with that, and
currently it is splitting the ParallelApplyWorkerEntry from the
ParallelApplyTxnHash.

~

1b.
/parameters for/Parameters for/

~~~

pa_attach_parallelized_txn_hash:

2.
+ MemoryContext oldctx;
+
+ if (parallelized_txns)
+ {
+ Assert(parallel_apply_dsa_area);
+ *pa_dsa_handle = dsa_get_handle(parallel_apply_dsa_area);
+ *pa_dshash_handle = dshash_get_hash_table_handle(parallelized_txns);
+ return;
+ }
+
+ /* Be sure any local memory allocated by DSA routines is persistent. */
+ oldctx = MemoryContextSwitchTo(ApplyContext);
+
+ if (am_leader_apply_worker())
+ {
+ /* Initialize dynamic shared hash table for parallelized transactions */
+ parallel_apply_dsa_area = dsa_create(LWTRANCHE_PARALLEL_APPLY_DSA);
+ dsa_pin(parallel_apply_dsa_area);
+ dsa_pin_mapping(parallel_apply_dsa_area);
+ parallelized_txns = dshash_create(parallel_apply_dsa_area, &dsh_params, NULL);
+
+ /* Store handles in shared memory for other backends to use. */
+ *pa_dsa_handle = dsa_get_handle(parallel_apply_dsa_area);
+ *pa_dshash_handle = dshash_get_hash_table_handle(parallelized_txns);
+ }
+ else if (am_parallel_apply_worker())
+ {
+ /* Attach to existing dynamic shared hash table. */
+ parallel_apply_dsa_area =
dsa_attach(MyParallelShared->parallel_apply_dsa_handle);
+ dsa_pin_mapping(parallel_apply_dsa_area);
+ parallelized_txns = dshash_attach(parallel_apply_dsa_area, &dsh_params,
+   MyParallelShared->parallelized_txns_handle,
+   NULL);
+ }
+
+ MemoryContextSwitchTo(oldctx);

2a.
This might be easier to read if rearranged to use if/else instead of
having the early return.

SUGGESTION
if (parallelized_txns)
{
  /* Hashtable is already available */
  ...
}
else
{
  /* Create or attach... */

  MemoryContext oldctx = ...

  if (am_leader_apply_worker())
  {
    /* Create... */
    ...
  }
  else (am_parallel_apply_worker())
  {
    /* Attach... */
    ...
  }

  MemoryContextSwitchTo(oldctx);
}

~~~

2b.
Can it be anything other than
am_leader_apply_worker/am_parallel_apply_worker here? Should there be
an Assert?

~~~

2c.
Since the `dsh_params` are already set up, shouldn't this code be using them?

BEFORE
dsa_create(LWTRANCHE_PARALLEL_APPLY_DSA);

SUGGESTION
dsa_create(dsh_params.tranche_id);

//////////
v19-0002

======
.../replication/logical/applyparallelworker.

1.
+void
+pa_wait_for_depended_transaction(TransactionId xid)
+{
+ ParallelizedTxnEntry *txn_entry;

The declaration of `txn_entry` can be done later within the loop where
it is used.

======
Kind Regards,
Peter Smith.
Fujitsu Australia

---

From: "Zhijie Hou (Fujitsu)" <[email-masked]>
Date: Mon, 15 Jun 2026 11:20:12 +0000

> Some review comments for v19-0001 and v19-0002

Thanks for the comments.

> //////////
> v19-0001
> 
> 1a.
> Maybe that ParallelizedTxnEntry should be moved to just immediately above
> 'dshash_parameters' because it seems to belong with that, and currently it is
> splitting the ParallelApplyWorkerEntry from the ParallelApplyTxnHash.

Moved as suggested.

> /parameters for/Parameters for/

Fixed.

> This might be easier to read if rearranged to use if/else instead of having the
> early return.

I'm not in favor of this style, as deeply nested condition blocks are harder for
me to understand. Please see existing examples like init_dsm_registry,
initGlobalChannelTable, and logicalrep_launcher_attach_dshmem for the preferred
coding style.

> am_leader_apply_worker/am_parallel_apply_worker here? Should there be

Only the leader and parallel apply workers need to coordinate in this context,
so I didn't add an Assert. Even if other processes use it, it will be a no-op,
which is fine.

> dsa_create(dsh_params.tranche_id);

I think dsa_create and dsh_params are two different objects. The former is for
pure shared memory allocation, while the latter is part of a hash table.
Therefore, it's not necessary to pass dsh_params to dsa_create. We could even
use different LWTRANCHE for them if we wanted, but we don't have a special need
to create another one at the moment.

> +pa_wait_for_depended_transaction(TransactionId xid) {
> +ParallelizedTxnEntry *txn_entry;
> 
> The declaration of `txn_entry` can be done later within the loop where it is
> used.

Moved.

Attaching the new version patch set.

Apart from addressing above comments, I also did the following changes:

0003:

- Fix detection of dependencies on toasted columns by properly copying column
  status.
- Ensure leader correctly handles table-wide dependencies when applying a
  transaction directly.
- Prevent infinite loop by skipping self-dependencies when building the
  dependency list.
- Expand test coverage for dependency tracking.
- Improve code readability through refactoring.
- Enhance comments for better documentation.

0008:
0009:

- Complete most TODOs and finalize the design to cover more cases. Add extensive
  comments to explain the design.

- Add more tests to cover the new code paths.

Best Regards,
Hou zj

---

From: "Zhijie Hou (Fujitsu)" <[email-masked]>
Date: Thu, 9 Jul 2026 08:08:31 +0000

> - Add more tests to cover the new code paths.

I've further reviewed and improved the patch recently; here is the updated V21 patch
set with the following changes:

0003: Fix missing dependency check for replica identity FULL.
0004: Add more test coverage for replica identity FULL.
0007: Improve parallel apply safety check to handle concurrent table schema
      changes.
0009: Fix missing dependency check when both sides of a foreign key are
      partitioned tables, and add more test coverage for partitioned tables.

Best Regards,
Hou zj