Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Make logrepl markers for (partial) decompressions
Added logical replication messages (PG14+) as markers for (partial) decompression events (mutual compression), which makes it possible to differentiate inserts happening as part of the decompression vs actual inserts by the user, and filter the former out of the event stream. While some tools may be interested in all events, synching the pure "state" (without internal behavior) is required for others.
- Loading branch information
Showing
3 changed files
with
218 additions
and
1 deletion.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,173 @@ | ||
| # This file and its contents are licensed under the Timescale License. | ||
| # Please see the included NOTICE for copyright information and | ||
| # LICENSE-TIMESCALE for a copy of the license. | ||
|
|
||
| if ($ENV{PG_VERSION_MAJOR} < 14) | ||
| { | ||
| done_testing(); | ||
| return; | ||
| } | ||
|
|
||
| use strict; | ||
| use warnings; | ||
| use TimescaleNode; | ||
| use Test::More; | ||
|
|
||
| # This test checks the creation of logical replication messages | ||
| # used to mark the start and end of inserts happening as a result | ||
| # of a (partial) decompression. | ||
|
|
||
| # Publishing node | ||
| my $publisher = | ||
| TimescaleNode->create('publisher', allows_streaming => 'logical'); | ||
|
|
||
| # Subscribing node | ||
| my $subscriber = | ||
| TimescaleNode->create('subscriber', allows_streaming => 'logical'); | ||
|
|
||
| # Setup test structures | ||
| $publisher->safe_psql( | ||
| 'postgres', | ||
| qq( | ||
| CREATE TABLE test (ts timestamptz NOT NULL PRIMARY KEY , val INT); | ||
| SELECT create_hypertable('test', 'ts', chunk_time_interval := INTERVAL '1day'); | ||
| ) | ||
| ); | ||
|
|
||
| # To kick off replication we need to fake the setup of a hypertable | ||
| $subscriber->safe_psql('postgres', | ||
| "CREATE TABLE _timescaledb_internal._hyper_1_1_chunk (ts timestamptz NOT NULL PRIMARY KEY , val INT)" | ||
| ); | ||
|
|
||
| # Initial data insert and preparation of the internal chunk tables | ||
| $publisher->safe_psql( | ||
| 'postgres', | ||
| qq( | ||
| INSERT INTO test | ||
| SELECT s.s, (random() * 100)::INT | ||
| FROM generate_series('2023-01-01'::timestamptz, '2023-01-02'::timestamptz, INTERVAL '3 hour') s; | ||
| ) | ||
| ); | ||
|
|
||
| # Setup logical replication | ||
| my $publisher_connstr = $publisher->connstr . ' dbname=postgres'; | ||
| $publisher->safe_psql('postgres', | ||
| "CREATE PUBLICATION tap_pub FOR TABLE _timescaledb_internal._hyper_1_1_chunk" | ||
| ); | ||
| $subscriber->safe_psql('postgres', | ||
| "CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr' PUBLICATION tap_pub WITH (binary = true)" | ||
| ); | ||
|
|
||
| # Wait for catchup and disable consumption of additional messages | ||
| $publisher->wait_for_catchup('tap_sub'); | ||
| $subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION tap_sub DISABLE"); | ||
| $publisher->poll_query_until( | ||
| 'postgres', | ||
| "SELECT COUNT(*) FROM pg_catalog.pg_replication_slots WHERE slot_name = 'tap_sub' AND active='f'", | ||
| 1); | ||
|
|
||
| # Compress chunks and consume replication stream explicitly | ||
| $publisher->safe_psql( | ||
| 'postgres', | ||
| qq( | ||
| ALTER TABLE test SET (timescaledb.compress); | ||
| SELECT compress_chunk((c.schema_name || '.' || c.table_name)::regclass, TRUE) | ||
| FROM _timescaledb_catalog.hypertable h | ||
| LEFT JOIN _timescaledb_catalog.chunk c ON c.hypertable_id = h.id WHERE h.table_name = 'test'; | ||
| ) | ||
| ); | ||
| $publisher->safe_psql( | ||
| 'postgres', | ||
| qq( | ||
| SELECT pg_logical_slot_get_binary_changes('tap_sub', NULL, NULL, | ||
| 'proto_version', '1', | ||
| 'publication_names', 'tap_pub', | ||
| 'messages', 'true' | ||
| ); | ||
| ) | ||
| ); | ||
|
|
||
| # Create a new entry which forces a decompression to happen | ||
| $publisher->safe_psql('postgres', | ||
| "INSERT INTO test VALUES ('2023-01-01 00:10:00', 5555)"); | ||
|
|
||
| # Retrieve the replication log messages | ||
| my $result = $publisher->safe_psql( | ||
| 'postgres', | ||
| qq( | ||
| SELECT get_byte(data, 0) | ||
| FROM pg_logical_slot_peek_binary_changes('tap_sub', NULL, NULL, | ||
| 'proto_version', '1', | ||
| 'publication_names', 'tap_pub', | ||
| 'messages', 'true' | ||
| ); | ||
| ) | ||
| ); | ||
|
|
||
| # Test: BEGIN, MESSAGE (start marker), RELATION, ... INSERT (decompression inserts x6) ..., MESSAGE (end marker), INSERT, COMMIT | ||
| is( $result, | ||
| qq(66 | ||
| 77 | ||
| 82 | ||
| 73 | ||
| 73 | ||
| 73 | ||
| 73 | ||
| 73 | ||
| 73 | ||
| 77 | ||
| 73 | ||
| 67), | ||
| 'messages on slot meet expectation <<BEGIN, MESSAGE (start marker), RELATION, ... INSERT (decompression inserts x6) ..., MESSAGE (end marker), INSERT, COMMIT>>' | ||
| ); | ||
|
|
||
| # Get initial message entry | ||
| $result = $publisher->safe_psql( | ||
| 'postgres', | ||
| qq( | ||
| SELECT get_byte(data, 1), encode(substr(data, 11, 33), 'escape') | ||
| FROM pg_logical_slot_peek_binary_changes('tap_sub', NULL, NULL, | ||
| 'proto_version', '1', | ||
| 'publication_names', 'tap_pub', | ||
| 'messages', 'true' | ||
| ) | ||
| OFFSET 1 LIMIT 1; | ||
| ) | ||
| ); | ||
| is( $result, | ||
| qq(1|::timescaledb-decompression-start), | ||
| 'first entry is decompression marker start message'); | ||
|
|
||
| # Get second message entry | ||
| $result = $publisher->safe_psql( | ||
| 'postgres', | ||
| qq( | ||
| SELECT get_byte(data, 1), encode(substr(data, 11, 31), 'escape') | ||
| FROM pg_logical_slot_peek_binary_changes('tap_sub', NULL, NULL, | ||
| 'proto_version', '1', | ||
| 'publication_names', 'tap_pub', | ||
| 'messages', 'true' | ||
| ) | ||
| OFFSET 9 LIMIT 1; | ||
| ) | ||
| ); | ||
| is( $result, | ||
| qq(1|::timescaledb-decompression-end), | ||
| '10th entry is decompression marker end message'); | ||
|
|
||
| # Get last insert entry to check it is the user executed insert (and value is 5555 or 35353535 in hex) | ||
| $result = $publisher->safe_psql( | ||
| 'postgres', | ||
| qq( | ||
| SELECT get_byte(data, 0), encode(substring(data from 41 for 44), 'hex') | ||
| FROM pg_logical_slot_peek_binary_changes('tap_sub', NULL, NULL, | ||
| 'proto_version', '1', | ||
| 'publication_names', 'tap_pub', | ||
| 'messages', 'true' | ||
| ) | ||
| OFFSET 10 LIMIT 1; | ||
| ) | ||
| ); | ||
| is($result, qq(73|35353535), '11th entry is an insert message'); | ||
|
|
||
| done_testing(); |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters