feat(datafusion): Add Boolean predicate pushdown support

This commit adds comprehensive support for pushing down Boolean predicates
to the Iceberg table scan layer, improving query performance by filtering
data at the storage level.

Changes:
- Enhanced expr_to_predicate.rs to handle boolean column expressions:
  * Bare boolean columns in filters (e.g., WHERE is_active) are converted
    to column = true predicates
  * NOT of boolean columns (e.g., WHERE NOT is_active) are converted to
    column = false predicates
  * Added Boolean scalar value to Datum conversion
- Added comprehensive sqllogictest (boolean_predicate_pushdown.slt) with:
  * Tests for is_active = true/false with EXPLAIN verification
  * Tests for is_active != true with EXPLAIN verification
  * Tests for combined predicates (AND/OR)
  * Tests for IS NULL/IS NOT NULL on boolean columns
- Created test_boolean_table in engine setup for testing
- Updated test schedule and show_tables baseline

All tests verify that predicates are successfully pushed down to
IcebergTableScan, not just executed in FilterExec.

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: viirya

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

@@ -51,8 +51,18 @@ pub fn convert_filters_to_predicate(filters: &[Expr]) -> Option<Predicate> {
 fn convert_filter_to_predicate(expr: &Expr) -> Option<Predicate> {
     match to_iceberg_predicate(expr) {
         TransformedResult::Predicate(predicate) => Some(predicate),
-        TransformedResult::Column(_) | TransformedResult::Literal(_) => {
-            unreachable!("Not a valid expression: {:?}", expr)
+        TransformedResult::Column(column) => {
+            // A bare column in a filter context represents a boolean column check
+            // Convert it to: column = true
+            Some(Predicate::Binary(BinaryExpression::new(
+                PredicateOperator::Eq,
+                column,
+                Datum::bool(true),
+            )))
+        }
+        TransformedResult::Literal(_) => {
+            // Literal values in filter context cannot be pushed down
+            None
         }
         _ => None,
     }
@@ -75,6 +85,14 @@ fn to_iceberg_predicate(expr: &Expr) -> TransformedResult {
             let expr = to_iceberg_predicate(exp);
             match expr {
                 TransformedResult::Predicate(p) => TransformedResult::Predicate(!p),
+                TransformedResult::Column(column) => {
+                    // NOT of a bare boolean column: NOT col => col = false
+                    TransformedResult::Predicate(Predicate::Binary(BinaryExpression::new(
+                        PredicateOperator::Eq,
+                        column,
+                        Datum::bool(false),
+                    )))
+                }
                 _ => TransformedResult::NotTransformed,
             }
         }
@@ -254,6 +272,7 @@ const MILLIS_PER_DAY: i64 = 24 * 60 * 60 * 1000;
 /// Convert a scalar value to an iceberg datum.
 fn scalar_value_to_datum(value: &ScalarValue) -> Option<Datum> {
     match value {
+        ScalarValue::Boolean(Some(v)) => Some(Datum::bool(*v)),
         ScalarValue::Int8(Some(v)) => Some(Datum::int(*v as i32)),
         ScalarValue::Int16(Some(v)) => Some(Datum::int(*v as i32)),
         ScalarValue::Int32(Some(v)) => Some(Datum::int(*v)),
@@ -509,6 +528,23 @@ mod tests {
         assert_eq!(predicate, expected_predicate);
     }
 
+    #[test]
+    fn test_scalar_value_to_datum_boolean() {
+        use datafusion::common::ScalarValue;
+
+        // Test boolean true
+        let datum = super::scalar_value_to_datum(&ScalarValue::Boolean(Some(true)));
+        assert_eq!(datum, Some(Datum::bool(true)));
+
+        // Test boolean false
+        let datum = super::scalar_value_to_datum(&ScalarValue::Boolean(Some(false)));
+        assert_eq!(datum, Some(Datum::bool(false)));
+
+        // Test None boolean
+        let datum = super::scalar_value_to_datum(&ScalarValue::Boolean(None));
+        assert_eq!(datum, None);
+    }
+
     #[test]
     fn test_predicate_conversion_with_like_starts_with() {
         let sql = "bar LIKE 'test%'";

crates/sqllogictest/src/engine/datafusion.rs

@@ -96,6 +96,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?;
+        Self::create_boolean_table(&catalog, &namespace).await?;
 
         Ok(Arc::new(
             IcebergCatalogProvider::try_new(Arc::new(catalog)).await?,
@@ -162,4 +163,32 @@ impl DataFusionEngine {
 
         Ok(())
     }
+
+    /// Create a test table with boolean type column
+    /// Used for testing boolean predicate pushdown
+    /// TODO: this can be removed when we support CREATE TABLE
+    async fn create_boolean_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, "is_active", Type::Primitive(PrimitiveType::Boolean)).into(),
+                NestedField::optional(3, "description", Type::Primitive(PrimitiveType::String)).into(),
+            ])
+            .build()?;
+
+        catalog
+            .create_table(
+                namespace,
+                TableCreation::builder()
+                    .name("test_boolean_table".to_string())
+                    .schema(schema)
+                    .build(),
+            )
+            .await?;
+
+        Ok(())
+    }
 }

crates/sqllogictest/testdata/schedules/df_test.toml

@@ -46,6 +46,10 @@ slt = "df_test/binary_predicate_pushdown.slt"
 engine = "df"
 slt = "df_test/like_predicate_pushdown.slt"
 
+[[steps]]
+engine = "df"
+slt = "df_test/boolean_predicate_pushdown.slt"
+
 [[steps]]
 engine = "df"
 slt = "df_test/drop_table.slt"

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

@@ -0,0 +1,122 @@
+# 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.
+
+# Insert test data into test_boolean_table
+statement ok
+INSERT INTO default.default.test_boolean_table VALUES
+  (1, true, 'Active user'),
+  (2, false, 'Inactive user'),
+  (3, true, 'Premium member'),
+  (4, false, 'Trial expired'),
+  (5, true, 'Verified account'),
+  (6, NULL, 'Pending verification')
+
+# Verify boolean equality predicate is pushed down to IcebergTableScan
+query TT
+EXPLAIN SELECT * FROM default.default.test_boolean_table WHERE is_active = true
+----
+logical_plan
+01)Filter: default.default.test_boolean_table.is_active
+02)--TableScan: default.default.test_boolean_table projection=[id, is_active, description], partial_filters=[default.default.test_boolean_table.is_active]
+physical_plan
+01)CoalesceBatchesExec: target_batch_size=8192
+02)--FilterExec: is_active@1
+03)----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+04)------CooperativeExec
+05)--------IcebergTableScan projection:[id,is_active,description] predicate:[is_active = true]
+
+# Query with is_active = true
+query ITT rowsort
+SELECT * FROM default.default.test_boolean_table WHERE is_active = true
+----
+1 true Active user
+3 true Premium member
+5 true Verified account
+
+# Verify boolean false predicate is pushed down to IcebergTableScan
+query TT
+EXPLAIN SELECT * FROM default.default.test_boolean_table WHERE is_active = false
+----
+logical_plan
+01)Filter: NOT default.default.test_boolean_table.is_active
+02)--TableScan: default.default.test_boolean_table projection=[id, is_active, description], partial_filters=[NOT default.default.test_boolean_table.is_active]
+physical_plan
+01)CoalesceBatchesExec: target_batch_size=8192
+02)--FilterExec: NOT is_active@1
+03)----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+04)------CooperativeExec
+05)--------IcebergTableScan projection:[id,is_active,description] predicate:[is_active = false]
+
+# Query with is_active = false
+query ITT rowsort
+SELECT * FROM default.default.test_boolean_table WHERE is_active = false
+----
+2 false Inactive user
+4 false Trial expired
+
+# Verify boolean NOT EQUAL predicate is pushed down
+query TT
+EXPLAIN SELECT * FROM default.default.test_boolean_table WHERE is_active != true
+----
+logical_plan
+01)Filter: NOT default.default.test_boolean_table.is_active
+02)--TableScan: default.default.test_boolean_table projection=[id, is_active, description], partial_filters=[NOT default.default.test_boolean_table.is_active]
+physical_plan
+01)CoalesceBatchesExec: target_batch_size=8192
+02)--FilterExec: NOT is_active@1
+03)----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+04)------CooperativeExec
+05)--------IcebergTableScan projection:[id,is_active,description] predicate:[is_active = false]
+
+# Query with is_active != true (includes false and NULL)
+query ITT rowsort
+SELECT * FROM default.default.test_boolean_table WHERE is_active != true
+----
+2 false Inactive user
+4 false Trial expired
+
+# Test combined boolean predicates with AND
+query ITT rowsort
+SELECT * FROM default.default.test_boolean_table WHERE is_active = true AND id > 2
+----
+3 true Premium member
+5 true Verified account
+
+# Test combined boolean predicates with OR
+query ITT rowsort
+SELECT * FROM default.default.test_boolean_table WHERE is_active = true OR id = 2
+----
+1 true Active user
+2 false Inactive user
+3 true Premium member
+5 true Verified account
+
+# Test IS NULL on boolean column
+query ITT
+SELECT * FROM default.default.test_boolean_table WHERE is_active IS NULL
+----
+6 NULL Pending verification
+
+# Test IS NOT NULL on boolean column
+query ITT rowsort
+SELECT * FROM default.default.test_boolean_table WHERE is_active IS NOT NULL
+----
+1 true Active user
+2 false Inactive user
+3 true Premium member
+4 false Trial expired
+5 true Verified account

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

@@ -28,6 +28,9 @@ 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_boolean_table BASE TABLE
+default default test_boolean_table$manifests BASE TABLE
+default default test_boolean_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