Logical Replication Of In-Progress Transactions

Logical Replication was introduced in PostgreSQL-10 and since then it is being improved with each version. Logical Replication is a method to replicate the data selectively unlike physical replication where the data of the entire cluster is copied. This can be used to build a multi-master or bi-directional replication solution. One of the main differences as compared with physical replication was that it allows replicating the transaction only at commit time. This leads to apply lag for large transactions where we need to wait to transfer the data till the transaction is finished. In the upcoming PostgreSQL-14 release, we are introducing a mechanism to stream the large in-progress transactions. We have seen the replication performance improved by 2 or more times due to this for large transactions especially due to early filtering. See the performance test results reported on hackers and in another blog on the same topic. This will reduce the apply lag to a good degree.

The first thing we need for this feature was to decide when to start streaming the WAL content. One could think if we have such a technology why not stream each change of transaction separately as and when we retrieve it from WAL but that would actually lead to sending much more data across the network because we need to send some additional transaction information with each change so that the apply-side can recognize the transaction to which the change belongs. To address this, in PostgreSQL-13, we have introduced a new GUC parameter logical_decoding_work_mem which allows users to specify the maximum amount of memory to be used by logical decoding, before which some of the decoded changes are either written to local disk or stream to the subscriber. The parameter is also used to control the memory used by logical decoding as explained in the blog.

The next thing that prevents incremental decoding was the delay in finding the association of subtransaction and top-level XID. During logical decoding, we accumulate all changes along with its (sub)transaction. Now, while sending the changes to the output plugin or stream to the other node, we need to combine all the changes that happened in the transaction which requires us to find the association of each top-level transaction with its subtransactions. Before PostgreSQL-14, we build this association at XLOG_XACT_ASSIGNMENT WAL record which we normally log after 64 subtransactions or at commit time because these are the only two times when we get such an association in the WAL. To find this association as it happened, we now also write the assignment info into WAL immediately, as part of the first WAL record for each subtransaction. This is done only when wal_level=logical to minimize the overhead.

Yet, another thing that is required for incremental decoding was to process invalidations at each command end. The basic idea of invalidations is that they make the caches (like relation cache) up-to-date to allow the next command to use up-to-date schema. This was required to correctly decode WAL incrementally as while decoding we will use the relation attributes from the caches. For this, when wal_level=logical, we write invalidations at the command end into WAL so that decoding can use this information. The invalidations are decoded and accumulated in top-transaction, and then executed during replay. This obviates the need to decode the invalidations as part of a commit record.

In previous paragraphs, the enhancements required in the server infrastructure to allow incremental decoding are explained. The next step was to provide APIs (stream methods) for out-of-core logical replication to stream large in-progress transactions. We added seven methods to the output plugin API to allow this. Those are: (stream_start_cb, stream_stop_cb, stream_abort_cb, stream_commit_cb and stream_change_cb) and two optional callbacks (stream_message_cb and stream_truncate_cb). For details about these APIs, refer to PostgreSQL docs.

When streaming an in-progress transaction, the changes (and messages) are streamed in blocks demarcated by stream_start_cb and stream_stop_cb callbacks. Once all the decoded changes are transmitted, the transaction can be committed using the stream_commit_cb callback (or possibly aborted using the stream_abort_cb callback). One example sequence of streaming transaction may look like the following:

/* Change logical_decoding_work_mem to 64kB in the session */
postgres=# show logical_decoding_work_mem;
 logical_decoding_work_mem
---------------------------
 64kB
(1 row)
postgres=# CREATE TABLE stream_test(data text);
CREATE TABLE

postgres=# SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'test_decoding');
 ?column?
----------
 init
(1 row)
postgres=# INSERT INTO stream_test SELECT repeat('a', 6000) || g.i FROM generate_series(1, 500) g(i);
INSERT 0 500
postgres=# SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL,NULL, 'include-xids', '1', 'skip-empty-xacts', '1', 'stream-changes', '1');
                       data
--------------------------------------------------
 opening a streamed block for transaction TXN 741
 streaming change for TXN 741
 streaming change for TXN 741
 streaming change for TXN 741
...
...
 streaming change for TXN 741
 streaming change for TXN 741
 streaming change for TXN 741
 closing a streamed block for transaction TXN 741
 opening a streamed block for transaction TXN 741
 streaming change for TXN 741
 streaming change for TXN 741
 streaming change for TXN 741
...
...
 streaming change for TXN 741
 streaming change for TXN 741
 closing a streamed block for transaction TXN 741
 committing streamed transaction TXN 741
(505 rows)

The actual sequence of callback calls may be more complicated depending on the server operations. There may be blocks for multiple streamed transactions, some of the transactions may get aborted, etc.

Note that streaming is triggered when the total amount of changes decoded from the WAL (for all in-progress transactions) exceeds the limit defined by the logical_decoding_work_mem setting. At that point, the largest top-level transaction (measured by the amount of memory currently used for decoded changes) is selected and streamed. However, in some cases we still have to spill to disk even if streaming is enabled because we exceed the memory threshold but still have not decoded the complete tuple e.g., only decoded toast table insert but not the main table insert or decoded speculative insert but not the corresponding confirm record. However, as soon as we get the complete tuple we stream the transaction including the serialized changes.

While streaming in-progress transactions, the concurrent aborts may cause failures when the output plugin (or decoding of WAL records) consults catalogs (both system and user-defined). Let me explain this with an example, suppose there is one catalog tuple with (xmin: 500, xmax: 0). Now, the transaction 501 updates the catalog tuple and after that we will have two tuples (xmin: 500, xmax: 501) and (xmin: 501, xmax: 0). Now, if 501 is aborted and some other transaction say 502 updates the same catalog tuple then the first tuple will be changed to (xmin: 500, xmax: 502). So, the problem is that when we try to decode the tuple inserted/updated in 501 after the catalog update, we will see the catalog tuple with (xmin: 500, xmax: 502) as visible because it will consider that the tuple is deleted by xid 502 which is not visible to our snapshot. And when we will try to decode with that catalog tuple, it can lead to a wrong result or a crash.  So, it is necessary to detect concurrent aborts to allow streaming of in-progress transactions. For detecting the concurrent abort, during catalog scan we can check the status of the xid and if it is aborted we will report a specific error so that we can stop streaming current transaction and discard the already streamed changes on such an error. We might have already streamed some of the changes for the aborted (sub)transaction, but that is fine because when we decode the abort we will stream the abort message to truncate the changes in the subscriber.

To add support for streaming of in-progress transactions into the built-in logical replication, we need to primarily do four things:

(a) Extend the logical replication protocol to identify in-progress transactions, and allow adding additional bits of information (e.g. XID of subtransactions). Refer to PostgreSQL docs for the protocol details. 

(b) Modify the output plugin (pgoutput) to implement the new stream API callbacks, by leveraging the extended replication protocol.

(c) Modify the replication apply worker, to properly handle streamed in-progress transaction by spilling the data to disk and then replaying them on commit.

(d) Provide a new option for streaming while creating a subscription.

The below example demonstrates how to set up the streaming via built-in logical replication:

Publisher node:

Set logical_decoding_work_mem = '64kB';
# Set up publication with some initial data

CREATE TABLE test_tab (a int primary key, b varchar);
INSERT INTO test_tab VALUES (1, 'foo'), (2, 'bar');
CREATE PUBLICATION tap_pub FOR TABLE test_tab;

Subscriber node:

CREATE TABLE test_tab (a int primary key, b varchar);
CREATE SUBSCRIPTION tap_sub CONNECTION 'host=localhost port=5432 dbname=postgres' PUBLICATION tap_pub WITH (streaming = on);

Publisher Node:

# Ensure the corresponding replication slot is created on publisher node
select slot_name, plugin, slot_type from pg_replication_slots;
 slot_name |  plugin  | slot_type
-----------+----------+-----------
 tap_sub   | pgoutput | logical
(1 row)

# Confirm there is no streamed bytes yet
postgres=# SELECT slot_name, stream_txns, stream_count, stream_bytes FROM pg_stat_replication_slots;
 slot_name | stream_txns | stream_count | stream_bytes
-----------+-------------+--------------+--------------
 tap_sub   |           0 |            0 |            0
(1 row)

# Insert, update and delete enough rows to exceed the logical_decoding_work_mem (64kB) limit.
BEGIN;
INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
DELETE FROM test_tab WHERE mod(a,3) = 0;

# Confirm that streaming happened
SELECT slot_name, stream_txns, stream_count, stream_bytes FROM pg_stat_replication_slots;
 slot_name | stream_txns | stream_count | stream_bytes
-----------+-------------+--------------+--------------
 tap_sub   |           1 |           22 |      1444410
(1 row)

Subscriber Node:
# The streamed data is still not visible.
select * from test_tab;
 a |  b
---+-----
 1 | foo
 2 | bar
(2 rows)

Publisher Node:
# Commit the large transactions
Commit;

Subscriber Node:
# The data must be visible on the subscriber
select count(*) from test_tab;
 count
-------
  3334
(1 row)

This feature was proposed in 2017 and committed in 2020 as part of various commits 0bead9af48, c55040ccd0, 45fdc9738b, 7259736a6e, and 464824323e. It took a long time to complete this feature because of the various infrastructure pieces required to achieve this. I would really like to thank all the people involved in this feature especially Tomas Vondra who has initially proposed it and then Dilip Kumar who along with me had completed various remaining parts and made it a reality. Then also to other people like Neha Sharma, Mahendra Singh Thalor, Ajin Cherian, and Kuntal Ghosh who helped throughout the project to do reviews and various tests. Also, special thanks to Andres Freund and other community members who have suggested solutions to some of the key problems of this feature. Last but not least, thanks to EDB and Fujitsu's management who encouraged me and some of the other members to work on this feature.

Improved (auto)vacuum in PostgreSQL 13

Vacuum is one of the sub-systems in PostgreSQL which gets improved with each release.  I have checked past five releases and each has quite a few improvements for vacuum.  Following the trend, there are a number of improvements in vacuum in the upcoming PostgreSQL release (v13) which are covered in this blog.

Improvement-1

---------------------

Vacuum will be allowed to process indexes in parallel. This means it can leverage multiple CPUs in order to perform index cleanup. This internally uses the background worker infrastructure of PostgreSQL to accomplish the work.  There is a new option PARALLEL which controls the parallelism used by vacuum.  Users can use the new option to specify the number of workers that can be used to perform the vacuum command which is limited by the minimum of (a) the number of indexes on a table and (b) max_parallel_maintenance_workers.  The parallelism for vacuum command is enabled by default which means it will be used even if the user didn't specify the PARALLEL option and it uses the number of workers equal to the number of indexes on the table being vacuumed.  We can disable parallelism for vacuum by specifying zero as the number of workers with the PARALLEL option.  The index can participate in parallel

vacuum iff it's size is greater than min_parallel_index_scan_size.

The PARALLEL option can't be used with the FULL option in vacuum command.  This feature won't be available via autovacuum, users need to use vacuum command to get the benefit provided by this feature.

There are a number of other blogs [1][2] written on this topic which shows the benefit of this feature.  A recent blog published by EnterpriseDB shows that vacuum could be approximately 4 times faster by using 7 workers especially when the relation is in dire need of vacuum, read that blog for more information about the test.

Here is a simple test to show the usage of this feature.

postgres=# create table pvac(c1 int, c2 text DEFAULT md5(random()::text), c3 text DEFAULT md5(random()::text));

CREATE TABLE

postgres=# create index pvac_1 on pvac(c1);

CREATE INDEX

postgres=# create index pvac_2 on pvac(c2);

CREATE INDEX

postgres=# create index pvac_3 on pvac(c3);

CREATE INDEX

postgres=# insert into pvac select i FROM generate_series(1,100000) as i;

INSERT 0 100000

postgres=# update pvac set c1=c1;

UPDATE 100000

postgres=# vacuum (parallel 4, verbose) pvac;

INFO:  vacuuming "public.pvac"

INFO:  launched 2 parallel vacuum workers for index vacuuming (planned: 2)

INFO:  scanned index "pvac_1" to remove 100000 row versions

..

..

In the above test, 2 parallel workers were used even though we have specified 4 and the reason is that the number of indexes is 3, and the number of workers can't be more than index which is further limited by max_parallel_maintenance_workers as you can see by command below.

postgres=# show max_parallel_maintenance_workers;

 max_parallel_maintenance_workers 

----------------------------------

 2

(1 row)

The other most important thing we have ensured in this feature is that the parallelism won't use more memory or I/O bandwidth as compared to non-parallel vacuum.

Improvement-2

---------------------

Allow inserts to trigger autovacuum activity.  This feature will be really helpful for the insert-only tables where anti-wraparound vacuums could be the first vacuum that the table ever receives and such a run would take a really long time.   This allows heap pages to be set as all-visible, which then allows index-only scans to skip heap fetches, and reduces the work necessary when the table needs to be frozen.  This is controlled by two new GUCs and reloptions; autovacuum_vacuum_insert_threshold and autovacuum_vacuum_insert_scale_factor.

Let me demonstrate the benefit of this feature with the help on an example.  Start the server with autovacuum_vacuum_insert_threshold = -1 (one can edit postgresql.conf file to change value of this parameter or can use Alter System command).  By connecting with psql, we can execute below commands to see the behavior.

postgres=# show autovacuum_vacuum_insert_threshold;

 autovacuum_vacuum_insert_threshold 

------------------------------------

 -1

(1 row)

postgres=# create table vac_ins(c1 int, c2 char(500));

CREATE TABLE

postgres=# create index idx_vac_ins on vac_ins(c1);

CREATE INDEX

postgres=# insert into vac_ins values(generate_series(1,20000),'aaaaaa');

INSERT 0 20000

After a few seconds, you can notice the below message in the server log which shows that autovacuum has performed analyze on the table.

LOG:  automatic analyze of table "postgres.public.vac_ins" system usage: CPU: user: 0.03 s, system: 0.11 s, elapsed: 0.15 s

After that, run below command:

postgres=# explain (analyze) select c1 from vac_ins where c1 < 100;

                                                         QUERY PLAN                                                          

-----------------------------------------------------------------------------------------------------------------------------

 Index Only Scan using idx_vac_ins on vac_ins  (cost=0.29..16.02 rows=99 width=4) (actual time=0.019..0.092 rows=99 loops=1)

   Index Cond: (c1 < 100)

   Heap Fetches: 99

 Planning Time: 0.269 ms

 Execution Time: 0.129 ms

(5 rows)

Here, "Heap Fetches: 99" shows that the above query need to visit heap to fetch the required information even though it is present in index and the scan type used is Index Only Scan.

postgres=# truncate vac_ins;

TRUNCATE TABLE

Now, restart the server with a default value of autovacuum_vacuum_insert_threshold and execute below commands from psql to see how the new feature helps:

postgres=# show autovacuum_vacuum_insert_threshold;

 autovacuum_vacuum_insert_threshold 

------------------------------------

 1000

(1 row)

postgres=# insert into vac_ins values(generate_series(1,20000),'aaaaaa');

INSERT 0 20000

After a few seconds, you can notice the below message in the server log which indicates that autovacuum has performed both vacuum and analyze on the table.

LOG:  automatic vacuum of table "postgres.public.vac_ins": index scans: 0

pages: 0 removed, 1334 remain, 0 skipped due to pins, 0 skipped frozen

...

LOG:  automatic analyze of table "postgres.public.vac_ins" system usage: CPU: user: 0.01 s, system: 0.15 s, elapsed: 0.21 s

After that, run below command:

postgres=# explain (analyze) select c1 from vac_ins where c1 < 100;

                                                         QUERY PLAN                                                         

----------------------------------------------------------------------------------------------------------------------------

 Index Only Scan using idx_vac_ins on vac_ins  (cost=0.29..6.02 rows=99 width=4) (actual time=0.016..0.039 rows=99 loops=1)

   Index Cond: (c1 < 100)

   Heap Fetches: 0

 Planning Time: 0.166 ms

 Execution Time: 0.074 ms

(5 rows)

Here, "Heap Fetches: 0" shows that the above query doesn't need to visit heap to fetch the required information. We can see that the time to complete the execution is reduced significantly in this case.

Improvement-3

---------------------

Allow an (auto)vacuum to display additional information about the heap or index in case of an error.  This feature could help users in case the database has some corruption.  For example, if one of the indexes on a relation has some corrupted data (due to bad hardware or some bug), it will let the user know the index information, and the user can take appropriate action like either Reindex or maybe drop and recreate that particular index to overcome the problem.  In the case of the heap, it displays the block number for which an error has occurred which makes it much easier for users and developers to detect the problem.  In the worst case, if any particular block is corrupted in a table, users can remove all the rows of that particular block and the table can be used.

vacuum pvac;

ERROR:  error induced to demonstrate use of information

CONTEXT:  while scanning block 2469 of relation "public.pvac"

Next, I used "SELECT * FROM heap_page_items(get_raw_page('pvac', 2469));" to find the information of all line pointers in the page and then removed them from using below query. 

postgres=# delete from pvac where ctid Between '(2469,1)' and '(2469,11)';

DELETE 11

Then, I again ran vacuum

postgres=# vacuum pvac;

ERROR:  error induced to demonstrate use of information

CONTEXT:  while scanning block 2468 of relation "public.pvac"

Now, you can see the block number is changed from 2469 to 2468 which means the vacuum could proceed.  As, in this case, I have manually induced the error by changing code, so it occurs for every block but in reality it would be for some particular block(s) and once user can get rid of those block(s), the table can be reused.  I don't want to say that is an ideal situation but at least it will allow users to proceed and it can help developers to narrow down the bug if there is any in the code.

Improvement-4

---------------------

Autovacuum will now log WAL usage statistics along with other information.  The WAL usage contains information on the total number of records, number of full pages images, and the total number of bytes.  The buffer usage and WAL usage stats combined gives us approximate usage of I/O by a particular autovacuum run. I took one example runs information which will be shown below, one can refer to "WAL usage" to check the information on newly added stats.

LOG:  automatic vacuum of table "postgres.public.pvac": index scans: 1

pages: 0 removed, 2470 remain, 0 skipped due to pins, 0 skipped frozen

tuples: 100000 removed, 100000 remain, 0 are dead but not yet removable, oldest xmin: 529

buffer usage: 9276 hits, 4 misses, 3 dirtied

avg read rate: 0.062 MB/s, avg write rate: 0.047 MB/s

system usage: CPU: user: 0.40 s, system: 0.00 s, elapsed: 0.50 s

WAL usage: 7909 records, 2 full page images, 2276323 bytes

Improvement-5

---------------------

Make vacuum buffer counters 64-bits wide to avoid overflow of buffer usage stats.  Without this feature, in extreme cases, if there exist tables that are large enough for 4 billion buffer accesses to be a possibility, the stats displayed are meaningless.  See below example of 'buffer usage' stats from pgsql-hackers:

LOG:  automatic vacuum of table "somtab.sf.foobar": index scans: 17

pages: 0 removed, 207650641 remain, 0 skipped due to pins, 13419403 skipped frozen

tuples: 141265419 removed, 3186614627 remain, 87783760 are dead but not yet removable

buffer usage: -2022059267 hits, -17141881 misses, 1252507767 dirtied

avg read rate: -0.043 MB/s, avg write rate: 3.146 MB/s

system usage: CPU 107819.92s/2932957.75u sec elapsed 3110498.10 sec

Improvement-6

---------------------

Add wait event 'VacuumDelay' to report on cost-based vacuum delay.  This will help us to monitor the [auto]vacuum throttling.  This is a small feature but quite important as right now we have no way to monitor [auto]vacuum throttling and it is not very uncommon to see this in user environments.

Parallelism, what next?

This blog post is about the journey of parallelism in PostgreSQL till now and what is in store for the future.  Since PostgreSQL 9.6 where the first feature of parallel query has arrived, each release improves it.  Below is a brief overview of the parallel query features added in each release.

PG9.6 has added Parallel execution of sequential scans, joins, and aggregates.

PG10 has added (a) Support parallel B-tree index scans, (b) Support parallel bitmap heap scans, (c) Allow merge joins to be performed in parallel, (d) Allow non-correlated subqueries to be run in parallel, (e) Improve ability of parallel workers to return pre-sorted data and (f) Increase parallel query usage in procedural language functions.

PG11 has added (a) Allow parallel building of a btree index, (b) Allow hash joins to be performed in parallel using a shared hash table, (c) Allow parallelization of commands CREATE TABLE ... AS, SELECT INTO, and CREATE MATERIALIZED VIEW, (d) Allow UNION to run each SELECT in parallel if the individual SELECTs cannot be parallelized, (e) Allow partition scans to more efficiently use parallel workers, (f) Allow LIMIT to be passed to parallel workers, this allows workers to reduce returned results and use targeted index scans, (g) Allow single-evaluation queries, e.g. WHERE clause aggregate queries, and functions in the target list to be parallelized.

PG12 has added Allow parallelized queries when in SERIALIZABLE isolation mode.

The progress for PG13 with respect to parallelism.  Some of the important advancements are:

(a) Parallel vacuum - This feature allows the vacuum to leverage multiple CPUs in order to process indexes.  This enables us to perform index vacuuming and index cleanup with background workers.  This adds a PARALLEL option to VACUUM command where the user can specify the number of workers that can be used to perform the command which is limited by the number of indexes on a table.  Specifying zero as a number of workers will disable parallelism.  For more information, see commit.

(b) Improve EXPLAIN's handling of per-worker details.  This allows displaying the worker information in a much better way.  The few visible side-effects as mentioned in the commit

* In text format, instead of something like

  Sort Method: external merge  Disk: 4920kB

  Worker 0:  Sort Method: external merge  Disk: 5880kB

  Worker 1:  Sort Method: external merge  Disk: 5920kB

  Buffers: shared hit=682 read=10188, temp read=1415 written=2101

  Worker 0:  actual time=130.058..130.324 rows=1324 loops=1

    Buffers: shared hit=337 read=3489, temp read=505 written=739

  Worker 1:  actual time=130.273..130.512 rows=1297 loops=1

    Buffers: shared hit=345 read=3507, temp read=505 written=744

you get

  Sort Method: external merge  Disk: 4920kB

  Buffers: shared hit=682 read=10188, temp read=1415 written=2101

  Worker 0:  actual time=130.058..130.324 rows=1324 loops=1

    Sort Method: external merge  Disk: 5880kB

    Buffers: shared hit=337 read=3489, temp read=505 written=739

  Worker 1:  actual time=130.273..130.512 rows=1297 loops=1

    Sort Method: external merge  Disk: 5920kB

    Buffers: shared hit=345 read=3507, temp read=505 written=744

(c) Avoid unnecessary shm writes in Parallel Hash Join.  This improves the performance of Parallel Hash Join by a significant amount on large systems running many-core joins.  Though this work has been back-patched to v11 where Parallel Hash Join was introduced, I mentioned it here as it is done during PG13 development.  For more information, see commit.

What is being discussed for the future:

(a) Parallel grouping sets - PostgreSQL already supports parallel aggregation by aggregating in two stages. First, each process participating in the parallel portion of the query performs an aggregation step, producing a partial result for each group of which that process is aware. Second, the partial results are transferred to the leader via the Gather node. Finally, the leader re-aggregates the results across all workers in order to produce the final result.

Next, there has been a discussion in the community to parallelize queries containing grouping sets in much the same way as we do parallel aggregation.

Basically, the aim is to parallelize queries like SELECT brand, size, sum(sales) FROM items_sold GROUP BY GROUPING SETS ((brand), (size), ());

This feature has been proposed for PG13, but yet not committed.

(b) Parallel copy - We also want to parallelize the Copy command, in particular "Copy <table_name> from .. ;" command.  This will help improve the bulk load operation in PostgreSQL.  Currently, we do a lot of work during the Copy command.  We read the file in 64KB chunks, then find the line endings and process that data line by line, where each line corresponds to one tuple.  We first form the tuple (in form of value/null array) from that line, check if it qualifies the where condition and if it qualifies, then perform constraint check and few other checks and then finally store it in local tuple array.  Once we reach 1000 tuples or consumed 64KB (whichever occurred first), we insert them together and then for each tuple insert into the index(es) and execute after row triggers.  The aim of this work is to parallelize as much as possible the work done during the copy.  There is an ongoing discussion in the community on this topic.

There is a small caveat here that to achieve parallel copy, we need to work on relation extension lock where parallel workers block each other while extending the relation which is not the case currently.  There is already a discussion on this topic in the community.

(c) Parallel file_fdw - The proposed work in this area allows file_fdw to divide its scan up for parallel workers, much like a parallel seq scan.

There are more areas where parallelism can be used like parallel DMLs (inserts, updates, deletes).  During a discussion with Thomas Munro, it came up that it would be beneficial if we can parallelize index creation and index scans for indexes other than btree especially gin and gist.  Note that we already support parallel index scans and parallel index creation for btree. I would not like to go in detail of these operations as till now we haven't seen any proposal for those.  Similarly, we can improve few things in our current parallel infrastructure (a) As of now, for each query the parallel workers are created and destroyed, instead we can have some pool of parallel query workers which can avoid the overhead of starting them for each query, (b) As of now, each worker can use up to work_mem of memory which might increase the overall memory usage of query.  We might want to improve this, but currently, there is no proposal for this.

Parallel Index Scans In PostgreSQL

There is a lot to say about parallelism in PostgreSQL. We have come a long way since I wrote my first post on this topic (Parallel Sequential Scans). Each of the past three releases (including PG-11, which is in its beta) have a parallel query as a major feature which in itself says how useful is this feature and the amount of work being done on this feature. You can read more about parallel query from the PostgreSQL docs or from a blog post on this topic by my colleague Robert Haas. The intent of this blog post is to talk about parallel index scans which were released in PostgreSQL 10. Currently, we have supported parallel scan for btree-indexes.

To demonstrate how the feature works, here is an example of TPC-H Q-6 at scale factor - 20 (which means approximately 20GB database). Q6 is a forecasting revenue change query. This query quantifies the amount of revenue increase that would have resulted from eliminating certain company-wide discounts in a given percentage range in a given year. Asking this type of "what if" query can be used to look for ways to increase revenues.

explain analyze

select sum(l_extendedprice * l_discount) as revenue
          from lineitem

          where l_shipdate >= date '1994-01-01' and

          l_shipdate < date '1994-01-01' + interval '1' year and

          l_discount between 0.02 - 0.01 and 0.02 + 0.01 and

          l_quantity < 24
          LIMIT 1;

Non-parallel version of plan
-------------------------------------
Limit
-> Aggregate
    -> Index Scan using idx_lineitem_shipdate on lineitem
         Index Cond: ((l_shipdate >= '1994-01-01'::date) AND (l_shipdate < '1995-01-01

         00:00:00'::timestamp without time zone) AND (l_discount >= 0.01) AND

         (l_discount <= 0.03)  AND  (l_quantity < '24'::numeric))

Planning Time: 0.406 ms
Execution Time: 35073.886 ms

Parallel version of plan
-------------------------------
Limit
-> Finalize Aggregate
    -> Gather

         Workers Planned: 2

         Workers Launched: 2
          -> Partial Aggregate
               -> Parallel Index Scan using idx_lineitem_shipdate on lineitem
                    Index Cond: ((l_shipdate >= '1994-01-01'::date) AND (l_shipdate < '1995-01-01 

                    00:00:00'::timestamp without time zone) AND (l_discount >= 0.01) AND

                    (l_discount <= 0.03) AND (l_quantity < '24'::numeric))

Planning Time: 0.420 ms
Execution Time: 15545.794 ms

We can see that the execution time is reduced by more than half for a parallel plan with two parallel workers. This query filters many rows and the work (CPU time) to perform that is divided among workers (and leader), leading to reduced time.

To further see the impact with a number of workers, we have used somewhat bigger dataset (scale_factor = 50). The setup has been done using TPC-H like benchmark for PostgreSQL. We have also created few additional indexes on columns (l_shipmode, l_shipdate, o_orderdate, o_comment)

Non-default parameter settings:
random_page_cost = seq_page_cost = 0.1
effective_cache_size = 10GB
shared_buffers = 8GB
work_mem = 1GB

The time is reduced almost linearly till 8 workers and then it reduced slowly. The further increase in workers won’t help unless the data to scan increases.

We have further evaluated the parallel index scan feature for all the queries in TPC-H benchmark and found that it is used in a number of queries and the impact is positive (reduced the execution time significantly). Below are results for TPC-H, scale factor - 20 with a number of parallel workers as 2. X-axis indicates (1: Q-6, 2: Q14, 3: Q18).

Under the Hood

The basic idea is quite similar to parallel heap scans where each worker (including leader whenever possible) will scan a block (all the tuples in a block) and then get the next block that is required to be scan. The parallelism is implemented at the leaf level of a btree. The first worker to start a btree scan will scan till it reaches the leaf and others will wait till the first worker has reached the leaf. Once, the first worker read the leaf block, it sets the next block to be read and wakes one of the workers waiting to scan blocks. Further, it proceeds scanning tuples from the block it has read. Henceforth, each worker after reading a block sets the next block to be read and wakes up the next waiting worker. This continues till no more pages are left to scan at which we end the parallel scan and notify all the workers.

A new guc min_parallel_index_scan_size has been introduced which indicates the minimum amount of index data that must be scanned in order for a parallel scan to be considered. Users can try changing the value of this parameter to see if the parallel index plan is effective for their queries. The number of parallel workers is decided based on the number of index pages to be scanned. The final cost of parallel plan considers the cost (CPU cost) to process the rows will be divided equally among workers.

In the end, I would like to thank the people (Rahila Syed and Robert Haas) who were involved in this work (along with me) and my employer EnterpriseDB who has supported this work. I would also like to thank Rafia Sabih who helped me in doing performance testing for this blog.

zheap: a storage engine to provide better control over bloat

In the past few years, PostgreSQL has advanced a lot in terms of features, performance, and scalability for many-core systems.  However, one of the problems that many enterprises still complain is that its size increases over time which is commonly referred to as bloat. PostgreSQL has a mechanism known as autovacuum wherein a dedicated process (or set of processes) tries to remove the dead rows from the relation in an attempt to reclaim the space, but it can’t completely reclaim the space in many cases.  In particular, it always creates a new version of a tuple on an update which must eventually be removed by periodic vacuuming or by HOT-pruning, but still in many cases space is never reclaimed completely.  A similar problem occurs for tuples that are deleted. This leads to bloat in the database.  My colleague Robert Haas has discussed some such cases in his blog DO or UNDO - there is no VACUUM where the PostgreSQL heap tends to bloat and has also mentioned the solution (zheap: a new storage format for PostgreSQL) on which EnterpriseDB is working to avoid the bloat whenever possible.  The intent of this blog post is to elaborate on that work in some more detail and show some results.

This project has three major objectives:

1. Provide better control over bloat.  zheap will prevent bloat (a) by allowing in-place updates in common cases and (b) by reusing space as soon as a transaction that has performed a delete or non-in-place update has committed.  In short, with this new storage, whenever possible, we’ll avoid creating bloat in the first place.

2. Reduce write amplification both by avoiding rewrites of heap pages and by making it possible to do an update that touches indexed columns without updating every index.

3. Reduce the tuple size by (a) shrinking the tuple header and (b) eliminating most alignment padding.

In this blog post, I will mainly focus on the first objective (Provide better control over bloat) and leave other things for future blog posts on this topic.

In-place updates will be supported except when (a) the new tuple is larger than the old tuple and the increase in size makes it impossible to fit the larger tuple onto the same page or (b) some column is modified which is covered by an index that has not been modified to support “delete-marking”.  Note that the work to support delete-marking in indexes is yet to start and we intend to support it at least for btree indexes. For in-place updates, we have to write the old tuple in the undo log and the new tuple in the zheap which help concurrent readers to read the old tuple from undo if the latest tuple is not yet visible to them.

Deletes write the complete tuple in the undo record even though we could get away with just writing the TID as we do for an insert operation. This allows us to reuse the space occupied by the deleted record as soon as the transaction that has performed the operation commits. Basically, if the delete is not yet visible to some concurrent transaction, it can read the tuple from undo and in heap, we can immediately (as soon as the transaction commits) reclaim the space occupied by the record.

Below are some of the graphs that compare the size of heap and zheap table when the table is constantly updated and there is a concurrent long-running transaction.  To perform these tests, we have used pgbench to initialize the data (at scale factor 1000) and then use the simple-update test (which comprises of one-update, one-select, one-insert) to perform updates.  You can refer to the PostgreSQL manual for more about how to use pgbench. These tests have been performed on a machine with an x86_64 architecture, 2-sockets, 14-cores per socket, 2-threads per-core and has 64-GB RAM.  The non-default configuration for the tests is shared_buffers=32GB, min_wal_size=15GB, max_wal_size=20GB, checkpoint_timeout=1200, maintenance_work_mem=1GB, checkpoint_completion_target=0.9, synchoronous_commit = off. The below graphs show the size of the table on which this test has performed updates.

In the above test, we can see that the initial size of the table was 13GB in heap and 11GB in zheap.  After running the test for 25 minutes (out of which there was an open transaction for first 15-minutes), the size in heap grows to 16GB at 8-client count test and to 20GB at 64-client count test whereas for zheap the size remains at 11GB for both the client-counts at the end of the test. The initial size of zheap is lesser because the tuple header size is smaller in zheap. Now, certainly for first 15 minutes, autovacuum can’t reclaim any space due to the open transaction, but it can’t reclaim it even after the open transaction is ended. On the other hand, the size of zheap remains constant and all the undo data generated is removed within seconds of the transaction ending.

Below are some more tests where the transaction has been kept open for a much longer duration.

After running the test for 40 minutes (out of which there was an open transaction for first 30-minutes), the size in heap grows to 19GB at 8-client count test and to 26GB at 64-client count test whereas for zheap the size remains at 11GB for both the client-counts at the end of test and all the undo generated during test gets discarded within a few seconds after the open transaction is ended.

After running the test for 55 minutes (out of which there was an open transaction for first 45-minutes), the size in heap grows to 22GB at 8-client count test and to 28GB at 64-client count test whereas for zheap the size remains at 11GB for both the client-counts at the end of test and all the undo generated during test gets discarded within few seconds after the open transaction is ended.

So from all the above three tests, it is clear that the size of heap keeps on growing as the time for a concurrent long-running transaction is increasing.  It was 13GB at the start of the test, grew to 20GB, then to 26GB, then to 28GB at 64-client count test as the duration of the open transaction has increased from 15-mins to 30-mins and then to 45-mins. We have done a few more tests on the above lines and found that as the duration of open-transaction increases, the size of heap keeps on increasing whereas zheap remains constant.  For example, similar to above, if we keep the transaction open 60-mins in a 70-min test, the size of heap increases to 30GB. The increase in size also depends on the number of updates that are happening as part of the test.

The above results show not only the impact on size, but we also noticed that the TPS (transactions per second) in zheap is also always better (up to ~45%) for the above tests.  In similar tests on some other high-end machine, we see much better results with zheap with respect to performance. I would like to defer the details about raw-performance of zheap vs. heap to another blog post as this blog has already become big. I would like to mention that the above results don't mean that zheap will be better in all cases than heap. For example, rollbacks will be costlier in zheap. Just to be clear, this storage format is proposed as another format alongside current heap, so that users can decide which storage they want to use for their use case.

The code for this project has been published and is proposed as a feature for PG-12 to PostgreSQL community.  Thanks to Kuntal Ghosh for doing the performance tests mentioned in this blog post.

Hash indexes are faster than Btree indexes?

PostgreSQL have supported Hash Index for a long time, but they are not much used in production mainly because they are not durable.  Now, with the next version of PostgreSQL, they will be durable.  The immediate question is how do they perform as compared to Btree indexes. There is a lot of work done in the coming version to make them faster. There are multiple ways in which we can compare the performance of Hash and Btree indexes, like the time taken for creation of the index, search or insertion in the index.  This blog will mainly focus on the search operation. By definition, hash indexes are O(1) and Btree indexes are O(log n), however with duplicates that is not exactly true.

To start with let us see the impact of work being done to improve the performance of hash indexes. Below is the performance data of the pgbench read-only workload to compare the performance difference of Hash indexes between 9.6 and HEAD on IBM POWER-8 having 24 cores, 192 hardware threads, 492GB RAM.

The workload is such that all the data fits in shared buffers (scale factor is 300 (~4.5GB) and shared_buffers is 8GB).  As we can see from the above graph, that the performance has increased at all client counts in the range of 7% to 81% and the impact is more pronounced at higher client counts. The main work which has led to this improvement is 6d46f478 (Improve hash index bucket split behavior.) and 293e24e5 (Cache hash index's metapage in rel->rd_amcache.).

The first commit 6d46f478 has changed the heavyweight locks (locks that are used for logical database objects to ensure the database ACID properties) to lightweight locks (locks to protect shared data structures) for scanning the bucket pages.  In general, acquiring the heavyweight lock is costlier as compare to lightweight locks.  In addition to reducing the locking cost, this also avoids locking out scans and inserts for the lifetime of the split.

The second commit 293e24e5 avoids a significant amount of contention for accessing metapage. Each search operation needs to access metapage to find the bucket that contains tuple being searched which leads to high contention around metapage.  Each access to metapage needs to further access buffer manager. This work avoids that contention by caching the metapage information in backend local cache which helps bypassing all the buffer manager related work and hence the major contention in accessing the metapage.


The next graph shows how the hash index performs as compared to the btree index.  In this run we have changed hash to btree index in pgbench read-only tests.

We can see here that the hash index performs better than the btree index and the performance difference is in the range of 10 to 22%.  In some other workloads we have seen a better performance like with hash index on varchar columns and even in the community, it has been reported that there is performance improvement in the range of 40-60% when hash indexes are used for unique index columns.


The important thing to note about the above data is that it is only on some of the specific workloads and it mainly covers Selects as that is the main area where performance improvement work has been done for PostgreSQL10.  The other interesting parameters to compare are the size of the index and update on the index which needs more study and experiments.

In the end, I would like to thank my colleagues who were directly involved in this work and my employer EnterpriseDB who has supported this work.  Firstly I would like to thank, Robert Haas who has envisioned all this work and is the committer of this work, and Mithun C Y who was the author of commit 293e24e5.  Also, I would like to extend sincere thanks to all the community members who are involved in this work and especially Jeff Janes and Jesper Pedersen who have reviewed and tested this work.

Troubleshooting waits in PostgreSQL

Currently when the PostgreSQL database becomes slow especially on systems with high load, it becomes difficult to find the exact reasons.  Currently one can use tools like perf, strace, dynamic tracing (http://www.postgresql.org/docs/devel/static/dynamic-trace.html), etc. to find out the reasons of slowdown, but most of the times they are quite inconvenient to use which lead to the development of the new feature to display wait events information in pg_stat_activity view.  Wait events are invented to capture the information of system blocks or waits to perform some action like waiting for another backend process to release the heavyweight or lightweight locks, waits to access data buffer when no other process can be examining the buffer, waits to read or write the data to disk, etc.  As part of initial feature, we have covered some of the common wait event types due to which there are waits in system, however it is designed such that it can be extended to capture other types of wait events as well.

I will briefly explain the wait event types covered as part of this feature and then explain with examples, how one can use this feature to find stalls or waits in the system.  First wait event type is lightweight lock which is used to protect a particular data structure in shared memory.  Second wait event type is named lightweight lock tranche, this indicates that the server process is waiting for one of a group of related lightweight locks. Third wait event type is heavyweight lock which is used to primarily protect SQL-visible objects such as tables.  Fourth type of wait event is BufferPin where the server process waits to access to a data buffer during a period when no other process can be examining that buffer.  For detail explanation, refer PostgreSQL documentation at http://www.postgresql.org/docs/devel/static/monitoring-stats.html#PG-STAT-ACTIVITY-VIEW

Now, let us try to understand with the help of simple examples, how to find waits in the system using this powerful tool.

Create table and insert data which will be used in below examples:

postgres=# create table wait_event_tbl(c1 int);

CREATE TABLE

postgres=# insert into wait_event_tbl values(1);

INSERT 0 1

wait event type - Lock (Heavyweight locks)

-------------------------------------------------

Scenario - 1

Let us try to examine the waits for a scenario where one of the session has acquired Access Exclusive Lock on a table and the other session wants to acquire Access Share Lock on the same table and is waiting for first session to complete it's transaction.

Session -1

postgres=# select pg_backend_pid();

 pg_backend_pid

----------------

           6088

(1 row)

postgres=# begin;

BEGIN

postgres=# Lock wait_event_tbl in Access Exclusive Mode;

LOCK TABLE

Session-2

postgres=# select pg_backend_pid();

 pg_backend_pid

----------------

           1152

(1 row)

postgres=# begin;

BEGIN

postgres=# Lock wait_event_tbl in Access Share Mode;

Session-3

postgres=# select pid, wait_event_type, wait_event from pg_stat_activity where wait_event is NOT NULL;

 pid  | wait_event_type | wait_event

------+-----------------+------------

 1152 | Lock            | relation

(1 row)

Here, via above statement, it is shown that session-2 is waiting for a Lock on a relation.  To know more information about relation, one can add "query" column in the above statement.

Scenario - 2

Three sessions try to update the same row, first one will be successful and the other two will be waiting.

Session -1

postgres=# select pg_backend_pid();

 pg_backend_pid

----------------

           6088

(1 row)

postgres=# begin;

BEGIN

postgres=# update wait_event_tbl set c1 = 2 where c1=1;

UPDATE 1

Session - 2

postgres=# select pg_backend_pid();

 pg_backend_pid

----------------

           1152

(1 row)

postgres=# begin;

BEGIN

postgres=# update wait_event_tbl set c1 = 3 where c1 = 1;

Session - 3

postgres=# select pg_backend_pid();

 pg_backend_pid

----------------

           5404

(1 row)

postgres=# begin;

BEGIN

postgres=# update wait_event_tbl set c1 = 4 where c1 = 1;

Session - 4

postgres=# select pid, wait_event_type, wait_event from pg_stat_activity where wait_event is NOT NULL;

 pid  | wait_event_type |  wait_event

------+-----------------+---------------

 1152 | Lock            | transactionid

 5404 | Lock            | tuple

(2 rows)

Here, above statement indicates that session-2 and session-3 are waiting.

To find detailed information about locks, you can join this table information with pg_locks as described in link:https://wiki.postgresql.org/wiki/Lock_Monitoring or some other similar way.

wait event type - LWLockName (Lightweight Locks)

----------------------------------------------------

One session trying to execute the update statement and other session is trying to execute select statement can block each other for short time.

Session - 1

postgres=# select pg_backend_pid();

 pg_backend_pid

----------------

           1152

(1 row)

postgres=# update wait_event_tbl set c1 = 2;

Session - 2

postgres=# select pg_backend_pid();

 pg_backend_pid

----------------

           6088

(1 row)

postgres=# select * from wait_event_tbl;

Session - 3

postgres=# select pid, wait_event_type, wait_event from pg_stat_activity where wait_event is NOT NULL;

 pid  | wait_event_type |  wait_event

------+-----------------+---------------

 1152 | LWLockNamed     | ProcArrayLock

(1 row)

I have created this scenario with the help of debugger, but it is quite possible to see such wait events during high load on the system.

One point to note for users who are using "waiting" column of pg_stat_activity to find blocking statements is that they need to change their queries for next version (presumably 9.6) of PostgreSQL  as waiting column is removed from pg_stat_activity.  This is an intentional decision taken by PostgreSQL community for the ease of use and or understanding of this feature especially for future versions.

This feature has been committed in PostgreSQL code.  For details, you can refer commit id - 53be0b1add7064ca5db3cd884302dfc3268d884e.  It took us approximately 9 months to complete this feature.  Thanks to all the PostgreSQL community members who have given their valuable feedback throughout the development of this feature and special thanks to Robert Haas and Ildus Kurbangaliev for giving tremendous support to me both by reviews and by helping in writing parts of code.  Also Thanks to Alexander Korotkov for review and inputs for this feature and last but not least Thanks to Thom Brown for inputs in documentation of this feature.