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 3cf57c4d2625e82cd393bc10b82d2226b7e42f5e Author: Ramin Gharib <[email protected]> AuthorDate: Tue Oct 27 15:13:45 2026 +0100 [FLINK-41243][table] Support VARIANT field access with a non-literal key VARIANT field access accepts any character string key, but the generated code only handled a literal. A column or another expression as the key, such as v[k] or $("v").at($("k")), failed with NoSuchElementException: None.get, and a NULL literal key failed with a NullPointerException. A key that is not a literal now reads the field with Variant#getField per row. Generated-by: Claude Code 2.1.311 (Claude Opus 5.5) --- .../planner/codegen/calls/ScalarOperatorGens.scala | 15 ++-- .../nodes/exec/stream/VariantSemanticTest.java | 89 ++++++++++++++++++++++ 2 files changed, 99 insertions(+), 5 deletions(-) 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 5fabb0ad3a5..b3dcd9b00bf 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 @@ -1747,11 +1747,17 @@ object ScalarOperatorGens { tmpValue ) } else if (isCharacterString(keyType)) { - val fieldName = key.literalValue.get.toString + 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())" + } generateCharacterStringKeyAccess( variantTerm, variantTypeTerm, - fieldName, + fieldAccess, resultTerm, nullTerm, tmpValue @@ -1777,14 +1783,13 @@ object ScalarOperatorGens { private def generateCharacterStringKeyAccess( variantTerm: String, variantTypeTerm: String, - fieldName: String, + fieldAccess: String, resultTerm: String, nullTerm: String, tmpValue: String): String = { - val escapedFieldName = EncodingUtils.escapeJava(fieldName) s""" | if ($variantTerm.isObject()){ - | $variantTypeTerm $tmpValue = $variantTerm.getField("$escapedFieldName"); + | $variantTypeTerm $tmpValue = $fieldAccess; | if ($tmpValue == null) { | $nullTerm = true; | } else { diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/VariantSemanticTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/VariantSemanticTest.java index c769e326f25..580201100f4 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/VariantSemanticTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/VariantSemanticTest.java @@ -35,6 +35,7 @@ import org.apache.flink.types.variant.VariantBuilder; import java.util.List; import static org.apache.flink.table.api.Expressions.$; +import static org.apache.flink.table.api.Expressions.nullOf; /** Semantic tests for {@link DataTypes#VARIANT()} type. */ public class VariantSemanticTest extends SemanticTestBase { @@ -477,6 +478,48 @@ public class VariantSemanticTest extends SemanticTestBase { "sink_t") .build(); + static final TableTestProgram VARIANT_ARRAY_ACCESS_WITH_NON_LITERAL_INDEX = + TableTestProgram.of( + "variant-array-access-with-non-literal-index", + "validates variant array element access with an index that is not a literal") + .setupTableSource( + SourceTestStep.newBuilder("t") + .addSchema("v VARIANT", "i INT") + .producedValues( + Row.of( + BUILDER.array() + .add(BUILDER.of(1)) + .add(BUILDER.of("hello")) + .build(), + 1), + Row.of( + BUILDER.array() + .add(BUILDER.of(10)) + .add(BUILDER.of("world")) + .build(), + 2), + Row.of(BUILDER.array().add(BUILDER.of(10)).build(), 2), + Row.of(BUILDER.array().add(BUILDER.of(10)).build(), 0)) + .build()) + .setupTableSink( + SinkTestStep.newBuilder("sink_t") + .addSchema("e VARIANT", "n VARIANT") + .consumedValues( + Row.of(BUILDER.of(1), null), + Row.of(BUILDER.of("world"), null), + Row.of(null, null), + Row.of(null, null)) + .build()) + .runSql("INSERT INTO sink_t SELECT v[i], v[CAST(NULL AS INT)] FROM t") + .runTableApi( + t -> + t.from("t") + .select( + $("v").at($("i")).as("e"), + $("v").at(nullOf(DataTypes.INT())).as("n")), + "sink_t") + .build(); + static final TableTestProgram VARIANT_OBJECT_ACCESS = TableTestProgram.of( "variant-object-access", @@ -494,6 +537,50 @@ public class VariantSemanticTest extends SemanticTestBase { "sink_t") .build(); + static final TableTestProgram VARIANT_OBJECT_ACCESS_WITH_NON_LITERAL_KEY = + TableTestProgram.of( + "variant-object-access-with-non-literal-key", + "validates variant object field access with a key that is not a literal") + .setupTableSource( + SourceTestStep.newBuilder("t") + .addSchema("v VARIANT", "k STRING") + .producedValues( + Row.of( + BUILDER.object() + .add("name", BUILDER.of("Alice")) + .add("age", BUILDER.of(30)) + .build(), + "name"), + Row.of( + BUILDER.object() + .add("name", BUILDER.of("Bob")) + .add("age", BUILDER.of(25)) + .build(), + "age"), + Row.of( + BUILDER.object() + .add("name", BUILDER.of("Carol")) + .build(), + "city")) + .build()) + .setupTableSink( + SinkTestStep.newBuilder("sink_t") + .addSchema("f VARIANT", "n VARIANT") + .consumedValues( + Row.of(BUILDER.of("Alice"), null), + Row.of(BUILDER.of(25), null), + Row.of(null, null)) + .build()) + .runSql("INSERT INTO sink_t SELECT v[k], v[CAST(NULL AS STRING)] FROM t") + .runTableApi( + t -> + t.from("t") + .select( + $("v").at($("k")).as("f"), + $("v").at(nullOf(DataTypes.STRING())).as("n")), + "sink_t") + .build(); + static final Variant COMPLEX_VARIANT = BUILDER.object() .add( @@ -609,7 +696,9 @@ public class VariantSemanticTest extends SemanticTestBase { VARIANT_FIELD_AS_JOIN_KEY, VARIANT_ARRAY_ACCESS, VARIANT_ARRAY_ACCESS_WITH_DIFFERENT_INDEX_TYPES, + VARIANT_ARRAY_ACCESS_WITH_NON_LITERAL_INDEX, VARIANT_OBJECT_ACCESS, + VARIANT_OBJECT_ACCESS_WITH_NON_LITERAL_KEY, VARIANT_NESTED_ACCESS, VARIANT_ARRAY_ERROR_ACCESS, VARIANT_OBJECT_ERROR_ACCESS,
