iceberg/spec/

transform.rs

// 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.

//! Transforms in iceberg.

use std::cmp::Ordering;
use std::fmt::{Display, Formatter};
use std::str::FromStr;

use chrono::Datelike;
use fnv::FnvHashSet;
use serde::{Deserialize, Deserializer, Serialize, Serializer};

use super::values::decimal_utils::decimal_from_i128_with_scale;
use super::values::temporal::date;
use super::{Datum, PrimitiveLiteral};
use crate::ErrorKind;
use crate::error::{Error, Result, invalid_data};
use crate::expr::{
    BinaryExpression, BoundPredicate, BoundReference, Predicate, PredicateOperator, Reference,
    SetExpression, UnaryExpression,
};
use crate::spec::Literal;
use crate::spec::datatypes::{PrimitiveType, Type};
use crate::transform::{BoxedTransformFunction, create_transform_function};

/// The year the Unix epoch falls in, which every temporal ordinal counts from.
const UNIX_EPOCH_YEAR: i32 = 1970;

/// Transform is used to transform predicates to partition predicates,
/// in addition to transforming data values.
///
/// Deriving partition predicates from column predicates on the table data
/// is used to separate the logical queries from physical storage: the
/// partitioning can change and the correct partition filters are always
/// derived from column predicates.
///
/// This simplifies queries because users don’t have to supply both logical
/// predicates and partition predicates.
///
/// All transforms must return `null` for a `null` input value.
#[derive(Debug, PartialEq, Eq, Clone, Copy, Hash)]
pub enum Transform {
    /// Source value, unmodified
    ///
    /// - Source type could be any type.
    /// - Return type is the same with source type.
    Identity,
    /// Hash of value, mod `N`.
    ///
    /// Bucket partition transforms use a 32-bit hash of the source value.
    /// The 32-bit hash implementation is the 32-bit Murmur3 hash, x86
    /// variant, seeded with 0.
    ///
    /// Transforms are parameterized by a number of buckets, N. The hash mod
    /// N must produce a positive value by first discarding the sign bit of
    /// the hash value. In pseudo-code, the function is:
    ///
    /// ```text
    /// def bucket_N(x) = (murmur3_x86_32_hash(x) & Integer.MAX_VALUE) % N
    /// ```
    ///
    /// - Source type could be `int`, `long`, `decimal`, `date`, `time`,
    ///   `timestamp`, `timestamptz`, `string`, `uuid`, `fixed`, `binary`.
    /// - Return type is `int`.
    Bucket(u32),
    /// Value truncated to width `W`
    ///
    /// For `int`:
    ///
    /// - `v - (v % W)` remainders must be positive
    /// - example: W=10: 1 ￫ 0, -1 ￫ -10
    /// - note: The remainder, v % W, must be positive.
    ///
    /// For `long`:
    ///
    /// - `v - (v % W)` remainders must be positive
    /// - example: W=10: 1 ￫ 0, -1 ￫ -10
    /// - note: The remainder, v % W, must be positive.
    ///
    /// For `decimal`:
    ///
    /// - `scaled_W = decimal(W, scale(v)) v - (v % scaled_W)`
    /// - example: W=50, s=2: 10.65 ￫ 10.50
    ///
    /// For `string`:
    ///
    /// - Substring of length L: `v.substring(0, L)`
    /// - example: L=3: iceberg ￫ ice
    /// - note: Strings are truncated to a valid UTF-8 string with no more
    ///   than L code points.
    ///
    /// - Source type could be `int`, `long`, `decimal`, `string`
    /// - Return type is the same with source type.
    Truncate(u32),
    /// Extract a date or timestamp year, as years from 1970
    ///
    /// - Source type could be `date`, `timestamp`, `timestamptz`
    /// - Return type is `int`
    Year,
    /// Extract a date or timestamp month, as months from 1970-01-01
    ///
    /// - Source type could be `date`, `timestamp`, `timestamptz`
    /// - Return type is `int`
    Month,
    /// Extract a date or timestamp day, as days from 1970-01-01
    ///
    /// - Source type could be `date`, `timestamp`, `timestamptz`
    /// - Return type is `int`
    Day,
    /// Extract a timestamp hour, as hours from 1970-01-01 00:00:00
    ///
    /// - Source type could be `timestamp`, `timestamptz`
    /// - Return type is `int`
    Hour,
    /// Always produces `null`
    ///
    /// The void transform may be used to replace the transform in an
    /// existing partition field so that the field is effectively dropped in
    /// v1 tables.
    ///
    /// - Source type could be any type..
    /// - Return type is Source type.
    Void,
    /// Used to represent some customized transform that can't be recognized or supported now.
    Unknown,
}

impl Transform {
    /// Returns a human-readable String representation of a transformed value.
    ///
    /// The temporal transforms store an ordinal count since the Unix epoch, and
    /// this method renders that count as a date so that partition paths and
    /// snapshot summary keys match the Java reference implementation:
    ///
    /// | Transform | Format          | Example         |
    /// |-----------|-----------------|-----------------|
    /// | `Year`    | `yyyy`          | `2017`          |
    /// | `Month`   | `yyyy-MM`       | `2017-06`       |
    /// | `Day`     | `yyyy-MM-dd`    | `2017-06-15`    |
    /// | `Hour`    | `yyyy-MM-dd-HH` | `2017-06-15-16` |
    ///
    /// `Void` renders as `null`, as does an absent value for any transform.
    ///
    /// # Example
    ///
    /// ```
    /// use iceberg::spec::{Literal, PrimitiveType, Transform, Type};
    ///
    /// let int = Type::Primitive(PrimitiveType::Int);
    /// let date = Type::Primitive(PrimitiveType::Date);
    ///
    /// // A stored value carries no logical type of its own. For transforms that do
    /// // not format it themselves, the declared field type decides how it renders.
    /// let stored = Literal::int(17332);
    /// assert_eq!(
    ///     Transform::Identity.to_human_string(&int, Some(&stored)),
    ///     "17332"
    /// );
    /// assert_eq!(
    ///     Transform::Identity.to_human_string(&date, Some(&stored)),
    ///     "2017-06-15"
    /// );
    ///
    /// // The temporal transforms format their own ordinal and ignore the declared
    /// // type. All four ordinals below are the same instant, 2017-06-15T16:00:00Z,
    /// // counted at four granularities.
    /// assert_eq!(
    ///     Transform::Year.to_human_string(&int, Some(&Literal::int(47))),
    ///     "2017"
    /// );
    /// assert_eq!(
    ///     Transform::Month.to_human_string(&int, Some(&Literal::int(569))),
    ///     "2017-06"
    /// );
    /// assert_eq!(
    ///     Transform::Day.to_human_string(&int, Some(&Literal::int(17332))),
    ///     "2017-06-15"
    /// );
    /// assert_eq!(
    ///     Transform::Hour.to_human_string(&int, Some(&Literal::int(415984))),
    ///     "2017-06-15-16"
    /// );
    ///
    /// // `Void` and an absent value render as `null`.
    /// assert_eq!(
    ///     Transform::Void.to_human_string(&int, Some(&Literal::int(47))),
    ///     "null"
    /// );
    /// assert_eq!(Transform::Year.to_human_string(&int, None), "null");
    /// ```
    pub fn to_human_string(&self, field_type: &Type, value: Option<&Literal>) -> String {
        let Some(value) = value.and_then(Literal::as_primitive_literal) else {
            return "null".to_string();
        };

        match (*self, value) {
            (Self::Void, _) => "null".to_string(),
            // The temporal transforms store an ordinal count since the Unix epoch,
            // which the datum arm below cannot render: it would show the raw count
            // for `Year`, `Month` and `Hour`, and would leave `Day` dependent on the
            // field type happening to be `date`. Java overrides `toHumanString` on
            // each of these four transforms and ignores the declared type, so do the
            // same. Any other literal falls through to the datum.
            (Self::Year, PrimitiveLiteral::Int(ordinal)) => Self::human_year(ordinal),
            (Self::Month, PrimitiveLiteral::Int(ordinal)) => Self::human_month(ordinal),
            (Self::Day, PrimitiveLiteral::Int(ordinal)) => Self::human_day(ordinal),
            (Self::Hour, PrimitiveLiteral::Int(ordinal)) => Self::human_hour(ordinal),
            (_, value) => {
                let field_type = field_type.as_primitive_type().unwrap();
                Datum::new(field_type.clone(), value).to_human_string()
            }
        }
    }

    /// Formats a year ordinal, the number of years since 1970, as `yyyy`.
    ///
    /// Mirrors the output of `TransformUtil.humanYear`.
    fn human_year(year_ordinal: i32) -> String {
        format!("{:04}", UNIX_EPOCH_YEAR + year_ordinal)
    }

    /// Formats a month ordinal, the number of months since 1970-01, as `yyyy-MM`.
    ///
    /// Mirrors the output of `TransformUtil.humanMonth`, which divides with
    /// `Math.floorDiv` and `Math.floorMod` rather than `/` and `%`. Truncating
    /// division rounds toward zero, which is the wrong direction before 1970:
    /// ordinal -1 is 1969-12, but truncating yields 1970-01. `div_euclid` and
    /// `rem_euclid` round toward negative infinity and so agree with Java.
    fn human_month(month_ordinal: i32) -> String {
        format!(
            "{:04}-{:02}",
            UNIX_EPOCH_YEAR + month_ordinal.div_euclid(12),
            1 + month_ordinal.rem_euclid(12)
        )
    }

    /// Formats a day ordinal, the number of days since 1970-01-01, as `yyyy-MM-dd`.
    ///
    /// Mirrors the output of `TransformUtil.humanDay`. Like the Java `Days`
    /// transform, whose signature is `toHumanString(Type alwaysDate, Integer
    /// value)`, this ignores the declared field type rather than relying on it being
    /// `date`.
    fn human_day(day_ordinal: i32) -> String {
        let date = date::days_to_date(day_ordinal);
        format!("{:04}-{:02}-{:02}", date.year(), date.month(), date.day())
    }

    /// Formats an hour ordinal, the number of hours since 1970-01-01T00:00:00Z,
    /// as `yyyy-MM-dd-HH`.
    ///
    /// Mirrors the output of `TransformUtil.humanHour`. `div_euclid` and `rem_euclid`
    /// split the ordinal into whole days and the hour within the day so that hours
    /// before 1970 round the way they do in Java.
    fn human_hour(hour_ordinal: i32) -> String {
        format!(
            "{}-{:02}",
            Self::human_day(hour_ordinal.div_euclid(24)),
            hour_ordinal.rem_euclid(24)
        )
    }

    /// Get the return type of transform given the input type.
    /// Returns `None` if it can't be transformed.
    pub fn result_type(&self, input_type: &Type) -> Result<Type> {
        match self {
            Transform::Identity => {
                if matches!(input_type, Type::Primitive(_)) {
                    Ok(input_type.clone())
                } else {
                    Err(invalid_data!(
                        "{input_type} is not a valid input type of identity transform",
                    ))
                }
            }
            Transform::Void => Ok(input_type.clone()),
            Transform::Unknown => Ok(Type::Primitive(PrimitiveType::String)),
            Transform::Bucket(_) => {
                if let Type::Primitive(p) = input_type {
                    match p {
                        PrimitiveType::Int
                        | PrimitiveType::Long
                        | PrimitiveType::Decimal { .. }
                        | PrimitiveType::Date
                        | PrimitiveType::Time
                        | PrimitiveType::Timestamp
                        | PrimitiveType::Timestamptz
                        | PrimitiveType::TimestampNs
                        | PrimitiveType::TimestamptzNs
                        | PrimitiveType::String
                        | PrimitiveType::Uuid
                        | PrimitiveType::Fixed(_)
                        | PrimitiveType::Binary => Ok(Type::Primitive(PrimitiveType::Int)),
                        _ => Err(invalid_data!(
                            "{input_type} is not a valid input type of bucket transform",
                        )),
                    }
                } else {
                    Err(invalid_data!(
                        "{input_type} is not a valid input type of bucket transform",
                    ))
                }
            }
            Transform::Truncate(_) => {
                if let Type::Primitive(p) = input_type {
                    match p {
                        PrimitiveType::Int
                        | PrimitiveType::Long
                        | PrimitiveType::String
                        | PrimitiveType::Binary
                        | PrimitiveType::Decimal { .. } => Ok(input_type.clone()),
                        _ => Err(invalid_data!(
                            "{input_type} is not a valid input type of truncate transform",
                        )),
                    }
                } else {
                    Err(invalid_data!(
                        "{input_type} is not a valid input type of truncate transform",
                    ))
                }
            }
            Transform::Year | Transform::Month => {
                if let Type::Primitive(p) = input_type {
                    match p {
                        PrimitiveType::Timestamp
                        | PrimitiveType::Timestamptz
                        | PrimitiveType::TimestampNs
                        | PrimitiveType::TimestamptzNs
                        | PrimitiveType::Date => Ok(Type::Primitive(PrimitiveType::Int)),
                        _ => Err(invalid_data!(
                            "{input_type} is not a valid input type of {self} transform",
                        )),
                    }
                } else {
                    Err(invalid_data!(
                        "{input_type} is not a valid input type of {self} transform",
                    ))
                }
            }
            Transform::Day => {
                if let Type::Primitive(p) = input_type {
                    match p {
                        PrimitiveType::Timestamp
                        | PrimitiveType::Timestamptz
                        | PrimitiveType::TimestampNs
                        | PrimitiveType::TimestamptzNs
                        | PrimitiveType::Date => Ok(Type::Primitive(PrimitiveType::Date)),
                        _ => Err(invalid_data!(
                            "{input_type} is not a valid input type of {self} transform",
                        )),
                    }
                } else {
                    Err(invalid_data!(
                        "{input_type} is not a valid input type of {self} transform",
                    ))
                }
            }
            Transform::Hour => {
                if let Type::Primitive(p) = input_type {
                    match p {
                        PrimitiveType::Timestamp
                        | PrimitiveType::Timestamptz
                        | PrimitiveType::TimestampNs
                        | PrimitiveType::TimestamptzNs => Ok(Type::Primitive(PrimitiveType::Int)),
                        _ => Err(invalid_data!(
                            "{input_type} is not a valid input type of {self} transform",
                        )),
                    }
                } else {
                    Err(invalid_data!(
                        "{input_type} is not a valid input type of {self} transform",
                    ))
                }
            }
        }
    }

    /// Whether the transform preserves the order of values.
    pub fn preserves_order(&self) -> bool {
        !matches!(
            self,
            Transform::Void | Transform::Bucket(_) | Transform::Unknown
        )
    }

    /// Return the unique transform name to check if similar transforms for the same source field
    /// are added multiple times in partition spec builder.
    pub fn dedup_name(&self) -> String {
        match self {
            Transform::Year | Transform::Month | Transform::Day | Transform::Hour => {
                "time".to_string()
            }
            _ => format!("{self}"),
        }
    }

    /// Whether ordering by this transform's result satisfies the ordering of another transform's
    /// result.
    ///
    /// For example, sorting by day(ts) will produce an ordering that is also by month(ts) or
    ///  year(ts). However, sorting by day(ts) will not satisfy the order of hour(ts) or identity(ts).
    pub fn satisfies_order_of(&self, other: &Self) -> bool {
        match self {
            Transform::Identity => other.preserves_order(),
            Transform::Hour => matches!(
                other,
                Transform::Hour | Transform::Day | Transform::Month | Transform::Year
            ),
            Transform::Day => matches!(other, Transform::Day | Transform::Month | Transform::Year),
            Transform::Month => matches!(other, Transform::Month | Transform::Year),
            _ => self == other,
        }
    }

    /// Strictly projects a given predicate according to the transformation
    /// specified by the `Transform` instance.
    ///
    /// This method ensures that the projected predicate is strictly aligned
    /// with the transformation logic, providing a more precise filtering
    /// mechanism for transformed data.
    ///
    /// # Example
    /// Suppose, we have row filter `a = 10`, and a partition spec
    /// `bucket(a, 37) as bs`, if one row matches `a = 10`, then its partition
    /// value should match `bucket(10, 37) as bs`, and we project `a = 10` to
    /// `bs = bucket(10, 37)`
    pub fn strict_project(
        &self,
        name: &str,
        predicate: &BoundPredicate,
    ) -> Result<Option<Predicate>> {
        let func = create_transform_function(self)?;

        match self {
            Transform::Identity => match predicate {
                BoundPredicate::Unary(expr) => Self::project_unary(expr.op(), name),
                BoundPredicate::Binary(expr) => Ok(Some(Predicate::Binary(BinaryExpression::new(
                    expr.op(),
                    Reference::new(name),
                    expr.literal().to_owned(),
                )))),
                BoundPredicate::Set(expr) => Ok(Some(Predicate::Set(SetExpression::new(
                    expr.op(),
                    Reference::new(name),
                    expr.literals().to_owned(),
                )))),
                _ => Ok(None),
            },
            Transform::Bucket(_) => match predicate {
                BoundPredicate::Unary(expr) => Self::project_unary(expr.op(), name),
                BoundPredicate::Binary(expr) => {
                    self.project_binary_expr(name, PredicateOperator::NotEq, expr, &func)
                }
                BoundPredicate::Set(expr) => {
                    self.project_set_expr(expr, PredicateOperator::NotIn, name, &func)
                }
                _ => Ok(None),
            },
            Transform::Truncate(width) => match predicate {
                BoundPredicate::Unary(expr) => Self::project_unary(expr.op(), name),
                BoundPredicate::Binary(expr) => {
                    if matches!(
                        expr.term().field().field_type.as_primitive_type(),
                        Some(&PrimitiveType::Int)
                            | Some(&PrimitiveType::Long)
                            | Some(&PrimitiveType::Decimal { .. })
                    ) {
                        self.truncate_number_strict(name, expr, &func)
                    } else if expr.op() == PredicateOperator::StartsWith {
                        let len = match expr.literal().literal() {
                            PrimitiveLiteral::String(s) => s.len(),
                            PrimitiveLiteral::Binary(b) => b.len(),
                            _ => {
                                return Err(invalid_data!(
                                    "Expected a string or binary literal, got: {:?}",
                                    expr.literal()
                                ));
                            }
                        };
                        match len.cmp(&(*width as usize)) {
                            Ordering::Less => Ok(Some(Predicate::Binary(BinaryExpression::new(
                                PredicateOperator::StartsWith,
                                Reference::new(name),
                                expr.literal().to_owned(),
                            )))),
                            Ordering::Equal => Ok(Some(Predicate::Binary(BinaryExpression::new(
                                PredicateOperator::Eq,
                                Reference::new(name),
                                expr.literal().to_owned(),
                            )))),
                            Ordering::Greater => Ok(None),
                        }
                    } else if expr.op() == PredicateOperator::NotStartsWith {
                        let len = match expr.literal().literal() {
                            PrimitiveLiteral::String(s) => s.len(),
                            PrimitiveLiteral::Binary(b) => b.len(),
                            _ => {
                                return Err(invalid_data!(
                                    "Expected a string or binary literal, got: {:?}",
                                    expr.literal()
                                ));
                            }
                        };
                        match len.cmp(&(*width as usize)) {
                            Ordering::Less => Ok(Some(Predicate::Binary(BinaryExpression::new(
                                PredicateOperator::NotStartsWith,
                                Reference::new(name),
                                expr.literal().to_owned(),
                            )))),
                            Ordering::Equal => Ok(Some(Predicate::Binary(BinaryExpression::new(
                                PredicateOperator::NotEq,
                                Reference::new(name),
                                expr.literal().to_owned(),
                            )))),
                            Ordering::Greater => {
                                Ok(Some(Predicate::Binary(BinaryExpression::new(
                                    expr.op(),
                                    Reference::new(name),
                                    func.transform_literal_result(expr.literal())?,
                                ))))
                            }
                        }
                    } else {
                        self.truncate_array_strict(name, expr, &func)
                    }
                }
                BoundPredicate::Set(expr) => {
                    self.project_set_expr(expr, PredicateOperator::NotIn, name, &func)
                }
                _ => Ok(None),
            },
            Transform::Year | Transform::Month | Transform::Day | Transform::Hour => {
                match predicate {
                    BoundPredicate::Unary(expr) => Self::project_unary(expr.op(), name),
                    BoundPredicate::Binary(expr) => self.truncate_number_strict(name, expr, &func),
                    BoundPredicate::Set(expr) => {
                        self.project_set_expr(expr, PredicateOperator::NotIn, name, &func)
                    }
                    _ => Ok(None),
                }
            }
            _ => Ok(None),
        }
    }

    /// Projects a given predicate according to the transformation
    /// specified by the `Transform` instance.
    ///
    /// This allows predicates to be effectively applied to data
    /// that has undergone transformation, enabling efficient querying
    /// and filtering based on the original, untransformed data.
    ///
    /// # Example
    /// Suppose, we have row filter `a = 10`, and a partition spec
    /// `bucket(a, 37) as bs`, if one row matches `a = 10`, then its partition
    /// value should match `bucket(10, 37) as bs`, and we project `a = 10` to
    /// `bs = bucket(10, 37)`
    pub fn project(&self, name: &str, predicate: &BoundPredicate) -> Result<Option<Predicate>> {
        let func = create_transform_function(self)?;

        match self {
            Transform::Identity => match predicate {
                BoundPredicate::Unary(expr) => Self::project_unary(expr.op(), name),
                BoundPredicate::Binary(expr) => Ok(Some(Predicate::Binary(BinaryExpression::new(
                    expr.op(),
                    Reference::new(name),
                    expr.literal().to_owned(),
                )))),
                BoundPredicate::Set(expr) => Ok(Some(Predicate::Set(SetExpression::new(
                    expr.op(),
                    Reference::new(name),
                    expr.literals().to_owned(),
                )))),
                _ => Ok(None),
            },
            Transform::Bucket(_) => match predicate {
                BoundPredicate::Unary(expr) => Self::project_unary(expr.op(), name),
                BoundPredicate::Binary(expr) => {
                    self.project_binary_expr(name, PredicateOperator::Eq, expr, &func)
                }
                BoundPredicate::Set(expr) => {
                    self.project_set_expr(expr, PredicateOperator::In, name, &func)
                }
                _ => Ok(None),
            },
            Transform::Truncate(width) => match predicate {
                BoundPredicate::Unary(expr) => Self::project_unary(expr.op(), name),
                BoundPredicate::Binary(expr) => {
                    self.project_binary_with_adjusted_boundary(name, expr, &func, Some(*width))
                }
                BoundPredicate::Set(expr) => {
                    self.project_set_expr(expr, PredicateOperator::In, name, &func)
                }
                _ => Ok(None),
            },
            Transform::Year | Transform::Month | Transform::Day | Transform::Hour => {
                match predicate {
                    BoundPredicate::Unary(expr) => Self::project_unary(expr.op(), name),
                    BoundPredicate::Binary(expr) => {
                        self.project_binary_with_adjusted_boundary(name, expr, &func, None)
                    }
                    BoundPredicate::Set(expr) => {
                        self.project_set_expr(expr, PredicateOperator::In, name, &func)
                    }
                    _ => Ok(None),
                }
            }
            _ => Ok(None),
        }
    }

    /// Check if `Transform` is applicable on datum's `PrimitiveType`
    fn can_transform(&self, datum: &Datum) -> bool {
        let input_type = datum.data_type().clone();
        self.result_type(&Type::Primitive(input_type)).is_ok()
    }

    /// Creates a unary predicate from a given operator and a reference name.
    fn project_unary(op: PredicateOperator, name: &str) -> Result<Option<Predicate>> {
        Ok(Some(Predicate::Unary(UnaryExpression::new(
            op,
            Reference::new(name),
        ))))
    }

    /// Attempts to create a binary predicate based on a binary expression,
    /// if applicable.
    ///
    /// This method evaluates a given binary expression and, if the operation
    /// is the given operator and the literal can be transformed, constructs a
    /// `Predicate::Binary`variant representing the binary operation.
    fn project_binary_expr(
        &self,
        name: &str,
        op: PredicateOperator,
        expr: &BinaryExpression<BoundReference>,
        func: &BoxedTransformFunction,
    ) -> Result<Option<Predicate>> {
        if expr.op() != op || !self.can_transform(expr.literal()) {
            return Ok(None);
        }

        Ok(Some(Predicate::Binary(BinaryExpression::new(
            expr.op(),
            Reference::new(name),
            func.transform_literal_result(expr.literal())?,
        ))))
    }

    /// Projects a binary expression to a predicate with an adjusted boundary.
    ///
    /// Checks if the literal within the given binary expression is
    /// transformable. If transformable, it proceeds to potentially adjust
    /// the boundary of the expression based on the comparison operator (`op`).
    /// The potential adjustments involve incrementing or decrementing the
    /// literal value and changing the `PredicateOperator` itself to its
    /// inclusive variant.
    fn project_binary_with_adjusted_boundary(
        &self,
        name: &str,
        expr: &BinaryExpression<BoundReference>,
        func: &BoxedTransformFunction,
        width: Option<u32>,
    ) -> Result<Option<Predicate>> {
        if !self.can_transform(expr.literal()) {
            return Ok(None);
        }

        let op = &expr.op();
        let datum = &expr.literal();

        if let Some(boundary) = Self::adjust_boundary(op, datum)? {
            let transformed_projection = func.transform_literal_result(&boundary)?;

            let adjusted_projection =
                self.adjust_time_projection(op, datum, &transformed_projection);

            let adjusted_operator = Self::adjust_operator(op, datum, width);

            if let Some(op) = adjusted_operator {
                let predicate = match adjusted_projection {
                    None => Predicate::Binary(BinaryExpression::new(
                        op,
                        Reference::new(name),
                        transformed_projection,
                    )),
                    Some(AdjustedProjection::Single(d)) => {
                        Predicate::Binary(BinaryExpression::new(op, Reference::new(name), d))
                    }
                    Some(AdjustedProjection::Set(d)) => Predicate::Set(SetExpression::new(
                        PredicateOperator::In,
                        Reference::new(name),
                        d,
                    )),
                };
                return Ok(Some(predicate));
            }
        };

        Ok(None)
    }

    /// Projects a set expression to a predicate,
    /// applying a transformation to each literal in the set.
    fn project_set_expr(
        &self,
        expr: &SetExpression<BoundReference>,
        op: PredicateOperator,
        name: &str,
        func: &BoxedTransformFunction,
    ) -> Result<Option<Predicate>> {
        if expr.op() != op || expr.literals().iter().any(|d| !self.can_transform(d)) {
            return Ok(None);
        }

        let mut new_set = FnvHashSet::default();

        for lit in expr.literals() {
            let datum = func.transform_literal_result(lit)?;

            if let Some(AdjustedProjection::Single(d)) =
                self.adjust_time_projection(&op, lit, &datum)
            {
                new_set.insert(d);
            };

            new_set.insert(datum);
        }

        Ok(Some(Predicate::Set(SetExpression::new(
            expr.op(),
            Reference::new(name),
            new_set,
        ))))
    }

    /// Adjusts the boundary value for comparison operations
    /// based on the specified `PredicateOperator` and `Datum`.
    ///
    /// This function modifies the boundary value for certain comparison
    /// operators (`LessThan`, `GreaterThan`) by incrementing or decrementing
    /// the literal value within the given `Datum`. For operators that do not
    /// imply a boundary shift (`Eq`, `LessThanOrEq`, `GreaterThanOrEq`,
    /// `StartsWith`, `NotStartsWith`), the original datum is returned
    /// unmodified.
    fn adjust_boundary(op: &PredicateOperator, datum: &Datum) -> Result<Option<Datum>> {
        let adjusted_boundary = match op {
            PredicateOperator::LessThan => match (datum.data_type(), datum.literal()) {
                (PrimitiveType::Int, PrimitiveLiteral::Int(v)) => Some(Datum::int(v - 1)),
                (PrimitiveType::Long, PrimitiveLiteral::Long(v)) => Some(Datum::long(v - 1)),
                (PrimitiveType::Decimal { .. }, PrimitiveLiteral::Int128(v)) => {
                    Some(Datum::decimal(decimal_from_i128_with_scale(v - 1, 0))?)
                }
                (PrimitiveType::Date, PrimitiveLiteral::Int(v)) => Some(Datum::date(v - 1)),
                (PrimitiveType::Timestamp, PrimitiveLiteral::Long(v)) => {
                    Some(Datum::timestamp_micros(v - 1))
                }
                (PrimitiveType::Timestamptz, PrimitiveLiteral::Long(v)) => {
                    Some(Datum::timestamptz_micros(v - 1))
                }
                (PrimitiveType::TimestampNs, PrimitiveLiteral::Long(v)) => {
                    Some(Datum::timestamp_nanos(v - 1))
                }
                (PrimitiveType::TimestamptzNs, PrimitiveLiteral::Long(v)) => {
                    Some(Datum::timestamptz_nanos(v - 1))
                }
                _ => Some(datum.to_owned()),
            },
            PredicateOperator::GreaterThan => match (datum.data_type(), datum.literal()) {
                (PrimitiveType::Int, PrimitiveLiteral::Int(v)) => Some(Datum::int(v + 1)),
                (PrimitiveType::Long, PrimitiveLiteral::Long(v)) => Some(Datum::long(v + 1)),
                (PrimitiveType::Decimal { .. }, PrimitiveLiteral::Int128(v)) => {
                    Some(Datum::decimal(decimal_from_i128_with_scale(v + 1, 0))?)
                }
                (PrimitiveType::Date, PrimitiveLiteral::Int(v)) => Some(Datum::date(v + 1)),
                (PrimitiveType::Timestamp, PrimitiveLiteral::Long(v)) => {
                    Some(Datum::timestamp_micros(v + 1))
                }
                (PrimitiveType::Timestamptz, PrimitiveLiteral::Long(v)) => {
                    Some(Datum::timestamptz_micros(v + 1))
                }
                (PrimitiveType::TimestampNs, PrimitiveLiteral::Long(v)) => {
                    Some(Datum::timestamp_nanos(v + 1))
                }
                (PrimitiveType::TimestamptzNs, PrimitiveLiteral::Long(v)) => {
                    Some(Datum::timestamptz_nanos(v + 1))
                }
                _ => Some(datum.to_owned()),
            },
            PredicateOperator::Eq
            | PredicateOperator::LessThanOrEq
            | PredicateOperator::GreaterThanOrEq
            | PredicateOperator::StartsWith
            | PredicateOperator::NotStartsWith => Some(datum.to_owned()),
            _ => None,
        };

        Ok(adjusted_boundary)
    }

    /// Adjusts the comparison operator based on the specified datum and an
    /// optional width constraint.
    ///
    /// This function modifies the comparison operator for `LessThan` and
    /// `GreaterThan` cases to their inclusive counterparts (`LessThanOrEq`,
    /// `GreaterThanOrEq`) unconditionally. For `StartsWith` and
    /// `NotStartsWith` operators acting on string literals, the operator may
    /// be adjusted to `Eq` or `NotEq` if the string length matches the
    /// specified width, indicating a precise match rather than a prefix
    /// condition.
    fn adjust_operator(
        op: &PredicateOperator,
        datum: &Datum,
        width: Option<u32>,
    ) -> Option<PredicateOperator> {
        match op {
            PredicateOperator::LessThan => Some(PredicateOperator::LessThanOrEq),
            PredicateOperator::GreaterThan => Some(PredicateOperator::GreaterThanOrEq),
            PredicateOperator::StartsWith => match datum.literal() {
                PrimitiveLiteral::String(s) => {
                    if let Some(w) = width
                        && s.len() == w as usize
                    {
                        return Some(PredicateOperator::Eq);
                    };
                    Some(*op)
                }
                _ => Some(*op),
            },
            PredicateOperator::NotStartsWith => match datum.literal() {
                PrimitiveLiteral::String(s) => {
                    if let Some(w) = width {
                        let w = w as usize;

                        if s.len() == w {
                            return Some(PredicateOperator::NotEq);
                        }

                        if s.len() < w {
                            return Some(*op);
                        }

                        return None;
                    };
                    Some(*op)
                }
                _ => Some(*op),
            },
            _ => Some(*op),
        }
    }

    /// Adjust projection for temporal transforms, align with Java
    /// implementation: https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/transforms/ProjectionUtil.java#L275
    fn adjust_time_projection(
        &self,
        op: &PredicateOperator,
        original: &Datum,
        transformed: &Datum,
    ) -> Option<AdjustedProjection> {
        let should_adjust = match self {
            Transform::Day => matches!(
                original.data_type(),
                PrimitiveType::Timestamp
                    | PrimitiveType::Timestamptz
                    | PrimitiveType::TimestampNs
                    | PrimitiveType::TimestamptzNs
            ),
            Transform::Year | Transform::Month => true,
            _ => false,
        };

        if should_adjust && let &PrimitiveLiteral::Int(v) = transformed.literal() {
            match op {
                PredicateOperator::LessThan
                | PredicateOperator::LessThanOrEq
                | PredicateOperator::In => {
                    if v < 0 {
                        // # TODO
                        // An ugly hack to fix. Refine the increment and decrement logic later.
                        match self {
                            Transform::Day => {
                                return Some(AdjustedProjection::Single(Datum::date(v + 1)));
                            }
                            _ => {
                                return Some(AdjustedProjection::Single(Datum::int(v + 1)));
                            }
                        }
                    };
                }
                PredicateOperator::Eq => {
                    if v < 0 {
                        let new_set = FnvHashSet::from_iter(vec![
                            transformed.to_owned(),
                            // # TODO
                            // An ugly hack to fix. Refine the increment and decrement logic later.
                            {
                                match self {
                                    Transform::Day => Datum::date(v + 1),
                                    _ => Datum::int(v + 1),
                                }
                            },
                        ]);
                        return Some(AdjustedProjection::Set(new_set));
                    }
                }
                _ => {
                    return None;
                }
            }
        };
        None
    }

    // Increment for Int, Long, Decimal, Date, Timestamp
    // Ignore other types
    #[inline]
    fn try_increment_number(datum: &Datum) -> Result<Datum> {
        match (datum.data_type(), datum.literal()) {
            (PrimitiveType::Int, PrimitiveLiteral::Int(v)) => Ok(Datum::int(v + 1)),
            (PrimitiveType::Long, PrimitiveLiteral::Long(v)) => Ok(Datum::long(v + 1)),
            (PrimitiveType::Decimal { .. }, PrimitiveLiteral::Int128(v)) => {
                Datum::decimal(decimal_from_i128_with_scale(v + 1, 0))
            }
            (PrimitiveType::Date, PrimitiveLiteral::Int(v)) => Ok(Datum::date(v + 1)),
            (PrimitiveType::Timestamp, PrimitiveLiteral::Long(v)) => {
                Ok(Datum::timestamp_micros(v + 1))
            }
            (PrimitiveType::TimestampNs, PrimitiveLiteral::Long(v)) => {
                Ok(Datum::timestamp_nanos(v + 1))
            }
            (PrimitiveType::Timestamptz, PrimitiveLiteral::Long(v)) => {
                Ok(Datum::timestamptz_micros(v + 1))
            }
            (PrimitiveType::TimestamptzNs, PrimitiveLiteral::Long(v)) => {
                Ok(Datum::timestamptz_nanos(v + 1))
            }
            (PrimitiveType::Int, _)
            | (PrimitiveType::Long, _)
            | (PrimitiveType::Decimal { .. }, _)
            | (PrimitiveType::Date, _)
            | (PrimitiveType::Timestamp, _) => Err(Error::new(
                ErrorKind::Unexpected,
                format!(
                    "Unsupported literal increment for type: {:?}",
                    datum.data_type()
                ),
            )),
            _ => Ok(datum.to_owned()),
        }
    }

    // Decrement for Int, Long, Decimal, Date, Timestamp
    // Ignore other types
    #[inline]
    fn try_decrement_number(datum: &Datum) -> Result<Datum> {
        match (datum.data_type(), datum.literal()) {
            (PrimitiveType::Int, PrimitiveLiteral::Int(v)) => Ok(Datum::int(v - 1)),
            (PrimitiveType::Long, PrimitiveLiteral::Long(v)) => Ok(Datum::long(v - 1)),
            (PrimitiveType::Decimal { .. }, PrimitiveLiteral::Int128(v)) => {
                Datum::decimal(decimal_from_i128_with_scale(v - 1, 0))
            }
            (PrimitiveType::Date, PrimitiveLiteral::Int(v)) => Ok(Datum::date(v - 1)),
            (PrimitiveType::Timestamp, PrimitiveLiteral::Long(v)) => {
                Ok(Datum::timestamp_micros(v - 1))
            }
            (PrimitiveType::TimestampNs, PrimitiveLiteral::Long(v)) => {
                Ok(Datum::timestamp_nanos(v - 1))
            }
            (PrimitiveType::Timestamptz, PrimitiveLiteral::Long(v)) => {
                Ok(Datum::timestamptz_micros(v - 1))
            }
            (PrimitiveType::TimestamptzNs, PrimitiveLiteral::Long(v)) => {
                Ok(Datum::timestamptz_nanos(v - 1))
            }
            (PrimitiveType::Int, _)
            | (PrimitiveType::Long, _)
            | (PrimitiveType::Decimal { .. }, _)
            | (PrimitiveType::Date, _)
            | (PrimitiveType::Timestamp, _) => Err(Error::new(
                ErrorKind::Unexpected,
                format!(
                    "Unsupported literal decrement for type: {:?}",
                    datum.data_type()
                ),
            )),
            _ => Ok(datum.to_owned()),
        }
    }

    fn truncate_number_strict(
        &self,
        name: &str,
        expr: &BinaryExpression<BoundReference>,
        func: &BoxedTransformFunction,
    ) -> Result<Option<Predicate>> {
        let boundary = expr.literal();

        if !matches!(
            boundary.data_type(),
            &PrimitiveType::Int
                | &PrimitiveType::Long
                | &PrimitiveType::Decimal { .. }
                | &PrimitiveType::Date
                | &PrimitiveType::Timestamp
                | &PrimitiveType::Timestamptz
                | &PrimitiveType::TimestampNs
                | &PrimitiveType::TimestamptzNs
        ) {
            return Err(invalid_data!(
                "Expected a numeric literal, got: {boundary:?}"
            ));
        }

        let predicate = match expr.op() {
            PredicateOperator::LessThan => Some(Predicate::Binary(BinaryExpression::new(
                PredicateOperator::LessThan,
                Reference::new(name),
                func.transform_literal_result(boundary)?,
            ))),
            PredicateOperator::LessThanOrEq => Some(Predicate::Binary(BinaryExpression::new(
                PredicateOperator::LessThan,
                Reference::new(name),
                func.transform_literal_result(&Self::try_increment_number(boundary)?)?,
            ))),
            PredicateOperator::GreaterThan => Some(Predicate::Binary(BinaryExpression::new(
                PredicateOperator::GreaterThan,
                Reference::new(name),
                func.transform_literal_result(boundary)?,
            ))),
            PredicateOperator::GreaterThanOrEq => Some(Predicate::Binary(BinaryExpression::new(
                PredicateOperator::GreaterThan,
                Reference::new(name),
                func.transform_literal_result(&Self::try_decrement_number(boundary)?)?,
            ))),
            PredicateOperator::NotEq => Some(Predicate::Binary(BinaryExpression::new(
                PredicateOperator::NotEq,
                Reference::new(name),
                func.transform_literal_result(boundary)?,
            ))),
            _ => None,
        };

        Ok(predicate)
    }

    fn truncate_array_strict(
        &self,
        name: &str,
        expr: &BinaryExpression<BoundReference>,
        func: &BoxedTransformFunction,
    ) -> Result<Option<Predicate>> {
        let boundary = expr.literal();

        match expr.op() {
            PredicateOperator::LessThan | PredicateOperator::LessThanOrEq => {
                Ok(Some(Predicate::Binary(BinaryExpression::new(
                    PredicateOperator::LessThan,
                    Reference::new(name),
                    func.transform_literal_result(boundary)?,
                ))))
            }
            PredicateOperator::GreaterThan | PredicateOperator::GreaterThanOrEq => {
                Ok(Some(Predicate::Binary(BinaryExpression::new(
                    PredicateOperator::GreaterThan,
                    Reference::new(name),
                    func.transform_literal_result(boundary)?,
                ))))
            }
            PredicateOperator::NotEq => Ok(Some(Predicate::Binary(BinaryExpression::new(
                PredicateOperator::NotEq,
                Reference::new(name),
                func.transform_literal_result(boundary)?,
            )))),
            _ => Ok(None),
        }
    }
}

impl Display for Transform {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        match self {
            Transform::Identity => write!(f, "identity"),
            Transform::Year => write!(f, "year"),
            Transform::Month => write!(f, "month"),
            Transform::Day => write!(f, "day"),
            Transform::Hour => write!(f, "hour"),
            Transform::Void => write!(f, "void"),
            Transform::Bucket(length) => write!(f, "bucket[{length}]"),
            Transform::Truncate(width) => write!(f, "truncate[{width}]"),
            Transform::Unknown => write!(f, "unknown"),
        }
    }
}

impl FromStr for Transform {
    type Err = Error;

    fn from_str(s: &str) -> Result<Self> {
        let t = match s {
            "identity" => Transform::Identity,
            "year" => Transform::Year,
            "month" => Transform::Month,
            "day" => Transform::Day,
            "hour" => Transform::Hour,
            "void" => Transform::Void,
            "unknown" => Transform::Unknown,
            v if v.starts_with("bucket") => {
                let length = v
                    .strip_prefix("bucket")
                    .expect("transform must starts with `bucket`")
                    .trim_start_matches('[')
                    .trim_end_matches(']')
                    .parse()
                    .map_err(|err| {
                        invalid_data!("transform bucket type {v:?} is invalid").with_source(err)
                    })?;

                Transform::Bucket(length)
            }
            v if v.starts_with("truncate") => {
                let width = v
                    .strip_prefix("truncate")
                    .expect("transform must starts with `truncate`")
                    .trim_start_matches('[')
                    .trim_end_matches(']')
                    .parse()
                    .map_err(|err| {
                        invalid_data!("transform truncate type {v:?} is invalid").with_source(err)
                    })?;

                Transform::Truncate(width)
            }
            v => {
                return Err(invalid_data!("transform {v:?} is invalid"));
            }
        };

        Ok(t)
    }
}

impl Serialize for Transform {
    fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
    where S: Serializer {
        serializer.serialize_str(format!("{self}").as_str())
    }
}

impl<'de> Deserialize<'de> for Transform {
    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
    where D: Deserializer<'de> {
        let s = String::deserialize(deserializer)?;
        s.parse().map_err(<D::Error as serde::de::Error>::custom)
    }
}

/// An enum representing the result of the adjusted projection.
/// Either being a single adjusted datum or a set.
#[derive(Debug)]
enum AdjustedProjection {
    Single(Datum),
    Set(FnvHashSet<Datum>),
}

#[cfg(test)]
mod tests {
    use super::*;

    fn check_boundary(op: PredicateOperator, input: Datum, expected: Datum) {
        let result = Transform::adjust_boundary(&op, &input).unwrap().unwrap();
        assert_eq!(result, expected);
    }

    #[test]
    fn test_adjust_boundary_timestamp_types() {
        for (datum, dec, inc) in [
            (
                Datum::timestamptz_micros(1000),
                Datum::timestamptz_micros(999),
                Datum::timestamptz_micros(1001),
            ),
            (
                Datum::timestamp_nanos(5000),
                Datum::timestamp_nanos(4999),
                Datum::timestamp_nanos(5001),
            ),
            (
                Datum::timestamptz_nanos(5000),
                Datum::timestamptz_nanos(4999),
                Datum::timestamptz_nanos(5001),
            ),
        ] {
            check_boundary(PredicateOperator::LessThan, datum.clone(), dec);
            check_boundary(PredicateOperator::GreaterThan, datum.clone(), inc);
            check_boundary(
                PredicateOperator::LessThanOrEq,
                datum.clone(),
                datum.clone(),
            );
            check_boundary(PredicateOperator::GreaterThanOrEq, datum.clone(), datum);
        }
    }

    /// Renders `ordinal` through the public API with the given declared type.
    fn human(transform: Transform, primitive: PrimitiveType, ordinal: i32) -> String {
        transform.to_human_string(&Type::Primitive(primitive), Some(&Literal::int(ordinal)))
    }

    /// Renders `ordinal` for a transform whose result type is `int`.
    fn human_int(transform: Transform, ordinal: i32) -> String {
        human(transform, PrimitiveType::Int, ordinal)
    }

    #[test]
    fn test_to_human_string_year() {
        assert_eq!(human_int(Transform::Year, -1970), "0000");
        assert_eq!(human_int(Transform::Year, -1), "1969");
        assert_eq!(human_int(Transform::Year, 0), "1970");
        assert_eq!(human_int(Transform::Year, 47), "2017");
    }

    #[test]
    fn test_to_human_string_month() {
        assert_eq!(human_int(Transform::Month, -1970 * 12), "0000-01");
        assert_eq!(human_int(Transform::Month, -13), "1968-12");
        assert_eq!(human_int(Transform::Month, -12), "1969-01");
        assert_eq!(human_int(Transform::Month, -1), "1969-12");
        assert_eq!(human_int(Transform::Month, 0), "1970-01");
        assert_eq!(human_int(Transform::Month, 11), "1970-12");
        assert_eq!(human_int(Transform::Month, 12), "1971-01");
        assert_eq!(human_int(Transform::Month, 569), "2017-06");
    }

    #[test]
    fn test_to_human_string_day() {
        assert_eq!(human_int(Transform::Day, -1), "1969-12-31");
        assert_eq!(human_int(Transform::Day, 0), "1970-01-01");
        assert_eq!(human_int(Transform::Day, 31), "1970-02-01");
        assert_eq!(human_int(Transform::Day, 17332), "2017-06-15");
    }

    #[test]
    fn test_to_human_string_hour() {
        assert_eq!(human_int(Transform::Hour, -24), "1969-12-31-00");
        assert_eq!(human_int(Transform::Hour, -1), "1969-12-31-23");
        assert_eq!(human_int(Transform::Hour, 0), "1970-01-01-00");
        assert_eq!(human_int(Transform::Hour, 23), "1970-01-01-23");
        assert_eq!(human_int(Transform::Hour, 24), "1970-01-02-00");
        assert_eq!(human_int(Transform::Hour, 1000), "1970-02-11-16");
        assert_eq!(human_int(Transform::Hour, 415984), "2017-06-15-16");
    }

    /// The temporal transforms ignore the declared field type, matching the Java
    /// signatures `toHumanString(Type alwaysInt, ..)` and
    /// `toHumanString(Type alwaysDate, ..)`.
    #[test]
    fn test_to_human_string_ignores_declared_type_for_temporal_transforms() {
        assert_eq!(human(Transform::Year, PrimitiveType::Date, 47), "2017");
        assert_eq!(human(Transform::Month, PrimitiveType::Date, 569), "2017-06");
        assert_eq!(
            human(Transform::Day, PrimitiveType::Int, 17332),
            "2017-06-15"
        );
        assert_eq!(
            human(Transform::Hour, PrimitiveType::Date, 415984),
            "2017-06-15-16"
        );
    }

    /// Transforms with no temporal format keep deferring to the datum, which
    /// renders according to the declared field type.
    #[test]
    fn test_to_human_string_defers_to_datum_for_other_transforms() {
        assert_eq!(
            human(Transform::Identity, PrimitiveType::Int, 17332),
            "17332"
        );
        assert_eq!(
            human(Transform::Identity, PrimitiveType::Date, 17332),
            "2017-06-15"
        );
        assert_eq!(human(Transform::Bucket(16), PrimitiveType::Int, 5), "5");

        // A literal that is not an `int` also defers to the datum rather than being
        // reported as `null`, so the temporal arms add no second behaviour change.
        assert_eq!(
            Transform::Year.to_human_string(
                &Type::Primitive(PrimitiveType::String),
                Some(&Literal::string("unformatted"))
            ),
            "unformatted"
        );
    }

    #[test]
    fn test_to_human_string_null_cases() {
        assert_eq!(human(Transform::Void, PrimitiveType::Int, 47), "null");
        assert_eq!(human(Transform::Void, PrimitiveType::Date, 17332), "null");
        for transform in [
            Transform::Year,
            Transform::Month,
            Transform::Day,
            Transform::Hour,
            Transform::Identity,
            Transform::Void,
        ] {
            assert_eq!(
                transform.to_human_string(&Type::Primitive(PrimitiveType::Int), None),
                "null"
            );
        }
    }
}