This is an automated email from the ASF dual-hosted git repository.
twalthr 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 21f6a078766 [FLINK-40492][table] Support casting from VARIANT to TIME
and nanosecond timestamps
21f6a078766 is described below
commit 21f6a0787663c7ab6ede99daf081e47065efc5e4
Author: Moritz Manner <[email protected]>
AuthorDate: Thu Sep 10 15:24:36 2026 +0200
[FLINK-40492][table] Support casting from VARIANT to TIME and nanosecond
timestamps
This closes #29080.
---
docs/content.zh/docs/sql/reference/data-types.md | 5 +-
docs/content/docs/sql/reference/data-types.md | 5 +-
.../flink/types/variant/BinaryVariantTest.java | 2 +-
.../types/logical/utils/LogicalTypeCasts.java | 1 +
.../apache/flink/table/utils/DateTimeUtils.java | 22 ++---
.../flink/table/types/LogicalTypeCastsTest.java | 3 +-
.../casting/VariantToPrimitiveCastRule.java | 13 ++-
.../functions/casting/CastRuleProviderTest.java | 7 +-
.../planner/functions/casting/CastRulesTest.java | 101 ++++++++++++++++++++-
.../table/runtime/functions/VariantCastUtils.java | 66 +++++++++++---
10 files changed, 189 insertions(+), 36 deletions(-)
diff --git a/docs/content.zh/docs/sql/reference/data-types.md
b/docs/content.zh/docs/sql/reference/data-types.md
index fa20d4c273f..26fc40150b2 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:
@@ -1874,7 +1877,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 e62f1997d29..1a687f6b972 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:
@@ -1883,7 +1886,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-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 5c4a6453dcc..d0d383604a5 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
@@ -128,7 +128,7 @@ class BinaryVariantTest {
// 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);
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 474e63b3670..e213c477164 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/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 8a161873e21..c458e1edac1 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,20 +1725,18 @@ public class DateTimeUtils {
}
public static TimestampData truncate(TimestampData ts, int precision) {
- String fraction = Integer.toString(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-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 f6022d3f2cb..f696227034e 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 @@ class LogicalTypeCastsTest {
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 9b42628498f..4a16d7fce81 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 @@ import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.str
* 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}.
*
- * <p>{@code CHARACTER_STRING} is handled by {@link VariantToStringCastRule};
{@code TIME} has no
- * variant counterpart and is unsupported.
+ * <p>{@code CHARACTER_STRING} is handled by {@link VariantToStringCastRule}.
*/
class VariantToPrimitiveCastRule extends
AbstractNullAwareCodeGeneratorCastRule<Variant, Object> {
@@ -73,6 +72,7 @@ class VariantToPrimitiveCastRule extends
AbstractNullAwareCodeGeneratorCastRule<
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 @@ class VariantToPrimitiveCastRule extends
AbstractNullAwareCodeGeneratorCastRule<
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 a743cf9ee55..0bebbd2c815 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;
@@ -132,11 +133,13 @@ class CastRuleProviderTest {
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();
+ // 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 f5e98a0fc0a..2aa04aeab2d 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
@@ -1846,6 +1846,24 @@ class CastRulesTest {
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(),
@@ -2021,7 +2039,28 @@ class CastRulesTest {
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).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)),
+ 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(
@@ -2029,7 +2068,14 @@ class CastRulesTest {
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(),
@@ -2062,6 +2108,35 @@ class CastRulesTest {
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))
+ .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(),
@@ -2080,6 +2155,28 @@ class CastRulesTest {
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 ddd86cdda64..561da100cee 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,13 @@ public final class VariantCastUtils {
*/
private static final double LONG_MAGNITUDE_LIMIT = -(double)
Long.MIN_VALUE;
- /** A variant stores a timestamp with microsecond precision. */
private static final int TIMESTAMP_PRECISION = 6;
+ private static final int TIMESTAMP_NANOS_PRECISION = 9;
+
+ // TIME is millisecond-of-day at runtime, so it renders with three
fractional-second digits.
+ private static final int TIME_PRECISION = 3;
+
private VariantCastUtils() {}
/**
@@ -229,24 +233,42 @@ public final class VariantCastUtils {
}
/**
- * 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()}
+ * 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) {
- throw unsupportedKind(variant, String.format("TIMESTAMP(%d)",
precision));
+ final Variant.Type type = variant.getType();
+ 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)
{
- if (variant.getType() != Variant.Type.TIMESTAMP_LTZ) {
- throw unsupportedKind(variant, String.format("TIMESTAMP_LTZ(%d)",
precision));
+ final Variant.Type type = variant.getType();
+ if (type == Variant.Type.TIMESTAMP_LTZ || type ==
Variant.Type.TIMESTAMP_LTZ_NS) {
+ return DateTimeUtils.truncate(
+ TimestampData.fromInstant(variant.getInstant()),
precision);
+ }
+ throw unsupportedKind(variant, String.format("TIMESTAMP_LTZ(%d)",
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) {
+ return DateTimeUtils.applyTimePrecisionTruncation(
+ DateTimeUtils.toInternal(variant.getTime()), precision);
}
- return
DateTimeUtils.truncate(TimestampData.fromInstant(variant.getInstant()),
precision);
+ throw unsupportedKind(variant, String.format("TIME(%d)", precision));
}
/**
@@ -377,15 +399,26 @@ public final class VariantCastUtils {
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(
@@ -393,6 +426,13 @@ public final class VariantCastUtils {
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.
