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 78b0df60e34 [FLINK-37926][table] Support casting from VARIANT to MAP
78b0df60e34 is described below
commit 78b0df60e341c7c0996b2bd9f511acf9e293a647
Author: Ramin Gharib <[email protected]>
AuthorDate: Mon Sep 7 12:11:24 2026 +0200
[FLINK-37926][table] Support casting from VARIANT to MAP
This closes #29118.
---
docs/content.zh/docs/sql/reference/data-types.md | 20 +-
docs/content/docs/sql/reference/data-types.md | 20 +-
.../types/logical/utils/LogicalTypeCasts.java | 8 +
.../flink/table/types/LogicalTypeCastsTest.java | 26 +++
.../functions/casting/CastRuleProvider.java | 1 +
.../functions/casting/VariantToMapCastRule.java | 208 +++++++++++++++++++++
.../planner/functions/CastFunctionITCase.java | 43 +++++
.../functions/casting/CastRuleProviderTest.java | 14 ++
.../planner/functions/casting/CastRulesTest.java | 88 ++++++++-
.../table/runtime/functions/VariantCastUtils.java | 18 ++
10 files changed, 443 insertions(+), 3 deletions(-)
diff --git a/docs/content.zh/docs/sql/reference/data-types.md
b/docs/content.zh/docs/sql/reference/data-types.md
index 61363906c27..9bc67dd733e 100644
--- a/docs/content.zh/docs/sql/reference/data-types.md
+++ b/docs/content.zh/docs/sql/reference/data-types.md
@@ -1641,6 +1641,24 @@ CAST(o AS ROW<`id` INT, `phone` STRING>) --
fails, the field 'phone' is n
CAST(o AS ROW<`id` VARIANT, `email` VARIANT>) -- (7, null), each field kept
as a variant, the variant null preserved
```
+A variant object also casts to `MAP<STRING, V>`, the schemaless read of an
object. Each field name
+becomes a key and each value casts to `V` by the same rules, recursively. The
key type must be a
+character string, since a variant object's keys are always strings, and a
non-string key type is
+rejected at validation. This is the way to read an object whose keys are not
known in advance.
+
+- A value present but set to a variant null maps to SQL `NULL` when `V` is
nullable and fails the
+ cast when `V` is `NOT NULL`. An empty object casts to an empty map.
+- A `MAP<STRING, VARIANT>` is the identity on its values: it shreds one level
and keeps each value a
+ variant, a variant null included.
+
+The following examples reuse `o` for `PARSE_JSON('{"id": 7, "name": "ada",
"email": null}')`:
+
+```sql
+CAST(o AS MAP<STRING, STRING>) -- {id=7, name=ada, email=NULL}, each value
rendered like the scalar cast
+CAST(o AS MAP<STRING, VARIANT>) -- values kept as variants, the variant null
included
+CAST(o AS MAP<INT, STRING>) -- fails at validation, a MAP key must be a
character string
+```
+
**Declaration**
{{< tabs "25c30432-8460-441d-a036-9416d8202882" >}}
@@ -1855,7 +1873,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 | Y | N |
+| `VARIANT` | 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 0c55f7eb9b8..8e43c2ec12e 100644
--- a/docs/content/docs/sql/reference/data-types.md
+++ b/docs/content/docs/sql/reference/data-types.md
@@ -1649,6 +1649,24 @@ CAST(o AS ROW<`id` INT, `phone` STRING>) --
fails, the field 'phone' is n
CAST(o AS ROW<`id` VARIANT, `email` VARIANT>) -- (7, null), each field kept
as a variant, the variant null preserved
```
+A variant object also casts to `MAP<STRING, V>`, the schemaless read of an
object. Each field name
+becomes a key and each value casts to `V` by the same rules, recursively. The
key type must be a
+character string, since a variant object's keys are always strings, and a
non-string key type is
+rejected at validation. This is the way to read an object whose keys are not
known in advance.
+
+- A value present but set to a variant null maps to SQL `NULL` when `V` is
nullable and fails the
+ cast when `V` is `NOT NULL`. An empty object casts to an empty map.
+- A `MAP<STRING, VARIANT>` is the identity on its values: it shreds one level
and keeps each value a
+ variant, a variant null included.
+
+The following examples reuse `o` for `PARSE_JSON('{"id": 7, "name": "ada",
"email": null}')`:
+
+```sql
+CAST(o AS MAP<STRING, STRING>) -- {id=7, name=ada, email=NULL}, each value
rendered like the scalar cast
+CAST(o AS MAP<STRING, VARIANT>) -- values kept as variants, the variant null
included
+CAST(o AS MAP<INT, STRING>) -- fails at validation, a MAP key must be a
character string
+```
+
**Declaration**
{{< tabs "25c30432-8460-441d-a036-9416d8202882" >}}
@@ -1864,7 +1882,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 | Y | N |
+| `VARIANT` | 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 8e6d1a47ce0..474e63b3670 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
@@ -69,6 +69,7 @@ import static
org.apache.flink.table.types.logical.LogicalTypeRoot.FLOAT;
import static org.apache.flink.table.types.logical.LogicalTypeRoot.INTEGER;
import static
org.apache.flink.table.types.logical.LogicalTypeRoot.INTERVAL_DAY_TIME;
import static
org.apache.flink.table.types.logical.LogicalTypeRoot.INTERVAL_YEAR_MONTH;
+import static org.apache.flink.table.types.logical.LogicalTypeRoot.MAP;
import static org.apache.flink.table.types.logical.LogicalTypeRoot.NULL;
import static org.apache.flink.table.types.logical.LogicalTypeRoot.RAW;
import static org.apache.flink.table.types.logical.LogicalTypeRoot.ROW;
@@ -690,6 +691,13 @@ public final class LogicalTypeCasts {
return allowExplicit
&& targetType.getChildren().stream()
.allMatch(field -> supportsCasting(sourceType,
field, true));
+ } else if (sourceRoot == VARIANT && targetRoot == MAP) {
+ // A variant object casts to MAP<STRING, V> when the key is a
character string, since
+ // VARIANT object keys are always strings, and VARIANT casts to
the value type V.
+ final List<LogicalType> mapChildren = targetType.getChildren();
+ return allowExplicit
+ && mapChildren.get(0).is(CHARACTER_STRING)
+ && supportsCasting(sourceType, mapChildren.get(1), 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 75a99bc5e22..f6022d3f2cb 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.MultisetType;
import org.apache.flink.table.types.logical.NullType;
import org.apache.flink.table.types.logical.RawType;
@@ -332,6 +333,31 @@ class LogicalTypeCastsTest {
.MONTH)))),
false,
false),
+ // A variant object casts to MAP<STRING, V> when the key is a
character string and
+ // the value is castable; a non-string key is rejected
+ Arguments.of(
+ new VariantType(),
+ new MapType(VarCharType.STRING_TYPE, new IntType()),
+ false,
+ true),
+ Arguments.of(
+ new VariantType(),
+ new MapType(VarCharType.STRING_TYPE, new
VariantType()),
+ false,
+ true),
+ Arguments.of(
+ new VariantType(),
+ new MapType(new IntType(), new CharType()),
+ false,
+ false),
+ Arguments.of(
+ new VariantType(),
+ new MapType(
+ VarCharType.STRING_TYPE,
+ 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/CastRuleProvider.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CastRuleProvider.java
index d16012b0afe..532c9bca994 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
@@ -100,6 +100,7 @@ public class CastRuleProvider {
.addRule(VariantToPrimitiveCastRule.INSTANCE)
.addRule(VariantToArrayCastRule.INSTANCE)
.addRule(VariantToRowCastRule.INSTANCE)
+ .addRule(VariantToMapCastRule.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/VariantToMapCastRule.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/VariantToMapCastRule.java
new file mode 100644
index 00000000000..71a8369de0d
--- /dev/null
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/VariantToMapCastRule.java
@@ -0,0 +1,208 @@
+/*
+ * 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.GenericMapData;
+import org.apache.flink.table.data.MapData;
+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.LogicalTypeFamily;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.MapType;
+import org.apache.flink.table.types.logical.utils.LogicalTypeChecks;
+import org.apache.flink.types.variant.Variant;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static
org.apache.flink.table.planner.codegen.CodeGenUtils.boxedTypeTermForType;
+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.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.strLiteral;
+
+/**
+ * {@link LogicalTypeRoot#VARIANT} to {@link LogicalTypeRoot#MAP} cast rule.
+ *
+ * <p>The variant must be an object. Each field name becomes a key and each
field value casts to the
+ * target value type by the full {@code VARIANT}-to-value rule, recursively.
The key type must be a
+ * character string, because a variant object's keys are always strings; a
non-string key type is
+ * rejected at validation. This is the only way to read an object whose keys
are dynamic or unknown
+ * at query time, and {@code MAP<STRING, VARIANT>} is the schemaless read of
an object.
+ *
+ * <p>A {@link LogicalTypeRoot#VARIANT} value type is the identity, so a
VARIANT null value is kept
+ * as a variant null rather than downgraded to SQL {@code NULL}, matching
{@code ARRAY<VARIANT>} and
+ * a {@code ROW} with {@code VARIANT} fields.
+ */
+class VariantToMapCastRule extends
AbstractVariantToConstructedCastRule<MapData> {
+
+ static final VariantToMapCastRule INSTANCE = new VariantToMapCastRule();
+
+ private VariantToMapCastRule() {
+ super(
+ CastRulePredicate.builder()
+ .predicate(
+ (input, target) ->
+ input.is(LogicalTypeRoot.VARIANT)
+ &&
target.is(LogicalTypeRoot.MAP)
+ && ((MapType) target)
+ .getKeyType()
+
.is(LogicalTypeFamily.CHARACTER_STRING)
+ && CastRuleProvider.resolve(
+ input,
+ ((MapType)
target).getValueType())
+ != null)
+ .build());
+ }
+
+ /* Example generated code for MAP<`STRING`, `INT`>. Each field name
becomes a key and each value
+ runs the leaf cast; a nullable value that is a VARIANT null is left as SQL
NULL:
+
+ org.apache.flink.table.runtime.functions.VariantCastUtils.requireObject(
+ variant$2, "MAP<STRING, INT>");
+ java.util.List fieldNames$4 = variant$2.getFieldNames();
+ int size$5 = fieldNames$4.size();
+ java.util.Map map$6 = new java.util.HashMap(size$5);
+ for (int i = 0; i < size$5; i++) {
+ java.lang.String name$7 = (java.lang.String) fieldNames$4.get(i);
+ org.apache.flink.types.variant.Variant valueVariant$8 =
variant$2.getField(name$7);
+ java.lang.Integer value$9 = null;
+ if (!valueVariant$8.isNull()) {
+ result$10 =
+ ((int)
org.apache.flink.table.runtime.functions.VariantCastUtils.toIntegral(
+ valueVariant$8, -2147483648L, 2147483647L,
"INTEGER"));
+ value$9 = result$10;
+ }
+ map$6.put(
+
org.apache.flink.table.runtime.functions.VariantCastUtils.variantKey(
+ name$7, 2147483647, false),
+ value$9);
+ }
+ result$3 = new org.apache.flink.table.data.GenericMapData(map$6);
+
+ A NOT NULL value type throws instead of leaving value$9 null for a VARIANT
null value. A VARIANT
+ value type takes the identity cast unconditionally, so a VARIANT null is
kept as a variant null.
+
+ */
+ @Override
+ protected String generateCodeBlockInternal(
+ CodeGeneratorCastRule.Context context,
+ String inputTerm,
+ String returnVariable,
+ LogicalType inputLogicalType,
+ LogicalType targetLogicalType) {
+ final MapType mapType = (MapType) targetLogicalType;
+ final LogicalType keyType = mapType.getKeyType();
+ final LogicalType valueType = mapType.getValueType();
+ final CodeGeneratorContext codeGeneratorContext =
context.getCodeGeneratorContext();
+
+ final int keyLength = LogicalTypeChecks.getLength(keyType);
+ final boolean charKey = keyType.is(LogicalTypeRoot.CHAR);
+ final String valueTypeTerm = boxedTypeTermForType(valueType);
+
+ final String namesTerm = newName(codeGeneratorContext, "fieldNames");
+ final String sizeTerm = newName(codeGeneratorContext, "size");
+ final String mapTerm = newName(codeGeneratorContext, "map");
+ final String nameTerm = newName(codeGeneratorContext, "name");
+ final String valueVariantTerm = newName(codeGeneratorContext,
"valueVariant");
+ final String valueTerm = newName(codeGeneratorContext, "value");
+
+ // The value is guaranteed non-null here, since a VARIANT null is
handled below, so the
+ // inner
+ // cast is the plain VARIANT-to-value rule. A VARIANT null value maps
to SQL NULL for a
+ // nullable value type, or fails the cast for a NOT NULL one.
+ final CastCodeBlock valueCast =
+ CastRuleProvider.generateAlwaysNonNullCodeBlock(
+ context, valueVariantTerm, inputLogicalType,
valueType);
+ final String putKey =
+ staticCall(
+ VariantCastUtils.class,
+ "variantKey",
+ nameTerm,
+ String.valueOf(keyLength),
+ charKey);
+
+ return new CastRuleUtils.CodeWriter()
+ .stmt(
+ staticCall(
+ VariantCastUtils.class,
+ "requireObject",
+ inputTerm,
+
strLiteral(targetLogicalType.asSummaryString())))
+ .declStmt(className(List.class), namesTerm,
methodCall(inputTerm, "getFieldNames"))
+ .declStmt(int.class, sizeTerm, methodCall(namesTerm, "size"))
+ .declStmt(className(Map.class), mapTerm,
constructorCall(HashMap.class, sizeTerm))
+ .forStmt(
+ sizeTerm,
+ (index, loopWriter) -> {
+ loopWriter
+ .declStmt(
+ String.class,
+ nameTerm,
+ cast(
+ className(String.class),
+ methodCall(namesTerm,
"get", index)))
+ .declStmt(
+ Variant.class,
+ valueVariantTerm,
+ methodCall(inputTerm, "getField",
nameTerm))
+ .declStmt(valueTypeTerm, valueTerm,
"null");
+ final String isPresent = "!" +
methodCall(valueVariantTerm, "isNull");
+ if (valueType.is(LogicalTypeRoot.VARIANT)) {
+ // The value cast is the identity, so a
present VARIANT null value
+ // is
+ // a valid variant null and is kept as-is
rather than downgraded to
+ // SQL NULL, matching ARRAY<VARIANT> and
ROW<VARIANT>.
+ loopWriter
+ .append(valueCast)
+ .assignStmt(valueTerm,
valueCast.getReturnTerm());
+ } else if (valueType.isNullable()) {
+ loopWriter.ifStmt(
+ isPresent,
+ thenWriter ->
+ thenWriter
+ .append(valueCast)
+ .assignStmt(
+ valueTerm,
+
valueCast.getReturnTerm()));
+ } else {
+ loopWriter.ifStmt(
+ isPresent,
+ thenWriter ->
+ thenWriter
+ .append(valueCast)
+ .assignStmt(
+ valueTerm,
+
valueCast.getReturnTerm()),
+ elseWriter ->
+ elseWriter.throwStmt(
+ "new
org.apache.flink.table.api.TableRuntimeException(\"Cannot cast a VARIANT null
object value to a NOT NULL map value type.\")"));
+ }
+ loopWriter.stmt(methodCall(mapTerm, "put", putKey,
valueTerm));
+ },
+ codeGeneratorContext)
+ .assignStmt(returnVariable,
constructorCall(GenericMapData.class, mapTerm))
+ .toString();
+ }
+}
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 2904c874f6c..b73054fbbf1 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
@@ -159,6 +159,7 @@ public class CastFunctionITCase extends
BuiltInFunctionTestBase {
specs.addAll(variantPrimitiveCasts());
specs.addAll(variantArrayCasts());
specs.addAll(variantRowCasts());
+ specs.addAll(variantMapCasts());
return specs;
}
@@ -675,6 +676,48 @@ public class CastFunctionITCase extends
BuiltInFunctionTestBase {
.notNull()));
}
+ private static List<TestSetSpec> variantMapCasts() {
+ final String obj = "{\"id\": 7, \"name\": \"ada\", \"active\": true}";
+ final String objNull = "{\"id\": 7, \"email\": null}";
+ final String mixed = "{\"a\": 1, \"b\": \"x\"}";
+ return List.of(
+ TestSetSpec.forExpression("Cast a VARIANT produced by
parseJson() to a MAP")
+ .onFieldsWithData("unused")
+ .andDataTypes(STRING())
+ // MAP: each field name becomes a key, each value
casts to V
+ .testResult(
+ lit(obj).parseJson().cast(MAP(STRING(),
STRING())),
+ "CAST(PARSE_JSON('" + obj + "') AS MAP<STRING,
STRING>)",
+ map(
+ entry("id", "7"),
+ entry("name", "ada"),
+ entry("active", "TRUE")),
+ MAP(STRING(), STRING()).notNull())
+ .testResult(
+ lit("{}").parseJson().cast(MAP(STRING(),
INT())),
+ "CAST(PARSE_JSON('{}') AS MAP<STRING, INT>)",
+ map(),
+ MAP(STRING(), INT()).notNull())
+ // a value present but set to a variant null maps to
SQL NULL when nullable
+ .testResult(
+ lit(objNull).parseJson().cast(MAP(STRING(),
STRING())),
+ "CAST(PARSE_JSON('" + objNull + "') AS
MAP<STRING, STRING>)",
+ map(entry("id", "7"), entry("email", null)),
+ MAP(STRING(), STRING()).notNull())
+ // and fails when the value type is NOT NULL
+ .testTableApiRuntimeError(
+ lit(objNull).parseJson().cast(MAP(STRING(),
STRING().notNull())),
+ "NOT NULL map value type")
+ // a value that is not an integer fails the cast
+ .testTableApiRuntimeError(
+ lit(mixed).parseJson().cast(MAP(STRING(),
INT())),
+ "does not change the type")
+ // a non-string map key is rejected at validation
+ .testTableApiValidationError(
+ lit(obj).parseJson().cast(MAP(INT(),
STRING())),
+ "Unsupported cast"));
+ }
+
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 19fcd689362..a743cf9ee55 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
@@ -36,6 +36,7 @@ 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.INTERVAL;
+import static org.apache.flink.table.api.DataTypes.MAP;
import static org.apache.flink.table.api.DataTypes.MONTH;
import static org.apache.flink.table.api.DataTypes.MULTISET;
import static org.apache.flink.table.api.DataTypes.ROW;
@@ -172,4 +173,17 @@ class CastRuleProviderTest {
VARIANT, ROW(FIELD("f0",
MULTISET(STRING()))).getLogicalType()))
.isFalse();
}
+
+ @Test
+ void testResolveVariantToMap() {
+ assertThat(CastRuleProvider.resolve(VARIANT, MAP(STRING(),
INT()).getLogicalType()))
+ .isSameAs(VariantToMapCastRule.INSTANCE);
+ // the value recurses through the VARIANT rules, including the
identity leaf
+ assertThat(CastRuleProvider.exists(VARIANT, MAP(STRING(),
VARIANT()).getLogicalType()))
+ .isTrue();
+
+ // a non-string map key is rejected
+ assertThat(CastRuleProvider.exists(VARIANT, MAP(INT(),
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 8ecf6556bb5..274626557d1 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
@@ -327,6 +327,24 @@ class CastRulesTest {
.build())
.build();
+ /** {@code {"a": 1, "b": 2}}, an all-numeric object. */
+ private static final Variant VARIANT_NUM_OBJECT =
+ VARIANT_BUILDER
+ .object()
+ .add("a", VARIANT_BUILDER.of(1))
+ .add("b", VARIANT_BUILDER.of(2))
+ .build();
+
+ private static final Variant VARIANT_EMPTY_OBJECT =
VARIANT_BUILDER.object().build();
+
+ /** {@code {"a": 1, "b": "x"}}, an object with a mixed integer and string
value. */
+ private static final Variant VARIANT_MIXED_OBJECT =
+ VARIANT_BUILDER
+ .object()
+ .add("a", VARIANT_BUILDER.of(1))
+ .add("b", VARIANT_BUILDER.of("x"))
+ .build();
+
private static final DataType MY_STRUCTURED_TYPE =
STRUCTURED(
MyStructuredType.class,
@@ -2257,7 +2275,75 @@ class CastRulesTest {
GenericRowData.of(
GenericRowData.of(1,
fromString("2020-01-01")),
new GenericArrayData(
- new Object[] {fromString("x"),
fromString("y")}))));
+ new Object[] {fromString("x"),
fromString("y")}))),
+ CastTestSpecBuilder.testCastTo(MAP(STRING(), STRING()))
+ .fromCase(VARIANT(), null, null)
+ .fromCase(
+ VARIANT(),
+ VARIANT_RECORD,
+ mapData(
+ entry(fromString("id"),
fromString("7")),
+ entry(fromString("name"),
fromString("ada")),
+ entry(fromString("active"),
fromString("TRUE"))))
+ // an empty object casts to an empty map
+ .fromCase(VARIANT(), VARIANT_EMPTY_OBJECT, mapData())
+ // a value present but set to a variant null maps to
SQL NULL when nullable
+ .fromCase(
+ VARIANT(),
+ VARIANT_RECORD_WITH_NULL,
+ mapData(
+ entry(fromString("id"),
fromString("7")),
+ entry(fromString("email"), null)))
+ // a mixed object renders every value to STRING
+ .fromCase(
+ VARIANT(),
+ VARIANT_MIXED_OBJECT,
+ mapData(
+ entry(fromString("a"),
fromString("1")),
+ entry(fromString("b"),
fromString("x"))))
+ // an array is not an object
+ .fail(
+ VARIANT(),
+ VARIANT_INT_ARRAY,
+ TableRuntimeException.class,
+ "requires an object"),
+ // a NOT NULL value type rejects a variant null value
+ CastTestSpecBuilder.testCastTo(MAP(STRING(),
STRING().notNull()))
+ .fail(
+ VARIANT(),
+ VARIANT_RECORD_WITH_NULL,
+ TableRuntimeException.class,
+ "NOT NULL map value type"),
+ // MAP<STRING, VARIANT> keeps each value a variant, one level
shredded
+ CastTestSpecBuilder.testCastTo(MAP(STRING(), VARIANT()))
+ .fromCase(
+ VARIANT(),
+ VARIANT_NUM_OBJECT,
+ mapData(
+ entry(fromString("a"),
VARIANT_NUM_OBJECT.getField("a")),
+ entry(fromString("b"),
VARIANT_NUM_OBJECT.getField("b"))))
+ // a VARIANT value keeps a variant null as a variant
null, not SQL NULL
+ .fromCase(
+ VARIANT(),
+ VARIANT_RECORD_WITH_NULL,
+ mapData(
+ entry(
+ fromString("id"),
+
VARIANT_RECORD_WITH_NULL.getField("id")),
+ entry(
+ fromString("email"),
+
VARIANT_RECORD_WITH_NULL.getField("email")))),
+ CastTestSpecBuilder.testCastTo(MAP(STRING(), INT()))
+ .fromCase(
+ VARIANT(),
+ VARIANT_NUM_OBJECT,
+ mapData(entry(fromString("a"), 1),
entry(fromString("b"), 2)))
+ // a value that is not an integer fails the cast
+ .fail(
+ VARIANT(),
+ VARIANT_MIXED_OBJECT,
+ TableRuntimeException.class,
+ "does not change the type"));
}
@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 f273575cdbe..f4920a4d353 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
@@ -84,6 +84,24 @@ public final class VariantCastUtils {
}
}
+ /**
+ * Renders an object field name as the map key, trimming or padding it to
a bounded target the
+ * same way a regular cast into the key type would. Object keys are always
strings, so no kind
+ * conversion happens here.
+ */
+ public static BinaryStringData variantKey(String name, int targetLength,
boolean charTarget) {
+ final BinaryStringData key = BinaryStringData.fromString(name);
+ final int length = key.numChars();
+ if (length > targetLength) {
+ return key.substring(0, targetLength);
+ }
+ if (charTarget && length < targetLength) {
+ return BinaryStringDataUtil.concat(
+ key, BinaryStringData.blankString(targetLength - length));
+ }
+ return key;
+ }
+
private static TableRuntimeException wrongShape(
Variant variant, String targetType, String required) {
return new TableRuntimeException(
