API: implement types timestamp_ns and timestamptz_ns

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

@@ -31,6 +31,7 @@ public class BoundLiteralPredicate<T> extends BoundPredicate<T> {
           Type.TypeID.LONG,
           Type.TypeID.DATE,
           Type.TypeID.TIME,
+          Type.TypeID.TIMESTAMP_NANO,
           Type.TypeID.TIMESTAMP);
 
   private static long toLong(Literal<?> lit) {

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

@@ -43,6 +43,7 @@ public class ExpressionUtil {
       Transforms.bucket(Integer.MAX_VALUE).bind(Types.StringType.get());
   private static final OffsetDateTime EPOCH = Instant.ofEpochSecond(0).atOffset(ZoneOffset.UTC);
   private static final long FIVE_MINUTES_IN_MICROS = TimeUnit.MINUTES.toMicros(5);
+  private static final long FIVE_MINUTES_IN_NANOS = TimeUnit.MINUTES.toNanos(5);
   private static final long THREE_DAYS_IN_HOURS = TimeUnit.DAYS.toHours(3);
   private static final long NINETY_DAYS_IN_HOURS = TimeUnit.DAYS.toHours(90);
   private static final Pattern DATE = Pattern.compile("\\d{4}-\\d{2}-\\d{2}");
@@ -52,6 +53,12 @@ public class ExpressionUtil {
   private static final Pattern TIMESTAMPTZ =
       Pattern.compile(
           "\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}(:\\d{2}(.\\d{1,9})?)?([-+]\\d{2}:\\d{2}|Z)");
+  private static final Pattern TIMESTAMPNS =
+      Pattern.compile("\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}(:\\d{2}(.\\d{7,9})?)?");
+  private static final Pattern TIMESTAMPTZNS =
+      Pattern.compile(
+          "\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}(:\\d{2}(.\\d{7,9})?)?([-+]\\d{2}:\\d{2}|Z)");
+
   static final int LONG_IN_PREDICATE_ABBREVIATION_THRESHOLD = 10;
   private static final int LONG_IN_PREDICATE_ABBREVIATION_MIN_GAIN = 5;
 
@@ -515,6 +522,8 @@ private static String sanitize(Type type, Object value, long now, int today) {
         return "(time)";
       case TIMESTAMP:
         return sanitizeTimestamp((long) value, now);
+      case TIMESTAMP_NANO:
+        return sanitizeTimestamp((long) value / 1000, now);
       case STRING:
         return sanitizeString((CharSequence) value, now, today);
       case BOOLEAN:
@@ -536,6 +545,8 @@ private static String sanitize(Literal<?> literal, long now, int today) {
       return sanitizeDate(((Literals.DateLiteral) literal).value(), 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);
     } else if (literal instanceof Literals.TimeLiteral) {
       return "(time)";
     } else if (literal instanceof Literals.IntegerLiteral) {
@@ -594,6 +605,12 @@ private static String sanitizeString(CharSequence value, long now, int today) {
       if (DATE.matcher(value).matches()) {
         Literal<Integer> date = Literal.of(value).to(Types.DateType.get());
         return sanitizeDate(date.value(), today);
+      } else if (TIMESTAMPNS.matcher(value).matches()) {
+        Literal<Long> ts = Literal.of(value).to(Types.TimestampNanoType.withoutZone());
+        return sanitizeTimestamp(Math.floorDiv(ts.value(), 1000), now);
+      } else if (TIMESTAMPTZNS.matcher(value).matches()) {
+        Literal<Long> ts = Literal.of(value).to(Types.TimestampNanoType.withZone());
+        return sanitizeTimestamp(Math.floorDiv(ts.value(), 1000), now);
       } else if (TIMESTAMP.matcher(value).matches()) {
         Literal<Long> ts = Literal.of(value).to(Types.TimestampType.withoutZone());
         return sanitizeTimestamp(ts.value(), now);

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

@@ -40,6 +40,7 @@
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.Types;
 import org.apache.iceberg.util.ByteBuffers;
+import org.apache.iceberg.util.DateTimeUtil;
 import org.apache.iceberg.util.NaNUtil;
 
 class Literals {
@@ -299,6 +300,8 @@ public <T> Literal<T> to(Type type) {
           return (Literal<T>) new TimeLiteral(value());
         case TIMESTAMP:
           return (Literal<T>) new TimestampLiteral(value());
+        case TIMESTAMP_NANO:
+          return (Literal<T>) new TimestampNanoLiteral(value());
         case DATE:
           if ((long) Integer.MAX_VALUE < value()) {
             return aboveMax();
@@ -442,6 +445,8 @@ public <T> Literal<T> to(Type type) {
                   (int)
                       ChronoUnit.DAYS.between(
                           EPOCH_DAY, EPOCH.plus(value(), ChronoUnit.MICROS).toLocalDate()));
+        case TIMESTAMP_NANO:
+          return (Literal<T>) new TimestampNanoLiteral(DateTimeUtil.microsToNanos(value()));
         default:
       }
       return null;
@@ -453,6 +458,34 @@ protected Type.TypeID typeId() {
     }
   }
 
+  static class TimestampNanoLiteral extends ComparableLiteral<Long> {
+    TimestampNanoLiteral(Long value) {
+      super(value);
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    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()));
+        case TIMESTAMP:
+          return (Literal<T>) new TimestampLiteral(DateTimeUtil.nanosToMicros(value()));
+        case TIMESTAMP_NANO:
+          return (Literal<T>) this;
+        default:
+      }
+      return null;
+    }
+
+    @Override
+    protected Type.TypeID typeId() {
+      return Type.TypeID.TIMESTAMP_NANO;
+    }
+  }
+
   static class DecimalLiteral extends ComparableLiteral<BigDecimal> {
     DecimalLiteral(BigDecimal value) {
       super(value);
@@ -515,6 +548,18 @@ public <T> Literal<T> to(Type type) {
             return (Literal<T>) new TimestampLiteral(timestampMicros);
           }
 
+        case TIMESTAMP_NANO:
+          if (((Types.TimestampNanoType) type).shouldAdjustToUTC()) {
+            return (Literal<T>) new TimestampNanoLiteral(DateTimeUtil.isoTimestampToNanos(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);
+          }
+
         case STRING:
           return (Literal<T>) this;
 

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

@@ -63,6 +63,8 @@ static <T, B extends Bucket<T> & SerializableFunction<T, Integer>> B get(
       case FIXED:
       case BINARY:
         return (B) new BucketByteBuffer(numBuckets);
+      case TIMESTAMP_NANO:
+        return (B) new BucketTimestampNano(numBuckets);
       case UUID:
         return (B) new BucketUUID(numBuckets);
       default:
@@ -107,6 +109,7 @@ public boolean canTransform(Type type) {
       case DATE:
       case TIME:
       case TIMESTAMP:
+      case TIMESTAMP_NANO:
       case STRING:
       case BINARY:
       case FIXED:
@@ -214,6 +217,20 @@ protected int hash(Long value) {
     }
   }
 
+  // In order to bucket TimestampNano the same as Timestamp, we divide these values by 1000.
+  private static class BucketTimestampNano extends Bucket<Long>
+      implements SerializableFunction<Long, Integer> {
+
+    private BucketTimestampNano(int numBuckets) {
+      super(numBuckets);
+    }
+
+    @Override
+    protected int hash(Long value) {
+      return BucketUtil.hash(Math.floorDiv(value, 1000));
+    }
+  }
+
   private static class BucketString extends Bucket<CharSequence>
       implements SerializableFunction<CharSequence, Integer> {
 

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

@@ -37,7 +37,9 @@ protected Transform<T, Integer> toEnum(Type type) {
       case DATE:
         return (Transform<T, Integer>) Dates.DAY;
       case TIMESTAMP:
-        return (Transform<T, Integer>) Timestamps.DAY;
+        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);
     }
@@ -55,7 +57,16 @@ public boolean satisfiesOrderOf(Transform<?, ?> other) {
     }
 
     if (other instanceof Timestamps) {
-      return Timestamps.DAY.satisfiesOrderOf(other);
+      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) {

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

@@ -33,16 +33,19 @@ static <T> Hours<T> get() {
   @Override
   @SuppressWarnings("unchecked")
   protected Transform<T, Integer> toEnum(Type type) {
-    if (type.typeId() == Type.TypeID.TIMESTAMP) {
-      return (Transform<T, Integer>) Timestamps.HOUR;
+    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);
     }
-
-    throw new IllegalArgumentException("Unsupported type: " + type);
   }
 
   @Override
   public boolean canTransform(Type type) {
-    return type.typeId() == Type.TypeID.TIMESTAMP;
+    return type.typeId() == Type.TypeID.TIMESTAMP || type.typeId() == Type.TypeID.TIMESTAMP_NANO;
   }
 
   @Override
@@ -57,7 +60,7 @@ public boolean satisfiesOrderOf(Transform<?, ?> other) {
     }
 
     if (other instanceof Timestamps) {
-      return other == Timestamps.HOUR;
+      return other == Timestamps.HOUR_FROM_MICROS || other == Timestamps.HOUR_FROM_NANOS;
     } else if (other instanceof Hours
         || other instanceof Days
         || other instanceof Months

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

@@ -37,7 +37,9 @@ protected Transform<T, Integer> toEnum(Type type) {
       case DATE:
         return (Transform<T, Integer>) Dates.MONTH;
       case TIMESTAMP:
-        return (Transform<T, Integer>) Timestamps.MONTH;
+        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);
     }
@@ -55,7 +57,16 @@ public boolean satisfiesOrderOf(Transform<?, ?> other) {
     }
 
     if (other instanceof Timestamps) {
-      return Timestamps.MONTH.satisfiesOrderOf(other);
+      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) {

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

@@ -122,16 +122,23 @@ 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
+        || transform == Timestamps.YEAR_FROM_MICROS
+        || transform == Timestamps.YEAR_FROM_NANOS
         || transform instanceof Years) {
       return visitor.year(field.fieldId(), sourceName, field.sourceId());
     } else if (transform == Dates.MONTH
-        || transform == Timestamps.MONTH
+        || transform == Timestamps.MONTH_FROM_MICROS
+        || transform == Timestamps.MONTH_FROM_NANOS
         || transform instanceof Months) {
       return visitor.month(field.fieldId(), sourceName, field.sourceId());
-    } else if (transform == Dates.DAY || transform == Timestamps.DAY || transform instanceof Days) {
+    } else if (transform == Dates.DAY
+        || transform == Timestamps.DAY_FROM_MICROS
+        || transform == Timestamps.DAY_FROM_NANOS
+        || transform instanceof Days) {
       return visitor.day(field.fieldId(), sourceName, field.sourceId());
-    } else if (transform == Timestamps.HOUR || transform instanceof Hours) {
+    } else if (transform == Timestamps.HOUR_FROM_MICROS
+        || transform == Timestamps.HOUR_FROM_NANOS
+        || transform instanceof Hours) {
       return visitor.hour(field.fieldId(), sourceName, field.sourceId());
     } else if (transform instanceof VoidTransform) {
       return visitor.alwaysNull(field.fieldId(), sourceName, field.sourceId());

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

@@ -85,21 +85,26 @@ static <R> List<R> visit(SortOrder sortOrder, SortOrderVisitor<R> visitor) {
             visitor.truncate(
                 sourceName, field.sourceId(), width, field.direction(), field.nullOrder()));
       } else if (transform == Dates.YEAR
-          || transform == Timestamps.YEAR
+          || transform == Timestamps.YEAR_FROM_MICROS
+          || transform == Timestamps.YEAR_FROM_NANOS
           || transform instanceof Years) {
         results.add(
             visitor.year(sourceName, field.sourceId(), field.direction(), field.nullOrder()));
       } else if (transform == Dates.MONTH
-          || transform == Timestamps.MONTH
+          || transform == Timestamps.MONTH_FROM_MICROS
+          || transform == Timestamps.MONTH_FROM_NANOS
           || transform instanceof Months) {
         results.add(
             visitor.month(sourceName, field.sourceId(), field.direction(), field.nullOrder()));
       } else if (transform == Dates.DAY
-          || transform == Timestamps.DAY
+          || transform == Timestamps.DAY_FROM_MICROS
+          || transform == Timestamps.DAY_FROM_NANOS
           || transform instanceof Days) {
         results.add(
             visitor.day(sourceName, field.sourceId(), field.direction(), field.nullOrder()));
-      } else if (transform == Timestamps.HOUR || transform instanceof Hours) {
+      } else if (transform == Timestamps.HOUR_FROM_MICROS
+          || transform == Timestamps.HOUR_FROM_NANOS
+          || transform instanceof Hours) {
         results.add(
             visitor.hour(sourceName, field.sourceId(), field.direction(), field.nullOrder()));
       } else if (transform instanceof UnknownTransform) {

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

@@ -39,7 +39,9 @@ public boolean preservesOrder() {
 
   @Override
   public boolean canTransform(Type type) {
-    return type.typeId() == Type.TypeID.DATE || type.typeId() == Type.TypeID.TIMESTAMP;
+    return type.typeId() == Type.TypeID.DATE
+        || type.typeId() == Type.TypeID.TIMESTAMP
+        || type.typeId() == Type.TypeID.TIMESTAMP_NANO;
   }
 
   @Override