diff --git a/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariant.java b/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariant.java index 97c11eedf52b1f..da4a5401ecfbf6 100644 --- a/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariant.java +++ b/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( diff --git a/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantBuilder.java b/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantBuilder.java index 4b7cd0558e48b4..71fed320dc1e1a 100644 --- a/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantBuilder.java +++ b/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(); } diff --git a/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantInternalBuilder.java b/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantInternalBuilder.java index 1674d166266f78..5b0cadddc3cc1b 100644 --- a/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantInternalBuilder.java +++ b/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); diff --git a/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantUtil.java b/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantUtil.java index c3ab8d29be811b..8bf7ef61fab992 100644 --- a/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantUtil.java +++ b/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantUtil.java @@ -43,93 +43,167 @@ * the below constants for all possible basic type and type info values. * *

The variant metadata includes a version id and a dictionary of distinct strings - * (case-sensitive). Its binary format is: - Version: 1-byte unsigned integer. The only acceptable - * value is 1 currently. - Dictionary size: 4-byte little-endian unsigned integer. The number of - * keys in the dictionary. - Offsets: (size + 1) * 4-byte little-endian unsigned integers. - * `offsets[i]` represents the starting position of string i, counting starting from the address of - * `offsets[0]`. Strings must be stored contiguously, so we don’t need to store the string size, - * instead, we compute it with `offset[i + 1] - offset[i]`. - UTF-8 string data. + * (case-sensitive). Its binary format is: + * + *

*/ @Internal public class BinaryVariantUtil { public static final int BASIC_TYPE_BITS = 2; public static final int BASIC_TYPE_MASK = 0x3; public static final int TYPE_INFO_MASK = 0x3F; - // The inclusive maximum value of the type info value. It is the size limit of `SHORT_STR`. + + /** + * The inclusive maximum value of the type info value. It is the size limit of {@link + * #SHORT_STR}. + */ public static final int MAX_SHORT_STR_SIZE = 0x3F; - // Below is all possible basic type values. - // Primitive value. The type info value must be one of the values in the below section. + // Below are all possible basic type values. + + /** Primitive value. The type info value must be one of the values in the below section. */ public static final int PRIMITIVE = 0; - // Short string value. The type info value is the string size, which must be in `[0, - // kMaxShortStrSize]`. - // The string content bytes directly follow the header byte. + + /** + * Short string value. The type info value is the string size, which must be in {@code [0, + * MAX_SHORT_STR_SIZE]}. The string content bytes directly follow the header byte. + */ public static final int SHORT_STR = 1; - // Object value. The content contains a size, a list of field ids, a list of field offsets, and - // the actual field data. The length of the id list is `size`, while the length of the offset - // list is `size + 1`, where the last offset represent the total size of the field data. The - // fields in an object must be sorted by the field name in alphabetical order. Duplicate field - // names in one object are not allowed. - // We use 5 bits in the type info to specify the integer type of the object header: it should - // be 0_b4_b3b2_b1b0 (MSB is 0), where: - // - b4 specifies the type of size. When it is 0/1, `size` is a little-endian 1/4-byte - // unsigned integer. - // - b3b2/b1b0 specifies the integer type of id and offset. When the 2 bits are 0/1/2, the - // list contains 1/2/3-byte little-endian unsigned integers. + + /** + * Object value. The content contains a size, a list of field ids, a list of field offsets, and + * the actual field data. The length of the id list is {@code size}, while the length of the + * offset list is {@code size + 1}, where the last offset represent the total size of the field + * data. The fields in an object must be sorted by the field name in alphabetical order. + * Duplicate field names in one object are not allowed. + * + *

We use 5 bits in the type info to specify the integer type of the object header: it should + * be 0_b4_b3b2_b1b0 (MSB is 0), where: + * + *

+ */ public static final int OBJECT = 2; - // Array value. The content contains a size, a list of field offsets, and the actual element - // data. It is similar to an object without the id list. The length of the offset list - // is `size + 1`, where the last offset represent the total size of the element data. - // Its type info should be: 000_b2_b1b0: - // - b2 specifies the type of size. - // - b1b0 specifies the integer type of offset. + + /** + * Array value. The content contains a size, a list of field offsets, and the actual element + * data. It is similar to an object without the id list. The length of the offset list is {@code + * size + 1}, where the last offset represent the total size of the element data. + * + *

Its type info should be: 000_b2_b1b0: + * + *

+ */ public static final int ARRAY = 3; - // Below is all possible type info values for `PRIMITIVE`. - // JSON Null value. Empty content. + // Below are all possible type info values for PRIMITIVE. + + /** JSON Null value. Empty content. */ public static final int NULL = 0; - // True value. Empty content. + + /** True value. Empty content. */ public static final int TRUE = 1; - // False value. Empty content. + + /** False value. Empty content. */ public static final int FALSE = 2; - // 1-byte little-endian signed integer. + + /** 1-byte little-endian signed integer. */ public static final int INT1 = 3; - // 2-byte little-endian signed integer. + + /** 2-byte little-endian signed integer. */ public static final int INT2 = 4; - // 4-byte little-endian signed integer. + + /** 4-byte little-endian signed integer. */ public static final int INT4 = 5; - // 4-byte little-endian signed integer. + + /** 8-byte little-endian signed integer. */ public static final int INT8 = 6; - // 8-byte IEEE double. + + /** 8-byte IEEE double. */ public static final int DOUBLE = 7; - // 4-byte decimal. Content is 1-byte scale + 4-byte little-endian signed integer. + + /** 4-byte decimal. Content is 1-byte scale + 4-byte little-endian signed integer. */ public static final int DECIMAL4 = 8; - // 8-byte decimal. Content is 1-byte scale + 8-byte little-endian signed integer. + + /** 8-byte decimal. Content is 1-byte scale + 8-byte little-endian signed integer. */ public static final int DECIMAL8 = 9; - // 16-byte decimal. Content is 1-byte scale + 16-byte little-endian signed integer. + + /** 16-byte decimal. Content is 1-byte scale + 16-byte little-endian signed integer. */ public static final int DECIMAL16 = 10; - // Date value. Content is 4-byte little-endian signed integer that represents the number of days - // from the Unix epoch. + + /** + * Date value. Content is 4-byte little-endian signed integer that represents the number of days + * from the Unix epoch. + */ public static final int DATE = 11; - // TimestampLTZ value. Content is 8-byte little-endian signed integer that represents the number - // of microseconds elapsed since the Unix epoch, 1970-01-01 00:00:00 UTC. It is displayed to - // users in their local time zones and may be displayed differently depending on the execution - // environment. + + /** + * TimestampLTZ value. Content is 8-byte little-endian signed integer that represents the number + * of microseconds elapsed since the Unix epoch, 1970-01-01 00:00:00 UTC. It is displayed to + * users in their local time zones and may be displayed differently depending on the execution + * environment. + */ public static final int TIMESTAMP_LTZ = 12; - // Timestamp value. It has the same content as `TIMESTAMP` but should always be interpreted - // as if the local time zone is UTC. + + /** + * Timestamp value. It has the same content as {@link #TIMESTAMP_LTZ} but should always be + * interpreted as if the local time zone is UTC. + */ public static final int TIMESTAMP = 13; - // 4-byte IEEE float. + + /** 4-byte IEEE float. */ public static final int FLOAT = 14; - // Binary value. The content is (4-byte little-endian unsigned integer representing the binary - // size) + (size bytes of binary content). + + /** + * Binary value. The content is (4-byte little-endian unsigned integer representing the binary + * size) + (size bytes of binary content). + */ public static final int BINARY = 15; - // Long string value. The content is (4-byte little-endian unsigned integer representing the - // string size) + (size bytes of string content). + + /** + * 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 {@link + * #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. + + /** The lower 4 bits of the first metadata byte contain the version. */ public static final byte VERSION_MASK = 0x0F; public static final int U8_MAX = 0xFF; @@ -138,7 +212,7 @@ public class BinaryVariantUtil { public static final int U24_SIZE = 3; public static final int U32_SIZE = 4; - // Both variant value and variant metadata need to be no longer than 16MiB. + /** Both variant value and variant metadata need to be no longer than 16MiB. */ public static final int SIZE_LIMIT = U24_MAX + 1; public static final int MAX_DECIMAL4_PRECISION = 9; @@ -160,8 +234,15 @@ public class BinaryVariantUtil { .appendOffset("+HH:MM", "+00:00") .toFormatter(Locale.US); - // Write the least significant `numBytes` bytes in `value` into `bytes[pos, pos + numBytes)` in - // little endian. + public static final DateTimeFormatter TIME_FORMATTER = + new DateTimeFormatterBuilder() + .append(DateTimeFormatter.ISO_LOCAL_TIME) + .toFormatter(Locale.US); + + /** + * Write the least significant {@code numBytes} bytes in {@code value} into {@code bytes[pos, + * pos + numBytes)} in little endian. + */ public static void writeLong(byte[] bytes, int pos, long value, int numBytes) { for (int i = 0; i < numBytes; ++i) { bytes[pos + i] = (byte) ((value >>> (8 * i)) & 0xFF); @@ -191,7 +272,10 @@ public static byte arrayHeader(boolean largeSize, int offsetSize) { | ARRAY); } - // An exception indicating that the variant value or metadata doesn't + /** + * An exception indicating that the variant value or metadata doesn't conform to the variant + * format. + */ static VariantTypeException malformedVariant() { return new VariantTypeException("MALFORMED_VARIANT"); } @@ -200,24 +284,28 @@ static VariantTypeException unknownPrimitiveTypeInVariant(int id) { return new VariantTypeException("UNKNOWN_PRIMITIVE_TYPE_IN_VARIANT, id: " + id); } - // An exception indicating that an external caller tried to call the Variant constructor with - // value or metadata exceeding the 16MiB size limit. We will never construct a Variant this - // large, - // so it should only be possible to encounter this exception when reading a Variant produced by - // another tool. + /** + * An exception indicating that an external caller tried to call the Variant constructor with + * value or metadata exceeding the 16MiB size limit. We will never construct a Variant this + * large, so it should only be possible to encounter this exception when reading a Variant + * produced by another tool. + */ static VariantTypeException variantConstructorSizeLimit() { return new VariantTypeException("VARIANT_CONSTRUCTOR_SIZE_LIMIT"); } - // Check the validity of an array index `pos`. Throw `MALFORMED_VARIANT` if it is out of bound, - // meaning that the variant is malformed. + /** + * Check the validity of an array index {@code pos}. + * + * @throws VariantTypeException if {@code pos} is out of bound, meaning the variant is malformed + */ static void checkIndex(int pos, int length) { if (pos < 0 || pos >= length) { throw malformedVariant(); } } - // Read a little-endian signed long value from `bytes[pos, pos + numBytes)`. + /** Read a little-endian signed long value from {@code bytes[pos, pos + numBytes)}. */ static long readLong(byte[] bytes, int pos, int numBytes) { checkIndex(pos, bytes.length); checkIndex(pos + numBytes - 1, bytes.length); @@ -234,8 +322,10 @@ static long readLong(byte[] bytes, int pos, int numBytes) { return result; } - // Read a little-endian unsigned int value from `bytes[pos, pos + numBytes)`. The value must fit - // into a non-negative int (`[0, Integer.MAX_VALUE]`). + /** + * Read a little-endian unsigned int value from {@code bytes[pos, pos + numBytes)}. The value + * must fit into a non-negative int ({@code [0, Integer.MAX_VALUE]}). + */ static int readUnsigned(byte[] bytes, int pos, int numBytes) { checkIndex(pos, bytes.length); checkIndex(pos + numBytes - 1, bytes.length); @@ -256,10 +346,13 @@ public static int getTypeInfo(byte[] value, int pos) { return (value[pos] >> BASIC_TYPE_BITS) & TYPE_INFO_MASK; } - // Get the value type of variant value `value[pos...]`. It is only legal to call `get*` if - // `getType` returns this type (for example, it is only legal to call `getLong` if `getType` - // returns `Type.Long`). - // Throw `MALFORMED_VARIANT` if the variant is malformed. + /** + * Get the value type of variant value {@code value[pos...]}. It is only legal to call {@code + * get*} if {@code getType} returns this type (for example, it is only legal to call {@link + * #getLong(byte[], int)} if {@code getType} returns {@code Type.Long}). + * + * @throws VariantTypeException if the variant is malformed or holds an unknown primitive type + */ public static Type getType(byte[] value, int pos) { checkIndex(pos, value.length); int basicType = value[pos] & BASIC_TYPE_MASK; @@ -304,15 +397,24 @@ 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); } } } - // Compute the size in bytes of the variant value `value[pos...]`. `value.length - pos` is an - // upper bound of the size, but the actual size can be smaller. - // Throw `MALFORMED_VARIANT` if the variant is malformed. + /** + * Compute the size in bytes of the variant value {@code value[pos...]}. {@code value.length - + * pos} is an upper bound of the size, but the actual size can be smaller. + * + * @throws VariantTypeException if the variant is malformed or holds an unknown primitive type + */ public static int valueSize(byte[] value, int pos) { checkIndex(pos, value.length); int basicType = value[pos] & BASIC_TYPE_MASK; @@ -360,6 +462,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; @@ -380,8 +485,11 @@ static VariantTypeException unexpectedType(Type type) { return new VariantTypeException("Expect type to be " + type); } - // Get a boolean value from variant value `value[pos...]`. - // Throw `MALFORMED_VARIANT` if the variant is malformed. + /** + * Get a boolean value from variant value {@code value[pos...]}. + * + * @throws VariantTypeException if the variant is malformed or does not hold a boolean + */ public static boolean getBoolean(byte[] value, int pos) { checkIndex(pos, value.length); int basicType = value[pos] & BASIC_TYPE_MASK; @@ -392,17 +500,31 @@ public static boolean getBoolean(byte[] value, int pos) { return typeInfo == TRUE; } - // 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. - // If the type is `TIMESTAMP/TIMESTAMP_LTZ`, the return value represents the number of - // microseconds from the Unix epoch. + /** + * Get a long value from variant value {@code value[pos...]}. It is only legal to call it for an + * integer, date, time, or timestamp value, as detailed below. + * + *

For an integer type ({@link Type#TINYINT}, {@link Type#SMALLINT}, {@link Type#INT}, {@link + * Type#BIGINT}), the return value is simply that integer widened to a long. + * + *

If the type is {@link Type#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 {@link Type#TIME}, the return value represents the number of microseconds + * since midnight. + * + *

If the type is {@link Type#TIMESTAMP}/{@link Type#TIMESTAMP_LTZ}, the return value + * represents the number of microseconds from the Unix epoch. + * + *

If the type is {@link Type#TIMESTAMP_NS}/{@link Type#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,16 +537,22 @@ 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); } } - // Get a double value from variant value `value[pos...]`. - // Throw `MALFORMED_VARIANT` if the variant is malformed. + /** + * Get a double value from variant value {@code value[pos...]}. + * + * @throws VariantTypeException if the variant is malformed or does not hold a double + */ public static double getDouble(byte[] value, int pos) { checkIndex(pos, value.length); int basicType = value[pos] & BASIC_TYPE_MASK; @@ -435,15 +563,18 @@ public static double getDouble(byte[] value, int pos) { return Double.longBitsToDouble(readLong(value, pos + 1, 8)); } - // Check whether the precision and scale of the decimal are within the limit. + /** Check whether the precision and scale of the decimal are within the limit. */ private static void checkDecimal(BigDecimal d, int maxPrecision) { if (d.precision() > maxPrecision || d.scale() > maxPrecision) { throw malformedVariant(); } } - // Get a decimal value from variant value `value[pos...]`. - // Throw `MALFORMED_VARIANT` if the variant is malformed. + /** + * Get a decimal value from variant value {@code value[pos...]}. + * + * @throws VariantTypeException if the variant is malformed or does not hold a decimal + */ public static BigDecimal getDecimalWithOriginalScale(byte[] value, int pos) { checkIndex(pos, value.length); int basicType = value[pos] & BASIC_TYPE_MASK; @@ -485,8 +616,11 @@ public static BigDecimal getDecimal(byte[] value, int pos) { return getDecimalWithOriginalScale(value, pos).stripTrailingZeros(); } - // Get a float value from variant value `value[pos...]`. - // Throw `MALFORMED_VARIANT` if the variant is malformed. + /** + * Get a float value from variant value {@code value[pos...]}. + * + * @throws VariantTypeException if the variant is malformed or does not hold a float + */ public static float getFloat(byte[] value, int pos) { checkIndex(pos, value.length); int basicType = value[pos] & BASIC_TYPE_MASK; @@ -497,8 +631,11 @@ public static float getFloat(byte[] value, int pos) { return Float.intBitsToFloat((int) readLong(value, pos + 1, 4)); } - // Get a binary value from variant value `value[pos...]`. - // Throw `MALFORMED_VARIANT` if the variant is malformed. + /** + * Get a binary value from variant value {@code value[pos...]}. + * + * @throws VariantTypeException if the variant is malformed or does not hold a binary value + */ public static byte[] getBinary(byte[] value, int pos) { checkIndex(pos, value.length); int basicType = value[pos] & BASIC_TYPE_MASK; @@ -512,8 +649,11 @@ public static byte[] getBinary(byte[] value, int pos) { return Arrays.copyOfRange(value, start, start + length); } - // Get a string value from variant value `value[pos...]`. - // Throw `MALFORMED_VARIANT` if the variant is malformed. + /** + * Get a string value from variant value {@code value[pos...]}. + * + * @throws VariantTypeException if the variant is malformed or does not hold a string + */ public static String getString(byte[] value, int pos) { checkIndex(pos, value.length); int basicType = value[pos] & BASIC_TYPE_MASK; @@ -534,7 +674,7 @@ public static String getString(byte[] value, int pos) { throw unexpectedType(Type.STRING); } - /** 1. */ + /** A handler that receives the decoded header fields of a variant object. */ public interface ObjectHandler { /** * @param size Number of object fields. @@ -547,8 +687,10 @@ public interface ObjectHandler { T apply(int size, int idSize, int offsetSize, int idStart, int offsetStart, int dataStart); } - // A helper function to access a variant object. It provides `handler` with its required - // parameters and returns what it returns. + /** + * A helper function to access a variant object. It provides {@code handler} with its required + * parameters and returns what it returns. + */ public static T handleObject(byte[] value, int pos, ObjectHandler handler) { checkIndex(pos, value.length); int basicType = value[pos] & BASIC_TYPE_MASK; @@ -572,7 +714,7 @@ public static T handleObject(byte[] value, int pos, ObjectHandler handler return handler.apply(size, idSize, offsetSize, idStart, offsetStart, dataStart); } - /** 1. */ + /** A handler that receives the decoded header fields of a variant array. */ public interface ArrayHandler { /** * @param size Number of array elements. @@ -583,7 +725,7 @@ public interface ArrayHandler { T apply(int size, int offsetSize, int offsetStart, int dataStart); } - // A helper function to access a variant array. + /** A helper function to access a variant array. */ public static T handleArray(byte[] value, int pos, ArrayHandler handler) { checkIndex(pos, value.length); int basicType = value[pos] & BASIC_TYPE_MASK; @@ -605,9 +747,12 @@ public static T handleArray(byte[] value, int pos, ArrayHandler handler) return handler.apply(size, offsetSize, offsetStart, dataStart); } - // Get a key at `id` in the variant metadata. - // Throw `MALFORMED_VARIANT` if the variant is malformed. An out-of-bound `id` is also - // considered a malformed variant because it is read from the corresponding variant value. + /** + * Get a key at {@code id} in the variant metadata. + * + * @throws VariantTypeException if the variant is malformed. An out-of-bound {@code id} is also + * considered a malformed variant because it is read from the corresponding variant value. + */ public static String getMetadataKey(byte[] metadata, int id) { checkIndex(0, metadata.length); // Extracts the highest 2 bits in the metadata header to determine the integer size of the diff --git a/flink-core/src/main/java/org/apache/flink/types/variant/Variant.java b/flink-core/src/main/java/org/apache/flink/types/variant/Variant.java index 1ff663caaf12aa..2503698fbaf97c 100644 --- a/flink-core/src/main/java/org/apache/flink/types/variant/Variant.java +++ b/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 } diff --git a/flink-core/src/main/java/org/apache/flink/types/variant/VariantBuilder.java b/flink-core/src/main/java/org/apache/flink/types/variant/VariantBuilder.java index 550aae226a59aa..d73d0fc3c253b2 100644 --- a/flink-core/src/main/java/org/apache/flink/types/variant/VariantBuilder.java +++ b/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(); diff --git a/flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantTest.java b/flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantTest.java index 4b4a23ad34d58a..02718297fa27ea 100644 --- a/flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantTest.java +++ b/flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantTest.java @@ -30,6 +30,8 @@ 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.Collections; @@ -99,10 +101,66 @@ void testScalarVariant() { assertThat(builder.of(localDate).getDate()).isEqualTo(localDate); assertThat(builder.of(localDate).get()).isEqualTo(localDate); + LocalTime localTime = LocalTime.now().truncatedTo(ChronoUnit.MICROS); + assertThat(builder.of(localTime).getTime()).isEqualTo(localTime); + assertThat(builder.of(localTime).get()).isEqualTo(localTime); + assertThat(builder.ofNull().get()).isEqualTo(null); assertThat(builder.ofNull().isNull()).isTrue(); } + @Test + void testNanosecondPrecisionVariant() { + // Microsecond-precision values keep using the compact TIMESTAMP/TIMESTAMP_LTZ encoding, + // matching the pre-existing on-wire format. + Instant microInstant = Instant.now().truncatedTo(ChronoUnit.MICROS); + assertThat(builder.of(microInstant).getType()).isEqualTo(Variant.Type.TIMESTAMP_LTZ); + assertThat(builder.of(microInstant).getInstant()).isEqualTo(microInstant); + + LocalDateTime microLocalDateTime = LocalDateTime.now().truncatedTo(ChronoUnit.MICROS); + assertThat(builder.of(microLocalDateTime).getType()).isEqualTo(Variant.Type.TIMESTAMP); + assertThat(builder.of(microLocalDateTime).getDateTime()).isEqualTo(microLocalDateTime); + + // Sub-microsecond precision must switch to the nanosecond-precision encoding rather than + // silently truncating. + Instant nanoInstant = Instant.now().truncatedTo(ChronoUnit.NANOS).plusNanos(123); + Variant instantVariant = builder.of(nanoInstant); + assertThat(instantVariant.getType()).isEqualTo(Variant.Type.TIMESTAMP_LTZ_NS); + assertThat(instantVariant.getInstantNanos()).isEqualTo(nanoInstant); + assertThat(instantVariant.get()).isEqualTo(nanoInstant); + assertThatThrownBy(instantVariant::getInstant).isInstanceOf(VariantTypeException.class); + + LocalDateTime nanoLocalDateTime = LocalDateTime.now().withNano(123456789); + Variant dateTimeVariant = builder.of(nanoLocalDateTime); + assertThat(dateTimeVariant.getType()).isEqualTo(Variant.Type.TIMESTAMP_NS); + assertThat(dateTimeVariant.getDateTimeNanos()).isEqualTo(nanoLocalDateTime); + assertThat(dateTimeVariant.get()).isEqualTo(nanoLocalDateTime); + assertThatThrownBy(dateTimeVariant::getDateTime).isInstanceOf(VariantTypeException.class); + } + + @Test + void testNanosecondPrecisionOutOfRange() { + // Nanosecond timestamps only span +/-292 years around 1970, beyond that must fail with + // proper exception + LocalDateTime outOfRangeLocalDateTime = LocalDateTime.of(2300, 1, 1, 0, 0, 0, 1); + assertThatThrownBy(() -> builder.of(outOfRangeLocalDateTime)) + .isInstanceOf(VariantTypeException.class) + .hasMessageContaining("nanosecond precision"); + + Instant outOfRangeInstant = outOfRangeLocalDateTime.toInstant(ZoneOffset.UTC); + assertThatThrownBy(() -> builder.of(outOfRangeInstant)) + .isInstanceOf(VariantTypeException.class) + .hasMessageContaining("nanosecond precision"); + } + + @Test + void testTimeSubMicrosecondTruncation() { + // A sub-microsecond LocalTime silently loses precision below the microsecond. + // TIME has no nanosecond-precision counterpart in the variant spec. + LocalTime nanoTime = LocalTime.of(23, 59, 59, 999999999); + assertThat(builder.of(nanoTime).getTime()).isEqualTo(LocalTime.of(23, 59, 59, 999999000)); + } + @Test void testArrayVariant() { Instant now = Instant.now().truncatedTo(ChronoUnit.MICROS); @@ -205,6 +263,9 @@ void testToJsonScalar() { Instant instant = Instant.EPOCH; LocalDateTime localDateTime = LocalDateTime.of(2000, 1, 1, 0, 0); LocalDate localDate = LocalDate.of(2000, 1, 1); + LocalTime localTime = LocalTime.of(13, 45, 30, 123456789); + Instant nanoInstant = Instant.EPOCH.plusNanos(123456789); + LocalDateTime nanoLocalDateTime = LocalDateTime.of(2000, 1, 1, 0, 0, 0, 123456789); assertThat(builder.of((byte) 1).toJson()).isEqualTo("1"); assertThat(builder.of((short) 1).toJson()).isEqualTo("1"); @@ -218,6 +279,11 @@ void testToJsonScalar() { assertThat(builder.of(instant).toJson()).isEqualTo("\"1970-01-01T00:00:00+00:00\""); assertThat(builder.of(localDateTime).toJson()).isEqualTo("\"2000-01-01T00:00:00\""); assertThat(builder.of(localDate).toJson()).isEqualTo("\"2000-01-01\""); + assertThat(builder.of(localTime).toJson()).isEqualTo("\"13:45:30.123456\""); + assertThat(builder.of(nanoInstant).toJson()) + .isEqualTo("\"1970-01-01T00:00:00.123456789+00:00\""); + assertThat(builder.of(nanoLocalDateTime).toJson()) + .isEqualTo("\"2000-01-01T00:00:00.123456789\""); assertThat(builder.of("hello".getBytes()).toJson()).isEqualTo("\"aGVsbG8=\""); assertThat(builder.ofNull().toJson()).isEqualTo("null"); }