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 4ad8e8372f2 [FLINK-37926][table] Support casting from VARIANT to ROW
and STRUCTURED
4ad8e8372f2 is described below
commit 4ad8e8372f28a8aeac2b4b28077bfed3049e9b7a
Author: Ramin Gharib <[email protected]>
AuthorDate: Mon Sep 7 10:30:06 2026 +0200
[FLINK-37926][table] Support casting from VARIANT to ROW and STRUCTURED
This closes #29092.
---
docs/content.zh/docs/sql/reference/data-types.md | 29 ++-
docs/content/docs/sql/reference/data-types.md | 29 ++-
.../types/logical/utils/LogicalTypeCasts.java | 5 +
.../flink/table/types/LogicalTypeCastsTest.java | 24 +++
.../AbstractVariantToConstructedCastRule.java | 6 +-
.../functions/casting/CastRuleProvider.java | 1 +
.../functions/casting/VariantToRowCastRule.java | 234 +++++++++++++++++++++
.../planner/functions/CastFunctionITCase.java | 109 ++++++++++
.../functions/casting/CastRuleProviderTest.java | 15 ++
.../planner/functions/casting/CastRulesTest.java | 187 +++++++++++++++-
.../table/runtime/functions/VariantCastUtils.java | 10 +
11 files changed, 643 insertions(+), 6 deletions(-)
diff --git a/docs/content.zh/docs/sql/reference/data-types.md
b/docs/content.zh/docs/sql/reference/data-types.md
index b1e938567f2..61363906c27 100644
--- a/docs/content.zh/docs/sql/reference/data-types.md
+++ b/docs/content.zh/docs/sql/reference/data-types.md
@@ -1614,6 +1614,33 @@ CAST(m AS ARRAY<STRING>) -- ['1', 'a'], a
heterogeneous array still renders ea
CAST(m AS ARRAY<VARIANT>) -- [1, "a"] as variants, one level shredded
```
+A variant object casts to `ROW` or `STRUCTURED`, which likewise imposes a
schema on it. The variant
+must be an object, otherwise the cast fails. Each field is itself a `VARIANT`,
so it casts to its
+declared type by the same rules, recursively. Fields match by name and name
matching is case
+sensitive. A `ROW` declared without field names uses the default names `f0`,
`f1`, and so on, which
+must then be present in the object.
+
+- A field absent from the object fails the cast, whether the target field is
nullable or not.
+- A field present but set to a variant null maps to SQL `NULL` when the field
is nullable and fails
+ the cast when the target is `NOT NULL`.
+- Object fields the target does not name are dropped, so the row is a
projection.
+- A `ROW` or `STRUCTURED` whose fields are `VARIANT` is the identity on those
fields: it shreds one
+ level and keeps the rest semi-structured. A field set to a variant null
stays a variant null
+ rather than becoming SQL `NULL`.
+
+If any field cast fails, the whole cast fails, and `TRY_CAST` returns `NULL`
for the entire value
+rather than a partial result.
+
+The following examples use `o` for `PARSE_JSON('{"id": 7, "name": "ada",
"email": null}')`:
+
+```sql
+CAST(o AS ROW<`id` INT, `name` STRING>) -- (7, 'ada')
+CAST(o AS ROW<`name` STRING, `id` INT>) -- ('ada', 7), the order of
target fields is free
+CAST(o AS ROW<`id` INT, `email` STRING>) -- (7, NULL), a field present
as a variant null maps to NULL
+CAST(o AS ROW<`id` INT, `phone` STRING>) -- fails, the field 'phone' is
not present in the VARIANT
+CAST(o AS ROW<`id` VARIANT, `email` VARIANT>) -- (7, null), each field kept
as a variant, the variant null preserved
+```
+
**Declaration**
{{< tabs "25c30432-8460-441d-a036-9416d8202882" >}}
@@ -1828,7 +1855,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 | Y | N |
+| `VARIANT` | 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 442536c212e..0c55f7eb9b8 100644
--- a/docs/content/docs/sql/reference/data-types.md
+++ b/docs/content/docs/sql/reference/data-types.md
@@ -1622,6 +1622,33 @@ CAST(m AS ARRAY<STRING>) -- ['1', 'a'], a
heterogeneous array still renders ea
CAST(m AS ARRAY<VARIANT>) -- [1, "a"] as variants, one level shredded
```
+A variant object casts to `ROW` or `STRUCTURED`, which likewise imposes a
schema on it. The variant
+must be an object, otherwise the cast fails. Each field is itself a `VARIANT`,
so it casts to its
+declared type by the same rules, recursively. Fields match by name and name
matching is case
+sensitive. A `ROW` declared without field names uses the default names `f0`,
`f1`, and so on, which
+must then be present in the object.
+
+- A field absent from the object fails the cast, whether the target field is
nullable or not.
+- A field present but set to a variant null maps to SQL `NULL` when the field
is nullable and fails
+ the cast when the target is `NOT NULL`.
+- Object fields the target does not name are dropped, so the row is a
projection.
+- A `ROW` or `STRUCTURED` whose fields are `VARIANT` is the identity on those
fields: it shreds one
+ level and keeps the rest semi-structured. A field set to a variant null
stays a variant null
+ rather than becoming SQL `NULL`.
+
+If any field cast fails, the whole cast fails, and `TRY_CAST` returns `NULL`
for the entire value
+rather than a partial result.
+
+The following examples use `o` for `PARSE_JSON('{"id": 7, "name": "ada",
"email": null}')`:
+
+```sql
+CAST(o AS ROW<`id` INT, `name` STRING>) -- (7, 'ada')
+CAST(o AS ROW<`name` STRING, `id` INT>) -- ('ada', 7), the order of
target fields is free
+CAST(o AS ROW<`id` INT, `email` STRING>) -- (7, NULL), a field present
as a variant null maps to NULL
+CAST(o AS ROW<`id` INT, `phone` STRING>) -- fails, the field 'phone' is
not present in the VARIANT
+CAST(o AS ROW<`id` VARIANT, `email` VARIANT>) -- (7, null), each field kept
as a variant, the variant null preserved
+```
+
**Declaration**
{{< tabs "25c30432-8460-441d-a036-9416d8202882" >}}
@@ -1837,7 +1864,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 | Y | N |
+| `VARIANT` | 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/logical/utils/LogicalTypeCasts.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeCasts.java
index 0e3e6b38e7b..8e6d1a47ce0 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
@@ -685,6 +685,11 @@ public final class LogicalTypeCasts {
// array cast rule; a per-element mismatch fails there, not here.
return allowExplicit
&& supportsCasting(sourceType, ((ArrayType)
targetType).getElementType(), true);
+ } else if (sourceRoot == VARIANT && (targetRoot == ROW || targetRoot
== STRUCTURED_TYPE)) {
+ // A variant object casts to ROW or STRUCTURED when VARIANT casts
to every field type.
+ return allowExplicit
+ && targetType.getChildren().stream()
+ .allMatch(field -> supportsCasting(sourceType,
field, true));
} else if (sourceRoot == RAW
&& !targetType.is(BINARY_STRING)
&& !targetType.is(CHARACTER_STRING)
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 ac293be2a25..75a99bc5e22 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
@@ -60,6 +60,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;
@@ -308,6 +309,29 @@ class LogicalTypeCastsTest {
YearMonthIntervalType.YearMonthResolution.MONTH)),
false,
false),
+ // A variant object casts to ROW or STRUCTURED when every
field is castable; an
+ // empty
+ // row is vacuously castable and matching is by name
+ Arguments.of(new VariantType(), new RowType(List.of()), false,
true),
+ Arguments.of(
+ new VariantType(),
+ new RowType(
+ List.of(
+ new RowField("f0", new IntType()),
+ new RowField("f1",
VarCharType.STRING_TYPE))),
+ false,
+ true),
+ Arguments.of(
+ new VariantType(),
+ new RowType(
+ List.of(
+ new RowField(
+ "f0",
+ new YearMonthIntervalType(
+
YearMonthIntervalType.YearMonthResolution
+ .MONTH)))),
+ false,
+ false),
// MULTISET has no variant counterpart and stays unsupported
Arguments.of(
new VariantType(), new
MultisetType(VarCharType.STRING_TYPE), false, false),
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/AbstractVariantToConstructedCastRule.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/AbstractVariantToConstructedCastRule.java
index f003b129ec2..e6d924617f3 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/AbstractVariantToConstructedCastRule.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/AbstractVariantToConstructedCastRule.java
@@ -30,9 +30,9 @@ import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.met
* a shape check at each level, so the recursion bottoms out at the same
scalar cast the primitive
* and string rules perform and no new leaf semantics are introduced.
*
- * <p>A constructed cast can always fail, on a shape mismatch, an unreadable
leaf, or a missing
- * {@code NOT NULL} field, so {@code TRY_CAST} wraps the whole value and
returns {@code NULL} for
- * any failure rather than a partial result.
+ * <p>A constructed cast can always fail, on a shape mismatch, an unreadable
leaf, a missing field,
+ * or a VARIANT {@code null} in a {@code NOT NULL} position, so {@code
TRY_CAST} wraps the whole
+ * value and returns {@code NULL} for any failure rather than a partial result.
*/
abstract class AbstractVariantToConstructedCastRule<OUT>
extends AbstractNullAwareCodeGeneratorCastRule<Variant, OUT> {
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 99fa9b63f2a..d16012b0afe 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
@@ -99,6 +99,7 @@ public class CastRuleProvider {
.addRule(VariantToStringCastRule.INSTANCE)
.addRule(VariantToPrimitiveCastRule.INSTANCE)
.addRule(VariantToArrayCastRule.INSTANCE)
+ .addRule(VariantToRowCastRule.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/VariantToRowCastRule.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/VariantToRowCastRule.java
new file mode 100644
index 00000000000..ade8e61f128
--- /dev/null
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/VariantToRowCastRule.java
@@ -0,0 +1,234 @@
+/*
+ * 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.api.TableRuntimeException;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.data.writer.BinaryRowWriter;
+import org.apache.flink.table.planner.codegen.CodeGeneratorContext;
+import org.apache.flink.table.runtime.functions.VariantCastUtils;
+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.util.List;
+
+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.binaryWriterWriteField;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.binaryWriterWriteNull;
+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.strLiteral;
+
+/**
+ * {@link LogicalTypeRoot#VARIANT} to {@link LogicalTypeRoot#ROW} and {@link
+ * LogicalTypeRoot#STRUCTURED_TYPE} cast rule.
+ *
+ * <p>The variant must be an object, otherwise the cast fails. Fields match
<b>by name</b> rather
+ * than by position, because a VARIANT object is unordered. Name matching is
case sensitive. A
+ * target field absent from the object fails the cast. A field present but set
to a VARIANT null
+ * maps to SQL {@code NULL} when it is nullable and fails the cast when it is
{@code NOT NULL}. A
+ * {@link LogicalTypeRoot#VARIANT} target field is the exception: there the
field cast is the
+ * identity, so a VARIANT null is kept as a variant null rather than
downgraded to SQL {@code NULL}.
+ * Object fields the target does not name are dropped, so the row is a
projection. A {@code
+ * STRUCTURED} target shares the {@code RowData} representation and is served
by the same rule.
+ */
+class VariantToRowCastRule extends
AbstractVariantToConstructedCastRule<RowData> {
+
+ static final VariantToRowCastRule INSTANCE = new VariantToRowCastRule();
+
+ private VariantToRowCastRule() {
+
super(CastRulePredicate.builder().predicate(VariantToRowCastRule::matches).build());
+ }
+
+ private static boolean matches(LogicalType input, LogicalType target) {
+ if (!(input.is(LogicalTypeRoot.VARIANT)
+ && (target.is(LogicalTypeRoot.ROW)
+ || target.is(LogicalTypeRoot.STRUCTURED_TYPE)))) {
+ return false;
+ }
+ return LogicalTypeChecks.getFieldTypes(target).stream()
+ .allMatch(fieldType -> CastRuleProvider.resolve(input,
fieldType) != null);
+ }
+
+ /* Example generated code for ROW<`id` INT, `raw` VARIANT>. The typed `id`
field runs the leaf
+ cast; the `raw` field runs the identity cast and is written as-is, so a
VARIANT null is kept
+ rather than turned into SQL NULL:
+
+ org.apache.flink.table.runtime.functions.VariantCastUtils.requireObject(
+ variant$2, "ROW<`id` INT, `raw` VARIANT>");
+ writer$4.reset();
+ org.apache.flink.types.variant.Variant f0Value$5 =
variant$2.getField("id");
+ if (f0Value$5 == null) {
+ throw new org.apache.flink.table.api.TableRuntimeException(
+ "Cannot cast the VARIANT object to ROW<`id` INT, `raw`
VARIANT> because the field "
+ + "'id' is not present in the VARIANT.");
+ } else {
+ if (f0Value$5.isNull()) {
+ writer$4.setNullAt(0);
+ } else {
+ result$6 =
+ ((int)
org.apache.flink.table.runtime.functions.VariantCastUtils.toIntegral(
+ f0Value$5, -2147483648L, 2147483647L, "INTEGER"));
+ if (!isNull$7) {
+ writer$4.writeInt(0, result$6);
+ } else {
+ writer$4.setNullAt(0);
+ }
+ }
+ }
+ org.apache.flink.types.variant.Variant f1Value$8 =
variant$2.getField("raw");
+ if (f1Value$8 == null) {
+ throw new org.apache.flink.table.api.TableRuntimeException(
+ "Cannot cast the VARIANT object to ROW<`id` INT, `raw`
VARIANT> because the field "
+ + "'raw' is not present in the VARIANT.");
+ } else {
+ writer$4.writeVariant(1, f1Value$8);
+ }
+ writer$4.complete();
+ result$3 = row$1.copy();
+
+ A NOT NULL typed field throws instead of the setNullAt(i) shown above when
its value is a
+ VARIANT null.
+
+ */
+ @Override
+ protected String generateCodeBlockInternal(
+ CodeGeneratorCastRule.Context context,
+ String inputTerm,
+ String returnVariable,
+ LogicalType inputLogicalType,
+ LogicalType targetLogicalType) {
+ final List<String> fieldNames =
LogicalTypeChecks.getFieldNames(targetLogicalType);
+ final List<LogicalType> fieldTypes =
LogicalTypeChecks.getFieldTypes(targetLogicalType);
+ final CodeGeneratorContext codeGeneratorContext =
context.getCodeGeneratorContext();
+
+ final String rowTerm = newName(codeGeneratorContext, "row");
+ final String writerTerm = newName(codeGeneratorContext, "writer");
+ context.declareClassField(
+ className(BinaryRowData.class),
+ rowTerm,
+ constructorCall(BinaryRowData.class, fieldTypes.size()));
+ context.declareClassField(
+ className(BinaryRowWriter.class),
+ writerTerm,
+ constructorCall(BinaryRowWriter.class, rowTerm));
+
+ final CastRuleUtils.CodeWriter writer =
+ new CastRuleUtils.CodeWriter()
+ .stmt(
+ staticCall(
+ VariantCastUtils.class,
+ "requireObject",
+ inputTerm,
+
strLiteral(targetLogicalType.asSummaryString())))
+ .stmt(methodCall(writerTerm, "reset"));
+
+ for (int i = 0; i < fieldTypes.size(); i++) {
+ final LogicalType fieldType = fieldTypes.get(i);
+ final String fieldName = fieldNames.get(i);
+ final String indexTerm = String.valueOf(i);
+ final String fieldTerm = newName(codeGeneratorContext, "f" +
indexTerm + "Value");
+
+ // The field is guaranteed present and non-null here, since a
missing field and a
+ // VARIANT
+ // null are handled below, so the inner cast is the plain
VARIANT-to-field rule.
+ final CastCodeBlock codeBlock =
+ CastRuleProvider.generateAlwaysNonNullCodeBlock(
+ context, fieldTerm, inputLogicalType, fieldType);
+ final String writeField =
+ binaryWriterWriteField(
+ context, writerTerm, fieldType, indexTerm,
codeBlock.getReturnTerm());
+ final String writeNull = binaryWriterWriteNull(writerTerm,
fieldType, indexTerm);
+
+ // getField returns Java null for a field the object does not
carry, which always fails
+ // the cast.
+ writer.declStmt(
+ Variant.class,
+ fieldTerm,
+ methodCall(inputTerm, "getField", strLiteral(fieldName)));
+
+ if (fieldType.is(LogicalTypeRoot.VARIANT)) {
+ // The field cast is the identity, so a present VARIANT null
field is a valid
+ // variant
+ // null and is kept as-is rather than downgraded to SQL NULL,
matching
+ // ARRAY<VARIANT>
+ // and the top-level VARIANT cast. An absent field still fails
the cast.
+ writer.ifStmt(
+ fieldTerm + " == null",
+ absentWriter ->
+ absentWriter.throwStmt(
+ missingFieldError(fieldName,
targetLogicalType)),
+ presentWriter ->
presentWriter.append(codeBlock).stmt(writeField));
+ continue;
+ }
+
+ // A field explicitly set to a VARIANT null returns a variant
whose isNull() is true; it
+ // maps to SQL NULL for a nullable field and fails the cast for a
NOT NULL one.
+ writer.ifStmt(
+ fieldTerm + " == null",
+ absentWriter ->
+
absentWriter.throwStmt(missingFieldError(fieldName, targetLogicalType)),
+ presentWriter ->
+ presentWriter.ifStmt(
+ methodCall(fieldTerm, "isNull"),
+ nullWriter -> {
+ if (fieldType.isNullable()) {
+ nullWriter.stmt(writeNull);
+ } else {
+ nullWriter.throwStmt(
+ nullFieldError(fieldName,
targetLogicalType));
+ }
+ },
+ valueWriter ->
+ valueWriter
+ .append(codeBlock)
+ .ifStmt(
+ "!" +
codeBlock.getIsNullTerm(),
+ thenWriter ->
+
thenWriter.stmt(writeField),
+ elseWriter ->
+
elseWriter.stmt(writeNull))));
+ }
+
+ writer.stmt(methodCall(writerTerm, "complete"))
+ .assignStmt(returnVariable, methodCall(rowTerm, "copy"));
+ return writer.toString();
+ }
+
+ private static String missingFieldError(String fieldName, LogicalType
targetType) {
+ final String message =
+ String.format(
+ "Cannot cast the VARIANT object to %s because the
field '%s' is not present in the VARIANT.",
+ targetType.asSummaryString(), fieldName);
+ return constructorCall(TableRuntimeException.class,
strLiteral(message));
+ }
+
+ private static String nullFieldError(String fieldName, LogicalType
targetType) {
+ final String message =
+ String.format(
+ "Cannot cast the VARIANT object to %s because the
field '%s' is a VARIANT null and the target does not accept NULL.",
+ targetType.asSummaryString(), fieldName);
+ return constructorCall(TableRuntimeException.class,
strLiteral(message));
+ }
+}
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 394827b5e60..2904c874f6c 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
@@ -158,6 +158,7 @@ public class CastFunctionITCase extends
BuiltInFunctionTestBase {
final List<TestSetSpec> specs = new ArrayList<>();
specs.addAll(variantPrimitiveCasts());
specs.addAll(variantArrayCasts());
+ specs.addAll(variantRowCasts());
return specs;
}
@@ -566,6 +567,114 @@ public class CastFunctionITCase extends
BuiltInFunctionTestBase {
BYTES().notNull()));
}
+ private static List<TestSetSpec> variantRowCasts() {
+ final String obj = "{\"id\": 7, \"name\": \"ada\", \"active\": true}";
+ final String objNull = "{\"id\": 7, \"name\": null}";
+ final String nested =
+ "{\"user\": {\"id\": 1, \"since\": \"2020-01-01\"}, \"tags\":
[\"x\", \"y\"]}";
+ return List.of(
+ TestSetSpec.forExpression("Cast a VARIANT produced by
parseJson() to a ROW")
+ .onFieldsWithData("unused")
+ .andDataTypes(STRING())
+ // ROW: fields match by name, order is free
+ .testResult(
+ lit(obj).parseJson()
+ .cast(ROW(FIELD("id", INT()),
FIELD("name", STRING()))),
+ "CAST(PARSE_JSON('" + obj + "') AS ROW<`id`
INT, `name` STRING>)",
+ Row.of(7, "ada"),
+ ROW(FIELD("id", INT()), FIELD("name",
STRING())).notNull())
+ .testResult(
+ lit(obj).parseJson()
+ .cast(ROW(FIELD("name", STRING()),
FIELD("id", INT()))),
+ "CAST(PARSE_JSON('" + obj + "') AS ROW<`name`
STRING, `id` INT>)",
+ Row.of("ada", 7),
+ ROW(FIELD("name", STRING()), FIELD("id",
INT())).notNull())
+ // a field absent from the object fails the cast
+ .testSqlRuntimeError(
+ "CAST(PARSE_JSON('"
+ + obj
+ + "') AS ROW<`id` INT, `non-existing`
STRING>)",
+ TableRuntimeException.class,
+ "is not present in the VARIANT")
+ // a field present but set to a variant null maps to
SQL NULL when nullable
+ .testResult(
+ lit(objNull)
+ .parseJson()
+ .cast(ROW(FIELD("id", INT()),
FIELD("name", STRING()))),
+ "CAST(PARSE_JSON('"
+ + objNull
+ + "') AS ROW<`id` INT, `name`
STRING>)",
+ Row.of(7, null),
+ ROW(FIELD("id", INT()), FIELD("name",
STRING())).notNull())
+ // and fails when that field is NOT NULL
+ .testSqlRuntimeError(
+ "CAST(PARSE_JSON('"
+ + objNull
+ + "') AS ROW<`id` INT, `name` STRING
NOT NULL>)",
+ TableRuntimeException.class,
+ "does not accept NULL")
+ // extra object fields are dropped, so the row is a
projection
+ .testResult(
+ lit(obj).parseJson().cast(ROW(FIELD("id",
INT()))),
+ "CAST(PARSE_JSON('" + obj + "') AS ROW<`id`
INT>)",
+ Row.of(7),
+ ROW(FIELD("id", INT())).notNull())
+ // an array is not an object
+ .testTableApiRuntimeError(
+ lit("[1, 2,
3]").parseJson().cast(ROW(FIELD("id", INT()))),
+ "requires an object")
+ // ROW<VARIANT> shreds one level, keeping each field a
variant that then
+ // casts back unchanged
+ .testResult(
+ lit(obj).parseJson()
+ .cast(ROW(FIELD("id", VARIANT()),
FIELD("name", VARIANT())))
+ .cast(ROW(FIELD("id", INT()),
FIELD("name", STRING()))),
+ "CAST(CAST(PARSE_JSON('"
+ + obj
+ + "') AS ROW<`id` VARIANT, `name`
VARIANT>)"
+ + " AS ROW<`id` INT, `name` STRING>)",
+ Row.of(7, "ada"),
+ ROW(FIELD("id", INT()), FIELD("name",
STRING())).notNull())
+ // a variant null field is kept as a variant null, so
casting it back to a
+ // concrete nullable type yields SQL NULL
+ .testResult(
+ lit(objNull)
+ .parseJson()
+ .cast(ROW(FIELD("id", VARIANT()),
FIELD("name", VARIANT())))
+ .cast(ROW(FIELD("id", INT()),
FIELD("name", STRING()))),
+ "CAST(CAST(PARSE_JSON('"
+ + objNull
+ + "') AS ROW<`id` VARIANT, `name`
VARIANT>)"
+ + " AS ROW<`id` INT, `name` STRING>)",
+ Row.of(7, null),
+ ROW(FIELD("id", INT()), FIELD("name",
STRING())).notNull())
+ // the recursion composes for nested rows and arrays
+ .testResult(
+ lit(nested)
+ .parseJson()
+ .cast(
+ ROW(
+ FIELD(
+ "user",
+ ROW(
+
FIELD("id", INT()),
+
FIELD("since", STRING()))),
+ FIELD("tags",
ARRAY(STRING())))),
+ "CAST(PARSE_JSON('"
+ + nested
+ + "') AS ROW<`user` ROW<`id` INT,
`since` STRING>,"
+ + " `tags` ARRAY<STRING>>)",
+ Row.of(Row.of(1, "2020-01-01"), new String[]
{"x", "y"}),
+ ROW(
+ FIELD(
+ "user",
+ ROW(
+ FIELD("id",
INT()),
+ FIELD("since",
STRING()))),
+ FIELD("tags", ARRAY(STRING())))
+ .notNull()));
+ }
+
private static List<TestSetSpec> allTypesBasic() {
return Arrays.asList(
CastTestSpecBuilder.testCastTo(BOOLEAN())
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 fabe82749f4..19fcd689362 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
@@ -157,4 +157,19 @@ class CastRuleProviderTest {
// MULTISET has no variant counterpart
assertThat(CastRuleProvider.exists(VARIANT,
MULTISET(STRING()).getLogicalType())).isFalse();
}
+
+ @Test
+ void testResolveVariantToRow() {
+ assertThat(CastRuleProvider.resolve(VARIANT, ROW(FIELD("f0",
INT())).getLogicalType()))
+ .isSameAs(VariantToRowCastRule.INSTANCE);
+ // a structured target shares the ROW rule
+ assertThat(CastRuleProvider.resolve(VARIANT, STRUCTURED))
+ .isSameAs(VariantToRowCastRule.INSTANCE);
+
+ // a field with no variant counterpart makes the whole cast
unresolvable
+ assertThat(
+ CastRuleProvider.exists(
+ VARIANT, ROW(FIELD("f0",
MULTISET(STRING()))).getLogicalType()))
+ .isFalse();
+ }
}
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 2ba4de50381..8ecf6556bb5 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.BinaryVariant;
import org.apache.flink.types.variant.Variant;
import org.apache.flink.types.variant.VariantBuilder;
@@ -261,6 +262,71 @@ class CastRulesTest {
private static final Variant VARIANT_EMPTY_ARRAY =
VARIANT_BUILDER.array().build();
+ /** {@code {"id": 7, "name": "ada", "active": true}}, the design's running
object value. */
+ private static final Variant VARIANT_RECORD =
+ Variant.newBuilder()
+ .object()
+ .add("id", Variant.newBuilder().of(7))
+ .add("name", Variant.newBuilder().of("ada"))
+ .add("active", Variant.newBuilder().of(true))
+ .build();
+
+ /**
+ * {@code {"id": 7, "email": null}}, an object with a field explicitly set
to a VARIANT null.
+ */
+ private static final Variant VARIANT_RECORD_WITH_NULL =
+ Variant.newBuilder()
+ .object()
+ .add("id", Variant.newBuilder().of(7))
+ .add("email", Variant.newBuilder().ofNull())
+ .build();
+
+ /** {@code {"user": {"id": 1, "since": "2020-01-01"}, "tags": ["x",
"y"]}}, a nested value. */
+ private static final Variant VARIANT_NESTED =
+ Variant.newBuilder()
+ .object()
+ .add(
+ "user",
+ Variant.newBuilder()
+ .object()
+ .add("id", Variant.newBuilder().of(1))
+ .add("since",
Variant.newBuilder().of("2020-01-01"))
+ .build())
+ .add(
+ "tags",
+ Variant.newBuilder()
+ .array()
+ .add(Variant.newBuilder().of("x"))
+ .add(Variant.newBuilder().of("y"))
+ .build())
+ .build();
+
+ /** {@code {"f0": 7, "f1": "ada"}}, an object keyed by the default {@code
ROW} field names. */
+ private static final Variant VARIANT_POSITIONAL_RECORD =
+ VARIANT_BUILDER
+ .object()
+ .add("f0", VARIANT_BUILDER.of(7))
+ .add("f1", VARIANT_BUILDER.of("ada"))
+ .build();
+
+ /**
+ * {@code {"a": 1, "b": 2, "c": "x", "d": ["p", "q"]}}, shaped for {@link
#MY_STRUCTURED_TYPE}.
+ */
+ private static final Variant VARIANT_STRUCT_RECORD =
+ VARIANT_BUILDER
+ .object()
+ .add("a", VARIANT_BUILDER.of(1L))
+ .add("b", VARIANT_BUILDER.of(2L))
+ .add("c", VARIANT_BUILDER.of("x"))
+ .add(
+ "d",
+ VARIANT_BUILDER
+ .array()
+ .add(VARIANT_BUILDER.of("p"))
+ .add(VARIANT_BUILDER.of("q"))
+ .build())
+ .build();
+
private static final DataType MY_STRUCTURED_TYPE =
STRUCTURED(
MyStructuredType.class,
@@ -283,6 +349,13 @@ class CastRulesTest {
"d",
ARRAY(STRING()).getLogicalType())))
.build());
+ // Rebuilds a variant object field at position 0, matching the form a ROW
cast produces when it
+ // serializes each field with BinaryRowWriter.writeVariant and reads it
back.
+ private static Variant rowFieldVariant(Variant fieldView) {
+ final BinaryVariant view = (BinaryVariant) fieldView;
+ return new BinaryVariant(view.getValue(), view.getMetadata());
+ }
+
Stream<CastTestSpecBuilder> testCases() {
return Stream.of(
CastTestSpecBuilder.testCastTo(TINYINT())
@@ -2072,7 +2145,119 @@ class CastRulesTest {
VARIANT_INT_ARRAY_WITH_NULL.getElement(0),
VARIANT_INT_ARRAY_WITH_NULL.getElement(1),
VARIANT_INT_ARRAY_WITH_NULL.getElement(2)
- })));
+ })),
+ CastTestSpecBuilder.testCastTo(ROW(FIELD("id", INT()),
FIELD("name", STRING())))
+ .fromCase(VARIANT(), null, null)
+ .fromCase(
+ VARIANT(), VARIANT_RECORD,
GenericRowData.of(7, fromString("ada")))
+ // an array or a scalar is not an object
+ .fail(
+ VARIANT(),
+ VARIANT_INT_ARRAY,
+ TableRuntimeException.class,
+ "requires an object"),
+ // field order of the target is free, since matching is by name
+ CastTestSpecBuilder.testCastTo(ROW(FIELD("name", STRING()),
FIELD("id", INT())))
+ .fromCase(
+ VARIANT(), VARIANT_RECORD,
GenericRowData.of(fromString("ada"), 7)),
+ // a field absent from the object fails the cast, nullable or
not
+ CastTestSpecBuilder.testCastTo(ROW(FIELD("id", INT()),
FIELD("email", STRING())))
+ .fail(
+ VARIANT(),
+ VARIANT_RECORD,
+ TableRuntimeException.class,
+ "is not present in the VARIANT"),
+ CastTestSpecBuilder.testCastTo(
+ ROW(FIELD("id", INT()), FIELD("email",
STRING().notNull())))
+ .fail(
+ VARIANT(),
+ VARIANT_RECORD,
+ TableRuntimeException.class,
+ "is not present in the VARIANT"),
+ // a field present but set to a VARIANT null maps to NULL when
nullable
+ CastTestSpecBuilder.testCastTo(ROW(FIELD("id", INT()),
FIELD("email", STRING())))
+ .fromCase(VARIANT(), VARIANT_RECORD_WITH_NULL,
GenericRowData.of(7, null)),
+ // and fails when the field is NOT NULL
+ CastTestSpecBuilder.testCastTo(
+ ROW(FIELD("id", INT()), FIELD("email",
STRING().notNull())))
+ .fail(
+ VARIANT(),
+ VARIANT_RECORD_WITH_NULL,
+ TableRuntimeException.class,
+ "does not accept NULL"),
+ // extra object fields are dropped, so the row is a projection
+ CastTestSpecBuilder.testCastTo(ROW(FIELD("id", INT())))
+ .fromCase(VARIANT(), VARIANT_RECORD,
GenericRowData.of(7)),
+ // a ROW<VARIANT> field is the identity: each field is kept as
a variant, one level
+ // shredded
+ CastTestSpecBuilder.testCastTo(
+ ROW(FIELD("id", VARIANT()), FIELD("name",
VARIANT())))
+ .fromCase(
+ VARIANT(),
+ VARIANT_RECORD,
+ GenericRowData.of(
+
rowFieldVariant(VARIANT_RECORD.getField("id")),
+
rowFieldVariant(VARIANT_RECORD.getField("name")))),
+ // a VARIANT target field keeps a variant null as a variant
null, not SQL NULL
+ CastTestSpecBuilder.testCastTo(
+ ROW(FIELD("id", VARIANT()), FIELD("email",
VARIANT())))
+ .fromCase(
+ VARIANT(),
+ VARIANT_RECORD_WITH_NULL,
+ GenericRowData.of(
+
rowFieldVariant(VARIANT_RECORD_WITH_NULL.getField("id")),
+ rowFieldVariant(
+
VARIANT_RECORD_WITH_NULL.getField("email")))),
+ // a typed field beside a VARIANT field keeps the variant null
only on the VARIANT
+ // side
+ CastTestSpecBuilder.testCastTo(ROW(FIELD("id", INT()),
FIELD("email", VARIANT())))
+ .fromCase(
+ VARIANT(),
+ VARIANT_RECORD_WITH_NULL,
+ GenericRowData.of(
+ 7,
+ rowFieldVariant(
+
VARIANT_RECORD_WITH_NULL.getField("email")))),
+ // a ROW without declared field names uses the default names
f0, f1, ...; matching
+ // is
+ // still by name, not by position
+ CastTestSpecBuilder.testCastTo(ROW(INT(), STRING()))
+ .fromCase(
+ VARIANT(),
+ VARIANT_POSITIONAL_RECORD,
+ GenericRowData.of(7, fromString("ada")))
+ // an object without the default names fails, position
is never used
+ .fail(
+ VARIANT(),
+ VARIANT_RECORD,
+ TableRuntimeException.class,
+ "is not present in the VARIANT"),
+ // a STRUCTURED target is served by the same rule, matching
attributes to object
+ // fields by name
+ CastTestSpecBuilder.testCastTo(MY_STRUCTURED_TYPE)
+ .fromCase(
+ VARIANT(),
+ VARIANT_STRUCT_RECORD,
+ GenericRowData.of(
+ 1L,
+ 2L,
+ fromString("x"),
+ new GenericArrayData(
+ new Object[] {fromString("p"),
fromString("q")}))),
+ // the recursion composes: a row of a row and an array with no
special case
+ CastTestSpecBuilder.testCastTo(
+ ROW(
+ FIELD(
+ "user",
+ ROW(FIELD("id", INT()),
FIELD("since", STRING()))),
+ FIELD("tags", ARRAY(STRING()))))
+ .fromCase(
+ VARIANT(),
+ VARIANT_NESTED,
+ GenericRowData.of(
+ GenericRowData.of(1,
fromString("2020-01-01")),
+ new GenericArrayData(
+ new Object[] {fromString("x"),
fromString("y")}))));
}
@TestFactory
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 ac8d550195c..f273575cdbe 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
@@ -74,6 +74,16 @@ public final class VariantCastUtils {
throw wrongShape(variant, targetType, "an array");
}
+ /**
+ * Fails when the variant is not an object, so a cast to {@code ROW},
{@code STRUCTURED}, or
+ * {@code MAP} reports a shape mismatch clearly. Only an object carries
named fields.
+ */
+ public static void requireObject(Variant variant, String targetType) {
+ if (!variant.isObject()) {
+ throw wrongShape(variant, targetType, "an object");
+ }
+ }
+
private static TableRuntimeException wrongShape(
Variant variant, String targetType, String required) {
return new TableRuntimeException(
