Handle Rows events with more than one row

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