Support table source being a SqlWriter expressionlibdbpp-1.0.3

5 files changed, 147 insertions, 22 deletions

diff --git a/libdbpp/sqlWriter.cpp b/libdbpp/sqlWriter.cpp
index d940ebf..8d588dd 100644
--- a/libdbpp/sqlWriter.cpp
+++ b/libdbpp/sqlWriter.cpp
@@ -5,3 +5,14 @@ DB::SqlWriter::bindParams(DB::Command *, unsigned int &)
 {
 }
 
+DB::StaticSqlWriter::StaticSqlWriter(const std::string & s) :
+	sql(s)
+{
+}
+
+void
+DB::StaticSqlWriter::writeSql(AdHoc::Buffer & buf)
+{
+	buf.append(sql);
+}
+
diff --git a/libdbpp/sqlWriter.h b/libdbpp/sqlWriter.h
index ca268c0..aacf3b3 100644
--- a/libdbpp/sqlWriter.h
+++ b/libdbpp/sqlWriter.h
@@ -10,6 +10,7 @@ namespace DB {
 	/// Base class of dynamic SQL constructors.
 	class DLL_PUBLIC SqlWriter {
 		public:
+			virtual ~SqlWriter() = default;
 			/// Append your SQL to the buffer.
 			/// @param buffer The buffer
 			virtual void writeSql(AdHoc::Buffer & buffer) = 0;
@@ -18,6 +19,19 @@ namespace DB {
 			/// @param offset The current bind offset.
 			virtual void bindParams(Command * cmd, unsigned int & offset);
 	};
+
+	/// A SQL Writer implementation that just writes static SQL.
+	class DLL_PUBLIC StaticSqlWriter : public SqlWriter {
+		public:
+			/// Construct with the SQL to write.
+			/// @param sql The SQL to write.
+			StaticSqlWriter(const std::string & sql);
+			/// Append the SQL to the buffer.
+			/// @param buffer The buffer
+			void writeSql(AdHoc::Buffer & buffer);
+
+			std::string sql;
+	};
 }
 
 #endif
diff --git a/libdbpp/tablepatch.cpp b/libdbpp/tablepatch.cpp
index 1f22fcf..40a1b41 100644
--- a/libdbpp/tablepatch.cpp
+++ b/libdbpp/tablepatch.cpp
@@ -8,6 +8,7 @@
 #include <boost/algorithm/string/join.hpp>
 
 DB::TablePatch::TablePatch() :
+	srcExpr(nullptr),
 	insteadOfDelete(nullptr),
 	where(nullptr),
 	order(nullptr),
@@ -27,11 +28,21 @@ DB::Connection::patchTable(TablePatch * tp)
 		throw TransactionRequired();
 	}
 	TransactionScope tx(this);
-	return {
+	bool ownedExpr = false;
+	if (!tp->srcExpr && !tp->src.empty()) {
+		tp->srcExpr = new DB::StaticSqlWriter(tp->src);
+		ownedExpr = true;
+	}
+	DB::PatchResult r {
 		tp->doDeletes ? patchDeletes(tp) : 0,
 		tp->doUpdates ? patchUpdates(tp) : 0,
 		tp->doInserts ? patchInserts(tp) : 0
 	};
+	if (ownedExpr) {
+		delete tp->srcExpr;
+		tp->srcExpr = nullptr;
+	}
+	return r;
 }
 
 template<typename Container>
@@ -107,12 +118,14 @@ DB::Connection::patchDeletes(TablePatch * tp)
 		AdHoc::Buffer toDelSql;
 		toDelSql.append("SELECT ");
 		append(toDelSql, tp->cols, ", ", "a.%s", selfCols);
-		toDelSql.appendbf(" FROM %s a LEFT OUTER JOIN %s b ON ",
-				tp->dest, tp->src);
+		toDelSql.appendbf(" FROM %s a LEFT OUTER JOIN ", tp->dest);
+		tp->srcExpr->writeSql(toDelSql);
+		toDelSql.append(" b ON ");
 		append(toDelSql, tp->pk, " AND ", " a.%s = b.%s", selfCols, selfCols);
 		patchDeletesSelect(toDelSql, tp);
 		auto del = select(toDelSql);
 		unsigned int offset = 0;
+		tp->srcExpr->bindParams(del.get(), offset);
 		if (tp->where) {
 			tp->where->bindParams(del.get(), offset);
 		}
@@ -144,8 +157,10 @@ DB::Connection::patchDeletes(TablePatch * tp)
 				// -----------------------------------------------------------------
 				toDelSql.append(") IN (SELECT ");
 				append(toDelSql, tp->pk, ", ", "a.%s", selfCols);
-				toDelSql.appendbf(" FROM %s a LEFT OUTER JOIN %s b ON ",
-						tp->dest, tp->src);
+				toDelSql.appendbf(" FROM %s a LEFT OUTER JOIN ",
+						tp->dest);
+				tp->srcExpr->writeSql(toDelSql);
+				toDelSql.append(" b ON ");
 				append(toDelSql, tp->pk, " AND ", " a.%s = b.%s", selfCols, selfCols);
 				patchDeletesSelect(toDelSql, tp);
 				toDelSql.append(")");
@@ -163,9 +178,9 @@ DB::Connection::patchDeletes(TablePatch * tp)
 							(bulkDeleteStyle() == BulkDeleteUsingUsingAlias ? "a" : tp->dest),
 							tp->dest);
 				}
-				toDelSql.appendbf(" LEFT OUTER JOIN %s b ",
-						tp->src);
-				toDelSql.append(" ON ");
+				toDelSql.append(" LEFT OUTER JOIN ");
+				tp->srcExpr->writeSql(toDelSql);
+				toDelSql.append(" b ON ");
 				append(toDelSql, tp->pk, " AND ", " a.%s = b.%s ", selfCols, selfCols);
 				if (tp->insteadOfDelete) {
 					tp->insteadOfDelete->writeSql(toDelSql);
@@ -176,6 +191,7 @@ DB::Connection::patchDeletes(TablePatch * tp)
 	}
 	auto del = ModifyCommandPtr(newModifyCommand(toDelSql));
 	unsigned int offset = 0;
+	tp->srcExpr->bindParams(del.get(), offset);
 	if (tp->insteadOfDelete) {
 		tp->insteadOfDelete->bindParams(del.get(), offset);
 	}
@@ -219,7 +235,9 @@ DB::Connection::patchUpdates(TablePatch * tp)
 		append(updSql, tp->pk, ", ", "a.%s", selfCols);
 		appendIf(updSql, tp->cols, isNotKey(tp), "", ", a.%1% old_%1%", selfCols);
 		appendIf(updSql, tp->cols, isNotKey(tp), "", ", b.%1% new_%1%", selfCols);
-		updSql.appendbf(" FROM %s a, %s b ", tp->dest, tp->src);
+		updSql.appendbf(" FROM %s a, ", tp->dest);
+		tp->srcExpr->writeSql(updSql);
+		updSql.append(" b ");
 		patchUpdatesSelect(updSql, tp);
 		if (tp->order) {
 			updSql.append(" ORDER BY ");
@@ -227,6 +245,7 @@ DB::Connection::patchUpdates(TablePatch * tp)
 		}
 		auto upd = select(updSql);
 		unsigned int offset = 0;
+		tp->srcExpr->bindParams(upd.get(), offset);
 		if (tp->where) {
 			tp->where->bindParams(upd.get(), offset);
 		}
@@ -245,14 +264,16 @@ DB::Connection::patchUpdates(TablePatch * tp)
 				updSql.appendbf("UPDATE %s a SET ",
 						tp->dest);
 				appendIf(updSql, tp->cols, isNotKey(tp), ", ", " %s = b.%s ", selfCols, selfCols);
-				updSql.appendbf(" FROM %s b ",
-						tp->src);
+				updSql.append(" FROM ");
+				tp->srcExpr->writeSql(updSql);
+				updSql.append(" b ");
 				patchUpdatesSelect(updSql, tp);
 				// -----------------------------------------------------------------
 				// Execute the bulk update command ---------------------------------
 				// -----------------------------------------------------------------
 				auto upd = ModifyCommandPtr(newModifyCommand(updSql));
 				unsigned int offset = 0;
+				tp->srcExpr->bindParams(upd.get(), offset);
 				if (tp->where) {
 					tp->where->bindParams(upd.get(), offset);
 				}
@@ -265,8 +286,10 @@ DB::Connection::patchUpdates(TablePatch * tp)
 				// Build SQL for list of updates to perform ------------------------
 				// -----------------------------------------------------------------
 				AdHoc::Buffer updSql;
-				updSql.appendbf("UPDATE %s a, %s b SET ",
-						tp->dest, tp->src);
+				updSql.appendbf("UPDATE %s a, ",
+						tp->dest);
+				tp->srcExpr->writeSql(updSql);
+				updSql.append(" b SET ");
 				appendIf(updSql, tp->cols, isNotKey(tp), ", ", " a.%s = b.%s ", selfCols, selfCols);
 				patchUpdatesSelect(updSql, tp);
 				if (tp->order) {
@@ -278,6 +301,7 @@ DB::Connection::patchUpdates(TablePatch * tp)
 				// -----------------------------------------------------------------
 				auto upd = ModifyCommandPtr(newModifyCommand(updSql));
 				unsigned int offset = 0;
+				tp->where->bindParams(upd.get(), offset);
 				if (tp->where) {
 					tp->where->bindParams(upd.get(), offset);
 				}
@@ -297,8 +321,10 @@ patchInsertsSelect(AdHoc::Buffer & toInsSql, DB::TablePatch * tp)
 {
 	toInsSql.append("SELECT ");
 	append(toInsSql, tp->cols, ", ", "b.%s", selfCols);
-	toInsSql.appendbf(" FROM %s b LEFT OUTER JOIN %s a ON ",
-			tp->src, tp->dest);
+	toInsSql.append(" FROM  ");
+	tp->srcExpr->writeSql(toInsSql);
+	toInsSql.appendbf(" b LEFT OUTER JOIN %s a ON ",
+			tp->dest);
 	append(toInsSql, tp->pk, " AND ", " a.%s = b.%s", selfCols, selfCols);
 	toInsSql.append(" WHERE ");
 	append(toInsSql, tp->pk, " AND ", " a.%s IS NULL", selfCols);
@@ -315,8 +341,9 @@ DB::Connection::patchInserts(TablePatch * tp)
 		AdHoc::Buffer toInsSql;
 		patchInsertsSelect(toInsSql, tp);
 		auto ins = select(toInsSql);
+		unsigned int offset = 0;
+		tp->srcExpr->bindParams(ins.get(), offset);
 		if (tp->order) {
-			unsigned int offset = 0;
 			tp->order->bindParams(ins.get(), offset);
 		}
 		tp->beforeInsert(ins);
@@ -331,8 +358,9 @@ DB::Connection::patchInserts(TablePatch * tp)
 	toInsSql.append(")\n");
 	patchInsertsSelect(toInsSql, tp);
 	auto ins = ModifyCommandPtr(newModifyCommand(toInsSql));
+	unsigned int offset = 0;
+	tp->srcExpr->bindParams(ins.get(), offset);
 	if (tp->order) {
-		unsigned int offset = 0;
 		tp->order->bindParams(ins.get(), offset);
 	}
 	return ins->execute();
diff --git a/libdbpp/tablepatch.h b/libdbpp/tablepatch.h
index c951cb5..84fada1 100644
--- a/libdbpp/tablepatch.h
+++ b/libdbpp/tablepatch.h
@@ -28,6 +28,8 @@ namespace DB {
 
 			/// Source table name.
 			TableName src;
+			/// Source expression.
+			SqlWriter * srcExpr;
 			/// Destination table name.
 			TableName dest;
 			/// Columns comprising the [effective] primary key.
diff --git a/libdbpp/unittests/testPatch.cpp b/libdbpp/unittests/testPatch.cpp
index cca8592..39ad5e3 100644
--- a/libdbpp/unittests/testPatch.cpp
+++ b/libdbpp/unittests/testPatch.cpp
@@ -18,12 +18,9 @@ class Mock : public PQ::Mock {
 		}
 };
 
-class OrderByA : public DB::SqlWriter {
+class OrderByA : public DB::StaticSqlWriter {
 	public:
-		void writeSql(AdHoc::Buffer & b) override
-		{
-			b.append("a");
-		}
+		OrderByA() : DB::StaticSqlWriter("a") { }
 };
 
 class WhereAequals1 : public DB::SqlWriter {
@@ -194,3 +191,76 @@ BOOST_AUTO_TEST_CASE( testInstead )
 	BOOST_REQUIRE_EQUAL(1, r.updates);
 }
 
+BOOST_AUTO_TEST_CASE( testSrcExprTable )
+{
+	Mock mock;
+	DB::StaticSqlWriter s("source");
+	auto db = DB::ConnectionPtr(DB::MockDatabase::openConnectionTo("pqmock"));
+	BOOST_REQUIRE(db);
+	DB::TablePatch tp;
+	tp.srcExpr = &s;
+	tp.dest = "target";
+	tp.cols = {"a", "b", "c", "d"};
+	tp.pk = {"a", "b"};
+	db->beginTx();
+	auto r = db->patchTable(&tp);
+	db->commitTx();
+	BOOST_REQUIRE_EQUAL(2, r.deletes);
+	BOOST_REQUIRE_EQUAL(2, r.inserts);
+	BOOST_REQUIRE_EQUAL(1, r.updates);
+}
+
+BOOST_AUTO_TEST_CASE( testSrcExprSelectTable )
+{
+	Mock mock;
+	DB::StaticSqlWriter s("(SELECT * FROM source)");
+	auto db = DB::ConnectionPtr(DB::MockDatabase::openConnectionTo("pqmock"));
+	BOOST_REQUIRE(db);
+	DB::TablePatch tp;
+	tp.srcExpr = &s;
+	tp.dest = "target";
+	tp.cols = {"a", "b", "c", "d"};
+	tp.pk = {"a", "b"};
+	db->beginTx();
+	auto r = db->patchTable(&tp);
+	db->commitTx();
+	BOOST_REQUIRE_EQUAL(2, r.deletes);
+	BOOST_REQUIRE_EQUAL(2, r.inserts);
+	BOOST_REQUIRE_EQUAL(1, r.updates);
+}
+
+class BindInt : public DB::StaticSqlWriter {
+	public:
+		BindInt(const std::string & s, int i) :
+			DB::StaticSqlWriter(s),
+			myInt(i)
+		{
+		}
+
+		void bindParams(DB::Command * c, unsigned int & offset) override
+		{
+			c->bindParamI(offset++, myInt);
+		}
+
+		int myInt;
+};
+
+BOOST_AUTO_TEST_CASE( testSrcExprSelectFilteredTable )
+{
+	Mock mock;
+	BindInt s("(SELECT s.* FROM source s WHERE s.a = ?)", 1);
+	auto db = DB::ConnectionPtr(DB::MockDatabase::openConnectionTo("pqmock"));
+	BOOST_REQUIRE(db);
+	DB::TablePatch tp;
+	tp.srcExpr = &s;
+	tp.dest = "target";
+	tp.cols = {"a", "b", "c", "d"};
+	tp.pk = {"a", "b"};
+	db->beginTx();
+	auto r = db->patchTable(&tp);
+	db->commitTx();
+	BOOST_REQUIRE_EQUAL(2, r.deletes);
+	BOOST_REQUIRE_EQUAL(1, r.inserts);
+	BOOST_REQUIRE_EQUAL(1, r.updates);
+}
+