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

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

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/integrations/datafusion/tests/integration_datafusion_test.rs

@@ -807,6 +807,55 @@ async fn test_insert_into_nested() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn test_binary_predicate_pushdown() -> Result<()> {
+    let iceberg_catalog = get_iceberg_catalog().await;
+    let namespace = NamespaceIdent::new("ns".to_string());
+    set_test_namespace(&iceberg_catalog, &namespace).await?;
+
+    // Create a schema with binary type
+    let schema = Schema::builder()
+        .with_schema_id(0)
+        .with_fields(vec![
+            NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(),
+            NestedField::optional(2, "data", Type::Primitive(PrimitiveType::Binary)).into(),
+        ])
+        .build()?;
+    let creation = get_table_creation(temp_path(), "binary_table", Some(schema))?;
+    iceberg_catalog.create_table(&namespace, creation).await?;
+
+    let client = Arc::new(iceberg_catalog);
+    let catalog = Arc::new(IcebergCatalogProvider::try_new(client).await?);
+
+    let ctx = SessionContext::new();
+    ctx.register_catalog("catalog", catalog);
+
+    // Test binary predicate pushdown using X'...' syntax for binary literals
+    let records = ctx
+        .sql("EXPLAIN select * from catalog.ns.binary_table where data = X'0102'")
+        .await
+        .unwrap()
+        .collect()
+        .await
+        .unwrap();
+    assert_eq!(1, records.len());
+    let plan = records[0]
+        .column(1)
+        .as_any()
+        .downcast_ref::<StringArray>()
+        .unwrap();
+    let plan_str = plan.value(1);
+
+    // The key validation: binary predicate should be pushed down to IcebergTableScan
+    // Without binary Datum conversion, this would stay in a FilterExec
+    assert!(
+        plan_str.contains("predicate:[data = ") || plan_str.contains("predicate:[(data = "),
+        "Binary predicate should be pushed down. Plan: {plan_str}"
+    );
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn test_insert_into_partitioned() -> Result<()> {
     let iceberg_catalog = get_iceberg_catalog().await;