Review fixes for timestamp_ns API changes

api/src/main/java/org/apache/iceberg/expressions/ExpressionUtil.java

@@ -524,7 +524,7 @@ private static String sanitize(Type type, Object value, long now, int today) {
       case TIMESTAMP:
         return sanitizeTimestamp((long) value, now);
       case TIMESTAMP_NANO:
-        return sanitizeTimestamp((long) value / 1000, now);
+        return sanitizeTimestamp(DateTimeUtil.nanosToMicros((long) value / 1000), now);
       case STRING:
         return sanitizeString((CharSequence) value, now, today);
       case BOOLEAN:
@@ -547,7 +547,8 @@ private static String sanitize(Literal<?> literal, long now, int today) {
     } else if (literal instanceof Literals.TimestampLiteral) {
       return sanitizeTimestamp(((Literals.TimestampLiteral) literal).value(), now);
     } else if (literal instanceof Literals.TimestampNanoLiteral) {
-      return sanitizeTimestamp(((Literals.TimestampNanoLiteral) literal).value() / 1000, now);
+      return sanitizeTimestamp(
+          DateTimeUtil.nanosToMicros(((Literals.TimestampNanoLiteral) literal).value()), now);
     } else if (literal instanceof Literals.TimeLiteral) {
       return "(time)";
     } else if (literal instanceof Literals.IntegerLiteral) {

api/src/main/java/org/apache/iceberg/expressions/Literals.java

@@ -24,7 +24,6 @@
 import java.nio.ByteBuffer;
 import java.time.Instant;
 import java.time.LocalDate;
-import java.time.LocalDateTime;
 import java.time.LocalTime;
 import java.time.OffsetDateTime;
 import java.time.ZoneOffset;
@@ -301,6 +300,7 @@ public <T> Literal<T> to(Type type) {
         case TIMESTAMP:
           return (Literal<T>) new TimestampLiteral(value());
         case TIMESTAMP_NANO:
+          // assume micros and convert to nanos to match the behavior in the timestamp case above
           return new TimestampLiteral(value()).to(type);
         case DATE:
           if ((long) Integer.MAX_VALUE < value()) {
@@ -440,11 +440,7 @@ public <T> Literal<T> to(Type type) {
         case TIMESTAMP:
           return (Literal<T>) this;
         case DATE:
-          return (Literal<T>)
-              new DateLiteral(
-                  (int)
-                      ChronoUnit.DAYS.between(
-                          EPOCH_DAY, EPOCH.plus(value(), ChronoUnit.MICROS).toLocalDate()));
+          return (Literal<T>) new DateLiteral(DateTimeUtil.microsToDays(value()));
         case TIMESTAMP_NANO:
           return (Literal<T>) new TimestampNanoLiteral(DateTimeUtil.microsToNanos(value()));
         default:
@@ -468,9 +464,7 @@ static class TimestampNanoLiteral extends ComparableLiteral<Long> {
     public <T> Literal<T> to(Type type) {
       switch (type.typeId()) {
         case DATE:
-          return (Literal<T>)
-              new DateLiteral(
-                  (int) ChronoUnit.DAYS.between(EPOCH_DAY, EPOCH.plusNanos(value()).toLocalDate()));
+          return (Literal<T>) new DateLiteral(DateTimeUtil.nanosToDays(value()));
         case TIMESTAMP:
           return (Literal<T>) new TimestampLiteral(DateTimeUtil.nanosToMicros(value()));
         case TIMESTAMP_NANO:
@@ -535,29 +529,18 @@ public <T> Literal<T> to(Type type) {
 
         case TIMESTAMP:
           if (((Types.TimestampType) type).shouldAdjustToUTC()) {
-            long timestampMicros =
-                ChronoUnit.MICROS.between(
-                    EPOCH, OffsetDateTime.parse(value(), DateTimeFormatter.ISO_DATE_TIME));
+            long timestampMicros = DateTimeUtil.isoTimestamptzToMicros(value().toString());
             return (Literal<T>) new TimestampLiteral(timestampMicros);
           } else {
-            long timestampMicros =
-                ChronoUnit.MICROS.between(
-                    EPOCH,
-                    LocalDateTime.parse(value(), DateTimeFormatter.ISO_LOCAL_DATE_TIME)
-                        .atOffset(ZoneOffset.UTC));
+            long timestampMicros = DateTimeUtil.isoTimestampToMicros(value().toString());
             return (Literal<T>) new TimestampLiteral(timestampMicros);
           }
 
         case TIMESTAMP_NANO:
           if (((Types.TimestampNanoType) type).shouldAdjustToUTC()) {
-            return (Literal<T>) new TimestampNanoLiteral(DateTimeUtil.isoTimestampToNanos(value()));
+            return (Literal<T>) new TimestampNanoLiteral(DateTimeUtil.isoTimestamptzToNanos(value()));
           } else {
-            long timestampNanos =
-                ChronoUnit.NANOS.between(
-                    EPOCH,
-                    LocalDateTime.parse(value(), DateTimeFormatter.ISO_LOCAL_DATE_TIME)
-                        .atOffset(ZoneOffset.UTC));
-            return (Literal<T>) new TimestampNanoLiteral(timestampNanos);
+            return (Literal<T>) new TimestampNanoLiteral(DateTimeUtil.isoTimestampToNanos(value()));
           }
 
         case STRING:

api/src/main/java/org/apache/iceberg/transforms/Bucket.java

@@ -33,6 +33,7 @@
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.Types;
 import org.apache.iceberg.util.BucketUtil;
+import org.apache.iceberg.util.DateTimeUtil;
 import org.apache.iceberg.util.SerializableFunction;
 
 class Bucket<T> implements Transform<T, Integer>, Serializable {
@@ -217,7 +218,7 @@ protected int hash(Long value) {
     }
   }
 
-  // In order to bucket TimestampNano the same as Timestamp, we divide these values by 1000.
+  // In order to bucket TimestampNano the same as Timestamp, convert to micros before hashing.
   private static class BucketTimestampNano extends Bucket<Long>
       implements SerializableFunction<Long, Integer> {
 
@@ -226,8 +227,8 @@ private BucketTimestampNano(int numBuckets) {
     }
 
     @Override
-    protected int hash(Long value) {
-      return BucketUtil.hash(Math.floorDiv(value, 1000));
+    protected int hash(Long nanos) {
+      return BucketUtil.hash(DateTimeUtil.nanosToMicros(nanos));
     }
   }
 

api/src/main/java/org/apache/iceberg/transforms/Dates.java

@@ -97,6 +97,10 @@ public Type getResultType(Type sourceType) {
     return Types.IntegerType.get();
   }
 
+  ChronoUnit granularity() {
+    return granularity;
+  }
+
   @Override
   public boolean preservesOrder() {
     return true;
@@ -109,11 +113,11 @@ public boolean satisfiesOrderOf(Transform<?, ?> other) {
     }
 
     if (other instanceof Dates) {
-      // test the granularity, in days. day(ts) => 1 day, months(ts) => 30 days, and day satisfies
-      // the order of months
-      Dates otherTransform = (Dates) other;
-      return granularity.getDuration().toDays()
-          <= otherTransform.granularity.getDuration().toDays();
+      return TransformUtil.satisfiesOrderOf(granularity, ((Dates) other).granularity());
+    } else if (other instanceof Timestamps) {
+      return TransformUtil.satisfiesOrderOf(granularity, ((Timestamps) other).granularity());
+    } else if (other instanceof TimeTransform) {
+      return TransformUtil.satisfiesOrderOf(granularity, ((TimeTransform<?>) other).granularity());
     }
 
     return false;

api/src/main/java/org/apache/iceberg/transforms/Days.java

@@ -19,6 +19,7 @@
 package org.apache.iceberg.transforms;
 
 import java.io.ObjectStreamException;
+import java.time.temporal.ChronoUnit;
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.Types;
 
@@ -31,49 +32,19 @@ static <T> Days<T> get() {
   }
 
   @Override
-  @SuppressWarnings("unchecked")
-  protected Transform<T, Integer> toEnum(Type type) {
-    switch (type.typeId()) {
-      case DATE:
-        return (Transform<T, Integer>) Dates.DAY;
-      case TIMESTAMP:
-        return (Transform<T, Integer>) Timestamps.DAY_FROM_MICROS;
-      case TIMESTAMP_NANO:
-        return (Transform<T, Integer>) Timestamps.DAY_FROM_NANOS;
-      default:
-        throw new IllegalArgumentException("Unsupported type: " + type);
-    }
+  protected ChronoUnit granularity() {
+    return ChronoUnit.DAYS;
   }
 
   @Override
-  public Type getResultType(Type sourceType) {
-    return Types.DateType.get();
+  protected Transform<T, Integer> toEnum(Type type) {
+    return (Transform<T, Integer>)
+        fromSourceType(type, Dates.DAY, Timestamps.MICROS_TO_DAY, Timestamps.NANOS_TO_DAY);
   }
 
   @Override
-  public boolean satisfiesOrderOf(Transform<?, ?> other) {
-    if (this == other) {
-      return true;
-    }
-
-    if (other instanceof Timestamps) {
-      Timestamps.ResultTypeUnit otherResultTypeUnit = ((Timestamps) other).resultTypeUnit();
-      switch (otherResultTypeUnit) {
-        case MICROS:
-          return Timestamps.DAY_FROM_MICROS.satisfiesOrderOf(other);
-        case NANOS:
-          return Timestamps.DAY_FROM_NANOS.satisfiesOrderOf(other);
-        default:
-          throw new UnsupportedOperationException(
-              "Unsupported timestamp unit: " + otherResultTypeUnit);
-      }
-    } else if (other instanceof Dates) {
-      return Dates.DAY.satisfiesOrderOf(other);
-    } else if (other instanceof Days || other instanceof Months || other instanceof Years) {
-      return true;
-    }
-
-    return false;
+  public Type getResultType(Type sourceType) {
+    return Types.DateType.get();
   }
 
   @Override

api/src/main/java/org/apache/iceberg/transforms/Hours.java

@@ -19,6 +19,7 @@
 package org.apache.iceberg.transforms;
 
 import java.io.ObjectStreamException;
+import java.time.temporal.ChronoUnit;
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.Types;
 
@@ -30,17 +31,16 @@ static <T> Hours<T> get() {
     return (Hours<T>) INSTANCE;
   }
 
+  @Override
+  protected ChronoUnit granularity() {
+    return ChronoUnit.HOURS;
+  }
+
   @Override
   @SuppressWarnings("unchecked")
   protected Transform<T, Integer> toEnum(Type type) {
-    switch (type.typeId()) {
-      case TIMESTAMP:
-        return (Transform<T, Integer>) Timestamps.HOUR_FROM_MICROS;
-      case TIMESTAMP_NANO:
-        return (Transform<T, Integer>) Timestamps.HOUR_FROM_NANOS;
-      default:
-        throw new IllegalArgumentException("Unsupported type: " + type);
-    }
+    return (Transform<T, Integer>)
+        fromSourceType(type, null, Timestamps.MICROS_TO_HOUR, Timestamps.NANOS_TO_HOUR);
   }
 
   @Override
@@ -53,24 +53,6 @@ public Type getResultType(Type sourceType) {
     return Types.IntegerType.get();
   }
 
-  @Override
-  public boolean satisfiesOrderOf(Transform<?, ?> other) {
-    if (this == other) {
-      return true;
-    }
-
-    if (other instanceof Timestamps) {
-      return other == Timestamps.HOUR_FROM_MICROS || other == Timestamps.HOUR_FROM_NANOS;
-    } else if (other instanceof Hours
-        || other instanceof Days
-        || other instanceof Months
-        || other instanceof Years) {
-      return true;
-    }
-
-    return false;
-  }
-
   @Override
   public String toHumanString(Type alwaysInt, Integer value) {
     return value != null ? TransformUtil.humanHour(value) : "null";

api/src/main/java/org/apache/iceberg/transforms/Months.java

@@ -19,6 +19,7 @@
 package org.apache.iceberg.transforms;
 
 import java.io.ObjectStreamException;
+import java.time.temporal.ChronoUnit;
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.Types;
 
@@ -31,49 +32,19 @@ static <T> Months<T> get() {
   }
 
   @Override
-  @SuppressWarnings("unchecked")
-  protected Transform<T, Integer> toEnum(Type type) {
-    switch (type.typeId()) {
-      case DATE:
-        return (Transform<T, Integer>) Dates.MONTH;
-      case TIMESTAMP:
-        return (Transform<T, Integer>) Timestamps.MONTH_FROM_MICROS;
-      case TIMESTAMP_NANO:
-        return (Transform<T, Integer>) Timestamps.MONTH_FROM_NANOS;
-      default:
-        throw new IllegalArgumentException("Unsupported type: " + type);
-    }
+  protected ChronoUnit granularity() {
+    return ChronoUnit.MONTHS;
   }
 
   @Override
-  public Type getResultType(Type sourceType) {
-    return Types.IntegerType.get();
+  protected Transform<T, Integer> toEnum(Type type) {
+    return (Transform<T, Integer>)
+        fromSourceType(type, Dates.MONTH, Timestamps.MICROS_TO_MONTH, Timestamps.NANOS_TO_MONTH);
   }
 
   @Override
-  public boolean satisfiesOrderOf(Transform<?, ?> other) {
-    if (this == other) {
-      return true;
-    }
-
-    if (other instanceof Timestamps) {
-      Timestamps.ResultTypeUnit otherResultTypeUnit = ((Timestamps) other).resultTypeUnit();
-      switch (otherResultTypeUnit) {
-        case MICROS:
-          return Timestamps.MONTH_FROM_MICROS.satisfiesOrderOf(other);
-        case NANOS:
-          return Timestamps.MONTH_FROM_NANOS.satisfiesOrderOf(other);
-        default:
-          throw new UnsupportedOperationException(
-              "Unsupported timestamp unit: " + otherResultTypeUnit);
-      }
-    } else if (other instanceof Dates) {
-      return Dates.MONTH.satisfiesOrderOf(other);
-    } else if (other instanceof Months || other instanceof Years) {
-      return true;
-    }
-
-    return false;
+  public Type getResultType(Type sourceType) {
+    return Types.IntegerType.get();
   }
 
   @Override

api/src/main/java/org/apache/iceberg/transforms/PartitionSpecVisitor.java

@@ -122,22 +122,22 @@ static <R> R visit(Schema schema, PartitionField field, PartitionSpecVisitor<R>
       int width = ((Truncate<?>) transform).width();
       return visitor.truncate(field.fieldId(), sourceName, field.sourceId(), width);
     } else if (transform == Dates.YEAR
-        || transform == Timestamps.YEAR_FROM_MICROS
-        || transform == Timestamps.YEAR_FROM_NANOS
+        || transform == Timestamps.MICROS_TO_YEAR
+        || transform == Timestamps.NANOS_TO_YEAR
         || transform instanceof Years) {
       return visitor.year(field.fieldId(), sourceName, field.sourceId());
     } else if (transform == Dates.MONTH
-        || transform == Timestamps.MONTH_FROM_MICROS
-        || transform == Timestamps.MONTH_FROM_NANOS
+        || transform == Timestamps.MICROS_TO_MONTH
+        || transform == Timestamps.NANOS_TO_MONTH
         || transform instanceof Months) {
       return visitor.month(field.fieldId(), sourceName, field.sourceId());
     } else if (transform == Dates.DAY
-        || transform == Timestamps.DAY_FROM_MICROS
-        || transform == Timestamps.DAY_FROM_NANOS
+        || transform == Timestamps.MICROS_TO_DAY
+        || transform == Timestamps.NANOS_TO_DAY
         || transform instanceof Days) {
       return visitor.day(field.fieldId(), sourceName, field.sourceId());
-    } else if (transform == Timestamps.HOUR_FROM_MICROS
-        || transform == Timestamps.HOUR_FROM_NANOS
+    } else if (transform == Timestamps.MICROS_TO_HOUR
+        || transform == Timestamps.NANOS_TO_HOUR
         || transform instanceof Hours) {
       return visitor.hour(field.fieldId(), sourceName, field.sourceId());
     } else if (transform instanceof VoidTransform) {