New operator pool branch

benchmark/micro/join/many_left_joins.benchmark

@@ -111,5 +111,3 @@ left join t5 ttt7 on (ttt6.c6=ttt7.c7)
 inner join t3 on (t2.c2=t3.c1)
 left join t4 tt4 on (t47.c1=tt4.c3)
 left join t4 tt5 on (tt4.c4=tt5.c5);
-
-result I

src/include/duckdb/optimizer/filter_pushdown.hpp

@@ -11,6 +11,7 @@
 #include "duckdb/common/unordered_set.hpp"
 #include "duckdb/optimizer/filter_combiner.hpp"
 #include "duckdb/optimizer/rule.hpp"
+#include "duckdb/optimizer/operator_pool.hpp"
 
 namespace duckdb {
 
@@ -70,6 +71,9 @@ class FilterPushdown {
 	// Finish pushing down at this operator, creating a LogicalFilter to store any of the stored filters and recursively
 	// pushing down into its children (if any)
 	unique_ptr<LogicalOperator> FinishPushdown(unique_ptr<LogicalOperator> op);
+	// This function is used when child operations are optimized with an optimizer that has
+	// should only
+	unique_ptr<LogicalOperator> FinishPushdownNoChildOptimization(unique_ptr<LogicalOperator> op);
 	//! Adds a filter to the set of filters. Returns FilterResult::UNSATISFIABLE if the subtree should be stripped, or
 	//! FilterResult::SUCCESS otherwise
 	FilterResult AddFilter(unique_ptr<Expression> expr);

src/include/duckdb/optimizer/operator_pool.hpp

@@ -0,0 +1,29 @@
+//===----------------------------------------------------------------------===//
+//                         DuckDB
+//
+// duckdb/optimizer/operator_pool.hpp
+//
+//
+//===----------------------------------------------------------------------===//
+
+#pragma once
+
+#include "duckdb/planner/logical_operator.hpp"
+#include "duckdb/common/unordered_set.hpp"
+
+namespace duckdb {
+
+class OperatorPool {
+public:
+	OperatorPool() {
+		seen_operators = unordered_set<idx_t>();
+	}
+
+	void AssertNotInPool(LogicalOperator *op);
+	bool InPool(LogicalOperator *op);
+	void EmptyOperatorPool();
+
+private:
+	unordered_set<idx_t> seen_operators;
+};
+} // namespace duckdb

src/include/duckdb/optimizer/optimizer.hpp

@@ -12,6 +12,7 @@
 #include "duckdb/planner/logical_operator.hpp"
 #include "duckdb/planner/logical_operator_visitor.hpp"
 #include "duckdb/common/enums/optimizer_type.hpp"
+#include "duckdb/optimizer/operator_pool.hpp"
 
 #include <functional>
 
@@ -23,6 +24,7 @@ class Optimizer {
 	Optimizer(Binder &binder, ClientContext &context);
 
 	unique_ptr<LogicalOperator> Optimize(unique_ptr<LogicalOperator> plan);
+	void AssertNotOptimized(LogicalOperator *op);
 
 	ClientContext &context;
 	Binder &binder;
@@ -33,6 +35,7 @@ class Optimizer {
 	void Verify(LogicalOperator &op);
 
 private:
+	OperatorPool seen_operators;
 	unique_ptr<LogicalOperator> plan;
 };
 

src/optimizer/CMakeLists.txt

@@ -18,6 +18,7 @@ add_library_unity(
   filter_pullup.cpp
   in_clause_rewriter.cpp
   optimizer.cpp
+  operator_pool.cpp
   expression_rewriter.cpp
   regex_range_filter.cpp
   remove_unused_columns.cpp

src/optimizer/filter_pushdown.cpp

@@ -13,6 +13,8 @@ FilterPushdown::FilterPushdown(Optimizer &optimizer) : optimizer(optimizer), com
 }
 
 unique_ptr<LogicalOperator> FilterPushdown::Rewrite(unique_ptr<LogicalOperator> op) {
+	// throws error in debug if operator has already been optimized with the current optimizer.
+	optimizer.AssertNotOptimized(op.get());
 	D_ASSERT(!combiner.HasFilters());
 	switch (op->type) {
 	case LogicalOperatorType::LOGICAL_AGGREGATE_AND_GROUP_BY:
@@ -124,6 +126,20 @@ unique_ptr<LogicalOperator> FilterPushdown::FinishPushdown(unique_ptr<LogicalOpe
 	return move(filter);
 }
 
+unique_ptr<LogicalOperator> FilterPushdown::FinishPushdownNoChildOptimization(unique_ptr<LogicalOperator> op) {
+	// now push any existing filters
+	if (filters.empty()) {
+		// no filters to push
+		return op;
+	}
+	auto filter = make_unique<LogicalFilter>();
+	for (auto &f : filters) {
+		filter->expressions.push_back(move(f->filter));
+	}
+	filter->children.push_back(move(op));
+	return move(filter);
+}
+
 void FilterPushdown::Filter::ExtractBindings() {
 	bindings.clear();
 	LogicalJoin::GetExpressionBindings(*filter, bindings);

src/optimizer/operator_pool.cpp

@@ -0,0 +1,21 @@
+#include "duckdb/optimizer/operator_pool.hpp"
+
+namespace duckdb {
+
+void OperatorPool::AssertNotInPool(LogicalOperator *op) {
+	D_ASSERT(!InPool(op));
+#ifdef DEBUG
+	seen_operators.insert((idx_t)op);
+#endif
+}
+
+bool OperatorPool::InPool(LogicalOperator *op) {
+	auto it = seen_operators.find((idx_t)op);
+	return it != seen_operators.end();
+}
+
+void OperatorPool::EmptyOperatorPool() {
+	seen_operators.clear();
+}
+
+} // namespace duckdb

src/optimizer/optimizer.cpp

@@ -26,7 +26,8 @@
 
 namespace duckdb {
 
-Optimizer::Optimizer(Binder &binder, ClientContext &context) : context(context), binder(binder), rewriter(context) {
+Optimizer::Optimizer(Binder &binder, ClientContext &context)
+    : context(context), binder(binder), rewriter(context), seen_operators() {
 	rewriter.rules.push_back(make_unique<ConstantFoldingRule>(rewriter));
 	rewriter.rules.push_back(make_unique<DistributivityRule>(rewriter));
 	rewriter.rules.push_back(make_unique<ArithmeticSimplificationRule>(rewriter));
@@ -60,6 +61,8 @@ void Optimizer::RunOptimizer(OptimizerType type, const std::function<void()> &ca
 	profiler.StartPhase(OptimizerTypeToString(type));
 	callback();
 	profiler.EndPhase();
+	seen_operators.EmptyOperatorPool();
+
 	if (plan) {
 		Verify(*plan);
 	}
@@ -69,6 +72,10 @@ void Optimizer::Verify(LogicalOperator &op) {
 	ColumnBindingResolver::Verify(op);
 }
 
+void Optimizer::AssertNotOptimized(LogicalOperator *op) {
+	seen_operators.AssertNotInPool(op);
+}
+
 unique_ptr<LogicalOperator> Optimizer::Optimize(unique_ptr<LogicalOperator> plan_p) {
 	Verify(*plan_p);
 	this->plan = move(plan_p);

src/optimizer/pushdown/pushdown_aggregate.cpp

@@ -65,7 +65,7 @@ unique_ptr<LogicalOperator> FilterPushdown::PushdownAggregate(unique_ptr<Logical
 	child_pushdown.GenerateFilters();
 
 	op->children[0] = child_pushdown.Rewrite(move(op->children[0]));
-	return FinishPushdown(move(op));
+	return FinishPushdownNoChildOptimization(move(op));
 }
 
 } // namespace duckdb

src/optimizer/pushdown/pushdown_left_join.cpp

@@ -120,6 +120,7 @@ unique_ptr<LogicalOperator> FilterPushdown::PushdownLeftJoin(unique_ptr<LogicalO
 		}
 	});
 	right_pushdown.GenerateFilters();
+
 	op->children[0] = left_pushdown.Rewrite(move(op->children[0]));
 	op->children[1] = right_pushdown.Rewrite(move(op->children[1]));
 	if (filters.empty()) {

src/optimizer/pushdown/pushdown_mark_join.cpp

@@ -77,7 +77,7 @@ unique_ptr<LogicalOperator> FilterPushdown::PushdownMarkJoin(unique_ptr<LogicalO
 	}
 	op->children[0] = left_pushdown.Rewrite(move(op->children[0]));
 	op->children[1] = right_pushdown.Rewrite(move(op->children[1]));
-	return FinishPushdown(move(op));
+	return FinishPushdownNoChildOptimization(move(op));
 }
 
 } // namespace duckdb

src/optimizer/pushdown/pushdown_single_join.cpp

@@ -23,7 +23,7 @@ unique_ptr<LogicalOperator> FilterPushdown::PushdownSingleJoin(unique_ptr<Logica
 	}
 	op->children[0] = left_pushdown.Rewrite(move(op->children[0]));
 	op->children[1] = right_pushdown.Rewrite(move(op->children[1]));
-	return FinishPushdown(move(op));
+	return FinishPushdownNoChildOptimization(move(op));
 }
 
 } // namespace duckdb

src/optimizer/regex_range_filter.cpp

@@ -16,7 +16,6 @@
 namespace duckdb {
 
 unique_ptr<LogicalOperator> RegexRangeFilter::Rewrite(unique_ptr<LogicalOperator> op) {
-
 	for (idx_t child_idx = 0; child_idx < op->children.size(); child_idx++) {
 		op->children[child_idx] = Rewrite(move(op->children[child_idx]));
 	}