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 855597a5460 [FLINK-40978][table] Cast from VARIANT through
FromVariantConverter
855597a5460 is described below
commit 855597a5460630eb8f72510277bfed7c0be0a004
Author: Ramin Gharib <[email protected]>
AuthorDate: Mon Oct 19 16:38:00 2026 +0200
[FLINK-40978][table] Cast from VARIANT through FromVariantConverter
This closes #29466.
---
.../table/data/utils/FromVariantConverter.java | 299 ++++++++++++++
.../flink/table/data/utils}/VariantCastUtils.java | 7 +-
.../table/data/utils/FromVariantConverterTest.java | 447 +++++++++++++++++++++
.../table/data/utils}/VariantCastUtilsTest.java | 4 +-
.../casting/AbstractCodeGeneratorCastRule.java | 6 +
.../AbstractExpressionCodeGeneratorCastRule.java | 6 +
.../AbstractVariantToConstructedCastRule.java | 71 ----
.../functions/casting/CastRuleProvider.java | 6 +-
.../functions/casting/CodeGeneratorCastRule.java | 7 +
.../functions/casting/FromVariantCastRule.java | 125 ++++++
.../functions/casting/VariantToArrayCastRule.java | 171 --------
.../functions/casting/VariantToMapCastRule.java | 208 ----------
.../casting/VariantToPrimitiveCastRule.java | 241 -----------
.../functions/casting/VariantToRowCastRule.java | 234 -----------
.../functions/casting/VariantToStringCastRule.java | 121 ------
.../planner/codegen/calls/ScalarOperatorGens.scala | 1 +
.../planner/functions/CastFunctionITCase.java | 44 +-
.../functions/casting/CastRuleProviderTest.java | 19 +-
.../flink/table/planner/utils/FilterUtils.java | 80 +---
19 files changed, 970 insertions(+), 1127 deletions(-)
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/FromVariantConverter.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/FromVariantConverter.java
new file mode 100644
index 00000000000..f612be4bf34
--- /dev/null
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/FromVariantConverter.java
@@ -0,0 +1,299 @@
+/*
+ * 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.data.utils;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.TableRuntimeException;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.GenericMapData;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.DistinctType;
+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.table.types.logical.utils.LogicalTypeUtils;
+import org.apache.flink.types.variant.BinaryVariant;
+import org.apache.flink.types.variant.Variant;
+import org.apache.flink.types.variant.VariantTypeException;
+
+import java.io.Serializable;
+import java.lang.reflect.Array;
+import java.time.ZoneId;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.TimeZone;
+
+/**
+ * Converts a {@link Variant} into internal data of a SQL type, for {@code
CAST(v AS T)}. {@link
+ * VariantCastUtils} defines how each leaf is read.
+ *
+ * <p>A {@code ROW} takes object fields by name and fails on a missing one. A
nested variant null
+ * becomes {@code NULL}, fails for a {@code NOT NULL} type, and stays a
variant null for a {@code
+ * VARIANT} type. A top-level variant null fails, so a cast to a nullable type
maps it to {@code
+ * NULL} first.
+ */
+@Internal
+public final class FromVariantConverter implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final ValueReader reader;
+
+ private FromVariantConverter(final ValueReader reader) {
+ this.reader = reader;
+ }
+
+ /**
+ * Creates a converter for a type that VARIANT casts to.
+ *
+ * @param sessionZone renders a {@code TIMESTAMP_LTZ} value read as a
string
+ * @throws IllegalArgumentException if VARIANT cannot be cast to the type
+ */
+ public static FromVariantConverter create(final LogicalType type, final
ZoneId sessionZone) {
+ return new FromVariantConverter(
+ createReader(type, TimeZone.getTimeZone(sessionZone), false));
+ }
+
+ /**
+ * Converts a variant into internal data.
+ *
+ * @throws TableRuntimeException if a value does not match its type
+ * @throws VariantTypeException if the variant is malformed, or a {@code
BOOLEAN}, {@code DATE}
+ * or binary value has another kind
+ */
+ public Object convert(final Variant variant) {
+ return reader.read(variant);
+ }
+
+ @FunctionalInterface
+ private interface ValueReader extends Serializable {
+ Object read(Variant variant);
+ }
+
+ // With copyVariants, a VARIANT is copied out of its parent rather than
returned as a view.
+ private static ValueReader createReader(
+ final LogicalType type, final TimeZone sessionZone, final boolean
copyVariants) {
+ switch (type.getTypeRoot()) {
+ case BOOLEAN:
+ return Variant::getBoolean;
+ case TINYINT:
+ return variant ->
+ (byte)
+ VariantCastUtils.toIntegral(
+ variant, Byte.MIN_VALUE,
Byte.MAX_VALUE, "TINYINT");
+ case SMALLINT:
+ return variant ->
+ (short)
+ VariantCastUtils.toIntegral(
+ variant, Short.MIN_VALUE,
Short.MAX_VALUE, "SMALLINT");
+ case INTEGER:
+ return variant ->
+ (int)
+ VariantCastUtils.toIntegral(
+ variant, Integer.MIN_VALUE,
Integer.MAX_VALUE, "INTEGER");
+ case BIGINT:
+ return variant ->
+ VariantCastUtils.toIntegral(
+ variant, Long.MIN_VALUE, Long.MAX_VALUE,
"BIGINT");
+ case FLOAT:
+ return VariantCastUtils::toFloat;
+ case DOUBLE:
+ return VariantCastUtils::toDouble;
+ case DECIMAL:
+ {
+ final int precision = LogicalTypeChecks.getPrecision(type);
+ final int scale = LogicalTypeChecks.getScale(type);
+ return variant -> VariantCastUtils.toDecimal(variant,
precision, scale);
+ }
+ case CHAR:
+ case VARCHAR:
+ {
+ final int length = LogicalTypeChecks.getLength(type);
+ final boolean charTarget = type.is(LogicalTypeRoot.CHAR);
+ return variant ->
+ VariantCastUtils.toStringValue(
+ variant, sessionZone, length, charTarget);
+ }
+ case BINARY:
+ case VARBINARY:
+ {
+ final int length = LogicalTypeChecks.getLength(type);
+ final boolean fixedLength =
type.is(LogicalTypeRoot.BINARY);
+ return variant -> VariantCastUtils.toBytes(variant,
length, fixedLength);
+ }
+ case DATE:
+ return variant -> (int) variant.getDate().toEpochDay();
+ case TIME_WITHOUT_TIME_ZONE:
+ {
+ final int precision = LogicalTypeChecks.getPrecision(type);
+ return variant -> VariantCastUtils.toTime(variant,
precision);
+ }
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ {
+ final int precision = LogicalTypeChecks.getPrecision(type);
+ return variant -> VariantCastUtils.toTimestamp(variant,
precision);
+ }
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ {
+ final int precision = LogicalTypeChecks.getPrecision(type);
+ return variant -> VariantCastUtils.toTimestampLtz(variant,
precision);
+ }
+ case UUID:
+ return VariantCastUtils::toUuid;
+ case VARIANT:
+ return copyVariants ? FromVariantConverter::copyVariant :
variant -> variant;
+ case ARRAY:
+ return createArrayReader((ArrayType) type, sessionZone,
copyVariants);
+ case MAP:
+ return createMapReader((MapType) type, sessionZone,
copyVariants);
+ case ROW:
+ case STRUCTURED_TYPE:
+ return createRowReader(type, sessionZone);
+ case DISTINCT_TYPE:
+ return createReader(
+ ((DistinctType) type).getSourceType(), sessionZone,
copyVariants);
+ default:
+ throw new IllegalArgumentException("Cannot cast VARIANT to " +
type + ".");
+ }
+ }
+
+ // A nested variant null becomes NULL, or fails for a NOT NULL type. A
VARIANT type keeps it.
+ private static ValueReader createChildReader(
+ final LogicalType type,
+ final TimeZone sessionZone,
+ final boolean copyVariants,
+ final String notNullMessage) {
+ final ValueReader reader = createReader(type, sessionZone,
copyVariants);
+ if (type.is(LogicalTypeRoot.VARIANT)) {
+ return reader;
+ }
+ if (type.isNullable()) {
+ return variant -> variant.isNull() ? null : reader.read(variant);
+ }
+ return variant -> {
+ if (variant.isNull()) {
+ throw new TableRuntimeException(notNullMessage);
+ }
+ return reader.read(variant);
+ };
+ }
+
+ private static ValueReader createArrayReader(
+ final ArrayType type, final TimeZone sessionZone, final boolean
copyVariants) {
+ final LogicalType elementType = type.getElementType();
+ final ValueReader elementReader =
+ createChildReader(
+ elementType,
+ sessionZone,
+ copyVariants,
+ "Cannot cast a VARIANT null array element to a NOT
NULL element type.");
+ // A typed array keeps GenericArrayData#toIntArray and its siblings
working.
+ final Class<?> elementClass =
LogicalTypeUtils.toInternalConversionClass(elementType);
+ final String arrayType = type.asSummaryString();
+ return variant -> {
+ final int size = VariantCastUtils.arraySize(variant, arrayType);
+ final Object[] elements = (Object[])
Array.newInstance(elementClass, size);
+ for (int i = 0; i < size; i++) {
+ elements[i] = elementReader.read(variant.getElement(i));
+ }
+ return new GenericArrayData(elements);
+ };
+ }
+
+ private static ValueReader createMapReader(
+ final MapType type, final TimeZone sessionZone, final boolean
copyVariants) {
+ final LogicalType keyType = type.getKeyType();
+ if (!keyType.is(LogicalTypeFamily.CHARACTER_STRING)) {
+ throw new IllegalArgumentException(
+ "Cannot cast VARIANT to " + type + ". A MAP key must be a
character string.");
+ }
+ final int keyLength = LogicalTypeChecks.getLength(keyType);
+ final boolean charKey = keyType.is(LogicalTypeRoot.CHAR);
+ final ValueReader valueReader =
+ createChildReader(
+ type.getValueType(),
+ sessionZone,
+ copyVariants,
+ "Cannot cast a VARIANT null object value to a NOT NULL
map value type.");
+ final String mapType = type.asSummaryString();
+ return variant -> {
+ VariantCastUtils.requireObject(variant, mapType);
+ final List<String> names = variant.getFieldNames();
+ final Map<Object, Object> map = new HashMap<>(names.size());
+ for (final String name : names) {
+ final Object value = valueReader.read(variant.getField(name));
+ map.put(VariantCastUtils.variantKey(name, keyLength, charKey),
value);
+ }
+ return new GenericMapData(map);
+ };
+ }
+
+ // A ROW copies every VARIANT below it, like a value read back from a
binary row.
+ private static ValueReader createRowReader(final LogicalType type, final
TimeZone sessionZone) {
+ final List<LogicalType> fieldTypes =
LogicalTypeChecks.getFieldTypes(type);
+ final String[] fieldNames =
LogicalTypeChecks.getFieldNames(type).toArray(new String[0]);
+ final int fieldCount = fieldTypes.size();
+ final String rowType = type.asSummaryString();
+ final ValueReader[] fieldReaders = new ValueReader[fieldCount];
+ for (int i = 0; i < fieldCount; i++) {
+ fieldReaders[i] =
+ createChildReader(
+ fieldTypes.get(i),
+ sessionZone,
+ true,
+ nullFieldError(rowType, fieldNames[i]));
+ }
+ return variant -> {
+ VariantCastUtils.requireObject(variant, rowType);
+ final GenericRowData row = new GenericRowData(fieldCount);
+ for (int i = 0; i < fieldCount; i++) {
+ // getField returns null for a field the object does not have.
+ final Variant field = variant.getField(fieldNames[i]);
+ if (field == null) {
+ throw new TableRuntimeException(missingFieldError(rowType,
fieldNames[i]));
+ }
+ row.setField(i, fieldReaders[i].read(field));
+ }
+ return row;
+ };
+ }
+
+ private static Variant copyVariant(final Variant variant) {
+ final BinaryVariant view = (BinaryVariant) variant;
+ return new BinaryVariant(view.getValue(), view.getMetadata());
+ }
+
+ private static String missingFieldError(final String rowType, final String
fieldName) {
+ return String.format(
+ "Cannot cast the VARIANT object to %s because the field '%s'
is not present in "
+ + "the VARIANT.",
+ rowType, fieldName);
+ }
+
+ private static String nullFieldError(final String rowType, final String
fieldName) {
+ return String.format(
+ "Cannot cast the VARIANT object to %s because the field '%s'
is a VARIANT null "
+ + "and the target does not accept NULL.",
+ rowType, fieldName);
+ }
+}
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/VariantCastUtils.java
similarity index 98%
rename from
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java
rename to
flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/VariantCastUtils.java
index 4ff53cc0b93..6c1b9c4118f 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/VariantCastUtils.java
@@ -16,16 +16,14 @@
* limitations under the License.
*/
-package org.apache.flink.table.runtime.functions;
+package org.apache.flink.table.data.utils;
import org.apache.flink.annotation.Internal;
import org.apache.flink.table.api.TableRuntimeException;
import org.apache.flink.table.data.DecimalData;
import org.apache.flink.table.data.TimestampData;
import org.apache.flink.table.data.binary.BinaryStringData;
-import org.apache.flink.table.data.binary.BinaryStringDataUtil;
import org.apache.flink.table.data.binary.StringUtf8Utils;
-import org.apache.flink.table.data.utils.ToVariantConverter;
import org.apache.flink.table.types.logical.utils.UuidUtils;
import org.apache.flink.table.utils.DateTimeUtils;
import org.apache.flink.table.utils.EncodingUtils;
@@ -109,8 +107,7 @@ public final class VariantCastUtils {
return key.substring(0, targetLength);
}
if (charTarget && length < targetLength) {
- return BinaryStringDataUtil.concat(
- key, BinaryStringData.blankString(targetLength - length));
+ return BinaryStringData.fromString(name + " ".repeat(targetLength
- length));
}
return key;
}
diff --git
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/utils/FromVariantConverterTest.java
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/utils/FromVariantConverterTest.java
new file mode 100644
index 00000000000..61997885049
--- /dev/null
+++
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/utils/FromVariantConverterTest.java
@@ -0,0 +1,447 @@
+/*
+ * 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.data.utils;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.TableRuntimeException;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.GenericMapData;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.DistinctType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.StructuredType;
+import org.apache.flink.table.types.logical.StructuredType.StructuredAttribute;
+import org.apache.flink.table.types.logical.utils.UuidUtils;
+import org.apache.flink.types.variant.BinaryVariant;
+import org.apache.flink.types.variant.Variant;
+import org.apache.flink.types.variant.VariantBuilder;
+import org.apache.flink.util.InstantiationUtil;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.math.BigDecimal;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.stream.Stream;
+
+import static org.apache.flink.table.api.DataTypes.FIELD;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.jupiter.params.provider.Arguments.arguments;
+
+class FromVariantConverterTest {
+
+ private static final VariantBuilder BUILDER = Variant.newBuilder();
+
+ private static final LocalDateTime DATE_TIME =
+ LocalDateTime.of(2026, 9, 25, 10, 15, 30, 123_000_000);
+
+ private static final UUID UUID_VALUE =
UUID.fromString("0e1f2a3b-4c5d-6e7f-8091-a2b3c4d5e6f7");
+
+ private static final LogicalType MONEY =
+ DistinctType.newBuilder(ObjectIdentifier.of("cat", "db", "Money"),
new IntType())
+ .build();
+
+ private static final LogicalType ORDER =
+ StructuredType.newBuilder(ObjectIdentifier.of("cat", "db",
"Order"))
+ .attributes(
+ List.of(
+ new StructuredAttribute("id", new
IntType()),
+ new StructuredAttribute("price", MONEY)))
+ .build();
+
+ @Test
+ void testLeavesApplyTheLengthAndPrecisionOfTheirType() {
+ final LogicalType type =
+ DataTypes.ROW(
+ FIELD("bool", DataTypes.BOOLEAN()),
+ FIELD("tiny", DataTypes.TINYINT()),
+ FIELD("small", DataTypes.SMALLINT()),
+ FIELD("int", DataTypes.INT()),
+ FIELD("big", DataTypes.BIGINT()),
+ FIELD("float", DataTypes.FLOAT()),
+ FIELD("double", DataTypes.DOUBLE()),
+ FIELD("decimal", DataTypes.DECIMAL(10, 2)),
+ FIELD("char", DataTypes.CHAR(4)),
+ FIELD("varchar", DataTypes.VARCHAR(2)),
+ FIELD("binary", DataTypes.BINARY(2)),
+ FIELD("bytes", DataTypes.BYTES()),
+ FIELD("date", DataTypes.DATE()),
+ FIELD("time", DataTypes.TIME(0)),
+ FIELD("ts", DataTypes.TIMESTAMP(0)),
+ FIELD("ltz", DataTypes.TIMESTAMP_LTZ(3)),
+ FIELD("uuid", DataTypes.UUID()),
+ FIELD("variant", DataTypes.VARIANT()))
+ .getLogicalType();
+ final Variant variant =
+ BUILDER.object()
+ .add("bool", BUILDER.of(true))
+ .add("tiny", BUILDER.of((byte) 1))
+ .add("small", BUILDER.of((short) 2))
+ .add("int", BUILDER.of(3))
+ .add("big", BUILDER.of(4L))
+ .add("float", BUILDER.of(1.5f))
+ .add("double", BUILDER.of(2.5d))
+ .add("decimal", BUILDER.of(new BigDecimal("12.5")))
+ .add("char", BUILDER.of("ab"))
+ .add("varchar", BUILDER.of("abc"))
+ .add("binary", BUILDER.of(new byte[] {7}))
+ .add("bytes", BUILDER.of(new byte[] {(byte) 0xCA,
(byte) 0xFE}))
+ .add("date", BUILDER.of(LocalDate.of(2026, 9, 25)))
+ .add("time", BUILDER.of(LocalTime.of(10, 15, 30,
123_456_000)))
+ .add("ts", BUILDER.of(DATE_TIME))
+ .add("ltz", BUILDER.of(Instant.ofEpochSecond(0,
123_456_000)))
+ .add("uuid", BUILDER.of(UUID_VALUE))
+ .add("variant", BUILDER.of("raw"))
+ .build();
+
+ final RowData row = (RowData) convert(type, variant);
+
+ assertThat(row.getBoolean(0)).isTrue();
+ assertThat(row.getByte(1)).isEqualTo((byte) 1);
+ assertThat(row.getShort(2)).isEqualTo((short) 2);
+ assertThat(row.getInt(3)).isEqualTo(3);
+ assertThat(row.getLong(4)).isEqualTo(4L);
+ assertThat(row.getFloat(5)).isEqualTo(1.5f);
+ assertThat(row.getDouble(6)).isEqualTo(2.5d);
+ assertThat(row.getDecimal(7, 10, 2)).isEqualTo(decimal("12.50", 10,
2));
+ assertThat(row.getString(8)).isEqualTo(StringData.fromString("ab "));
+ assertThat(row.getString(9)).isEqualTo(StringData.fromString("ab"));
+ assertThat(row.getBinary(10)).containsExactly(7, 0);
+ assertThat(row.getBinary(11)).containsExactly((byte) 0xCA, (byte)
0xFE);
+ assertThat(row.getInt(12)).isEqualTo((int) LocalDate.of(2026, 9,
25).toEpochDay());
+ assertThat(row.getInt(13)).isEqualTo(LocalTime.of(10, 15,
30).toSecondOfDay() * 1000);
+ assertThat(row.getTimestamp(14, 0))
+
.isEqualTo(TimestampData.fromLocalDateTime(DATE_TIME.withNano(0)));
+ assertThat(row.getTimestamp(15,
3)).isEqualTo(TimestampData.fromEpochMillis(123));
+ assertThat(row.getBinary(16)).isEqualTo(UuidUtils.toBytes(UUID_VALUE));
+ assertThat(row.getVariant(17).getString()).isEqualTo("raw");
+ }
+
+ @Test
+ void testIntegerLeavesCheckTheRangeOfTheirType() {
+ assertThatThrownBy(() -> convert(DataTypes.TINYINT(), BUILDER.of(128)))
+ .isInstanceOf(TableRuntimeException.class)
+ .hasMessage("Casting the VARIANT value 128 to TINYINT
overflowed.");
+ assertThatThrownBy(() -> convert(DataTypes.SMALLINT(),
BUILDER.of(32768)))
+ .isInstanceOf(TableRuntimeException.class)
+ .hasMessage("Casting the VARIANT value 32768 to SMALLINT
overflowed.");
+ assertThatThrownBy(() -> convert(DataTypes.INT(),
BUILDER.of(2147483648L)))
+ .isInstanceOf(TableRuntimeException.class)
+ .hasMessage("Casting the VARIANT value 2147483648 to INTEGER
overflowed.");
+ assertThatThrownBy(
+ () ->
+ convert(
+ DataTypes.BIGINT(),
+ BUILDER.of(new
BigDecimal("9223372036854775808"))))
+ .isInstanceOf(TableRuntimeException.class)
+ .hasMessage("Casting the VARIANT value 9223372036854775808 to
BIGINT overflowed.");
+ }
+
+ @Test
+ void testMapKeysFollowTheLengthOfTheKeyType() {
+ final Variant variant = BUILDER.object().add("ab",
BUILDER.of(1)).build();
+
+ final MapData charKey =
+ (MapData) convert(DataTypes.MAP(DataTypes.CHAR(3),
DataTypes.INT()), variant);
+ final MapData varcharKey =
+ (MapData) convert(DataTypes.MAP(DataTypes.VARCHAR(1),
DataTypes.INT()), variant);
+
+
assertThat(charKey.keyArray().getString(0)).isEqualTo(StringData.fromString("ab
"));
+
assertThat(varcharKey.keyArray().getString(0)).isEqualTo(StringData.fromString("a"));
+ }
+
+ @Test
+ void testArrayIsTypedByItsElement() {
+ final Variant variant =
BUILDER.array().add(BUILDER.of(1)).add(BUILDER.of(2)).build();
+
+ final GenericArrayData array =
+ (GenericArrayData)
convert(DataTypes.ARRAY(DataTypes.INT().notNull()), variant);
+
+ assertThat(array.toIntArray()).containsExactly(1, 2);
+ }
+
+ @Test
+ void testNotNullConstructedChildFails() {
+ final Variant array = BUILDER.array().add(BUILDER.ofNull()).build();
+
+ assertThatThrownBy(
+ () ->
+ convert(
+
DataTypes.ARRAY(DataTypes.ARRAY(DataTypes.INT()).notNull()),
+ array))
+ .isInstanceOf(TableRuntimeException.class)
+ .hasMessage("Cannot cast a VARIANT null array element to a NOT
NULL element type.");
+ }
+
+ @Test
+ void testNotNullVariantChildKeepsVariantNull() {
+ final Variant array = BUILDER.array().add(BUILDER.ofNull()).build();
+
+ final ArrayData result =
+ (ArrayData)
convert(DataTypes.ARRAY(DataTypes.VARIANT().notNull()), array);
+
+ assertThat(result.getVariant(0).isNull()).isTrue();
+ }
+
+ @Test
+ void testRowCopiesEveryVariantBelowIt() {
+ final Variant list = BUILDER.array().add(BUILDER.of("x")).build();
+ final Variant object =
+ BUILDER.object()
+ .add("v", BUILDER.of("x"))
+ .add("list", list)
+ .add("map", BUILDER.object().add("k",
BUILDER.of("x")).build())
+ .build();
+ final DataType type =
+ DataTypes.ROW(
+ FIELD("v", DataTypes.VARIANT()),
+ FIELD("list", DataTypes.ARRAY(DataTypes.VARIANT())),
+ FIELD("map", DataTypes.MAP(DataTypes.STRING(),
DataTypes.VARIANT())));
+
+ final RowData row = (RowData) convert(type, object);
+
+ assertThat(row.getVariant(0)).isEqualTo(copyOf(object.getField("v")));
+ assertThat(row.getArray(1).getVariant(0))
+ .isEqualTo(copyOf(object.getField("list").getElement(0)));
+ assertThat(row.getMap(2).valueArray().getVariant(0))
+ .isEqualTo(copyOf(object.getField("map").getField("k")));
+ // outside a ROW, an element stays a view into its parent
+ assertThat(((ArrayData) convert(DataTypes.ARRAY(DataTypes.VARIANT()),
list)).getVariant(0))
+ .isEqualTo(list.getElement(0));
+ }
+
+ @Test
+ void testAbsentVariantFieldFails() {
+ final Variant object = BUILDER.object().add("b",
BUILDER.of(1)).build();
+
+ assertThatThrownBy(() -> convert(DataTypes.ROW(FIELD("a",
DataTypes.VARIANT())), object))
+ .isInstanceOf(TableRuntimeException.class)
+ .hasMessage(
+ "Cannot cast the VARIANT object to ROW<`a` VARIANT>
because the field 'a' "
+ + "is not present in the VARIANT.");
+ }
+
+ @Test
+ void testTopLevelVariantNullReachesTheLeaf() {
+ final Variant nullVariant = BUILDER.ofNull();
+
+ // a cast to a nullable type maps a variant null to NULL before it
calls the converter
+ assertThatThrownBy(() -> convert(DataTypes.INT(), nullVariant))
+ .isInstanceOf(TableRuntimeException.class)
+ .hasMessageStartingWith("Cannot cast a VARIANT NULL value to
INTEGER.");
+ assertThatThrownBy(() -> convert(DataTypes.INT().notNull(),
nullVariant))
+ .isInstanceOf(TableRuntimeException.class)
+ .hasMessageStartingWith("Cannot cast a VARIANT NULL value to
INTEGER.");
+ assertThatThrownBy(() ->
convert(DataTypes.ARRAY(DataTypes.INT()).notNull(), nullVariant))
+ .isInstanceOf(TableRuntimeException.class)
+ .hasMessageStartingWith(
+ "Cannot cast a VARIANT NULL value to ARRAY<INT> NOT
NULL because the "
+ + "target requires an array.");
+ assertThat(convert(DataTypes.VARIANT(),
nullVariant)).isSameAs(nullVariant);
+ }
+
+ @Test
+ void testNestedWrongShapeNamesItsOwnType() {
+ final Variant array = BUILDER.array().add(BUILDER.of(1)).build();
+
+ assertThatThrownBy(() ->
convert(DataTypes.ARRAY(DataTypes.ARRAY(DataTypes.INT())), array))
+ .isInstanceOf(TableRuntimeException.class)
+ .hasMessage(
+ "Cannot cast a VARIANT INT value to ARRAY<INT> because
the target "
+ + "requires an array. Only a variant array
casts to ARRAY and "
+ + "only a variant object casts to ROW,
STRUCTURED, or MAP.");
+ }
+
+ @Test
+ void testStructuredAndDistinctTypes() {
+ final Variant variant =
+ BUILDER.object().add("price", BUILDER.of(42)).add("id",
BUILDER.of(1)).build();
+
+ assertThat(convert(ORDER, variant)).isEqualTo(GenericRowData.of(1,
42));
+ assertThat(convert(MONEY, BUILDER.of(42))).isEqualTo(42);
+ }
+
+ @Test
+ void testCreateRejectsTypesThatVariantDoesNotCastTo() {
+ assertThatThrownBy(
+ () ->
+ FromVariantConverter.create(
+
DataTypes.MULTISET(DataTypes.INT()).getLogicalType(),
+ ZoneOffset.UTC))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Cannot cast VARIANT to MULTISET<INT>.");
+ assertThatThrownBy(
+ () ->
+ FromVariantConverter.create(
+
DataTypes.ARRAY(DataTypes.NULL()).getLogicalType(),
+ ZoneOffset.UTC))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Cannot cast VARIANT to NULL.");
+ assertThatThrownBy(
+ () ->
+ FromVariantConverter.create(
+ DataTypes.MAP(DataTypes.INT(),
DataTypes.STRING())
+ .getLogicalType(),
+ ZoneOffset.UTC))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageEndingWith("A MAP key must be a character string.");
+ }
+
+ @Test
+ void testConverterSurvivesJavaSerialization() throws Exception {
+ final LogicalType type =
+ DataTypes.ROW(
+ FIELD("int", DataTypes.INT()),
+ FIELD("ltz", DataTypes.STRING()),
+ FIELD("list",
DataTypes.ARRAY(DataTypes.INT().notNull())),
+ FIELD("map", DataTypes.MAP(DataTypes.CHAR(2),
DataTypes.INT())),
+ FIELD("variant", DataTypes.VARIANT()),
+ FIELD("order", DataTypes.of(ORDER)))
+ .getLogicalType();
+ final Variant variant =
+ BUILDER.object()
+ .add("int", BUILDER.of(1))
+ .add("ltz", BUILDER.of(Instant.EPOCH))
+ .add("list",
BUILDER.array().add(BUILDER.of(2)).build())
+ .add("map", BUILDER.object().add("k",
BUILDER.ofNull()).build())
+ .add("variant", BUILDER.ofNull())
+ .add(
+ "order",
+ BUILDER.object()
+ .add("id", BUILDER.of(3))
+ .add("price", BUILDER.of(4))
+ .build())
+ .build();
+ final FromVariantConverter converter =
+ FromVariantConverter.create(type, ZoneId.of("Asia/Tokyo"));
+
+ final RowData actual = (RowData)
InstantiationUtil.clone(converter).convert(variant);
+
+ assertThat(actual).isEqualTo(converter.convert(variant));
+ // the session zone survives too
+ assertThat(actual.getString(1).toString()).startsWith("1970-01-01
09:00:00");
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("roundTrips")
+ void testRoundTripThroughToVariantConverter(final DataType type, final
Object value) {
+ final LogicalType logicalType = type.getLogicalType();
+ final Variant variant =
ToVariantConverter.create(logicalType).convert(value);
+
+ assertThat(FromVariantConverter.create(logicalType,
ZoneOffset.UTC).convert(variant))
+ .isEqualTo(value);
+ }
+
+ private static Stream<Arguments> roundTrips() {
+ return Stream.of(
+ arguments(DataTypes.BOOLEAN(), true),
+ arguments(DataTypes.TINYINT(), (byte) -1),
+ arguments(DataTypes.SMALLINT(), (short) 300),
+ arguments(DataTypes.INT(), 70_000),
+ arguments(DataTypes.BIGINT(), Long.MIN_VALUE),
+ arguments(DataTypes.FLOAT(), 1.5f),
+ arguments(DataTypes.DOUBLE(), -2.25d),
+ arguments(DataTypes.DECIMAL(10, 2), decimal("12.50", 10, 2)),
+ arguments(
+ DataTypes.DECIMAL(38, 10),
+ decimal("1234567890123456789012345678.0123456789", 38,
10)),
+ arguments(DataTypes.CHAR(4), StringData.fromString("ab ")),
+ arguments(DataTypes.STRING(), StringData.fromString("héllo
👋")),
+ arguments(DataTypes.BINARY(2), new byte[] {1, 2}),
+ arguments(DataTypes.BYTES(), new byte[] {(byte) 0xCA, (byte)
0xFE}),
+ arguments(DataTypes.DATE(), (int) LocalDate.of(2026, 9,
25).toEpochDay()),
+ arguments(DataTypes.TIME(3), LocalTime.of(10, 15,
30).toSecondOfDay() * 1000 + 123),
+ arguments(DataTypes.TIMESTAMP(3),
TimestampData.fromLocalDateTime(DATE_TIME)),
+ arguments(DataTypes.TIMESTAMP(9),
TimestampData.fromEpochMillis(1, 234_567)),
+ arguments(DataTypes.TIMESTAMP_LTZ(6),
TimestampData.fromEpochMillis(-1, 1_000)),
+ arguments(DataTypes.TIMESTAMP_LTZ(9),
TimestampData.fromEpochMillis(0, 1)),
+ arguments(DataTypes.UUID(), UuidUtils.toBytes(UUID_VALUE)),
+ arguments(
+ DataTypes.ARRAY(DataTypes.INT()),
+ new GenericArrayData(new Object[] {1, null, 3})),
+ arguments(
+ DataTypes.ARRAY(DataTypes.ARRAY(DataTypes.STRING())),
+ new GenericArrayData(
+ new Object[] {
+ new GenericArrayData(new Object[]
{StringData.fromString("a")}),
+ null
+ })),
+ arguments(
+ DataTypes.MAP(DataTypes.STRING(), DataTypes.BIGINT()),
+ mapOf(StringData.fromString("a"), 1L)),
+ arguments(
+ DataTypes.MAP(DataTypes.STRING(), DataTypes.BIGINT()),
+ mapOf(StringData.fromString("a"), null)),
+ arguments(
+ DataTypes.ROW(
+ FIELD("id", DataTypes.INT().notNull()),
+ FIELD("name", DataTypes.STRING()),
+ FIELD("tags",
DataTypes.ARRAY(DataTypes.STRING()))),
+ GenericRowData.of(
+ 7,
+ null,
+ new GenericArrayData(new Object[]
{StringData.fromString("x")}))),
+ arguments(DataTypes.of(ORDER), GenericRowData.of(1, 42)),
+ arguments(DataTypes.of(MONEY), 42));
+ }
+
+ private static Variant copyOf(final Variant view) {
+ final BinaryVariant binary = (BinaryVariant) view;
+ return new BinaryVariant(binary.getValue(), binary.getMetadata());
+ }
+
+ private static DecimalData decimal(final String value, final int
precision, final int scale) {
+ return DecimalData.fromBigDecimal(new BigDecimal(value), precision,
scale);
+ }
+
+ private static GenericMapData mapOf(final Object key, final Object value) {
+ final Map<Object, Object> map = new HashMap<>();
+ map.put(key, value);
+ return new GenericMapData(map);
+ }
+
+ private static Object convert(final DataType dataType, final Variant
variant) {
+ return convert(dataType.getLogicalType(), variant);
+ }
+
+ private static Object convert(final LogicalType type, final Variant
variant) {
+ return FromVariantConverter.create(type,
ZoneOffset.UTC).convert(variant);
+ }
+}
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/functions/VariantCastUtilsTest.java
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/utils/VariantCastUtilsTest.java
similarity index 96%
rename from
flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/functions/VariantCastUtilsTest.java
rename to
flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/utils/VariantCastUtilsTest.java
index f07452165d3..89db5d619ef 100644
---
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/functions/VariantCastUtilsTest.java
+++
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/utils/VariantCastUtilsTest.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.flink.table.runtime.functions;
+package org.apache.flink.table.data.utils;
import org.apache.flink.types.variant.BinaryVariant;
import org.apache.flink.types.variant.Variant;
@@ -27,7 +27,7 @@ import org.junit.jupiter.api.Test;
import java.time.Instant;
import java.util.TimeZone;
-import static
org.apache.flink.table.runtime.functions.VariantCastUtils.toPrintString;
+import static org.apache.flink.table.data.utils.VariantCastUtils.toPrintString;
import static org.apache.flink.types.variant.BinaryVariantUtil.primitiveHeader;
import static org.apache.flink.types.variant.BinaryVariantUtil.shortStrHeader;
import static org.assertj.core.api.Assertions.assertThat;
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/AbstractCodeGeneratorCastRule.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/AbstractCodeGeneratorCastRule.java
index 4390717de40..55903ab390e 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/AbstractCodeGeneratorCastRule.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/AbstractCodeGeneratorCastRule.java
@@ -27,6 +27,7 @@ import
org.apache.flink.table.runtime.typeutils.InternalSerializers;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.util.FlinkRuntimeException;
+import java.time.ZoneId;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashMap;
@@ -208,6 +209,11 @@ abstract class AbstractCodeGeneratorCastRule<IN, OUT>
extends AbstractCastRule<I
+ "\")";
}
+ @Override
+ public ZoneId getSessionZoneId() {
+ return castRuleCtx.getSessionZoneId();
+ }
+
@Override
public String declareVariable(String type, String variablePrefix) {
String variableName = variablePrefix + "$" + variableIndex;
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/AbstractExpressionCodeGeneratorCastRule.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/AbstractExpressionCodeGeneratorCastRule.java
index feda119a1d6..8d67dfdf9d2 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/AbstractExpressionCodeGeneratorCastRule.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/AbstractExpressionCodeGeneratorCastRule.java
@@ -24,6 +24,7 @@ import org.apache.flink.table.runtime.generated.CompileUtils;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.utils.LogicalTypeUtils;
+import java.time.ZoneId;
import java.util.Collections;
import static
org.apache.flink.table.planner.codegen.CodeGenUtils.boxedTypeTermForType;
@@ -119,6 +120,11 @@ abstract class AbstractExpressionCodeGeneratorCastRule<IN,
OUT>
return "java.util.TimeZone.getTimeZone(\"" +
ctx.getSessionZoneId().getId() + "\")";
}
+ @Override
+ public ZoneId getSessionZoneId() {
+ return ctx.getSessionZoneId();
+ }
+
@Override
public String declareVariable(String type, String variablePrefix) {
throw new UnsupportedOperationException(
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
deleted file mode 100644
index e6d924617f3..00000000000
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/AbstractVariantToConstructedCastRule.java
+++ /dev/null
@@ -1,71 +0,0 @@
-/*
- * 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.types.logical.LogicalType;
-import org.apache.flink.table.types.logical.LogicalTypeRoot;
-import org.apache.flink.types.variant.Variant;
-
-import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.methodCall;
-
-/**
- * Base class for the rules that cast a {@link LogicalTypeRoot#VARIANT} to a
constructed target,
- * imposing a schema on a variant. A constructed cast is the scalar cast
applied to every leaf plus
- * 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, 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> {
-
- protected AbstractVariantToConstructedCastRule(CastRulePredicate
predicate) {
- super(predicate);
- }
-
- @Override
- public boolean canFail(LogicalType inputLogicalType, LogicalType
targetLogicalType) {
- return true;
- }
-
- /**
- * Treats a variant that stores a {@code null} as a {@code NULL} input, so
a top-level VARIANT
- * null casts to SQL {@code NULL} before any shape check runs. Only
applied for a nullable
- * target: a {@code NOT NULL} result cannot carry {@code NULL}, so a
null-valued variant then
- * fails the shape check as a regular 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);
- }
-}
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 485b1ab89b1..dcd6cdb891c 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
@@ -96,11 +96,7 @@ public class CastRuleProvider {
.addRule(MapToMapAndMultisetToMultisetCastRule.INSTANCE)
.addRule(RowToRowCastRule.INSTANCE)
// Variant rules
- .addRule(VariantToStringCastRule.INSTANCE)
- .addRule(VariantToPrimitiveCastRule.INSTANCE)
- .addRule(VariantToArrayCastRule.INSTANCE)
- .addRule(VariantToRowCastRule.INSTANCE)
- .addRule(VariantToMapCastRule.INSTANCE)
+ .addRule(FromVariantCastRule.INSTANCE)
.addRule(ToVariantCastRule.INSTANCE)
// Bitmap rules
.addRule(BitmapToStringCastRule.INSTANCE)
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CodeGeneratorCastRule.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CodeGeneratorCastRule.java
index 17a56dcc623..eb05de6b63d 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CodeGeneratorCastRule.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/CodeGeneratorCastRule.java
@@ -22,6 +22,8 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.table.planner.codegen.CodeGeneratorContext;
import org.apache.flink.table.types.logical.LogicalType;
+import java.time.ZoneId;
+
/**
* Cast rule that has code generation capabilities.
*
@@ -62,6 +64,11 @@ public interface CodeGeneratorCastRule<IN, OUT> extends
CastRule<IN, OUT> {
*/
String getSessionTimeZoneTerm();
+ /**
+ * @return the session time zone.
+ */
+ ZoneId getSessionZoneId();
+
/**
* Declare a new variable accessible within the scope of the caller of
{@link
* #generateCodeBlock(Context, String, String, LogicalType,
LogicalType)}.
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/FromVariantCastRule.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/FromVariantCastRule.java
new file mode 100644
index 00000000000..14985f60337
--- /dev/null
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/FromVariantCastRule.java
@@ -0,0 +1,125 @@
+/*
+ * 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.utils.FromVariantConverter;
+import org.apache.flink.table.data.utils.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.utils.LogicalTypeCasts;
+import org.apache.flink.types.variant.Variant;
+
+import static
org.apache.flink.table.planner.codegen.CodeGenUtils.boxedTypeTermForType;
+import static
org.apache.flink.table.planner.codegen.calls.BuiltInMethods.BINARY_STRING_DATA_FROM_STRING;
+import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.cast;
+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.unbox;
+
+/**
+ * Cast rule from {@link LogicalTypeRoot#VARIANT} to every type that {@link
LogicalTypeCasts}
+ * allows. The generated code calls a {@link FromVariantConverter}, which also
trims and pads a
+ * {@code CHAR(n)} or {@code VARCHAR(n)} target. Printing calls {@link
+ * VariantCastUtils#toPrintString} instead, because a printed value must never
fail.
+ */
+class FromVariantCastRule extends
AbstractNullAwareCodeGeneratorCastRule<Variant, Object> {
+
+ static final FromVariantCastRule INSTANCE = new FromVariantCastRule();
+
+ private FromVariantCastRule() {
+
super(CastRulePredicate.builder().predicate(FromVariantCastRule::matches).build());
+ }
+
+ private static boolean matches(final LogicalType input, final LogicalType
target) {
+ return input.is(LogicalTypeRoot.VARIANT)
+ && !target.is(LogicalTypeRoot.VARIANT)
+ && LogicalTypeCasts.supportsExplicitCast(input, target);
+ }
+
+ private static boolean isPrinting(
+ final CodeGeneratorCastRule.Context context, final LogicalType
target) {
+ return context.isPrinting() &&
target.is(LogicalTypeFamily.CHARACTER_STRING);
+ }
+
+ @Override
+ public boolean canFail(
+ final LogicalType inputLogicalType, final LogicalType
targetLogicalType) {
+ return true;
+ }
+
+ /** Maps a variant null to SQL {@code NULL} for a nullable target, except
when printing. */
+ @Override
+ public CastCodeBlock generateCodeBlock(
+ final CodeGeneratorCastRule.Context context,
+ final String inputTerm,
+ final String inputIsNullTerm,
+ final LogicalType inputLogicalType,
+ final LogicalType targetLogicalType) {
+ if (!targetLogicalType.isNullable() || isPrinting(context,
targetLogicalType)) {
+ return super.generateCodeBlock(
+ context, inputTerm, inputIsNullTerm, inputLogicalType,
targetLogicalType);
+ }
+ final String isNullTerm =
+ "(" + inputIsNullTerm + " || " + methodCall(inputTerm,
"isNull") + ")";
+ return super.generateCodeBlock(
+ context, inputTerm, isNullTerm, inputLogicalType,
targetLogicalType);
+ }
+
+ /* Example generated code for INT, inside the null check of the base class:
+
+ result$1 =
((java.lang.Integer)(fromVariantConverter$2.convert(variant$0))).intValue();
+
+ */
+ @Override
+ protected String generateCodeBlockInternal(
+ final CodeGeneratorCastRule.Context context,
+ final String inputTerm,
+ final String returnVariable,
+ final LogicalType inputLogicalType,
+ final LogicalType targetLogicalType) {
+ if (isPrinting(context, targetLogicalType)) {
+ // toPrintString returns a String, which the result has to wrap.
+ return new CastRuleUtils.CodeWriter()
+ .assignStmt(
+ returnVariable,
+ staticCall(
+ BINARY_STRING_DATA_FROM_STRING(),
+ staticCall(
+ VariantCastUtils.class,
+ "toPrintString",
+ inputTerm,
+ context.getSessionTimeZoneTerm())))
+ .toString();
+ }
+ final String converterTerm =
+ context.declareReusableObject(
+ FromVariantConverter.create(targetLogicalType,
context.getSessionZoneId()),
+ "fromVariantConverter");
+ return new CastRuleUtils.CodeWriter()
+ .assignStmt(
+ returnVariable,
+ unbox(
+ cast(
+
boxedTypeTermForType(targetLogicalType),
+ methodCall(converterTerm, "convert",
inputTerm)),
+ targetLogicalType))
+ .toString();
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/VariantToArrayCastRule.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/VariantToArrayCastRule.java
deleted file mode 100644
index 5761cb55044..00000000000
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/VariantToArrayCastRule.java
+++ /dev/null
@@ -1,171 +0,0 @@
-/*
- * 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.ArrayData;
-import org.apache.flink.table.data.GenericArrayData;
-import org.apache.flink.table.planner.codegen.CodeGenUtils;
-import org.apache.flink.table.runtime.functions.VariantCastUtils;
-import org.apache.flink.table.types.logical.ArrayType;
-import org.apache.flink.table.types.logical.LogicalType;
-import org.apache.flink.table.types.logical.LogicalTypeRoot;
-import org.apache.flink.types.variant.Variant;
-
-import static org.apache.flink.table.planner.codegen.CodeGenUtils.newName;
-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.newArray;
-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#ARRAY} cast rule.
- *
- * <p>The variant must be an array, otherwise the cast fails. Each element is
itself a variant and
- * casts to the target element type by the full {@code VARIANT}-to-element
rule, recursively. An
- * element that stores a {@code null} maps to SQL {@code NULL} when the
element type is nullable and
- * fails the cast when it is {@code NOT NULL}. The exception is a {@code
VARIANT} element type:
- * there the element cast is the identity, so a VARIANT null element is kept
as a variant null
- * rather than downgraded to SQL {@code NULL}.
- */
-class VariantToArrayCastRule extends
AbstractVariantToConstructedCastRule<ArrayData> {
-
- static final VariantToArrayCastRule INSTANCE = new
VariantToArrayCastRule();
-
- private VariantToArrayCastRule() {
- super(
- CastRulePredicate.builder()
- .predicate(
- (input, target) ->
- input.is(LogicalTypeRoot.VARIANT)
- &&
target.is(LogicalTypeRoot.ARRAY)
- && CastRuleProvider.resolve(
- input,
- ((ArrayType)
target)
-
.getElementType())
- != null)
- .build());
- }
-
- /* Example generated code for ARRAY<INT>:
-
- int arraySize$2 =
-
org.apache.flink.table.runtime.functions.VariantCastUtils.arraySize(
- variant$1, "ARRAY<INT>");
- java.lang.Integer[] objArray$3 = new java.lang.Integer[arraySize$2];
- for (int i$4 = 0; i$4 < arraySize$2; i$4++) {
- org.apache.flink.types.variant.Variant element$5 =
variant$1.getElement(i$4);
- if (!element$5.isNull()) {
- result$6 =
- ((int)
org.apache.flink.table.runtime.functions.VariantCastUtils.toIntegral(
- element$5, -2147483648L, 2147483647L, "INTEGER"));
- objArray$3[i$4] = result$6;
- }
- }
- result$0 = new org.apache.flink.table.data.GenericArrayData(objArray$3);
-
- A VARIANT null element leaves the slot null (SQL NULL); a NOT NULL element
type throws instead.
-
- */
- @Override
- protected String generateCodeBlockInternal(
- CodeGeneratorCastRule.Context context,
- String inputTerm,
- String returnVariable,
- LogicalType inputLogicalType,
- LogicalType targetLogicalType) {
- final LogicalType elementType = ((ArrayType)
targetLogicalType).getElementType();
- final String elementTypeTerm = arrayElementType(elementType);
- final String sizeTerm = newName(context.getCodeGeneratorContext(),
"arraySize");
- final String arrayTerm = newName(context.getCodeGeneratorContext(),
"objArray");
- final String elementTerm = newName(context.getCodeGeneratorContext(),
"element");
-
- // For a typed element the VARIANT null is handled in the loop below,
so the inner cast is
- // the plain VARIANT-to-element rule on a non-null variant: a nullable
element maps it to
- // SQL NULL, a NOT NULL element fails. For a VARIANT element the inner
cast is the identity,
- // which keeps a variant null as-is.
- final CastCodeBlock elementCast =
- CastRuleProvider.generateAlwaysNonNullCodeBlock(
- context, elementTerm, inputLogicalType, elementType);
-
- return new CastRuleUtils.CodeWriter()
- .declStmt(
- int.class,
- sizeTerm,
- staticCall(
- VariantCastUtils.class,
- "arraySize",
- inputTerm,
-
strLiteral(targetLogicalType.asSummaryString())))
- .declStmt(elementTypeTerm + "[]", arrayTerm,
newArray(elementTypeTerm, sizeTerm))
- .forStmt(
- sizeTerm,
- (index, loopWriter) -> {
- loopWriter.declStmt(
- Variant.class,
- elementTerm,
- methodCall(inputTerm, "getElement",
index));
- if (elementType.is(LogicalTypeRoot.VARIANT)) {
- // The element cast is the identity, so a
VARIANT null element is a
- // valid variant null and is kept as-is rather
than downgraded to
- // SQL NULL, matching the top-level VARIANT
cast.
- loopWriter
- .append(elementCast)
- .assignArrayStmt(
- arrayTerm, index,
elementCast.getReturnTerm());
- return;
- }
- final String isPresent = "!" +
methodCall(elementTerm, "isNull");
- if (elementType.isNullable()) {
- loopWriter.ifStmt(
- isPresent,
- thenWriter ->
- thenWriter
- .append(elementCast)
- .assignArrayStmt(
- arrayTerm,
- index,
-
elementCast.getReturnTerm()));
- } else {
- loopWriter.ifStmt(
- isPresent,
- thenWriter ->
- thenWriter
- .append(elementCast)
- .assignArrayStmt(
- arrayTerm,
- index,
-
elementCast.getReturnTerm()),
- elseWriter ->
- elseWriter.throwStmt(
- "new
org.apache.flink.table.api.TableRuntimeException(\"Cannot cast a VARIANT null
array element to a NOT NULL element type.\")"));
- }
- },
- context.getCodeGeneratorContext())
- .assignStmt(returnVariable,
constructorCall(GenericArrayData.class, arrayTerm))
- .toString();
- }
-
- private static String arrayElementType(LogicalType elementType) {
- if (elementType.isNullable()) {
- return CodeGenUtils.boxedTypeTermForType(elementType);
- }
- return CodeGenUtils.primitiveTypeTermForType(elementType);
- }
-}
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
deleted file mode 100644
index 71a8369de0d..00000000000
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/VariantToMapCastRule.java
+++ /dev/null
@@ -1,208 +0,0 @@
-/*
- * 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/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
deleted file mode 100644
index b0bf6703e64..00000000000
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/VariantToPrimitiveCastRule.java
+++ /dev/null
@@ -1,241 +0,0 @@
-/*
- * 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.planner.functions.casting.CastRuleUtils.CodeWriter;
-import org.apache.flink.table.runtime.functions.VariantCastUtils;
-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 static
org.apache.flink.table.planner.codegen.CodeGenUtils.primitiveTypeTermForType;
-import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.cast;
-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 primitive type cast rule.
- *
- * <p>A cast succeeds only when the target holds the stored value without
altering it, otherwise it
- * fails {@code CAST} and yields {@code null} for {@code TRY_CAST}. An integer
widens or narrows as
- * long as it stays in range, a {@code DECIMAL} has to fit the precision and
scale without rounding,
- * and a timestamp has to fit the target precision. {@code FLOAT} and {@code
DOUBLE} are approximate
- * by definition, so they take any numeric kind and reject only a magnitude
out of range. Changing
- * the kind itself is not implicit, so a decimal is not read as an integer and
a {@code TIMESTAMP}
- * is not read as a {@code TIMESTAMP_LTZ}.
- *
- * <p>{@code CHARACTER_STRING} is handled by {@link VariantToStringCastRule}.
- */
-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 TIME_WITHOUT_TIME_ZONE:
- case TIMESTAMP_WITHOUT_TIME_ZONE:
- case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
- case UUID:
- 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:
- writer.assignStmt(
- returnVariable,
- cast(
- primitiveTypeTermForType(targetLogicalType),
- staticCall(
- VariantCastUtils.class,
- "toIntegral",
- inputTerm,
- // A long literal needs the suffix to
compile, since
- // Long.MIN_VALUE does not fit an int
literal.
- integralMin(targetLogicalType) + "L",
- integralMax(targetLogicalType) + "L",
-
strLiteral(targetLogicalType.getTypeRoot().name()))));
- break;
- case FLOAT:
- writer.assignStmt(
- returnVariable, staticCall(VariantCastUtils.class,
"toFloat", inputTerm));
- break;
- case DOUBLE:
- writer.assignStmt(
- returnVariable, staticCall(VariantCastUtils.class,
"toDouble", inputTerm));
- break;
- case DECIMAL:
- final DecimalType decimalType = (DecimalType)
targetLogicalType;
- writer.assignStmt(
- returnVariable,
- staticCall(
- VariantCastUtils.class,
- "toDecimal",
- inputTerm,
- decimalType.getPrecision(),
- decimalType.getScale()));
- break;
- case BINARY:
- case VARBINARY:
- writer.assignStmt(
- returnVariable,
- staticCall(
- VariantCastUtils.class,
- "toBytes",
- inputTerm,
- LogicalTypeChecks.getLength(targetLogicalType),
- targetLogicalType.is(LogicalTypeRoot.BINARY)));
- break;
- case DATE:
- writer.assignStmt(
- returnVariable,
- cast("int", methodCall(methodCall(inputTerm,
"getDate"), "toEpochDay")));
- break;
- case TIME_WITHOUT_TIME_ZONE:
- writer.assignStmt(
- returnVariable,
- staticCall(
- VariantCastUtils.class,
- "toTime",
- inputTerm,
-
LogicalTypeChecks.getPrecision(targetLogicalType)));
- break;
- case TIMESTAMP_WITHOUT_TIME_ZONE:
- writer.assignStmt(
- returnVariable,
- staticCall(
- VariantCastUtils.class,
- "toTimestamp",
- inputTerm,
-
LogicalTypeChecks.getPrecision(targetLogicalType)));
- break;
- case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
- writer.assignStmt(
- returnVariable,
- staticCall(
- VariantCastUtils.class,
- "toTimestampLtz",
- inputTerm,
-
LogicalTypeChecks.getPrecision(targetLogicalType)));
- break;
- case UUID:
- writer.assignStmt(
- returnVariable, staticCall(VariantCastUtils.class,
"toUuid", inputTerm));
- break;
- default:
- throw new IllegalArgumentException(
- "Unsupported target type for casting from VARIANT: " +
targetLogicalType);
- }
- return writer.toString();
- }
-
- private static long integralMin(LogicalType target) {
- switch (target.getTypeRoot()) {
- case TINYINT:
- return Byte.MIN_VALUE;
- case SMALLINT:
- return Short.MIN_VALUE;
- case INTEGER:
- return Integer.MIN_VALUE;
- default:
- return Long.MIN_VALUE;
- }
- }
-
- private static long integralMax(LogicalType target) {
- switch (target.getTypeRoot()) {
- case TINYINT:
- return Byte.MAX_VALUE;
- case SMALLINT:
- return Short.MAX_VALUE;
- case INTEGER:
- return Integer.MAX_VALUE;
- default:
- return Long.MAX_VALUE;
- }
- }
-}
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
deleted file mode 100644
index ade8e61f128..00000000000
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/VariantToRowCastRule.java
+++ /dev/null
@@ -1,234 +0,0 @@
-/*
- * 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/main/java/org/apache/flink/table/planner/functions/casting/VariantToStringCastRule.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/VariantToStringCastRule.java
deleted file mode 100644
index 955f6530c0f..00000000000
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/VariantToStringCastRule.java
+++ /dev/null
@@ -1,121 +0,0 @@
-/*
- * 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.StringData;
-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.utils.LogicalTypeChecks;
-import org.apache.flink.types.variant.Variant;
-
-import static
org.apache.flink.table.planner.codegen.calls.BuiltInMethods.BINARY_STRING_DATA_FROM_STRING;
-import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.methodCall;
-import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.staticCall;
-
-/**
- * {@link LogicalTypeRoot#VARIANT} to {@link
LogicalTypeFamily#CHARACTER_STRING} cast rule.
- *
- * <p>Renders the scalar value the way a regular SQL cast of the stored kind
would, so a boolean
- * becomes {@code TRUE}, a timestamp uses the SQL format, and a {@code
TIMESTAMP_LTZ} is shifted
- * into the session time zone and a binary value is read as UTF-8. A variant
holding an object or
- * array has no scalar form, so it renders like a regular {@code MAP} or
{@code ARRAY} to string
- * cast, an array as {@code [e1, e2]} and an object as {@code {k1=v1, k2=v2}},
with strings unquoted
- * at every depth. This is a SQL rendering, not JSON; use {@code JSON_STRING}
for the JSON form.
- *
- * <p>A binary value that is not well-formed UTF-8 fails instead of decoding
to {@code U+FFFD}. Cast
- * it to {@code BYTES} to inspect the raw value, or wrap that in {@code
MAKE_VALID_UTF8} to accept
- * the lenient decode explicitly.
- *
- * <p>A value that does not fit a {@code CHAR(n)} or {@code VARCHAR(n)} target
is trimmed, and a
- * {@code CHAR} target pads a shorter value to its fixed width. The length is
enforced here rather
- * than by {@link CharVarCharTrimPadCastRule}, so that a variant storing a
JSON {@code null} still
- * reaches SQL {@code NULL} and so that trimming counts code points rather
than UTF-16 units.
- *
- * <p>Printing uses {@link VariantCastUtils#toPrintString}, which renders the
same way but never
- * fails.
- */
-class VariantToStringCastRule extends
AbstractExpressionCodeGeneratorCastRule<Variant, StringData> {
-
- static final VariantToStringCastRule INSTANCE = new
VariantToStringCastRule();
-
- private VariantToStringCastRule() {
- super(
- CastRulePredicate.builder()
- .input(LogicalTypeRoot.VARIANT)
- .target(LogicalTypeRoot.CHAR)
- .target(LogicalTypeRoot.VARCHAR)
- .build());
- }
-
- @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 the text {@code null}. Only applied for a
nullable target: a {@code
- * NOT NULL} result cannot carry {@code NULL}, so a null-valued variant
then fails. Printing
- * skips this, so the formatter shows {@code NULL}.
- */
- @Override
- public CastCodeBlock generateCodeBlock(
- CodeGeneratorCastRule.Context context,
- String inputTerm,
- String inputIsNullTerm,
- LogicalType inputLogicalType,
- LogicalType targetLogicalType) {
- if (!targetLogicalType.isNullable() || context.isPrinting()) {
- return super.generateCodeBlock(
- context, inputTerm, inputIsNullTerm, inputLogicalType,
targetLogicalType);
- }
- final String isNullTerm =
- "(" + inputIsNullTerm + " || " + methodCall(inputTerm,
"isNull") + ")";
- return super.generateCodeBlock(
- context, inputTerm, isNullTerm, inputLogicalType,
targetLogicalType);
- }
-
- @Override
- public String generateExpression(
- CodeGeneratorCastRule.Context context,
- String inputTerm,
- LogicalType inputLogicalType,
- LogicalType targetLogicalType) {
- if (context.isPrinting()) {
- // toPrintString returns a String, so it needs the wrap that
toStringValue applies
- // itself.
- return staticCall(
- BINARY_STRING_DATA_FROM_STRING(),
- staticCall(
- VariantCastUtils.class,
- "toPrintString",
- inputTerm,
- context.getSessionTimeZoneTerm()));
- }
- return staticCall(
- VariantCastUtils.class,
- "toStringValue",
- inputTerm,
- context.getSessionTimeZoneTerm(),
- LogicalTypeChecks.getLength(targetLogicalType),
- targetLogicalType.is(LogicalTypeRoot.CHAR));
- }
-}
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala
index bff1d79dd1c..5fabb0ad3a5 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala
@@ -2147,6 +2147,7 @@ object ScalarOperatorGens {
override def isPrinting(): Boolean = false
override def legacyBehaviour(): Boolean =
isLegacyCastBehaviourEnabled(ctx)
override def getSessionTimeZoneTerm: String =
ctx.addReusableSessionTimeZone()
+ override def getSessionZoneId: ZoneId =
TableConfigUtils.getLocalTimeZone(ctx.tableConfig)
override def declareVariable(ty: String, variablePrefix: String): String
=
ctx.addReusableLocalVariable(ty, variablePrefix)
override def declareTypeSerializer(ty: LogicalType): String =
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 714df428b58..e3eba6e9a73 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
@@ -518,6 +518,12 @@ public class CastFunctionITCase extends
BuiltInFunctionTestBase {
"CAST(CAST(f5 AS VARIANT) AS
TIMESTAMP_LTZ(3))",
instant,
TIMESTAMP_LTZ(3))
+ // a TIMESTAMP_LTZ renders as a string in the session
time zone
+ .testResult(
+ $("f5").cast(VARIANT()).cast(STRING()),
+ "CAST(CAST(f5 AS VARIANT) AS STRING)",
+ "2026-09-25 18:15:30.123000",
+ STRING())
.testResult(
$("f6").cast(VARIANT()).cast(UUID()),
"CAST(CAST(f6 AS VARIANT) AS UUID)",
@@ -1052,6 +1058,7 @@ public class CastFunctionITCase extends
BuiltInFunctionTestBase {
final String objNull = "{\"id\": 7, \"name\": null}";
final String nested =
"{\"user\": {\"id\": 1, \"since\": \"2020-01-01\"}, \"tags\":
[\"x\", \"y\"]}";
+ final DataType stringRow = ROW(FIELD("d", STRING()));
return List.of(
TestSetSpec.forExpression("Cast a VARIANT produced by
parseJson() to a ROW")
.onFieldsWithData("null")
@@ -1155,7 +1162,42 @@ public class CastFunctionITCase extends
BuiltInFunctionTestBase {
FIELD(
"user",
ROW(FIELD("id", INT()),
FIELD("since", STRING()))),
- FIELD("tags", ARRAY(STRING())))));
+ FIELD("tags", ARRAY(STRING())))),
+ // A ROW cast from VARIANT compares field by field, like a ROW
built with ROW(...):
+ // NaN never equals NaN, and 0.0 equals -0.0. PARSE_JSON
rejects NaN and
+ // infinities, so each VARIANT is built from a row of strings.
Each pair spells its
+ // value differently, because the planner merges equal inputs
and folds x = x to
+ // TRUE.
+ TestSetSpec.forExpression("Compare ROW values cast from
VARIANT")
+ .onFieldsWithData(
+ Row.of("NaN"),
+ Row.of("nan"),
+ Row.of("0.0"),
+ Row.of("-0.0"),
+ Row.of("Infinity"),
+ Row.of("inf"),
+ Row.of("-Infinity"))
+ .andDataTypes(
+ stringRow, stringRow, stringRow, stringRow,
stringRow, stringRow,
+ stringRow)
+ // NaN = NaN
+ .testSqlResult(variantRowsEqual("f0", "f1"), false,
BOOLEAN())
+ // 0.0 = -0.0
+ .testSqlResult(variantRowsEqual("f2", "f3"), true,
BOOLEAN())
+ // Infinity = Infinity
+ .testSqlResult(variantRowsEqual("f4", "f5"), true,
BOOLEAN())
+ // Infinity = -Infinity
+ .testSqlResult(variantRowsEqual("f4", "f6"), false,
BOOLEAN()));
+ }
+
+ /** Casts two rows of strings to {@code ROW<d DOUBLE>} through a VARIANT
and compares them. */
+ private static String variantRowsEqual(final String left, final String
right) {
+ return variantRow(left) + " = " + variantRow(right);
+ }
+
+ private static String variantRow(final String field) {
+ return String.format(
+ "CAST(CAST(CAST(%s AS ROW<d DOUBLE>) AS VARIANT) AS ROW<d
DOUBLE>)", field);
}
private static List<TestSetSpec> variantMapCasts() {
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 5d9915c16cd..3ce28626217 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
@@ -168,10 +168,9 @@ class CastRuleProviderTest {
@Test
void testResolveVariantToPrimitive() {
- assertThat(CastRuleProvider.resolve(VARIANT, INT))
- .isSameAs(VariantToPrimitiveCastRule.INSTANCE);
+ assertThat(CastRuleProvider.resolve(VARIANT,
INT)).isSameAs(FromVariantCastRule.INSTANCE);
assertThat(CastRuleProvider.resolve(VARIANT,
BOOLEAN().getLogicalType()))
- .isSameAs(VariantToPrimitiveCastRule.INSTANCE);
+ .isSameAs(FromVariantCastRule.INSTANCE);
assertThat(CastRuleProvider.exists(VARIANT, DECIMAL(10,
2).getLogicalType())).isTrue();
assertThat(CastRuleProvider.exists(VARIANT,
DATE().getLogicalType())).isTrue();
assertThat(CastRuleProvider.exists(VARIANT,
TIMESTAMP().getLogicalType())).isTrue();
@@ -184,9 +183,11 @@ class CastRuleProviderTest {
// INTERVAL has no VARIANT counterpart, so it is not a castable target
assertThat(CastRuleProvider.exists(VARIANT,
INTERVAL(DataTypes.DAY()).getLogicalType()))
.isFalse();
- // character strings keep going through the display-oriented rule
+ // character strings go through the same rule, including a bounded
length
assertThat(CastRuleProvider.resolve(VARIANT, STRING_TYPE))
- .isSameAs(VariantToStringCastRule.INSTANCE);
+ .isSameAs(FromVariantCastRule.INSTANCE);
+ assertThat(CastRuleProvider.resolve(VARIANT, CHAR(3).getLogicalType()))
+ .isSameAs(FromVariantCastRule.INSTANCE);
}
@Test
@@ -317,7 +318,7 @@ class CastRuleProviderTest {
@Test
void testResolveVariantToArray() {
assertThat(CastRuleProvider.resolve(VARIANT,
ARRAY(INT()).getLogicalType()))
- .isSameAs(VariantToArrayCastRule.INSTANCE);
+ .isSameAs(FromVariantCastRule.INSTANCE);
// the element recurses through the VARIANT rules, including the
identity leaf and nesting
assertThat(CastRuleProvider.exists(VARIANT,
ARRAY(VARIANT()).getLogicalType())).isTrue();
@@ -334,10 +335,10 @@ class CastRuleProviderTest {
@Test
void testResolveVariantToRow() {
assertThat(CastRuleProvider.resolve(VARIANT, ROW(FIELD("f0",
INT())).getLogicalType()))
- .isSameAs(VariantToRowCastRule.INSTANCE);
+ .isSameAs(FromVariantCastRule.INSTANCE);
// a structured target shares the ROW rule
assertThat(CastRuleProvider.resolve(VARIANT, STRUCTURED))
- .isSameAs(VariantToRowCastRule.INSTANCE);
+ .isSameAs(FromVariantCastRule.INSTANCE);
// a field with no variant counterpart makes the whole cast
unresolvable
assertThat(
@@ -349,7 +350,7 @@ class CastRuleProviderTest {
@Test
void testResolveVariantToMap() {
assertThat(CastRuleProvider.resolve(VARIANT, MAP(STRING(),
INT()).getLogicalType()))
- .isSameAs(VariantToMapCastRule.INSTANCE);
+ .isSameAs(FromVariantCastRule.INSTANCE);
// the value recurses through the VARIANT rules, including the
identity leaf
assertThat(CastRuleProvider.exists(VARIANT, MAP(STRING(),
VARIANT()).getLogicalType()))
.isTrue();
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/utils/FilterUtils.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/utils/FilterUtils.java
index 6965d55cef7..b2087b32de8 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/utils/FilterUtils.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/utils/FilterUtils.java
@@ -19,6 +19,8 @@
package org.apache.flink.table.planner.utils;
import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.utils.FromVariantConverter;
import org.apache.flink.table.expressions.CallExpression;
import org.apache.flink.table.expressions.Expression;
import org.apache.flink.table.expressions.FieldReferenceExpression;
@@ -27,26 +29,22 @@ import
org.apache.flink.table.expressions.ResolvedExpression;
import org.apache.flink.table.expressions.ValueLiteralExpression;
import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
import org.apache.flink.table.functions.FunctionDefinition;
-import org.apache.flink.table.runtime.functions.VariantCastUtils;
-import org.apache.flink.table.types.logical.CharType;
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.VarCharType;
import org.apache.flink.types.variant.Variant;
import org.apache.flink.util.Preconditions;
import javax.annotation.Nullable;
import java.lang.reflect.Array;
-import java.util.EnumMap;
+import java.time.ZoneOffset;
+import java.util.EnumSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
-import java.util.TimeZone;
-import java.util.function.BiFunction;
import java.util.function.Function;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.AT;
@@ -58,57 +56,18 @@ import static
org.apache.flink.table.functions.BuiltInFunctionDefinitions.UPPER;
/** Utils for catalog and source to filter partition or row. */
public class FilterUtils {
- private static final TimeZone UTC = TimeZone.getTimeZone("UTC");
-
- /**
- * Casts of a VARIANT element that can be evaluated, by target type, using
the same {@link
- * VariantCastUtils} helpers as the generated code of {@code
VariantToPrimitiveCastRule}.
- */
- private static final Map<LogicalTypeRoot, BiFunction<Variant, LogicalType,
Object>>
- VARIANT_CASTS = new EnumMap<>(LogicalTypeRoot.class);
-
- static {
- VARIANT_CASTS.put(LogicalTypeRoot.BOOLEAN, (variant, type) ->
variant.getBoolean());
- VARIANT_CASTS.put(
- LogicalTypeRoot.TINYINT,
- (variant, type) ->
- (byte)
- VariantCastUtils.toIntegral(
- variant, Byte.MIN_VALUE,
Byte.MAX_VALUE, "TINYINT"));
- VARIANT_CASTS.put(
- LogicalTypeRoot.SMALLINT,
- (variant, type) ->
- (short)
- VariantCastUtils.toIntegral(
- variant, Short.MIN_VALUE,
Short.MAX_VALUE, "SMALLINT"));
- VARIANT_CASTS.put(
- LogicalTypeRoot.INTEGER,
- (variant, type) ->
- (int)
- VariantCastUtils.toIntegral(
- variant, Integer.MIN_VALUE,
Integer.MAX_VALUE, "INT"));
- VARIANT_CASTS.put(
- LogicalTypeRoot.BIGINT,
- (variant, type) ->
- VariantCastUtils.toIntegral(
- variant, Long.MIN_VALUE, Long.MAX_VALUE,
"BIGINT"));
- VARIANT_CASTS.put(
- LogicalTypeRoot.FLOAT, (variant, type) ->
VariantCastUtils.toFloat(variant));
- VARIANT_CASTS.put(
- LogicalTypeRoot.DOUBLE, (variant, type) ->
VariantCastUtils.toDouble(variant));
- VARIANT_CASTS.put(
- LogicalTypeRoot.CHAR,
- (variant, type) ->
- VariantCastUtils.toStringValue(
- variant, UTC, ((CharType)
type).getLength(), true)
- .toString());
- VARIANT_CASTS.put(
- LogicalTypeRoot.VARCHAR,
- (variant, type) ->
- VariantCastUtils.toStringValue(
- variant, UTC, ((VarCharType)
type).getLength(), false)
- .toString());
- }
+ /** Cast targets of a VARIANT element whose values compare with a filter
literal. */
+ private static final Set<LogicalTypeRoot> VARIANT_CAST_TARGETS =
+ EnumSet.of(
+ LogicalTypeRoot.BOOLEAN,
+ LogicalTypeRoot.TINYINT,
+ LogicalTypeRoot.SMALLINT,
+ LogicalTypeRoot.INTEGER,
+ LogicalTypeRoot.BIGINT,
+ LogicalTypeRoot.FLOAT,
+ LogicalTypeRoot.DOUBLE,
+ LogicalTypeRoot.CHAR,
+ LogicalTypeRoot.VARCHAR);
public static boolean shouldPushDown(ResolvedExpression expr, Set<String>
filterableFields) {
if (expr instanceof CallExpression && expr.getChildren().size() == 2) {
@@ -190,7 +149,7 @@ public class FilterUtils {
// A cast makes an element of a VARIANT comparable, e.g. CAST(v['k']
AS INT).
if (isCall(expr, CAST) || isCall(expr, TRY_CAST)) {
final ResolvedExpression element =
expr.getResolvedChildren().get(0);
- return VARIANT_CASTS.containsKey(
+ return VARIANT_CAST_TARGETS.contains(
expr.getOutputDataType().getLogicalType().getTypeRoot())
&&
element.getOutputDataType().getLogicalType().is(LogicalTypeRoot.VARIANT)
&& isElementAccess(element, filterableFields);
@@ -297,7 +256,10 @@ public class FilterUtils {
return null;
}
try {
- return VARIANT_CASTS.get(targetType.getTypeRoot()).apply(variant,
targetType);
+ final Object value =
+ FromVariantConverter.create(targetType,
ZoneOffset.UTC).convert(variant);
+ // A string filter literal is a String.
+ return value instanceof StringData ? value.toString() : value;
} catch (RuntimeException e) {
if (tryCast) {
return null;
