API: implement types timestamp_ns and timestamptz_ns

api/src/main/java/org/apache/iceberg/Schema.java

@@ -54,6 +54,8 @@ public class Schema implements Serializable {
   private static final Joiner NEWLINE = Joiner.on('\n');
   private static final String ALL_COLUMNS = "*";
   private static final int DEFAULT_SCHEMA_ID = 0;
+  private static final Map<Type.TypeID, Integer> MIN_FORMAT_VERSIONS =
+      ImmutableMap.of(Type.TypeID.TIMESTAMP_NANO, 3);
 
   private final StructType struct;
   private final int schemaId;
@@ -573,4 +575,27 @@ private List<NestedField> reassignIds(List<NestedField> columns, TypeUtil.GetID
             });
     return res.asStructType().fields();
   }
+
+  /**
+   * Check the compatibility of the schema with a format version.
+   *
+   * <p>This validates that the schema does not contain types that were released in later format
+   * versions.
+   *
+   * @param schema a Schema
+   * @param formatVersion table format version
+   */
+  public static void checkCompatibility(Schema schema, int formatVersion) {
+    // check the type in each field
+    for (NestedField field : schema.lazyIdToField().values()) {
+      Integer minFormatVersion = MIN_FORMAT_VERSIONS.get(field.type().typeId());
+      Preconditions.checkState(
+          minFormatVersion == null || formatVersion >= minFormatVersion,
+          "Invalid type in v%s schema: %s %s is not supported until v%s",
+          formatVersion,
+          schema.findColumnName(field.fieldId()),
+          field.type(),
+          minFormatVersion);
+    }
+  }
 }

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

@@ -36,13 +36,15 @@
 import org.apache.iceberg.transforms.Transforms;
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.DateTimeUtil;
 
 /** Expression utility methods. */
 public class ExpressionUtil {
   private static final Function<Object, Integer> HASH_FUNC =
       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 +54,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 +523,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(DateTimeUtil.nanosToMicros((long) value / 1000), now);
       case STRING:
         return sanitizeString((CharSequence) value, now, today);
       case BOOLEAN:
@@ -536,6 +546,9 @@ 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(
+          DateTimeUtil.nanosToMicros(((Literals.TimestampNanoLiteral) literal).value()), now);
     } else if (literal instanceof Literals.TimeLiteral) {
       return "(time)";
     } else if (literal instanceof Literals.IntegerLiteral) {
@@ -594,6 +607,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(DateTimeUtil.nanosToMicros(ts.value()), now);
+      } else if (TIMESTAMPTZNS.matcher(value).matches()) {
+        Literal<Long> ts = Literal.of(value).to(Types.TimestampNanoType.withZone());
+        return sanitizeTimestamp(DateTimeUtil.nanosToMicros(ts.value()), 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

@@ -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;
@@ -40,6 +39,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 +299,9 @@ public <T> Literal<T> to(Type type) {
           return (Literal<T>) new TimeLiteral(value());
         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()) {
             return aboveMax();
@@ -437,11 +440,9 @@ 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:
       }
       return null;
@@ -453,6 +454,32 @@ 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(DateTimeUtil.nanosToDays(value()));
+        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);
@@ -502,19 +529,21 @@ 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.isoTimestamptzToNanos(value()));
+          } else {
+            return (Literal<T>) new TimestampNanoLiteral(DateTimeUtil.isoTimestampToNanos(value()));
+          }
+
         case STRING:
           return (Literal<T>) this;
 

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 {
@@ -63,6 +64,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 +110,7 @@ public boolean canTransform(Type type) {
       case DATE:
       case TIME:
       case TIMESTAMP:
+      case TIMESTAMP_NANO:
       case STRING:
       case BINARY:
       case FIXED:
@@ -214,6 +218,20 @@ protected int hash(Long value) {
     }
   }
 
+  // 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> {
+
+    private BucketTimestampNano(int numBuckets) {
+      super(numBuckets);
+    }
+
+    @Override
+    protected int hash(Long nanos) {
+      return BucketUtil.hash(DateTimeUtil.nanosToMicros(nanos));
+    }
+  }
+
   private static class BucketString extends Bucket<CharSequence>
       implements SerializableFunction<CharSequence, Integer> {
 

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,38 +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;
-      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) {
-      return Timestamps.DAY.satisfiesOrderOf(other);
-    } 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,44 +31,28 @@ 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) {
-    if (type.typeId() == Type.TypeID.TIMESTAMP) {
-      return (Transform<T, Integer>) Timestamps.HOUR;
-    }
-
-    throw new IllegalArgumentException("Unsupported type: " + type);
+    return (Transform<T, Integer>)
+        fromSourceType(type, null, Timestamps.MICROS_TO_HOUR, Timestamps.NANOS_TO_HOUR);
   }
 
   @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
   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;
-    } 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";