finish expression filter integration

Author: evertlammerts

src/duckdb_py/arrow/arrow_array_stream.cpp

@@ -73,18 +73,20 @@ unique_ptr<ArrowArrayStreamWrapper> PythonTableArrowArrayStreamFactory::Produce(
 		auto filters = parameters.filters;
 		bool filters_pushed = false;
 
-		// Translate DuckDB filters to Polars expressions and push into the lazy plan
+		// Translate DuckDB filters to Polars expressions and push into the lazy plan.
+		// The walker only fails (throws / returns py::none()) for filters that are not
+		// required for correctness — optional/runtime wrappers it skips, or shapes the
+		// optimizer keeps above the scan. A throw here would mean the optimizer fully
+		// pushed something we can't translate (a correctness bug), so we let it surface
+		// rather than silently returning unfiltered rows — the arrow scan does not
+		// re-apply pushed filters. Mirrors the pyarrow ProduceScanner path.
 		if (filters && filters->HasFilters()) {
-			try {
-				auto filter_expr = PolarsFilterPushdown::TransformFilter(
-				    *filters, parameters.projected_columns.projection_map, parameters.projected_columns.filter_to_col,
-				    factory->client_properties);
-				if (!filter_expr.is(py::none())) {
-					lf = lf.attr("filter")(filter_expr);
-					filters_pushed = true;
-				}
-			} catch (...) {
-				// Fallback: DuckDB handles filtering post-scan
+			auto filter_expr = PolarsFilterPushdown::TransformFilter(
+			    *filters, parameters.projected_columns.projection_map, parameters.projected_columns.filter_to_col,
+			    factory->client_properties);
+			if (!filter_expr.is(py::none())) {
+				lf = lf.attr("filter")(filter_expr);
+				filters_pushed = true;
 			}
 		}
 

src/duckdb_py/arrow/filter_pushdown_visitor.cpp

@@ -6,13 +6,8 @@
 #include "duckdb/planner/expression/bound_constant_expression.hpp"
 #include "duckdb/planner/expression/bound_function_expression.hpp"
 #include "duckdb/planner/expression/bound_operator_expression.hpp"
-#include "duckdb/planner/expression/bound_reference_expression.hpp"
-#include "duckdb/planner/filter/conjunction_filter.hpp"
-#include "duckdb/planner/filter/constant_filter.hpp"
 #include "duckdb/planner/filter/expression_filter.hpp"
-#include "duckdb/planner/filter/in_filter.hpp"
-#include "duckdb/planner/filter/optional_filter.hpp"
-#include "duckdb/planner/filter/struct_filter.hpp"
+#include "duckdb/planner/filter/table_filter_functions.hpp"
 
 namespace duckdb {
 
@@ -101,6 +96,45 @@ py::object TransformExpression(const Expression &expression, const vector<string
 			return EmitCompare(backend, expression_type, std::move(col), constant_side->value, resolved.leaf_type,
 			                   timezone_config);
 		}
+
+		// Internal table-filter functions. Since the table-filter -> expression-filter
+		// migration in core, optional / dynamic / bloom / perfect-hash-join / prefix-range
+		// filters no longer have dedicated TableFilter subtypes. They arrive as scalar
+		// function wrappers inside the ExpressionFilter expression tree (see
+		// table_filter_functions.hpp).
+		const auto &func_name = bound_function_expression.function.GetName();
+
+		// OPTIONAL / SELECTIVITY_OPTIONAL wrap a child predicate that lives in `bind_info`
+		// (their `children` hold only a placeholder column ref). An optional filter is never
+		// required for correctness, so if its child can't be translated we push nothing for
+		// it rather than failing the whole scan.
+		if (func_name == OptionalFilterScalarFun::NAME || func_name == SelectivityOptionalFilterScalarFun::NAME) {
+			optional_ptr<const Expression> child;
+			if (bound_function_expression.bind_info) {
+				if (func_name == OptionalFilterScalarFun::NAME) {
+					child =
+					    bound_function_expression.bind_info->Cast<OptionalFilterFunctionData>().child_filter_expr.get();
+				} else {
+					child = bound_function_expression.bind_info->Cast<SelectivityOptionalFilterFunctionData>()
+					            .child_filter_expr.get();
+				}
+			}
+			if (!child) {
+				return py::none();
+			}
+			try {
+				return TransformExpression(*child, column_path, backend, arrow_type, timezone_config);
+			} catch (const NotImplementedException &) {
+				return py::none();
+			}
+		}
+
+		// DYNAMIC / BLOOM / PERFECT_HASH_JOIN / PREFIX_RANGE are runtime filters with no
+		// static pyarrow/polars equivalent. They are not required for correctness (the
+		// engine applies them above the scan), so skip them.
+		if (TableFilterFunctions::IsTableFilterFunction(func_name)) {
+			return py::none();
+		}
 	}
 
 	if (expression_class == ExpressionClass::BOUND_OPERATOR) {
@@ -130,19 +164,27 @@ py::object TransformExpression(const Expression &expression, const vector<string
 
 	if (expression_class == ExpressionClass::BOUND_CONJUNCTION) {
 		if (expression_type == ExpressionType::CONJUNCTION_OR || expression_type == ExpressionType::CONJUNCTION_AND) {
+			const bool is_and = expression_type == ExpressionType::CONJUNCTION_AND;
 			auto &conj_expr = expression.Cast<BoundConjunctionExpression>();
 			py::object result = py::none();
 			for (idx_t i = 0; i < conj_expr.children.size(); i++) {
 				py::object child_expression =
 				    TransformExpression(*conj_expr.children[i], column_path, backend, arrow_type, timezone_config);
 				if (child_expression.is(py::none())) {
-					// An OR branch that can't be translated (e.g. DYNAMIC_FILTER) means the pushed-down
-					// predicate would be stricter than the engine intends — fall back to no pushdown.
+					if (is_and) {
+						// A conjunct we can't push can simply be dropped: the remaining AND
+						// terms still form a correct (if weaker) filter, and the engine
+						// re-applies the rest above the scan.
+						continue;
+					}
+					// An OR branch that can't be translated (e.g. a dynamic filter) would
+					// make the pushed-down predicate stricter than the engine intends —
+					// fall back to no pushdown for the whole disjunction.
 					return py::none();
 				}
 				if (result.is(py::none())) {
 					result = std::move(child_expression);
-				} else if (expression_type == ExpressionType::CONJUNCTION_AND) {
+				} else if (is_and) {
 					result = backend.And(std::move(result), std::move(child_expression));
 				} else {
 					result = backend.Or(std::move(result), std::move(child_expression));

src/duckdb_py/include/duckdb_python/arrow/filter_pushdown_visitor.hpp

@@ -16,21 +16,22 @@
 
 namespace duckdb {
 
-// A FilterBackend abstracts the Python side of a `TableFilter` → expression
-// translation. The shared walker in this file handles the structural
-// recursion (CONJUNCTION_AND/OR, STRUCT_EXTRACT, OPTIONAL_FILTER, the
-// ExpressionFilter sub-walker) and dispatches leaf operations to the backend.
+// A FilterBackend abstracts the Python side of an `ExpressionFilter` →
+// expression translation. The shared walker in this file handles the
+// structural recursion (CONJUNCTION_AND/OR, struct_extract column paths, the
+// optional / selectivity-optional filter wrappers, and the internal runtime
+// filter functions) and dispatches leaf operations to the backend.
 //
 // Two backends exist today: PyArrowBackend (emits pyarrow.dataset.Expression)
 // and PolarsBackend (emits polars.Expr). Adding a new backend is purely a
 // matter of implementing this interface; the walker itself is reused.
 //
 // Convention: a backend method that cannot push the given filter must throw
-// `NotImplementedException`. The walker catches it at `OPTIONAL_FILTER`
-// boundaries (silently swallow) and at the top-level entry points (returning
-// `py::none()` for the affected column). Throwing keeps the "I can't push
-// this" path uniform across backends, replacing the old polars walker's ad
-// hoc `return py::none()` style.
+// `NotImplementedException`. The walker swallows it at optional-filter
+// boundaries (an optional filter is not required for correctness) and the
+// top-level entry points catch it too, returning `py::none()` for the affected
+// column. Throwing keeps the "I can't push this" path uniform across backends,
+// replacing the old polars walker's ad hoc `return py::none()` style.
 struct FilterBackend {
 	virtual ~FilterBackend() = default;
 
@@ -68,19 +69,25 @@ struct FilterBackend {
 	virtual py::object Or(py::object a, py::object b) = 0;
 };
 
-// Walk a TableFilter and emit a backend-specific expression.
-// - `column_path` is the path accumulated through STRUCT_EXTRACT.
+// Walk a TableFilter and emit a backend-specific expression. Since the
+// table-filter -> expression-filter migration in core, the only runtime filter
+// type is `EXPRESSION_FILTER`; this unwraps it and walks the expression tree.
+// - `column_path` is the top-level column name; struct paths are accumulated
+//    inside the expression walk via struct_extract.
 // - `arrow_type` is the ArrowType for the current path leaf (nullable for
 //    backends that don't track Arrow types).
 // - Returns `py::none()` if no part of the filter could be pushed.
 py::object TransformFilter(const TableFilter &filter, vector<string> column_path, FilterBackend &backend,
                            const ArrowType *arrow_type, const string &timezone_config);
 
-// Walk a bound Expression tree (the contents of an `ExpressionFilter`) and
-// emit a backend-specific expression. Handles BOUND_FUNCTION (comparisons),
+// Walk a bound Expression tree (the contents of an `ExpressionFilter`) and emit
+// a backend-specific expression. Handles BOUND_FUNCTION comparisons,
 // BOUND_OPERATOR (IS_NULL / IS_NOT_NULL / COMPARE_IN), BOUND_CONJUNCTION
-// (AND/OR), and recurses through struct_extract chains to build the column
-// path before invoking the backend.
+// (AND/OR), struct_extract column chains, the optional / selectivity-optional
+// wrappers (unwrapped from `bind_info`; an untranslatable child is swallowed),
+// and the internal runtime filter functions (dynamic / bloom / perfect-hash-join
+// / prefix-range, which are skipped). Returns `py::none()` for an optional or
+// runtime filter that can't be pushed.
 py::object TransformExpression(const Expression &expression, const vector<string> &column_path, FilterBackend &backend,
                                const ArrowType *arrow_type, const string &timezone_config);
 

tests/fast/arrow/test_filter_pushdown.py

@@ -190,7 +190,12 @@ def test_or_with_like_does_not_push(self, duckdb_cursor):
 
 
 class TestInPushdown:
-    """IN (...) reaches the walker as ``BOUND_OPERATOR`` with ``COMPARE_IN``."""
+    """IN (...) pushdown test.
+
+    IN (...) reaches the walker as a ``COMPARE_IN`` ``BOUND_OPERATOR``, wrapped in an
+    ``__internal_tablefilter_optional`` function that the walker must unwrap before it
+    sees the operator.
+    """
 
     def test_basic(self, duckdb_cursor):
         duckdb_cursor.execute("CREATE TABLE _t AS SELECT range a FROM range(1000)")