Hacktoberfest 2026: những issue maintainer đã đánh dấu cho tháng Mười, đang mở và phù hợp người mới. Xem issue Hacktoberfest

Filters on Arrow tables with run-end encoded columns get pushed down and fail (some invalidate the database)

Đang mở
#666 0 bình luận 0 reaction 0 người được giao Xem trên GitHub

Maintainer thường phản hồi trong vòng 1 ngày

Chưa có ai nhận issue này.

Đánh giá

Độ khó
3/5
Thời gian dự kiến
Nửa ngày
Mức phù hợp với người mới
45/100
Loại issue
Lỗi
Độ rõ ràng
Đặc tả rõ ràng
Mức độ hoạt động
Sôi nổi
Công nghệ
cpp, python
Lĩnh vực
data, databases

Hướng nghiên cứu

Start in src/function/table/arrow.cpp at ArrowTableFunction::ArrowPushdownType and the HasViewType helper it calls, then check ArrowType::SetRunEndEncoded to see why the duck type hides the run-end struct info. The proposed fix and the Catch test in test/arrow/arrow_filter_pushdown.cpp are already written, but the core change belongs in duckdb/duckdb, so it needs coordination. Done means the REE Catch cases show a FILTER above the scan and the pytest REE cases pass.

Do mô hình lập chỉ mục viết ra từ nội dung của issue.

Mô tả

What happens?

Scanning a pyarrow table with a run-end encoded column works, but filtering it mostly doesn't. The filters get pushed into the scan, and pyarrow's compare and filter kernels don't support run-end encoded arrays. What breaks depends on the column's value type:

  • For integer, float, bool, date and plain timestamp values, a filter on the REE column fails in pyarrow (Function 'equal' has no kernel matching input types (run_end_encoded<...>, int8), or less / is_in). A filter on any other column fails in array_filter, because pyarrow filters every column of the batch.
  • For string and binary values, any comparison or Top-N on the table, on any column, throws INTERNAL Error: Failed to cast ArrowTypeInfo, type mismatch (expected: STRING, got: STRUCT) while planning. EXPLAIN throws it too.
  • For decimals wider than 18 digits and timezone-aware timestamps (s/ms/us units; any unit on 1.5.6), comparing the column throws the same kind of internal error (expected: DECIMAL / expected: DATE_TIME), and the database is invalidated after that.

Joins against these tables fail as well, partly because of a separate bug in join filter pushdown: #665.

From reading the code: ArrowType::SetRunEndEncoded() gives an REE column its values' duck type but leaves its type info as the struct of run ends and values. ArrowPushdownType only looks at the duck type, so it allows the filter. For strings, HasViewType reads that struct info as ArrowStringInfo, which is where the STRING error comes from. For wide decimals and tz timestamps the filter gets as far as MakePyArrowScalar in this repo, which reads it as ArrowDecimalInfo / ArrowDateTimeInfo. Narrow decimals are only safe because CanPushdown already refuses them.

SET disabled_optimizers = 'filter_pushdown' fixes the WHERE cases, but not Top-N or joins.

Suggested fix

This is also a duckdb/duckdb change: treat REE columns like string_view and turn off pushdown for the whole table, the same approach as duckdb/duckdb#20703 for #227. The REE check has to come before the duck type is read. I renamed the two helpers since they're no longer only about view types.

diff --git a/src/function/table/arrow.cpp b/src/function/table/arrow.cpp
--- a/src/function/table/arrow.cpp
+++ b/src/function/table/arrow.cpp
@@ -266,7 +266,14 @@
 		return false;
 	}
 }
-static bool HasViewType(const ArrowType &type) {
+// Whether PyArrow's array_filter kernel cannot filter an array of this type: a string_view/binary_view array, a
+// run-end encoded array, or a nested array with such a child.
+static bool HasUnfilterableType(const ArrowType &type) {
+	if (type.RunEndEncoded()) {
+		// checked first: the duck type of a run-end encoded array is its values' type, but its type info is the
+		// struct of its run ends and values
+		return true;
+	}
 	auto duck_type = type.GetDuckType();
 	switch (duck_type.id()) {
 	case LogicalTypeId::VARCHAR:
@@ -277,7 +284,7 @@
 	case LogicalTypeId::UNION: {
 		const auto &struct_info = type.GetTypeInfo<ArrowStructInfo>();
 		for (idx_t i = 0; i < struct_info.ChildCount(); i++) {
-			if (HasViewType(struct_info.GetChild(i))) {
+			if (HasUnfilterableType(struct_info.GetChild(i))) {
 				return true;
 			}
 		}
@@ -286,20 +293,20 @@
 	case LogicalTypeId::LIST:
 	case LogicalTypeId::MAP: {
 		const auto &list_info = type.GetTypeInfo<ArrowListInfo>();
-		return HasViewType(list_info.GetChild());
+		return HasUnfilterableType(list_info.GetChild());
 	}
 	case LogicalTypeId::ARRAY: {
 		const auto &array_info = type.GetTypeInfo<ArrowArrayInfo>();
-		return HasViewType(array_info.GetChild());
+		return HasUnfilterableType(array_info.GetChild());
 	}
 	default:
 		return false;
 	}
 }
 
-static bool TableHasViewTypes(const arrow_column_map_t &column_info) {
+static bool TableHasUnfilterableTypes(const arrow_column_map_t &column_info) {
 	for (const auto &col : column_info) {
-		if (HasViewType(*col.second)) {
+		if (HasUnfilterableType(*col.second)) {
 			return true;
 		}
 	}
@@ -309,11 +316,11 @@
 bool ArrowTableFunction::ArrowPushdownType(const FunctionData &bind_data, idx_t col_idx) {
 	auto &arrow_bind_data = bind_data.Cast<ArrowScanFunctionData>();
 	const auto &column_info = arrow_bind_data.arrow_table.GetColumns();
-	// PyArrow's array_filter kernel doesn't support string_view/binary_view types.
-	// The filter is applied to ALL columns in a record batch, so if any column has a
-	// view type, we must disable filter pushdown for the entire table.
+	// PyArrow's array_filter kernel doesn't support string_view/binary_view or run-end encoded arrays.
+	// The filter is applied to ALL columns in a record batch, so if any column has such a
+	// type, we must disable filter pushdown for the entire table.
 	// See https://github.com/duckdb/duckdb-python/issues/227
-	if (TableHasViewTypes(column_info)) {
+	if (TableHasUnfilterableTypes(column_info)) {
 		return false;
 	}
 	const auto column_type = column_info.at(col_idx);

With this the reproducer returns the right rows and the connection stays usable. I also ran a bigger matrix of 172 cases covering 19 value types, int16/32/64 run ends, top-level and struct-field columns, several kinds of filter, and table/reader/dataset/scanner inputs. 96 of them fail on 1.5.6 and the dev build, 6 of those invalidating the database. With the patch all 172 give the same rows as the same data without REE.

The Catch test below builds REE arrays by hand and checks EXPLAIN for a FILTER above the scan. The pytest cases cover int64, utf8, string_view, decimal(30,2) and timestamptz, and 12 of the 15 fail without the patch. [arrow], test/optimizer/* and tests/fast/arrow still pass. It applies to main and to 939d6af, with or without the patch from #665.

Core test
diff --git a/test/arrow/arrow_filter_pushdown.cpp b/test/arrow/arrow_filter_pushdown.cpp
--- a/test/arrow/arrow_filter_pushdown.cpp
+++ b/test/arrow/arrow_filter_pushdown.cpp
@@ -1,6 +1,7 @@
 #include "catch.hpp"
 
 #include "arrow/arrow_test_helper.hpp"
+#include "duckdb/common/adbc/single_batch_array_stream.hpp"
 
 using namespace duckdb;
 
@@ -23,8 +24,8 @@
 }
 
 // Helper: get the EXPLAIN output for an arrow_scan with a filter
-static string GetExplainForFilter(Connection &con, ArrowTestFactory &factory, const string &filter_expr) {
-	const auto params = ArrowTestHelper::ConstructArrowScan(factory);
+static string GetExplainForFilter(Connection &con, const shared_ptr<ArrowScanFactory> &params,
+                                  const string &filter_expr) {
 	const auto rel = con.TableFunction("arrow_scan", {}, {}, params)->Filter(filter_expr);
 	// render operator names raw/upper-case so the string checks below (ARROW_SCAN, FILTER) match
 	REQUIRE(!con.Query("SET profiling_renderer_settings = MAP {'operator_casing': 'upper'}")->HasError());
@@ -32,6 +33,10 @@
 	REQUIRE(!explain_result->HasError());
 	auto &mat = *explain_result;
 	return mat.Collection().GetValue(1, 0).ToString();
+}
+
+static string GetExplainForFilter(Connection &con, ArrowTestFactory &factory, const string &filter_expr) {
+	return GetExplainForFilter(con, ArrowTestHelper::ConstructArrowScan(factory), filter_expr);
 }
 
 // Helper: check for a standalone FILTER operator node in the explain output
@@ -107,3 +112,165 @@
 		REQUIRE(!FilterInScan(explain_str));
 	}
 }
+
+namespace arrow_filter_pushdown_ree_test {
+
+// Stack-allocated Arrow structs need a no-op release.
+void ReleaseSchema(ArrowSchema *s) {
+	s->release = nullptr;
+}
+void ReleaseArray(ArrowArray *a) {
+	a->release = nullptr;
+}
+
+const int32_t X_VALUES[5] = {0, 1, 2, 3, 4};
+// d holds 7, 7, 9, 9, 9 or "red", "red", "blue", "blue", "blue": two runs, ending at rows 2 and 5
+const int32_t RUN_ENDS[2] = {2, 5};
+const int32_t INT32_VALUES[2] = {7, 9};
+const int32_t UTF8_OFFSETS[3] = {0, 3, 7};
+const char UTF8_DATA[] = "redblue";
+
+enum class RunValues { INT32, UTF8 };
+
+// One batch of two columns, x INTEGER 0..4 and a run-end encoded column d with int32 or utf8 values, as a
+// single-batch stream for binding the scan.
+struct RunEndTable {
+	explicit RunEndTable(RunValues values) {
+		x_schema.format = "i";
+		x_schema.name = "x";
+		x_schema.release = ReleaseSchema;
+		x_buffers[0] = nullptr;
+		x_buffers[1] = X_VALUES;
+		x_array.length = 5;
+		x_array.n_buffers = 2;
+		x_array.buffers = x_buffers;
+		x_array.release = ReleaseArray;
+
+		run_ends_schema.format = "i";
+		run_ends_schema.name = "run_ends";
+		run_ends_schema.release = ReleaseSchema;
+		run_ends_buffers[0] = nullptr;
+		run_ends_buffers[1] = RUN_ENDS;
+		run_ends_array.length = 2;
+		run_ends_array.n_buffers = 2;
+		run_ends_array.buffers = run_ends_buffers;
+		run_ends_array.release = ReleaseArray;
+
+		values_schema.name = "values";
+		values_schema.flags = ARROW_FLAG_NULLABLE;
+		values_schema.release = ReleaseSchema;
+		values_buffers[0] = nullptr;
+		switch (values) {
+		case RunValues::INT32:
+			values_schema.format = "i";
+			values_buffers[1] = INT32_VALUES;
+			values_array.n_buffers = 2;
+			break;
+		case RunValues::UTF8:
+			values_schema.format = "u";
+			values_buffers[1] = UTF8_OFFSETS;
+			values_buffers[2] = UTF8_DATA;
+			values_array.n_buffers = 3;
+			break;
+		}
+		values_array.length = 2;
+		values_array.buffers = values_buffers;
+		values_array.release = ReleaseArray;
+
+		d_schema_children[0] = &run_ends_schema;
+		d_schema_children[1] = &values_schema;
+		d_schema.format = "+r";
+		d_schema.name = "d";
+		d_schema.n_children = 2;
+		d_schema.children = d_schema_children;
+		d_schema.release = ReleaseSchema;
+		// a run-end encoded array has no buffers of its own
+		d_array_children[0] = &run_ends_array;
+		d_array_children[1] = &values_array;
+		d_array.length = 5;
+		d_array.n_children = 2;
+		d_array.children = d_array_children;
+		d_array.release = ReleaseArray;
+
+		schema_children[0] = &x_schema;
+		schema_children[1] = &d_schema;
+		record_schema.format = "+s";
+		record_schema.n_children = 2;
+		record_schema.children = schema_children;
+		record_schema.release = ReleaseSchema;
+		array_children[0] = &x_array;
+		array_children[1] = &d_array;
+		record_array.length = 5;
+		record_array.n_buffers = 1;
+		record_array.buffers = record_buffers;
+		record_array.n_children = 2;
+		record_array.children = array_children;
+		record_array.release = ReleaseArray;
+		AdbcError err {};
+		REQUIRE(duckdb_adbc::BatchToArrayStream(&record_array, &record_schema, &stream, &err) == ADBC_STATUS_OK);
+	}
+	RunEndTable(const RunEndTable &) = delete;
+	~RunEndTable() {
+		// an EXPLAIN only binds the scan, so the stream is still ours to release
+		if (stream.release) {
+			stream.release(&stream);
+		}
+	}
+
+	const void *x_buffers[2];
+	const void *run_ends_buffers[2];
+	const void *values_buffers[3];
+	const void *record_buffers[1] = {nullptr};
+	ArrowSchema *d_schema_children[2];
+	ArrowArray *d_array_children[2];
+	ArrowSchema *schema_children[2];
+	ArrowArray *array_children[2];
+	ArrowSchema x_schema {};
+	ArrowSchema run_ends_schema {};
+	ArrowSchema values_schema {};
+	ArrowSchema d_schema {};
+	ArrowSchema record_schema {};
+	ArrowArray x_array {};
+	ArrowArray run_ends_array {};
+	ArrowArray values_array {};
+	ArrowArray d_array {};
+	ArrowArray record_array {};
+	ArrowArrayStream stream {};
+};
+
+} // namespace arrow_filter_pushdown_ree_test
+
+TEST_CASE("Arrow filter pushdown - run-end encoded columns disable pushdown", "[arrow]") {
+	using namespace arrow_filter_pushdown_ree_test;
+	DuckDB db;
+	Connection con(db);
+
+	SECTION("Filter on a run-end encoded int32 column: filter above scan (not pushed)") {
+		RunEndTable table(RunValues::INT32);
+		auto explain_str = GetExplainForFilter(con, ArrowTestHelper::ConstructArrowScan(table.stream), "d = 7");
+		REQUIRE(StandaloneFilter(explain_str));
+		REQUIRE(!FilterInScan(explain_str));
+	}
+
+	SECTION("Filter on another column of a table with a run-end encoded int32 column: filter above scan") {
+		// pyarrow's array_filter, which filters every column of the batch, has no kernel for run-end encoded arrays
+		RunEndTable table(RunValues::INT32);
+		auto explain_str = GetExplainForFilter(con, ArrowTestHelper::ConstructArrowScan(table.stream), "x > 2");
+		REQUIRE(StandaloneFilter(explain_str));
+		REQUIRE(!FilterInScan(explain_str));
+	}
+
+	SECTION("Filter on a run-end encoded utf8 column: filter above scan (not pushed)") {
+		RunEndTable table(RunValues::UTF8);
+		auto explain_str = GetExplainForFilter(con, ArrowTestHelper::ConstructArrowScan(table.stream), "d = 'red'");
+		REQUIRE(StandaloneFilter(explain_str));
+		REQUIRE(!FilterInScan(explain_str));
+	}
+
+	SECTION("Filter on another column of a table with a run-end encoded utf8 column: filter above scan") {
+		RunEndTable table(RunValues::UTF8);
+		auto explain_str = GetExplainForFilter(con, ArrowTestHelper::ConstructArrowScan(table.stream), "x > 2");
+		REQUIRE(StandaloneFilter(explain_str));
+		REQUIRE(!FilterInScan(explain_str));
+	}
+}
Python test
diff --git a/tests/fast/arrow/test_filter_pushdown.py b/tests/fast/arrow/test_filter_pushdown.py
--- a/tests/fast/arrow/test_filter_pushdown.py
+++ b/tests/fast/arrow/test_filter_pushdown.py
@@ -14,8 +14,8 @@
   (one- and multi-level), optional filter, dynamic top-N filter, join
   filter pushdown.
 * **Unsupported-type fallback** — ``UHUGEINT``, ``string_view``,
-  ``binary_view`` columns must not crash; the filter is applied above the
-  scan instead.
+  ``binary_view`` and run-end encoded columns must not crash; the filter is
+  applied above the scan instead.
 * **Regressions** — issues that were fixed previously and need to stay
   fixed.
 * **Canaries** — markers for behaviour we expect to change upstream
@@ -31,6 +31,7 @@
 
 import datetime as dt
 import re
+from decimal import Decimal
 
 import pytest
 from _pushdown_helpers import (
@@ -648,6 +649,59 @@
         assert res == [("abc",)]
 
 
+class TestRunEndEncoded:
+    """Filters on a table with a run-end encoded column are applied above the scan.
+
+    pyarrow has no compare kernels for run-end encoded arrays, and its ``array_filter``, which filters every
+    column of a batch, has none either, so no filter on such a table can be pushed into the scan.
+    """
+
+    # (pyarrow value type, the value of d in rows 0 and 1, the value of d in rows 2 and 4, a SQL literal for the first)
+    CASES = (
+        pytest.param(pa.int64(), 7, 9, "7", id="int64"),
+        pytest.param(pa.string(), "red", "blue", "'red'", id="utf8"),
+        pytest.param(pa.string_view(), "red", "blue", "'red'", id="string_view"),
+        pytest.param(pa.decimal128(30, 2), Decimal("1.50"), Decimal("2.50"), "1.50::DECIMAL(30,2)", id="decimal"),
+        pytest.param(
+            pa.timestamp("us", tz="UTC"),
+            dt.datetime(2024, 1, 1, tzinfo=dt.timezone.utc),
+            dt.datetime(2024, 6, 1, tzinfo=dt.timezone.utc),
+            "'2024-01-01 00:00:00+00'::TIMESTAMPTZ",
+            id="timestamptz",
+        ),
+    )
+
+    @staticmethod
+    def _connect(value_type, first, second) -> duckdb.DuckDBPyConnection:
+        """A connection with ``t``: ``x`` 0..4 and a run-end encoded ``d`` of first, first, second, NULL, second."""
+        run_end_encoded = pa.RunEndEncodedArray.from_arrays(
+            pa.array([2, 3, 4, 5], pa.int32()), pa.array([first, second, None, second], type=value_type)
+        )
+        con = duckdb.connect()
+        con.register("t", pa.table({"x": pa.array(range(5), pa.int64()), "d": run_end_encoded}))
+        return con
+
+    @pytest.mark.parametrize(("value_type", "first", "second", "literal"), CASES)
+    def test_filter_on_run_end_encoded_column(self, value_type, first, second, literal):
+        con = self._connect(value_type, first, second)
+        query = f"SELECT x FROM t WHERE d = {literal} ORDER BY x"
+        assert not _was_pushed(con, query)
+        assert con.execute(query).fetchall() == [(0,), (1,)]
+        assert con.execute("SELECT 42").fetchall() == [(42,)]
+
+    @pytest.mark.parametrize(("value_type", "first", "second", "literal"), CASES)
+    def test_filter_on_other_column(self, value_type, first, second, literal):
+        con = self._connect(value_type, first, second)
+        query = "SELECT x FROM t WHERE x = 2"
+        assert not _was_pushed(con, query)
+        assert con.execute(query).fetchall() == [(2,)]
+
+    @pytest.mark.parametrize(("value_type", "first", "second", "literal"), CASES)
+    def test_top_n_on_other_column(self, value_type, first, second, literal):
+        con = self._connect(value_type, first, second)
+        assert con.execute("SELECT x FROM t ORDER BY x DESC LIMIT 2").fetchall() == [(4,), (3,)]
+
+
 # ===========================================================================
 # 11. Projection / scanner-path interactions
 # ===========================================================================
To Reproduce
import decimal

import duckdb
import pyarrow as pa


def run_end_encoded(values, value_type):
    return pa.RunEndEncodedArray.from_arrays(pa.array([2, 3, 5], pa.int32()), pa.array(values, type=value_type))


con = duckdb.connect()
con.register("ints", pa.table({"x": [1, 2, 3, 4, 5], "d": run_end_encoded([10, 20, 10], pa.int64())}))
con.register("strs", pa.table({"x": [1, 2, 3, 4, 5], "d": run_end_encoded(["a", "b", "a"], pa.string())}))
dec = [decimal.Decimal("1.25"), decimal.Decimal("2.50"), decimal.Decimal("1.25")]
con.register("decs", pa.table({"x": [1, 2, 3, 4, 5], "d": run_end_encoded(dec, pa.decimal128(30, 2))}))

print(con.sql("SELECT x, d FROM ints ORDER BY x").fetchall())  # reads fine
for query in (
    "SELECT x FROM ints WHERE d = 10",  # duckdb.Error: ArrowNotImplementedError: Function 'equal' ...
    "SELECT x FROM ints WHERE x = 3",  # InvalidInputException: ... 'array_filter' has no kernel ...
    "SELECT x FROM strs WHERE x = 3",  # InternalException: ... (expected: STRING, got: STRUCT)
    "SELECT x FROM decs WHERE d = 1.25::DECIMAL(30, 2)",  # InternalException: ... (expected: DECIMAL, got: STRUCT)
    "SELECT 42",  # FatalException: database has been invalidated
):
    try:
        print(con.sql(query).fetchall())
    except duckdb.Error as err:
        print(f"{query}\n  {type(err).__name__}: {err}")

The queries should return [(1,), (2,), (4,), (5,)], [(3,)], [(3,)] and [(1,), (2,), (4,), (5,)], and SELECT 42 should still work afterwards.

OS:

macOS 26.5.1, arm64

DuckDB Package Version:

1.5.6 and 2.0.0.dev2610011535, and a source build of v2.0-cyanoptera at 8bbac25

Python Version:

3.14.7 (pyarrow 26.0.0)

Full Name:

Will Frey

Affiliation:

dataplor

What is the latest build you tested with? If possible, we recommend testing with the latest nightly build.

I have tested with a nightly build

Did you include all relevant data sets for reproducing the issue?

Not applicable - the reproduction does not require a data set

Did you include all code required to reproduce the issue?
  • Yes, I have
Did you include all relevant configuration to reproduce the issue?
  • Yes, I have
Ngôn ngữ chính
Python
Star
191
Fork
120
Merge trung bình
19 giờ 52 phút
Pull request đã merge (30 ngày)
49

Chuẩn bị môi trường

Bắt đầu từ đâu

  1. Đọc hết issue, rồi đọc hướng dẫn đóng góp của dự án.
  2. Bình luận trên issue rằng bạn sẽ nhận — tránh hai người làm cùng một việc.
  3. Fork repository và làm thay đổi trên một nhánh.
  4. Mở pull request có tham chiếu số hiệu của issue.

Issue khác của duckdb/duckdb-python

Tất cả issue của duckdb/duckdb-python

Issue tương tự

Thêm issue về Python

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.