################################################################################
# MDEV-32014 Rename binlog cache to binlog file
#
# It verifies that the binlog caches which are larger
# than binlog_large_commit_threshold can be move to a binlog file
# successfully. With a successful rename,
# - it rotates the binlog and the cache is renamed to the new binlog file
# - an ignorable event is generated just after the Gtid_log_event of the
#   transaction to take the reserved spaces which is unused.
#
# It also verifies that rename is not supported in below cases
# though the cache is larger than the threshold
# - both statement and transaction cache should be flushed.
# - the cache's checksum option is not same to binlog_checksum
# - binlog_legacy_event_pos is enabled.
################################################################################
--source include/have_binlog_format_row.inc
--source include/have_innodb.inc
--source include/master-slave.inc

--echo # Prepare
SET @saved_binlog_large_commit_threshold= @@GLOBAL.binlog_large_commit_threshold;
SET @saved_binlog_checksum= @@GLOBAL.binlog_checksum;

SET GLOBAL binlog_checksum = "NONE";

CREATE TABLE t1 (c1 LONGTEXT) ENGINE = InnoDB;
CREATE TABLE t2 (c1 LONGTEXT) ENGINE = MyISAM;

INSERT INTO t1 values(repeat("1", 5242880));
INSERT INTO t1 values(repeat("1", 5242880));
INSERT INTO t2 values(repeat("1", 5242880));
INSERT INTO t2 values(repeat("1", 5242880));

FLUSH BINARY LOGS;

--echo # Not renamed to binlog, since the binlog cache is not larger than the
--echo # threshold. And it should works well after ROLLBACK TO SAVEPOINT
BEGIN;
SAVEPOINT s1;
UPDATE t1 SET c1 = repeat('1', 5242880);
ROLLBACK TO SAVEPOINT s1;
UPDATE t1 SET c1 = repeat('2', 5242880);
SAVEPOINT s2;
UPDATE t1 SET c1 = repeat('3', 5242880);
UPDATE t1 SET c1 = repeat('4', 5242880);
ROLLBACK TO SAVEPOINT s2;
COMMIT;

--let $binlog_file= query_get_value(SHOW MASTER STATUS, File, 1)
--let $assert_cond= "$binlog_file" = "master-bin.000003"
--let $assert_text= Binlog is not rotated
--source include/assert.inc

--echo #
--echo # Test binlog cache rename to binlog file with checksum off
--echo #
--source include/sync_slave_sql_with_master.inc
--source include/stop_slave.inc

SET @saved_binlog_large_commit_threshold = @@GLOBAL.binlog_large_commit_threshold;
SET @saved_slave_parallel_workers = @@GLOBAL.slave_parallel_workers;
SET @saved_slave_parallel_mode = @@GLOBAL.slave_parallel_mode;
SET @saved_slave_parallel_max_queued = @@GLOBAL.slave_parallel_max_queued;

SET GLOBAL binlog_large_commit_threshold = 10 * 1024 * 1024;
SET GLOBAL slave_parallel_max_queued = 100 * 1024 * 1024;
SET GLOBAL slave_parallel_workers = 4;
SET GLOBAL slave_parallel_mode = "aggressive";
--source include/start_slave.inc
# Block all DML on slave
BEGIN;
DELETE FROM t1;

--connection master
SET GLOBAL binlog_large_commit_threshold = 10 * 1024 * 1024;

--echo # Transaction cache can be renamed and works well with ROLLBACK TO SAVEPOINT
BEGIN;
SAVEPOINT s1;
UPDATE t1 SET c1 = repeat('2', 5242880);
ROLLBACK TO s1;
UPDATE t1 SET c1 = repeat('3', 5242880);
SAVEPOINT s2;
UPDATE t1 SET c1 = repeat('4', 5242880);
UPDATE t1 SET c1 = repeat('5', 5242880);
UPDATE t1 SET c1 = repeat('6', 5242880);
ROLLBACK TO SAVEPOINT s2;
COMMIT;
INSERT INTO t1 VALUES("after_update_t1");

--let $gtid_end_pos= query_get_value(SHOW BINLOG EVENTS IN 'master-bin.000004' LIMIT 4, End_log_pos, 4)
--let $assert_cond= $gtid_end_pos = 4096
--let $assert_text= Rename is executed.
--source include/assert.inc

--echo # statement cache can be renamed
--connection master
BEGIN;
UPDATE t2 SET c1 = repeat('4', 5242880);
INSERT INTO t1 VALUES("after_update_t2");
COMMIT;
--let $gtid_end_pos= query_get_value(SHOW BINLOG EVENTS IN 'master-bin.000005' LIMIT 4, End_log_pos, 4)
--let $assert_cond= $gtid_end_pos = 4096
--let $assert_text= Rename is executed.
--source include/assert.inc

--connection slave
# UPDATE t2 should be waiting for prior transactions to commit.
let $wait_condition=
      SELECT count(*) = 1 FROM information_schema.processlist
      WHERE State = "Waiting for prior transaction to commit";
--source include/wait_condition.inc
ROLLBACK;

--connection master
--source include/sync_slave_sql_with_master.inc

--let $gtid_end_pos= query_get_value(SHOW BINLOG EVENTS IN 'slave-bin.000002' LIMIT 4, End_log_pos, 4)
--let $assert_cond= $gtid_end_pos = 4096
--let $assert_text= Rename is executed.
--source include/assert.inc

--let $gtid_end_pos= query_get_value(SHOW BINLOG EVENTS IN 'slave-bin.000003' LIMIT 4, End_log_pos, 4)
--let $assert_cond= $gtid_end_pos = 4096
--let $assert_text= Rename is executed.
--source include/assert.inc

--let $binlog_file= slave-bin.000002
--let $skip_checkpoint_events= 1
--source include/show_binlog_events.inc
--let $binlog_file= slave-bin.000003
--source include/show_binlog_events.inc

--source include/stop_slave.inc
SET GLOBAL binlog_large_commit_threshold = @saved_binlog_large_commit_threshold;
SET GLOBAL slave_parallel_workers = @saved_slave_parallel_workers;
SET GLOBAL slave_parallel_max_queued = @saved_slave_parallel_max_queued;
SET GLOBAL slave_parallel_mode = @saved_slave_parallel_mode;
--source include/start_slave.inc

--echo # CREATE SELECT works well
--connection master
CREATE TABLE t3 SELECT * FROM t1;
--let $gtid_end_pos= query_get_value(SHOW BINLOG EVENTS IN 'master-bin.000006' LIMIT 4, End_log_pos, 4)
--let $assert_cond= $gtid_end_pos = 4096
--let $assert_text= Rename is executed.
--source include/assert.inc

CREATE TABLE t4 SELECT * FROM t2;
--let $gtid_end_pos= query_get_value(SHOW BINLOG EVENTS IN 'master-bin.000007' LIMIT 4, End_log_pos, 4)
--let $assert_cond= $gtid_end_pos = 4096
--let $assert_text= Rename is executed.
--source include/assert.inc

--echo # XA statement works well
XA START "test-a-long-xid========================================";
UPDATE t1 SET c1 = repeat('1', 5242880);
XA END "test-a-long-xid========================================";
XA PREPARE "test-a-long-xid========================================";
XA COMMIT "test-a-long-xid========================================";
--let $gtid_end_pos= query_get_value(SHOW BINLOG EVENTS IN 'master-bin.000008' LIMIT 4, End_log_pos, 4)
--let $assert_cond= $gtid_end_pos = 4096
--let $assert_text= Rename is executed.
--source include/assert.inc

XA START "test-xid";
UPDATE t1 SET c1 = repeat('2', 5242880);
XA END "test-xid";
XA COMMIT "test-xid" ONE PHASE;
--let $gtid_end_pos= query_get_value(SHOW BINLOG EVENTS IN 'master-bin.000009' LIMIT 4, End_log_pos, 4)
--let $assert_cond= $gtid_end_pos = 4096
--let $assert_text= Rename is executed.
--source include/assert.inc

--echo #
--echo #  It works well in the situation that binlog header is larger than
--echo #  IO_SIZE and binlog file's buffer.
--echo #
--disable_query_log

# make Gtid_list_event larger than 64K(binlog file's buffer)
--let $server_id= 100000
while ($server_id < 104096)
{
  eval SET SESSION server_id = $server_id;
  eval UPDATE t1 SET c1 = "$server_id" LIMIT 1;
  --inc $server_id
}

--enable_query_log

# After flush, reserved space should be updated.
FLUSH BINARY LOGS;

SET SESSION server_id = 1;
UPDATE t1 SET c1 = repeat('3', 5242880);

--let $gtid_end_pos= query_get_value(SHOW BINLOG EVENTS IN 'master-bin.000011' LIMIT 4, End_log_pos, 4)
# 69632 is 65K which is larger, binlog's buffer is 64K
--let $assert_cond= $gtid_end_pos = 69632
--let $assert_text= Rename is executed.
--source include/assert.inc

--echo #
--echo # RESET MASTER should work well. It also verifies binlog checksum mechanism.
--echo #
--source include/rpl_reset.inc

--echo #
--echo # Test binlog cache rename to binlog file with checksum on
--echo #
SET GLOBAL binlog_checksum = "CRC32";

--echo # It will not rename the cache to file, since the cache's checksum was
--echo # initialized when reset the cache at the end of previous transaction.
UPDATE t1 SET c1 = repeat('5', 5242880);
--let $binlog_file= query_get_value(SHOW MASTER STATUS, File, 1)
--let $assert_cond= "$binlog_file" = "master-bin.000002"
--let $assert_text= Binlog is not rotated
--source include/assert.inc

--echo #
--echo # Not rename to binlog file If the cache's checksum is not same
--echo # to binlog_checksum
--echo #
BEGIN;
UPDATE t1 SET c1 = repeat('6', 5242880);
SET GLOBAL binlog_checksum = "NONE";
COMMIT;
--let $binlog_file= query_get_value(SHOW MASTER STATUS, File, 1)
--let $assert_cond= "$binlog_file" = "master-bin.000003"
--let $assert_text= Binlog is not rotated
--source include/assert.inc

BEGIN;
UPDATE t1 SET c1 = repeat('7', 5242880);
SET GLOBAL binlog_checksum = "CRC32";
COMMIT;
--let $binlog_file= query_get_value(SHOW MASTER STATUS, File, 1)
--let $assert_cond= "$binlog_file" = "master-bin.000004"
--let $assert_text= Binlog is not rotated
--source include/assert.inc

--echo #
--echo # Not rename to binlog file If both stmt and trx cache are not empty
--echo #
UPDATE t1, t2 SET t1.c1 = repeat('8', 5242880), t2.c1 = repeat('7', 5242880);
--let $binlog_file= query_get_value(SHOW MASTER STATUS, File, 1)
--let $assert_cond= "$binlog_file" = "master-bin.000004"
--let $assert_text= Binlog is not rotated
--source include/assert.inc

--echo #
--echo # Not rename to binlog file If binlog_legacy_event_pos is on
--echo #
SET GLOBAL binlog_legacy_event_pos = ON;
UPDATE t1 SET c1 = repeat('9', 5242880);
SET GLOBAL binlog_legacy_event_pos = OFF;
--let $binlog_file= query_get_value(SHOW MASTER STATUS, File, 1)
--let $assert_cond= "$binlog_file" = "master-bin.000004"
--let $assert_text= Binlog is not rotated
--source include/assert.inc

# cleanup
DROP TABLE t1, t2, t3, t4;
SET GLOBAL binlog_large_commit_threshold = @saved_binlog_large_commit_threshold;
SET GLOBAL binlog_checksum = @saved_binlog_checksum;
--let $binlog_file=
--let $skip_checkpoint_events=0
--source include/rpl_end.inc