Skip to content

[FLINK-37925][table] Support casting from VARIANT to primitive types #28733

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

Merged
snuyanzin merged 5 commits into from
Jul 15, 2026
Merged

[FLINK-37925][table] Support casting from VARIANT to primitive types #28733

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
d1f1a22
[FLINK-37925][table] Support casting from VARIANT to primitive types
raminqaf Jul 13, 2026
6c422be
[FLINK-37925][table] Add tests for casting from VARIANT to primitive …
raminqaf Jul 13, 2026
cca9d90
[FLINK-37925][docs] Document casting from VARIANT to primitive types
raminqaf Jul 13, 2026
3bfbb43
[FLINK-37925][table] Cast a null-valued VARIANT to NULL and hint JSON…
raminqaf Jul 13, 2026
0ebbdb1
[FLINK-37925] Add more tests in LogicalTypeCastsTest
raminqaf Jul 13, 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
8 changes: 7 additions & 1 deletion docs/content.zh/docs/sql/reference/data-types.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1515,6 +1515,12 @@ without requiring upfront schema definition. For example, if a new field is adde
can be directly incorporated into the `VARIANT` data without modifying the table schema. This is
particularly useful in dynamic environments where schemas may evolve over time.

A primitive-valued `VARIANT` can be converted to a scalar type with `CAST` or `TRY_CAST`. Numeric
targets are lenient: a variant holding any numeric value casts to any numeric type, so a JSON integer
such as `PARSE_JSON('42')` casts to `INT` or `BIGINT`. Other targets require the stored value to be of
the matching kind. When the value cannot be converted, `CAST` fails the job and `TRY_CAST` returns
`NULL`. Use the `JSON_STRING` function to obtain the JSON string representation of a `VARIANT`.

**Declaration**

{{< tabs "25c30432-8460-441d-a036-9416d8202882" >}}
Expand Down Expand Up @@ -1729,7 +1735,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 | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N |
| `VARIANT` | N | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | N | ! | ! | N | N | N | 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 |

备注：
Expand Down
8 changes: 7 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 @@ -1523,6 +1523,12 @@ without requiring upfront schema definition. For example, if a new field is adde
can be directly incorporated into the `VARIANT` data without modifying the table schema. This is
particularly useful in dynamic environments where schemas may evolve over time.

A primitive-valued `VARIANT` can be converted to a scalar type with `CAST` or `TRY_CAST`. Numeric
targets are lenient: a variant holding any numeric value casts to any numeric type, so a JSON integer
such as `PARSE_JSON('42')` casts to `INT` or `BIGINT`. Other targets require the stored value to be of
the matching kind. When the value cannot be converted, `CAST` fails the job and `TRY_CAST` returns
`NULL`. Use the `JSON_STRING` function to obtain the JSON string representation of a `VARIANT`.

**Declaration**

{{< tabs "25c30432-8460-441d-a036-9416d8202882" >}}
Expand Down Expand Up @@ -1738,7 +1744,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 | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N |
| `VARIANT` | N | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | N | ! | ! | N | N | N | 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:
Expand Down
10 changes: 10 additions & 0 deletions ...rc/main/java/org/apache/flink/table/types/inference/strategies/CastInputTypeStrategy.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@
import java.util.List;
import java.util.Optional;

import static org.apache.flink.table.types.logical.utils.LogicalTypeCasts.getUnsupportedCastHint;
import static org.apache.flink.table.types.logical.utils.LogicalTypeCasts.supportsExplicitCast;

/**
Expand Down Expand Up @@ -69,6 +70,15 @@ public Optional<List<DataType>> inferInputTypes(
return Optional.of(argumentDataTypes);
}
if (!supportsExplicitCast(fromType, toType)) {
final Optional<String> hint = getUnsupportedCastHint(fromType, toType);
if (hint.isPresent()) {
return callContext.fail(
throwOnFailure,
"Unsupported cast from '%s' to '%s'. %s",
fromType,
toType,
hint.get());
}
return callContext.fail(
throwOnFailure, "Unsupported cast from '%s' to '%s'.", fromType, toType);
}
Expand Down
44 changes: 44 additions & 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 @@ -37,6 +37,7 @@
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.function.BiFunction;
import java.util.function.BiPredicate;
Expand Down Expand Up @@ -79,6 +80,7 @@
import static org.apache.flink.table.types.logical.LogicalTypeRoot.TINYINT;
import static org.apache.flink.table.types.logical.LogicalTypeRoot.VARBINARY;
import static org.apache.flink.table.types.logical.LogicalTypeRoot.VARCHAR;
import static org.apache.flink.table.types.logical.LogicalTypeRoot.VARIANT;
import static org.apache.flink.table.types.logical.utils.LogicalTypeChecks.getDayPrecision;
import static org.apache.flink.table.types.logical.utils.LogicalTypeChecks.getFractionalPrecision;
import static org.apache.flink.table.types.logical.utils.LogicalTypeChecks.getLength;
Expand Down Expand Up @@ -646,6 +648,9 @@ private static boolean supportsCasting(
// BITMAP can only be cast to BYTES (unbounded VARBINARY), because trimming or padding
// would corrupt the serialized bitmap data.
return allowExplicit && getLength(targetType) == VarBinaryType.MAX_LENGTH;
} else if (sourceRoot == VARIANT) {
// a VARIANT can only be explicitly cast to a supported scalar type
return allowExplicit && supportsVariantToScalarCast(targetType);
}

if (implicitCastingRules.get(targetRoot).contains(sourceRoot)) {
Expand Down Expand Up @@ -726,6 +731,45 @@ private static boolean supportsConstructedCasting(
return false;
}

private static boolean supportsVariantToScalarCast(LogicalType targetType) {
switch (targetType.getTypeRoot()) {
case BOOLEAN:
case TINYINT:
case SMALLINT:
case INTEGER:
case BIGINT:
case FLOAT:
case DOUBLE:
case DECIMAL:
case BINARY:
case VARBINARY:
case DATE:
case TIMESTAMP_WITHOUT_TIME_ZONE:
case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
return true;
default:
// TIME has no counterpart in the Variant type model. CHARACTER_STRING is handled by

@davidradl davidradl Jul 13, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I wonder if the cast is to a string then we could issue a helpful exception message indicating which SQL functions the user should use.

@raminqaf raminqaf Jul 13, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Even though this is a good idea and helpful for the user. Implementing it is a bit messy. Still I kept the changes for you to review. Since it adds lots of nose in the method. But let me know what you think. See getUnsupportedCastHint and CastInputTypeStrategy.

// the display-oriented VariantToStringCastRule and is intentionally not offered as
// a
// user-facing cast here.
return false;
}
}

/**
* Returns a hint pointing to the function that performs a conceptually related operation when
* an explicit cast is unsupported, or empty when no specific hint applies.
*/
public static Optional<String> getUnsupportedCastHint(
LogicalType sourceType, LogicalType targetType) {
if (sourceType.is(VARIANT) && targetType.is(CHARACTER_STRING)) {
return Optional.of(
"Use the JSON_STRING function to convert a VARIANT to its JSON string "
+ "representation.");
}
return Optional.empty();
}

private static CastingRuleBuilder castTo(LogicalTypeRoot targetType) {
return new CastingRuleBuilder(targetType);
}
Expand Down
35 changes: 34 additions & 1 deletion ...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 @@ -34,6 +34,7 @@
import org.apache.flink.table.types.logical.IntType;
import org.apache.flink.table.types.logical.LocalZonedTimestampType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.MapType;
import org.apache.flink.table.types.logical.NullType;
import org.apache.flink.table.types.logical.RawType;
import org.apache.flink.table.types.logical.RowType;
Expand All @@ -46,6 +47,7 @@
import org.apache.flink.table.types.logical.TinyIntType;
import org.apache.flink.table.types.logical.VarBinaryType;
import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.table.types.logical.VariantType;
import org.apache.flink.table.types.logical.YearMonthIntervalType;
import org.apache.flink.table.types.logical.ZonedTimestampType;
import org.apache.flink.table.types.logical.utils.LogicalTypeCasts;
Expand All @@ -57,6 +59,7 @@
import org.junit.jupiter.params.provider.MethodSource;

import java.util.Arrays;
import java.util.List;
import java.util.stream.Stream;

import static org.assertj.core.api.Assertions.assertThat;
Expand Down Expand Up @@ -262,7 +265,37 @@ private static Stream<Arguments> testData() {
new RawType<>(Integer.class, IntSerializer.INSTANCE),
VarCharType.STRING_TYPE,
false,
true));
true),

// variant to scalar is explicit only
Arguments.of(new VariantType(), new BooleanType(), false, true),
Arguments.of(new VariantType(), new TinyIntType(), false, true),
Arguments.of(new VariantType(), new IntType(), false, true),
Arguments.of(new VariantType(), new BigIntType(), false, true),
Arguments.of(new VariantType(), new DoubleType(), false, true),
Arguments.of(new VariantType(), new DecimalType(10, 2), false, true),
Arguments.of(new VariantType(), new DateType(), false, true),
Arguments.of(new VariantType(), new TimestampType(), false, true),
Comment thread
raminqaf marked this conversation as resolved.
Arguments.of(new VariantType(), new TimestampType(3), false, true),
Arguments.of(new VariantType(), new LocalZonedTimestampType(), false, true),
Arguments.of(new VariantType(), new LocalZonedTimestampType(9), false, true),
Arguments.of(

@snuyanzin snuyanzin Jul 13, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

is interval intentionally skipped?

@raminqaf raminqaf Jul 13, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Variant.Type has no interval kind. Please refer to: https://cwiki.apache.org/confluence/spaces/FLINK/pages/349637099/FLIP-521+Integrating+Variant+Type+into+Flink+Enabling+Efficient+Semi-Structured+Data+Processing#FLIP521:IntegratingVariantTypeintoFlink:EnablingEfficientSemiStructuredDataProcessing-Casting

new VariantType(),
new VarBinaryType(VarBinaryType.MAX_LENGTH),
false,
true),
// variant identity cast is implicit
Arguments.of(new VariantType(), new VariantType(), true, true),
// TIME, character strings and constructed targets are not castable from variant
Arguments.of(new VariantType(), new TimeType(), false, false),
Arguments.of(new VariantType(), VarCharType.STRING_TYPE, false, false),
Arguments.of(new VariantType(), new ArrayType(new IntType()), false, false),
Arguments.of(new VariantType(), new RowType(List.of()), false, false),
Arguments.of(
new VariantType(),
new MapType(new IntType(), new CharType()),
false,
false));
}

@ParameterizedTest(name = "{index}: [From: {0}, To: {1}, Implicit: {2}, Explicit: {3}]")
Expand Down
1 change: 1 addition & 0 deletions ...nner/src/main/java/org/apache/flink/table/planner/functions/casting/CastRuleProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,6 +97,7 @@ public class CastRuleProvider {
.addRule(RowToRowCastRule.INSTANCE)
// Variant rules
.addRule(VariantToStringCastRule.INSTANCE)
.addRule(VariantToPrimitiveCastRule.INSTANCE)
// Bitmap rules
.addRule(BitmapToStringCastRule.INSTANCE)
.addRule(BitmapToBinaryCastRule.INSTANCE)
Expand Down
Loading