Handle Rows events with more than one row

4 files changed, 25 insertions, 48 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 &);