summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--lib/output/dumpToConsole.cpp24
-rw-r--r--lib/output/dumpToConsole.h1
-rw-r--r--lib/output/pq/updateDatabase.cpp45
-rw-r--r--lib/output/pq/updateDatabase.h3
-rw-r--r--lib/rawDataReader.h6
-rw-r--r--lib/row.cpp29
-rw-r--r--lib/row.h17
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<TableId, MyGrate::MariaDB_Event_Ptr>;
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<std::vector<DbValue>> && out)
- {
- std::copy(r.begin(), r.end(), out);
- }
-
- void
- UpdateDatabase::copyKeys(
- const Row & r, const TableDefPtr & td, std::back_insert_iterator<std::vector<DbValue>> && out)
+ UpdateDatabase::copyKeys(const Row & r, const TableDefPtr & td, std::back_insert_iterator<Row> && 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<DbValue> 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<ReplicationError>(out->update->rows() == 1, "Wrong number of rows updated.");
+ auto rows {Row::fromRowsEvent(e->event.rows, table_map->event.table_map)};
+ verify<ReplicationError>(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<ReplicationError>(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<DbValue> 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<ReplicationError>(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<ReplicationError>(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<DbValue> 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<ReplicationError>(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<ReplicationError>(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<std::vector<DbValue>> &&);
- static void copyKeys(const Row & r, const TableDefPtr &, std::back_insert_iterator<std::vector<DbValue>> &&);
+ static void copyKeys(const Row & r, const TableDefPtr &, std::back_insert_iterator<Row> &&);
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 <string>
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> {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<DbValue> {
public:
- Row(const st_mariadb_rpl_rows_event &, const st_mariadb_rpl_table_map_event &);
+ using Rows = std::vector<Row>;
+ 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<DbValue>::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<Row, Row> {
- 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
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-22 18:18:20 +0000