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

Hi,

During logical decoding, if there is a large write transaction, some
spill files will be written to disk,
depending on the setting of max_changes_in_memory.

This behavior can effectively avoid OOM, but if the transaction
generates a lot of change before commit,
a large number of files may fill the disk. For example, you can update
a TB-level table.
Of course,   this is also inevitable.

But I found an inelegant phenomenon. If the updated large table is not
published, its changes will also
be written with a large number of spill files. Look at an example below:

publisher:
```
create table tbl_pub(id int, val1 text, val2 text,val3 text);
create table tbl_t1(id int, val1 text, val2 text,val3 text);
CREATE PUBLICATION mypub FOR TABLE public.tbl_pub;
```

subscriber:
```
create table tbl_pub(id int, val1 text, val2 text,val3 text);
create table tbl_t1(id int, val1 text, val2 text,val3 text);
CREATE SUBSCRIPTION mysub CONNECTION 'host=127.0.0.1 port=5432
user=postgres dbname=postgres' PUBLICATION mypub;
```

publisher:
```
begin;
insert into tbl_t1 select i,repeat('xyzzy', i),repeat('abcba',
i),repeat('dfds', i) from generate_series(0,999999) i;
```

Later you will see a large number of spill files in the
"/$PGDATA/pg_replslot/mysub/" directory.
```
$ll -sh
total 4.5G
4.0K -rw------- 1 postgres postgres 200 Nov 30 09:24 state
17M -rw------- 1 postgres postgres 17M Nov 30 08:22 xid-750-lsn-0-10000000.spill
12M -rw------- 1 postgres postgres 12M Nov 30 08:20 xid-750-lsn-0-1000000.spill
17M -rw------- 1 postgres postgres 17M Nov 30 08:23 xid-750-lsn-0-11000000.spill
......
```

We can see that table tbl_t1 is not published in mypub. It is also not
sent downstream because it is subscribed.
After the transaction is reorganized, the pgoutput decoding plug-in
filters out these change  of unpublished relation
when sending logical changes. see function pgoutput_change.

Above all, if after constructing a change and before queuing a change
into a transaction, we filter out unpublished
relation-related changes, will it make logical decoding less laborious
and avoid disk growth as much as possible?


This is just an immature idea. I haven't started to implement it yet.
Maybe it was designed this way because there
are key factors that I didn't consider. So I want to hear everyone's
opinions, especially the designers of logic decoding.

> This is just an immature idea. I haven't started to implement it yet.
> Maybe it was designed this way because there
> are key factors that I didn't consider. So I want to hear everyone's
> opinions, especially the designers of logic decoding.

Attached is the patch I used to implement this optimization.
The main designs are as follows:
1. Added a callback LogicalDecodeFilterByRelCB for the output plugin.

2. Added this callback function pgoutput_table_filter for the pgoutput plugin.
    Its main implementation is based on the table filter in the
pgoutput_change function.

3. After constructing a change and before Queue a change into a transaction,
    use RelidByRelfilenumber to obtain the relation associated with the change,
    just like obtaining the relation in the ReorderBufferProcessTXN function.

4. Relation may be a toast, and there is no good way to get its real
table relation
   based on toast relation. Here, I get the real table oid through
toast relname, and
   then get the real table relation.

5. This filtering takes into account INSERT/UPDATE/INSERT. Other
changes have not
   been considered yet and can be expanded in the future.

6.  The table filter in pgoutput_change and the get relation in
ReorderBufferProcessTXN
  can be deleted. This has not been done yet. This is the next step.

Sincerely look forward to your feedback.
Regards, lijie

On Fri, Dec 1, 2023 at 1:55 PM li jie <ggysxcq@gmail.com> wrote:
>
> > This is just an immature idea. I haven't started to implement it yet.
> > Maybe it was designed this way because there
> > are key factors that I didn't consider. So I want to hear everyone's
> > opinions, especially the designers of logic decoding.
>
> Attached is the patch I used to implement this optimization.
> The main designs are as follows:
> 1. Added a callback LogicalDecodeFilterByRelCB for the output plugin.
>
> 2. Added this callback function pgoutput_table_filter for the pgoutput plugin.
>     Its main implementation is based on the table filter in the
> pgoutput_change function.
>
> 3. After constructing a change and before Queue a change into a transaction,
>     use RelidByRelfilenumber to obtain the relation associated with the change,
>     just like obtaining the relation in the ReorderBufferProcessTXN function.
>
> 4. Relation may be a toast, and there is no good way to get its real
> table relation
>    based on toast relation. Here, I get the real table oid through
> toast relname, and
>    then get the real table relation.
>

This may be helpful for the case you have mentioned but how about
cases where there is nothing to filter by relation? It will add
overhead related to the transaction start/end and others for each
change. Currently, we do that just once for all the changes that need
to be processed. I wonder why the spilling can't be avoided with GUC
logical_decoding_work_mem?

--
With Regards,
Amit Kapila.

>
> This may be helpful for the case you have mentioned but how about
> cases where there is nothing to filter by relation?

You can see the complete antecedent in the email [1]. Relation that has
not been published will also generate changes and put them into the entire
transaction group, which will increase invalid memory or disk space.

> It will add
> overhead related to the transaction start/end and others for each
> change. Currently, we do that just once for all the changes that need
> to be processed.

Yes, it will only be processed once at present. It is done when applying
each change when the transaction is committed. This patch hopes to
advance it to the time when constructing the change, and determines the
change queue into a based on whether the relation is published.

> I wonder why the spilling can't be avoided with GUC
> logical_decoding_work_mem?

Of course you can, but this will only convert disk space into memory space.
 For details, please see the case in Email [1].

Regards, lijie

> This may be helpful for the case you have mentioned but how about
> cases where there is nothing to filter by relation?

You can see the complete antecedent in the email [1]. Relation that has
not been published will also generate changes and put them into the entire
transaction group, which will increase invalid memory or disk space.

> It will add
> overhead related to the transaction start/end and others for each
> change. Currently, we do that just once for all the changes that need
> to be processed.

Yes, it will only be processed once at present. It is done when applying
each change when the transaction is committed. This patch hopes to
advance it to the time when constructing the change, and determines the
change queue into a based on whether the relation is published.

> I wonder why the spilling can't be avoided with GUC
> logical_decoding_work_mem?

Of course you can, but this will only convert disk space into memory space.
 For details, please see the case in Email [1].

[1] https://www.postgresql.org/message-id/CAGfChW51P944nM5h0HTV9HistvVfwBxNaMt_s-OZ9t%3DuXz%2BZbg%40mail.gmail.com

Regards, lijie

Amit Kapila <amit.kapila16@gmail.com> 于2023年12月2日周六 12:11写道：
>
> On Fri, Dec 1, 2023 at 1:55 PM li jie <ggysxcq@gmail.com> wrote:
> >
> > > This is just an immature idea. I haven't started to implement it yet.
> > > Maybe it was designed this way because there
> > > are key factors that I didn't consider. So I want to hear everyone's
> > > opinions, especially the designers of logic decoding.
> >
> > Attached is the patch I used to implement this optimization.
> > The main designs are as follows:
> > 1. Added a callback LogicalDecodeFilterByRelCB for the output plugin.
> >
> > 2. Added this callback function pgoutput_table_filter for the pgoutput plugin.
> >     Its main implementation is based on the table filter in the
> > pgoutput_change function.
> >
> > 3. After constructing a change and before Queue a change into a transaction,
> >     use RelidByRelfilenumber to obtain the relation associated with the change,
> >     just like obtaining the relation in the ReorderBufferProcessTXN function.
> >
> > 4. Relation may be a toast, and there is no good way to get its real
> > table relation
> >    based on toast relation. Here, I get the real table oid through
> > toast relname, and
> >    then get the real table relation.
> >
>
> This may be helpful for the case you have mentioned but how about
> cases where there is nothing to filter by relation? It will add
> overhead related to the transaction start/end and others for each
> change. Currently, we do that just once for all the changes that need
> to be processed. I wonder why the spilling can't be avoided with GUC
> logical_decoding_work_mem?
>
> --
> With Regards,
> Amit Kapila.

Dear Li,

Thanks for proposing and sharing the PoC. Here are my high-level comments.

1.
What if ALTER PUBLICATION ... ADD is executed in parallel?
Should we publish added tables if the altering is done before the transaction is
committed? The current patch seems unable to do so because changes for added
tables have not been queued at COMMIT.
If we should not publish such tables, why?

2.
This patch could not apply as-is. Please rebase.

3. FilterByTable()

```
+    if (ctx->callbacks.filter_by_origin_cb == NULL)
+        return false;
```

filter_by_table_cb should be checked instead of filter_by_origin_cb.
Current patch crashes if the filter_by_table_cb() is not implemented.

4. DecodeSpecConfirm()

```
+    if (FilterByTable(ctx, change))
+        return;
+
```

I'm not sure it is needed. Can you explain the reason why you added?

5. FilterByTable

```
+    switch (change->action)
+    {
+            /* intentionally fall through */
+        case REORDER_BUFFER_CHANGE_INSERT:
+        case REORDER_BUFFER_CHANGE_UPDATE:
+        case REORDER_BUFFER_CHANGE_DELETE:
+            break;
+        default:
+            return false;
+    }
```

IIUC, REORDER_BUFFER_CHANGE_TRUNCATE also targes the user table, so I think
it should be accepted. Thought?

6.

I got strange errors when I tested the feature. I thought this implied there were
bugs in your patch.

1. implemented no-op filter atop test_decoding like attached
2. ran `make check` for test_decoding modle
3. some tests failed. Note that "filter" was a test added by me.
  regression.diffs was also attached.

```
not ok 1     - ddl                                       970 ms
ok 2         - xact                                       36 ms
not ok 3     - rewrite                                   525 ms
not ok 4     - toast                                     736 ms
ok 5         - permissions                                50 ms
ok 6         - decoding_in_xact                           39 ms
not ok 7     - decoding_into_rel                          57 ms
ok 8         - binary                                     21 ms
not ok 9     - prepared                                   33 ms
ok 10        - replorigin                                 93 ms
ok 11        - time                                       25 ms
ok 12        - messages                                   47 ms
ok 13        - spill                                    8063 ms
ok 14        - slot                                      124 ms
ok 15        - truncate                                   37 ms
not ok 16    - stream                                     60 ms
ok 17        - stats                                     157 ms
ok 18        - twophase                                  122 ms
not ok 19    - twophase_stream                            57 ms
not ok 20    - filter                                     20 ms
```

Currently I'm not 100% sure the reason, but I think it may read the latest system
catalog even if ALTER SUBSCRIPTION is executed after changes.
In below example, an attribute is altered text->somenum, after inserting data.
But get_changes() outputs as somenum.

```
  BEGIN
- table public.replication_example: INSERT: id[integer]:1 somedata[integer]:1 text[character varying]:'1'
- table public.replication_example: INSERT: id[integer]:2 somedata[integer]:1 text[character varying]:'2'
+ table public.replication_example: INSERT: id[integer]:1 somedata[integer]:1 somenum[character varying]:'1'
+ table public.replication_example: INSERT: id[integer]:2 somedata[integer]:1 somenum[character varying]:'2'
  COMMIT
```

Also, if the relfilenuber of the relation is changed, an ERROR is raised.

```
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts',
'1');
-                                    data                                    
-----------------------------------------------------------------------------
- BEGIN
- table public.tr_pkey: INSERT: id2[integer]:2 data[integer]:1 id[integer]:2
- COMMIT
- BEGIN
- table public.tr_pkey: DELETE: id[integer]:1
- table public.tr_pkey: DELETE: id[integer]:2
- COMMIT
-(7 rows)
-
+ERROR:  could not map filenumber "base/16384/16397" to relation OID
```

Best Regards,
Hayato Kuroda
FUJITSU LIMITED
https://www.fujitsu.com/ 

Hi Ajin,

Some review comments for patch v12-0001.

======
Commit message

1.
Track transactions which have snapshot changes with a new flag
RBTXN_HAS_SNAPSHOT_CHANGES

~

The commit message only says *what* it does, but not *why* this patch
even exists.  TBH, I don't understand why this patch needs to be
separated from your patch 0002, because 0001 makes no independent use
of the flag, nor is it separately tested.

Anyway, if it is going to remain separated then IMO at least the the
message should explain the intended purpose e.g. why the subsequent
patches require this flagged info and how they will use it.

======
src/include/replication/reorderbuffer.h

2.
+/* Does this transaction have snapshot changes? */
+#define rbtxn_has_snapshot_changes(txn) \
+( \
+ ((txn)->txn_flags & RBTXN_HAS_SNAPSHOT_CHANGES) != 0 \
+)
+

Is the below wording maybe a more plain way to say that:

/* Does this transaction make changes to the current snapshot? */

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