Pushdown or operators

benchmark/imdb/init/load.sql

@@ -18,4 +18,5 @@ CREATE TABLE movie_link AS SELECT * FROM read_parquet('https://github.com/duckdb
 CREATE TABLE name AS SELECT * FROM read_parquet('https://github.com/duckdb/duckdb-data/releases/download/v1.0/job_name.parquet');
 CREATE TABLE person_info AS SELECT * FROM read_parquet('https://github.com/duckdb/duckdb-data/releases/download/v1.0/job_person_info.parquet');
 CREATE TABLE role_type AS SELECT * FROM read_parquet('https://github.com/duckdb/duckdb-data/releases/download/v1.0/job_role_type.parquet');
-CREATE TABLE title AS SELECT * FROM read_parquet('https://github.com/duckdb/duckdb-data/releases/download/v1.0/job_title.parquet');
+CREATE TABLE title AS SELECT * FROM read_parquet('https://github.com/duckdb/duckdb-data/releases/download/v1.0/job_title.parquet');
+

benchmark/imdb_plan_cost/queries/18a.sql

@@ -1,3 +1,4 @@
+pragma threads=1;
 SELECT MIN(mi.info) AS movie_budget,
        MIN(mi_idx.info) AS movie_votes,
        MIN(t.title) AS movie_title

src/include/duckdb/planner/table_filter.hpp

@@ -12,6 +12,7 @@
 #include "duckdb/common/types.hpp"
 #include "duckdb/common/unordered_map.hpp"
 #include "duckdb/common/enums/filter_propagate_result.hpp"
+#include "duckdb/common/enums/expression_type.hpp"
 
 namespace duckdb {
 class BaseStatistics;
@@ -63,12 +64,14 @@ class TableFilter {
 	}
 };
 
+//!
 class TableFilterSet {
 public:
 	unordered_map<idx_t, unique_ptr<TableFilter>> filters;
 
 public:
-	void PushFilter(idx_t table_index, unique_ptr<TableFilter> filter);
+	void PushFilter(idx_t column_index, unique_ptr<TableFilter> filter,
+	                TableFilterType conjunction_type = TableFilterType::CONJUNCTION_AND);
 
 	bool Equals(TableFilterSet &other) {
 		if (filters.size() != other.filters.size()) {
@@ -97,6 +100,7 @@ class TableFilterSet {
 
 	void Serialize(Serializer &serializer) const;
 	static TableFilterSet Deserialize(Deserializer &deserializer);
+	static bool ExpressionSupportsPushdown(ExpressionType comparison);
 };
 
 } // namespace duckdb

src/optimizer/filter_combiner.cpp

@@ -14,6 +14,7 @@
 #include "duckdb/planner/filter/constant_filter.hpp"
 #include "duckdb/planner/filter/null_filter.hpp"
 #include "duckdb/optimizer/optimizer.hpp"
+#include "duckdb/common/types/value.hpp"
 
 namespace duckdb {
 
@@ -391,20 +392,29 @@ bool FilterCombiner::HasFilters() {
 // 	return zonemap_checks;
 // }
 
+bool TableFilterSet::ExpressionSupportsPushdown(ExpressionType comparison) {
+	return (comparison == ExpressionType::COMPARE_EQUAL || comparison == ExpressionType::COMPARE_GREATERTHAN ||
+	        comparison == ExpressionType::COMPARE_GREATERTHANOREQUALTO ||
+	        comparison == ExpressionType::COMPARE_LESSTHAN || comparison == ExpressionType::COMPARE_LESSTHANOREQUALTO);
+}
+
+static bool ValueTypeSupportsPushown(const Value &value) {
+	return TypeIsNumeric(value.type().InternalType()) || value.type().InternalType() == PhysicalType::VARCHAR ||
+	       value.type().InternalType() == PhysicalType::BOOL;
+}
+
+static bool LeftConstValRightBoundColref(Expression &left, Expression &right) {
+	return (left.type == ExpressionType::VALUE_CONSTANT && right.type == ExpressionType::BOUND_COLUMN_REF);
+}
+
 TableFilterSet FilterCombiner::GenerateTableScanFilters(vector<idx_t> &column_ids) {
 	TableFilterSet table_filters;
 	//! First, we figure the filters that have constant expressions that we can push down to the table scan
 	for (auto &constant_value : constant_values) {
 		if (!constant_value.second.empty()) {
 			auto filter_exp = equivalence_map.end();
-			if ((constant_value.second[0].comparison_type == ExpressionType::COMPARE_EQUAL ||
-			     constant_value.second[0].comparison_type == ExpressionType::COMPARE_GREATERTHAN ||
-			     constant_value.second[0].comparison_type == ExpressionType::COMPARE_GREATERTHANOREQUALTO ||
-			     constant_value.second[0].comparison_type == ExpressionType::COMPARE_LESSTHAN ||
-			     constant_value.second[0].comparison_type == ExpressionType::COMPARE_LESSTHANOREQUALTO) &&
-			    (TypeIsNumeric(constant_value.second[0].constant.type().InternalType()) ||
-			     constant_value.second[0].constant.type().InternalType() == PhysicalType::VARCHAR ||
-			     constant_value.second[0].constant.type().InternalType() == PhysicalType::BOOL)) {
+			if (TableFilterSet::ExpressionSupportsPushdown(constant_value.second[0].comparison_type) &&
+			    ValueTypeSupportsPushown(constant_value.second[0].constant)) {
 				//! Here we check if these filters are column references
 				filter_exp = equivalence_map.find(constant_value.first);
 				if (filter_exp->second.size() == 1 &&
@@ -433,6 +443,7 @@ TableFilterSet FilterCombiner::GenerateTableScanFilters(vector<idx_t> &column_id
 		}
 	}
 	//! Here we look for LIKE or IN filters
+	vector<idx_t> remaining_filters_to_remove;
 	for (idx_t rem_fil_idx = 0; rem_fil_idx < remaining_filters.size(); rem_fil_idx++) {
 		auto &remaining_filter = remaining_filters[rem_fil_idx];
 		if (remaining_filter->expression_class == ExpressionClass::BOUND_FUNCTION) {
@@ -522,6 +533,14 @@ TableFilterSet FilterCombiner::GenerateTableScanFilters(vector<idx_t> &column_id
 			// e.g. if we have x IN (1, 2, 3, 4, 5) we transform this into x >= 1 AND x <= 5
 			if (!type.IsIntegral()) {
 				continue;
+//				for (idx_t i = 1; i < func.children.size(); i++) {
+//					auto &child = func.children[i];
+//					auto &const_value_expr = child->Cast<BoundConstantExpression>();
+//					auto filter = make_uniq<ConstantFilter>(ExpressionType::COMPARE_EQUAL, const_value_expr.value);
+//					table_filters.PushFilter(column_index, std::move(filter), TableFilterType::CONJUNCTION_OR);
+//				}
+//				remaining_filters_to_remove.push_back(rem_fil_idx);
+//				continue;
 			}
 
 			bool can_simplify_in_clause = true;
@@ -546,6 +565,13 @@ TableFilterSet FilterCombiner::GenerateTableScanFilters(vector<idx_t> &column_id
 				}
 			}
 			if (!can_simplify_in_clause) {
+				// make one big conjunction OR
+				for (auto &in_val : in_values) {
+					auto filter =
+					    make_uniq<ConstantFilter>(ExpressionType::COMPARE_EQUAL, Value::Numeric(type, in_val));
+					table_filters.PushFilter(column_index, std::move(filter), TableFilterType::CONJUNCTION_OR);
+				}
+				remaining_filters_to_remove.push_back(rem_fil_idx);
 				continue;
 			}
 			auto lower_bound = make_uniq<ConstantFilter>(ExpressionType::COMPARE_GREATERTHANOREQUALTO,
@@ -556,10 +582,74 @@ TableFilterSet FilterCombiner::GenerateTableScanFilters(vector<idx_t> &column_id
 			table_filters.PushFilter(column_index, std::move(upper_bound));
 			table_filters.PushFilter(column_index, make_uniq<IsNotNullFilter>());
 
-			remaining_filters.erase(remaining_filters.begin() + rem_fil_idx);
+			remaining_filters_to_remove.push_back(rem_fil_idx);
+		} else if (remaining_filter->type == ExpressionType::CONJUNCTION_OR) {
+//			continue;
+			unordered_set<idx_t> columns_that_need_and_null;
+			auto &conj = remaining_filter->Cast<BoundConjunctionExpression>();
+			auto or_filter = make_uniq<ConjunctionOrFilter>();
+			bool all_can_pushdown = true;
+			idx_t column_id = DConstants::INVALID_INDEX;
+			for (auto &expr : conj.children) {
+				// go through every child of the conjunction and check that it can be pushed down
+				// if not a direct comparison skip
+				if (expr->expression_class != ExpressionClass::BOUND_COMPARISON) {
+					all_can_pushdown = false;
+					continue;
+				}
+				auto &comp = expr->Cast<BoundComparisonExpression>();
+				BoundColumnRefExpression *colref = nullptr;
+				BoundConstantExpression *value = nullptr;
+				bool can_pushdown = false;
+				// if not a simple comparison between bound column index and constant value, skip
+				if (TableFilterSet::ExpressionSupportsPushdown(comp.type)) {
+					if (LeftConstValRightBoundColref(*comp.left, *comp.right)) {
+						value = &comp.left->Cast<BoundConstantExpression>();
+						colref = &comp.right->Cast<BoundColumnRefExpression>();
+						can_pushdown = true;
+					} else if (LeftConstValRightBoundColref(*comp.right, *comp.left)) {
+						value = &comp.right->Cast<BoundConstantExpression>();
+						colref = &comp.left->Cast<BoundColumnRefExpression>();
+						can_pushdown = true;
+					}
+				}
+				if (!can_pushdown) {
+					all_can_pushdown = false;
+					break;
+				}
+				auto column_index = column_ids[colref->binding.column_index];
+				if (column_index == COLUMN_IDENTIFIER_ROW_ID) {
+					all_can_pushdown = false;
+					break;
+				}
+				// if conjunction has two separate column_ids, skip
+				// (i.e a = 5 or b = 7)
+				if (column_id == DConstants::INVALID_INDEX) {
+					column_id = column_index;
+				}
+				if (column_id != column_index) {
+					all_can_pushdown = false;
+					break;
+				}
+				auto column_filter = make_uniq<ConstantFilter>(comp.type, value->value);
+				or_filter->child_filters.push_back(std::move(column_filter));
+				columns_that_need_and_null.insert(column_index);
+			}
+			// if all of the expressions in the conjunction or can be pushed down, then add add them to the
+			// table filters. Otherwise leave it as a filter on the read.
+			if (all_can_pushdown) {
+				table_filters.PushFilter(column_id, std::move(or_filter), TableFilterType::CONJUNCTION_OR);
+//				table_filters.PushFilter(column_id, make_uniq<IsNotNullFilter>(), TableFilterType::CONJUNCTION_AND);
+				remaining_filters_to_remove.push_back(rem_fil_idx);
+			}
 		}
 	}
 
+	for (int i = remaining_filters_to_remove.size() - 1; i >= 0; i--) {
+		auto remaining_filter_index = remaining_filters_to_remove.at(i);
+		remaining_filters.erase(remaining_filters.begin() + remaining_filter_index);
+	}
+
 	//	GenerateORFilters(table_filters, column_ids);
 
 	return table_filters;

src/optimizer/optimizer.cpp

@@ -176,7 +176,7 @@ unique_ptr<LogicalOperator> Optimizer::Optimize(unique_ptr<LogicalOperator> plan
 		column_lifetime.VisitOperator(*plan);
 	});
 
-	// compress data based on statistics for materializing operators
+//	// compress data based on statistics for materializing operators
 	RunOptimizer(OptimizerType::COMPRESSED_MATERIALIZATION, [&]() {
 		CompressedMaterialization compressed_materialization(context, binder, std::move(statistics_map));
 		compressed_materialization.Compress(plan);

src/planner/table_filter.cpp

@@ -5,17 +5,39 @@
 
 namespace duckdb {
 
-void TableFilterSet::PushFilter(idx_t column_index, unique_ptr<TableFilter> filter) {
+void TableFilterSet::PushFilter(idx_t column_index, unique_ptr<TableFilter> filter, TableFilterType conjunction_type) {
 	auto entry = filters.find(column_index);
 	if (entry == filters.end()) {
 		// no filter yet: push the filter directly
 		filters[column_index] = std::move(filter);
 	} else {
 		// there is already a filter: AND it together
 		if (entry->second->filter_type == TableFilterType::CONJUNCTION_AND) {
+			if (conjunction_type == TableFilterType::CONJUNCTION_OR) {
+				throw InternalException("Conjunction AND should not be connecting conjunction ors.");
+			}
 			auto &and_filter = entry->second->Cast<ConjunctionAndFilter>();
 			and_filter.child_filters.push_back(std::move(filter));
+		} else if (entry->second->filter_type == TableFilterType::CONJUNCTION_OR) {
+			// Or filter connecting ands
+			if (conjunction_type == TableFilterType::CONJUNCTION_AND) {
+				auto and_filter = make_uniq<ConjunctionAndFilter>();
+				and_filter->child_filters.push_back(std::move(entry->second));
+				and_filter->child_filters.push_back(std::move(filter));
+				filters[column_index] = std::move(and_filter);
+				return;
+			}
+			auto &or_filter = entry->second->Cast<ConjunctionOrFilter>();
+			or_filter.child_filters.push_back(std::move(filter));
 		} else {
+			if (conjunction_type == TableFilterType::CONJUNCTION_OR) {
+				auto or_filter = make_uniq<ConjunctionOrFilter>();
+				or_filter->child_filters.push_back(std::move(entry->second));
+				or_filter->child_filters.push_back(std::move(filter));
+				filters[column_index] = std::move(or_filter);
+				return;
+			}
+			D_ASSERT(conjunction_type == TableFilterType::CONJUNCTION_AND);
 			auto and_filter = make_uniq<ConjunctionAndFilter>();
 			and_filter->child_filters.push_back(std::move(entry->second));
 			and_filter->child_filters.push_back(std::move(filter));

src/storage/table/column_segment.cpp

@@ -340,13 +340,32 @@ idx_t ColumnSegment::FilterSelection(SelectionVector &sel, Vector &result, const
 		// similar to the CONJUNCTION_AND, but we need to take care of the SelectionVectors (OR all of them)
 		idx_t count_total = 0;
 		SelectionVector result_sel(approved_tuple_count);
+
+//		SelectionVector current_sel = result_sel;
+//		idx_t current_count = approved_tuple_count;
+//		idx_t result_count = 0;
+//
+//		unique_ptr<SelectionVector> temp_true, temp_false;
+//		unique_ptr<SelectionVector> false_sel;
+//
+//		temp_true = make_uniq<SelectionVector>(STANDARD_VECTOR_SIZE);
+//
+//		temp_false = make_uniq<SelectionVector>(STANDARD_VECTOR_SIZE);
+
 		auto &conjunction_or = filter.Cast<ConjunctionOrFilter>();
 		for (auto &child_filter : conjunction_or.child_filters) {
 			SelectionVector temp_sel;
 			temp_sel.Initialize(sel);
 			idx_t temp_tuple_count = approved_tuple_count;
 			idx_t temp_count = FilterSelection(temp_sel, result, *child_filter, temp_tuple_count, mask);
 			// tuples passed, move them into the actual result vector
+//			if (temp_count > 0) {
+//				for (idx_t i = 0; i < temp_count; i++) {
+//					result_sel.set_index(result_count++, temp_true->get_index(i));
+//				}
+//				current_count -= temp_count;
+//
+//			}
 			for (idx_t i = 0; i < temp_count; i++) {
 				auto new_idx = temp_sel.get_index(i);
 				bool is_new_idx = true;

src/storage/table/row_group.cpp

@@ -20,6 +20,7 @@
 #include "duckdb/common/serializer/serializer.hpp"
 #include "duckdb/common/serializer/deserializer.hpp"
 #include "duckdb/common/serializer/binary_serializer.hpp"
+#include "iostream"
 
 namespace duckdb {
 
@@ -267,7 +268,7 @@ unique_ptr<RowGroup> RowGroup::AlterType(RowGroupCollection &new_collection, con
 }
 
 unique_ptr<RowGroup> RowGroup::AddColumn(RowGroupCollection &new_collection, ColumnDefinition &new_column,
-                                         ExpressionExecutor &executor, Expression &default_value, Vector &result) {
+                                          ExpressionExecutor &executor, Expression &default_value, Vector &result) {
 	Verify();
 
 	// construct a new column data for the new column
@@ -394,11 +395,12 @@ void RowGroup::TemplatedScan(TransactionData transaction, CollectionScanState &s
 	const bool ALLOW_UPDATES = TYPE != TableScanType::TABLE_SCAN_COMMITTED_ROWS_DISALLOW_UPDATES &&
 	                           TYPE != TableScanType::TABLE_SCAN_COMMITTED_ROWS_OMIT_PERMANENTLY_DELETED;
 	auto table_filters = state.GetFilters();
+
 	const auto &column_ids = state.GetColumnIds();
 	auto adaptive_filter = state.GetAdaptiveFilter();
 	while (true) {
 		if (state.vector_index * STANDARD_VECTOR_SIZE >= state.max_row_group_row) {
-			// exceeded the amount of rows to scan
+			// exceeded the amount of rows to scan 
 			return;
 		}
 		idx_t current_row = state.vector_index * STANDARD_VECTOR_SIZE;
@@ -462,12 +464,46 @@ void RowGroup::TemplatedScan(TransactionData transaction, CollectionScanState &s
 			if (table_filters) {
 				D_ASSERT(adaptive_filter);
 				D_ASSERT(ALLOW_UPDATES);
+
+				// are we apply the whole OR conjunction filter at once. and then slice it.
+				// What we should do instead is, if the filter is a conjunction OR filter, apply every individual child
+				// update the selection vector with tuples that pass, then apply the next conjunction to the indexes that did not pass.
 				for (idx_t i = 0; i < table_filters->filters.size(); i++) {
+
 					auto tf_idx = adaptive_filter->permutation[i];
 					auto col_idx = column_ids[tf_idx];
 					auto &col_data = GetColumn(col_idx);
-					col_data.Select(transaction, state.vector_index, state.column_scans[tf_idx], result.data[tf_idx],
-					                sel, approved_tuple_count, *table_filters->filters[tf_idx]);
+					auto &filter = table_filters->filters[tf_idx];
+
+					switch (filter->filter_type) {
+					case TableFilterType::CONJUNCTION_OR: {
+
+						auto &or_filter = filter->Cast<ConjunctionOrFilter>();
+						SelectionVector current_sel = sel;
+						SelectionVector *true_sel = nullptr;
+						SelectionVector *false_sel = nullptr;
+
+						// scan and flatten the vector.
+						// apply children one by one unti
+						for (idx_t i = 0; i < or_filter.child_filters.size(); i++) {
+							idx_t tcount = 0;
+							auto &child_filter_expression = or_filter.child_filters.at(i);
+							if (child_filter_expression->filter_type == TableFilterType::CONSTANT_COMPARISON) {
+								auto context = transaction.transaction.get()->context;
+								auto expression_executor = ExpressionExecutor(&context);
+								ColumnSegment::FilterSelection(sel, result.data.at(tf_idx), *child_filter_expression,
+								                               tcount, FlatVector::Validity(result.data.at(tf_idx)));
+							}
+						}
+
+
+						}
+					}
+					case TableFilterType::CONJUNCTION_AND:
+					default: {
+						col_data.Select(transaction, state.vector_index, state.column_scans[tf_idx], result.data[tf_idx],
+						                sel, approved_tuple_count, *table_filters->filters[tf_idx]);
+					}
 				}
 				for (auto &table_filter : table_filters->filters) {
 					result.data[table_filter.first].Slice(sel, approved_tuple_count);