You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "JinkunLiu (via GitHub)" <gi...@apache.org> on 2025/12/09 16:17:36 UTC
[PR] [FLINK-37924][sql]Introduce Built-in Function to Access field or element in the Variant [flink]
JinkunLiu opened a new pull request, #27330:
URL: https://github.com/apache/flink/pull/27330
## What is the purpose of the change
Introduce Built-in Function to Access field or element in the Variant
SQL Function | Table Function | Description
-- | -- | --
variant ‘[’ INT ‘]’ | VARIANT.at(INT) | If the VARIANT is an ARRAY value, returns a VARIANT whose value is the element at the specified index. Otherwise, this operation returns NULL
variant ‘[’ STRING ‘]’ | VARIANT.at(STRING) | If the VARIANT is a MAP value that has an element with this key, a VARIANT holding the associated value is returned. Otherwise, NULL is returned.
## Brief change log
- **Support accessing VARIANT elements using [INT] or VARIANT.at(INT) when the VARIANT contains an ARRAY value in SQL or table expressions.**
- **Support accessing VARIANT elements using ['STRING'] or VARIANT.at('STRING') when the VARIANT contains a MAP value in SQL or table expressions.**
## Verifying this change
This change added tests and can be verified as follows:
- *Added six new test at `org.apache.flink.table.planner.plan.nodes.exec.stream.VariantSemanticTest`*
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no)
- The serializers: (no)
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
- The S3 file system connector: (no)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #27330:
URL: https://github.com/apache/flink/pull/27330#discussion_r2605242708
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ItemAtIndexArgumentTypeStrategy.java:
##########
@@ -86,12 +89,36 @@ public Optional<DataType> inferArgumentType(
}
}
+ if (collectionType.is(LogicalTypeRoot.VARIANT)) {
+ if (indexType.getLogicalType().is(LogicalTypeFamily.INTEGER_NUMERIC)) {
+
+ if (callContext.isArgumentLiteral(1)) {
+ Optional<Integer> literalVal = callContext.getArgumentValue(1, Integer.class);
+ if (literalVal.isPresent() && literalVal.get() <= 0) {
+ return callContext.fail(
+ throwOnFailure,
+ "The provided index must be a valid SQL index starting from 1, but was '%s'",
+ literalVal.get());
+ }
+ }
+
+ return Optional.of(indexType);
+ } else if (indexType.getLogicalType().is(LogicalTypeFamily.CHARACTER_STRING)) {
+ return Optional.of(indexType);
+ } else {
+ return callContext.fail(
+ throwOnFailure,
+ "Variant values can only be accessed with a CHARACTER STRING key or indexed with an INTEGER NUMERIC index.");
+ }
+ }
+
return Optional.empty();
}
@Override
public Signature.Argument getExpectedArgument(
FunctionDefinition functionDefinition, int argumentPos) {
- return Signature.Argument.of("[<INTEGER NUMERIC> | <MAP_KEY_TYPE>]");
+ return Signature.Argument.of(
+ "[<INTEGER NUMERIC> | <MAP_KEY_TYPE> | <CHARACTER STRING> or <INTEGER NUMERIC>]");
Review Comment:
My original intent was to clearly illustrate the one-to-one correspondence between logical root types and their allowed index/key argument types, for example:
- `LogicalTypeRoot.ARRAY` → `<INTEGER NUMERIC>`
- `LogicalTypeRoot.MAP` → `<MAP_KEY_TYPE>`
- `LogicalTypeRoot.VARIANT` → `<CHARACTER STRING>` or `<INTEGER NUMERIC>`
The expected function signatures in old code are:
```
at([<ARRAY> | <MAP> | <VARIANT>], [<INTEGER NUMERIC> | <MAP_KEY_TYPE> | <CHARACTER STRING> or <INTEGER NUMERIC>])
```
However, I have realized that this mapping is already effectively conveyed through the error messages generated by `inferArgumentType()`. I have updated the implementation to reflect this.
Thanks for your reminder!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "yunfengzhou-hub (via GitHub)" <gi...@apache.org>.
yunfengzhou-hub merged PR #27330:
URL: https://github.com/apache/flink/pull/27330
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #27330:
URL: https://github.com/apache/flink/pull/27330#discussion_r2732971894
##########
docs/data/sql_functions.yml:
##########
@@ -1167,6 +1167,16 @@ variant:
parser will keep the last occurrence of all fields with the same key, otherwise when
`allowDuplicateKeys` is false it will throw an error. The default value of
`allowDuplicateKeys` is false.
+ - sql: variant '[' INT ']'
+ table: VARIANT.at(INT)
+ description: |
+ If the VARIANT is an ARRAY value, returns a VARIANT whose value is the element at
+ the specified index. The index starts from 1, If the index is out of range, it returns NULL.
+ - sql: variant '[' STRING ']'
+ table: VARIANT.at(STRING)
+ description: |
+ If the VARIANT is a MAP value that has an element with this key, a VARIANT holding
Review Comment:
Thanks for you review. About duplicate keys has already been explained in detail in the PARSE_JSON method at line 1166~1169. I don't think it needs to be repeated here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #27330:
URL: https://github.com/apache/flink/pull/27330#discussion_r2702341234
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala:
##########
@@ -743,7 +743,11 @@ class ExprCodeGenerator(ctx: CodeGeneratorContext, nullableInput: Boolean)
case LogicalTypeRoot.ROW | LogicalTypeRoot.STRUCTURED_TYPE =>
generateDot(ctx, operands)
- case _ => throw new CodeGenException("Expect an array, a map or a row.")
+ case LogicalTypeRoot.VARIANT =>
+ val key = operands(1)
+ generateVariantGet(ctx, operands.head, key)
Review Comment:
Thanks for the suggestion. Handling nested access in a single codegen pass would likely require some special-case logic in upper-level call sites, which feels a bit hacky and harder to maintain.
For shallow cases, users can use `item access`. For deeper accesses, we plan to provide a dedicated `variantQuery` facility (FLIP-521) for more efficient multi-level extraction:
https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=349637099#FLIP521:IntegratingVariantTypeintoFlink:EnablingEfficientSemiStructuredDataProcessing-Introducebuilt-infunctionsVARIANT_VALUEandVARIANT_QUERYtoextractvaluesfromavariant
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #27330:
URL: https://github.com/apache/flink/pull/27330#discussion_r2673245798
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala:
##########
@@ -1668,6 +1668,108 @@ object ScalarOperatorGens {
generateUnaryOperatorIfNotNull(ctx, resultType, map)(_ => s"${map.resultTerm}.size()")
}
+ def generateVariantGet(
+ ctx: CodeGeneratorContext,
+ variant: GeneratedExpression,
+ key: GeneratedExpression): GeneratedExpression = {
+ val Seq(resultTerm, nullTerm) = newNames(ctx, "result", "isNull")
+ val binaryVariantTerm = newName(ctx, "binaryVariantTerm")
+
+ val variantType = variant.resultType.asInstanceOf[VariantType]
+ val keyType = key.resultType
+
+ val variantTypeTerm = primitiveTypeTermForType(variantType)
+ val variantDefault = primitiveDefaultValue(variantType)
+ val variantTerm = variant.resultTerm
+ val keyTerm = key.resultTerm
+
+ val accessCode = if (isInteger(keyType)) {
+ generateIntegerKeyAccess(
+ variantTerm,
+ binaryVariantTerm,
+ keyTerm,
+ resultTerm,
+ nullTerm
+ )
+ } else if (isCharacterString(keyType)) {
+ generateCharacterStringKeyAccess(
+ variantTerm,
+ binaryVariantTerm,
+ keyTerm,
+ resultTerm,
+ nullTerm
+ )
+ } else {
+ throw new CodeGenException(s"Unsupported key type for variant: $keyType")
+ }
+
+ val finalCode =
+ s"""
+ |${variant.code}
+ |${key.code}
+ |boolean $nullTerm = (${variant.nullTerm}|| ${key.nullTerm});
+ |$variantTypeTerm $resultTerm = $variantDefault;
+ |if (!$nullTerm) {
+ | $accessCode
+ |}
+ """.stripMargin
+
+ GeneratedExpression(resultTerm, nullTerm, finalCode, variantType)
+ }
+
+ private def generateCharacterStringKeyAccess(
+ variantTerm: String,
+ binaryVariantTerm: String,
+ keyTerm: String,
+ resultTerm: String,
+ nullTerm: String): String = {
+ s"""
+ | if ($variantTerm.isObject()){
+ | if ($variantTerm instanceof $BINARY_VARIANT) {
+ | $BINARY_VARIANT $binaryVariantTerm = ($BINARY_VARIANT) $variantTerm;
+ |
+ | String keyStr = null;
+ | if ($keyTerm instanceof $BINARY_STRING) {
+ | keyStr = (($BINARY_STRING) $keyTerm).toString();
+ | } else {
+ | keyStr = String.valueOf($keyTerm);
+ | }
+ |
+ | if (keyStr != null) {
+ | $resultTerm = $binaryVariantTerm.getField(keyStr);
+ | } else {
+ | $nullTerm = true;
+ | }
+ | } else {
+ | $nullTerm = true;
+ | }
+ | } else {
+ | throw new org.apache.flink.table.api.TableRuntimeException("String key access on variant requires an object variant, but a non-object variant was provided.");
+ | }
+ """.stripMargin
+ }
+
+ private def generateIntegerKeyAccess(
+ variantTerm: String,
+ binaryVariantTerm: String,
+ keyTerm: String,
+ resultTerm: String,
+ nullTerm: String): String = {
+ s"""
+ | if ($variantTerm.isArray()){
+ | if ($variantTerm instanceof $BINARY_VARIANT) {
+ | $BINARY_VARIANT $binaryVariantTerm = ($BINARY_VARIANT) $variantTerm;
+ | $resultTerm = $binaryVariantTerm.getElement((int)$keyTerm - 1);
Review Comment:
Thanks for the comment. In Flink SQL, array indexing is 1-based, and I think we should keep it consistent here and also use 1-based indexing. Similarly, in Calcite, `VARIANT` indexing is also 1-based.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "yunfengzhou-hub (via GitHub)" <gi...@apache.org>.
yunfengzhou-hub commented on code in PR #27330:
URL: https://github.com/apache/flink/pull/27330#discussion_r2719475541
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGenUtils.scala:
##########
@@ -43,7 +43,7 @@ import org.apache.flink.table.types.logical.utils.LogicalTypeUtils.toInternalCon
import org.apache.flink.table.types.utils.DataTypeUtils.isInternal
import org.apache.flink.table.utils.EncodingUtils
import org.apache.flink.types.{ColumnList, Row, RowKind}
-import org.apache.flink.types.variant.Variant
+import org.apache.flink.types.variant.{BinaryVariant, Variant}
Review Comment:
The `BinaryVariant` seems unused and can be removed.
##########
docs/data/sql_functions.yml:
##########
@@ -1167,6 +1167,16 @@ variant:
parser will keep the last occurrence of all fields with the same key, otherwise when
`allowDuplicateKeys` is false it will throw an error. The default value of
`allowDuplicateKeys` is false.
+ - sql: variant '[' INT ']'
+ table: VARIANT.at(INT)
+ description: |
+ If the VARIANT is an ARRAY value, returns a VARIANT whose value is the element at
+ the specified index. Otherwise, NULL is returned.
+ - sql: variant '[' STRING ']'
Review Comment:
Apart from `variant ‘[’ INT ‘]’` and `variant ‘[’ STRING ‘]’`, FLIP-521 also mentioned `variant.field` as a built-in function to access field or element in the Variant. I'm also OK if we wants to add support for this function later in a separated PR.
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ItemAtIndexArgumentTypeStrategy.java:
##########
@@ -86,12 +89,36 @@ public Optional<DataType> inferArgumentType(
}
}
+ if (collectionType.is(LogicalTypeRoot.VARIANT)) {
+ if (indexType.getLogicalType().is(LogicalTypeFamily.INTEGER_NUMERIC)) {
+
+ if (callContext.isArgumentLiteral(1)) {
+ Optional<Integer> literalVal = callContext.getArgumentValue(1, Integer.class);
+ if (literalVal.isPresent() && literalVal.get() <= 0) {
+ return callContext.fail(
+ throwOnFailure,
+ "The provided index must be a valid SQL index starting from 1, but was '%s'",
+ literalVal.get());
+ }
+ }
+
+ return Optional.of(indexType);
+ } else if (indexType.getLogicalType().is(LogicalTypeFamily.CHARACTER_STRING)) {
Review Comment:
Would the following be better in aspect of extensibility?
```java
if (LogicalTypeCasts.supportsImplicitCast(indexType.getLogicalType(), new VarCharType())) {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #27330:
URL: https://github.com/apache/flink/pull/27330#discussion_r2603559997
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/VariantSemanticTest.java:
##########
@@ -253,6 +256,185 @@ public class VariantSemanticTest extends SemanticTestBase {
.runSql("INSERT INTO sink_t SELECT k, SUM(v) AS total FROM t GROUP BY k")
.build();
+ static final TableTestProgram VARIANT_ARRAY_ACCESS_SQL;
+
+ static final TableTestProgram VARIANT_ARRAY_ACCESS_TABLE_API;
+
+ static final TableTestProgram VARIANT_OBJECT_ACCESS_SQL;
+
+ static final TableTestProgram VARIANT_OBJECT_ACCESS_TABLE_API;
+
+ static final TableTestProgram VARIANT_NESTED_ACCESS_SQL;
+
+ static final TableTestProgram VARIANT_NESTED_ACCESS_TABLE_API;
+
+ public static final SourceTestStep VARIANT_ARRAY_SOURCE =
Review Comment:
could we have some error tests please. I am thinking.
- supply an integer to a map
- supply a string to an array
- supply an invalid type.
- make sure the error message makes sense for nested failures.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on code in PR #27330:
URL: https://github.com/apache/flink/pull/27330#discussion_r2726271959
##########
flink-table/flink-sql-parser/src/main/java/org/apache/calcite/sql/fun/SqlItemOperator.java:
##########
@@ -0,0 +1,218 @@
+/*
+ * 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.calcite.sql.fun;
+
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlCallBinding;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlOperandCountRange;
+import org.apache.calcite.sql.SqlOperatorBinding;
+import org.apache.calcite.sql.SqlSpecialOperator;
+import org.apache.calcite.sql.SqlWriter;
+import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.calcite.sql.type.OperandTypes;
+import org.apache.calcite.sql.type.SqlOperandCountRanges;
+import org.apache.calcite.sql.type.SqlSingleOperandTypeChecker;
+import org.apache.calcite.sql.type.SqlTypeFamily;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.calcite.sql.type.SqlTypeUtil;
+
+import java.util.Arrays;
+
+import static java.util.Objects.requireNonNull;
+import static org.apache.calcite.sql.type.NonNullableAccessors.getComponentTypeOrThrow;
+import static org.apache.calcite.sql.validate.SqlNonNullableAccessors.getOperandLiteralValueOrThrow;
+
+/**
+ * The item operator {@code [ ... ]}, used to access a given element of an array, map or struct. For
+ * example, {@code myArray[3]}, {@code "myMap['foo']"}, {@code myStruct[2]} or {@code
+ * myStruct['fieldName']}.
+ *
+ * <p>This class was copied over from Calcite 1.39.0 version to support access variant
+ * (FLINK-37924).
+ *
+ * <p>Line 148, CALCITE-7325, should be removed after upgrading Calcite to 1.42.0.
+ */
+public class SqlItemOperator extends SqlSpecialOperator {
+ public final int offset;
+ public final boolean safe;
+
+ public SqlItemOperator(
+ String name, SqlSingleOperandTypeChecker operandTypeChecker, int offset, boolean safe) {
+ super(name, SqlKind.ITEM, 100, true, null, null, operandTypeChecker);
+ this.offset = offset;
+ this.safe = safe;
+ }
+
+ @Override
+ public ReduceResult reduceExpr(int ordinal, TokenSequence list) {
+ SqlNode left = list.node(ordinal - 1);
+ SqlNode right = list.node(ordinal + 1);
+ return new ReduceResult(
+ ordinal - 1,
+ ordinal + 2,
+ createCall(
+ SqlParserPos.sum(
+ Arrays.asList(
+ requireNonNull(left, "left").getParserPosition(),
+ requireNonNull(right, "right").getParserPosition(),
+ list.pos(ordinal))),
+ left,
+ right));
+ }
+
+ @Override
+ public void unparse(SqlWriter writer, SqlCall call, int leftPrec, int rightPrec) {
+ call.operand(0).unparse(writer, leftPrec, 0);
+ final SqlWriter.Frame frame = writer.startList("[", "]");
+ if (!this.getName().equals("ITEM")) {
+ final SqlWriter.Frame offsetFrame = writer.startFunCall(this.getName());
+ call.operand(1).unparse(writer, 0, 0);
+ writer.endFunCall(offsetFrame);
+ } else {
+ call.operand(1).unparse(writer, 0, 0);
+ }
+ writer.endList(frame);
+ }
+
+ @Override
+ public SqlOperandCountRange getOperandCountRange() {
+ return SqlOperandCountRanges.of(2);
+ }
+
+ @Override
+ public boolean checkOperandTypes(SqlCallBinding callBinding, boolean throwOnFailure) {
+ final SqlNode left = callBinding.operand(0);
+ final SqlNode right = callBinding.operand(1);
+ if (!getOperandTypeChecker().checkSingleOperandType(callBinding, left, 0, throwOnFailure)) {
+ return false;
+ }
+ final SqlSingleOperandTypeChecker checker = getChecker(callBinding);
+ return checker.checkSingleOperandType(callBinding, right, 0, throwOnFailure);
+ }
+
+ @Override
+ public SqlSingleOperandTypeChecker getOperandTypeChecker() {
+ return (SqlSingleOperandTypeChecker)
+ requireNonNull(super.getOperandTypeChecker(), "operandTypeChecker");
+ }
+
+ private static SqlSingleOperandTypeChecker getChecker(SqlCallBinding callBinding) {
+ final RelDataType operandType = callBinding.getOperandType(0);
+ switch (operandType.getSqlTypeName()) {
+ case ARRAY:
+ return OperandTypes.family(SqlTypeFamily.INTEGER);
+ case MAP:
+ RelDataType keyType =
+ requireNonNull(operandType.getKeyType(), "operandType.getKeyType()");
+ SqlTypeName sqlTypeName = keyType.getSqlTypeName();
+ return OperandTypes.family(
+ requireNonNull(
+ sqlTypeName.getFamily(),
+ () ->
+ "keyType.getSqlTypeName().getFamily() null, type is "
+ + sqlTypeName));
+ case ROW:
+ case ANY:
+ case DYNAMIC_STAR:
+ case VARIANT:
+ return OperandTypes.family(SqlTypeFamily.INTEGER)
+ .or(OperandTypes.family(SqlTypeFamily.CHARACTER));
+ default:
+ throw callBinding.newValidationSignatureError();
+ }
+ }
+
+ @Override
+ public String getAllowedSignatures(String name) {
+ if (name.equals("ITEM")) {
+ return "<ARRAY>[<INTEGER>]\n"
Review Comment:
nit replace it with the following and update the changed line number scope in class annotation.
```
// FLINK MODIFICATION BEGIN
return "<ARRAY>[<INTEGER>]\n"
+ "<MAP>[<ANY>]\n"
+ "<ROW>[<CHARACTER>|<INTEGER>]\n"
+ "<VARIANT>[<CHARACTER>|<INTEGER>]";
// FLINK MODIFICATION END
```
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ItemAtIndexArgumentTypeStrategy.java:
##########
@@ -41,7 +41,10 @@
* or {@link LogicalTypeRoot#MULTISET}
*
* <p>the type to be equal to the key type of {@link LogicalTypeRoot#MAP} if the first argument is a
- * map.
+ * map
+ *
+ * <p>a {@link LogicalTypeFamily#NUMERIC} type or {LogicalTypeFamily.CHARACTER_STRING} type if the
Review Comment:
nit `{@link LogicalTypeFamily#CHARACTER_STRING}`
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala:
##########
@@ -1668,6 +1668,100 @@ object ScalarOperatorGens {
generateUnaryOperatorIfNotNull(ctx, resultType, map)(_ => s"${map.resultTerm}.size()")
}
+ def generateVariantGet(
+ ctx: CodeGeneratorContext,
+ variant: GeneratedExpression,
+ key: GeneratedExpression): GeneratedExpression = {
+ val Seq(resultTerm, nullTerm) = newNames(ctx, "result", "isNull")
+ val tmpValue = newName(ctx, "tmpValue")
+
+ val variantType = variant.resultType.asInstanceOf[VariantType]
+ val variantTerm = variant.resultTerm
+ val variantTypeTerm = primitiveTypeTermForType(variantType)
+ val variantDefaultTerm = primitiveDefaultValue(variantType)
+
+ val keyTerm = key.resultTerm
+ val keyType = key.resultType
+
+ val accessCode = if (isInteger(keyType)) {
Review Comment:
Should we align this with `indexType.getLogicalType().is(LogicalTypeFamily.INTEGER_NUMERIC)` in `ItemAtIndexArgumentTypeStrategy`? Some `INTEGER_NUMERIC` types like smallint and bigint will pass the check in type strategy but fail here.
##########
docs/data/sql_functions.yml:
##########
@@ -1167,6 +1167,16 @@ variant:
parser will keep the last occurrence of all fields with the same key, otherwise when
`allowDuplicateKeys` is false it will throw an error. The default value of
`allowDuplicateKeys` is false.
+ - sql: variant '[' INT ']'
+ table: VARIANT.at(INT)
+ description: |
+ If the VARIANT is an ARRAY value, returns a VARIANT whose value is the element at
Review Comment:
I think we should also mention the index starts from 1 like `array '[' INT ']'` because most developers might think they're starting from 0.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #27330:
URL: https://github.com/apache/flink/pull/27330#discussion_r2733070390
##########
docs/data/sql_functions.yml:
##########
@@ -1167,6 +1167,16 @@ variant:
parser will keep the last occurrence of all fields with the same key, otherwise when
`allowDuplicateKeys` is false it will throw an error. The default value of
`allowDuplicateKeys` is false.
+ - sql: variant '[' INT ']'
+ table: VARIANT.at(INT)
+ description: |
+ If the VARIANT is an ARRAY value, returns a VARIANT whose value is the element at
+ the specified index. The index starts from 1, If the index is out of range, it returns NULL.
+ - sql: variant '[' STRING ']'
+ table: VARIANT.at(STRING)
+ description: |
+ If the VARIANT is a MAP value that has an element with this key, a VARIANT holding
+ the associated value is returned. Otherwise, NULL is returned.
Review Comment:
We do not support null values in the current design. When the key is valid, the variant always returns another variant.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #27330:
URL: https://github.com/apache/flink/pull/27330#discussion_r2605242708
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ItemAtIndexArgumentTypeStrategy.java:
##########
@@ -86,12 +89,36 @@ public Optional<DataType> inferArgumentType(
}
}
+ if (collectionType.is(LogicalTypeRoot.VARIANT)) {
+ if (indexType.getLogicalType().is(LogicalTypeFamily.INTEGER_NUMERIC)) {
+
+ if (callContext.isArgumentLiteral(1)) {
+ Optional<Integer> literalVal = callContext.getArgumentValue(1, Integer.class);
+ if (literalVal.isPresent() && literalVal.get() <= 0) {
+ return callContext.fail(
+ throwOnFailure,
+ "The provided index must be a valid SQL index starting from 1, but was '%s'",
+ literalVal.get());
+ }
+ }
+
+ return Optional.of(indexType);
+ } else if (indexType.getLogicalType().is(LogicalTypeFamily.CHARACTER_STRING)) {
+ return Optional.of(indexType);
+ } else {
+ return callContext.fail(
+ throwOnFailure,
+ "Variant values can only be accessed with a CHARACTER STRING key or indexed with an INTEGER NUMERIC index.");
+ }
+ }
+
return Optional.empty();
}
@Override
public Signature.Argument getExpectedArgument(
FunctionDefinition functionDefinition, int argumentPos) {
- return Signature.Argument.of("[<INTEGER NUMERIC> | <MAP_KEY_TYPE>]");
+ return Signature.Argument.of(
+ "[<INTEGER NUMERIC> | <MAP_KEY_TYPE> | <CHARACTER STRING> or <INTEGER NUMERIC>]");
Review Comment:
My original intent was to clearly illustrate the one-to-one correspondence between logical root types and their allowed index/key argument types, for example:
- `LogicalTypeRoot.ARRAY` → `<INTEGER NUMERIC>`
- `LogicalTypeRoot.MAP` → `<MAP_KEY_TYPE>`
- `LogicalTypeRoot.VARIANT` → `<CHARACTER STRING>` or `<INTEGER NUMERIC>`
The expected function signatures in old code are:
```
at([<ARRAY> | <MAP> | <VARIANT>], [<INTEGER NUMERIC> | <MAP_KEY_TYPE> | <CHARACTER STRING> or <INTEGER NUMERIC>])
```
However, I have realized that this mapping is already effectively conveyed through the error messages generated by `inferArgumentType()`. I have updated the implementation to reflect this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #27330:
URL: https://github.com/apache/flink/pull/27330#discussion_r2702371143
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ItemAtTypeStrategy.java:
##########
@@ -33,27 +34,29 @@
/**
* An output type strategy for {@link BuiltInFunctionDefinitions#AT}.
*
- * <p>Returns either the element of an {@link LogicalTypeFamily#COLLECTION} type or the value of
- * {@link LogicalTypeRoot#MAP}.
+ * <p>Returns either the element of an {@link LogicalTypeFamily#COLLECTION} type, the value of
+ * {@link LogicalTypeRoot#MAP}, or the variant itself for {@link LogicalTypeRoot#VARIANT}.
Review Comment:
I’ve updated the comment.
Could you please check whether the current wording is accurate?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #27330:
URL: https://github.com/apache/flink/pull/27330#discussion_r2702341234
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala:
##########
@@ -743,7 +743,11 @@ class ExprCodeGenerator(ctx: CodeGeneratorContext, nullableInput: Boolean)
case LogicalTypeRoot.ROW | LogicalTypeRoot.STRUCTURED_TYPE =>
generateDot(ctx, operands)
- case _ => throw new CodeGenException("Expect an array, a map or a row.")
+ case LogicalTypeRoot.VARIANT =>
+ val key = operands(1)
+ generateVariantGet(ctx, operands.head, key)
Review Comment:
Thanks for the suggestion. Handling nested access in a single codegen pass would likely require some special-case logic in upper-level call sites, which feels a bit hacky and harder to maintain.
For shallow cases, users can use ITEM. For deeper accesses, we plan to provide a dedicated variantQuery facility (FLIP-521) for more efficient multi-level extraction:
https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=349637099#FLIP521:IntegratingVariantTypeintoFlink:EnablingEfficientSemiStructuredDataProcessing-Introducebuilt-infunctionsVARIANT_VALUEandVARIANT_QUERYtoextractvaluesfromavariant
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #27330:
URL: https://github.com/apache/flink/pull/27330#discussion_r2721286165
##########
docs/data/sql_functions.yml:
##########
@@ -1167,6 +1167,16 @@ variant:
parser will keep the last occurrence of all fields with the same key, otherwise when
`allowDuplicateKeys` is false it will throw an error. The default value of
`allowDuplicateKeys` is false.
+ - sql: variant '[' INT ']'
+ table: VARIANT.at(INT)
+ description: |
+ If the VARIANT is an ARRAY value, returns a VARIANT whose value is the element at
+ the specified index. Otherwise, NULL is returned.
+ - sql: variant '[' STRING ']'
Review Comment:
In Flink and other SQL-capable projects (e.g. `Calcite` or `Spark`), `.field` is typically meant for accessing fields of structured types whose field names are static and known during analysis/compile time (e.g. `ROW` in Flink). Therefore I’m still evaluating whether we should implement this for `VARIANT`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #27330:
URL: https://github.com/apache/flink/pull/27330#discussion_r2732086234
##########
docs/data/sql_functions.yml:
##########
@@ -1167,6 +1167,16 @@ variant:
parser will keep the last occurrence of all fields with the same key, otherwise when
`allowDuplicateKeys` is false it will throw an error. The default value of
`allowDuplicateKeys` is false.
+ - sql: variant '[' INT ']'
+ table: VARIANT.at(INT)
+ description: |
+ If the VARIANT is an ARRAY value, returns a VARIANT whose value is the element at
+ the specified index. The index starts from 1, If the index is out of range, it returns NULL.
+ - sql: variant '[' STRING ']'
+ table: VARIANT.at(STRING)
+ description: |
+ If the VARIANT is a MAP value that has an element with this key, a VARIANT holding
Review Comment:
I am curious if there are duplicate keys I assume one (the last maybe) will be used here. If would be worth extending the words to be explicit about this scenario.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on code in PR #27330:
URL: https://github.com/apache/flink/pull/27330#discussion_r2744516860
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala:
##########
@@ -1668,6 +1668,100 @@ object ScalarOperatorGens {
generateUnaryOperatorIfNotNull(ctx, resultType, map)(_ => s"${map.resultTerm}.size()")
}
+ def generateVariantGet(
+ ctx: CodeGeneratorContext,
+ variant: GeneratedExpression,
+ key: GeneratedExpression): GeneratedExpression = {
+ val Seq(resultTerm, nullTerm) = newNames(ctx, "result", "isNull")
+ val tmpValue = newName(ctx, "tmpValue")
+
+ val variantType = variant.resultType.asInstanceOf[VariantType]
+ val variantTerm = variant.resultTerm
+ val variantTypeTerm = primitiveTypeTermForType(variantType)
+ val variantDefaultTerm = primitiveDefaultValue(variantType)
+
+ val keyTerm = key.resultTerm
+ val keyType = key.resultType
+
+ val accessCode = if (isIntegerNumeric(keyType)) {
+ generateIntegerKeyAccess(
+ variantTerm,
+ variantTypeTerm,
+ keyTerm,
+ resultTerm,
+ nullTerm,
+ tmpValue
+ )
+ } else if (isCharacterString(keyType)) {
+ val fieldName = key.literalValue.get.toString
+ generateCharacterStringKeyAccess(
+ variantTerm,
+ variantTypeTerm,
+ fieldName,
+ resultTerm,
+ nullTerm,
+ tmpValue
+ )
+ } else {
+ throw new CodeGenException(s"Unsupported key type for variant: $keyType")
+ }
+
+ val finalCode =
+ s"""
+ |${variant.code}
+ |${key.code}
+ |boolean $nullTerm = (${variant.nullTerm} || ${key.nullTerm});
+ |$variantTypeTerm $resultTerm = $variantDefaultTerm;
+ |if (!$nullTerm) {
+ | $accessCode
+ |}
+ """.stripMargin
+
+ GeneratedExpression(resultTerm, nullTerm, finalCode, variantType)
+ }
+
+ private def generateCharacterStringKeyAccess(
+ variantTerm: String,
+ variantTypeTerm: String,
+ fieldName: String,
+ resultTerm: String,
+ nullTerm: String,
+ tmpValue: String): String = {
+ s"""
+ | if ($variantTerm.isObject()){
+ | $variantTypeTerm $tmpValue = $variantTerm.getField("$fieldName");
+ | if ($tmpValue == null) {
+ | $nullTerm = true;
+ | } else {
+ | $resultTerm = $tmpValue;
+ | }
+ | } else {
+ | throw new org.apache.flink.table.api.TableRuntimeException("String key access on variant requires an object variant, but a non-object variant was provided.");
+ | }
+ """.stripMargin
+ }
+
+ private def generateIntegerKeyAccess(
+ variantTerm: String,
+ variantTypeTerm: String,
+ keyTerm: String,
+ resultTerm: String,
+ nullTerm: String,
+ tmpValue: String): String = {
+ s"""
+ | if ($variantTerm.isArray()){
+ | $variantTypeTerm $tmpValue = $variantTerm.getElement((int) $keyTerm - 1);
Review Comment:
Nit what about adding tests for other integer numeric? For example
```
VARIANT_ARRAY_ACCESS_WITH_DIFFERENT_INDEX_TYPES =
TableTestProgram.of(
"variant-array-access-with-different-index-types",
"validates variant array access with different index types using [] operator in sql and at() in table api")
.setupTableSource(VARIANT_ARRAY_SOURCE)
.setupTableSink(VARIANT_ARRAY_SINK)
.runSql(
"INSERT INTO sink_t SELECT v[cast(1 as tinyint)], v[cast(2 as smallint)], v[cast(4 as bigint)] FROM t")
.runTableApi(
t ->
t.from("t")
.select(
$("v").at((byte) 1).as("v1"),
$("v").at((short) 2).as("v2"),
$("v").at((long) 3).as("v3")),
"sink_t")
.build();
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on PR #27330:
URL: https://github.com/apache/flink/pull/27330#issuecomment-3790471246
Hi @davidradl , thanks for the review! Could you take an another look. Thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #27330:
URL: https://github.com/apache/flink/pull/27330#issuecomment-3633164794
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "efb13bf473db71fa58dad90be7151fdbacb704c7",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "efb13bf473db71fa58dad90be7151fdbacb704c7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* efb13bf473db71fa58dad90be7151fdbacb704c7 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27330:
URL: https://github.com/apache/flink/pull/27330#discussion_r2701066244
##########
flink-table/flink-sql-parser/src/main/java/org/apache/calcite/sql/fun/SqlItemOperator.java:
##########
@@ -0,0 +1,218 @@
+/*
+ * 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.calcite.sql.fun;
+
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlCallBinding;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlOperandCountRange;
+import org.apache.calcite.sql.SqlOperatorBinding;
+import org.apache.calcite.sql.SqlSpecialOperator;
+import org.apache.calcite.sql.SqlWriter;
+import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.calcite.sql.type.OperandTypes;
+import org.apache.calcite.sql.type.SqlOperandCountRanges;
+import org.apache.calcite.sql.type.SqlSingleOperandTypeChecker;
+import org.apache.calcite.sql.type.SqlTypeFamily;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.calcite.sql.type.SqlTypeUtil;
+
+import java.util.Arrays;
+
+import static java.util.Objects.requireNonNull;
+import static org.apache.calcite.sql.type.NonNullableAccessors.getComponentTypeOrThrow;
+import static org.apache.calcite.sql.validate.SqlNonNullableAccessors.getOperandLiteralValueOrThrow;
+
+/**
+ * The item operator {@code [ ... ]}, used to access a given element of an array, map or struct. For
+ * example, {@code myArray[3]}, {@code "myMap['foo']"}, {@code myStruct[2]} or {@code
+ * myStruct['fieldName']}.
+ *
+ * <p>This class was copied over from Calcite 1.39.0 version to support access variant
+ * (FLINK-37924).
+ *
+ * <p>Line 148, CALCITE-7325, should be removed after upgrading Calcite to 1.42.0.
Review Comment:
Bumping Calcite is not an easy task unfortunately (giving the fact that Flink very aggressively customizes it), so I wouldn't expect we upgrade to 1.42 fast
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "yunfengzhou-hub (via GitHub)" <gi...@apache.org>.
yunfengzhou-hub commented on code in PR #27330:
URL: https://github.com/apache/flink/pull/27330#discussion_r2671378203
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ItemAtTypeStrategy.java:
##########
@@ -33,27 +34,29 @@
/**
* An output type strategy for {@link BuiltInFunctionDefinitions#AT}.
*
- * <p>Returns either the element of an {@link LogicalTypeFamily#COLLECTION} type or the value of
- * {@link LogicalTypeRoot#MAP}.
+ * <p>Returns either the element of an {@link LogicalTypeFamily#COLLECTION} type, the value of
+ * {@link LogicalTypeRoot#MAP}, or the variant itself for {@link LogicalTypeRoot#VARIANT}.
Review Comment:
The returned type might not equal to the original variant in aspect of nullability.
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala:
##########
@@ -743,7 +743,11 @@ class ExprCodeGenerator(ctx: CodeGeneratorContext, nullableInput: Boolean)
case LogicalTypeRoot.ROW | LogicalTypeRoot.STRUCTURED_TYPE =>
generateDot(ctx, operands)
- case _ => throw new CodeGenException("Expect an array, a map or a row.")
+ case LogicalTypeRoot.VARIANT =>
+ val key = operands(1)
+ generateVariantGet(ctx, operands.head, key)
Review Comment:
Would it be better to directly support nested field access in one codegen invocation? Currently this method only resolves the uppermost level of expressions like `v['k0']['kk1'][1]`, and leaves the rest levels to the next ExprCodeGenerator invocation.
##########
flink-table/flink-sql-parser/src/main/java/org/apache/calcite/sql/fun/SqlItemOperator.java:
##########
@@ -0,0 +1,218 @@
+/*
+ * 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.calcite.sql.fun;
+
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlCallBinding;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlOperandCountRange;
+import org.apache.calcite.sql.SqlOperatorBinding;
+import org.apache.calcite.sql.SqlSpecialOperator;
+import org.apache.calcite.sql.SqlWriter;
+import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.calcite.sql.type.OperandTypes;
+import org.apache.calcite.sql.type.SqlOperandCountRanges;
+import org.apache.calcite.sql.type.SqlSingleOperandTypeChecker;
+import org.apache.calcite.sql.type.SqlTypeFamily;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.calcite.sql.type.SqlTypeUtil;
+
+import java.util.Arrays;
+
+import static java.util.Objects.requireNonNull;
+import static org.apache.calcite.sql.type.NonNullableAccessors.getComponentTypeOrThrow;
+import static org.apache.calcite.sql.validate.SqlNonNullableAccessors.getOperandLiteralValueOrThrow;
+
+/**
+ * The item operator {@code [ ... ]}, used to access a given element of an array, map or struct. For
+ * example, {@code myArray[3]}, {@code "myMap['foo']"}, {@code myStruct[2]} or {@code
+ * myStruct['fieldName']}.
+ *
+ * <p>This class was copied over from Calcite 1.39.0 version to support access variant
+ * (FLINK-37924).
+ *
+ * <p>Line 148, CALCITE-7325, should be removed after upgrading Calcite to 1.42.0.
Review Comment:
Hi @snuyanzin, do you think it would be better to update Calcite version in or before this PR, so that this PR could avoid adding this class? I see that you made the last upgrade of Calcite dependency, and that PR does seem to have been of great workload.
https://github.com/apache/flink/pull/26827
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala:
##########
@@ -1668,6 +1668,108 @@ object ScalarOperatorGens {
generateUnaryOperatorIfNotNull(ctx, resultType, map)(_ => s"${map.resultTerm}.size()")
}
+ def generateVariantGet(
+ ctx: CodeGeneratorContext,
+ variant: GeneratedExpression,
+ key: GeneratedExpression): GeneratedExpression = {
+ val Seq(resultTerm, nullTerm) = newNames(ctx, "result", "isNull")
+ val binaryVariantTerm = newName(ctx, "binaryVariantTerm")
+
+ val variantType = variant.resultType.asInstanceOf[VariantType]
+ val keyType = key.resultType
+
+ val variantTypeTerm = primitiveTypeTermForType(variantType)
+ val variantDefault = primitiveDefaultValue(variantType)
+ val variantTerm = variant.resultTerm
+ val keyTerm = key.resultTerm
+
+ val accessCode = if (isInteger(keyType)) {
+ generateIntegerKeyAccess(
+ variantTerm,
+ binaryVariantTerm,
+ keyTerm,
+ resultTerm,
+ nullTerm
+ )
+ } else if (isCharacterString(keyType)) {
+ generateCharacterStringKeyAccess(
+ variantTerm,
+ binaryVariantTerm,
+ keyTerm,
+ resultTerm,
+ nullTerm
+ )
+ } else {
+ throw new CodeGenException(s"Unsupported key type for variant: $keyType")
+ }
+
+ val finalCode =
+ s"""
+ |${variant.code}
+ |${key.code}
+ |boolean $nullTerm = (${variant.nullTerm}|| ${key.nullTerm});
+ |$variantTypeTerm $resultTerm = $variantDefault;
+ |if (!$nullTerm) {
+ | $accessCode
+ |}
+ """.stripMargin
+
+ GeneratedExpression(resultTerm, nullTerm, finalCode, variantType)
+ }
+
+ private def generateCharacterStringKeyAccess(
+ variantTerm: String,
+ binaryVariantTerm: String,
+ keyTerm: String,
+ resultTerm: String,
+ nullTerm: String): String = {
+ s"""
+ | if ($variantTerm.isObject()){
+ | if ($variantTerm instanceof $BINARY_VARIANT) {
+ | $BINARY_VARIANT $binaryVariantTerm = ($BINARY_VARIANT) $variantTerm;
+ |
+ | String keyStr = null;
+ | if ($keyTerm instanceof $BINARY_STRING) {
+ | keyStr = (($BINARY_STRING) $keyTerm).toString();
+ | } else {
+ | keyStr = String.valueOf($keyTerm);
+ | }
+ |
+ | if (keyStr != null) {
+ | $resultTerm = $binaryVariantTerm.getField(keyStr);
+ | } else {
+ | $nullTerm = true;
+ | }
+ | } else {
+ | $nullTerm = true;
+ | }
+ | } else {
+ | throw new org.apache.flink.table.api.TableRuntimeException("String key access on variant requires an object variant, but a non-object variant was provided.");
+ | }
+ """.stripMargin
+ }
+
+ private def generateIntegerKeyAccess(
+ variantTerm: String,
+ binaryVariantTerm: String,
+ keyTerm: String,
+ resultTerm: String,
+ nullTerm: String): String = {
+ s"""
+ | if ($variantTerm.isArray()){
+ | if ($variantTerm instanceof $BINARY_VARIANT) {
+ | $BINARY_VARIANT $binaryVariantTerm = ($BINARY_VARIANT) $variantTerm;
+ | $resultTerm = $binaryVariantTerm.getElement((int)$keyTerm - 1);
Review Comment:
The index should be 0 based, and no `- 1` is needed.
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala:
##########
@@ -1668,6 +1668,108 @@ object ScalarOperatorGens {
generateUnaryOperatorIfNotNull(ctx, resultType, map)(_ => s"${map.resultTerm}.size()")
}
+ def generateVariantGet(
+ ctx: CodeGeneratorContext,
+ variant: GeneratedExpression,
+ key: GeneratedExpression): GeneratedExpression = {
+ val Seq(resultTerm, nullTerm) = newNames(ctx, "result", "isNull")
+ val binaryVariantTerm = newName(ctx, "binaryVariantTerm")
+
+ val variantType = variant.resultType.asInstanceOf[VariantType]
+ val keyType = key.resultType
+
+ val variantTypeTerm = primitiveTypeTermForType(variantType)
+ val variantDefault = primitiveDefaultValue(variantType)
+ val variantTerm = variant.resultTerm
+ val keyTerm = key.resultTerm
+
+ val accessCode = if (isInteger(keyType)) {
+ generateIntegerKeyAccess(
+ variantTerm,
+ binaryVariantTerm,
+ keyTerm,
+ resultTerm,
+ nullTerm
+ )
+ } else if (isCharacterString(keyType)) {
+ generateCharacterStringKeyAccess(
+ variantTerm,
+ binaryVariantTerm,
+ keyTerm,
+ resultTerm,
+ nullTerm
+ )
+ } else {
+ throw new CodeGenException(s"Unsupported key type for variant: $keyType")
+ }
+
+ val finalCode =
+ s"""
+ |${variant.code}
+ |${key.code}
+ |boolean $nullTerm = (${variant.nullTerm}|| ${key.nullTerm});
+ |$variantTypeTerm $resultTerm = $variantDefault;
+ |if (!$nullTerm) {
+ | $accessCode
+ |}
+ """.stripMargin
+
+ GeneratedExpression(resultTerm, nullTerm, finalCode, variantType)
+ }
+
+ private def generateCharacterStringKeyAccess(
+ variantTerm: String,
+ binaryVariantTerm: String,
+ keyTerm: String,
+ resultTerm: String,
+ nullTerm: String): String = {
+ s"""
+ | if ($variantTerm.isObject()){
+ | if ($variantTerm instanceof $BINARY_VARIANT) {
+ | $BINARY_VARIANT $binaryVariantTerm = ($BINARY_VARIANT) $variantTerm;
+ |
+ | String keyStr = null;
+ | if ($keyTerm instanceof $BINARY_STRING) {
Review Comment:
It should be avoided to rely on concrete subclasses like `BinaryStringData` and `BinaryVariant`. Better use `StringData` and `Variant` instead.
Furthermore, code like follows might get you rid of complex toString implementations.
```scala
val fieldName = key.literalValue.get.toString
val inputCode =
s"""
|$nullTerm = $inputTerm.isNull();
|$fieldTerm = null;
|if (!$nullTerm) {
| $fieldTerm = $inputTerm.getField("$fieldName");
|}
""".stripMargin.trim
GeneratedExpression(fieldTerm, nullTerm, inputCode, variantType)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #27330:
URL: https://github.com/apache/flink/pull/27330#discussion_r2732093350
##########
docs/data/sql_functions.yml:
##########
@@ -1167,6 +1167,16 @@ variant:
parser will keep the last occurrence of all fields with the same key, otherwise when
`allowDuplicateKeys` is false it will throw an error. The default value of
`allowDuplicateKeys` is false.
+ - sql: variant '[' INT ']'
+ table: VARIANT.at(INT)
+ description: |
+ If the VARIANT is an ARRAY value, returns a VARIANT whose value is the element at
+ the specified index. The index starts from 1, If the index is out of range, it returns NULL.
+ - sql: variant '[' STRING ']'
+ table: VARIANT.at(STRING)
+ description: |
+ If the VARIANT is a MAP value that has an element with this key, a VARIANT holding
+ the associated value is returned. Otherwise, NULL is returned.
Review Comment:
do we support null values? Do we need to differentiate between an non existent key and an entry that exists with a null value
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on PR #27330:
URL: https://github.com/apache/flink/pull/27330#issuecomment-3772685367
Hi @yunfengzhou-hub, thanks for the review! I have updated the code according to your suggestions. Could you please take another look at this PR when you have time? Thank you very much!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #27330:
URL: https://github.com/apache/flink/pull/27330#discussion_r2603527093
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ItemAtIndexArgumentTypeStrategy.java:
##########
@@ -86,12 +89,36 @@ public Optional<DataType> inferArgumentType(
}
}
+ if (collectionType.is(LogicalTypeRoot.VARIANT)) {
+ if (indexType.getLogicalType().is(LogicalTypeFamily.INTEGER_NUMERIC)) {
+
+ if (callContext.isArgumentLiteral(1)) {
+ Optional<Integer> literalVal = callContext.getArgumentValue(1, Integer.class);
+ if (literalVal.isPresent() && literalVal.get() <= 0) {
+ return callContext.fail(
+ throwOnFailure,
+ "The provided index must be a valid SQL index starting from 1, but was '%s'",
+ literalVal.get());
+ }
+ }
+
+ return Optional.of(indexType);
+ } else if (indexType.getLogicalType().is(LogicalTypeFamily.CHARACTER_STRING)) {
+ return Optional.of(indexType);
+ } else {
+ return callContext.fail(
+ throwOnFailure,
+ "Variant values can only be accessed with a CHARACTER STRING key or indexed with an INTEGER NUMERIC index.");
+ }
+ }
+
return Optional.empty();
}
@Override
public Signature.Argument getExpectedArgument(
FunctionDefinition functionDefinition, int argumentPos) {
- return Signature.Argument.of("[<INTEGER NUMERIC> | <MAP_KEY_TYPE>]");
+ return Signature.Argument.of(
+ "[<INTEGER NUMERIC> | <MAP_KEY_TYPE> | <CHARACTER STRING> or <INTEGER NUMERIC>]");
Review Comment:
Why do we need `<INTEGER NUMERIC>` twice here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #27330:
URL: https://github.com/apache/flink/pull/27330#discussion_r2603548047
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java:
##########
@@ -2442,7 +2442,8 @@ ANY, and(logical(LogicalTypeRoot.BOOLEAN), LITERAL)
sequence(
or(
logical(LogicalTypeRoot.ARRAY),
- logical(LogicalTypeRoot.MAP)),
+ logical(LogicalTypeRoot.MAP),
Review Comment:
do we need to update the docs for this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "fxxowo (via GitHub)" <gi...@apache.org>.
fxxowo commented on code in PR #27330:
URL: https://github.com/apache/flink/pull/27330#discussion_r2702340723
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala:
##########
@@ -743,7 +743,11 @@ class ExprCodeGenerator(ctx: CodeGeneratorContext, nullableInput: Boolean)
case LogicalTypeRoot.ROW | LogicalTypeRoot.STRUCTURED_TYPE =>
generateDot(ctx, operands)
- case _ => throw new CodeGenException("Expect an array, a map or a row.")
+ case LogicalTypeRoot.VARIANT =>
+ val key = operands(1)
+ generateVariantGet(ctx, operands.head, key)
Review Comment:
Thanks for the suggestion. Handling nested access in a single codegen pass would likely require some special-case logic in upper-level call sites, which feels a bit hacky and harder to maintain.
For shallow cases, users can use ITEM. For deeper accesses, we plan to provide a dedicated variantQuery facility (FLIP-521) for more efficient multi-level extraction:
https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=349637099#FLIP521:IntegratingVariantTypeintoFlink:EnablingEfficientSemiStructuredDataProcessing-Introducebuilt-infunctionsVARIANT_VALUEandVARIANT_QUERYtoextractvaluesfromavariant
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala:
##########
@@ -743,7 +743,11 @@ class ExprCodeGenerator(ctx: CodeGeneratorContext, nullableInput: Boolean)
case LogicalTypeRoot.ROW | LogicalTypeRoot.STRUCTURED_TYPE =>
generateDot(ctx, operands)
- case _ => throw new CodeGenException("Expect an array, a map or a row.")
+ case LogicalTypeRoot.VARIANT =>
+ val key = operands(1)
+ generateVariantGet(ctx, operands.head, key)
Review Comment:
Thanks for the suggestion. Handling nested access in a single codegen pass would likely require some special-case logic in upper-level call sites, which feels a bit hacky and harder to maintain.
For shallow cases, users can use ITEM. For deeper accesses, we plan to provide a dedicated variantQuery facility (FLIP-521) for more efficient multi-level extraction:
https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=349637099#FLIP521:IntegratingVariantTypeintoFlink:EnablingEfficientSemiStructuredDataProcessing-Introducebuilt-infunctionsVARIANT_VALUEandVARIANT_QUERYtoextractvaluesfromavariant
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #27330:
URL: https://github.com/apache/flink/pull/27330#discussion_r2721377941
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ItemAtIndexArgumentTypeStrategy.java:
##########
@@ -86,12 +89,36 @@ public Optional<DataType> inferArgumentType(
}
}
+ if (collectionType.is(LogicalTypeRoot.VARIANT)) {
+ if (indexType.getLogicalType().is(LogicalTypeFamily.INTEGER_NUMERIC)) {
+
+ if (callContext.isArgumentLiteral(1)) {
+ Optional<Integer> literalVal = callContext.getArgumentValue(1, Integer.class);
+ if (literalVal.isPresent() && literalVal.get() <= 0) {
+ return callContext.fail(
+ throwOnFailure,
+ "The provided index must be a valid SQL index starting from 1, but was '%s'",
+ literalVal.get());
+ }
+ }
+
+ return Optional.of(indexType);
+ } else if (indexType.getLogicalType().is(LogicalTypeFamily.CHARACTER_STRING)) {
Review Comment:
Thanks for your advice. I think using LogicalType.typeRoot and LogicalTypeFamily is a better way to classify types, because it aligns better with the abstraction level of Flink's type system and is more standard and maintainable.
At the same time, the CHARACTER_STRING family can cover string types such as CHAR and VARCHAR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #27330:
URL: https://github.com/apache/flink/pull/27330#discussion_r2721377941
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ItemAtIndexArgumentTypeStrategy.java:
##########
@@ -86,12 +89,36 @@ public Optional<DataType> inferArgumentType(
}
}
+ if (collectionType.is(LogicalTypeRoot.VARIANT)) {
+ if (indexType.getLogicalType().is(LogicalTypeFamily.INTEGER_NUMERIC)) {
+
+ if (callContext.isArgumentLiteral(1)) {
+ Optional<Integer> literalVal = callContext.getArgumentValue(1, Integer.class);
+ if (literalVal.isPresent() && literalVal.get() <= 0) {
+ return callContext.fail(
+ throwOnFailure,
+ "The provided index must be a valid SQL index starting from 1, but was '%s'",
+ literalVal.get());
+ }
+ }
+
+ return Optional.of(indexType);
+ } else if (indexType.getLogicalType().is(LogicalTypeFamily.CHARACTER_STRING)) {
Review Comment:
Thanks for your advice. I think using LogicalType.typeRoot and LogicalTypeFamily is a better way to classify types, because it aligns better with the abstraction level of Flink\’s type system and is more standard and maintainable.
At the same time, the CHARACTER_STRING family can cover string types such as CHAR and VARCHAR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #27330:
URL: https://github.com/apache/flink/pull/27330#discussion_r2744563056
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala:
##########
@@ -1668,6 +1668,100 @@ object ScalarOperatorGens {
generateUnaryOperatorIfNotNull(ctx, resultType, map)(_ => s"${map.resultTerm}.size()")
}
+ def generateVariantGet(
+ ctx: CodeGeneratorContext,
+ variant: GeneratedExpression,
+ key: GeneratedExpression): GeneratedExpression = {
+ val Seq(resultTerm, nullTerm) = newNames(ctx, "result", "isNull")
+ val tmpValue = newName(ctx, "tmpValue")
+
+ val variantType = variant.resultType.asInstanceOf[VariantType]
+ val variantTerm = variant.resultTerm
+ val variantTypeTerm = primitiveTypeTermForType(variantType)
+ val variantDefaultTerm = primitiveDefaultValue(variantType)
+
+ val keyTerm = key.resultTerm
+ val keyType = key.resultType
+
+ val accessCode = if (isIntegerNumeric(keyType)) {
+ generateIntegerKeyAccess(
+ variantTerm,
+ variantTypeTerm,
+ keyTerm,
+ resultTerm,
+ nullTerm,
+ tmpValue
+ )
+ } else if (isCharacterString(keyType)) {
+ val fieldName = key.literalValue.get.toString
+ generateCharacterStringKeyAccess(
+ variantTerm,
+ variantTypeTerm,
+ fieldName,
+ resultTerm,
+ nullTerm,
+ tmpValue
+ )
+ } else {
+ throw new CodeGenException(s"Unsupported key type for variant: $keyType")
+ }
+
+ val finalCode =
+ s"""
+ |${variant.code}
+ |${key.code}
+ |boolean $nullTerm = (${variant.nullTerm} || ${key.nullTerm});
+ |$variantTypeTerm $resultTerm = $variantDefaultTerm;
+ |if (!$nullTerm) {
+ | $accessCode
+ |}
+ """.stripMargin
+
+ GeneratedExpression(resultTerm, nullTerm, finalCode, variantType)
+ }
+
+ private def generateCharacterStringKeyAccess(
+ variantTerm: String,
+ variantTypeTerm: String,
+ fieldName: String,
+ resultTerm: String,
+ nullTerm: String,
+ tmpValue: String): String = {
+ s"""
+ | if ($variantTerm.isObject()){
+ | $variantTypeTerm $tmpValue = $variantTerm.getField("$fieldName");
+ | if ($tmpValue == null) {
+ | $nullTerm = true;
+ | } else {
+ | $resultTerm = $tmpValue;
+ | }
+ | } else {
+ | throw new org.apache.flink.table.api.TableRuntimeException("String key access on variant requires an object variant, but a non-object variant was provided.");
+ | }
+ """.stripMargin
+ }
+
+ private def generateIntegerKeyAccess(
+ variantTerm: String,
+ variantTypeTerm: String,
+ keyTerm: String,
+ resultTerm: String,
+ nullTerm: String,
+ tmpValue: String): String = {
+ s"""
+ | if ($variantTerm.isArray()){
+ | $variantTypeTerm $tmpValue = $variantTerm.getElement((int) $keyTerm - 1);
Review Comment:
Good suggestion! I have finished it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #27330:
URL: https://github.com/apache/flink/pull/27330#discussion_r2607296455
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/VariantSemanticTest.java:
##########
@@ -253,6 +256,185 @@ public class VariantSemanticTest extends SemanticTestBase {
.runSql("INSERT INTO sink_t SELECT k, SUM(v) AS total FROM t GROUP BY k")
.build();
+ static final TableTestProgram VARIANT_ARRAY_ACCESS_SQL;
+
+ static final TableTestProgram VARIANT_ARRAY_ACCESS_TABLE_API;
+
+ static final TableTestProgram VARIANT_OBJECT_ACCESS_SQL;
+
+ static final TableTestProgram VARIANT_OBJECT_ACCESS_TABLE_API;
+
+ static final TableTestProgram VARIANT_NESTED_ACCESS_SQL;
+
+ static final TableTestProgram VARIANT_NESTED_ACCESS_TABLE_API;
+
+ public static final SourceTestStep VARIANT_ARRAY_SOURCE =
Review Comment:
Done~
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #27330:
URL: https://github.com/apache/flink/pull/27330#discussion_r2605246979
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ItemAtIndexArgumentTypeStrategy.java:
##########
@@ -86,12 +89,36 @@ public Optional<DataType> inferArgumentType(
}
}
+ if (collectionType.is(LogicalTypeRoot.VARIANT)) {
+ if (indexType.getLogicalType().is(LogicalTypeFamily.INTEGER_NUMERIC)) {
+
+ if (callContext.isArgumentLiteral(1)) {
+ Optional<Integer> literalVal = callContext.getArgumentValue(1, Integer.class);
+ if (literalVal.isPresent() && literalVal.get() <= 0) {
+ return callContext.fail(
+ throwOnFailure,
+ "The provided index must be a valid SQL index starting from 1, but was '%s'",
+ literalVal.get());
+ }
+ }
+
+ return Optional.of(indexType);
+ } else if (indexType.getLogicalType().is(LogicalTypeFamily.CHARACTER_STRING)) {
+ return Optional.of(indexType);
+ } else {
+ return callContext.fail(
+ throwOnFailure,
+ "Variant values can only be accessed with a CHARACTER STRING key or indexed with an INTEGER NUMERIC index.");
Review Comment:
Yeah, this makes the error clearer for users. Done, thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #27330:
URL: https://github.com/apache/flink/pull/27330#discussion_r2603545971
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ItemAtIndexArgumentTypeStrategy.java:
##########
@@ -86,12 +89,36 @@ public Optional<DataType> inferArgumentType(
}
}
+ if (collectionType.is(LogicalTypeRoot.VARIANT)) {
+ if (indexType.getLogicalType().is(LogicalTypeFamily.INTEGER_NUMERIC)) {
+
+ if (callContext.isArgumentLiteral(1)) {
+ Optional<Integer> literalVal = callContext.getArgumentValue(1, Integer.class);
+ if (literalVal.isPresent() && literalVal.get() <= 0) {
+ return callContext.fail(
+ throwOnFailure,
+ "The provided index must be a valid SQL index starting from 1, but was '%s'",
+ literalVal.get());
+ }
+ }
+
+ return Optional.of(indexType);
+ } else if (indexType.getLogicalType().is(LogicalTypeFamily.CHARACTER_STRING)) {
+ return Optional.of(indexType);
+ } else {
+ return callContext.fail(
+ throwOnFailure,
+ "Variant values can only be accessed with a CHARACTER STRING key or indexed with an INTEGER NUMERIC index.");
Review Comment:
can we include what type was given.
nit: wording suggestion - not need to take unless you want to.
Incorrect type {} supplied for the variant value. Variant values can only be accessed with a CHARACTER STRING map key or an INTEGER NUMERIC array index.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #27330:
URL: https://github.com/apache/flink/pull/27330#discussion_r2603559997
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/VariantSemanticTest.java:
##########
@@ -253,6 +256,185 @@ public class VariantSemanticTest extends SemanticTestBase {
.runSql("INSERT INTO sink_t SELECT k, SUM(v) AS total FROM t GROUP BY k")
.build();
+ static final TableTestProgram VARIANT_ARRAY_ACCESS_SQL;
+
+ static final TableTestProgram VARIANT_ARRAY_ACCESS_TABLE_API;
+
+ static final TableTestProgram VARIANT_OBJECT_ACCESS_SQL;
+
+ static final TableTestProgram VARIANT_OBJECT_ACCESS_TABLE_API;
+
+ static final TableTestProgram VARIANT_NESTED_ACCESS_SQL;
+
+ static final TableTestProgram VARIANT_NESTED_ACCESS_TABLE_API;
+
+ public static final SourceTestStep VARIANT_ARRAY_SOURCE =
Review Comment:
could we have some error tests please. I am thinking.
-supply an integer to a map
-supply a string to an array
-supply an invalid type.
-make sure the error message makes sense for nested failures.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #27330:
URL: https://github.com/apache/flink/pull/27330#discussion_r2603559997
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/VariantSemanticTest.java:
##########
@@ -253,6 +256,185 @@ public class VariantSemanticTest extends SemanticTestBase {
.runSql("INSERT INTO sink_t SELECT k, SUM(v) AS total FROM t GROUP BY k")
.build();
+ static final TableTestProgram VARIANT_ARRAY_ACCESS_SQL;
+
+ static final TableTestProgram VARIANT_ARRAY_ACCESS_TABLE_API;
+
+ static final TableTestProgram VARIANT_OBJECT_ACCESS_SQL;
+
+ static final TableTestProgram VARIANT_OBJECT_ACCESS_TABLE_API;
+
+ static final TableTestProgram VARIANT_NESTED_ACCESS_SQL;
+
+ static final TableTestProgram VARIANT_NESTED_ACCESS_TABLE_API;
+
+ public static final SourceTestStep VARIANT_ARRAY_SOURCE =
Review Comment:
could we have some error tests please. I am thinking.
-supply an integer to a map
-supply a string to an array
-supply an invalid type.
- make sure the error message makes sense for nested failures.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on PR #27330:
URL: https://github.com/apache/flink/pull/27330#issuecomment-3685065984
Hello @Sxnan Could you take a look at this PR when you have time?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37924][table]Introduce Built-in Function to Access field or element in the Variant [flink]
Posted by "JinkunLiu (via GitHub)" <gi...@apache.org>.
JinkunLiu commented on code in PR #27330:
URL: https://github.com/apache/flink/pull/27330#discussion_r2702309749
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala:
##########
@@ -1668,6 +1668,108 @@ object ScalarOperatorGens {
generateUnaryOperatorIfNotNull(ctx, resultType, map)(_ => s"${map.resultTerm}.size()")
}
+ def generateVariantGet(
+ ctx: CodeGeneratorContext,
+ variant: GeneratedExpression,
+ key: GeneratedExpression): GeneratedExpression = {
+ val Seq(resultTerm, nullTerm) = newNames(ctx, "result", "isNull")
+ val binaryVariantTerm = newName(ctx, "binaryVariantTerm")
+
+ val variantType = variant.resultType.asInstanceOf[VariantType]
+ val keyType = key.resultType
+
+ val variantTypeTerm = primitiveTypeTermForType(variantType)
+ val variantDefault = primitiveDefaultValue(variantType)
+ val variantTerm = variant.resultTerm
+ val keyTerm = key.resultTerm
+
+ val accessCode = if (isInteger(keyType)) {
+ generateIntegerKeyAccess(
+ variantTerm,
+ binaryVariantTerm,
+ keyTerm,
+ resultTerm,
+ nullTerm
+ )
+ } else if (isCharacterString(keyType)) {
+ generateCharacterStringKeyAccess(
+ variantTerm,
+ binaryVariantTerm,
+ keyTerm,
+ resultTerm,
+ nullTerm
+ )
+ } else {
+ throw new CodeGenException(s"Unsupported key type for variant: $keyType")
+ }
+
+ val finalCode =
+ s"""
+ |${variant.code}
+ |${key.code}
+ |boolean $nullTerm = (${variant.nullTerm}|| ${key.nullTerm});
+ |$variantTypeTerm $resultTerm = $variantDefault;
+ |if (!$nullTerm) {
+ | $accessCode
+ |}
+ """.stripMargin
+
+ GeneratedExpression(resultTerm, nullTerm, finalCode, variantType)
+ }
+
+ private def generateCharacterStringKeyAccess(
+ variantTerm: String,
+ binaryVariantTerm: String,
+ keyTerm: String,
+ resultTerm: String,
+ nullTerm: String): String = {
+ s"""
+ | if ($variantTerm.isObject()){
+ | if ($variantTerm instanceof $BINARY_VARIANT) {
+ | $BINARY_VARIANT $binaryVariantTerm = ($BINARY_VARIANT) $variantTerm;
+ |
+ | String keyStr = null;
+ | if ($keyTerm instanceof $BINARY_STRING) {
Review Comment:
Thank you for the suggestion.
I have replaced `BinaryVariant` with `Variant` and improved the implementation for obtaining fieldName.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org