#
#   Ensure that executing row-injected events (i.e. via BINLOG statments and
# row-based binlog events) uses historical partitions. That is, for tables
# which use system versioning and system_time partitions, MDEV-35096 reported
# that row-injected events would not be stored into the correct historical
# partition. This test considers both use cases of row-injected events.
#
#   The test setup creates a system-versioned table with system_time-based
# partitioning and fills the table up with enough records that bypass the size
# limit of each historical partition.
#
#   To test BINLOG statements, a series of BINLOG statements are used to delete
# all the records in the test tables, and the resulting partitions are analyzed
# to ensure that they match the partition specification.  The BINLOG events
# were collected by running an original set of delete statements on the table
# data, and taking their binlog data from mysqlbinlog. Note these binary log
# events are actually Update events, because system versioning just archives
# the rows, rather than deleting them.
#
#   To test row-based event replication, a slave replicates the master's
# events, and the partitions are compared between the slave and master for
# consistency.
#
# Note that the TIMESTAMP of this test is fixed so the BINLOG statements can
# identify the correct rows to delete (system versioning adds implicit fields
# `row_start` and `row_end`, which are automatically populated using the current
# timestamp).
#
#
# References:
#   MDEV-35096: History is stored in different partitions on different nodes
#               when using SYSTEM VERSION
#
--source include/have_64bit_timestamp.inc
--source include/have_binlog_format_row.inc
--source include/have_innodb.inc
--source include/have_partition.inc
--source include/master-slave.inc

--echo #
--echo # Initialize system-versioned and partitioned table and its data
--connection master

# Fix the timestamp for the system versioned row_start and row_end fields, so
# the later hard-coded BINLOG base64 data can find the rows.
SET timestamp=UNIX_TIMESTAMP('2025-01-01 01:00:00.000000');
RESET MASTER;

create table t1 (x int) engine=InnoDB with system versioning partition by system_time limit 3 partitions 5;
insert into t1 values(1);
insert into t1 values(2);
insert into t1 values(3);
insert into t1 values(4);
insert into t1 values(5);
--let $master_total_size= `select count(*) from t1`
--let $master_p0_size= `select count(*) from t1 partition (p0)`
--let $master_p1_size= `select count(*) from t1 partition (p1)`
--let $master_p2_size= `select count(*) from t1 partition (p2)`

--echo # Verifying master partitions are correct after data insertion..
if ($master_total_size != 5)
{
  --echo # Master t1 count: $master_total_size
  --die Master table t1 should have 5 entries
}
if ($master_p0_size)
{
  --echo # Master t1,p0 count: $master_p0_size
  --die Master t1 partition p0 should be empty
}
if ($master_p1_size)
{
  --echo # Master t1,p1 count: $master_p1_size
  --die Master t1 partition p1 should be empty
}
if ($master_p2_size)
{
  --echo # Master t1,p2 count: $master_p2_size
  --die Master t1 partition p2 should be empty
}
--echo # .. done

--sync_slave_with_master

--connection slave
--let $slave_total_size= `select count(*) from t1`
--let $slave_p0_size= `select count(*) from t1 partition (p0)`
--let $slave_p1_size= `select count(*) from t1 partition (p1)`
--let $slave_p2_size= `select count(*) from t1 partition (p2)`

--echo # Verifying partitions of master and slave match on data setup..
if ($slave_total_size != $master_total_size)
{
  --connection master
  select count(*) from t0;
  --connection slave
  select count(*) from t1;
  --die Size of t1 differs between master and slave
}
if ($slave_p0_size != $master_p0_size)
{
  --connection master
  select count(*) from t1 partition (p0);
  --connection slave
  select count(*) from t1 partition (p0);
  --die Size of t1 partition p0 differs between master and slave
}
if ($slave_p1_size != $master_p1_size)
{
  --connection master
  select count(*) from t1 partition (p1);
  --connection slave
  select count(*) from t1 partition (p1);
  --die Size of t1 partition p1 differs between master and slave
}
if ($slave_p2_size != $master_p2_size)
{
  --connection master
  select count(*) from t1 partition (p2);
  --connection slave
  select count(*) from t1 partition (p2);
  --die Size of t1 partition p2 differs between master and slave
}
--echo # .. done

--echo #
--echo # "Delete" each row -- these are the BINLOG commands generated by
--echo # mysqlbinlog from `delete from t1 where x=<n>` statments. Because the
--echo # table uses system versioning and system_time partition, the actual
--echo # events are updates, with added fields for the `row_start` and `row_end`
--echo # columns.
--connection master

--echo # BINLOG for Format Description event
BINLOG '
APZ0Zw8BAAAA/AAAAAABAAAAAAQAMTEuNy4yLU1hcmlhREItZGVidWctbG9nAAAAAAAAAAAAAAAA
AAAAAAAAAAAAAAAAAAAA9nRnEzgNAAgAEgAEBAQEEgAA5AAEGggAAAAICAgCAAAACgoKAAAAAAAA
CgoKAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA
AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA
AAAAAAAAAAAEEwQADQgICAoKCgGuUmo6
';

--echo # BINLOG for delete from t1 where x=1;
BINLOG '
APZ0ZxMBAAAAMQAAAAAAAAAAACEAAAAAAAEABHRlc3QAAnQxAAMDERECBgYBoCHyJQ==
APZ0ZxgBAAAASAAAAAAAAAAAACEAAAAAAAEAAwcH+AEAAABndPYAAAAA/////w9CP/gBAAAAZ3T2
AAAAAGd09gAAAAAuqwNP
';

--echo # BINLOG for delete from t1 where x=2;
BINLOG '
APZ0ZxMBAAAAMQAAAAAAAAAAACEAAAAAAAEABHRlc3QAAnQxAAMDERECBgYBoCHyJQ==
APZ0ZxgBAAAASAAAAAAAAAAAACEAAAAAAAEAAwcH+AIAAABndPYAAAAA/////w9CP/gCAAAAZ3T2
AAAAAGd09gAAAAAsSeT/
';


--echo # BINLOG for delete from t1 where x=3;
BINLOG '
APZ0ZxMBAAAAMQAAAAAAAAAAACEAAAAAAAEABHRlc3QAAnQxAAMDERECBgYBoCHyJQ==
APZ0ZxgBAAAASAAAAAAAAAAAACEAAAAAAAEAAwcH+AMAAABndPYAAAAA/////w9CP/gDAAAAZ3T2
AAAAAGd09gAAAADS6EaQ
';

--echo # BINLOG for delete from t1 where x=4;
BINLOG '
APZ0ZxMBAAAAMQAAAAAAAAAAACEAAAAAAAEABHRlc3QAAnQxAAMDERECBgYBoCHyJQ==
APZ0ZxgBAAAASAAAAAAAAAAAACEAAAAAAAEAAwcH+AQAAABndPYAAAAA/////w9CP/gEAAAAZ3T2
AAAAAGd09gAAAABpi1pF
';

--echo # BINLOG for delete from t1 where x=5;
BINLOG '
APZ0ZxMBAAAAMQAAAAAAAAAAACEAAAAAAAEABHRlc3QAAnQxAAMDERECBgYBoCHyJQ==
APZ0ZxgBAAAASAAAAAAAAAAAACEAAAAAAAEAAwcH+AUAAABndPYAAAAA/////w9CP/gFAAAAZ3T2
AAAAAGd09gAAAACXKvgq
';

--let $master_total_size= `select count(*) from t1`
--let $master_p0_size= `select count(*) from t1 partition (p0)`
--let $master_p1_size= `select count(*) from t1 partition (p1)`
--let $master_p2_size= `select count(*) from t1 partition (p2)`
--echo # Verifying master partitions are correct after deletion BINLOG stmts..
if ($master_total_size > 0)
{
  --echo # Master t1 count: $master_total_size
  --die Master table t1 should have 0 count
}
if ($master_p0_size != 3)
{
  --echo # Master t1,p0 count: $master_p0_size
  --die Master t1 partition p0 should have 3 entries
}
if ($master_p1_size != 2)
{
  --echo # Master t1,p1 count: $master_p1_size
  --die Master t1 partition p1 should have 2 entries
}
if ($master_p2_size)
{
  --echo # Master t1,p2 count: $master_p2_size
  --die Master t1 partition p2 should be empty
}
--echo # .. done
--sync_slave_with_master

--connection slave
--let $slave_total_size= `select count(*) from t1`
--let $slave_p0_size= `select count(*) from t1 partition (p0)`
--let $slave_p1_size= `select count(*) from t1 partition (p1)`
--let $slave_p2_size= `select count(*) from t1 partition (p2)`

if ($slave_total_size != $master_total_size)
{
  --connection master
  select count(*) from t1;
  --connection slave
  select count(*) from t1;
  --die Size of t1 differs between master and slave
}
if ($slave_p0_size != $master_p0_size)
{
  --connection master
  select count(*) from t1 partition (p0);
  --connection slave
  select count(*) from t1 partition (p0);
  --die Size of t1 partition p0 differs between master and slave
}
if ($slave_p1_size != $master_p1_size)
{
  --connection master
  select count(*) from t1 partition (p1);
  --connection slave
  select count(*) from t1 partition (p1);
  --die Size of t1 partition p1 differs between master and slave
}
if ($slave_p2_size != $master_p2_size)
{
  --connection master
  select count(*) from t1 partition (p2);
  --connection slave
  select count(*) from t1 partition (p2);
  --die Size of t1 partition p2 differs between master and slave
}

--connection master
drop table t1;

--source include/rpl_end.inc