568 operation converter optimizer with semi anti pushdown

src/execution/physical_plan/plan_set_operation.cpp

@@ -10,19 +10,6 @@
 
 namespace duckdb {
 
-static vector<unique_ptr<Expression>> CreatePartitionedRowNumExpression(const vector<LogicalType> &types) {
-	vector<unique_ptr<Expression>> res;
-	auto expr =
-	    make_uniq<BoundWindowExpression>(ExpressionType::WINDOW_ROW_NUMBER, LogicalType::BIGINT, nullptr, nullptr);
-	expr->start = WindowBoundary::UNBOUNDED_PRECEDING;
-	expr->end = WindowBoundary::UNBOUNDED_FOLLOWING;
-	for (idx_t i = 0; i < types.size(); i++) {
-		expr->partitions.push_back(make_uniq<BoundReferenceExpression>(types[i], i));
-	}
-	res.push_back(std::move(expr));
-	return res;
-}
-
 static JoinCondition CreateNotDistinctComparison(const LogicalType &type, idx_t i) {
 	JoinCondition cond;
 	cond.left = make_uniq<BoundReferenceExpression>(type, i);
@@ -43,6 +30,8 @@ unique_ptr<PhysicalOperator> PhysicalPlanGenerator::CreatePlan(LogicalSetOperati
 		throw InvalidInputException("Type mismatch for SET OPERATION");
 	}
 
+	// can't swich logical unions to semi/anti join
+	// also if the operation is a INTERSECT ALL or EXCEPT ALL
 	switch (op.type) {
 	case LogicalOperatorType::LOGICAL_UNION:
 		// UNION
@@ -51,54 +40,8 @@ unique_ptr<PhysicalOperator> PhysicalPlanGenerator::CreatePlan(LogicalSetOperati
 		break;
 	case LogicalOperatorType::LOGICAL_EXCEPT:
 	case LogicalOperatorType::LOGICAL_INTERSECT: {
-		auto &types = left->GetTypes();
-		vector<JoinCondition> conditions;
-		// create equality condition for all columns
-		for (idx_t i = 0; i < types.size(); i++) {
-			conditions.push_back(CreateNotDistinctComparison(types[i], i));
-		}
-		// For EXCEPT ALL / INTERSECT ALL we push a window operator with a ROW_NUMBER into the scans and join to get bag
-		// semantics.
-		if (op.setop_all) {
-			vector<LogicalType> window_types = types;
-			window_types.push_back(LogicalType::BIGINT);
-
-			auto window_left = make_uniq<PhysicalWindow>(window_types, CreatePartitionedRowNumExpression(types),
-			                                             left->estimated_cardinality);
-			window_left->children.push_back(std::move(left));
-			left = std::move(window_left);
-
-			auto window_right = make_uniq<PhysicalWindow>(window_types, CreatePartitionedRowNumExpression(types),
-			                                              right->estimated_cardinality);
-			window_right->children.push_back(std::move(right));
-			right = std::move(window_right);
-
-			// add window expression result to join condition
-			conditions.push_back(CreateNotDistinctComparison(LogicalType::BIGINT, types.size()));
-			// join (created below) now includes the row number result column
-			op.types.push_back(LogicalType::BIGINT);
-		}
-
-		// EXCEPT is ANTI join
-		// INTERSECT is SEMI join
-		PerfectHashJoinStats join_stats; // used in inner joins only
-
-		JoinType join_type = op.type == LogicalOperatorType::LOGICAL_EXCEPT ? JoinType::ANTI : JoinType::SEMI;
-		result = make_uniq<PhysicalHashJoin>(op, std::move(left), std::move(right), std::move(conditions), join_type,
-		                                     op.estimated_cardinality, join_stats);
-
-		// For EXCEPT ALL / INTERSECT ALL we need to remove the row number column again
-		if (op.setop_all) {
-			vector<unique_ptr<Expression>> projection_select_list;
-			for (idx_t i = 0; i < types.size(); i++) {
-				projection_select_list.push_back(make_uniq<BoundReferenceExpression>(types[i], i));
-			}
-			auto projection =
-			    make_uniq<PhysicalProjection>(types, std::move(projection_select_list), op.estimated_cardinality);
-			projection->children.push_back(std::move(result));
-			result = std::move(projection);
-		}
-		break;
+		throw InternalException(
+		    "Logical Except/Intersect should have been transformed to semi anti before the physical planning phase");
 	}
 	default:
 		throw InternalException("Unexpected operator type for set operation");

src/include/duckdb/planner/operator/logical_execute.hpp

@@ -34,7 +34,7 @@ class LogicalExecute : public LogicalOperator {
 
 protected:
 	void ResolveTypes() override {
-		// already resolved
+		types = prepared->types;
 	}
 	vector<ColumnBinding> GetColumnBindings() override {
 		return GenerateColumnBindings(0, types.size());

src/include/duckdb/planner/operator/logical_filter.hpp

@@ -9,9 +9,46 @@
 #pragma once
 
 #include "duckdb/planner/logical_operator.hpp"
+#include "duckdb/planner/operator/logical_comparison_join.hpp"
+#include "duckdb/planner/expression/bound_columnref_expression.hpp"
 
 namespace duckdb {
 
+static bool CanFiltersPropogateRightSide(LogicalOperator &op) {
+	if (op.type != LogicalOperatorType::LOGICAL_COMPARISON_JOIN) {
+		return false;
+	}
+	auto &join = op.Cast<LogicalComparisonJoin>();
+	if (join.join_type != JoinType::SEMI) {
+		return false;
+	}
+	auto left_bindings = op.children[0]->GetColumnBindings();
+	auto right_bindings = op.children[1]->GetColumnBindings();
+	D_ASSERT(left_bindings.size() == right_bindings.size());
+	// make sure we are comparing every column
+	if (join.conditions.size() != left_bindings.size()) {
+		return false;
+	}
+	auto &conditions = join.conditions;
+	for (idx_t i = 0; i < conditions.size(); i++) {
+		auto &cond = conditions[i];
+		auto &left = cond.left;
+		auto &right = cond.right;
+		if (cond.comparison == ExpressionType::COMPARE_NOT_DISTINCT_FROM) {
+			if (left->type == ExpressionType::BOUND_COLUMN_REF && right->type == ExpressionType::BOUND_COLUMN_REF) {
+				auto &left_expr = left->Cast<BoundColumnRefExpression>();
+				auto &right_expr = right->Cast<BoundColumnRefExpression>();
+				auto left_match = left_expr.binding == left_bindings[i];
+				auto right_match = right_expr.binding == right_bindings[i];
+				if (!(left_match && right_match)) {
+					return false;
+				}
+			}
+		}
+	}
+	return true;
+}
+
 //! LogicalFilter represents a filter operation (e.g. WHERE or HAVING clause)
 class LogicalFilter : public LogicalOperator {
 public:

src/main/config.cpp

@@ -15,7 +15,7 @@
 namespace duckdb {
 
 #ifdef DEBUG
-bool DBConfigOptions::debug_print_bindings = false;
+bool DBConfigOptions::debug_print_bindings = true;
 #endif
 
 #define DUCKDB_GLOBAL(_PARAM)                                                                                          \

src/optimizer/filter_pullup.cpp

@@ -1,5 +1,6 @@
 #include "duckdb/optimizer/filter_pullup.hpp"
 #include "duckdb/planner/operator/logical_join.hpp"
+#include "duckdb/planner/operator/logical_filter.hpp"
 
 namespace duckdb {
 
@@ -40,8 +41,13 @@ unique_ptr<LogicalOperator> FilterPullup::PullupJoin(unique_ptr<LogicalOperator>
 	case JoinType::INNER:
 		return PullupInnerJoin(std::move(op));
 	case JoinType::LEFT:
-	case JoinType::ANTI:
+	case JoinType::ANTI: {
+		return PullupFromLeft(std::move(op));
+	}
 	case JoinType::SEMI: {
+		if (CanFiltersPropogateRightSide(*op)) {
+			return PullupBothSide(std::move(op));
+		}
 		return PullupFromLeft(std::move(op));
 	}
 	default:

src/optimizer/join_order/relation_statistics_helper.cpp

@@ -244,7 +244,8 @@ RelationStats RelationStatisticsHelper::CombineStatsOfNonReorderableOperator(Log
 	}
 	ret.stats_initialized = true;
 	ret.filter_strength = 1;
-	ret.table_name = child_stats[0].table_name + " joined with " + child_stats[1].table_name;
+	ret.table_name =
+	    "(" + child_stats[0].table_name + LogicalOperatorToString(op.type) + child_stats[1].table_name + ")";
 	for (auto &stats : child_stats) {
 		// MARK joins are nonreorderable. They won't return initialized stats
 		// continue in this case.

src/optimizer/pushdown/pushdown_semi_anti_join.cpp

@@ -9,16 +9,55 @@ namespace duckdb {
 
 using Filter = FilterPushdown::Filter;
 
+static void ReplaceBindings(vector<ColumnBinding> &bindings, Filter &filter, Expression &expr,
+                            vector<ColumnBinding> &replacement_bindings) {
+	if (expr.type == ExpressionType::BOUND_COLUMN_REF) {
+		auto &colref = expr.Cast<BoundColumnRefExpression>();
+		D_ASSERT(colref.depth == 0);
+
+		// rewrite the binding by looking into the bound_tables list of the subquery
+		idx_t binding_index = 0;
+		for (idx_t i = 0; i < bindings.size(); i++) {
+			if (bindings[i] == colref.binding) {
+				binding_index = i;
+				break;
+			}
+		}
+		colref.binding = replacement_bindings[binding_index];
+		filter.bindings.insert(colref.binding.table_index);
+		return;
+	}
+	ExpressionIterator::EnumerateChildren(
+	    expr, [&](Expression &child) { ReplaceBindings(bindings, filter, child, replacement_bindings); });
+}
+
 unique_ptr<LogicalOperator> FilterPushdown::PushdownSemiAntiJoin(unique_ptr<LogicalOperator> op) {
 	auto &join = op->Cast<LogicalJoin>();
 	if (op->type == LogicalOperatorType::LOGICAL_DELIM_JOIN) {
 		return FinishPushdown(std::move(op));
 	}
 
-	// push all current filters down the left side
-	op->children[0] = Rewrite(std::move(op->children[0]));
-	FilterPushdown right_pushdown(optimizer);
-	op->children[1] = right_pushdown.Rewrite(std::move(op->children[1]));
+	if (CanFiltersPropogateRightSide(*op)) {
+		auto left_bindings = op->children[0]->GetColumnBindings();
+		auto right_bindings = op->children[1]->GetColumnBindings();
+		FilterPushdown right_pushdown(optimizer);
+		for (idx_t i = 0; i < filters.size(); i++) {
+			// first create a copy of the filter
+			auto right_filter = make_uniq<Filter>();
+			right_filter->filter = filters[i]->filter->Copy();
+
+			ReplaceBindings(left_bindings, *right_filter, *right_filter->filter, right_bindings);
+			right_filter->ExtractBindings();
+
+			// move the filters into the child pushdown nodes
+			right_pushdown.filters.push_back(std::move(right_filter));
+		}
+		op->children[0] = Rewrite(std::move(op->children[0]));
+		op->children[1] = right_pushdown.Rewrite(std::move(op->children[1]));
+	} else {
+		// push all current filters down the left side
+		op->children[0] = Rewrite(std::move(op->children[0]));
+	}
 
 	bool left_empty = op->children[0]->type == LogicalOperatorType::LOGICAL_EMPTY_RESULT;
 	bool right_empty = op->children[1]->type == LogicalOperatorType::LOGICAL_EMPTY_RESULT;

src/planner/binder/query_node/plan_setop.cpp

@@ -2,11 +2,32 @@
 #include "duckdb/planner/expression/bound_cast_expression.hpp"
 #include "duckdb/planner/expression/bound_columnref_expression.hpp"
 #include "duckdb/planner/operator/logical_projection.hpp"
+#include "duckdb/planner/operator/logical_window.hpp"
+#include "duckdb/planner/expression/bound_reference_expression.hpp"
+#include "duckdb/planner/expression/bound_window_expression.hpp"
 #include "duckdb/planner/operator/logical_set_operation.hpp"
 #include "duckdb/planner/query_node/bound_set_operation_node.hpp"
 
 namespace duckdb {
 
+static unique_ptr<LogicalWindow> CreateWindowWithPartitionedRowNum(idx_t window_table_index,
+                                                                   unique_ptr<LogicalOperator> op) {
+	// instead create a logical projection on top of whatever to add the window expression, then
+	auto window = make_uniq<LogicalWindow>(window_table_index);
+	auto row_number =
+	    make_uniq<BoundWindowExpression>(ExpressionType::WINDOW_ROW_NUMBER, LogicalType::BIGINT, nullptr, nullptr);
+	row_number->start = WindowBoundary::UNBOUNDED_PRECEDING;
+	row_number->end = WindowBoundary::CURRENT_ROW_ROWS;
+	auto bindings = op->GetColumnBindings();
+	auto types = op->types;
+	for (idx_t i = 0; i < types.size(); i++) {
+		row_number->partitions.push_back(make_uniq<BoundColumnRefExpression>(types[i], bindings[i]));
+	}
+	window->expressions.push_back(std::move(row_number));
+	window->AddChild(std::move(op));
+	return window;
+}
+
 // Optionally push a PROJECTION operator
 unique_ptr<LogicalOperator> Binder::CastLogicalOperatorToTypes(vector<LogicalType> &source_types,
                                                                vector<LogicalType> &target_types,
@@ -116,9 +137,104 @@ unique_ptr<LogicalOperator> Binder::CreatePlan(BoundSetOperationNode &node) {
 		break;
 	}
 
+	// here we convert the set operation to anti semi if required. Using the node.setop all we know what conversion we
+	// need.
 	auto root = make_uniq<LogicalSetOperation>(node.setop_index, node.types.size(), std::move(left_node),
 	                                           std::move(right_node), logical_type, node.setop_all);
+	root->ResolveOperatorTypes();
+
+	unique_ptr<LogicalOperator> op;
+
+	// if we have an intersect or except, immediately translate it to a semi or anti join.
+	// Unions stay as they are.
+	if (logical_type == LogicalOperatorType::LOGICAL_INTERSECT || logical_type == LogicalOperatorType::LOGICAL_EXCEPT) {
+		auto &left = root->children[0];
+		auto &right = root->children[1];
+		auto left_types = root->children[0]->types;
+		auto right_types = root->children[1]->types;
+		auto old_bindings = root->GetColumnBindings();
+		if (node.setop_all) {
+			auto window_left_table_id = GenerateTableIndex();
+			root->children[0] = CreateWindowWithPartitionedRowNum(window_left_table_id, std::move(root->children[0]));
+
+			auto window_right_table_id = GenerateTableIndex();
+			root->children[1] = CreateWindowWithPartitionedRowNum(window_right_table_id, std::move(root->children[1]));
 
+			root->types.push_back(LogicalType::BIGINT);
+			root->column_count += 1;
+		}
+
+		auto left_bindings = left->GetColumnBindings();
+		auto right_bindings = right->GetColumnBindings();
+		D_ASSERT(left_bindings.size() == right_bindings.size());
+
+		vector<JoinCondition> conditions;
+		// create equality condition for all columns
+		idx_t binding_offset = node.setop_all ? 1 : 0;
+		for (idx_t i = 0; i < left_bindings.size() - binding_offset; i++) {
+			auto cond_type_left = LogicalType(LogicalType::UNKNOWN);
+			auto cond_type_right = LogicalType(LogicalType::UNKNOWN);
+			JoinCondition cond;
+			cond.left = make_uniq<BoundColumnRefExpression>(left_types[i], left_bindings[i]);
+			cond.right = make_uniq<BoundColumnRefExpression>(right_types[i], right_bindings[i]);
+			cond.comparison = ExpressionType::COMPARE_NOT_DISTINCT_FROM;
+			conditions.push_back(std::move(cond));
+		}
+
+		// create condition for the row number as well.
+		if (node.setop_all) {
+			JoinCondition cond;
+			cond.left =
+			    make_uniq<BoundColumnRefExpression>(LogicalType::BIGINT, left_bindings[left_bindings.size() - 1]);
+			cond.right =
+			    make_uniq<BoundColumnRefExpression>(LogicalType::BIGINT, right_bindings[right_bindings.size() - 1]);
+			cond.comparison = ExpressionType::COMPARE_NOT_DISTINCT_FROM;
+			conditions.push_back(std::move(cond));
+		}
+
+		JoinType join_type = root->type == LogicalOperatorType::LOGICAL_EXCEPT ? JoinType::ANTI : JoinType::SEMI;
+
+		auto join_op = make_uniq<LogicalComparisonJoin>(join_type);
+		join_op->children.push_back(std::move(left));
+		join_op->children.push_back(std::move(right));
+		join_op->conditions = std::move(conditions);
+		join_op->ResolveOperatorTypes();
+
+		op = std::move(join_op);
+
+		// create projection to remove row_id.
+		if (node.setop_all) {
+			vector<unique_ptr<Expression>> projection_select_list;
+			auto bindings = op->GetColumnBindings();
+			for (idx_t i = 0; i < bindings.size() - 1; i++) {
+				projection_select_list.push_back(make_uniq<BoundColumnRefExpression>(op->types[i], bindings[i]));
+			}
+			auto projection = make_uniq<LogicalProjection>(node.setop_index, std::move(projection_select_list));
+			projection->children.push_back(std::move(op));
+			op = std::move(projection);
+		}
+
+		if (!node.setop_all) {
+			// push a distinct operator on the join
+			auto &types = op->types;
+			auto join_bindings = op->GetColumnBindings();
+			vector<unique_ptr<Expression>> distinct_targets;
+			vector<unique_ptr<Expression>> select_list;
+			for (idx_t i = 0; i < join_bindings.size(); i++) {
+				distinct_targets.push_back(make_uniq<BoundColumnRefExpression>(types[i], join_bindings[i]));
+				select_list.push_back(make_uniq<BoundColumnRefExpression>(types[i], join_bindings[i]));
+			}
+			auto distinct = make_uniq<LogicalDistinct>(std::move(distinct_targets), DistinctType::DISTINCT);
+			distinct->children.push_back(std::move(op));
+			op = std::move(distinct);
+
+			auto projection = make_uniq<LogicalProjection>(node.setop_index, std::move(select_list));
+			projection->children.push_back(std::move(op));
+			op = std::move(projection);
+			op->ResolveOperatorTypes();
+		}
+		return VisitQueryNode(node, std::move(op));
+	}
 	return VisitQueryNode(node, std::move(root));
 }