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

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));
+	}
+}

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
 # ===========================================================================