Skip to content

[FLINK-40492][table] Support casting from VARIANT to TIME and nanosecond timestamps #29080

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 6 commits into
base: master
Choose a base branch
from
Open

[FLINK-40492][table] Support casting from VARIANT to TIME and nanosecond timestamps #29080

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
0eac3c3
[FLINK-40492][docs] Fix VARIANT to CHARACTER_STRING entry in the cast…
manner Sep 2, 2026
e03946b
[FLINK-40492][table] Support casting from VARIANT to TIME and nanosec…
manner Sep 2, 2026
e1fa3a9
[FLINK-40492][table] Fix timestamp truncation for fractions with lead…
manner Sep 3, 2026
1a3b0aa
[FLINK-40492][table] Address review comments
manner Sep 3, 2026
b0cdca4
[FLINK-40492][table] Simplify Timestamp truncation
manner Sep 4, 2026
18d7a63
[hotfix][core] Fix flaky nanosecond-precision variant test
manner Sep 7, 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
5 changes: 4 additions & 1 deletion docs/content.zh/docs/sql/reference/data-types.md
View file
Open in desktop
Comment thread
manner marked this conversation as resolved.
Original file line number Diff line number Diff line change
Expand Up @@ -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 |
Expand All @@ -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:
Expand Down Expand Up @@ -1873,7 +1876,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 | 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 |

备注：
Expand Down
5 changes: 4 additions & 1 deletion docs/content/docs/sql/reference/data-types.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 |
Expand All @@ -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:
Expand Down Expand Up @@ -1882,7 +1885,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 | 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:
Expand Down
2 changes: 1 addition & 1 deletion flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
1 change: 1 addition & 0 deletions ...ble-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeCasts.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();

Expand Down
22 changes: 10 additions & 12 deletions flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/DateTimeUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1725,20 +1725,18 @@ private static void int4(StringBuilder buf, int i) {
}

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) {
Expand Down
3 changes: 1 addition & 2 deletions ...e/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeCastsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -289,10 +289,9 @@ private static Stream<Arguments> 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),
Expand Down
13 changes: 11 additions & 2 deletions ...ain/java/org/apache/flink/table/planner/functions/casting/VariantToPrimitiveCastRule.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}.
*
* <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> {

Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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,
Expand Down
7 changes: 5 additions & 2 deletions .../src/test/java/org/apache/flink/table/planner/functions/casting/CastRuleProviderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -132,11 +133,13 @@ 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();
Comment thread
manner marked this conversation as resolved.
// 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);
Expand Down
Loading