From 25e2428754a3ac6c02650812564c857d1e25df55 Mon Sep 17 00:00:00 2001 From: Dan Goodliffe Date: Mon, 2 Aug 2021 21:13:43 +0100 Subject: Handle Rows events with more than one row --- lib/output/dumpToConsole.cpp | 24 ++++++--------------- lib/output/dumpToConsole.h | 1 - lib/output/pq/updateDatabase.cpp | 45 ++++++++++++++++------------------------ lib/output/pq/updateDatabase.h | 3 +-- lib/rawDataReader.h | 6 ++++++ lib/row.cpp | 29 ++++++++------------------ lib/row.h | 17 +++------------ 7 files changed, 43 insertions(+), 82 deletions(-) diff --git a/lib/output/dumpToConsole.cpp b/lib/output/dumpToConsole.cpp index 656063b..304e35a 100644 --- a/lib/output/dumpToConsole.cpp +++ b/lib/output/dumpToConsole.cpp @@ -33,7 +33,7 @@ namespace MyGrate::Output { { const auto & rs = event->event.rows; scprintf<"Update %?\n">(std::cout, rs.table_id); - dumpRowPairData(event->event.rows); + dumpRowData(event->event.rows); } void @@ -68,22 +68,10 @@ namespace MyGrate::Output { void DumpToConsole::dumpRowData(const st_mariadb_rpl_rows_event & row) const { - Row r {row, tableMaps.at(row.table_id)->event.table_map}; - std::for_each(r.begin(), r.end(), [](auto && fv) { - fv.visit(write {}); - }); - } - - void - DumpToConsole::dumpRowPairData(const st_mariadb_rpl_rows_event & row) const - { - RowPair rp {row, tableMaps.at(row.table_id)->event.table_map}; - std::for_each(rp.first.begin(), rp.first.end(), [](auto && fv) { - fv.visit(write {}); - }); - std::for_each(rp.second.begin(), rp.second.end(), [](auto && fv) { - fv.visit(write {}); - }); + for (const auto & r : Row::fromRowsEvent(row, tableMaps.at(row.table_id)->event.table_map)) { + std::for_each(r.begin(), r.end(), [](auto && fv) { + fv.visit(write {}); + }); + } } - } diff --git a/lib/output/dumpToConsole.h b/lib/output/dumpToConsole.h index bbbac44..23fa6ac 100644 --- a/lib/output/dumpToConsole.h +++ b/lib/output/dumpToConsole.h @@ -20,7 +20,6 @@ namespace MyGrate::Output { using TableMaps = std::map; void dumpRowData(const st_mariadb_rpl_rows_event & row) const; - void dumpRowPairData(const st_mariadb_rpl_rows_event & row) const; TableMaps tableMaps; }; diff --git a/lib/output/pq/updateDatabase.cpp b/lib/output/pq/updateDatabase.cpp index eb912d8..232e8f9 100644 --- a/lib/output/pq/updateDatabase.cpp +++ b/lib/output/pq/updateDatabase.cpp @@ -284,14 +284,7 @@ namespace MyGrate::Output::Pq { } void - UpdateDatabase::copyAll(const Row & r, std::back_insert_iterator> && out) - { - std::copy(r.begin(), r.end(), out); - } - - void - UpdateDatabase::copyKeys( - const Row & r, const TableDefPtr & td, std::back_insert_iterator> && out) + UpdateDatabase::copyKeys(const Row & r, const TableDefPtr & td, std::back_insert_iterator && out) { std::copy_if(r.begin(), r.end(), out, [c = td->columns.begin()](auto &&) mutable { return (c++)->get()->is_pk; @@ -324,13 +317,13 @@ namespace MyGrate::Output::Pq { out->update = prepare(ou.str().c_str(), kordinal); } beforeEvent(e); - std::vector updateValues; - updateValues.reserve(out->columns.size() + out->keys); - RowPair rp {e->event.rows, table_map->event.table_map}; - copyAll(rp.second, std::back_inserter(updateValues)); - copyKeys(rp.first, out, std::back_inserter(updateValues)); - out->update->execute(updateValues); - verify(out->update->rows() == 1, "Wrong number of rows updated."); + auto rows {Row::fromRowsEvent(e->event.rows, table_map->event.table_map)}; + verify(rows.size() % 2 == 0, "Odd number of update rows"); + for (auto rp = rows.begin(); rp != rows.end(); rp++) { + copyKeys(*rp, out, std::back_inserter(*rp)); + out->update->execute(*rp); + verify(out->update->rows() == 1, "Wrong number of rows updated."); + } afterEvent(e); } } @@ -356,12 +349,12 @@ namespace MyGrate::Output::Pq { out->deleteFrom = prepare(ou.str().c_str(), kordinal); } beforeEvent(e); - std::vector updateValues; - updateValues.reserve(out->keys); - Row rp {e->event.rows, table_map->event.table_map}; - copyKeys(rp, out, std::back_inserter(updateValues)); - out->deleteFrom->execute(updateValues); - verify(out->deleteFrom->rows() == 1, "Wrong number of rows deleted."); + for (auto & r : Row::fromRowsEvent(e->event.rows, table_map->event.table_map)) { + Row keys; + copyKeys(r, out, std::back_inserter(keys)); + out->deleteFrom->execute(keys); + verify(out->deleteFrom->rows() == 1, "Wrong number of rows deleted."); + } afterEvent(e); } } @@ -390,12 +383,10 @@ namespace MyGrate::Output::Pq { out->insertInto = prepare(ou.str().c_str(), out->columns.size()); } beforeEvent(e); - std::vector updateValues; - updateValues.reserve(out->columns.size()); - Row rp {e->event.rows, table_map->event.table_map}; - copyAll(rp, std::back_inserter(updateValues)); - out->insertInto->execute(updateValues); - verify(out->insertInto->rows() == 1, "Wrong number of rows updated."); + for (const auto & r : Row::fromRowsEvent(e->event.rows, table_map->event.table_map)) { + out->insertInto->execute(r); + verify(out->insertInto->rows() == 1, "Wrong number of rows updated."); + } afterEvent(e); } } diff --git a/lib/output/pq/updateDatabase.h b/lib/output/pq/updateDatabase.h index bc3a6bc..d2b4872 100644 --- a/lib/output/pq/updateDatabase.h +++ b/lib/output/pq/updateDatabase.h @@ -73,8 +73,7 @@ namespace MyGrate::Output::Pq { UpdateDatabase(PqConn &&, uint64_t source, RecordSetPtr cfg); static void verifyRow(const MariaDB_Event_Ptr & e, const TableDefPtr &); - static void copyAll(const Row & r, std::back_insert_iterator> &&); - static void copyKeys(const Row & r, const TableDefPtr &, std::back_insert_iterator> &&); + static void copyKeys(const Row & r, const TableDefPtr &, std::back_insert_iterator &&); void copyTableContent(Input::MySQLConn *, const char * tableName, const TableDefPtr &); diff --git a/lib/rawDataReader.h b/lib/rawDataReader.h index 906eaef..58724ff 100644 --- a/lib/rawDataReader.h +++ b/lib/rawDataReader.h @@ -56,6 +56,12 @@ namespace MyGrate { void discard(size_t); + bool + more() const + { + return offset < len; + } + private: void offsetSizeCheck(size_t) const; diff --git a/lib/row.cpp b/lib/row.cpp index 9d81906..ba8045d 100644 --- a/lib/row.cpp +++ b/lib/row.cpp @@ -10,17 +10,6 @@ #include namespace MyGrate { - Row::Row(const st_mariadb_rpl_rows_event & row, const st_mariadb_rpl_table_map_event & tm) : - Row(row, tm, RawDataReader {tm.metadata}, RawDataReader {row.row_data, row.row_data_size}) - { - } - - Row::Row(const st_mariadb_rpl_rows_event & row, const st_mariadb_rpl_table_map_event & tm, - MyGrate::RawDataReader && md, MyGrate::RawDataReader && data) : - Row {row, tm, md, data} - { - } - Row::Row(const st_mariadb_rpl_rows_event & row, const st_mariadb_rpl_table_map_event & tm, MyGrate::RawDataReader & md, MyGrate::RawDataReader & data) { @@ -114,15 +103,15 @@ namespace MyGrate { } } - RowPair::RowPair(const st_mariadb_rpl_rows_event & row, const st_mariadb_rpl_table_map_event & tm) : - RowPair(row, tm, RawDataReader {tm.metadata}, RawDataReader {tm.metadata}, - RawDataReader {row.row_data, row.row_data_size}) - { - } - - RowPair::RowPair(const st_mariadb_rpl_rows_event & row, const st_mariadb_rpl_table_map_event & tm, - MyGrate::RawDataReader && md1, MyGrate::RawDataReader && md2, MyGrate::RawDataReader && data) : - std::pair {Row {row, tm, md1, data}, Row {row, tm, md2, data}} + Row::Rows + Row::fromRowsEvent(const st_mariadb_rpl_rows_event & row, const st_mariadb_rpl_table_map_event & tm) { + Rows rtn; + RawDataReader data {row.row_data, row.row_data_size}; + while (data.more()) { + RawDataReader md {tm.metadata}; + rtn.emplace_back(row, tm, md, data); + } + return rtn; } } diff --git a/lib/row.h b/lib/row.h index 35d8bcd..908a52f 100644 --- a/lib/row.h +++ b/lib/row.h @@ -13,24 +13,13 @@ struct st_mariadb_rpl_table_map_event; namespace MyGrate { class Row : public std::vector { public: - Row(const st_mariadb_rpl_rows_event &, const st_mariadb_rpl_table_map_event &); + using Rows = std::vector; + static Rows fromRowsEvent(const st_mariadb_rpl_rows_event &, const st_mariadb_rpl_table_map_event &); - private: - friend class RowPair; - Row(const st_mariadb_rpl_rows_event &, const st_mariadb_rpl_table_map_event &, MyGrate::RawDataReader && md, - MyGrate::RawDataReader && data); + using std::vector::vector; Row(const st_mariadb_rpl_rows_event &, const st_mariadb_rpl_table_map_event &, MyGrate::RawDataReader & md, MyGrate::RawDataReader & data); }; - - class RowPair : public std::pair { - public: - RowPair(const st_mariadb_rpl_rows_event &, const st_mariadb_rpl_table_map_event &); - - private: - RowPair(const st_mariadb_rpl_rows_event &, const st_mariadb_rpl_table_map_event &, - MyGrate::RawDataReader && md1, MyGrate::RawDataReader && md2, MyGrate::RawDataReader && data); - }; } #endif -- cgit v1.2.3