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

Add support for converting DataFusion Timestamp scalar values to Iceberg Datum
for predicate pushdown. This enables timestamp literal comparisons to be pushed
down to the storage layer when DataFusion provides timestamp literals in
ScalarValue form, improving query performance.

Changes:
- Add conversion for all four timestamp time units (Second, Millisecond,
  Microsecond, Nanosecond) to Iceberg's native microsecond representation
- Convert timestamps to Datum::timestamp_micros with appropriate scaling:
  * TimestampSecond: multiply by 1,000,000
  * TimestampMillisecond: multiply by 1,000
  * TimestampMicrosecond: use directly
  * TimestampNanosecond: divide by 1,000 (truncates sub-microsecond precision)
- Add comprehensive unit tests covering all time units and edge cases
- Add sqllogictest (timestamp_predicate_pushdown.slt) to validate end-to-end
  timestamp predicate filtering and pushdown behavior
- Create test_timestamp_table in sqllogictest engine setup
- Update show_tables.slt to include test_timestamp_table

Implementation details:
- Iceberg uses microseconds as the native timestamp representation
- Nanosecond timestamps lose sub-microsecond precision when converted
- Timezone information in ScalarValue is preserved but not used in conversion
  (Iceberg timestamp type is timezone-agnostic)
- Currently DataFusion casts timestamp literals to strings in predicates, but
  this implementation provides the foundation for direct timestamp literal
  support when DataFusion evolves to provide timestamp ScalarValues

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

Author: viirya

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

@@ -201,6 +201,10 @@ fn reverse_predicate_operator(op: PredicateOperator) -> PredicateOperator {
 }
 
 const MILLIS_PER_DAY: i64 = 24 * 60 * 60 * 1000;
+const MICROS_PER_SECOND: i64 = 1_000_000;
+const NANOS_PER_MICRO: i64 = 1_000;
+const MICROS_PER_MILLI: i64 = 1_000;
+
 /// Convert a scalar value to an iceberg datum.
 fn scalar_value_to_datum(value: &ScalarValue) -> Option<Datum> {
     match value {
@@ -216,6 +220,17 @@ fn scalar_value_to_datum(value: &ScalarValue) -> Option<Datum> {
         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)),
+        // Timestamp conversions - convert to microseconds (Iceberg's native timestamp unit)
+        ScalarValue::TimestampSecond(Some(v), _) => {
+            Some(Datum::timestamp_micros(v * MICROS_PER_SECOND))
+        }
+        ScalarValue::TimestampMillisecond(Some(v), _) => {
+            Some(Datum::timestamp_micros(v * MICROS_PER_MILLI))
+        }
+        ScalarValue::TimestampMicrosecond(Some(v), _) => Some(Datum::timestamp_micros(*v)),
+        ScalarValue::TimestampNanosecond(Some(v), _) => {
+            Some(Datum::timestamp_micros(v / NANOS_PER_MICRO))
+        }
         _ => None,
     }
 }
@@ -432,6 +447,51 @@ mod tests {
         assert_eq!(predicate, None);
     }
 
+    #[test]
+    fn test_scalar_value_to_datum_timestamp() {
+        use datafusion::common::ScalarValue;
+
+        // Test TimestampSecond - convert to microseconds
+        let ts_seconds = 1672876800i64; // 2023-01-05 00:00:00 UTC in seconds
+        let datum = super::scalar_value_to_datum(&ScalarValue::TimestampSecond(
+            Some(ts_seconds),
+            None,
+        ));
+        assert_eq!(
+            datum,
+            Some(Datum::timestamp_micros(ts_seconds * 1_000_000))
+        );
+
+        // Test TimestampMillisecond - convert to microseconds
+        let ts_millis = 1672876800000i64; // 2023-01-05 00:00:00 UTC in milliseconds
+        let datum = super::scalar_value_to_datum(&ScalarValue::TimestampMillisecond(
+            Some(ts_millis),
+            None,
+        ));
+        assert_eq!(datum, Some(Datum::timestamp_micros(ts_millis * 1_000)));
+
+        // Test TimestampMicrosecond - use directly
+        let ts_micros = 1672876800000000i64; // 2023-01-05 00:00:00 UTC in microseconds
+        let datum = super::scalar_value_to_datum(&ScalarValue::TimestampMicrosecond(
+            Some(ts_micros),
+            None,
+        ));
+        assert_eq!(datum, Some(Datum::timestamp_micros(ts_micros)));
+
+        // Test TimestampNanosecond - convert to microseconds (truncate)
+        let ts_nanos = 1672876800000000500i64; // 2023-01-05 00:00:00.000000500 UTC in nanoseconds
+        let datum = super::scalar_value_to_datum(&ScalarValue::TimestampNanosecond(
+            Some(ts_nanos),
+            None,
+        ));
+        // Nanoseconds are truncated when converting to microseconds
+        assert_eq!(datum, Some(Datum::timestamp_micros(1672876800000000)));
+
+        // Test None timestamp
+        let datum = super::scalar_value_to_datum(&ScalarValue::TimestampMicrosecond(None, None));
+        assert_eq!(datum, None);
+    }
+
     #[test]
     fn test_scalar_value_to_datum_binary() {
         use datafusion::common::ScalarValue;

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_timestamp_table(&catalog, &namespace).await?;
 
         Ok(Arc::new(
             IcebergCatalogProvider::try_new(Arc::new(catalog)).await?,
@@ -162,4 +163,31 @@ impl DataFusionEngine {
 
         Ok(())
     }
+
+    /// Create a test table with timestamp type column
+    /// Used for testing timestamp predicate pushdown
+    /// TODO: this can be removed when we support CREATE TABLE
+    async fn create_timestamp_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, "ts", Type::Primitive(PrimitiveType::Timestamp)).into(),
+            ])
+            .build()?;
+
+        catalog
+            .create_table(
+                namespace,
+                TableCreation::builder()
+                    .name("test_timestamp_table".to_string())
+                    .schema(schema)
+                    .build(),
+            )
+            .await?;
+
+        Ok(())
+    }
 }

crates/sqllogictest/testdata/schedules/df_test.toml

@@ -34,6 +34,10 @@ slt = "df_test/insert_into.slt"
 engine = "df"
 slt = "df_test/binary_predicate_pushdown.slt"
 
+[[steps]]
+engine = "df"
+slt = "df_test/timestamp_predicate_pushdown.slt"
+
 [[steps]]
 engine = "df"
 slt = "df_test/drop_table.slt"

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

@@ -31,6 +31,9 @@ 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
+default default test_timestamp_table BASE TABLE
+default default test_timestamp_table$manifests BASE TABLE
+default default test_timestamp_table$snapshots BASE TABLE
 default information_schema columns VIEW
 default information_schema df_settings VIEW
 default information_schema parameters VIEW

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

@@ -0,0 +1,133 @@
+# 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 timestamp predicate pushdown
+# This validates that timestamp predicates are pushed down to the storage layer
+# Note: Timestamp scalar value conversion to Datum::timestamp_micros supports
+# direct timestamp literals when DataFusion provides them in ScalarValue form
+
+# Insert test data with timestamps
+# Note: Iceberg timestamps are stored in microseconds
+# We use CAST to convert string timestamps to proper timestamp values
+query I
+INSERT INTO default.default.test_timestamp_table
+VALUES
+  (1, CAST('2023-01-01 00:00:00' AS TIMESTAMP)),
+  (2, CAST('2023-01-05 12:30:00' AS TIMESTAMP)),
+  (3, CAST('2023-01-10 15:45:30' AS TIMESTAMP)),
+  (4, CAST('2023-01-15 09:00:00' AS TIMESTAMP)),
+  (5, CAST('2023-01-20 18:20:10' AS TIMESTAMP))
+----
+5
+
+# Verify timestamp equality predicate is pushed down to IcebergTableScan
+query TT
+EXPLAIN SELECT * FROM default.default.test_timestamp_table WHERE ts = CAST('2023-01-05 12:30:00' AS TIMESTAMP)
+----
+logical_plan
+01)Filter: default.default.test_timestamp_table.ts = TimestampMicrosecond(1672921800000000, None)
+02)--TableScan: default.default.test_timestamp_table projection=[id, ts], partial_filters=[default.default.test_timestamp_table.ts = TimestampMicrosecond(1672921800000000, None)]
+physical_plan
+01)CoalesceBatchesExec: target_batch_size=8192
+02)--FilterExec: ts@1 = 1672921800000000
+03)----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+04)------CooperativeExec
+05)--------IcebergTableScan projection:[id,ts] predicate:[ts = 2023-01-05 12:30:00]
+
+# Verify timestamp equality filtering works
+query I?
+SELECT * FROM default.default.test_timestamp_table WHERE ts = CAST('2023-01-05 12:30:00' AS TIMESTAMP)
+----
+2 2023-01-05T12:30:00
+
+# Verify timestamp greater than predicate is pushed down to IcebergTableScan
+query TT
+EXPLAIN SELECT * FROM default.default.test_timestamp_table WHERE ts > CAST('2023-01-10 00:00:00' AS TIMESTAMP)
+----
+logical_plan
+01)Filter: default.default.test_timestamp_table.ts > TimestampMicrosecond(1673308800000000, None)
+02)--TableScan: default.default.test_timestamp_table projection=[id, ts], partial_filters=[default.default.test_timestamp_table.ts > TimestampMicrosecond(1673308800000000, None)]
+physical_plan
+01)CoalesceBatchesExec: target_batch_size=8192
+02)--FilterExec: ts@1 > 1673308800000000
+03)----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+04)------CooperativeExec
+05)--------IcebergTableScan projection:[id,ts] predicate:[ts > 2023-01-10 00:00:00]
+
+# Verify timestamp greater than filtering
+query I? rowsort
+SELECT * FROM default.default.test_timestamp_table WHERE ts > CAST('2023-01-10 00:00:00' AS TIMESTAMP)
+----
+3 2023-01-10T15:45:30
+4 2023-01-15T09:00:00
+5 2023-01-20T18:20:10
+
+# Test timestamp less than or equal filtering
+query I? rowsort
+SELECT * FROM default.default.test_timestamp_table WHERE ts <= CAST('2023-01-05 12:30:00' AS TIMESTAMP)
+----
+1 2023-01-01T00:00:00
+2 2023-01-05T12:30:00
+
+# Verify timestamp range predicate (AND of two comparisons) is pushed down
+query TT
+EXPLAIN SELECT * FROM default.default.test_timestamp_table
+WHERE ts >= CAST('2023-01-05 00:00:00' AS TIMESTAMP)
+  AND ts <= CAST('2023-01-15 23:59:59' AS TIMESTAMP)
+----
+logical_plan
+01)Filter: default.default.test_timestamp_table.ts >= TimestampMicrosecond(1672876800000000, None) AND default.default.test_timestamp_table.ts <= TimestampMicrosecond(1673827199000000, None)
+02)--TableScan: default.default.test_timestamp_table projection=[id, ts], partial_filters=[default.default.test_timestamp_table.ts >= TimestampMicrosecond(1672876800000000, None), default.default.test_timestamp_table.ts <= TimestampMicrosecond(1673827199000000, None)]
+physical_plan
+01)CoalesceBatchesExec: target_batch_size=8192
+02)--FilterExec: ts@1 >= 1672876800000000 AND ts@1 <= 1673827199000000
+03)----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+04)------CooperativeExec
+05)--------IcebergTableScan projection:[id,ts] predicate:[(ts >= 2023-01-05 00:00:00) AND (ts <= 2023-01-15 23:59:59)]
+
+# Test timestamp range predicate filtering
+query I? rowsort
+SELECT * FROM default.default.test_timestamp_table
+WHERE ts >= CAST('2023-01-05 00:00:00' AS TIMESTAMP)
+  AND ts <= CAST('2023-01-15 23:59:59' AS TIMESTAMP)
+----
+2 2023-01-05T12:30:00
+3 2023-01-10T15:45:30
+4 2023-01-15T09:00:00
+
+# Test timestamp predicate combined with other predicates
+query I? rowsort
+SELECT * FROM default.default.test_timestamp_table
+WHERE ts >= CAST('2023-01-10 00:00:00' AS TIMESTAMP) AND id < 5
+----
+3 2023-01-10T15:45:30
+4 2023-01-15T09:00:00
+
+# Test timestamp NOT EQUAL predicate
+query I? rowsort
+SELECT * FROM default.default.test_timestamp_table WHERE ts != CAST('2023-01-05 12:30:00' AS TIMESTAMP)
+----
+1 2023-01-01T00:00:00
+3 2023-01-10T15:45:30
+4 2023-01-15T09:00:00
+5 2023-01-20T18:20:10
+
+# Test timestamp less than filtering
+query I? rowsort
+SELECT * FROM default.default.test_timestamp_table WHERE ts < CAST('2023-01-05 00:00:00' AS TIMESTAMP)
+----
+1 2023-01-01T00:00:00