twalthr commented on a change in pull request #12649:
URL: https://github.com/apache/flink/pull/12649#discussion_r440252362
##########
File path:
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/InputTypeStrategies.java
##########
@@ -198,6 +201,36 @@ public static InputTypeStrategy comparable(
*/
public static final LiteralArgumentTypeStrategy LITERAL_OR_NULL = new
LiteralArgumentTypeStrategy(true);
+ /**
+ * Strategy that checks that the argument has a composite type.
+ */
+ public static final ArgumentTypeStrategy COMPOSITE = new
ArgumentTypeStrategy() {
Review comment:
nit: move to separate class like all other argument strategies
##########
File path:
flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/expressions/CompositeTypeAccessExpressionITCase.java
##########
@@ -0,0 +1,264 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.expressions;
+
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.FunctionHint;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.ScalarFunction;
+import org.apache.flink.table.functions.TableFunction;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Suite;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static java.util.Collections.singletonMap;
+import static org.apache.flink.table.api.DataTypes.ARRAY;
+import static org.apache.flink.table.api.DataTypes.BIGINT;
+import static org.apache.flink.table.api.DataTypes.FIELD;
+import static org.apache.flink.table.api.DataTypes.MAP;
+import static org.apache.flink.table.api.DataTypes.ROW;
+import static org.apache.flink.table.api.DataTypes.STRING;
+import static org.apache.flink.table.api.Expressions.$;
+import static org.apache.flink.table.api.Expressions.call;
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThat;
+
+/**
+ * Tests for functions that access nested fields/elements of
composite/collection types.
+ */
+@RunWith(Suite.class)
[email protected](
+ {
+ CompositeTypeAccessExpressionITCase.TableFieldAccess.class,
+ CompositeTypeAccessExpressionITCase.CallFieldAccess.class
+ }
+)
+public class CompositeTypeAccessExpressionITCase {
+
+ /**
+ * Regular tests. See also {@link CallFieldAccess} for tests that
access a nested field of an expression or
+ * for {@link BuiltInFunctionDefinitions#FLATTEN} which produces
multiple columns from a single one.
+ */
+ public static class TableFieldAccess extends BuiltInFunctionTestBase {
+ @Parameterized.Parameters(name = "{index}: {0}")
+ public static List<TestSpec> testData() {
+ return Arrays.asList(
+
+ // Actually in case of SQL it does not use the
GET method, but
+ // a custom logic for accessing nested fields
of a Table.
+
TestSpec.forFunction(BuiltInFunctionDefinitions.GET)
+ .onFieldsWithData(null, Row.of(1))
+ .andDataTypes(
+ ROW(FIELD("nested",
BIGINT().notNull())).nullable(),
+ ROW(FIELD("nested",
BIGINT().notNull())).notNull()
+ )
+
.testTableApiResult($("f0").get("nested"), null, BIGINT().nullable())
+
.testTableApiResult($("f1").get("nested"), 1L, BIGINT().notNull())
+ .testSqlResult("f0.nested", null,
BIGINT().nullable())
+ .testSqlResult("f1.nested", 1L,
BIGINT().notNull()),
+
+ // In Calcite it maps to
FlinkSqlOperatorTable.ITEM
+
TestSpec.forFunction(BuiltInFunctionDefinitions.AT)
+ .onFieldsWithData(null, new int[] {1},
null, singletonMap("nested", 1), null, Row.of(1))
+ .andDataTypes(
+
ARRAY(BIGINT().notNull()).nullable(),
+
ARRAY(BIGINT().notNull()).notNull(),
+ MAP(STRING(),
BIGINT().notNull()).nullable(),
+ MAP(STRING(),
BIGINT().notNull()).notNull(),
+ ROW(FIELD("nested",
BIGINT().notNull())).nullable(),
+ ROW(FIELD("nested",
BIGINT().notNull())).notNull()
+ )
+ // accessing elements of MAP or ARRAY
is a runtime operations,
+ // we do not know about the size or
contents during the inference
+ // therefore the results are always
nullable
+ .testSqlResult("f0[1]", null,
BIGINT().nullable())
Review comment:
should we port `AT` as well or in a separate PR?
##########
File path:
flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/expressions/CompositeTypeAccessExpressionITCase.java
##########
@@ -0,0 +1,264 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.expressions;
+
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.FunctionHint;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.ScalarFunction;
+import org.apache.flink.table.functions.TableFunction;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Suite;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static java.util.Collections.singletonMap;
+import static org.apache.flink.table.api.DataTypes.ARRAY;
+import static org.apache.flink.table.api.DataTypes.BIGINT;
+import static org.apache.flink.table.api.DataTypes.FIELD;
+import static org.apache.flink.table.api.DataTypes.MAP;
+import static org.apache.flink.table.api.DataTypes.ROW;
+import static org.apache.flink.table.api.DataTypes.STRING;
+import static org.apache.flink.table.api.Expressions.$;
+import static org.apache.flink.table.api.Expressions.call;
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThat;
+
+/**
+ * Tests for functions that access nested fields/elements of
composite/collection types.
+ */
+@RunWith(Suite.class)
[email protected](
+ {
+ CompositeTypeAccessExpressionITCase.TableFieldAccess.class,
+ CompositeTypeAccessExpressionITCase.CallFieldAccess.class
+ }
+)
+public class CompositeTypeAccessExpressionITCase {
+
+ /**
+ * Regular tests. See also {@link CallFieldAccess} for tests that
access a nested field of an expression or
+ * for {@link BuiltInFunctionDefinitions#FLATTEN} which produces
multiple columns from a single one.
+ */
+ public static class TableFieldAccess extends BuiltInFunctionTestBase {
+ @Parameterized.Parameters(name = "{index}: {0}")
+ public static List<TestSpec> testData() {
+ return Arrays.asList(
+
+ // Actually in case of SQL it does not use the
GET method, but
+ // a custom logic for accessing nested fields
of a Table.
+
TestSpec.forFunction(BuiltInFunctionDefinitions.GET)
+ .onFieldsWithData(null, Row.of(1))
+ .andDataTypes(
+ ROW(FIELD("nested",
BIGINT().notNull())).nullable(),
+ ROW(FIELD("nested",
BIGINT().notNull())).notNull()
+ )
+
.testTableApiResult($("f0").get("nested"), null, BIGINT().nullable())
+
.testTableApiResult($("f1").get("nested"), 1L, BIGINT().notNull())
+ .testSqlResult("f0.nested", null,
BIGINT().nullable())
+ .testSqlResult("f1.nested", 1L,
BIGINT().notNull()),
+
+ // In Calcite it maps to
FlinkSqlOperatorTable.ITEM
+
TestSpec.forFunction(BuiltInFunctionDefinitions.AT)
+ .onFieldsWithData(null, new int[] {1},
null, singletonMap("nested", 1), null, Row.of(1))
+ .andDataTypes(
+
ARRAY(BIGINT().notNull()).nullable(),
+
ARRAY(BIGINT().notNull()).notNull(),
+ MAP(STRING(),
BIGINT().notNull()).nullable(),
+ MAP(STRING(),
BIGINT().notNull()).notNull(),
+ ROW(FIELD("nested",
BIGINT().notNull())).nullable(),
+ ROW(FIELD("nested",
BIGINT().notNull())).notNull()
+ )
+ // accessing elements of MAP or ARRAY
is a runtime operations,
+ // we do not know about the size or
contents during the inference
+ // therefore the results are always
nullable
+ .testSqlResult("f0[1]", null,
BIGINT().nullable())
+ .testSqlResult("f1[1]", 1L,
BIGINT().nullable())
+ .testSqlResult("f2['nested']", null,
BIGINT().nullable())
+ .testSqlResult("f3['nested']", 1L,
BIGINT().nullable())
+
+ // we know all the fields of a type up
front, therefore we can
+ // derive more accurate types during
the inference
+ .testSqlResult("f4['nested']", null,
BIGINT().nullable())
+ .testSqlResult("f5['nested']", 1L,
BIGINT().notNull())
+ );
+ }
+ }
+
+ /**
+ * A class for customized tests.
+ */
+ public static class CallFieldAccess {
Review comment:
nit: very good tests in general but can we rename the two suits to
`FieldAccessAfterCall` and `FieldAccessFromTable` or similar to highlight the
difference?
##########
File path:
flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
##########
@@ -972,13 +972,28 @@
new BuiltInFunctionDefinition.Builder()
.name("flatten")
.kind(OTHER)
- .outputTypeStrategy(TypeStrategies.MISSING)
+
.inputTypeStrategy(sequence(InputTypeStrategies.COMPOSITE))
+ .outputTypeStrategy(callContext -> {
+ throw new
UnsupportedOperationException("FLATTEN should be resolved to GET expressions");
+ })
.build();
public static final BuiltInFunctionDefinition GET =
new BuiltInFunctionDefinition.Builder()
.name("get")
.kind(OTHER)
- .outputTypeStrategy(TypeStrategies.MISSING)
+ .inputTypeStrategy(
+ sequence(
+ InputTypeStrategies.COMPOSITE,
+ and(
+ InputTypeStrategies.LITERAL,
+ or(
+
logical(LogicalTypeRoot.INTEGER),
+
logical(LogicalTypeFamily.CHARACTER_STRING, false)
Review comment:
remove `false`? the nullability is already excluded by `LITERAL`
##########
File path:
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/InputTypeStrategies.java
##########
@@ -198,6 +201,36 @@ public static InputTypeStrategy comparable(
*/
public static final LiteralArgumentTypeStrategy LITERAL_OR_NULL = new
LiteralArgumentTypeStrategy(true);
+ /**
+ * Strategy that checks that the argument has a composite type.
+ */
+ public static final ArgumentTypeStrategy COMPOSITE = new
ArgumentTypeStrategy() {
+ @Override
+ public Optional<DataType> inferArgumentType(
+ CallContext callContext,
+ int argumentPos,
+ boolean throwOnFailure) {
+ DataType dataType =
callContext.getArgumentDataTypes().get(argumentPos);
Review comment:
nit: wrong indention
##########
File path:
flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/expressions/CompositeTypeAccessExpressionITCase.java
##########
@@ -0,0 +1,264 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.expressions;
+
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.FunctionHint;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.ScalarFunction;
+import org.apache.flink.table.functions.TableFunction;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Suite;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static java.util.Collections.singletonMap;
+import static org.apache.flink.table.api.DataTypes.ARRAY;
+import static org.apache.flink.table.api.DataTypes.BIGINT;
+import static org.apache.flink.table.api.DataTypes.FIELD;
+import static org.apache.flink.table.api.DataTypes.MAP;
+import static org.apache.flink.table.api.DataTypes.ROW;
+import static org.apache.flink.table.api.DataTypes.STRING;
+import static org.apache.flink.table.api.Expressions.$;
+import static org.apache.flink.table.api.Expressions.call;
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThat;
+
+/**
+ * Tests for functions that access nested fields/elements of
composite/collection types.
+ */
+@RunWith(Suite.class)
[email protected](
+ {
+ CompositeTypeAccessExpressionITCase.TableFieldAccess.class,
+ CompositeTypeAccessExpressionITCase.CallFieldAccess.class
+ }
+)
+public class CompositeTypeAccessExpressionITCase {
+
+ /**
+ * Regular tests. See also {@link CallFieldAccess} for tests that
access a nested field of an expression or
+ * for {@link BuiltInFunctionDefinitions#FLATTEN} which produces
multiple columns from a single one.
+ */
+ public static class TableFieldAccess extends BuiltInFunctionTestBase {
+ @Parameterized.Parameters(name = "{index}: {0}")
+ public static List<TestSpec> testData() {
+ return Arrays.asList(
+
+ // Actually in case of SQL it does not use the
GET method, but
+ // a custom logic for accessing nested fields
of a Table.
+
TestSpec.forFunction(BuiltInFunctionDefinitions.GET)
+ .onFieldsWithData(null, Row.of(1))
+ .andDataTypes(
+ ROW(FIELD("nested",
BIGINT().notNull())).nullable(),
+ ROW(FIELD("nested",
BIGINT().notNull())).notNull()
+ )
+
.testTableApiResult($("f0").get("nested"), null, BIGINT().nullable())
Review comment:
use the newly introduced `testResult` such that a result must be defined
only once
##########
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala
##########
@@ -724,6 +724,9 @@ class ExprCodeGenerator(ctx: CodeGeneratorContext,
nullableInput: Boolean)
val key = operands(1)
generateMapGet(ctx, operands.head, key)
+ case t: LogicalType if TypeCheckUtils.isRow(t) =>
Review comment:
use `LogicalTypeChecks` instead
##########
File path:
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/TypeStrategies.java
##########
@@ -315,6 +317,36 @@ public static TypeStrategy nullable(TypeStrategy
initialStrategy) {
return Optional.of(fromLogicalToDataType(inferredType));
};
+ /**
+ * Type strategy that returns a type of a field nested inside a
composite type that is described by the second argument.
+ * The second argument must be a literal that describes either the
nested field name or index.
+ */
+ public static final TypeStrategy GET = callContext -> {
+ List<DataType> argumentDataTypes =
callContext.getArgumentDataTypes();
+ DataType rowDataType = argumentDataTypes.get(0);
+ TableSchema nestedSchema =
DataTypeUtils.expandCompositeTypeToSchema(rowDataType);
Review comment:
I don't like that we use `TableSchema` here. This class doesn't belong
here. Furthermore, it does not support distinct type yet. Maybe we can just
introduce a `DataTypeUtils.getFieldType(int) and
DataTypeUtils.getFieldType(name)`. Can we make it support distinct type (an
`AtomicDataType`)?
##########
File path:
flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/InputTypeStrategiesTest.java
##########
@@ -582,7 +587,43 @@
"My constraint
says %s must be nullable.",
args ->
args.get(0).getLogicalType().isNullable()))))
.calledWithArgumentTypes(DataTypes.BOOLEAN().notNull())
- .expectErrorMessage("My constraint says BOOLEAN
NOT NULL must be nullable.")
+ .expectErrorMessage("My constraint says BOOLEAN
NOT NULL must be nullable."),
+
+ TestSpec
+ .forStrategy(
+ "Composite type strategy with ROW",
+ sequence(InputTypeStrategies.COMPOSITE)
+ )
+
.calledWithArgumentTypes(DataTypes.ROW(DataTypes.FIELD("f0",
DataTypes.BIGINT())))
+ .expectSignature("f(<COMPOSITE>)")
+
.expectArgumentTypes(DataTypes.ROW(DataTypes.FIELD("f0", DataTypes.BIGINT()))),
+
+ TestSpec
+ .forStrategy(
+ "Composite type strategy with
STRUCTURED type",
+ sequence(InputTypeStrategies.COMPOSITE)
+ )
+ .calledWithArgumentTypes(new FieldsDataType(
Review comment:
shall we upgrade the test base to accept `AbstractDataType` similar to
`DataStructureConverterTest`? it would make the code more readable.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
