This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 26c1062dfc4 [FLINK-40491][core] Extend `BinaryVariant` with `TIME`,
`TIMESTAMP_NS`, `TIMESTAMP_LTZ_NS` primitives
26c1062dfc4 is described below
commit 26c1062dfc464c86267f5d47a61af8c8bbd97847
Author: Moritz Manner <[email protected]>
AuthorDate: Thu Sep 3 15:22:29 2026 +0200
[FLINK-40491][core] Extend `BinaryVariant` with `TIME`, `TIMESTAMP_NS`,
`TIMESTAMP_LTZ_NS` primitives
---
.../apache/flink/types/variant/BinaryVariant.java | 66 +++-
.../flink/types/variant/BinaryVariantBuilder.java | 49 ++-
.../variant/BinaryVariantInternalBuilder.java | 24 ++
.../flink/types/variant/BinaryVariantUtil.java | 353 +++++++++++++++------
.../org/apache/flink/types/variant/Variant.java | 23 +-
.../apache/flink/types/variant/VariantBuilder.java | 4 +
.../flink/types/variant/BinaryVariantTest.java | 89 ++++++
7 files changed, 491 insertions(+), 117 deletions(-)
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 97c11eedf52..6cccd45d524 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.math.BigDecimal;
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.BINARY_SEARCH_THR
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;
@@ -181,16 +183,39 @@ public final class BinaryVariant implements Variant {
@Override
public LocalDateTime getDateTime() throws VariantTypeException {
- checkType(Type.TIMESTAMP, getType());
- return microsToInstant(BinaryVariantUtil.getLong(value, pos))
- .atZone(ZoneOffset.UTC)
- .toLocalDateTime();
+ Type type = getType();
+ Instant instant;
+ if (type == Type.TIMESTAMP) {
+ instant = microsToInstant(BinaryVariantUtil.getLong(value, pos));
+ } else if (type == Type.TIMESTAMP_NS) {
+ instant = nanosToInstant(BinaryVariantUtil.getLong(value, pos));
+ } else {
+ throw new VariantTypeException(
+ String.format(
+ "Expected type %s or %s but got %s",
+ Type.TIMESTAMP, Type.TIMESTAMP_NS, type));
+ }
+ return instant.atZone(ZoneOffset.UTC).toLocalDateTime();
}
@Override
public Instant getInstant() throws VariantTypeException {
- checkType(Type.TIMESTAMP_LTZ, getType());
- return microsToInstant(BinaryVariantUtil.getLong(value, pos));
+ Type type = getType();
+ if (type == Type.TIMESTAMP_LTZ) {
+ return microsToInstant(BinaryVariantUtil.getLong(value, pos));
+ } else if (type == Type.TIMESTAMP_LTZ_NS) {
+ return nanosToInstant(BinaryVariantUtil.getLong(value, pos));
+ }
+ throw new VariantTypeException(
+ String.format(
+ "Expected type %s or %s but got %s",
+ Type.TIMESTAMP_LTZ, Type.TIMESTAMP_LTZ_NS, type));
+ }
+
+ @Override
+ public LocalTime getTime() throws VariantTypeException {
+ checkType(Type.TIME, getType());
+ return LocalTime.ofNanoOfDay(BinaryVariantUtil.getLong(value, pos) *
1000);
}
@Override
@@ -224,9 +249,13 @@ public final class BinaryVariant implements Variant {
return getString();
case DATE:
return getDate();
+ case TIME:
+ return getTime();
case TIMESTAMP:
+ case TIMESTAMP_NS:
return getDateTime();
case TIMESTAMP_LTZ:
+ case TIMESTAMP_LTZ_NS:
return getInstant();
case BYTES:
return getBytes();
@@ -391,6 +420,27 @@ public final class BinaryVariant implements Variant {
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 +468,10 @@ public final class BinaryVariant implements Variant {
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 4b7cd0558e4..c2b9ef59faf 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.math.BigDecimal;
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 class BinaryVariantBuilder implements
VariantBuilder {
@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(microsSinceEpoch(instant));
+ } else {
+ builder.appendTimestampLtzNanos(nanosSinceEpoch(instant));
+ }
return builder.build();
}
@@ -120,8 +125,46 @@ public class BinaryVariantBuilder implements
VariantBuilder {
@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(microsSinceEpoch(instant));
+ } else {
+ builder.appendTimestampNanos(nanosSinceEpoch(instant));
+ }
+ return builder.build();
+ }
+
+ private static long microsSinceEpoch(Instant instant) {
+ try {
+ return Math.addExact(
+ Math.multiplyExact(instant.getEpochSecond(), 1_000_000L),
+ instant.getNano() / 1000);
+ } catch (ArithmeticException e) {
+ throw new VariantTypeException(
+ String.format(
+ "%s is outside the range supported by microsecond
precision variant "
+ + "timestamps.",
+ instant));
+ }
+ }
+
+ 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 1674d166266..5b0cadddc3c 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.MAX_SHORT_STR_SIZ
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 class BinaryVariantInternalBuilder {
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 c3ab8d29be8..8bf7ef61fab 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 @@ import java.util.Locale;
* the below constants for all possible basic type and type info values.
*
* <p>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:
+ *
+ * <ul>
+ * <li>Version: 1-byte unsigned integer. The only acceptable value is 1
currently.
+ * <li>Dictionary size: 4-byte little-endian unsigned integer. The number of
keys in the
+ * dictionary.
+ * <li>Offsets: (size + 1) * 4-byte little-endian unsigned integers. {@code
offsets[i]} represents
+ * the starting position of string i, counting starting from the address
of {@code
+ * offsets[0]}. Strings must be stored contiguously, so we don’t need to
store the string
+ * size, instead, we compute it with {@code offset[i + 1] - offset[i]}.
+ * <li>UTF-8 string data.
+ * </ul>
*/
@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.
+ *
+ * <p>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:
+ *
+ * <ul>
+ * <li>b4 specifies the type of size. When it is 0/1, {@code size} is a
little-endian 1/4-byte
+ * unsigned integer.
+ * <li>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.
+ * </ul>
+ */
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.
+ *
+ * <p>Its type info should be: 000_b2_b1b0:
+ *
+ * <ul>
+ * <li>b2 specifies the type of size.
+ * <li>b1b0 specifies the integer type of offset.
+ * </ul>
+ */
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 class BinaryVariantUtil {
| 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 @@ public class BinaryVariantUtil {
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 @@ public class BinaryVariantUtil {
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 class BinaryVariantUtil {
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 class BinaryVariantUtil {
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 class BinaryVariantUtil {
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 @@ public class BinaryVariantUtil {
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 class BinaryVariantUtil {
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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>If the type is {@link Type#TIME}, the return value represents the
number of microseconds
+ * since midnight.
+ *
+ * <p>If the type is {@link Type#TIMESTAMP}/{@link Type#TIMESTAMP_LTZ},
the return value
+ * represents the number of microseconds from the Unix epoch.
+ *
+ * <p>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 class BinaryVariantUtil {
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 class BinaryVariantUtil {
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 class BinaryVariantUtil {
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 class BinaryVariantUtil {
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 class BinaryVariantUtil {
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 class BinaryVariantUtil {
throw unexpectedType(Type.STRING);
}
- /** 1. */
+ /** A handler that receives the decoded header fields of a variant object.
*/
public interface ObjectHandler<T> {
/**
* @param size Number of object fields.
@@ -547,8 +687,10 @@ public class BinaryVariantUtil {
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> T handleObject(byte[] value, int pos, ObjectHandler<T>
handler) {
checkIndex(pos, value.length);
int basicType = value[pos] & BASIC_TYPE_MASK;
@@ -572,7 +714,7 @@ public class BinaryVariantUtil {
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<T> {
/**
* @param size Number of array elements.
@@ -583,7 +725,7 @@ public class BinaryVariantUtil {
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> T handleArray(byte[] value, int pos, ArrayHandler<T>
handler) {
checkIndex(pos, value.length);
int basicType = value[pos] & BASIC_TYPE_MASK;
@@ -605,9 +747,12 @@ public class BinaryVariantUtil {
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 1ff663caaf1..cad26314ad3 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.math.BigDecimal;
import java.time.Instant;
import java.time.LocalDate;
import java.time.LocalDateTime;
+import java.time.LocalTime;
import java.util.List;
/**
@@ -135,22 +136,33 @@ public interface Variant extends Serializable {
/**
* Get the scalar value of variant as {@link LocalDateTime}, if the
variant type is {@link
- * Type#TIMESTAMP}. The returned value has microsecond precision.
+ * Type#TIMESTAMP} or {@link Type#TIMESTAMP_NS}. The returned value has
microsecond or
+ * nanosecond precision, matching the variant's actual type.
*
* @throws VariantTypeException If this variant is not a scalar value or
is not {@link
- * Type#TIMESTAMP}.
+ * Type#TIMESTAMP} or {@link Type#TIMESTAMP_NS}.
*/
LocalDateTime getDateTime() throws VariantTypeException;
/**
* Get the scalar value of variant as {@link Instant}, if the variant type
is {@link
- * Type#TIMESTAMP_LTZ}. The returned value has microsecond precision.
+ * Type#TIMESTAMP_LTZ} or {@link Type#TIMESTAMP_LTZ_NS}. The returned
value has microsecond or
+ * nanosecond precision, matching the variant's actual type.
*
* @throws VariantTypeException If this variant is not a scalar value or
is not {@link
- * Type#TIMESTAMP_LTZ}.
+ * Type#TIMESTAMP_LTZ} or {@link Type#TIMESTAMP_LTZ_NS}.
*/
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 byte array, if the variant type is
{@link Type#BYTES}.
*
@@ -232,8 +244,11 @@ public interface Variant extends Serializable {
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 550aae226a5..d73d0fc3c25 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.math.BigDecimal;
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 4b4a23ad34d..76c5222c168 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.nio.charset.StandardCharsets;
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,89 @@ class BinaryVariantTest {
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 switches to the nanosecond encoding
instead of truncating,
+ // but getInstant()/getDateTime() still work regardless of which
encoding was picked.
+ 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.getInstant()).isEqualTo(nanoInstant);
+ assertThat(instantVariant.get()).isEqualTo(nanoInstant);
+
+ LocalDateTime nanoLocalDateTime =
LocalDateTime.now().withNano(123456789);
+ Variant dateTimeVariant = builder.of(nanoLocalDateTime);
+
assertThat(dateTimeVariant.getType()).isEqualTo(Variant.Type.TIMESTAMP_NS);
+ assertThat(dateTimeVariant.getDateTime()).isEqualTo(nanoLocalDateTime);
+ assertThat(dateTimeVariant.get()).isEqualTo(nanoLocalDateTime);
+ }
+
+ @Test
+ void testNanosecondTimestampPrecisionRange() {
+ // 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 testMicrosecondTimestampPrecisionRange() {
+ // Microsecond precision spans = +/-292.000 years around 1970
+ LocalDateTime inRangeLocalDateTime = LocalDateTime.of(150_000, 1, 1,
0, 0, 0, 0);
+ Variant inRangeLocalDateTimeVariant = builder.of(inRangeLocalDateTime);
+
assertThat(inRangeLocalDateTimeVariant.getType()).isEqualTo(Variant.Type.TIMESTAMP);
+
assertThat(inRangeLocalDateTimeVariant.getDateTime()).isEqualTo(inRangeLocalDateTime);
+
+ Instant inRangeInstant =
inRangeLocalDateTime.toInstant(ZoneOffset.UTC);
+ Variant inRangeInstantVariant = builder.of(inRangeInstant);
+
assertThat(inRangeInstantVariant.getType()).isEqualTo(Variant.Type.TIMESTAMP_LTZ);
+
assertThat(inRangeInstantVariant.getInstant()).isEqualTo(inRangeInstant);
+
+ // beyond that must fail with proper exception
+ LocalDateTime outOfRangeLocalDateTime = LocalDateTime.of(350_000, 1,
1, 0, 0, 0, 0);
+ assertThatThrownBy(() -> builder.of(outOfRangeLocalDateTime))
+ .isInstanceOf(VariantTypeException.class)
+ .hasMessageContaining("microsecond precision");
+
+ Instant outOfRangeInstant =
outOfRangeLocalDateTime.toInstant(ZoneOffset.UTC);
+ assertThatThrownBy(() -> builder.of(outOfRangeInstant))
+ .isInstanceOf(VariantTypeException.class)
+ .hasMessageContaining("microsecond 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 +286,9 @@ class BinaryVariantTest {
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 +302,11 @@ class BinaryVariantTest {
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");
}
