This is an automated email from the ASF dual-hosted git repository.
snuyanzin 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 76e7898e484 [FLINK-37925][table] Support casting from `VARIANT` to
primitive types
76e7898e484 is described below
commit 76e7898e484cb03aa6e030180310c4a941d41a83
Author: Ramin Gharib <[email protected]>
AuthorDate: Wed Jul 15 13:59:57 2026 +0200
[FLINK-37925][table] Support casting from `VARIANT` to primitive types
---
docs/content.zh/docs/sql/reference/data-types.md | 8 +-
docs/content/docs/sql/reference/data-types.md | 8 +-
.../strategies/CastInputTypeStrategy.java | 10 +
.../types/logical/utils/LogicalTypeCasts.java | 44 ++++
.../flink/table/types/LogicalTypeCastsTest.java | 35 ++-
.../functions/casting/CastRuleProvider.java | 1 +
.../casting/VariantToPrimitiveCastRule.java | 237 +++++++++++++++++++++
.../planner/functions/CastFunctionITCase.java | 105 ++++++++-
.../functions/casting/CastRuleProviderTest.java | 28 +++
.../planner/functions/casting/CastRulesTest.java | 98 ++++++++-
10 files changed, 569 insertions(+), 5 deletions(-)
diff --git a/docs/content.zh/docs/sql/reference/data-types.md
b/docs/content.zh/docs/sql/reference/data-types.md
index 1cbc6f8bae6..6ad9cd02c7a 100644
--- a/docs/content.zh/docs/sql/reference/data-types.md
+++ b/docs/content.zh/docs/sql/reference/data-types.md
@@ -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" >}}
@@ -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 |
备注：
diff --git a/docs/content/docs/sql/reference/data-types.md
b/docs/content/docs/sql/reference/data-types.md
index a849ee5dac8..45ace31e365 100644
--- a/docs/content/docs/sql/reference/data-types.md
+++ b/docs/content/docs/sql/reference/data-types.md
@@ -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" >}}
@@ -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:
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/CastInputTypeStrategy.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/CastInputTypeStrategy.java
index fbd4bf188d7..3bd66e01cd5 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/CastInputTypeStrategy.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/CastInputTypeStrategy.java
@@ -35,6 +35,7 @@ import java.util.Collections;
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;
/**
@@ -69,6 +70,15 @@ class CastInputTypeStrategy implements InputTypeStrategy {
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);
}
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 a0afca2caa0..58f474ec0eb 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
@@ -37,6 +37,7 @@ import java.util.HashMap;
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;
@@ -79,6 +80,7 @@ import static
org.apache.flink.table.types.logical.LogicalTypeRoot.TIME_WITHOUT_
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;
@@ -646,6 +648,9 @@ public final class LogicalTypeCasts {
// 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)) {
@@ -726,6 +731,45 @@ public final class LogicalTypeCasts {
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
+ // 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);
}
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 3d0dec23997..6fb3574b1f5 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
@@ -34,6 +34,7 @@ import org.apache.flink.table.types.logical.FloatType;
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;
@@ -46,6 +47,7 @@ import org.apache.flink.table.types.logical.TimestampType;
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;
@@ -57,6 +59,7 @@ import org.junit.jupiter.params.provider.Arguments;
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;
@@ -262,7 +265,37 @@ class LogicalTypeCastsTest {
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),
+ 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(
+ 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}]")
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CastRuleProvider.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CastRuleProvider.java
index b18b44e126e..133542ec6f5 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CastRuleProvider.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CastRuleProvider.java
@@ -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)
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
new file mode 100644
index 00000000000..8a278f41498
--- /dev/null
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/VariantToPrimitiveCastRule.java
@@ -0,0 +1,237 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.functions.casting;
+
+import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.TimestampData;
+import
org.apache.flink.table.planner.functions.casting.CastRuleUtils.CodeWriter;
+import org.apache.flink.table.types.logical.DecimalType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.utils.LogicalTypeChecks;
+import org.apache.flink.types.variant.Variant;
+
+import java.math.BigDecimal;
+import java.util.Arrays;
+
+import static org.apache.flink.table.planner.codegen.CodeGenUtils.className;
+import static org.apache.flink.table.planner.codegen.CodeGenUtils.newName;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.arrayLength;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.cast;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.constructorCall;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.methodCall;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.staticCall;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.ternaryOperator;
+
+/**
+ * {@link LogicalTypeRoot#VARIANT} to primitive type cast rule.
+ *
+ * <p>Numeric targets are lenient and follow regular numeric cast semantics;
other targets require
+ * the stored value to match the target kind. On a mismatch {@code CAST} fails
and {@code TRY_CAST}
+ * returns {@code null}.
+ *
+ * <p>{@code CHARACTER_STRING} is handled by {@link VariantToStringCastRule};
{@code TIME} has no
+ * variant counterpart and is unsupported.
+ */
+class VariantToPrimitiveCastRule extends
AbstractNullAwareCodeGeneratorCastRule<Variant, Object> {
+
+ static final VariantToPrimitiveCastRule INSTANCE = new
VariantToPrimitiveCastRule();
+
+ private VariantToPrimitiveCastRule() {
+ super(
+ CastRulePredicate.builder()
+ .predicate(
+ (input, target) ->
+ input.is(LogicalTypeRoot.VARIANT)
+ && isSupportedTarget(target))
+ .build());
+ }
+
+ private static boolean isSupportedTarget(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:
+ return false;
+ }
+ }
+
+ @Override
+ public boolean canFail(LogicalType inputLogicalType, LogicalType
targetLogicalType) {
+ return true;
+ }
+
+ /**
+ * Treats a variant that stores a JSON {@code null} as a {@code NULL}
input, so it casts to SQL
+ * {@code NULL} instead of failing in the type-specific accessor. Only
applied for a nullable
+ * target: a {@code NOT NULL} result cannot carry {@code NULL}, so a
null-valued variant then
+ * fails as a regular type mismatch.
+ */
+ @Override
+ public CastCodeBlock generateCodeBlock(
+ CodeGeneratorCastRule.Context context,
+ String inputTerm,
+ String inputIsNullTerm,
+ LogicalType inputLogicalType,
+ LogicalType targetLogicalType) {
+ if (!targetLogicalType.isNullable()) {
+ return super.generateCodeBlock(
+ context, inputTerm, inputIsNullTerm, inputLogicalType,
targetLogicalType);
+ }
+ final String isNullTerm =
+ "(" + inputIsNullTerm + " || " + methodCall(inputTerm,
"isNull") + ")";
+ return super.generateCodeBlock(
+ context, inputTerm, isNullTerm, inputLogicalType,
targetLogicalType);
+ }
+
+ @Override
+ protected String generateCodeBlockInternal(
+ CodeGeneratorCastRule.Context context,
+ String inputTerm,
+ String returnVariable,
+ LogicalType inputLogicalType,
+ LogicalType targetLogicalType) {
+ final CodeWriter writer = new CastRuleUtils.CodeWriter();
+ switch (targetLogicalType.getTypeRoot()) {
+ case BOOLEAN:
+ writer.assignStmt(returnVariable, methodCall(inputTerm,
"getBoolean"));
+ break;
+ case TINYINT:
+ case SMALLINT:
+ case INTEGER:
+ case BIGINT:
+ case FLOAT:
+ case DOUBLE:
+ case DECIMAL:
+ writer.assignStmt(returnVariable, numericExpression(inputTerm,
targetLogicalType));
+ break;
+ case BINARY:
+ case VARBINARY:
+ generateToBytes(context, inputTerm, returnVariable,
targetLogicalType, writer);
+ break;
+ case DATE:
+ writer.assignStmt(
+ returnVariable,
+ cast("int", methodCall(methodCall(inputTerm,
"getDate"), "toEpochDay")));
+ break;
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ writer.assignStmt(
+ returnVariable,
+ staticCall(
+ TimestampData.class,
+ "fromLocalDateTime",
+ methodCall(inputTerm, "getDateTime")));
+ break;
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ writer.assignStmt(
+ returnVariable,
+ staticCall(
+ TimestampData.class,
+ "fromInstant",
+ methodCall(inputTerm, "getInstant")));
+ break;
+ default:
+ throw new IllegalArgumentException(
+ "Unsupported target type for casting from VARIANT: " +
targetLogicalType);
+ }
+ return writer.toString();
+ }
+
+ /**
+ * Converts a numeric variant to the numeric {@code target} via the
matching {@link Number}
+ * accessor, mirroring regular numeric cast semantics. A non-numeric
variant raises {@link
+ * ClassCastException}, failing {@code CAST} and yielding {@code null} for
{@code TRY_CAST}.
+ */
+ private static String numericExpression(String inputTerm, LogicalType
target) {
+ final String number = cast(className(Number.class),
methodCall(inputTerm, "get"));
+ if (!target.is(LogicalTypeRoot.DECIMAL)) {
+ return methodCall(number, numberAccessor(target));
+ }
+ final DecimalType decimalType = (DecimalType) target;
+ return staticCall(
+ DecimalData.class,
+ "fromBigDecimal",
+ constructorCall(BigDecimal.class, methodCall(number,
"toString")),
+ decimalType.getPrecision(),
+ decimalType.getScale());
+ }
+
+ private static String numberAccessor(LogicalType target) {
+ switch (target.getTypeRoot()) {
+ case TINYINT:
+ return "byteValue";
+ case SMALLINT:
+ return "shortValue";
+ case INTEGER:
+ return "intValue";
+ case BIGINT:
+ return "longValue";
+ case FLOAT:
+ return "floatValue";
+ case DOUBLE:
+ return "doubleValue";
+ default:
+ throw new IllegalArgumentException(
+ "Unsupported numeric target for casting from VARIANT:
" + target);
+ }
+ }
+
+ private static void generateToBytes(
+ CodeGeneratorCastRule.Context context,
+ String inputTerm,
+ String returnVariable,
+ LogicalType targetLogicalType,
+ CodeWriter writer) {
+ final int targetLength =
LogicalTypeChecks.getLength(targetLogicalType);
+ // Read the bytes once to avoid decoding the variant twice.
+ final String bytesTerm = newName(context.getCodeGeneratorContext(),
"variantBytes");
+ writer.declStmt("byte[]", bytesTerm, methodCall(inputTerm,
"getBytes"));
+ if (BinaryToBinaryCastRule.couldPad(targetLogicalType, targetLength)) {
+ // BINARY(n): pad or trim to the exact target length.
+ writer.assignStmt(
+ returnVariable,
+ ternaryOperator(
+ arrayLength(bytesTerm) + " == " + targetLength,
+ bytesTerm,
+ staticCall(Arrays.class, "copyOf", bytesTerm,
targetLength)));
+ } else if (BinaryToBinaryCastRule.couldTrim(targetLength)) {
+ // VARBINARY(n): trim only when longer than the target length.
+ writer.assignStmt(
+ returnVariable,
+ ternaryOperator(
+ arrayLength(bytesTerm) + " <= " + targetLength,
+ bytesTerm,
+ staticCall(Arrays.class, "copyOf", bytesTerm,
targetLength)));
+ } else {
+ writer.assignStmt(returnVariable, bytesTerm);
+ }
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java
index 5420ce1987f..cb41ad083c7 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java
@@ -78,6 +78,7 @@ import static org.apache.flink.table.api.DataTypes.VARBINARY;
import static org.apache.flink.table.api.DataTypes.VARCHAR;
import static org.apache.flink.table.api.DataTypes.YEAR;
import static org.apache.flink.table.api.Expressions.$;
+import static org.apache.flink.table.api.Expressions.call;
import static org.apache.flink.util.CollectionUtil.entry;
import static org.apache.flink.util.CollectionUtil.map;
import static org.assertj.core.api.Assertions.assertThat;
@@ -132,9 +133,111 @@ public class CastFunctionITCase extends
BuiltInFunctionTestBase {
specs.addAll(numericBounds());
specs.addAll(constructedTypes());
specs.addAll(bitmapCasts());
+ specs.addAll(variantCasts());
return specs.stream();
}
+ private static List<TestSetSpec> variantCasts() {
+ // A variant is produced with PARSE_JSON so that the source column
stays a STRING literal;
+ // there is no VARIANT literal to feed a source column directly. A
JSON integer is stored in
+ // the smallest integer type that fits (42 -> TINYINT), and numeric
casts are lenient, so it
+ // still widens to INT.
+ return List.of(
+ TestSetSpec.forExpression("Cast a VARIANT produced by
PARSE_JSON to a primitive")
+ .onFieldsWithData("unused")
+ .andDataTypes(STRING())
+ .testResult(
+ call("PARSE_JSON", "42").cast(INT()),
+ "CAST(PARSE_JSON('42') AS INT)",
+ 42,
+ INT().notNull())
+ // Integer overflow wraps around (Java narrowing),
like a regular numeric
+ // cast.
+ .testResult(
+ call("PARSE_JSON", "40000").cast(SMALLINT()),
+ "CAST(PARSE_JSON('40000') AS SMALLINT)",
+ (short) -25536,
+ SMALLINT().notNull())
+ .testResult(
+ call("PARSE_JSON", "128").cast(TINYINT()),
+ "CAST(PARSE_JSON('128') AS TINYINT)",
+ (byte) -128,
+ TINYINT().notNull())
+ .testResult(
+ call("PARSE_JSON", "2147483648").cast(INT()),
+ "CAST(PARSE_JSON('2147483648') AS INT)",
+ -2147483648,
+ INT().notNull())
+ .testResult(
+ call("PARSE_JSON",
"9223372036854775808").cast(BIGINT()),
+ "CAST(PARSE_JSON('9223372036854775808') AS
BIGINT)",
+ -9223372036854775808L,
+ BIGINT().notNull())
+ // An out-of-range floating point value saturates when
cast to an integer,
+ // and a fractional value is truncated toward zero.
+ .testResult(
+ call("PARSE_JSON", "1e20").cast(INT()),
+ "CAST(PARSE_JSON('1e20') AS INT)",
+ 2147483647,
+ INT().notNull())
+ .testResult(
+ call("PARSE_JSON", "3.9").cast(INT()),
+ "CAST(PARSE_JSON('3.9') AS INT)",
+ 3,
+ INT().notNull())
+ // A value beyond the FLOAT range becomes infinity.
+ .testResult(
+ call("PARSE_JSON", "1e40").cast(FLOAT()),
+ "CAST(PARSE_JSON('1e40') AS FLOAT)",
+ Float.POSITIVE_INFINITY,
+ FLOAT().notNull())
+ // DECIMAL overflow yields NULL instead of wrapping;
TRY_CAST surfaces it.
+ .testResult(
+ call("PARSE_JSON",
"123.456").tryCast(DECIMAL(4, 2)),
+ "TRY_CAST(PARSE_JSON('123.456') AS DECIMAL(4,
2))",
+ null,
+ DECIMAL(4, 2))
+ .testResult(
+ call("PARSE_JSON", "42").cast(BIGINT()),
+ "CAST(PARSE_JSON('42') AS BIGINT)",
+ 42L,
+ BIGINT().notNull())
+ .testResult(
+ call("PARSE_JSON", "true").cast(BOOLEAN()),
+ "CAST(PARSE_JSON('true') AS BOOLEAN)",
+ true,
+ BOOLEAN().notNull())
+ // TRY_CAST of a value whose kind does not match the
target returns NULL
+ .testResult(
+ call("PARSE_JSON", "\"foo\"").tryCast(INT()),
+ "TRY_CAST(PARSE_JSON('\"foo\"') AS INT)",
+ null,
+ INT())
+ // A variant that stores a JSON null casts to SQL NULL
when the target is
+ // nullable: a nullable variant source, or TRY_CAST
which forces nullable.
+ .testResult(
+ call("TRY_PARSE_JSON", "null").cast(INT()),
+ "CAST(TRY_PARSE_JSON('null') AS INT)",
+ null,
+ INT())
+ .testResult(
+ call("PARSE_JSON", "null").tryCast(BOOLEAN()),
+ "TRY_CAST(PARSE_JSON('null') AS BOOLEAN)",
+ null,
+ BOOLEAN())
+ // A nullable variant with a concrete value still
casts normally.
+ .testResult(
+ call("TRY_PARSE_JSON", "42").cast(INT()),
+ "CAST(TRY_PARSE_JSON('42') AS INT)",
+ 42,
+ INT())
+ // Casting a VARIANT to a string points the user to
JSON_STRING.
+ .testTableApiValidationError(
+ call("PARSE_JSON", "42").cast(STRING()),
+ "Use the JSON_STRING function to convert a
VARIANT to its JSON "
+ + "string representation."));
+ }
+
private static List<TestSetSpec> allTypesBasic() {
return Arrays.asList(
CastTestSpecBuilder.testCastTo(BOOLEAN())
@@ -1269,7 +1372,7 @@ public class CastFunctionITCase extends
BuiltInFunctionTestBase {
}
private static List<TestSetSpec> decimalCasts() {
- return Collections.singletonList(
+ return List.of(
CastTestSpecBuilder.testCastTo(DECIMAL(8, 4))
.fromCase(STRING(), null, null)
// rounding
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 4e230667111..adc44c474be 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
@@ -28,13 +28,20 @@ import org.apache.flink.table.types.logical.VarCharType;
import org.junit.jupiter.api.Test;
import static org.apache.flink.table.api.DataTypes.BIGINT;
+import static org.apache.flink.table.api.DataTypes.BOOLEAN;
+import static org.apache.flink.table.api.DataTypes.BYTES;
+import static org.apache.flink.table.api.DataTypes.DATE;
+import static org.apache.flink.table.api.DataTypes.DECIMAL;
import static org.apache.flink.table.api.DataTypes.FIELD;
import static org.apache.flink.table.api.DataTypes.INT;
import static org.apache.flink.table.api.DataTypes.ROW;
import static org.apache.flink.table.api.DataTypes.STRING;
import static org.apache.flink.table.api.DataTypes.STRUCTURED;
import static org.apache.flink.table.api.DataTypes.TIME;
+import static org.apache.flink.table.api.DataTypes.TIMESTAMP;
+import static org.apache.flink.table.api.DataTypes.TIMESTAMP_LTZ;
import static org.apache.flink.table.api.DataTypes.TINYINT;
+import static org.apache.flink.table.api.DataTypes.VARIANT;
import static org.apache.flink.table.types.logical.VarCharType.STRING_TYPE;
import static org.assertj.core.api.Assertions.assertThat;
@@ -48,6 +55,7 @@ class CastRuleProviderTest {
.build();
private static final LogicalType INT = INT().getLogicalType();
private static final LogicalType TINYINT = TINYINT().getLogicalType();
+ private static final LogicalType VARIANT = VARIANT().getLogicalType();
private static final LogicalType ROW =
ROW(FIELD("a", INT()), FIELD("b",
TINYINT().notNull())).getLogicalType();
private static final LogicalType STRUCTURED =
@@ -108,4 +116,24 @@ class CastRuleProviderTest {
assertThat(CastRuleProvider.canFail(inputType, ROW(INT(),
STRING()).getLogicalType()))
.isFalse();
}
+
+ @Test
+ void testResolveVariantToPrimitive() {
+ assertThat(CastRuleProvider.resolve(VARIANT, INT))
+ .isSameAs(VariantToPrimitiveCastRule.INSTANCE);
+ assertThat(CastRuleProvider.resolve(VARIANT,
BOOLEAN().getLogicalType()))
+ .isSameAs(VariantToPrimitiveCastRule.INSTANCE);
+ assertThat(CastRuleProvider.exists(VARIANT, DECIMAL(10,
2).getLogicalType())).isTrue();
+ 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,
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 2aa6e198a98..8459a804163 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
@@ -41,6 +41,7 @@ import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.StructuredType;
import org.apache.flink.table.utils.DateTimeUtils;
import org.apache.flink.types.bitmap.Bitmap;
+import org.apache.flink.types.variant.Variant;
import org.junit.jupiter.api.DynamicTest;
import org.junit.jupiter.api.TestFactory;
@@ -50,6 +51,7 @@ import org.junit.jupiter.api.parallel.ExecutionMode;
import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
+import java.time.Instant;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
@@ -95,6 +97,7 @@ import static
org.apache.flink.table.api.DataTypes.TIMESTAMP_LTZ;
import static org.apache.flink.table.api.DataTypes.TINYINT;
import static org.apache.flink.table.api.DataTypes.VARBINARY;
import static org.apache.flink.table.api.DataTypes.VARCHAR;
+import static org.apache.flink.table.api.DataTypes.VARIANT;
import static org.apache.flink.table.api.DataTypes.YEAR;
import static org.apache.flink.table.data.DecimalData.fromBigDecimal;
import static org.apache.flink.table.data.StringData.fromString;
@@ -1540,7 +1543,100 @@ class CastRulesTest {
CastTestSpecBuilder.testCastTo(BYTES())
.fromCase(BITMAP(), DEFAULT_BITMAP,
DEFAULT_BITMAP.toBytes())
.fromCase(BITMAP(), Bitmap.empty(),
Bitmap.empty().toBytes())
- .fromCase(BITMAP(), null, null));
+ .fromCase(BITMAP(), null, null),
+ // From VARIANT to primitive types. Numeric targets are
lenient: a variant holding
+ // any numeric kind converts to the requested numeric type
(widening and narrowing).
+ // Non-numeric targets are strict: the stored kind must match,
otherwise the cast
+ // fails and TRY_CAST returns null.
+ CastTestSpecBuilder.testCastTo(BOOLEAN())
+ .fromCase(VARIANT(), Variant.newBuilder().of(true),
true)
+ .fromCase(VARIANT(), Variant.newBuilder().of(false),
false)
+ .fail(VARIANT(), Variant.newBuilder().of(1),
TableRuntimeException.class),
+ CastTestSpecBuilder.testCastTo(TINYINT())
+ .fromCase(VARIANT(), Variant.newBuilder().of((byte)
42), (byte) 42)
+ .fromCase(VARIANT(), Variant.newBuilder().of(42),
(byte) 42)
+ .fail(VARIANT(), Variant.newBuilder().of("x"),
TableRuntimeException.class),
+ CastTestSpecBuilder.testCastTo(SMALLINT())
+ .fromCase(VARIANT(), Variant.newBuilder().of((short)
42), (short) 42)
+ .fromCase(VARIANT(), Variant.newBuilder().of((byte)
42), (short) 42)
+ .fail(
+ VARIANT(),
+ Variant.newBuilder().of(true),
+ TableRuntimeException.class),
+ CastTestSpecBuilder.testCastTo(INT())
+ // widening: a JSON integer is stored in the smallest
type but still casts
+ // up
+ .fromCase(VARIANT(), Variant.newBuilder().of((byte)
42), 42)
+ .fromCase(VARIANT(), Variant.newBuilder().of((short)
42), 42)
+ .fromCase(VARIANT(), Variant.newBuilder().of(42), 42)
+ .fromCase(VARIANT(), Variant.newBuilder().of(42L), 42)
+ // narrowing from a floating point or decimal value
truncates
+ .fromCase(VARIANT(), Variant.newBuilder().of(3.9d), 3)
+ .fromCase(VARIANT(), Variant.newBuilder().of(new
BigDecimal("7.2")), 7)
+ // a non-numeric variant cannot be cast to a number
+ .fail(
+ VARIANT(),
+ Variant.newBuilder().of("foo"),
+ TableRuntimeException.class)
+ .fail(
+ VARIANT(),
+ Variant.newBuilder().of(true),
+ TableRuntimeException.class),
+ CastTestSpecBuilder.testCastTo(BIGINT())
+ .fromCase(VARIANT(), Variant.newBuilder().of(42L), 42L)
+ .fromCase(VARIANT(), Variant.newBuilder().of(42), 42L)
+ .fail(VARIANT(), Variant.newBuilder().of("x"),
TableRuntimeException.class),
+ CastTestSpecBuilder.testCastTo(FLOAT())
+ .fromCase(VARIANT(), Variant.newBuilder().of(1.5f),
1.5f)
+ .fromCase(VARIANT(), Variant.newBuilder().of(1.5d),
1.5f)
+ .fromCase(VARIANT(), Variant.newBuilder().of(3), 3.0f)
+ .fail(VARIANT(), Variant.newBuilder().of("x"),
TableRuntimeException.class),
+ CastTestSpecBuilder.testCastTo(DOUBLE())
+ .fromCase(VARIANT(), Variant.newBuilder().of(1.5d),
1.5d)
+ .fromCase(VARIANT(), Variant.newBuilder().of(1.5f),
1.5d)
+ .fromCase(VARIANT(), Variant.newBuilder().of(3), 3.0d)
+ .fail(VARIANT(), Variant.newBuilder().of("x"),
TableRuntimeException.class),
+ CastTestSpecBuilder.testCastTo(DECIMAL(5, 2))
+ .fromCase(VARIANT(), null, null)
+ .fromCase(
+ VARIANT(),
+ Variant.newBuilder().of(new
BigDecimal("123.45")),
+ DecimalData.fromBigDecimal(new
BigDecimal("123.45"), 5, 2))
+ .fromCase(
+ VARIANT(),
+ Variant.newBuilder().of(42),
+ DecimalData.fromBigDecimal(new
BigDecimal("42"), 5, 2))
+ .fail(VARIANT(), Variant.newBuilder().of("x"),
TableRuntimeException.class),
+ CastTestSpecBuilder.testCastTo(BYTES())
+ .fromCase(VARIANT(), null, null)
+ .fromCase(
+ VARIANT(),
+ Variant.newBuilder().of(new byte[] {1, 2, 3}),
+ new byte[] {1, 2, 3})
+ .fail(
+ VARIANT(),
+ Variant.newBuilder().of("foo"),
+ TableRuntimeException.class),
+ CastTestSpecBuilder.testCastTo(DATE())
+ .fromCase(
+ VARIANT(),
+ Variant.newBuilder().of(LocalDate.of(2020, 1,
1)),
+ (int) LocalDate.of(2020, 1, 1).toEpochDay())
+ .fail(VARIANT(), Variant.newBuilder().of(1),
TableRuntimeException.class),
+ CastTestSpecBuilder.testCastTo(TIMESTAMP())
+ .fromCase(VARIANT(), null, null)
+ .fromCase(
+ VARIANT(),
+ Variant.newBuilder().of(LocalDateTime.of(2020,
1, 1, 12, 0, 0)),
+ TimestampData.fromLocalDateTime(
+ LocalDateTime.of(2020, 1, 1, 12, 0,
0)))
+ .fail(VARIANT(), Variant.newBuilder().of(1),
TableRuntimeException.class),
+ CastTestSpecBuilder.testCastTo(TIMESTAMP_LTZ())
+ .fromCase(
+ VARIANT(),
+
Variant.newBuilder().of(Instant.ofEpochSecond(1_600_000_000L)),
+
TimestampData.fromInstant(Instant.ofEpochSecond(1_600_000_000L)))
+ .fail(VARIANT(), Variant.newBuilder().of(1),
TableRuntimeException.class));
}
@TestFactory
