Arrow scan: filter on a dictionary column with string_view values is pushed down and fails #663

--- a/test/arrow/arrow_filter_pushdown.cpp
+++ b/test/arrow/arrow_filter_pushdown.cpp
@@ -1,6 +1,8 @@
 #include "catch.hpp"
 
 #include "arrow/arrow_test_helper.hpp"
+#include "duckdb/common/adbc/single_batch_array_stream.hpp"
+#include "duckdb/common/types/arrow_string_view_type.hpp"
 
 using namespace duckdb;
 
@@ -23,8 +25,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());
@@ -34,6 +36,10 @@
 	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
 static bool StandaloneFilter(const std::string &explain_str) {
 	// Look for FILTER as a standalone operator node name (not "Filters:" which is a scan attribute)
@@ -107,3 +113,142 @@
 		REQUIRE(!FilterInScan(explain_str));
 	}
 }
+
+namespace arrow_filter_pushdown_dict_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};
+const int8_t D_INDICES[5] = {0, 1, 2, 0, 1};
+const int32_t UTF8_OFFSETS[4] = {0, 3, 8, 12};
+const char UTF8_DATA[] = "redgreenblue";
+// a string view array with no variadic data buffers still ends with the (empty) buffer-sizes buffer
+const int64_t NO_BUFFER_SIZES[1] = {0};
+
+enum class DictValues { UTF8, STRING_VIEW };
+
+// One batch of two columns: x INTEGER 0..4, and d holding "red", "green", "blue", "red", "green" as a dictionary
+// with int8 indices whose values are utf8 or string view, as a single-batch stream for binding the scan.
+struct DictTable {
+	explicit DictTable(DictValues values)
+	    : views {arrow_string_view_t(3, "red"), arrow_string_view_t(5, "green"), arrow_string_view_t(4, "blue")} {
+		dict_buffers[0] = nullptr;
+		switch (values) {
+		case DictValues::UTF8:
+			dict_schema.format = "u";
+			dict_buffers[1] = UTF8_OFFSETS;
+			dict_buffers[2] = UTF8_DATA;
+			break;
+		case DictValues::STRING_VIEW:
+			// every value fits inline, so the views buffer is followed only by the buffer sizes
+			dict_schema.format = "vu";
+			dict_buffers[1] = views;
+			dict_buffers[2] = NO_BUFFER_SIZES;
+			break;
+		}
+		dict_schema.release = ReleaseSchema;
+		dict_array.length = 3;
+		dict_array.n_buffers = 3;
+		dict_array.buffers = dict_buffers;
+		dict_array.release = ReleaseArray;
+
+		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;
+
+		d_schema.format = "c";
+		d_schema.name = "d";
+		d_schema.dictionary = &dict_schema;
+		d_schema.release = ReleaseSchema;
+		d_buffers[0] = nullptr;
+		d_buffers[1] = D_INDICES;
+		d_array.length = 5;
+		d_array.n_buffers = 2;
+		d_array.buffers = d_buffers;
+		d_array.dictionary = &dict_array;
+		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);
+	}
+	DictTable(const DictTable &) = delete;
+	~DictTable() {
+		// an EXPLAIN only binds the scan, so the stream is still ours to release
+		if (stream.release) {
+			stream.release(&stream);
+		}
+	}
+
+	arrow_string_view_t views[3];
+	const void *dict_buffers[3];
+	const void *x_buffers[2];
+	const void *d_buffers[2];
+	const void *record_buffers[1] = {nullptr};
+	ArrowSchema *schema_children[2];
+	ArrowArray *array_children[2];
+	ArrowSchema dict_schema {};
+	ArrowSchema x_schema {};
+	ArrowSchema d_schema {};
+	ArrowSchema record_schema {};
+	ArrowArray dict_array {};
+	ArrowArray x_array {};
+	ArrowArray d_array {};
+	ArrowArray record_array {};
+	ArrowArrayStream stream {};
+};
+
+} // namespace arrow_filter_pushdown_dict_test
+
+TEST_CASE("Arrow filter pushdown - dictionary columns are judged by their value type", "[arrow]") {
+	using namespace arrow_filter_pushdown_dict_test;
+	DuckDB db;
+	Connection con(db);
+
+	SECTION("Filter on a dictionary column with string view values: filter above scan (not pushed)") {
+		DictTable table(DictValues::STRING_VIEW);
+		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 string view dictionary: filter pushed into scan") {
+		// pyarrow filters a dictionary column by its indices and never reads the values
+		DictTable table(DictValues::STRING_VIEW);
+		auto explain_str = GetExplainForFilter(con, ArrowTestHelper::ConstructArrowScan(table.stream), "x > 2");
+		REQUIRE(!StandaloneFilter(explain_str));
+		REQUIRE(FilterInScan(explain_str));
+	}
+
+	SECTION("Filter on a dictionary column with utf8 values: filter pushed into scan") {
+		DictTable table(DictValues::UTF8);
+		auto explain_str = GetExplainForFilter(con, ArrowTestHelper::ConstructArrowScan(table.stream), "d = 'red'");
+		REQUIRE(!StandaloneFilter(explain_str));
+		REQUIRE(FilterInScan(explain_str));
+	}
+}