Re: Perform streaming logical transactions by background workers and parallel apply

Below are some review comments for patch v14-0003:

========
v14-0003
========

3.1 Commit message

If any of the following checks are violated, an error will be reported.
1. The unique columns between publisher and subscriber are difference.
2. There is any non-immutable function present in expression in
subscriber's relation. Check from the following 4 items:
a. The function in triggers;
b. Column default value expressions and domain constraints;
c. Constraint expressions.
d. The foreign keys.

SUGGESTION (rewording to match the docs and the code).

Add some checks before using apply background worker to apply changes.
streaming=parallel mode has two requirements:
1) The unique columns must be the same between publisher and subscriber
2) There cannot be any non-immutable functions in the subscriber-side
replicated table. Look for functions in the following places:
* a. Trigger functions
* b. Column default value expressions and domain constraints
* c. Constraint expressions
* d. Foreign keys

======

3.2 doc/src/sgml/ref/create_subscription.sgml

+ To run in this mode, there are following two requirements. The first
+ is that the unique column should be the same between publisher and
+ subscriber; the second is that there should not be any non-immutable
+ function in subscriber-side replicated table.

SUGGESTION
Parallel mode has two requirements: 1) the unique columns must be the
same between publisher and subscriber; 2) there cannot be any
non-immutable functions in the subscriber-side replicated table.

======

3.3 .../replication/logical/applybgworker.c - apply_bgworker_relation_check

+ * Check if changes on this logical replication relation can be applied by
+ * apply background worker.

SUGGESTION
Check if changes on this relation can be applied by an apply background worker.

~~~

3.4

+ * Although we maintains the commit order by allowing only one process to
+ * commit at a time, our access order to the relation has changed.

SUGGESTION
Although the commit order is maintained only allowing one process to
commit at a time, the access order to the relation has changed.

~~~

3.5

+ /* Check only we are in apply bgworker. */
+ if (!am_apply_bgworker())
+ return;

SUGGESTION
/* Skip check if not an apply background worker. */

~~~

3.6

+ /*
+ * If it is a partitioned table, we do not check it, we will check its
+ * partition later.
+ */

This comment is lacking useful details:

/* Partition table checks are done later in (?????) */

~~~

3.7

+ if (!rel->sameunique)
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("cannot replicate relation with different unique index"),
+ errhint("Please change the streaming option to 'on' instead of
'parallel'.")));

Maybe the first message should change slightly so it is worded
consistently with the other one.

SUGGESTION
errmsg("cannot replicate relation. Unique indexes must be the same"),

======

3.8 src/backend/replication/logical/proto.c

-#define LOGICALREP_IS_REPLICA_IDENTITY 1
+#define LOGICALREP_IS_REPLICA_IDENTITY 0x0001
+#define LOGICALREP_IS_UNIQUE 0x0002

I think these constants should named differently to reflect that they
are just attribute flags. They should should use similar bitset styles
to the other nearby constants.

SUGGESTION
#define ATTR_IS_REPLICA_IDENTITY (1 << 0)
#define ATTR_IS_UNIQUE (1 << 1)

~~~

3.9 src/backend/replication/logical/proto.c - logicalrep_write_attrs

This big slab of new code to get the BMS looks very similar to
RelationGetIdentityKeyBitmap. So perhaps this code should be
encapsulated in another function like that one (in relcache.c?) and
then just called from logicalrep_write_attrs

======

3.10 src/backend/replication/logical/relation.c -
logicalrep_relmap_reset_volatility_cb

+/*
+ * Reset the flag volatility of all existing entry in the relation map cache.
+ */
+static void
+logicalrep_relmap_reset_volatility_cb(Datum arg, int cacheid, uint32 hashvalue)

SUGGESTION
Reset the volatility flag of all entries in the relation map cache.

~~~

3.11 src/backend/replication/logical/relation.c -
logicalrep_rel_mark_safe_in_apply_bgworker

+/*
+ * Check if unique index/constraint matches and mark sameunique and volatility
+ * flag.
+ *
+ * Don't throw any error here just mark the relation entry as not sameunique or
+ * FUNCTION_NONIMMUTABLE as we only check these in apply background worker.
+ */
+static void
+logicalrep_rel_mark_safe_in_apply_bgworker(LogicalRepRelMapEntry *entry)

SUGGESTION
Check if unique index/constraint matches and assign 'sameunique' flag.
Check if there are any non-immutable functions and assign the
'volatility' flag. Note: Don't throw any error here - these flags will
be checked in the apply background worker.

~~~

3.12 src/backend/replication/logical/relation.c -
logicalrep_rel_mark_safe_in_apply_bgworker

I did not really understand why you used an enum for one flag
(volatility) but not the other one (sameunique); shouldn’t both of
these be tri-values: unknown/yes/no?

For E.g. there is a quick exit from this function if the
FUNCTION_UNKNOWN, but there is no equivalent quick exit for the
sameunique? It seems inconsistent.

~~~

3.13 src/backend/replication/logical/relation.c -
logicalrep_rel_mark_safe_in_apply_bgworker

+ /*
+ * Check whether there is any non-immutable function in the local table.
+ *
+ * a. The function in triggers;
+ * b. Column default value expressions and domain constraints;
+ * c. Constraint expressions;
+ * d. Foreign keys.
+ */

SUGGESTION
* Check if there is any non-immutable function in the local table.
* Look for functions in the following places:
* a. trigger functions
* b. Column default value expressions and domain constraints
* c. Constraint expressions
* d. Foreign keys

~~~

3.14 src/backend/replication/logical/relation.c -
logicalrep_rel_mark_safe_in_apply_bgworker

There are lots of places setting FUNCTION_NONIMMUTABLE, so I think
this code might be tidier if you just have a single return at the end
of this function and 'goto' it.

e.g.
if (...)
goto function_not_immutable;

...

return;

function_not_immutable:
entry->volatility = FUNCTION_NONIMMUTABLE;
======

3.15 src/backend/replication/logical/worker.c - apply_handle_stream_stop

+ /*
+ * Unlike stream_commit, we don't need to wait here for stream_stop to
+ * finish. Allowing the other transaction to be applied before stream_stop
+ * is finished can only lead to failures if the unique index/constraint is
+ * different between publisher and subscriber. But for such cases, we don't
+ * allow streamed transactions to be applied in parallel. See
+ * apply_bgworker_relation_check.
+ */

"can only lead to failures" -> "can lead to failures"

~~~

3.16 src/backend/replication/logical/worker.c - apply_handle_tuple_routing

@@ -2534,13 +2548,14 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
}
MemoryContextSwitchTo(oldctx);

+ part_entry = logicalrep_partition_open(relmapentry, partrel,
+ attrmap);
+
+ apply_bgworker_relation_check(part_entry);
+
/* Check if we can do the update or delete on the leaf partition. */
if (operation == CMD_UPDATE || operation == CMD_DELETE)
- {
- part_entry = logicalrep_partition_open(relmapentry, partrel,
- attrmap);
check_relation_updatable(part_entry);
- }

Perhaps the apply_bgworker_relation_check(part_entry); should be done
AFTER the CMD_UPDATE/CMD_DELETE check because then it will not change
the existing errors for those cases.

======

3.17 src/backend/utils/cache/typcache.c

+/*
+ * GetDomainConstraints --- get DomainConstraintState list of
specified domain type
+ */
+List *
+GetDomainConstraints(Oid type_id)

This is an unusual-looking function header comment, with the function
name and the "---".

======

3.18 src/include/replication/logicalrelation.h

+/*
+ * States to determine volatility of the function in expressions in one
+ * relation.
+ */
+typedef enum RelFuncVolatility
+{
+ FUNCTION_UNKNOWN = 0, /* initializing */
+ FUNCTION_IMMUTABLE, /* all functions are immutable function */
+ FUNCTION_NONIMMUTABLE /* at least one non-immutable function */
+} RelFuncVolatility;
+

I think the comments can be improved, and also the values can be more
self-explanatory. e.g.

typedef enum RelFuncVolatility
{
FUNCTION_UNKNOWN_IMMUATABLE, /* unknown */
FUNCTION_ALL_MUTABLE, /* all functions are immutable */
FUNCTION_NOT_ALL_IMMUTABLE /* not all functions are immuatble */
} RelFuncVolatility;

~~~

3.18

RelFuncVolatility should be added to typedefs.list

~~~

3.19

@@ -31,6 +42,11 @@ typedef struct LogicalRepRelMapEntry
Relation localrel; /* relcache entry (NULL when closed) */
AttrMap *attrmap; /* map of local attributes to remote ones */
bool updatable; /* Can apply updates/deletes? */
+ bool sameunique; /* Are all unique columns of the local
+ relation contained by the unique columns in
+ remote? */

(This is similar to review comment 3.12)

I felt it was inconsistent for this to be a boolean but for the
'volatility' member to be an enum. AFAIK these 2 flags are similar
kinds – e.g. essentially tri-state flags unknown/true/false so I
thought they should be treated the same. E.g. both enums?

~~~

3.20

+ RelFuncVolatility volatility; /* all functions in localrel are
+ immutable function? */

SUGGESTION
/* Indicator of local relation function volatility */

======

3.21 .../subscription/t/022_twophase_cascade.pl

+ if ($streaming_mode eq 'parallel')
+ {
+ $node_C->safe_psql(
+ 'postgres', "
+ ALTER TABLE test_tab ALTER c DROP DEFAULT");
+ }
+

Indentation of the ALTER does not seem right.

======

3.22 .../subscription/t/032_streaming_apply.pl

3.22.a
+# Setup structure on publisher

"structure"?

3.22.b
+# Setup structure on subscriber

"structure"?

~~~

3.23

+# Check that a background worker starts if "streaming" option is specified as
+# "parallel". We have to look for the DEBUG1 log messages about that, so
+# temporarily bump up the log verbosity.
+$node_subscriber->append_conf('postgresql.conf', "log_min_messages = debug1");
+$node_subscriber->reload;
+
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(1,
5000) s(i)"
+);
+
+$node_subscriber->wait_for_log(qr/\[Apply BGW #\d+\] started/, 0);
+$node_subscriber->append_conf('postgresql.conf',
+ "log_min_messages = warning");
+$node_subscriber->reload;

I didn't really think it was necessary to bump this log level, and to
verify that the bgworker is started, because this test is anyway going
to ensure that the ERROR "cannot replicate relation with different
unique index" happens, so that is already implicitly ensuring the
bgworker was used.

~~~

3.24

+# Then we check the unique index on partition table.
+$node_subscriber->safe_psql(
+ 'postgres', qq{
+CREATE TRIGGER insert_trig
+BEFORE INSERT ON test_tab_partition
+FOR EACH ROW EXECUTE PROCEDURE trigger_func();
+ALTER TABLE test_tab_partition ENABLE REPLICA TRIGGER insert_trig;
+});

Looks like the wrong comment. I think it should say something like
"Check the trigger on the partition table."

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