You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by dw...@apache.org on 2023/11/06 08:53:24 UTC
(flink) branch master updated: [FLINK-18286] Implement type inference for functions on composite types
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new fa8ea3f9a44 [FLINK-18286] Implement type inference for functions on composite types
fa8ea3f9a44 is described below
commit fa8ea3f9a443f555de5c5479d725e16befefc957
Author: Dawid Wysakowicz <dw...@apache.org>
AuthorDate: Mon Oct 30 13:45:20 2023 +0100
[FLINK-18286] Implement type inference for functions on composite types
---
.../functions/BuiltInFunctionDefinitions.java | 18 +++-
.../table/types/inference/InputTypeStrategies.java | 4 +
.../strategies/ArrayElementTypeStrategy.java | 49 ++++++++++
.../ItemAtIndexArgumentTypeStrategy.java | 97 +++++++++++++++++++
.../inference/strategies/ItemAtTypeStrategy.java | 61 ++++++++++++
.../strategies/SpecificTypeStrategies.java | 7 +-
.../types/inference/strategies/StrategyUtils.java | 20 ++++
.../ItemAtIndexArgumentTypeStrategyTest.java | 77 +++++++++++++++
.../types/inference/TypeStrategiesTestBase.java | 15 ++-
.../strategies/ArrayElementTypeStrategyTest.java | 49 ++++++++++
.../strategies/ItemAtTypeStrategyTest.java | 42 +++++++++
.../expressions/PlannerExpressionConverter.scala | 12 ---
.../table/planner/expressions/collection.scala | 105 ---------------------
.../table/planner/expressions/ArrayTypeTest.scala | 10 +-
.../validation/ArrayTypeValidationTest.scala | 6 --
15 files changed, 436 insertions(+), 136 deletions(-)
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
index 2d49dc30725..4f9953f45c9 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
@@ -1826,14 +1826,25 @@ public final class BuiltInFunctionDefinitions {
BuiltInFunctionDefinition.newBuilder()
.name("at")
.kind(SCALAR)
- .outputTypeStrategy(TypeStrategies.MISSING)
+ .inputTypeStrategy(
+ sequence(
+ or(
+ logical(LogicalTypeRoot.ARRAY),
+ logical(LogicalTypeRoot.MAP)),
+ InputTypeStrategies.ITEM_AT_INDEX))
+ .outputTypeStrategy(SpecificTypeStrategies.ITEM_AT)
.build();
public static final BuiltInFunctionDefinition CARDINALITY =
BuiltInFunctionDefinition.newBuilder()
.name("cardinality")
.kind(SCALAR)
- .outputTypeStrategy(TypeStrategies.MISSING)
+ .inputTypeStrategy(
+ sequence(
+ or(
+ logical(LogicalTypeFamily.COLLECTION),
+ logical(LogicalTypeRoot.MAP))))
+ .outputTypeStrategy(nullableIfArgs(TypeStrategies.explicit(DataTypes.INT())))
.build();
public static final BuiltInFunctionDefinition ARRAY =
@@ -1848,7 +1859,8 @@ public final class BuiltInFunctionDefinitions {
BuiltInFunctionDefinition.newBuilder()
.name("element")
.kind(SCALAR)
- .outputTypeStrategy(TypeStrategies.MISSING)
+ .inputTypeStrategy(sequence(logical(LogicalTypeRoot.ARRAY)))
+ .outputTypeStrategy(forceNullable(SpecificTypeStrategies.ARRAY_ELEMENT))
.build();
public static final BuiltInFunctionDefinition MAP =
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/InputTypeStrategies.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/InputTypeStrategies.java
index 7ed5ba771f5..f5477c508d4 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/InputTypeStrategies.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/InputTypeStrategies.java
@@ -32,6 +32,7 @@ import org.apache.flink.table.types.inference.strategies.CompositeArgumentTypeSt
import org.apache.flink.table.types.inference.strategies.ConstraintArgumentTypeStrategy;
import org.apache.flink.table.types.inference.strategies.ExplicitArgumentTypeStrategy;
import org.apache.flink.table.types.inference.strategies.FamilyArgumentTypeStrategy;
+import org.apache.flink.table.types.inference.strategies.ItemAtIndexArgumentTypeStrategy;
import org.apache.flink.table.types.inference.strategies.LiteralArgumentTypeStrategy;
import org.apache.flink.table.types.inference.strategies.OrArgumentTypeStrategy;
import org.apache.flink.table.types.inference.strategies.OrInputTypeStrategy;
@@ -369,6 +370,9 @@ public final class InputTypeStrategies {
return new ArrayComparableElementTypeStrategy(StructuredComparison.FULL);
}
+ /** @see ItemAtIndexArgumentTypeStrategy */
+ public static final ArgumentTypeStrategy ITEM_AT_INDEX = new ItemAtIndexArgumentTypeStrategy();
+
// --------------------------------------------------------------------------------------------
private InputTypeStrategies() {
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ArrayElementTypeStrategy.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ArrayElementTypeStrategy.java
new file mode 100644
index 00000000000..b0515961e9c
--- /dev/null
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ArrayElementTypeStrategy.java
@@ -0,0 +1,49 @@
+/*
+ * 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.types.inference.strategies;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.types.CollectionDataType;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.TypeStrategy;
+import org.apache.flink.table.types.logical.LogicalTypeFamily;
+
+import java.util.Optional;
+
+/** Returns the element of an {@link LogicalTypeFamily#COLLECTION} type. */
+@Internal
+public final class ArrayElementTypeStrategy implements TypeStrategy {
+ @Override
+ public Optional<DataType> inferType(CallContext callContext) {
+
+ DataType arrayType = callContext.getArgumentDataTypes().get(0);
+ final Optional<DataType> legacyArrayElement =
+ StrategyUtils.extractLegacyArrayElement(arrayType);
+ if (legacyArrayElement.isPresent()) {
+ return legacyArrayElement;
+ }
+
+ if (!arrayType.getLogicalType().is(LogicalTypeFamily.COLLECTION)) {
+ return Optional.empty();
+ }
+
+ return Optional.of(((CollectionDataType) arrayType).getElementDataType().nullable());
+ }
+}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ItemAtIndexArgumentTypeStrategy.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ItemAtIndexArgumentTypeStrategy.java
new file mode 100644
index 00000000000..82b360ccc43
--- /dev/null
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ItemAtIndexArgumentTypeStrategy.java
@@ -0,0 +1,97 @@
+/*
+ * 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.types.inference.strategies;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.KeyValueDataType;
+import org.apache.flink.table.types.inference.ArgumentTypeStrategy;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeFamily;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.MapType;
+import org.apache.flink.table.types.logical.utils.LogicalTypeCasts;
+
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * An {@link ArgumentTypeStrategy} that expects:
+ *
+ * <p>a {@link LogicalTypeFamily#NUMERIC} if the first argument is an {@link LogicalTypeRoot#ARRAY}
+ * 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.
+ */
+@Internal
+public final class ItemAtIndexArgumentTypeStrategy implements ArgumentTypeStrategy {
+ @Override
+ public Optional<DataType> inferArgumentType(
+ CallContext callContext, int argumentPos, boolean throwOnFailure) {
+ final List<DataType> argumentDataTypes = callContext.getArgumentDataTypes();
+ final LogicalType collectionType = argumentDataTypes.get(0).getLogicalType();
+ final DataType indexType = argumentDataTypes.get(1);
+
+ if (collectionType.is(LogicalTypeRoot.ARRAY)) {
+ 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 {
+ return callContext.fail(
+ throwOnFailure, "Array can be indexed only using an INTEGER NUMERIC type.");
+ }
+ }
+
+ if (collectionType.is(LogicalTypeRoot.MAP)) {
+ MapType mapType = (MapType) collectionType;
+ if (LogicalTypeCasts.supportsImplicitCast(
+ indexType.getLogicalType(), mapType.getKeyType())) {
+ final KeyValueDataType mapDataType = (KeyValueDataType) argumentDataTypes.get(0);
+ return Optional.of(mapDataType.getKeyDataType());
+ } else {
+ return callContext.fail(
+ throwOnFailure,
+ "Expected index for a MAP to be of type: %s",
+ mapType.getKeyType());
+ }
+ }
+
+ return Optional.empty();
+ }
+
+ @Override
+ public Signature.Argument getExpectedArgument(
+ FunctionDefinition functionDefinition, int argumentPos) {
+ return Signature.Argument.of("[<INTEGER NUMERIC> | <MAP_KEY_TYPE>]");
+ }
+}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ItemAtTypeStrategy.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ItemAtTypeStrategy.java
new file mode 100644
index 00000000000..dc5a3ddebf8
--- /dev/null
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ItemAtTypeStrategy.java
@@ -0,0 +1,61 @@
+/*
+ * 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.types.inference.strategies;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.types.CollectionDataType;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.KeyValueDataType;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.TypeStrategy;
+import org.apache.flink.table.types.logical.LogicalTypeFamily;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+
+import java.util.Optional;
+
+/**
+ * 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}.
+ */
+@Internal
+public final class ItemAtTypeStrategy implements TypeStrategy {
+ @Override
+ public Optional<DataType> inferType(CallContext callContext) {
+
+ DataType arrayOrMapType = callContext.getArgumentDataTypes().get(0);
+ final Optional<DataType> legacyArrayElement =
+ StrategyUtils.extractLegacyArrayElement(arrayOrMapType);
+
+ if (legacyArrayElement.isPresent()) {
+ return legacyArrayElement;
+ }
+
+ if (arrayOrMapType.getLogicalType().is(LogicalTypeRoot.ARRAY)) {
+ return Optional.of(
+ ((CollectionDataType) arrayOrMapType).getElementDataType().nullable());
+ } else if (arrayOrMapType instanceof KeyValueDataType) {
+ return Optional.of(((KeyValueDataType) arrayOrMapType).getValueDataType().nullable());
+ }
+
+ return Optional.empty();
+ }
+}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificTypeStrategies.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificTypeStrategies.java
index 911f0d643e0..d6fba4409a5 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificTypeStrategies.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificTypeStrategies.java
@@ -59,12 +59,9 @@ public final class SpecificTypeStrategies {
public static final TypeStrategy ARRAY = new ArrayTypeStrategy();
/** Type strategy specific for array element. */
- public static final TypeStrategy ARRAY_ELEMENT =
- callContext ->
- Optional.of(
- ((CollectionDataType) callContext.getArgumentDataTypes().get(0))
- .getElementDataType());
+ public static final TypeStrategy ARRAY_ELEMENT = new ArrayElementTypeStrategy();
+ public static final TypeStrategy ITEM_AT = new ItemAtTypeStrategy();
/** See {@link GetTypeStrategy}. */
public static final TypeStrategy GET = new GetTypeStrategy();
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/StrategyUtils.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/StrategyUtils.java
index fe2b3d85ca4..a4b302fc5b0 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/StrategyUtils.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/StrategyUtils.java
@@ -18,18 +18,22 @@
package org.apache.flink.table.types.inference.strategies;
+import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.inference.CallContext;
import org.apache.flink.table.types.logical.BinaryType;
import org.apache.flink.table.types.logical.CharType;
import org.apache.flink.table.types.logical.DecimalType;
+import org.apache.flink.table.types.logical.LegacyTypeInformationType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.LogicalTypeFamily;
import org.apache.flink.table.types.logical.LogicalTypeRoot;
import org.apache.flink.table.types.logical.VarBinaryType;
import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.table.types.logical.utils.LogicalTypeCasts;
+import org.apache.flink.table.types.utils.TypeConversions;
import javax.annotation.Nullable;
@@ -197,6 +201,22 @@ final class StrategyUtils {
}
}
+ static Optional<DataType> extractLegacyArrayElement(DataType arrayType) {
+ final LogicalType logicalType = arrayType.getLogicalType();
+ if (logicalType instanceof LegacyTypeInformationType) {
+ final TypeInformation<?> typeInformation =
+ ((LegacyTypeInformationType<?>) logicalType).getTypeInformation();
+ if (typeInformation instanceof BasicArrayTypeInfo) {
+ return Optional.of(
+ TypeConversions.fromLegacyInfoToDataType(
+ ((BasicArrayTypeInfo<?, ?>) typeInformation)
+ .getComponentInfo())
+ .nullable());
+ }
+ }
+ return Optional.empty();
+ }
+
private StrategyUtils() {
// no instantiation
}
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/ItemAtIndexArgumentTypeStrategyTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/ItemAtIndexArgumentTypeStrategyTest.java
new file mode 100644
index 00000000000..d38ebd011f3
--- /dev/null
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/ItemAtIndexArgumentTypeStrategyTest.java
@@ -0,0 +1,77 @@
+/*
+ * 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.types.inference;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.types.inference.strategies.ItemAtIndexArgumentTypeStrategy;
+import org.apache.flink.table.types.utils.DataTypeFactoryMock;
+
+import java.util.stream.Stream;
+
+/** Tests for {@link ItemAtIndexArgumentTypeStrategy}. */
+class ItemAtIndexArgumentTypeStrategyTest extends InputTypeStrategiesTestBase {
+
+ private static final InputTypeStrategy ITEM_AT_INPUT_STRATEGY =
+ BuiltInFunctionDefinitions.AT
+ .getTypeInference(new DataTypeFactoryMock())
+ .getInputTypeStrategy();
+
+ @Override
+ protected Stream<TestSpec> testData() {
+
+ return Stream.of(
+ TestSpec.forStrategy("Validate integer index for an array", ITEM_AT_INPUT_STRATEGY)
+ .calledWithArgumentTypes(
+ DataTypes.ARRAY(DataTypes.STRING().notNull()),
+ DataTypes.SMALLINT().notNull())
+ .expectSignature(
+ "f([<ARRAY> | <MAP>], [<INTEGER NUMERIC> | <MAP_KEY_TYPE>])")
+ .expectArgumentTypes(
+ DataTypes.ARRAY(DataTypes.STRING().notNull()),
+ DataTypes.SMALLINT().notNull()),
+ TestSpec.forStrategy(
+ "Validate not an integer index for an array",
+ ITEM_AT_INPUT_STRATEGY)
+ .calledWithArgumentTypes(
+ DataTypes.ARRAY(DataTypes.STRING().notNull()), DataTypes.STRING())
+ .expectErrorMessage(
+ "Array can be indexed only using an INTEGER NUMERIC type."),
+ TestSpec.forStrategy("Validate correct map key", ITEM_AT_INPUT_STRATEGY)
+ .calledWithArgumentTypes(
+ DataTypes.MAP(DataTypes.BIGINT(), DataTypes.STRING().notNull()),
+ DataTypes.SMALLINT())
+ .expectSignature(
+ "f([<ARRAY> | <MAP>], [<INTEGER NUMERIC> | <MAP_KEY_TYPE>])")
+ .expectArgumentTypes(
+ DataTypes.MAP(DataTypes.BIGINT(), DataTypes.STRING().notNull()),
+ DataTypes.BIGINT()),
+ TestSpec.forStrategy("Validate incorrect map key", ITEM_AT_INPUT_STRATEGY)
+ .calledWithArgumentTypes(
+ DataTypes.MAP(DataTypes.BIGINT(), DataTypes.STRING().notNull()),
+ DataTypes.STRING())
+ .expectErrorMessage("Expected index for a MAP to be of type: BIGINT"),
+ TestSpec.forStrategy("Validate incorrect index", ITEM_AT_INPUT_STRATEGY)
+ .calledWithArgumentTypes(
+ DataTypes.ARRAY(DataTypes.BIGINT()), DataTypes.INT().notNull())
+ .calledWithLiteralAt(1, 0)
+ .expectErrorMessage(
+ "The provided index must be a valid SQL index starting from 1, but was '0'"));
+ }
+}
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/TypeStrategiesTestBase.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/TypeStrategiesTestBase.java
index 94a95e186de..310c1faa7ea 100644
--- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/TypeStrategiesTestBase.java
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/TypeStrategiesTestBase.java
@@ -54,8 +54,13 @@ public abstract class TypeStrategiesTestBase {
anyCauseMatches(
ValidationException.class, testSpec.expectedErrorMessage));
} else if (testSpec.expectedDataType != null) {
- assertThat(runTypeInference(testSpec).getOutputDataType())
- .isEqualTo(testSpec.expectedDataType);
+ if (testSpec.compareConversionClass) {
+ assertThat(runTypeInference(testSpec).getOutputDataType())
+ .isEqualTo(testSpec.expectedDataType);
+ } else {
+ assertThat(runTypeInference(testSpec).getOutputDataType().getLogicalType())
+ .isEqualTo(testSpec.expectedDataType.getLogicalType());
+ }
}
}
@@ -114,6 +119,7 @@ public abstract class TypeStrategiesTestBase {
private @Nullable Object literalValue;
private boolean isGroupedAggregation;
+ private boolean compareConversionClass = false;
private TestSpec(@Nullable String description, TypeStrategy strategy) {
this.description = description;
@@ -154,6 +160,11 @@ public abstract class TypeStrategiesTestBase {
return this;
}
+ public TestSpec compareConversionClass() {
+ this.compareConversionClass = true;
+ return this;
+ }
+
@Override
public String toString() {
return description != null ? description : "";
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/ArrayElementTypeStrategyTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/ArrayElementTypeStrategyTest.java
new file mode 100644
index 00000000000..6f0f518f766
--- /dev/null
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/ArrayElementTypeStrategyTest.java
@@ -0,0 +1,49 @@
+/*
+ * 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.types.inference.strategies;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.types.inference.TypeStrategiesTestBase;
+
+import java.util.stream.Stream;
+
+/** Tests for {@link ArrayElementTypeStrategy}. */
+class ArrayElementTypeStrategyTest extends TypeStrategiesTestBase {
+
+ @Override
+ protected Stream<TestSpec> testData() {
+ return Stream.of(
+ TestSpec.forStrategy(
+ "Infer an element of an array type",
+ SpecificTypeStrategies.ARRAY_ELEMENT)
+ .inputTypes(DataTypes.ARRAY(DataTypes.BIGINT().notNull()))
+ .expectDataType(DataTypes.BIGINT().nullable()),
+ TestSpec.forStrategy(
+ "Infer an element of a multiset type",
+ SpecificTypeStrategies.ARRAY_ELEMENT)
+ .inputTypes(DataTypes.MULTISET(DataTypes.STRING().notNull()))
+ .expectDataType(DataTypes.STRING().nullable()),
+ TestSpec.forStrategy(
+ "Error on non collection type",
+ SpecificTypeStrategies.ARRAY_ELEMENT)
+ .inputTypes(DataTypes.BIGINT())
+ .expectErrorMessage(
+ "Could not infer an output type for the given arguments."));
+ }
+}
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/ItemAtTypeStrategyTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/ItemAtTypeStrategyTest.java
new file mode 100644
index 00000000000..a908d98526b
--- /dev/null
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/ItemAtTypeStrategyTest.java
@@ -0,0 +1,42 @@
+/*
+ * 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.types.inference.strategies;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.types.inference.TypeStrategiesTestBase;
+
+import java.util.stream.Stream;
+
+/** Tests for {@link ItemAtTypeStrategy}. */
+class ItemAtTypeStrategyTest extends TypeStrategiesTestBase {
+
+ @Override
+ protected Stream<TestSpec> testData() {
+ return Stream.of(
+ TestSpec.forStrategy("Infer an item at array type", SpecificTypeStrategies.ITEM_AT)
+ .inputTypes(DataTypes.ARRAY(DataTypes.BIGINT().notNull()), DataTypes.INT())
+ .expectDataType(DataTypes.BIGINT().nullable()),
+ TestSpec.forStrategy("Infer an item at map type", SpecificTypeStrategies.ITEM_AT)
+ .inputTypes(
+ DataTypes.MAP(
+ DataTypes.STRING().notNull(), DataTypes.BIGINT().notNull()),
+ DataTypes.STRING())
+ .expectDataType(DataTypes.BIGINT().nullable()));
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/PlannerExpressionConverter.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/PlannerExpressionConverter.scala
index 953da5ede4f..7996be7dd87 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/PlannerExpressionConverter.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/PlannerExpressionConverter.scala
@@ -139,18 +139,6 @@ class PlannerExpressionConverter private extends ApiExpressionVisitor[PlannerExp
assert(args.size == 1)
Collect(args.head)
- case AT =>
- assert(args.size == 2)
- ItemAt(args.head, args.last)
-
- case CARDINALITY =>
- assert(args.size == 1)
- Cardinality(args.head)
-
- case ARRAY_ELEMENT =>
- assert(args.size == 1)
- ArrayElement(args.head)
-
case ORDER_ASC =>
assert(args.size == 1)
Asc(args.head)
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/collection.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/collection.scala
deleted file mode 100644
index 35bf9c2d721..00000000000
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/collection.scala
+++ /dev/null
@@ -1,105 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.table.planner.expressions
-
-import org.apache.flink.api.common.typeinfo.{BasicArrayTypeInfo, BasicTypeInfo, PrimitiveArrayTypeInfo, TypeInformation}
-import org.apache.flink.api.common.typeinfo.BasicTypeInfo.INT_TYPE_INFO
-import org.apache.flink.api.java.typeutils.{MapTypeInfo, ObjectArrayTypeInfo, RowTypeInfo}
-import org.apache.flink.table.planner.typeutils.TypeInfoCheckUtils.{isArray, isMap}
-import org.apache.flink.table.planner.validate.{ValidationFailure, ValidationResult, ValidationSuccess}
-
-case class ArrayElement(array: PlannerExpression) extends PlannerExpression {
-
- override private[flink] def children: Seq[PlannerExpression] = Seq(array)
-
- override def toString = s"($array).element()"
-
- override private[flink] def resultType = array.resultType match {
- case oati: ObjectArrayTypeInfo[_, _] => oati.getComponentInfo
- case bati: BasicArrayTypeInfo[_, _] => bati.getComponentInfo
- case pati: PrimitiveArrayTypeInfo[_] => pati.getComponentType
- }
-
- override private[flink] def validateInput(): ValidationResult = {
- array.resultType match {
- case ati: TypeInformation[_] if isArray(ati) => ValidationSuccess
- case other @ _ => ValidationFailure(s"Array expected but was '$other'.")
- }
- }
-}
-
-case class Cardinality(container: PlannerExpression) extends PlannerExpression {
-
- override private[flink] def children: Seq[PlannerExpression] = Seq(container)
-
- override def toString = s"($container).cardinality()"
-
- override private[flink] def resultType = BasicTypeInfo.INT_TYPE_INFO
-
- override private[flink] def validateInput(): ValidationResult = {
- container.resultType match {
- case mti: TypeInformation[_] if isMap(mti) => ValidationSuccess
- case ati: TypeInformation[_] if isArray(ati) => ValidationSuccess
- case other @ _ => ValidationFailure(s"Array or map expected but was '$other'.")
- }
- }
-}
-
-case class ItemAt(container: PlannerExpression, key: PlannerExpression) extends PlannerExpression {
-
- override private[flink] def children: Seq[PlannerExpression] = Seq(container, key)
-
- override def toString = s"($container).at($key)"
-
- override private[flink] def resultType = container.resultType match {
- case mti: MapTypeInfo[_, _] => mti.getValueTypeInfo
- case oati: ObjectArrayTypeInfo[_, _] => oati.getComponentInfo
- case bati: BasicArrayTypeInfo[_, _] => bati.getComponentInfo
- case pati: PrimitiveArrayTypeInfo[_] => pati.getComponentType
- }
-
- override private[flink] def validateInput(): ValidationResult = {
- container.resultType match {
-
- case ati: TypeInformation[_] if isArray(ati) =>
- if (key.resultType == INT_TYPE_INFO) {
- // check for common user mistake
- key match {
- case Literal(value: Int, INT_TYPE_INFO) if value < 1 =>
- ValidationFailure(
- s"Array element access needs an index starting at 1 but was $value.")
- case _ => ValidationSuccess
- }
- } else {
- ValidationFailure(
- s"Array element access needs an integer index but was '${key.resultType}'.")
- }
-
- case mti: MapTypeInfo[_, _] =>
- if (key.resultType == mti.getKeyTypeInfo) {
- ValidationSuccess
- } else {
- ValidationFailure(
- s"Map entry access needs a valid key of type " +
- s"'${mti.getKeyTypeInfo}', found '${key.resultType}'.")
- }
-
- case other @ _ => ValidationFailure(s"Array or map expected but was '$other'.")
- }
- }
-}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/ArrayTypeTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/ArrayTypeTest.scala
index d5465104c1d..bc4b13615e3 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/ArrayTypeTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/ArrayTypeTest.scala
@@ -17,10 +17,12 @@
*/
package org.apache.flink.table.planner.expressions
+import org.apache.flink.core.testutils.FlinkAssertions.anyCauseMatches
import org.apache.flink.table.api._
import org.apache.flink.table.planner.expressions.utils.ArrayTypeTestBase
import org.apache.flink.table.planner.utils.DateTimeTestUtil.{localDate, localDateTime, localTime => gLocalTime}
+import org.assertj.core.api.Assertions.assertThatThrownBy
import org.junit.jupiter.api.Test
import java.time.{LocalDateTime => JLocalDateTime}
@@ -226,9 +228,11 @@ class ArrayTypeTest extends ArrayTypeTestBase {
@Test
def testArrayIndexStaticCheckForTable(): Unit = {
- testExpectedTableApiException(
- 'f2.at(0),
- "Array element access needs an index starting at 1 but was 0.")
+ assertThatThrownBy(() => testTableApi('f2.at(0), "1"))
+ .satisfies(
+ anyCauseMatches(
+ classOf[ValidationException],
+ "The provided index must be a valid SQL index starting from 1"))
}
@Test
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/validation/ArrayTypeValidationTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/validation/ArrayTypeValidationTest.scala
index ce47498ebfe..ae71cfbb604 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/validation/ArrayTypeValidationTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/validation/ArrayTypeValidationTest.scala
@@ -31,12 +31,6 @@ class ArrayTypeValidationTest extends ArrayTypeTestBase {
.isThrownBy(() => testSqlApi("ARRAY['string', 12]", "FAIL"))
}
- @Test
- def testObviousInvalidIndexTableApi(): Unit = {
- assertThatExceptionOfType(classOf[ValidationException])
- .isThrownBy(() => testTableApi('f2.at(0), "FAIL"))
- }
-
@Test
def testEmptyArraySql(): Unit = {
assertThatExceptionOfType(classOf[ValidationException])