[FLINK-40491][core] Extend BinaryVariant with TIME, TIMESTAMP_NS, TIMESTAMP_LTZ_NS primitives

flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariant.java

@@ -29,6 +29,7 @@
 import java.time.Instant;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
+import java.time.LocalTime;
 import java.time.ZoneId;
 import java.time.ZoneOffset;
 import java.time.temporal.ChronoUnit;
@@ -42,6 +43,7 @@
 import static org.apache.flink.types.variant.BinaryVariantUtil.SIZE_LIMIT;
 import static org.apache.flink.types.variant.BinaryVariantUtil.TIMESTAMP_FORMATTER;
 import static org.apache.flink.types.variant.BinaryVariantUtil.TIMESTAMP_LTZ_FORMATTER;
+import static org.apache.flink.types.variant.BinaryVariantUtil.TIME_FORMATTER;
 import static org.apache.flink.types.variant.BinaryVariantUtil.VERSION;
 import static org.apache.flink.types.variant.BinaryVariantUtil.VERSION_MASK;
 import static org.apache.flink.types.variant.BinaryVariantUtil.checkIndex;
@@ -193,6 +195,26 @@ public Instant getInstant() throws VariantTypeException {
         return microsToInstant(BinaryVariantUtil.getLong(value, pos));
     }
 
+    @Override
+    public LocalTime getTime() throws VariantTypeException {
+        checkType(Type.TIME, getType());
+        return LocalTime.ofNanoOfDay(BinaryVariantUtil.getLong(value, pos) * 1000);
+    }
+
+    @Override
+    public LocalDateTime getDateTimeNanos() throws VariantTypeException {
+        checkType(Type.TIMESTAMP_NS, getType());
+        return nanosToInstant(BinaryVariantUtil.getLong(value, pos))
+                .atZone(ZoneOffset.UTC)
+                .toLocalDateTime();
+    }
+
+    @Override
+    public Instant getInstantNanos() throws VariantTypeException {
+        checkType(Type.TIMESTAMP_LTZ_NS, getType());
+        return nanosToInstant(BinaryVariantUtil.getLong(value, pos));
+    }
+
     @Override
     public byte[] getBytes() throws VariantTypeException {
         checkType(Type.BYTES, getType());
@@ -224,10 +246,16 @@ public Object get() throws VariantTypeException {
                 return getString();
             case DATE:
                 return getDate();
+            case TIME:
+                return getTime();
             case TIMESTAMP:
                 return getDateTime();
             case TIMESTAMP_LTZ:
                 return getInstant();
+            case TIMESTAMP_NS:
+                return getDateTimeNanos();
+            case TIMESTAMP_LTZ_NS:
+                return getInstantNanos();
             case BYTES:
                 return getBytes();
             default:
@@ -391,6 +419,27 @@ private static void toJsonImpl(
                                 microsToInstant(BinaryVariantUtil.getLong(value, pos))
                                         .atZone(ZoneOffset.UTC)));
                 break;
+            case TIME:
+                appendQuoted(
+                        sb,
+                        TIME_FORMATTER.format(
+                                LocalTime.ofNanoOfDay(
+                                        BinaryVariantUtil.getLong(value, pos) * 1000)));
+                break;
+            case TIMESTAMP_LTZ_NS:
+                appendQuoted(
+                        sb,
+                        TIMESTAMP_LTZ_FORMATTER.format(
+                                nanosToInstant(BinaryVariantUtil.getLong(value, pos))
+                                        .atZone(zoneId)));
+                break;
+            case TIMESTAMP_NS:
+                appendQuoted(
+                        sb,
+                        TIMESTAMP_FORMATTER.format(
+                                nanosToInstant(BinaryVariantUtil.getLong(value, pos))
+                                        .atZone(ZoneOffset.UTC)));
+                break;
             case FLOAT:
                 {
                     final float f = BinaryVariantUtil.getFloat(value, pos);
@@ -418,6 +467,10 @@ private static Instant microsToInstant(long timestamp) {
         return Instant.EPOCH.plus(timestamp, ChronoUnit.MICROS);
     }
 
+    private static Instant nanosToInstant(long timestamp) {
+        return Instant.EPOCH.plus(timestamp, ChronoUnit.NANOS);
+    }
+
     private void checkType(Type expected, Type actual) {
         if (expected != actual) {
             throw new VariantTypeException(

flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantBuilder.java

@@ -25,6 +25,7 @@
 import java.time.Instant;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
+import java.time.LocalTime;
 import java.time.ZoneOffset;
 import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
@@ -106,7 +107,11 @@ public Variant of(BigDecimal bigDecimal) {
     @Override
     public Variant of(Instant instant) {
         BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false);
-        builder.appendTimestampLtz(ChronoUnit.MICROS.between(Instant.EPOCH, instant));
+        if (instant.getNano() % 1000 == 0) {
+            builder.appendTimestampLtz(ChronoUnit.MICROS.between(Instant.EPOCH, instant));
+        } else {
+            builder.appendTimestampLtzNanos(nanosSinceEpoch(instant));
+        }
         return builder.build();
     }
 
@@ -120,8 +125,32 @@ public Variant of(LocalDate localDate) {
     @Override
     public Variant of(LocalDateTime localDateTime) {
         BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false);
-        builder.appendTimestamp(
-                ChronoUnit.MICROS.between(Instant.EPOCH, localDateTime.toInstant(ZoneOffset.UTC)));
+        Instant instant = localDateTime.toInstant(ZoneOffset.UTC);
+        if (localDateTime.getNano() % 1000 == 0) {
+            builder.appendTimestamp(ChronoUnit.MICROS.between(Instant.EPOCH, instant));
+        } else {
+            builder.appendTimestampNanos(nanosSinceEpoch(instant));
+        }
+        return builder.build();
+    }
+
+    private static long nanosSinceEpoch(Instant instant) {
+        try {
+            return ChronoUnit.NANOS.between(Instant.EPOCH, instant);
+        } catch (ArithmeticException e) {
+            throw new VariantTypeException(
+                    String.format(
+                            "%s is outside the +/-292 year range (1677-09-21 to 2262-04-11) "
+                                    + "supported by nanosecond precision variant timestamps. Use "
+                                    + "microsecond precision instead.",
+                            instant));
+        }
+    }
+
+    @Override
+    public Variant of(LocalTime localTime) {
+        BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false);
+        builder.appendTime(localTime.toNanoOfDay() / 1000);
         return builder.build();
     }
 

flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantInternalBuilder.java

@@ -58,8 +58,11 @@
 import static org.apache.flink.types.variant.BinaryVariantUtil.NULL;
 import static org.apache.flink.types.variant.BinaryVariantUtil.OBJECT;
 import static org.apache.flink.types.variant.BinaryVariantUtil.SIZE_LIMIT;
+import static org.apache.flink.types.variant.BinaryVariantUtil.TIME;
 import static org.apache.flink.types.variant.BinaryVariantUtil.TIMESTAMP;
 import static org.apache.flink.types.variant.BinaryVariantUtil.TIMESTAMP_LTZ;
+import static org.apache.flink.types.variant.BinaryVariantUtil.TIMESTAMP_LTZ_NS;
+import static org.apache.flink.types.variant.BinaryVariantUtil.TIMESTAMP_NS;
 import static org.apache.flink.types.variant.BinaryVariantUtil.TRUE;
 import static org.apache.flink.types.variant.BinaryVariantUtil.U16_MAX;
 import static org.apache.flink.types.variant.BinaryVariantUtil.U24_MAX;
@@ -288,6 +291,27 @@ public void appendTimestamp(long microsSinceEpoch) {
         writePos += 8;
     }
 
+    public void appendTime(long microsSinceMidnight) {
+        checkCapacity(1 + 8);
+        writeBuffer[writePos++] = primitiveHeader(TIME);
+        writeLong(writeBuffer, writePos, microsSinceMidnight, 8);
+        writePos += 8;
+    }
+
+    public void appendTimestampLtzNanos(long nanosSinceEpoch) {
+        checkCapacity(1 + 8);
+        writeBuffer[writePos++] = primitiveHeader(TIMESTAMP_LTZ_NS);
+        writeLong(writeBuffer, writePos, nanosSinceEpoch, 8);
+        writePos += 8;
+    }
+
+    public void appendTimestampNanos(long nanosSinceEpoch) {
+        checkCapacity(1 + 8);
+        writeBuffer[writePos++] = primitiveHeader(TIMESTAMP_NS);
+        writeLong(writeBuffer, writePos, nanosSinceEpoch, 8);
+        writePos += 8;
+    }
+
     public void appendFloat(float f) {
         checkCapacity(1 + 4);
         writeBuffer[writePos++] = primitiveHeader(FLOAT);

flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantUtil.java

@@ -127,6 +127,16 @@ public class BinaryVariantUtil {
     // Long string value. The content is (4-byte little-endian unsigned integer representing the
     // string size) + (size bytes of string content).
     public static final int LONG_STR = 16;
+    // Time value, no time zone. Content is 8-byte little-endian signed integer that represents
+    // the number of microseconds since midnight.
+    public static final int TIME = 17;
+    // TimestampLTZ value with nanosecond precision. Content is 8-byte little-endian signed
+    // integer that represents the number of nanoseconds elapsed since the Unix epoch,
+    // 1970-01-01 00:00:00 UTC.
+    public static final int TIMESTAMP_LTZ_NS = 18;
+    // Timestamp value with nanosecond precision. It has the same content as `TIMESTAMP_LTZ_NS`
+    // but should always be interpreted as if the local time zone is UTC.
+    public static final int TIMESTAMP_NS = 19;
 
     public static final byte VERSION = 1;
     // The lower 4 bits of the first metadata byte contain the version.
@@ -160,6 +170,11 @@ public class BinaryVariantUtil {
                     .appendOffset("+HH:MM", "+00:00")
                     .toFormatter(Locale.US);
 
+    public static final DateTimeFormatter TIME_FORMATTER =
+            new DateTimeFormatterBuilder()
+                    .append(DateTimeFormatter.ISO_LOCAL_TIME)
+                    .toFormatter(Locale.US);
+
     // Write the least significant `numBytes` bytes in `value` into `bytes[pos, pos + numBytes)` in
     // little endian.
     public static void writeLong(byte[] bytes, int pos, long value, int numBytes) {
@@ -304,6 +319,12 @@ public static Type getType(byte[] value, int pos) {
                         return Type.BYTES;
                     case LONG_STR:
                         return Type.STRING;
+                    case TIME:
+                        return Type.TIME;
+                    case TIMESTAMP_LTZ_NS:
+                        return Type.TIMESTAMP_LTZ_NS;
+                    case TIMESTAMP_NS:
+                        return Type.TIMESTAMP_NS;
                     default:
                         throw unknownPrimitiveTypeInVariant(typeInfo);
                 }
@@ -360,6 +381,9 @@ public static int valueSize(byte[] value, int pos) {
                     case DOUBLE:
                     case TIMESTAMP_LTZ:
                     case TIMESTAMP:
+                    case TIME:
+                    case TIMESTAMP_LTZ_NS:
+                    case TIMESTAMP_NS:
                         return 9;
                     case DECIMAL4:
                         return 6;
@@ -393,16 +417,22 @@ public static boolean getBoolean(byte[] value, int pos) {
     }
 
     // Get a long value from variant value `value[pos...]`.
-    // It is only legal to call it if `getType` returns one of `Type.LONG/DATE/TIMESTAMP/
-    // TIMESTAMP_LTZ`. If the type is `DATE`, the return value is guaranteed to fit into an int and
-    // represents the number of days from the Unix epoch.
+    // It is only legal to call it if `getType` returns one of `Type.LONG/DATE/TIME/TIMESTAMP/
+    // TIMESTAMP_LTZ/TIMESTAMP_NS/TIMESTAMP_LTZ_NS`.
+    // If the type is `DATE`, the return value is
+    // guaranteed to fit into an int and represents the number of days from the Unix epoch.
+    // If the type is `TIME`, the return value represents the number of microseconds since
+    // midnight.
     // If the type is `TIMESTAMP/TIMESTAMP_LTZ`, the return value represents the number of
     // microseconds from the Unix epoch.
+    // If the type is `TIMESTAMP_NS/TIMESTAMP_LTZ_NS`, the return value represents the number of
+    // nanoseconds from the Unix epoch.
     public static long getLong(byte[] value, int pos) {
         checkIndex(pos, value.length);
         int basicType = value[pos] & BASIC_TYPE_MASK;
         int typeInfo = (value[pos] >> BASIC_TYPE_BITS) & TYPE_INFO_MASK;
-        String exceptionMessage = "Expect type to be LONG/DATE/TIMESTAMP/TIMESTAMP_LTZ";
+        String exceptionMessage =
+                "Expect type to be LONG/DATE/TIME/TIMESTAMP/TIMESTAMP_LTZ/TIMESTAMP_NS/TIMESTAMP_LTZ_NS";
         if (basicType != PRIMITIVE) {
             throw new IllegalStateException(exceptionMessage);
         }
@@ -415,8 +445,11 @@ public static long getLong(byte[] value, int pos) {
             case DATE:
                 return readLong(value, pos + 1, 4);
             case INT8:
+            case TIME:
             case TIMESTAMP_LTZ:
             case TIMESTAMP:
+            case TIMESTAMP_LTZ_NS:
+            case TIMESTAMP_NS:
                 return readLong(value, pos + 1, 8);
             default:
                 throw new IllegalStateException(exceptionMessage);

flink-core/src/main/java/org/apache/flink/types/variant/Variant.java

@@ -25,6 +25,7 @@
 import java.time.Instant;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
+import java.time.LocalTime;
 import java.util.List;
 
 /**
@@ -151,6 +152,33 @@ public interface Variant extends Serializable {
      */
     Instant getInstant() throws VariantTypeException;
 
+    /**
+     * Get the scalar value of variant as {@link LocalTime}, if the variant type is {@link
+     * Type#TIME}. The returned value has microsecond precision.
+     *
+     * @throws VariantTypeException If this variant is not a scalar value or is not {@link
+     *     Type#TIME}.
+     */
+    LocalTime getTime() throws VariantTypeException;
+
+    /**
+     * Get the scalar value of variant as {@link LocalDateTime}, if the variant type is {@link
+     * Type#TIMESTAMP_NS}. The returned value has nanosecond precision.
+     *
+     * @throws VariantTypeException If this variant is not a scalar value or is not {@link
+     *     Type#TIMESTAMP_NS}.
+     */
+    LocalDateTime getDateTimeNanos() throws VariantTypeException;
+
+    /**
+     * Get the scalar value of variant as {@link Instant}, if the variant type is {@link
+     * Type#TIMESTAMP_LTZ_NS}. The returned value has nanosecond precision.
+     *
+     * @throws VariantTypeException If this variant is not a scalar value or is not {@link
+     *     Type#TIMESTAMP_LTZ_NS}.
+     */
+    Instant getInstantNanos() throws VariantTypeException;
+
     /**
      * Get the scalar value of variant as byte array, if the variant type is {@link Type#BYTES}.
      *
@@ -232,8 +260,11 @@ enum Type {
         DECIMAL,
         STRING,
         DATE,
+        TIME,
         TIMESTAMP,
         TIMESTAMP_LTZ,
+        TIMESTAMP_NS,
+        TIMESTAMP_LTZ_NS,
         BYTES
     }
 

flink-core/src/main/java/org/apache/flink/types/variant/VariantBuilder.java

@@ -24,6 +24,7 @@
 import java.time.Instant;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
+import java.time.LocalTime;
 
 /** Builder for variants. */
 @PublicEvolving
@@ -68,6 +69,9 @@ public interface VariantBuilder {
     /** Create a variant from a LocalDateTime. */
     Variant of(LocalDateTime localDateTime);
 
+    /** Create a variant from a LocalTime. Sub-microsecond precision is truncated. */
+    Variant of(LocalTime localTime);
+
     /** Create a variant of null. */
     Variant ofNull();