Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
manner wants to merge 7 commits into
base: master
Choose a base branch
from
+498 −107
Open

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

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
5b4a097
[FLINK-40491][core] Extend BinaryVariant with TIME, TIMESTAMP_NS, TIM…
manner Aug 31, 2026
c7dc74a
[FLINK-40491][core] Fix formatting
manner Aug 31, 2026
ffd3b20
[FLINK-40491][core] Add comment about truncating LocalTime
manner Aug 31, 2026
058498a
[FLINK-40491][core] Reject out-of-range nanosecond-precision timestamps
manner Aug 31, 2026
19c6ec8
[FLINK-40491][core] Add nano precision to Time to check truncation
manner Sep 1, 2026
5eea899
[FLINK-40491][core] Refactor comments to Javadoc and fix inconsistencies
manner Sep 1, 2026
758db23
[FLINK-40491][core] Fix microsecond timestamp overflow
manner Sep 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 53 additions & 0 deletions flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariant.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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());
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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(
Expand Down
35 changes: 32 additions & 3 deletions flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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();
}

Expand All @@ -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();
}

Expand Down
24 changes: 24 additions & 0 deletions flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantInternalBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand Down
Loading