Re: Proposal: Filter irrelevant change before reassemble transactions during logical decoding

On Mon, 17 Feb 2025 at 16:20, Ajin Cherian <itsajin(at)gmail(dot)com> wrote:
>
> On Mon, Feb 17, 2025 at 10:08 AM Peter Smith <smithpb2250(at)gmail(dot)com> wrote:
> >
> > Hi Ajin.
> >
> > FYI - Patch set v13* no longer applies cleanly. Needs rebasing.
> >
> I've rebased the patch. I've also merged patch 1 into patch 2 as the
> functionality of the changes in patch 1 are only n patch 2.
>
> On Mon, Feb 17, 2025 at 10:08 AM Peter Smith <smithpb2250(at)gmail(dot)com> wrote:
> >
> > Hi Ajin.
> >
> > FYI - Patch set v13* no longer applies cleanly. Needs rebasing.
> >
> I've rebased the patch. I've also merged patch 1 into patch 2 as the
> functionality of the changes in patch 1 are only in patch 2. So only 2
> patches in this version.
>
>
> On Fri, Feb 14, 2025 at 6:18 PM Amit Kapila <amit(dot)kapila16(at)gmail(dot)com> wrote:
> >
> > Again, can you share the link of performance data for this overhead
> > and if you have not published then please share it and also mention it
> > in commit message?
>
> I will run the performance tests and post an update with the results.
>
> > Can we try different thresholds for this like 10, 50, 100, 200, etc.
> > to decide what is a good threshold value to skip filtering changes?
> >
> Ok, will do this. For this patch, I reset the count to 0, else the
> test case fails as filtering could be skipped due to throttling. I
> think we need a way for the user to set this threshold via a GUC and
> that can be used for testing.
>

Hi Ajin,

I have reviewed the v14-0001 patch, I have following comment:

1. We need to initialize the 'rb->unfiltered_changes_count = 0', initially
inside function 'ReorderBufferAllocate', otherwise it will get set to garbage
value. While debugging I found out it was setting to 128 by default, so
'++rb->unfiltered_changes_count' was setting it to '-127'. As a result it
might not enter the if condition here for some initial inserts:

+ /*
+ * If filtering was suspended and we've crossed the change threshold,
+ * attempt to filter again
+ */
+ if (!rb->can_filter_change && (++rb->unfiltered_changes_count
+ >= CHANGES_THRESHOLD_FOR_FILTER))
+ {
+ rb->can_filter_change = true;
+ rb->unfiltered_changes_count = 0;
+ }

2. After applying only 0001 patch, then I tried to insert in a published table.
I am getting following error:

postgres=# insert into s1.t1 values(15);
INSERT 0 1
postgres=# insert into s1.t1 values(15);
WARNING: terminating connection because of crash of another server process
DETAIL: The postmaster has commanded this server process to roll back
the current transaction and exit, because another server process
exited abnormally and possibly corrupted shared memory.
HINT: In a moment you should be able to reconnect to the database and
repeat your command.
server closed the connection unexpectedly
This probably means the server terminated abnormally
before or while processing the request.
postgres=?#

Stack trace:
#0 __pthread_kill_implementation (no_tid=0, signo=6,
threadid=132913961858944) at ./nptl/pthread_kill.c:44
#1 __pthread_kill_internal (signo=6, threadid=132913961858944) at
./nptl/pthread_kill.c:78
#2 __GI___pthread_kill (threadid=132913961858944,
signo=signo(at)entry=6) at ./nptl/pthread_kill.c:89
#3 0x000078e270c42476 in __GI_raise (sig=sig(at)entry=6) at
../sysdeps/posix/raise.c:26
#4 0x000078e270c287f3 in __GI_abort () at ./stdlib/abort.c:79
#5 0x00005c9c56842f53 in ExceptionalCondition
(conditionName=0x5c9c56c10ed5 "MemoryContextIsValid(context)",
fileName=0x5c9c56c10d5f "mcxt.c",
lineNumber=1323) at assert.c:66
#6 0x00005c9c5690e99c in palloc (size=48) at mcxt.c:1323
#7 0x00005c9c55277564 in systable_beginscan
(heapRelation=0x78e2716561d8, indexId=2685, indexOK=true,
snapshot=0x0, nkeys=1, key=0x7ffd9a4446b0)
at genam.c:403
#8 0x00005c9c567cb6ca in SearchCatCacheMiss (cache=0x5c9c58af0e80,
nkeys=1, hashValue=1393055558, hashIndex=6, v1=16384, v2=0, v3=0,
v4=0)
at catcache.c:1533
#9 0x00005c9c567cb43d in SearchCatCacheInternal
(cache=0x5c9c58af0e80, nkeys=1, v1=16384, v2=0, v3=0, v4=0) at
catcache.c:1464
#10 0x00005c9c567cabd8 in SearchCatCache1 (cache=0x5c9c58af0e80,
v1=16384) at catcache.c:1332
#11 0x00005c9c5682a055 in SearchSysCache1 (cacheId=38, key1=16384) at
syscache.c:228
#12 0x00005c9c567e3b6f in get_namespace_name (nspid=16384) at lsyscache.c:3397
#13 0x00005c9c5608eb5e in logicalrep_write_namespace
(out=0x5c9c58b5d5c8, nspid=16384) at proto.c:1033
#14 0x00005c9c5608d0c2 in logicalrep_write_rel (out=0x5c9c58b5d5c8,
xid=0, rel=0x78e27167ede0, columns=0x0,
include_gencols_type=PUBLISH_GENCOLS_NONE) at proto.c:683
#15 0x000078e270fc9301 in send_relation_and_attrs
(relation=0x78e27167ede0, xid=0, ctx=0x5c9c58b4d8f0,
relentry=0x5c9c58b92278) at pgoutput.c:798
#16 0x000078e270fc8ed9 in maybe_send_schema (ctx=0x5c9c58b4d8f0,
change=0x5c9c58b8f2e0, relation=0x78e27167ede0,
relentry=0x5c9c58b92278)
at pgoutput.c:752
#17 0x000078e270fce338 in pgoutput_change (ctx=0x5c9c58b4d8f0,
txn=0x5c9c58b8b2c0, relation=0x78e27167ede0, change=0x5c9c58b8f2e0)
at pgoutput.c:1572
#18 0x00005c9c5607b265 in change_cb_wrapper (cache=0x5c9c58b4f900,
txn=0x5c9c58b8b2c0, relation=0x78e27167ede0, change=0x5c9c58b8f2e0)
at logical.c:1116
#19 0x00005c9c560a2cff in ReorderBufferApplyChange (rb=0x5c9c58b4f900,
txn=0x5c9c58b8b2c0, relation=0x78e27167ede0, change=0x5c9c58b8f2e0,
streaming=false) at reorderbuffer.c:2175
#20 0x00005c9c560a4856 in ReorderBufferProcessTXN (rb=0x5c9c58b4f900,
txn=0x5c9c58b8b2c0, commit_lsn=8380824, snapshot_now=0x5c9c58b736b8,
command_id=0, streaming=false) at reorderbuffer.c:2511
#21 0x00005c9c560a6aef in ReorderBufferReplay (txn=0x5c9c58b8b2c0,
rb=0x5c9c58b4f900, xid=752, commit_lsn=8380824, end_lsn=8380872,
commit_time=793368288662227, origin_id=0, origin_lsn=0) at
reorderbuffer.c:2973
#22 0x00005c9c560a6b71 in ReorderBufferCommit (rb=0x5c9c58b4f900,
xid=752, commit_lsn=8380824, end_lsn=8380872,
commit_time=793368288662227,
origin_id=0, origin_lsn=0) at reorderbuffer.c:2997
#23 0x00005c9c5606804a in DecodeCommit (ctx=0x5c9c58b4d8f0,
buf=0x7ffd9a445010, parsed=0x7ffd9a444e90, xid=752, two_phase=false)
at decode.c:730
#24 0x00005c9c56064d04 in xact_decode (ctx=0x5c9c58b4d8f0,
buf=0x7ffd9a445010) at decode.c:242
#25 0x00005c9c56063fe2 in LogicalDecodingProcessRecord
(ctx=0x5c9c58b4d8f0, record=0x5c9c58b4dd18) at decode.c:116
#26 0x00005c9c56135980 in XLogSendLogical () at walsender.c:3382
#27 0x00005c9c56133895 in WalSndLoop (send_data=0x5c9c56135836
<XLogSendLogical>) at walsender.c:2788
#28 0x00005c9c56130762 in StartLogicalReplication (cmd=0x5c9c58ab9a10)
at walsender.c:1496
#29 0x00005c9c56131ec3 in exec_replication_command (
cmd_string=0x5c9c58a83b90 "START_REPLICATION SLOT \"test1\"
LOGICAL 0/0 (proto_version '4', streaming 'parallel', origin 'any',
publication_names '\"pub1\"')") at walsender.c:2119
#30 0x00005c9c562abbe8 in PostgresMain (dbname=0x5c9c58abd598
"postgres", username=0x5c9c58abd580 "ubuntu") at postgres.c:4687

Thanks and Regards,
Shlok Kyal