PBS-42 feature: Implement basic support for COM_BINLOG_DUMP packet handling (part 5)

README.md

@@ -712,7 +712,7 @@ Running the utility for the second time (in any mode) results in resuming stream
 
 ### Graceful termination
 
-The user can request the utility operating in either `fetch` or `pull` mode to be gracefully terminated leaving storage in consistent state. For this, the utility sets custom handlers for the the following POSIX signals.
+The user can request the utility operating in either `fetch` or `pull` mode to be gracefully terminated leaving storage in consistent state. For this, the utility sets custom handlers for the following POSIX signals.
 - `SIGINT` - for processing `^C` in console.
 - `SIGTERM` - for processing `kill <pid>`.
 

mtr/binlog_streaming/include/rs_compare_dumps.inc

@@ -0,0 +1,61 @@
+# ==== Purpose ====
+#
+# Dumps binary logs starting from the specified (binlog name, position)
+# checkpoint up to the last binlog file both from the Binlog Server
+# replication source and directly from the MySQL Server and compares
+# generated event sequences.
+#
+# ==== Usage ====
+#
+# --let $checkpoint_label = after inserting the first batch of rows
+# --let $checkpoint_binlog_name = binlog.000001
+# --let $checkpoint_binlog_position = 4
+# [--let $saved_roundtrip_dump = $MYSQL_TMP_DIR/roundtrip.sql]
+# --source ../include/rs_compare_dumps.inc
+#
+# An empty $checkpoint_binlog_name means "start from the oldest binlog file
+# available" (in this case $checkpoint_binlog_position must be 4).
+#
+# If $saved_roundtrip_dump is specified, the unfiltered dump received from
+# the Binlog Server replication source is copied to this location before
+# the cleanup (e.g. to be restored later). It is the responsibility of the
+# caller to remove this file. $saved_roundtrip_dump is reset at the end.
+#
+# Expected inputs set by set_up_binsrv_environment.inc:
+#   $binsrv_connection_host
+#   $binsrv_connection_user
+#   $binsrv_replication_source_port
+#   $binsrv_auth_user
+#   $binsrv_auth_password
+
+--echo
+--echo *** Comparing event sequences from the PBS and MySQL Server starting
+--echo *** from checkpoint '$checkpoint_label'.
+--let $checkpoint_roundtrip_dump = $MYSQL_TMP_DIR/rs_compare_dumps_roundtrip.sql
+exec $MYSQL_BINLOG
+  --read-from-remote-server --host=$binsrv_connection_host --port=$binsrv_replication_source_port
+  --user=$binsrv_auth_user --password=$binsrv_auth_password --default-auth=caching_sha2_password
+  --start-position=$checkpoint_binlog_position --to-last-log "$checkpoint_binlog_name" > $checkpoint_roundtrip_dump 2>/dev/null;
+--let $checkpoint_roundtrip_dump_filtered = $checkpoint_roundtrip_dump.filtered
+--exec cat $checkpoint_roundtrip_dump | grep -E "^#[[:digit:]]+" > $checkpoint_roundtrip_dump_filtered
+
+--let $checkpoint_direct_dump = $MYSQL_TMP_DIR/rs_compare_dumps_direct.sql
+exec $MYSQL_BINLOG
+  --read-from-remote-server --host=$binsrv_connection_host --user=$binsrv_connection_user
+  --start-position=$checkpoint_binlog_position --to-last-log "$checkpoint_binlog_name" > $checkpoint_direct_dump 2>/dev/null;
+--let $checkpoint_direct_dump_filtered = $checkpoint_direct_dump.filtered
+--exec cat $checkpoint_direct_dump | grep -E "^#[[:digit:]]+" > $checkpoint_direct_dump_filtered
+
+--diff_files $checkpoint_roundtrip_dump_filtered $checkpoint_direct_dump_filtered
+
+if ($saved_roundtrip_dump)
+{
+  --copy_file $checkpoint_roundtrip_dump $saved_roundtrip_dump
+}
+
+--remove_file $checkpoint_roundtrip_dump
+--remove_file $checkpoint_roundtrip_dump_filtered
+--remove_file $checkpoint_direct_dump
+--remove_file $checkpoint_direct_dump_filtered
+
+--let $saved_roundtrip_dump =

mtr/binlog_streaming/include/rs_position_based_consistency_save.inc

@@ -0,0 +1,40 @@
+# ==== Purpose ====
+#
+# Appends a (binlog name, position) checkpoint to the '@checkpoints'
+# JSON array session variable initialized by
+# 'rs_position_based_consistency.test'. Setting a user variable does not
+# write anything to the binary log, so saving a checkpoint does not affect
+# the content of binary log files.
+#
+# ==== Usage ====
+#
+# --let $checkpoint_label = after inserting the first batch of rows
+# [--let $checkpoint_binlog_name = binlog.000001]
+# [--let $checkpoint_binlog_position = 4]
+# --source ../include/rs_position_based_consistency_save.inc
+#
+# If $checkpoint_binlog_position is not specified, the current binary log
+# name and position are taken from $stmt_show_binary_log_status (set by
+# v80_v84_compatibility_defines.inc). If $checkpoint_binlog_position is
+# specified, $checkpoint_binlog_name is used as is (an empty
+# $checkpoint_binlog_name with position 4 means "from the beginning of the
+# oldest binlog file").
+#
+# The $checkpoint_label must not contain single quotes.
+
+if ($checkpoint_binlog_position == "")
+{
+  --let $checkpoint_binlog_name = query_get_value($stmt_show_binary_log_status, File, 1)
+  --let $checkpoint_binlog_position = query_get_value($stmt_show_binary_log_status, Position, 1)
+}
+
+--disable_query_log
+eval SET @checkpoints = JSON_ARRAY_APPEND(@checkpoints, '\$',
+  JSON_OBJECT('binlog_file', '$checkpoint_binlog_name',
+              'position', $checkpoint_binlog_position,
+              'test_label', '$checkpoint_label'));
+--enable_query_log
+
+--let $checkpoint_binlog_name =
+--let $checkpoint_binlog_position =
+--let $checkpoint_label =

mtr/binlog_streaming/include/rs_roundtrip_body.inc

@@ -14,10 +14,6 @@
 #   $binsrv_replication_source_port
 #   $stmt_show_binary_log_status
 
---echo
---echo *** Determining the first binary log name.
---let $first_binlog = query_get_value($stmt_show_binary_log_status, File, 1)
-
 --echo
 --echo *** Creating source_db.tbl and inserting the first batch of rows.
 CREATE DATABASE source_db;
@@ -29,36 +25,81 @@ CREATE TABLE source_db.tbl(
 INSERT INTO source_db.tbl VALUES (1, 'one'), (2, 'two'), (3, 'three');
 
 --echo
---echo *** Flushing the source binary log.
+--echo *** Flushing the source binary log for the first time.
 FLUSH BINARY LOGS;
 
 --echo
 --echo *** Inserting the second batch of rows.
 INSERT INTO source_db.tbl VALUES (4, 'four'), (5, 'five'), (6, 'six');
 
 --echo
---echo *** Fetching the source binary logs in position mode.
+--echo *** Determining binary log name and position for the second part.
+--let $second_part_binlog_name = query_get_value($stmt_show_binary_log_status, File, 1)
+--let $second_part_binlog_position = query_get_value($stmt_show_binary_log_status, Position, 1)
+
+--echo
+--echo *** Fetching the source binary logs in position mode for the first time.
 --exec $BINSRV fetch $binsrv_config_file_path > /dev/null
 
 --echo
---echo *** Starting Binlog Server Utility in background in pull mode.
+--echo *** Starting Binlog Server Utility in background in pull mode for the first time.
 --let $proc_command_line = $BINSRV pull $binsrv_config_file_path > /dev/null
 --source ../include/start_proc_in_background.inc
 
 --echo
 --echo *** Waiting for the replication source listener port ready to accept
---echo *** connections.
+--echo *** connections for the first time.
 --let $listening_port = $binsrv_replication_source_port
 --source ../include/wait_for_listening_port.inc
 
+--let $first_roundtrip_dump = $MYSQL_TMP_DIR/first_rs_roundtrip.sql
+--let $checkpoint_label = the beginning of the oldest binlog file
+--let $checkpoint_binlog_name =
+--let $checkpoint_binlog_position = 4
+--let $saved_roundtrip_dump = $first_roundtrip_dump
+--source ../include/rs_compare_dumps.inc
+
+--echo
+--echo *** Sending SIGTERM signal to the Binlog Server Utility and waiting for
+--echo *** the process to terminate for the first time.
+--source ../include/terminate_proc.inc
+
+--echo
+--echo *** Inserting the third batch of rows.
+INSERT INTO source_db.tbl VALUES (7, 'seven'), (8, 'eight'), (9, 'nine');
+
+--echo
+--echo *** Flushing the source binary log for the second time.
+FLUSH BINARY LOGS;
+
 --echo
---echo *** Dumping the binary logs from the Binlog Server replication source.
---let $roundtrip_dump = $MYSQL_TMP_DIR/rs_roundtrip.sql
---exec $MYSQL_BINLOG --read-from-remote-server --host=127.0.0.1 --port=$binsrv_replication_source_port --user=rpl --password=password --default-auth=caching_sha2_password $first_binlog > $roundtrip_dump 2>/dev/null
+--echo *** Inserting the fourth batch of rows.
+INSERT INTO source_db.tbl VALUES (10, 'ten'), (11, 'eleven'), (12, 'twelve');
+
+--echo
+--echo *** Fetching the source binary logs in position mode for the second time.
+--exec $BINSRV fetch $binsrv_config_file_path > /dev/null
+
+--echo *** Starting Binlog Server Utility in background in pull mode for the second time.
+--let $proc_command_line = $BINSRV pull $binsrv_config_file_path > /dev/null
+--source ../include/start_proc_in_background.inc
+
+--echo
+--echo *** Waiting for the replication source listener port ready to accept
+--echo *** connections for the second time.
+--let $listening_port = $binsrv_replication_source_port
+--source ../include/wait_for_listening_port.inc
+
+--let $second_roundtrip_dump = $MYSQL_TMP_DIR/second_rs_roundtrip.sql
+--let $checkpoint_label = after inserting the second batch of rows
+--let $checkpoint_binlog_name = $second_part_binlog_name
+--let $checkpoint_binlog_position = $second_part_binlog_position
+--let $saved_roundtrip_dump = $second_roundtrip_dump
+--source ../include/rs_compare_dumps.inc
 
 --echo
 --echo *** Sending SIGTERM signal to the Binlog Server Utility and waiting for
---echo *** the process to terminate
+--echo *** the process to terminate for the second time.
 --source ../include/terminate_proc.inc
 
 --echo
@@ -68,8 +109,9 @@ RENAME TABLE source_db.tbl TO backup_db.tbl;
 DROP DATABASE source_db;
 
 --echo
---echo *** Restoring source_db from the replication source dump.
---exec $MYSQL < $roundtrip_dump
+--echo *** Restoring source_db from two replication source dumps.
+--exec $MYSQL < $first_roundtrip_dump
+--exec $MYSQL < $second_roundtrip_dump
 
 --echo
 --echo *** Comparing the restored and backup tables.
@@ -81,4 +123,7 @@ DROP DATABASE source_db;
 DROP DATABASE source_db;
 DROP DATABASE backup_db;
 
---remove_file $roundtrip_dump
+--echo
+--echo *** Cleaning up.
+--remove_file $first_roundtrip_dump
+--remove_file $second_roundtrip_dump

mtr/binlog_streaming/r/rs_position_based_consistency.result

@@ -0,0 +1,106 @@
+*** Resetting replication at the very beginning of the test.
+
+*** Generating a configuration file in JSON format for the Binlog
+*** Server utility.
+
+*** Determining binlog file directory from the server.
+
+*** Creating a temporary directory <BINSRV_STORAGE_PATH> for storing
+*** binlog files downloaded via the Binlog Server utility.
+
+*** Initializing the @checkpoints session variable (a JSON array) for
+*** storing (binlog name, position) checkpoints.
+
+*** Saving the ["":4] checkpoint (everything from the beginning of the
+*** oldest binary log file).
+
+*** Step 0: Saving the initial binary log position.
+
+*** Step 1: Creating tbl.
+CREATE TABLE tbl(
+id INT UNSIGNED NOT NULL,
+value VARCHAR(32) NOT NULL,
+PRIMARY KEY(id)
+) ENGINE=InnoDB;
+
+*** Step 2: Inserting the first batch of rows.
+INSERT INTO tbl VALUES (1, 'one'), (2, 'two'), (3, 'three');
+
+*** Step 3: Flushing the source binary log for the first time.
+FLUSH BINARY LOGS;
+
+*** Step 4: Inserting the second batch of rows.
+INSERT INTO tbl VALUES (4, 'four'), (5, 'five'), (6, 'six');
+
+*** Step 5: Flushing the source binary log for the second time.
+FLUSH BINARY LOGS;
+
+*** Step 6: Flushing the source binary log for the third time (producing an
+*** empty binary log file).
+FLUSH BINARY LOGS;
+
+*** Step 7: Inserting the third batch of rows.
+INSERT INTO tbl VALUES (7, 'seven'), (8, 'eight'), (9, 'nine');
+
+*** Saving end-of-file positions (sizes) of all binary log files as
+*** checkpoints.
+include/assert.inc [There must be exactly 4 binary log files]
+
+*** Fetching the source binary logs in position mode.
+
+*** Starting Binlog Server Utility in background in pull mode.
+include/read_file_to_var.inc
+
+*** Waiting for the replication source listener port ready to accept
+*** connections.
+
+*** Comparing event sequences from the PBS and MySQL Server starting
+*** from checkpoint 'from the beginning of the oldest binary log file'.
+
+*** Comparing event sequences from the PBS and MySQL Server starting
+*** from checkpoint 'at the very beginning'.
+
+*** Comparing event sequences from the PBS and MySQL Server starting
+*** from checkpoint 'after creating tbl'.
+
+*** Comparing event sequences from the PBS and MySQL Server starting
+*** from checkpoint 'after inserting the first batch of rows'.
+
+*** Comparing event sequences from the PBS and MySQL Server starting
+*** from checkpoint 'after the first FLUSH BINARY LOGS'.
+
+*** Comparing event sequences from the PBS and MySQL Server starting
+*** from checkpoint 'after inserting the second batch of rows'.
+
+*** Comparing event sequences from the PBS and MySQL Server starting
+*** from checkpoint 'after the second FLUSH BINARY LOGS'.
+
+*** Comparing event sequences from the PBS and MySQL Server starting
+*** from checkpoint 'after the third FLUSH BINARY LOGS'.
+
+*** Comparing event sequences from the PBS and MySQL Server starting
+*** from checkpoint 'after inserting the third batch of rows'.
+
+*** Comparing event sequences from the PBS and MySQL Server starting
+*** from checkpoint 'end of binary log file 1'.
+
+*** Comparing event sequences from the PBS and MySQL Server starting
+*** from checkpoint 'end of binary log file 2'.
+
+*** Comparing event sequences from the PBS and MySQL Server starting
+*** from checkpoint 'end of binary log file 3'.
+
+*** Comparing event sequences from the PBS and MySQL Server starting
+*** from checkpoint 'end of binary log file 4'.
+
+*** Sending SIGTERM signal to the Binlog Server Utility and waiting for
+*** the process to terminate.
+
+*** Dropping the table.
+DROP TABLE tbl;
+
+*** Removing the Binlog Server utility storage directory.
+
+*** Removing the Binlog Server utility log file.
+
+*** Removing the Binlog Server utility configuration file.