fix(scan): promote date column bounds to timestamp

crates/iceberg/src/expr/visitors/inclusive_metrics_evaluator.rs

@@ -122,7 +122,7 @@ impl<'a> InclusiveMetricsEvaluator<'a> {
         };
 
         if let Some(bound) = bound {
-            let bound = &*Self::promote_bound(bound, reference);
+            let bound = &*Self::promote_bound(bound, reference)?;
             if cmp_fn(bound, datum) {
                 return ROWS_MIGHT_MATCH;
             }
@@ -137,13 +137,19 @@ impl<'a> InclusiveMetricsEvaluator<'a> {
     /// narrower type (e.g. an `int` column later promoted to `long`) compares
     /// against a predicate bound to the wider type. Borrows the bound when nothing
     /// is promoted, so unpromoted bounds (e.g. strings/binaries) are not copied.
-    fn promote_bound<'b>(bound: &'b Datum, reference: &BoundReference) -> Cow<'b, Datum> {
-        reference
-            .field()
-            .field_type
-            .as_primitive_type()
-            .and_then(|primitive_type| bound.promote_to(primitive_type))
-            .map_or(Cow::Borrowed(bound), Cow::Owned)
+    /// A `date` bound read for a column promoted to `timestamp` / `timestamp_ns` is
+    /// converted to midnight of that day; one outside the timestamp range is an error.
+    fn promote_bound<'b>(
+        bound: &'b Datum,
+        reference: &BoundReference,
+    ) -> crate::Result<Cow<'b, Datum>> {
+        let promoted = match reference.field().field_type.as_primitive_type() {
+            Some(primitive_type) => bound
+                .promote_to(primitive_type)
+                .map_err(|err| err.with_context("field", reference.field().name.clone()))?,
+            None => None,
+        };
+        Ok(promoted.map_or(Cow::Borrowed(bound), Cow::Owned))
     }
 }
 
@@ -275,7 +281,7 @@ impl BoundPredicateVisitor for InclusiveMetricsEvaluator<'_> {
         }
 
         if let Some(lower_bound) = self.lower_bound(field_id) {
-            let lower_bound = &*Self::promote_bound(lower_bound, reference);
+            let lower_bound = &*Self::promote_bound(lower_bound, reference)?;
             if lower_bound.is_nan() {
                 // NaN indicates unreliable bounds.
                 // See the InclusiveMetricsEvaluator docs for more.
@@ -286,7 +292,7 @@ impl BoundPredicateVisitor for InclusiveMetricsEvaluator<'_> {
         }
 
         if let Some(upper_bound) = self.upper_bound(field_id) {
-            let upper_bound = &*Self::promote_bound(upper_bound, reference);
+            let upper_bound = &*Self::promote_bound(upper_bound, reference)?;
             if upper_bound.is_nan() {
                 // NaN indicates unreliable bounds.
                 // See the InclusiveMetricsEvaluator docs for more.
@@ -457,10 +463,12 @@ impl BoundPredicateVisitor for InclusiveMetricsEvaluator<'_> {
 
         let lower_bound = self
             .lower_bound(field_id)
-            .map(|bound| Self::promote_bound(bound, reference));
+            .map(|bound| Self::promote_bound(bound, reference))
+            .transpose()?;
         let upper_bound = self
             .upper_bound(field_id)
-            .map(|bound| Self::promote_bound(bound, reference));
+            .map(|bound| Self::promote_bound(bound, reference))
+            .transpose()?;
 
         // A NaN bound is unreliable on that side only. Drop it to unbounded
         // so a valid bound can still prune.
@@ -618,6 +626,132 @@ mod test {
         assert!(InclusiveMetricsEvaluator::eval(&predicate, &data_file, true).unwrap());
     }
 
+    // 19_723 days after the epoch is 2024-01-01. Bounds under test: [19_723, 19_730].
+    const MICROS_PER_DAY: i64 = 86_400_000_000;
+    const NANOS_PER_DAY: i64 = 86_400_000_000_000;
+
+    #[test]
+    fn eval_promotes_date_bounds() {
+        // A column promoted date->timestamp keeps date-typed (day count) bounds in
+        // pre-promotion manifests. Compared as raw numbers, a day count is far below
+        // any timestamp in micros/nanos, so inequalities and set predicates would
+        // wrongly prune the file without promotion. Each operator gets a literal
+        // inside the bounds (kept) and one outside (pruned), for both targets.
+        let data_file = data_file_with_bounds(Datum::date(19_723), Datum::date(19_730));
+
+        for primitive in [PrimitiveType::Timestamp, PrimitiveType::TimestampNs] {
+            let schema = promotion_schema(primitive.clone());
+            let ts = |days: i64| match primitive {
+                PrimitiveType::Timestamp => Datum::timestamp_micros(days * MICROS_PER_DAY),
+                PrimitiveType::TimestampNs => Datum::timestamp_nanos(days * NANOS_PER_DAY),
+                _ => unreachable!(),
+            };
+
+            let cases = [
+                (
+                    "less-than inside",
+                    Predicate::Binary(BinaryExpression::new(
+                        LessThan,
+                        Reference::new("a"),
+                        ts(19_725),
+                    )),
+                    true,
+                ),
+                (
+                    "less-than outside",
+                    Predicate::Binary(BinaryExpression::new(
+                        LessThan,
+                        Reference::new("a"),
+                        ts(19_722),
+                    )),
+                    false,
+                ),
+                (
+                    "greater-than inside",
+                    Predicate::Binary(BinaryExpression::new(
+                        GreaterThan,
+                        Reference::new("a"),
+                        ts(19_725),
+                    )),
+                    true,
+                ),
+                (
+                    "greater-than outside",
+                    Predicate::Binary(BinaryExpression::new(
+                        GreaterThan,
+                        Reference::new("a"),
+                        ts(19_731),
+                    )),
+                    false,
+                ),
+                (
+                    "eq inside",
+                    Predicate::Binary(BinaryExpression::new(Eq, Reference::new("a"), ts(19_725))),
+                    true,
+                ),
+                (
+                    "eq outside",
+                    Predicate::Binary(BinaryExpression::new(Eq, Reference::new("a"), ts(19_722))),
+                    false,
+                ),
+                (
+                    "in inside",
+                    Predicate::Set(SetExpression::new(
+                        In,
+                        Reference::new("a"),
+                        FnvHashSet::from_iter([ts(19_700), ts(19_725)]),
+                    )),
+                    true,
+                ),
+                (
+                    "in outside",
+                    Predicate::Set(SetExpression::new(
+                        In,
+                        Reference::new("a"),
+                        FnvHashSet::from_iter([ts(19_700), ts(19_710)]),
+                    )),
+                    false,
+                ),
+            ];
+
+            for (label, predicate, might_match) in cases {
+                let bound = predicate.bind(schema.clone(), true).unwrap();
+                assert_eq!(
+                    InclusiveMetricsEvaluator::eval(&bound, &data_file, true).unwrap(),
+                    might_match,
+                    "{primitive:?}: {label}"
+                );
+            }
+        }
+    }
+
+    #[test]
+    fn eval_errors_on_date_bound_out_of_timestamp_ns_range() {
+        // A date in year 3000 has no `timestamp_ns` value; evaluating must fail rather
+        // than compare a wrapped bound.
+        let schema = promotion_schema(PrimitiveType::TimestampNs);
+        let data_file = data_file_with_bounds(Datum::date(19_723), Datum::date(376_200));
+        let predicate = Predicate::Binary(BinaryExpression::new(
+            LessThan,
+            Reference::new("a"),
+            Datum::timestamp_nanos(19_725 * NANOS_PER_DAY),
+        ))
+        .bind(schema.clone(), true)
+        .unwrap();
+        // `<` only reads the lower bound, which is in range.
+        assert!(InclusiveMetricsEvaluator::eval(&predicate, &data_file, true).unwrap());
+
+        let predicate = Predicate::Binary(BinaryExpression::new(
+            GreaterThan,
+            Reference::new("a"),
+            Datum::timestamp_nanos(19_725 * NANOS_PER_DAY),
+        ))
+        .bind(schema, true)
+        .unwrap();
+        let err = InclusiveMetricsEvaluator::eval(&predicate, &data_file, true).unwrap_err();
+        assert_eq!(err.kind(), crate::ErrorKind::DataInvalid);
+    }
+
     #[test]
     fn test_all_nulls() {
         let result =

crates/iceberg/src/spec/manifest/mod.rs

@@ -1906,6 +1906,98 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_parse_manifest_promotes_date_bounds_under_timestamp_schema() {
+        // Under a timestamp / timestamp_ns schema, a 4-byte bound is a date. Promotion
+        // does not rewrite bounds, so a merged manifest can carry date bytes that must
+        // decode to midnight of that day rather than failing to load.
+        let schema = Arc::new(
+            Schema::builder()
+                .with_fields(vec![
+                    NestedField::optional(1, "ts", Type::Primitive(PrimitiveType::Timestamp))
+                        .into(),
+                    NestedField::optional(2, "ns", Type::Primitive(PrimitiveType::TimestampNs))
+                        .into(),
+                ])
+                .build()
+                .unwrap(),
+        );
+        let partition_spec = PartitionSpec::builder(schema.clone())
+            .with_spec_id(0)
+            .build()
+            .unwrap();
+        let metadata = ManifestMetadata {
+            schema_id: 0,
+            schema,
+            partition_spec,
+            content: ManifestContentType::Data,
+            format_version: FormatVersion::V2,
+        };
+        let days = 19_723_i32;
+        let mut entry = expected_entry(Struct::empty());
+        entry.data_file.lower_bounds =
+            HashMap::from([(1, Datum::date(days)), (2, Datum::date(days))]);
+        entry.data_file.upper_bounds =
+            HashMap::from([(1, Datum::date(days)), (2, Datum::date(days))]);
+
+        let partition_type = metadata
+            .partition_spec
+            .partition_type(&metadata.schema)
+            .unwrap();
+        let avro_schema = manifest_schema_v2(&partition_type).unwrap();
+        let mut writer = Writer::with_codec(&avro_schema, Vec::new(), Codec::Snappy).unwrap();
+        for (key, value) in [
+            ("schema", to_vec(&metadata.schema).unwrap()),
+            ("schema-id", metadata.schema_id.to_string().into_bytes()),
+            (
+                "partition-spec",
+                to_vec(&metadata.partition_spec.fields()).unwrap(),
+            ),
+            (
+                "partition-spec-id",
+                metadata.partition_spec.spec_id().to_string().into_bytes(),
+            ),
+            (
+                "format-version",
+                (metadata.format_version as u8).to_string().into_bytes(),
+            ),
+            ("content", metadata.content.to_string().into_bytes()),
+        ] {
+            writer.add_user_metadata(key.to_string(), value).unwrap();
+        }
+        let value = to_value(
+            _serde::ManifestEntryV2::try_from(entry, &Type::Struct(partition_type)).unwrap(),
+        )
+        .unwrap()
+        .resolve(&avro_schema)
+        .unwrap();
+        writer.append_value(value).unwrap();
+        let bs = writer.into_inner().unwrap();
+
+        let manifest = Manifest::parse_avro(&bs).unwrap();
+        let file = &manifest.entries()[0].data_file;
+        assert_eq!(
+            file.lower_bounds.get(&1),
+            Some(&Datum::timestamp_micros(i64::from(days) * 86_400_000_000))
+        );
+        assert_eq!(
+            file.lower_bounds.get(&2),
+            Some(&Datum::timestamp_nanos(
+                i64::from(days) * 86_400_000_000_000
+            ))
+        );
+        assert_eq!(
+            file.upper_bounds.get(&1),
+            Some(&Datum::timestamp_micros(i64::from(days) * 86_400_000_000))
+        );
+        assert_eq!(
+            file.upper_bounds.get(&2),
+            Some(&Datum::timestamp_nanos(
+                i64::from(days) * 86_400_000_000_000
+            ))
+        );
+    }
+
     #[test]
     fn test_parse_manifest_reads_required_writer_fields_as_optional() {
         // Fields that are optional in the reader schema are written without a

crates/iceberg/src/spec/values/datum.rs

@@ -33,7 +33,9 @@ use super::decimal_utils::{
     decimal_precision, decimal_scale, i128_from_be_bytes, i128_to_be_bytes_min,
 };
 use super::literal::Literal;
-use super::primitive::PrimitiveLiteral;
+use super::primitive::{
+    MICROSECONDS_PER_DAY, NANOSECONDS_PER_DAY, PrimitiveLiteral, date_to_timestamp,
+};
 use super::serde::_serde::RawLiteral;
 use super::temporal::{date, time, timestamp, timestamptz};
 use crate::ensure_data_valid;
@@ -422,13 +424,31 @@ impl Datum {
             PrimitiveType::Date => PrimitiveLiteral::Int(i32::from_le_bytes(bytes.try_into()?)),
             PrimitiveType::Time => PrimitiveLiteral::Long(i64::from_le_bytes(bytes.try_into()?)),
             PrimitiveType::Timestamp => {
-                PrimitiveLiteral::Long(i64::from_le_bytes(bytes.try_into()?))
+                if bytes.len() == 4 {
+                    // Bound written as a date under a later timestamp schema.
+                    date_to_timestamp(
+                        i32::from_le_bytes(bytes.try_into()?),
+                        MICROSECONDS_PER_DAY,
+                        &data_type,
+                    )?
+                } else {
+                    PrimitiveLiteral::Long(i64::from_le_bytes(bytes.try_into()?))
+                }
             }
             PrimitiveType::Timestamptz => {
                 PrimitiveLiteral::Long(i64::from_le_bytes(bytes.try_into()?))
             }
             PrimitiveType::TimestampNs => {
-                PrimitiveLiteral::Long(i64::from_le_bytes(bytes.try_into()?))
+                if bytes.len() == 4 {
+                    // Bound written as a date under a later timestamp_ns schema.
+                    date_to_timestamp(
+                        i32::from_le_bytes(bytes.try_into()?),
+                        NANOSECONDS_PER_DAY,
+                        &data_type,
+                    )?
+                } else {
+                    PrimitiveLiteral::Long(i64::from_le_bytes(bytes.try_into()?))
+                }
             }
             PrimitiveType::TimestamptzNs => {
                 PrimitiveLiteral::Long(i64::from_le_bytes(bytes.try_into()?))
@@ -1230,14 +1250,45 @@ impl Datum {
     }
 
     /// Returns a copy whose literal is widened to `target` for the value
-    /// promotions `int` -> `long` and `float` -> `double`, so a value written
-    /// under the narrower type compares against a predicate bound to the wider
-    /// one. Returns `None` when `target` does not widen this value (decimals
-    /// already compare across precisions), so callers can avoid a copy.
-    pub(crate) fn promote_to(&self, target: &PrimitiveType) -> Option<Datum> {
-        self.literal
-            .promote_to(target)
-            .map(|literal| Datum::new(target.clone(), literal))
+    /// promotions `int` -> `long`, `float` -> `double`, and `date` -> `timestamp` /
+    /// `timestamp_ns`, so a value written under the narrower type compares against
+    /// a predicate bound to the wider one. Returns `Ok(None)` when `target` does not
+    /// widen this value (decimals already compare across precisions), so callers
+    /// can avoid a copy.
+    ///
+    /// Matches on `(self.data_type(), target)` so an `int` literal is not treated as
+    /// a date.
+    ///
+    /// # Errors
+    /// Returns `DataInvalid` if a date is outside the range of the target timestamp
+    /// type.
+    pub(crate) fn promote_to(&self, target: &PrimitiveType) -> Result<Option<Datum>> {
+        match (self.data_type(), target) {
+            (PrimitiveType::Int, PrimitiveType::Long)
+            | (PrimitiveType::Float, PrimitiveType::Double) => Ok(self
+                .literal
+                .widen()
+                .map(|literal| Datum::new(target.clone(), literal))),
+            (PrimitiveType::Date, PrimitiveType::Timestamp) => {
+                let PrimitiveLiteral::Int(days) = self.literal else {
+                    return Ok(None);
+                };
+                Ok(Some(Datum::new(
+                    target.clone(),
+                    date_to_timestamp(days, MICROSECONDS_PER_DAY, target)?,
+                )))
+            }
+            (PrimitiveType::Date, PrimitiveType::TimestampNs) => {
+                let PrimitiveLiteral::Int(days) = self.literal else {
+                    return Ok(None);
+                };
+                Ok(Some(Datum::new(
+                    target.clone(),
+                    date_to_timestamp(days, NANOSECONDS_PER_DAY, target)?,
+                )))
+            }
+            _ => Ok(None),
+        }
     }
 
     /// Returns true if the Literal represents a primitive type