feat(datafusion): Add Binary scalar value conversion for predicate pushdown (apache#2048)

## Which issue does this PR close?

- Closes #.

## What changes are included in this PR?

Add support for converting Binary and LargeBinary DataFusion ScalarValue
types to Iceberg Datum, enabling binary predicates to be pushed down to
the Iceberg storage layer.

This conversion allows SQL queries with binary hex literals (X'...') to
push predicates down to Iceberg, improving query performance by
filtering data at the storage level rather than in DataFusion.

The integration test verifies that binary predicates are successfully
pushed down end-to-end:
- Without conversion: predicate stays in FilterExec with predicate:[]
- With conversion: predicate pushed to IcebergTableScan

Other scalar types (Boolean, Timestamp, Decimal) were investigated but
excluded because they are not reachable through practical usage:
- Boolean: DataFusion aggressively optimizes comparisons (e.g., x=true
becomes just x) before reaching the converter
- Timestamp/Decimal: SQL literals are converted to strings/other types
before reaching the converter

## Are these changes tested?

---------

Co-authored-by: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: viirya

crates/integrations/datafusion/src/physical_plan/expr_to_predicate.rs

@@ -212,6 +212,8 @@ fn scalar_value_to_datum(value: &ScalarValue) -> Option<Datum> {
         ScalarValue::Float64(Some(v)) => Some(Datum::double(*v)),
         ScalarValue::Utf8(Some(v)) => Some(Datum::string(v.clone())),
         ScalarValue::LargeUtf8(Some(v)) => Some(Datum::string(v.clone())),
+        ScalarValue::Binary(Some(v)) => Some(Datum::binary(v.clone())),
+        ScalarValue::LargeBinary(Some(v)) => Some(Datum::binary(v.clone())),
         ScalarValue::Date32(Some(v)) => Some(Datum::date(*v)),
         ScalarValue::Date64(Some(v)) => Some(Datum::date((*v / MILLIS_PER_DAY) as i32)),
         _ => None,
@@ -429,4 +431,31 @@ mod tests {
         let predicate = convert_to_iceberg_predicate(sql);
         assert_eq!(predicate, None);
     }
+
+    #[test]
+    fn test_scalar_value_to_datum_binary() {
+        use datafusion::common::ScalarValue;
+
+        let bytes = vec![1u8, 2u8, 3u8];
+        let datum = super::scalar_value_to_datum(&ScalarValue::Binary(Some(bytes.clone())));
+        assert_eq!(datum, Some(Datum::binary(bytes.clone())));
+
+        let datum = super::scalar_value_to_datum(&ScalarValue::LargeBinary(Some(bytes.clone())));
+        assert_eq!(datum, Some(Datum::binary(bytes)));
+
+        let datum = super::scalar_value_to_datum(&ScalarValue::Binary(None));
+        assert_eq!(datum, None);
+    }
+
+    #[test]
+    fn test_predicate_conversion_with_binary() {
+        let sql = "foo = 1 and bar = X'0102'";
+        let predicate = convert_to_iceberg_predicate(sql).unwrap();
+        // Binary literals are converted to Datum::binary
+        // Note: SQL literal 1 is converted to Long by DataFusion
+        let expected_predicate = Reference::new("foo")
+            .equal_to(Datum::long(1))
+            .and(Reference::new("bar").equal_to(Datum::binary(vec![1u8, 2u8])));
+        assert_eq!(predicate, expected_predicate);
+    }
 }

crates/sqllogictest/src/engine/datafusion.rs

@@ -95,6 +95,7 @@ impl DataFusionEngine {
 
         // Create partitioned test table (unpartitioned tables are now created via SQL)
         Self::create_partitioned_table(&catalog, &namespace).await?;
+        Self::create_binary_table(&catalog, &namespace).await?;
 
         Ok(Arc::new(
             IcebergCatalogProvider::try_new(Arc::new(catalog)).await?,
@@ -134,4 +135,31 @@ impl DataFusionEngine {
 
         Ok(())
     }
+
+    /// Create a test table with binary type column
+    /// Used for testing binary predicate pushdown
+    /// TODO: this can be removed when we support CREATE TABLE
+    async fn create_binary_table(
+        catalog: &impl Catalog,
+        namespace: &NamespaceIdent,
+    ) -> anyhow::Result<()> {
+        let schema = Schema::builder()
+            .with_fields(vec![
+                NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(),
+                NestedField::optional(2, "data", Type::Primitive(PrimitiveType::Binary)).into(),
+            ])
+            .build()?;
+
+        catalog
+            .create_table(
+                namespace,
+                TableCreation::builder()
+                    .name("test_binary_table".to_string())
+                    .schema(schema)
+                    .build(),
+            )
+            .await?;
+
+        Ok(())
+    }
 }

crates/sqllogictest/testdata/schedules/df_test.toml

@@ -29,3 +29,7 @@ slt = "df_test/create_table.slt"
 [[steps]]
 engine = "df"
 slt = "df_test/insert_into.slt"
+
+[[steps]]
+engine = "df"
+slt = "df_test/binary_predicate_pushdown.slt"

crates/sqllogictest/testdata/slts/df_test/binary_predicate_pushdown.slt

@@ -0,0 +1,69 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# Test that binary predicates are pushed down to IcebergTableScan
+# This validates that ScalarValue::Binary is correctly converted to Datum::binary
+
+# Verify EXPLAIN shows binary predicate is pushed down to IcebergTableScan
+query TT
+EXPLAIN SELECT * FROM default.default.test_binary_table WHERE data = X'0102'
+----
+logical_plan
+01)Filter: default.default.test_binary_table.data = LargeBinary("1,2")
+02)--TableScan: default.default.test_binary_table projection=[id, data], partial_filters=[default.default.test_binary_table.data = LargeBinary("1,2")]
+physical_plan
+01)CoalesceBatchesExec: target_batch_size=8192
+02)--FilterExec: data@1 = 0102
+03)----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+04)------CooperativeExec
+05)--------IcebergTableScan projection:[id,data] predicate:[data = 0102]
+
+# Verify empty result from empty table
+query I?
+SELECT * FROM default.default.test_binary_table WHERE data = X'0102'
+----
+
+# Insert test data
+query I
+INSERT INTO default.default.test_binary_table VALUES (1, X'0102')
+----
+1
+
+# Verify binary predicate filtering works
+query I?
+SELECT * FROM default.default.test_binary_table WHERE data = X'0102'
+----
+1 0102
+
+# Insert more rows with different binary values
+query I
+INSERT INTO default.default.test_binary_table VALUES (2, X'0304'), (3, X'0102'), (4, X'0506')
+----
+3
+
+# Verify binary predicate filters correctly
+query I? rowsort
+SELECT * FROM default.default.test_binary_table WHERE data = X'0102'
+----
+1 0102
+3 0102
+
+# Verify different binary value
+query I?
+SELECT * FROM default.default.test_binary_table WHERE data = X'0506'
+----
+4 0506

crates/sqllogictest/testdata/slts/df_test/show_tables.slt

@@ -25,6 +25,9 @@ datafusion information_schema routines VIEW
 datafusion information_schema schemata VIEW
 datafusion information_schema tables VIEW
 datafusion information_schema views VIEW
+default default test_binary_table BASE TABLE
+default default test_binary_table$manifests BASE TABLE
+default default test_binary_table$snapshots BASE TABLE
 default default test_partitioned_table BASE TABLE
 default default test_partitioned_table$manifests BASE TABLE
 default default test_partitioned_table$snapshots BASE TABLE