This is an automated email from the ASF dual-hosted git repository. raminqaf pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 43a9d4bc31931e3808cbf9d879d1af503815c1e4 Author: Ramin Gharib <[email protected]> AuthorDate: Tue Oct 27 13:20:44 2026 +0100 [FLINK-41241][table] Use VariantFieldGetter for v['key'] and CAST to ROW The generated code for v['literal'] keeps one VariantFieldGetter per call site as a reusable member, and the VARIANT to ROW and STRUCTURED cast keeps one per field. The key now reaches the generated code inside the serialized getter, so it no longer needs escaping. A key that is not a literal passes the UTF-8 bytes of its StringData to BinaryVariant#getFieldByUtf8Key instead of decoding them into a String. The getter assumes a BinaryVariant, like the row writer and the serializer. A UDF that returns another Variant implementation now fails with a ClassCastException on v['k'] in the same projection. Generated-by: Claude Code 2.1.311 (Claude Opus 5.5) --- .../apache/flink/table/data/utils/FromVariantConverter.java | 5 ++++- .../table/planner/codegen/calls/ScalarOperatorGens.scala | 13 ++++++++----- 2 files changed, 12 insertions(+), 6 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 index f612be4bf34..e2909990a75 100644 --- 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 @@ -33,6 +33,7 @@ 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.VariantFieldGetter; import org.apache.flink.types.variant.VariantTypeException; import java.io.Serializable; @@ -254,8 +255,10 @@ public final class FromVariantConverter implements Serializable { final String[] fieldNames = LogicalTypeChecks.getFieldNames(type).toArray(new String[0]); final int fieldCount = fieldTypes.size(); final String rowType = type.asSummaryString(); + final VariantFieldGetter[] fieldGetters = new VariantFieldGetter[fieldCount]; final ValueReader[] fieldReaders = new ValueReader[fieldCount]; for (int i = 0; i < fieldCount; i++) { + fieldGetters[i] = new VariantFieldGetter(fieldNames[i]); fieldReaders[i] = createChildReader( fieldTypes.get(i), @@ -268,7 +271,7 @@ public final class FromVariantConverter implements Serializable { 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]); + final Variant field = fieldGetters[i].getField(variant); if (field == null) { throw new TableRuntimeException(missingFieldError(rowType, fieldNames[i])); } 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 b3dcd9b00bf..891f9c4b19a 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 @@ -40,8 +40,8 @@ import org.apache.flink.table.types.logical.utils.LogicalTypeChecks import org.apache.flink.table.types.logical.utils.LogicalTypeChecks.{getFieldTypes, getPrecision, getScale} import org.apache.flink.table.types.logical.utils.LogicalTypeMerging.findCommonType import org.apache.flink.table.utils.DateTimeUtils.MILLIS_PER_DAY -import org.apache.flink.table.utils.EncodingUtils import org.apache.flink.types.ColumnList +import org.apache.flink.types.variant.{BinaryVariant, VariantFieldGetter} import org.apache.flink.util.Preconditions.checkArgument import java.time.ZoneId @@ -1749,10 +1749,13 @@ object ScalarOperatorGens { } else if (isCharacterString(keyType)) { val fieldAccess = key.literalValue match { case Some(fieldName) if fieldName != null => - s"""$variantTerm.getField("${EncodingUtils.escapeJava(fieldName.toString)}")""" - // A key that is not a literal, or a NULL literal, whose access code never runs. The key is - // a StringData, and getField takes a String. - case _ => s"$variantTerm.getField($keyTerm.toString())" + val getterTerm = + ctx.addReusableObject(new VariantFieldGetter(fieldName.toString), "variantFieldGetter") + s"$getterTerm.getField($variantTerm)" + // A key that is not a literal, or a NULL literal, whose access code never runs. The + // lookup compares the UTF-8 bytes of the StringData key without decoding them. + case _ => + s"((${className[BinaryVariant]}) $variantTerm).getFieldByUtf8Key($keyTerm.toBytes())" } generateCharacterStringKeyAccess( variantTerm,
