Python: Add year, month, day, and hour transforms

python/pyiceberg/transforms.py

@@ -58,6 +58,10 @@
 VOID = "void"
 BUCKET = "bucket"
 TRUNCATE = "truncate"
+YEAR = "year"
+MONTH = "month"
+DAY = "day"
+HOUR = "hour"
 
 BUCKET_PARSER = ParseNumberFromBrackets(BUCKET)
 TRUNCATE_PARSER = ParseNumberFromBrackets(TRUNCATE)
@@ -92,6 +96,14 @@ def validate(cls, v: Any):
                 return BucketTransform(num_buckets=BUCKET_PARSER.match(v))
             elif v.startswith(TRUNCATE):
                 return TruncateTransform(width=TRUNCATE_PARSER.match(v))
+            elif v == YEAR:
+                return YearTransform()
+            elif v == MONTH:
+                return MonthTransform()
+            elif v == DAY:
+                return DayTransform()
+            elif v == HOUR:
+                return HourTransform()
             else:
                 return UnknownTransform(transform=v)
         return v
@@ -215,6 +227,243 @@ def __repr__(self) -> str:
         return f"BucketTransform(num_buckets={self._num_buckets})"
 
 
+class YearTransform(Transform[S, int]):
+    """Transforms a datetime value into a year value.
+
+    Example:
+        >>> transform = YearTransform()
+        >>> transform.transform(TimestampType())(1512151975038194)
+        47
+    """
+
+    __root__: Literal["year"] = Field(default="year")
+    _source_type: IcebergType = PrivateAttr()
+
+    def transform(self, source: IcebergType) -> Callable[[Optional[S]], Optional[int]]:
+        source_type = type(source)
+        if source_type == DateType:
+
+            def year_func(v):
+                return datetime.days_to_years(v)
+
+        elif source_type in {TimestampType, TimestamptzType}:
+
+            def year_func(v):
+                return datetime.micros_to_years(v)
+
+        else:
+            raise ValueError(f"Cannot apply year transform for type: {source}")
+
+        return lambda v: year_func(v) if v else None
+
+    def can_transform(self, source: IcebergType) -> bool:
+        return type(source) in {
+            DateType,
+            TimestampType,
+            TimestamptzType,
+        }
+
+    def result_type(self, source: IcebergType) -> IcebergType:
+        return IntegerType()
+
+    @property
+    def preserves_order(self) -> bool:
+        return True
+
+    @property
+    def time_order(self) -> int:
+        return 3
+
+    def satisfies_order_of(self, other: Transform) -> bool:
+        return self.time_order <= other.time_order if hasattr(other, "time_order") else False
+
+    def to_human_string(self, _: IcebergType, value: Optional[S]) -> str:
+        return datetime.to_human_year(value) if isinstance(value, int) else "null"
+
+    @property
+    def dedup_name(self) -> str:
+        return "time"
+
+    def __repr__(self) -> str:
+        return "YearTransform()"
+
+
+class MonthTransform(Transform[S, int]):
+    """Transforms a datetime value into a month value.
+
+    Example:
+        >>> transform = MonthTransform()
+        >>> transform.transform(DateType())(17501)
+        575
+    """
+
+    __root__: Literal["month"] = Field(default="month")
+    _source_type: IcebergType = PrivateAttr()
+
+    def transform(self, source: IcebergType) -> Callable[[Optional[S]], Optional[int]]:
+        source_type = type(source)
+        if source_type == DateType:
+
+            def month_func(v):
+                return datetime.days_to_months(v)
+
+        elif source_type in {TimestampType, TimestamptzType}:
+
+            def month_func(v):
+                return datetime.micros_to_months(v)
+
+        else:
+            raise ValueError(f"Cannot apply month transform for type: {source}")
+
+        return lambda v: month_func(v) if v else None
+
+    def can_transform(self, source: IcebergType) -> bool:
+        return type(source) in {
+            DateType,
+            TimestampType,
+            TimestamptzType,
+        }
+
+    def result_type(self, source: IcebergType) -> IcebergType:
+        return IntegerType()
+
+    @property
+    def preserves_order(self) -> bool:
+        return True
+
+    @property
+    def time_order(self) -> int:
+        return 2
+
+    def satisfies_order_of(self, other: Transform) -> bool:
+        return self.time_order <= other.time_order if hasattr(other, "time_order") else False
+
+    def to_human_string(self, _: IcebergType, value: Optional[S]) -> str:
+        return datetime.to_human_month(value) if isinstance(value, int) else "null"
+
+    @property
+    def dedup_name(self) -> str:
+        return "time"
+
+    def __repr__(self) -> str:
+        return "MonthTransform()"
+
+
+class DayTransform(Transform[S, int]):
+    """Transforms a datetime value into a day value.
+
+    Example:
+        >>> transform = MonthTransform()
+        >>> transform.transform(DateType())(17501)
+        17501
+    """
+
+    __root__: Literal["day"] = Field(default="day")
+    _source_type: IcebergType = PrivateAttr()
+
+    def transform(self, source: IcebergType) -> Callable[[Optional[S]], Optional[int]]:
+        source_type = type(source)
+        if source_type == DateType:
+
+            def day_func(v):
+                return v
+
+        elif source_type in {TimestampType, TimestamptzType}:
+
+            def day_func(v):
+                return datetime.micros_to_days(v)
+
+        else:
+            raise ValueError(f"Cannot apply day transform for type: {source}")
+
+        return lambda v: day_func(v) if v else None
+
+    def can_transform(self, source: IcebergType) -> bool:
+        return type(source) in {
+            DateType,
+            TimestampType,
+            TimestamptzType,
+        }
+
+    def result_type(self, source: IcebergType) -> IcebergType:
+        return DateType()
+
+    @property
+    def preserves_order(self) -> bool:
+        return True
+
+    @property
+    def time_order(self) -> int:
+        return 1
+
+    def satisfies_order_of(self, other: Transform) -> bool:
+        return self.time_order <= other.time_order if hasattr(other, "time_order") else False
+
+    def to_human_string(self, _: IcebergType, value: Optional[S]) -> str:
+        return datetime.to_human_day(value) if isinstance(value, int) else "null"
+
+    @property
+    def dedup_name(self) -> str:
+        return "time"
+
+    def __repr__(self) -> str:
+        return "DayTransform()"
+
+
+class HourTransform(Transform[S, int]):
+    """Transforms a datetime value into a hour value.
+
+    Example:
+        >>> transform = HourTransform()
+        >>> transform.transform(TimestampType())(1512151975038194)
+        420042
+    """
+
+    __root__: Literal["hour"] = Field(default="hour")
+    _source_type: IcebergType = PrivateAttr()
+
+    def transform(self, source: IcebergType) -> Callable[[Optional[S]], Optional[int]]:
+        if type(source) in {TimestampType, TimestamptzType}:
+
+            def hour_func(v):
+                return datetime.micros_to_hours(v)
+
+        else:
+            raise ValueError(f"Cannot apply hour transform for type: {source}")
+
+        return lambda v: hour_func(v) if v else None
+
+    def can_transform(self, source: IcebergType) -> bool:
+        return type(source) in {
+            TimestampType,
+            TimestamptzType,
+        }
+
+    def result_type(self, source: IcebergType) -> IcebergType:
+        return IntegerType()
+
+    @property
+    def preserves_order(self) -> bool:
+        return True
+
+    @property
+    def time_order(self) -> int:
+        return 0
+
+    def satisfies_order_of(self, other: Transform) -> bool:
+        return self.time_order <= other.time_order if hasattr(other, "time_order") else False
+
+    def to_human_string(self, _: IcebergType, value: Optional[S]) -> str:
+        return datetime.to_human_hour(value) if isinstance(value, int) else "null"
+
+    @property
+    def dedup_name(self) -> str:
+        return "time"
+
+    def __repr__(self) -> str:
+        return "HourTransform()"
+
+
 def _base64encode(buffer: bytes) -> str:
     """Converts bytes to base64 string"""
     return base64.b64encode(buffer).decode("ISO-8859-1")

python/pyiceberg/utils/datetime.py

@@ -98,11 +98,26 @@ def micros_to_timestamptz(micros: int):
     return EPOCH_TIMESTAMPTZ + dt
 
 
+def to_human_year(year_ordinal: int) -> str:
+    """Converts a DateType value to human string"""
+    return f"{EPOCH_TIMESTAMP.year + year_ordinal:0=4d}"
+
+
+def to_human_month(month_ordinal: int) -> str:
+    """Converts a DateType value to human string"""
+    return f"{EPOCH_TIMESTAMP.year + int(month_ordinal / 12):0=4d}-{1 + int(month_ordinal % 12):0=2d}"
+
+
 def to_human_day(day_ordinal: int) -> str:
     """Converts a DateType value to human string"""
     return (EPOCH_DATE + timedelta(days=day_ordinal)).isoformat()
 
 
+def to_human_hour(hour_ordinal: int) -> str:
+    """Converts a DateType value to human string"""
+    return (EPOCH_TIMESTAMP + timedelta(hours=hour_ordinal)).isoformat("-", "hours")
+
+
 def to_human_time(micros_from_midnight: int) -> str:
     """Converts a TimeType value to human string"""
     return micros_to_time(micros_from_midnight).isoformat()
@@ -116,3 +131,33 @@ def to_human_timestamptz(timestamp_micros: int) -> str:
 def to_human_timestamp(timestamp_micros: int) -> str:
     """Converts a TimestampType value to human string"""
     return (EPOCH_TIMESTAMP + timedelta(microseconds=timestamp_micros)).isoformat()
+
+
+def micros_to_hours(timestamp: int) -> int:
+    """Converts a timestamp in microseconds to a date in hours"""
+    return int((datetime.utcfromtimestamp(timestamp / 1_000_000) - EPOCH_TIMESTAMP).total_seconds() / 3600)
+
+
+def days_to_months(days: int) -> int:
+    """Creates a date from the number of days from 1970-01-01"""
+    dt = datetime.utcfromtimestamp(days * 86400)
+    return (dt.year - EPOCH_TIMESTAMP.year) * 12 + (dt.month - EPOCH_TIMESTAMP.month) - (1 if dt.day < EPOCH_TIMESTAMP.day else 0)
+
+
+def micros_to_months(timestamp: int) -> int:
+    dt = datetime.utcfromtimestamp(timestamp / 1_000_000)
+    return (dt.year - EPOCH_TIMESTAMP.year) * 12 + (dt.month - EPOCH_TIMESTAMP.month) - (1 if dt.day < EPOCH_TIMESTAMP.day else 0)
+
+
+def days_to_years(days: int) -> int:
+    dt = datetime.utcfromtimestamp(days * 86400)
+    return (dt.year - EPOCH_TIMESTAMP.year) - (
+        1 if dt.month < EPOCH_TIMESTAMP.month or (dt.month == EPOCH_TIMESTAMP.month and dt.day < EPOCH_TIMESTAMP.day) else 0
+    )
+
+
+def micros_to_years(timestamp: int) -> int:
+    dt = datetime.utcfromtimestamp(timestamp / 1000000)
+    return (dt.year - EPOCH_TIMESTAMP.year) - (
+        1 if dt.month < EPOCH_TIMESTAMP.month or (dt.month == EPOCH_TIMESTAMP.month and dt.day < EPOCH_TIMESTAMP.day) else 0
+    )