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 +-- 4 files changed, 25 insertions(+), 48 deletions(-) (limited to 'lib/output') 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 &); -- cgit v1.2.3