From 0eac3c358e3c8b1f78c068b62353f21b8c370c14 Mon Sep 17 00:00:00 2001 From: Moritz Manner Date: Wed, 2 Sep 2026 16:23:01 +0200 Subject: [PATCH 1/6] [FLINK-40492][docs] Fix VARIANT to CHARACTER_STRING entry in the cast matrix --- docs/content.zh/docs/sql/reference/data-types.md | 2 +- docs/content/docs/sql/reference/data-types.md | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/content.zh/docs/sql/reference/data-types.md b/docs/content.zh/docs/sql/reference/data-types.md index 9bc67dd733e92..3967bb0df6728 100644 --- a/docs/content.zh/docs/sql/reference/data-types.md +++ b/docs/content.zh/docs/sql/reference/data-types.md @@ -1873,7 +1873,7 @@ COALESCE(TRY_CAST('non-number' AS INT), 0) --- 结果返回数字 0 的 INT 格 | `ROW` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N | N | | `STRUCTURED` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N | | `RAW` | Y | ! | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | Y⁴ | N | N | -| `VARIANT` | N | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | N | ! | ! | N | !³ | N | !³ | !³ | !³ | N | Y | N | +| `VARIANT` | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | N | ! | ! | N | !³ | N | !³ | !³ | !³ | N | Y | N | | `BITMAP` | Y | Y⁷ | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | 备注： diff --git a/docs/content/docs/sql/reference/data-types.md b/docs/content/docs/sql/reference/data-types.md index 8e43c2ec12ee2..80540c583379b 100644 --- a/docs/content/docs/sql/reference/data-types.md +++ b/docs/content/docs/sql/reference/data-types.md @@ -1882,7 +1882,7 @@ The matrix below describes the supported cast pairs, where "Y" means supported, | `ROW` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N | N | | `STRUCTURED` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N | | `RAW` | Y | ! | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | Y⁴ | N | N | -| `VARIANT` | N | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | N | ! | ! | N | !³ | N | !³ | !³ | !³ | N | Y | N | +| `VARIANT` | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | N | ! | ! | N | !³ | N | !³ | !³ | !³ | N | Y | N | | `BITMAP` | Y | Y⁷ | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | Notes: From e03946b24ead7f7152d864b43bf0d1b1b1a44040 Mon Sep 17 00:00:00 2001 From: Moritz Manner Date: Wed, 2 Sep 2026 17:17:16 +0200 Subject: [PATCH 2/6] [FLINK-40492][table] Support casting from VARIANT to TIME and nanosecond timestamps --- .../docs/sql/reference/data-types.md | 2 +- docs/content/docs/sql/reference/data-types.md | 2 +- .../types/logical/utils/LogicalTypeCasts.java | 1 + .../table/types/LogicalTypeCastsTest.java | 3 +- .../casting/VariantToPrimitiveCastRule.java | 13 ++- .../casting/CastRuleProviderTest.java | 3 +- .../functions/casting/CastRulesTest.java | 82 ++++++++++++++++++- .../runtime/functions/VariantCastUtils.java | 60 ++++++++++++-- 8 files changed, 149 insertions(+), 17 deletions(-) diff --git a/docs/content.zh/docs/sql/reference/data-types.md b/docs/content.zh/docs/sql/reference/data-types.md index 3967bb0df6728..81aec696f8dec 100644 --- a/docs/content.zh/docs/sql/reference/data-types.md +++ b/docs/content.zh/docs/sql/reference/data-types.md @@ -1873,7 +1873,7 @@ COALESCE(TRY_CAST('non-number' AS INT), 0) --- 结果返回数字 0 的 INT 格 | `ROW` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N | N | | `STRUCTURED` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N | | `RAW` | Y | ! | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | Y⁴ | N | N | -| `VARIANT` | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | N | ! | ! | N | !³ | N | !³ | !³ | !³ | N | Y | N | +| `VARIANT` | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | N | !³ | N | !³ | !³ | !³ | N | Y | N | | `BITMAP` | Y | Y⁷ | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | 备注： diff --git a/docs/content/docs/sql/reference/data-types.md b/docs/content/docs/sql/reference/data-types.md index 80540c583379b..5ff9e0d8100d5 100644 --- a/docs/content/docs/sql/reference/data-types.md +++ b/docs/content/docs/sql/reference/data-types.md @@ -1882,7 +1882,7 @@ The matrix below describes the supported cast pairs, where "Y" means supported, | `ROW` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N | N | | `STRUCTURED` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N | | `RAW` | Y | ! | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | Y⁴ | N | N | -| `VARIANT` | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | N | ! | ! | N | !³ | N | !³ | !³ | !³ | N | Y | N | +| `VARIANT` | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | N | !³ | N | !³ | !³ | !³ | N | Y | N | | `BITMAP` | Y | Y⁷ | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | Notes: diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeCasts.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeCasts.java index 474e63b36709e..e213c47716435 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeCasts.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeCasts.java @@ -359,6 +359,7 @@ public final class LogicalTypeCasts { castTo(TIME_WITHOUT_TIME_ZONE) .implicitFrom(TIME_WITHOUT_TIME_ZONE, TIMESTAMP_WITHOUT_TIME_ZONE) .explicitFromFamily(TIME, TIMESTAMP, CHARACTER_STRING) + .explicitFrom(VARIANT) .injectiveFrom(WHEN_PRECISION_MATCHES, TIME_WITHOUT_TIME_ZONE) .build(); diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeCastsTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeCastsTest.java index f6022d3f2cb20..f696227034e38 100644 --- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeCastsTest.java +++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeCastsTest.java @@ -289,10 +289,9 @@ private static Stream testData() { true), Arguments.of(new VariantType(), new CharType(), false, true), Arguments.of(new VariantType(), VarCharType.STRING_TYPE, false, true), + Arguments.of(new VariantType(), new TimeType(), false, true), // variant identity cast is implicit Arguments.of(new VariantType(), new VariantType(), true, true), - // TIME has no variant counterpart, so it is not castable from variant - Arguments.of(new VariantType(), new TimeType(), false, false), // A variant imposes a schema on a constructed target, explicit only, recursing on // every leaf, which is itself a VARIANT cast Arguments.of(new VariantType(), new ArrayType(new IntType()), false, true), diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/VariantToPrimitiveCastRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/VariantToPrimitiveCastRule.java index 9b42628498f9d..4a16d7fce8189 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/VariantToPrimitiveCastRule.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/VariantToPrimitiveCastRule.java @@ -43,8 +43,7 @@ * the kind itself is not implicit, so a decimal is not read as an integer and a {@code TIMESTAMP} * is not read as a {@code TIMESTAMP_LTZ}. * - *

{@code CHARACTER_STRING} is handled by {@link VariantToStringCastRule}; {@code TIME} has no - * variant counterpart and is unsupported. + *

{@code CHARACTER_STRING} is handled by {@link VariantToStringCastRule}. */ class VariantToPrimitiveCastRule extends AbstractNullAwareCodeGeneratorCastRule { @@ -73,6 +72,7 @@ private static boolean isSupportedTarget(LogicalType targetType) { case BINARY: case VARBINARY: case DATE: + case TIME_WITHOUT_TIME_ZONE: case TIMESTAMP_WITHOUT_TIME_ZONE: case TIMESTAMP_WITH_LOCAL_TIME_ZONE: return true; @@ -174,6 +174,15 @@ protected String generateCodeBlockInternal( returnVariable, cast("int", methodCall(methodCall(inputTerm, "getDate"), "toEpochDay"))); break; + case TIME_WITHOUT_TIME_ZONE: + writer.assignStmt( + returnVariable, + staticCall( + VariantCastUtils.class, + "toTime", + inputTerm, + LogicalTypeChecks.getPrecision(targetLogicalType))); + break; case TIMESTAMP_WITHOUT_TIME_ZONE: writer.assignStmt( returnVariable, diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRuleProviderTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRuleProviderTest.java index a743cf9ee5527..1fe1a23786091 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRuleProviderTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRuleProviderTest.java @@ -132,11 +132,10 @@ void testResolveVariantToPrimitive() { assertThat(CastRuleProvider.exists(VARIANT, DATE().getLogicalType())).isTrue(); assertThat(CastRuleProvider.exists(VARIANT, TIMESTAMP().getLogicalType())).isTrue(); assertThat(CastRuleProvider.exists(VARIANT, TIMESTAMP_LTZ().getLogicalType())).isTrue(); + assertThat(CastRuleProvider.exists(VARIANT, TIME().getLogicalType())).isTrue(); assertThat(CastRuleProvider.exists(VARIANT, BYTES().getLogicalType())).isTrue(); assertThat(CastRuleProvider.canFail(VARIANT, INT)).isTrue(); - // TIME has no variant counterpart and is not castable - assertThat(CastRuleProvider.exists(VARIANT, TIME().getLogicalType())).isFalse(); // character strings keep going through the display-oriented rule assertThat(CastRuleProvider.resolve(VARIANT, STRING_TYPE)) .isSameAs(VariantToStringCastRule.INSTANCE); diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java index 274626557d172..52ccc2a2d02b3 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java @@ -1831,6 +1831,24 @@ Stream testCases() { CET_CONTEXT, VARIANT_BUILDER.of(TIMESTAMP.toInstant()), TIMESTAMP_STRING_CET) + // a time renders at millisecond resolution, the same as a regular TIME to + // string cast, so the stored microseconds are truncated + .fromCase( + VARIANT(), + Variant.newBuilder().of(LocalTime.of(12, 34, 56, 123_456_000)), + fromString("12:34:56.123")) + // a nanosecond timestamp keeps its full precision when rendered + .fromCase( + VARIANT(), + Variant.newBuilder() + .of(LocalDateTime.of(2021, 9, 24, 12, 34, 56, 123_456_789)), + fromString("2021-09-24 12:34:56.123456789")) + .fromCase( + VARIANT(), + CET_CONTEXT, + Variant.newBuilder() + .of(Instant.parse("2021-09-24T12:34:56.123456789Z")), + fromString("2021-09-24 14:34:56.123456789")) // a binary value is read as UTF-8, like a regular BINARY to string cast .fromCase( VARIANT(), @@ -1985,7 +2003,26 @@ Stream testCases() { VARIANT(), VARIANT_BUILDER.of(LocalDate.of(2020, 1, 1)), (int) LocalDate.of(2020, 1, 1).toEpochDay()) - .fail(VARIANT(), VARIANT_BUILDER.of(1), TableRuntimeException.class), + .fail(VARIANT(), Variant.newBuilder().of(1), TableRuntimeException.class), + // A variant keeps microseconds for TIME, so fractional seconds beyond the target + // precision are truncated, matching a regular cast into a narrower TIME. + CastTestSpecBuilder.testCastTo(TIME(3)) + .fromCase( + VARIANT(), + Variant.newBuilder().of(LocalTime.of(12, 0, 0, 123_000_000)), + DateTimeUtils.toInternal(LocalTime.of(12, 0, 0, 123_000_000))) + .fromCase( + VARIANT(), + Variant.newBuilder().of(LocalTime.of(12, 0, 0, 123_456_000)), + DateTimeUtils.toInternal(LocalTime.of(12, 0, 0, 123_000_000))) + .fail(VARIANT(), Variant.newBuilder().of(1), TableRuntimeException.class), + // TIME has no runtime representation finer than milliseconds, so a target + // precision above 3 truncates no further than TIME(3) already does. + CastTestSpecBuilder.testCastTo(TIME(6)) + .fromCase( + VARIANT(), + Variant.newBuilder().of(LocalTime.of(12, 0, 0, 123_456_789)), + DateTimeUtils.toInternal(LocalTime.of(12, 0, 0, 123_000_000))), CastTestSpecBuilder.testCastTo(TIMESTAMP()) .fromCase(VARIANT(), null, null) .fromCase( @@ -2026,6 +2063,27 @@ Stream testCases() { LocalDateTime.of(2020, 1, 1, 12, 0, 0, 123000000)), TimestampData.fromLocalDateTime( LocalDateTime.of(2020, 1, 1, 12, 0, 0))), + // The cast accepts either storage kind: TIMESTAMP_NS for a value that needs + // nanosecond precision, plain TIMESTAMP when microseconds already hold it exactly. + CastTestSpecBuilder.testCastTo(TIMESTAMP(9)) + .fromCase( + VARIANT(), + Variant.newBuilder() + .of(LocalDateTime.of(2020, 1, 1, 12, 0, 0, 123456789)), + TimestampData.fromLocalDateTime( + LocalDateTime.of(2020, 1, 1, 12, 0, 0, 123456789))) + .fromCase( + VARIANT(), + Variant.newBuilder() + .of(LocalDateTime.of(2020, 1, 1, 12, 0, 0, 123456000)), + TimestampData.fromLocalDateTime( + LocalDateTime.of(2020, 1, 1, 12, 0, 0, 123456000))) + // a TIMESTAMP_LTZ_NS is a different kind and is not read as a TIMESTAMP + .fail( + VARIANT(), + Variant.newBuilder() + .of(Instant.ofEpochSecond(1_600_000_000L, 123456789)), + TableRuntimeException.class), CastTestSpecBuilder.testCastTo(TIMESTAMP_LTZ()) .fromCase( VARIANT(), @@ -2044,6 +2102,28 @@ Stream testCases() { Instant.ofEpochSecond(1_600_000_000L, 123456000)), TimestampData.fromInstant( Instant.ofEpochSecond(1_600_000_000L, 123000000))), + // The cast accepts either storage kind: TIMESTAMP_LTZ_NS for a value that needs + // nanosecond precision, plain TIMESTAMP_LTZ when microseconds already hold it + // exactly. + CastTestSpecBuilder.testCastTo(TIMESTAMP_LTZ(9)) + .fromCase( + VARIANT(), + Variant.newBuilder() + .of(Instant.ofEpochSecond(1_600_000_000L, 123456789)), + TimestampData.fromInstant( + Instant.ofEpochSecond(1_600_000_000L, 123456789))) + .fromCase( + VARIANT(), + Variant.newBuilder() + .of(Instant.ofEpochSecond(1_600_000_000L, 123456000)), + TimestampData.fromInstant( + Instant.ofEpochSecond(1_600_000_000L, 123456000))) + // a TIMESTAMP_NS is a different kind and is not read as a TIMESTAMP_LTZ + .fail( + VARIANT(), + Variant.newBuilder() + .of(LocalDateTime.of(2020, 1, 1, 12, 0, 0, 123456789)), + TableRuntimeException.class), // A binary target pads a shorter value and truncates a longer one, matching a // regular cast into the same type. CastTestSpecBuilder.testCastTo(BINARY(4)) diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java index f4920a4d3532c..be3001edff183 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java @@ -58,9 +58,18 @@ public final class VariantCastUtils { */ private static final double LONG_MAGNITUDE_LIMIT = -(double) Long.MIN_VALUE; - /** A variant stores a timestamp with microsecond precision. */ + /** A microsecond timestamp variant renders with six fractional-second digits. */ private static final int TIMESTAMP_PRECISION = 6; + /** A nanosecond timestamp variant renders with nine fractional-second digits. */ + private static final int TIMESTAMP_NANOS_PRECISION = 9; + + /** + * A time variant keeps microseconds, but the runtime TIME representation is millisecond-of-day, + * so it renders with three fractional-second digits, the same as a regular TIME to string cast. + */ + private static final int TIME_PRECISION = 3; + private VariantCastUtils() {} /** @@ -229,12 +238,14 @@ private static String decimalTarget(int precision, int scale) { } /** - * Reads a timestamp variant as the target {@code TIMESTAMP}. A variant keeps microseconds, so - * fractional seconds beyond the target precision are truncated, the same as a regular {@code - * TIMESTAMP} to {@code TIMESTAMP(p)} cast. + * Reads a timestamp variant as the target {@code TIMESTAMP}. {@link Variant#getDateTime()} + * already accepts both the microsecond ({@link Variant.Type#TIMESTAMP}) and nanosecond ({@link + * Variant.Type#TIMESTAMP_NS}) encodings. Fractional seconds beyond the target precision are + * truncated, the same as a regular {@code TIMESTAMP} to {@code TIMESTAMP(p)} cast. */ public static TimestampData toTimestamp(Variant variant, int precision) { - if (variant.getType() != Variant.Type.TIMESTAMP) { + final Variant.Type type = variant.getType(); + if (type != Variant.Type.TIMESTAMP && type != Variant.Type.TIMESTAMP_NS) { throw unsupportedKind(variant, String.format("TIMESTAMP(%d)", precision)); } return DateTimeUtils.truncate( @@ -243,12 +254,27 @@ public static TimestampData toTimestamp(Variant variant, int precision) { /** Reads a timestamp with local time zone variant. See {@link #toTimestamp(Variant, int)}. */ public static TimestampData toTimestampLtz(Variant variant, int precision) { - if (variant.getType() != Variant.Type.TIMESTAMP_LTZ) { + final Variant.Type type = variant.getType(); + if (type != Variant.Type.TIMESTAMP_LTZ && type != Variant.Type.TIMESTAMP_LTZ_NS) { throw unsupportedKind(variant, String.format("TIMESTAMP_LTZ(%d)", precision)); } return DateTimeUtils.truncate(TimestampData.fromInstant(variant.getInstant()), precision); } + /** + * Reads a time variant as the target {@code TIME}. The runtime TIME representation is + * millisecond-of-day, so a variant's microseconds are dropped and any fractional seconds beyond + * the target precision are then truncated, the same as a regular {@code TIME} to {@code + * TIME(p)} cast. + */ + public static int toTime(Variant variant, int precision) { + if (variant.getType() != Variant.Type.TIME) { + throw unsupportedKind(variant, String.format("TIME(%d)", precision)); + } + return DateTimeUtils.applyTimePrecisionTruncation( + DateTimeUtils.toInternal(variant.getTime()), precision); + } + /** * Reads a binary variant as the target binary type. {@code BINARY} is fixed width, so a shorter * value is padded with zero bytes, and either target truncates a value longer than {@code @@ -340,15 +366,26 @@ private static String getVariantTypeAsString( case DATE: value = DateTimeUtils.formatDate((int) variant.getDate().toEpochDay()); break; + case TIME: + value = + DateTimeUtils.formatTimestampMillis( + DateTimeUtils.toInternal(variant.getTime()), TIME_PRECISION); + break; case TIMESTAMP: - // A wall-clock value needs no zone shift, which is what UTC_ZONE achieves here. A - // variant keeps microseconds, so the precision is always 6. + // A wall-clock value needs no zone shift, which is what UTC_ZONE achieves here. value = DateTimeUtils.formatTimestamp( TimestampData.fromLocalDateTime(variant.getDateTime()), DateTimeUtils.UTC_ZONE, TIMESTAMP_PRECISION); break; + case TIMESTAMP_NS: + value = + DateTimeUtils.formatTimestamp( + TimestampData.fromLocalDateTime(variant.getDateTime()), + DateTimeUtils.UTC_ZONE, + TIMESTAMP_NANOS_PRECISION); + break; case TIMESTAMP_LTZ: value = DateTimeUtils.formatTimestamp( @@ -356,6 +393,13 @@ private static String getVariantTypeAsString( sessionZone, TIMESTAMP_PRECISION); break; + case TIMESTAMP_LTZ_NS: + value = + DateTimeUtils.formatTimestamp( + TimestampData.fromInstant(variant.getInstant()), + sessionZone, + TIMESTAMP_NANOS_PRECISION); + break; case NULL: // Only reachable for a NOT NULL target. A nullable target maps a null-valued // variant to SQL NULL before this method is called. From e1fa3a9ba672f4e2a006430d39fe3664fa544379 Mon Sep 17 00:00:00 2001 From: Moritz Manner Date: Thu, 3 Sep 2026 16:38:09 +0200 Subject: [PATCH 3/6] [FLINK-40492][table] Fix timestamp truncation for fractions with leading zeros --- .../java/org/apache/flink/table/utils/DateTimeUtils.java | 3 ++- .../table/planner/functions/casting/CastRulesTest.java | 8 ++++++++ 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/DateTimeUtils.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/DateTimeUtils.java index 8a161873e21f2..245c707767154 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/DateTimeUtils.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/DateTimeUtils.java @@ -1725,7 +1725,8 @@ private static void int4(StringBuilder buf, int i) { } public static TimestampData truncate(TimestampData ts, int precision) { - String fraction = Integer.toString(ts.toLocalDateTime().getNano()); + // Zero-pad to nine digits so leading zeros are not dropped and the precision understated. + String fraction = String.format("%09d", ts.toLocalDateTime().getNano()); if (fraction.length() <= precision) { return ts; } else { diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java index 52ccc2a2d02b3..fe832e71e3bdf 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java @@ -2063,6 +2063,14 @@ Stream testCases() { LocalDateTime.of(2020, 1, 1, 12, 0, 0, 123000000)), TimestampData.fromLocalDateTime( LocalDateTime.of(2020, 1, 1, 12, 0, 0))), + // A fraction with leading zeros (.000123456) is still truncated to the precision. + CastTestSpecBuilder.testCastTo(TIMESTAMP(6)) + .fromCase( + VARIANT(), + Variant.newBuilder() + .of(LocalDateTime.of(2020, 1, 1, 12, 0, 0, 123_456)), + TimestampData.fromLocalDateTime( + LocalDateTime.of(2020, 1, 1, 12, 0, 0, 123_000))), // The cast accepts either storage kind: TIMESTAMP_NS for a value that needs // nanosecond precision, plain TIMESTAMP when microseconds already hold it exactly. CastTestSpecBuilder.testCastTo(TIMESTAMP(9)) From 1a3b0aa1f6110532f0c21b67ee99eeb6ae3096e2 Mon Sep 17 00:00:00 2001 From: Moritz Manner Date: Thu, 3 Sep 2026 17:46:02 +0200 Subject: [PATCH 4/6] [FLINK-40492][table] Address review comments --- .../docs/sql/reference/data-types.md | 3 ++ docs/content/docs/sql/reference/data-types.md | 3 ++ .../casting/CastRuleProviderTest.java | 4 +++ .../functions/casting/CastRulesTest.java | 15 ++++++++-- .../runtime/functions/VariantCastUtils.java | 30 ++++++++----------- 5 files changed, 35 insertions(+), 20 deletions(-) diff --git a/docs/content.zh/docs/sql/reference/data-types.md b/docs/content.zh/docs/sql/reference/data-types.md index 81aec696f8dec..1f39596b54c1e 100644 --- a/docs/content.zh/docs/sql/reference/data-types.md +++ b/docs/content.zh/docs/sql/reference/data-types.md @@ -1554,6 +1554,7 @@ to make it fit. Otherwise `CAST` fails and `TRY_CAST` returns `NULL`. | numeric kinds | any numeric target that holds the value | | `BOOLEAN` | `BOOLEAN` | | `DATE` | `DATE` | +| `TIME` | `TIME(p)` | | `TIMESTAMP` | `TIMESTAMP(p)` | | `TIMESTAMP_LTZ` | `TIMESTAMP_LTZ(p)` | | `BYTES` | `BINARY(n)`, `VARBINARY(n)`, and a character string | @@ -1572,6 +1573,8 @@ The conditions above mean: - A **length or precision** is adjusted the same way a regular cast into that type would: a value longer than the target is trimmed, fractional seconds beyond the target precision are truncated, and the fixed width types `CHAR(n)` and `BINARY(n)` pad a shorter value. +- A **`TIME`** value keeps only millisecond precision, the resolution Flink's runtime `TIME` type + supports, so `TIME(4)` through `TIME(9)` behave like `TIME(3)`. To reach a type the table does not list, wrap the cast in a regular cast. Only the inner cast is a `VARIANT` cast, so the outer one applies the usual rules and may round, truncate, or overflow: diff --git a/docs/content/docs/sql/reference/data-types.md b/docs/content/docs/sql/reference/data-types.md index 5ff9e0d8100d5..865f2d8fd0c44 100644 --- a/docs/content/docs/sql/reference/data-types.md +++ b/docs/content/docs/sql/reference/data-types.md @@ -1562,6 +1562,7 @@ to make it fit. Otherwise `CAST` fails and `TRY_CAST` returns `NULL`. | numeric kinds | any numeric target that holds the value | | `BOOLEAN` | `BOOLEAN` | | `DATE` | `DATE` | +| `TIME` | `TIME(p)` | | `TIMESTAMP` | `TIMESTAMP(p)` | | `TIMESTAMP_LTZ` | `TIMESTAMP_LTZ(p)` | | `BYTES` | `BINARY(n)`, `VARBINARY(n)`, and a character string | @@ -1580,6 +1581,8 @@ The conditions above mean: - A **length or precision** is adjusted the same way a regular cast into that type would: a value longer than the target is trimmed, fractional seconds beyond the target precision are truncated, and the fixed width types `CHAR(n)` and `BINARY(n)` pad a shorter value. +- A **`TIME`** value keeps only millisecond precision, the resolution Flink's runtime `TIME` type + supports, so `TIME(4)` through `TIME(9)` behave like `TIME(3)`. To reach a type the table does not list, wrap the cast in a regular cast. Only the inner cast is a `VARIANT` cast, so the outer one applies the usual rules and may round, truncate, or overflow: diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRuleProviderTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRuleProviderTest.java index 1fe1a23786091..0bebbd2c815b7 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRuleProviderTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRuleProviderTest.java @@ -18,6 +18,7 @@ package org.apache.flink.table.planner.functions.casting; +import org.apache.flink.table.api.DataTypes; import org.apache.flink.table.catalog.ObjectIdentifier; import org.apache.flink.table.types.logical.ArrayType; import org.apache.flink.table.types.logical.CharType; @@ -136,6 +137,9 @@ void testResolveVariantToPrimitive() { assertThat(CastRuleProvider.exists(VARIANT, BYTES().getLogicalType())).isTrue(); assertThat(CastRuleProvider.canFail(VARIANT, INT)).isTrue(); + // INTERVAL has no VARIANT counterpart, so it is not a castable target + assertThat(CastRuleProvider.exists(VARIANT, INTERVAL(DataTypes.DAY()).getLogicalType())) + .isFalse(); // character strings keep going through the display-oriented rule assertThat(CastRuleProvider.resolve(VARIANT, STRING_TYPE)) .isSameAs(VariantToStringCastRule.INSTANCE); diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java index fe832e71e3bdf..cb4c8b46826d7 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/casting/CastRulesTest.java @@ -2009,8 +2009,10 @@ Stream testCases() { CastTestSpecBuilder.testCastTo(TIME(3)) .fromCase( VARIANT(), - Variant.newBuilder().of(LocalTime.of(12, 0, 0, 123_000_000)), - DateTimeUtils.toInternal(LocalTime.of(12, 0, 0, 123_000_000))) + Variant.newBuilder() + .of(LocalTime.of(12, 0, 0).plus(Duration.ofMillis(123))), + DateTimeUtils.toInternal( + LocalTime.of(12, 0, 0).plus(Duration.ofMillis(123)))) .fromCase( VARIANT(), Variant.newBuilder().of(LocalTime.of(12, 0, 0, 123_456_000)), @@ -2030,7 +2032,14 @@ Stream testCases() { VARIANT_BUILDER.of(LocalDateTime.of(2020, 1, 1, 12, 0, 0)), TimestampData.fromLocalDateTime( LocalDateTime.of(2020, 1, 1, 12, 0, 0))) - .fail(VARIANT(), VARIANT_BUILDER.of(1), TableRuntimeException.class) + // the default precision keeps microseconds, truncating the nanoseconds + .fromCase( + VARIANT(), + Variant.newBuilder() + .of(LocalDateTime.of(2020, 1, 1, 12, 0, 0, 123456789)), + TimestampData.fromLocalDateTime( + LocalDateTime.of(2020, 1, 1, 12, 0, 0, 123456000))) + .fail(VARIANT(), Variant.newBuilder().of(1), TableRuntimeException.class) // a TIMESTAMP_LTZ is a different kind and is not read as a TIMESTAMP .fail( VARIANT(), diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java index be3001edff183..d7db1c2119763 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java @@ -58,16 +58,11 @@ public final class VariantCastUtils { */ private static final double LONG_MAGNITUDE_LIMIT = -(double) Long.MIN_VALUE; - /** A microsecond timestamp variant renders with six fractional-second digits. */ private static final int TIMESTAMP_PRECISION = 6; - /** A nanosecond timestamp variant renders with nine fractional-second digits. */ private static final int TIMESTAMP_NANOS_PRECISION = 9; - /** - * A time variant keeps microseconds, but the runtime TIME representation is millisecond-of-day, - * so it renders with three fractional-second digits, the same as a regular TIME to string cast. - */ + // TIME is millisecond-of-day at runtime, so it renders with three fractional-second digits. private static final int TIME_PRECISION = 3; private VariantCastUtils() {} @@ -245,20 +240,21 @@ private static String decimalTarget(int precision, int scale) { */ public static TimestampData toTimestamp(Variant variant, int precision) { final Variant.Type type = variant.getType(); - if (type != Variant.Type.TIMESTAMP && type != Variant.Type.TIMESTAMP_NS) { - throw unsupportedKind(variant, String.format("TIMESTAMP(%d)", precision)); + if (type == Variant.Type.TIMESTAMP || type == Variant.Type.TIMESTAMP_NS) { + return DateTimeUtils.truncate( + TimestampData.fromLocalDateTime(variant.getDateTime()), precision); } - return DateTimeUtils.truncate( - TimestampData.fromLocalDateTime(variant.getDateTime()), precision); + throw unsupportedKind(variant, String.format("TIMESTAMP(%d)", precision)); } /** Reads a timestamp with local time zone variant. See {@link #toTimestamp(Variant, int)}. */ public static TimestampData toTimestampLtz(Variant variant, int precision) { final Variant.Type type = variant.getType(); - if (type != Variant.Type.TIMESTAMP_LTZ && type != Variant.Type.TIMESTAMP_LTZ_NS) { - throw unsupportedKind(variant, String.format("TIMESTAMP_LTZ(%d)", precision)); + if (type == Variant.Type.TIMESTAMP_LTZ || type == Variant.Type.TIMESTAMP_LTZ_NS) { + return DateTimeUtils.truncate( + TimestampData.fromInstant(variant.getInstant()), precision); } - return DateTimeUtils.truncate(TimestampData.fromInstant(variant.getInstant()), precision); + throw unsupportedKind(variant, String.format("TIMESTAMP_LTZ(%d)", precision)); } /** @@ -268,11 +264,11 @@ public static TimestampData toTimestampLtz(Variant variant, int precision) { * TIME(p)} cast. */ public static int toTime(Variant variant, int precision) { - if (variant.getType() != Variant.Type.TIME) { - throw unsupportedKind(variant, String.format("TIME(%d)", precision)); + if (variant.getType() == Variant.Type.TIME) { + return DateTimeUtils.applyTimePrecisionTruncation( + DateTimeUtils.toInternal(variant.getTime()), precision); } - return DateTimeUtils.applyTimePrecisionTruncation( - DateTimeUtils.toInternal(variant.getTime()), precision); + throw unsupportedKind(variant, String.format("TIME(%d)", precision)); } /** From b0cdca490882dd60738a4e66b0cc243b3e893e42 Mon Sep 17 00:00:00 2001 From: Moritz Manner Date: Fri, 4 Sep 2026 16:30:45 +0200 Subject: [PATCH 5/6] [FLINK-40492][table] Simplify Timestamp truncation --- .../flink/table/utils/DateTimeUtils.java | 23 ++++++++----------- .../runtime/functions/VariantCastUtils.java | 2 +- 2 files changed, 11 insertions(+), 14 deletions(-) diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/DateTimeUtils.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/DateTimeUtils.java index 245c707767154..c458e1edac1d9 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/DateTimeUtils.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/DateTimeUtils.java @@ -1725,21 +1725,18 @@ private static void int4(StringBuilder buf, int i) { } public static TimestampData truncate(TimestampData ts, int precision) { - // Zero-pad to nine digits so leading zeros are not dropped and the precision understated. - String fraction = String.format("%09d", ts.toLocalDateTime().getNano()); - if (fraction.length() <= precision) { + // A timestamp holds at most nine fractional-second digits, so a higher precision keeps it + // unchanged. Otherwise, the digits beyond the precision are zeroed out. + if (precision >= 9) { return ts; - } else { - // need to truncate - if (precision <= 3) { - return TimestampData.fromEpochMillis( - zeroLastDigits(ts.getMillisecond(), 3 - precision)); - } else { - return TimestampData.fromEpochMillis( - ts.getMillisecond(), - (int) zeroLastDigits(ts.getNanoOfMillisecond(), 9 - precision)); - } } + if (precision <= 3) { + return TimestampData.fromEpochMillis( + zeroLastDigits(ts.getMillisecond(), 3 - precision)); + } + return TimestampData.fromEpochMillis( + ts.getMillisecond(), + (int) zeroLastDigits(ts.getNanoOfMillisecond(), 9 - precision)); } public static int truncate(int time, int precision) { diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java index d7db1c2119763..c26847c101696 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java @@ -234,7 +234,7 @@ private static String decimalTarget(int precision, int scale) { /** * Reads a timestamp variant as the target {@code TIMESTAMP}. {@link Variant#getDateTime()} - * already accepts both the microsecond ({@link Variant.Type#TIMESTAMP}) and nanosecond ({@link + * accepts both the microsecond ({@link Variant.Type#TIMESTAMP}) and nanosecond ({@link * Variant.Type#TIMESTAMP_NS}) encodings. Fractional seconds beyond the target precision are * truncated, the same as a regular {@code TIMESTAMP} to {@code TIMESTAMP(p)} cast. */ From 18d7a636ca273a049f68b1014ba6518b32eb2bb1 Mon Sep 17 00:00:00 2001 From: Moritz Manner Date: Mon, 7 Sep 2026 14:55:11 +0200 Subject: [PATCH 6/6] [hotfix][core] Fix flaky nanosecond-precision variant test --- .../java/org/apache/flink/types/variant/BinaryVariantTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 76c5222c168fb..8f9f35b3bf37d 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 @@ -123,7 +123,7 @@ void testNanosecondPrecisionVariant() { // 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); + Instant nanoInstant = Instant.now().truncatedTo(ChronoUnit.MICROS).plusNanos(123); Variant instantVariant = builder.of(nanoInstant); assertThat(instantVariant.getType()).isEqualTo(Variant.Type.TIMESTAMP_LTZ_NS); assertThat(instantVariant.getInstant()).isEqualTo(nanoInstant);