From fdc6de263a58480519d683e67bb3dfa262d9b230 Mon Sep 17 00:00:00 2001 From: Dan Goodliffe Date: Thu, 5 Aug 2021 23:30:09 +0100 Subject: Handle gtid and xid transaction events --- lib/output/pq/updateDatabase.cpp | 30 ++++++++++++++++++++--- lib/output/pq/updateDatabase.h | 3 +++ test/sql/simpleCreate.sql | 5 ++++ test/sql/simpleDelete.sql | 2 ++ test/sql/simpleDeleteSome.sql | 2 ++ test/sql/simpleInsert.sql | 1 + test/sql/simpleSelectAll.sql | 3 +++ test/sql/simpleUpdate.sql | 1 + test/sql/simpleUpdateAll.sql | 1 + test/test-e2e.cpp | 51 ++++++++++++++++++++++++++++++++++++++++ 10 files changed, 96 insertions(+), 3 deletions(-) create mode 100644 test/sql/simpleCreate.sql create mode 100644 test/sql/simpleDelete.sql create mode 100644 test/sql/simpleDeleteSome.sql create mode 100644 test/sql/simpleInsert.sql create mode 100644 test/sql/simpleSelectAll.sql create mode 100644 test/sql/simpleUpdate.sql create mode 100644 test/sql/simpleUpdateAll.sql diff --git a/lib/output/pq/updateDatabase.cpp b/lib/output/pq/updateDatabase.cpp index 232e8f9..ea8797b 100644 --- a/lib/output/pq/updateDatabase.cpp +++ b/lib/output/pq/updateDatabase.cpp @@ -27,6 +27,8 @@ #include namespace MyGrate::Output::Pq { + constexpr uint8_t STANDALONE {1}; + #ifndef __cpp_aggregate_paren_init ColumnDef::ColumnDef(std::string n, std::size_t o, bool p) : name {std::move(n)}, ordinal(o), is_pk(p) { } #endif @@ -266,14 +268,18 @@ namespace MyGrate::Output::Pq { void UpdateDatabase::beforeEvent(const MariaDB_Event_Ptr &) { - beginTx(); + if (!intx) { + beginTx(); + } } void UpdateDatabase::afterEvent(const MariaDB_Event_Ptr & e) { - output::pq::sql::updateSourcePosition::execute(this, e->next_event_pos, source); - commitTx(); + if (!intx) { + output::pq::sql::updateSourcePosition::execute(this, e->next_event_pos, source); + commitTx(); + } } void @@ -401,4 +407,22 @@ namespace MyGrate::Output::Pq { afterEvent(e); } } + + void + UpdateDatabase::gtid(MariaDB_Event_Ptr e) + { + beforeEvent(e); + if (!(e->event.gtid.flags & STANDALONE)) { + intx = true; + } + afterEvent(e); + } + + void + UpdateDatabase::xid(MariaDB_Event_Ptr e) + { + beforeEvent(e); + intx = false; + afterEvent(e); + } } diff --git a/lib/output/pq/updateDatabase.h b/lib/output/pq/updateDatabase.h index d2b4872..716d7f6 100644 --- a/lib/output/pq/updateDatabase.h +++ b/lib/output/pq/updateDatabase.h @@ -59,6 +59,8 @@ namespace MyGrate::Output::Pq { void insertRow(MariaDB_Event_Ptr) override; void tableMap(MariaDB_Event_Ptr) override; void rotate(MariaDB_Event_Ptr) override; + void gtid(MariaDB_Event_Ptr) override; + void xid(MariaDB_Event_Ptr) override; const uint64_t source; const std::string schema; @@ -81,6 +83,7 @@ namespace MyGrate::Output::Pq { Tables tables; Tables::const_iterator selected; MariaDB_Event_Ptr table_map; + bool intx {false}; }; } diff --git a/test/sql/simpleCreate.sql b/test/sql/simpleCreate.sql new file mode 100644 index 0000000..c2a89e9 --- /dev/null +++ b/test/sql/simpleCreate.sql @@ -0,0 +1,5 @@ +CREATE TABLE test( + id int auto_increment, + val varchar(20), + primary key(id) +) diff --git a/test/sql/simpleDelete.sql b/test/sql/simpleDelete.sql new file mode 100644 index 0000000..1c1bd85 --- /dev/null +++ b/test/sql/simpleDelete.sql @@ -0,0 +1,2 @@ +DELETE FROM test WHERE id = ? + diff --git a/test/sql/simpleDeleteSome.sql b/test/sql/simpleDeleteSome.sql new file mode 100644 index 0000000..ed69fa8 --- /dev/null +++ b/test/sql/simpleDeleteSome.sql @@ -0,0 +1,2 @@ +DELETE FROM test WHERE id < ? + diff --git a/test/sql/simpleInsert.sql b/test/sql/simpleInsert.sql new file mode 100644 index 0000000..f695332 --- /dev/null +++ b/test/sql/simpleInsert.sql @@ -0,0 +1 @@ +INSERT INTO test(val) VALUES(?) diff --git a/test/sql/simpleSelectAll.sql b/test/sql/simpleSelectAll.sql new file mode 100644 index 0000000..9fe07bc --- /dev/null +++ b/test/sql/simpleSelectAll.sql @@ -0,0 +1,3 @@ +SELECT id, val +FROM testout.test +ORDER BY id diff --git a/test/sql/simpleUpdate.sql b/test/sql/simpleUpdate.sql new file mode 100644 index 0000000..700f127 --- /dev/null +++ b/test/sql/simpleUpdate.sql @@ -0,0 +1 @@ +UPDATE test SET val = ? WHERE id = ? diff --git a/test/sql/simpleUpdateAll.sql b/test/sql/simpleUpdateAll.sql new file mode 100644 index 0000000..1ac2f13 --- /dev/null +++ b/test/sql/simpleUpdateAll.sql @@ -0,0 +1 @@ +UPDATE test SET val = ? diff --git a/test/test-e2e.cpp b/test/test-e2e.cpp index 37fa924..31f3b95 100644 --- a/test/test-e2e.cpp +++ b/test/test-e2e.cpp @@ -5,11 +5,19 @@ #include "semaphore.h" #include "testdb-mysql.h" #include "testdb-postgresql.h" +#include +#include #include #include #include #include #include +#include +#include +#include +#include +#include +#include #include class TestUpdateDatabase : public MyGrate::Output::Pq::UpdateDatabase { @@ -76,3 +84,46 @@ BOOST_AUTO_TEST_CASE(e2e, *boost::unit_test::timeout(15)) src->stopEvents(); repl.join(); } + +BOOST_AUTO_TEST_CASE(txns, *boost::unit_test::timeout(15)) +{ + const char * const target_schema {"testout"}; + using namespace MyGrate::Testing; + MySQLDB my; + PqConnDB pq {ROOT "/db/schema.sql"}; + + auto pqm = pq.mock(); + auto mym = my.mock(); + + MyGrate::sql::simpleCreate::execute(&mym); + + TestUpdateDatabase out {pqm.connstr.c_str(), + MyGrate::Output::Pq::UpdateDatabase::createNew(&pqm, MySQLDB::SERVER, MySQLDB::USER, MySQLDB::PASSWORD, + MySQLDB::PORT, my.mockname.c_str(), 100, target_schema) + .source}; + BOOST_CHECK_EQUAL(out.source, 1); + auto src = out.getSource(); + BOOST_REQUIRE(src); + + out.addTable(&mym, "test"); + + std::thread repl {&MyGrate::EventSourceBase::readEvents, src.get(), std::ref(out)}; + MyGrate::Tx {&mym}([&]() { + for (unsigned int x = 0; x < 10; x += 1) { + MyGrate::sql::simpleInsert::execute(&mym, MyGrate::scprintf<"some string %?">(x)); + } + MyGrate::sql::simpleUpdateAll::execute(&mym, "Same"); + MyGrate::sql::simpleDeleteSome::execute(&mym, 5); + }); + out.waitFor(14); + + src->stopEvents(); + repl.join(); + + auto recs = MyGrate::sql::simpleSelectAll::execute(&pqm); + BOOST_REQUIRE_EQUAL(recs->rows(), 6); + BOOST_CHECK_EQUAL(recs->at(0, 0).get(), 5); + BOOST_CHECK_EQUAL(recs->at(5, 0).get(), 10); + BOOST_CHECK_EQUAL(recs->at(1, 1).get(), "Same"); + BOOST_CHECK_EQUAL(recs->at(3, 1).get(), "Same"); +} -- cgit v1.2.3