You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sn...@apache.org on 2023/03/26 21:35:07 UTC
[flink] branch master updated: [FLINK-31102][table] Add ARRAY_REMOVE function (#22144)
This is an automated email from the ASF dual-hosted git repository.
snuyanzin 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 f3c653ed2e4 [FLINK-31102][table] Add ARRAY_REMOVE function (#22144)
f3c653ed2e4 is described below
commit f3c653ed2e4264315ed83a5b4b2494a7dcc41474
Author: Jacky Lau <li...@gmail.com>
AuthorDate: Mon Mar 27 05:34:55 2023 +0800
[FLINK-31102][table] Add ARRAY_REMOVE function (#22144)
---
docs/data/sql_functions.yml | 3 +
.../docs/reference/pyflink.table/expressions.rst | 1 +
flink-python/pyflink/table/expression.py | 7 ++
.../flink/table/api/internal/BaseExpressions.java | 11 +++
.../functions/BuiltInFunctionDefinitions.java | 15 +++
.../functions/CollectionFunctionsITCase.java | 109 ++++++++++++++++++++-
.../functions/scalar/ArrayRemoveFunction.java | 95 ++++++++++++++++++
7 files changed, 240 insertions(+), 1 deletion(-)
diff --git a/docs/data/sql_functions.yml b/docs/data/sql_functions.yml
index 7dd6d8d446a..15c30eda42d 100644
--- a/docs/data/sql_functions.yml
+++ b/docs/data/sql_functions.yml
@@ -634,6 +634,9 @@ collection:
- sql: ARRAY_DISTINCT(haystack)
table: haystack.arrayDistinct()
description: Returns an array with unique elements. If the array itself is null, the function will return null. Keeps ordering of elements.
+ - sql: ARRAY_REMOVE(haystack, needle)
+ table: haystack.arrayRemove(needle)
+ description: Removes all elements that equal to element from array. If the array itself is null, the function will return null. Keeps ordering of elements.
- sql: MAP_KEYS(map)
table: MAP.mapKeys()
description: Returns the keys of the map as array. No order guaranteed.
diff --git a/flink-python/docs/reference/pyflink.table/expressions.rst b/flink-python/docs/reference/pyflink.table/expressions.rst
index 981ab77cc23..f04c16433d4 100644
--- a/flink-python/docs/reference/pyflink.table/expressions.rst
+++ b/flink-python/docs/reference/pyflink.table/expressions.rst
@@ -227,6 +227,7 @@ advanced type helper functions
Expression.element
Expression.array_contains
Expression.array_distinct
+ Expression.array_remove
Expression.map_keys
Expression.map_values
diff --git a/flink-python/pyflink/table/expression.py b/flink-python/pyflink/table/expression.py
index 34ea0716603..ab608c5cd02 100644
--- a/flink-python/pyflink/table/expression.py
+++ b/flink-python/pyflink/table/expression.py
@@ -1487,6 +1487,13 @@ class Expression(Generic[T]):
"""
return _binary_op("arrayDistinct")(self)
+ def array_remove(self, needle) -> 'Expression':
+ """
+ Removes all elements that equal to element from array.
+ If the array itself is null, the function will return null. Keeps ordering of elements.
+ """
+ return _binary_op("arrayRemove")(self, needle)
+
@property
def map_keys(self) -> 'Expression':
"""
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
index 30899f49b68..c43307dcb3d 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
@@ -56,6 +56,7 @@ import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.AND;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.ARRAY_CONTAINS;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.ARRAY_DISTINCT;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.ARRAY_ELEMENT;
+import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.ARRAY_REMOVE;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.ASCII;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.ASIN;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.AT;
@@ -1361,6 +1362,16 @@ public abstract class BaseExpressions<InType, OutType> {
return toApiSpecificExpression(unresolvedCall(ARRAY_DISTINCT, toExpr()));
}
+ /**
+ * Removes all elements that equal to element from array.
+ *
+ * <p>If the array itself is null, the function will return null. Keeps ordering of elements.
+ */
+ public OutType arrayRemove(InType needle) {
+ return toApiSpecificExpression(
+ unresolvedCall(ARRAY_REMOVE, toExpr(), objectToExpression(needle)));
+ }
+
/** Returns the keys of the map as an array. */
public OutType mapKeys() {
return toApiSpecificExpression(unresolvedCall(MAP_KEYS, toExpr()));
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 7cd7800a4f0..8860fe02a17 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
@@ -219,6 +219,21 @@ public final class BuiltInFunctionDefinitions {
.runtimeClass(
"org.apache.flink.table.runtime.functions.scalar.ArrayDistinctFunction")
.build();
+
+ public static final BuiltInFunctionDefinition ARRAY_REMOVE =
+ BuiltInFunctionDefinition.newBuilder()
+ .name("ARRAY_REMOVE")
+ .kind(SCALAR)
+ .inputTypeStrategy(
+ sequence(
+ Arrays.asList("haystack", "needle"),
+ Arrays.asList(
+ logical(LogicalTypeRoot.ARRAY), ARRAY_ELEMENT_ARG)))
+ .outputTypeStrategy(nullableIfArgs(ConstantArgumentCount.of(0), argument(0)))
+ .runtimeClass(
+ "org.apache.flink.table.runtime.functions.scalar.ArrayRemoveFunction")
+ .build();
+
public static final BuiltInFunctionDefinition INTERNAL_REPLICATE_ROWS =
BuiltInFunctionDefinition.newBuilder()
.name("$REPLICATE_ROWS$1")
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CollectionFunctionsITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CollectionFunctionsITCase.java
index 684338f062d..da216006c08 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CollectionFunctionsITCase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CollectionFunctionsITCase.java
@@ -21,13 +21,16 @@ package org.apache.flink.table.planner.functions;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
import org.apache.flink.types.Row;
+import org.apache.flink.util.CollectionUtil;
import java.time.LocalDate;
+import java.util.Map;
import java.util.stream.Stream;
import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.lit;
import static org.apache.flink.table.api.Expressions.row;
+import static org.apache.flink.util.CollectionUtil.entry;
/** Tests for {@link BuiltInFunctionDefinitions} around arrays. */
class CollectionFunctionsITCase extends BuiltInFunctionTestBase {
@@ -178,6 +181,110 @@ class CollectionFunctionsITCase extends BuiltInFunctionTestBase {
null
},
DataTypes.ARRAY(
- DataTypes.ROW(DataTypes.BOOLEAN(), DataTypes.DATE()))));
+ DataTypes.ROW(DataTypes.BOOLEAN(), DataTypes.DATE()))),
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.ARRAY_REMOVE)
+ .onFieldsWithData(
+ new Integer[] {1, 2, 2},
+ null,
+ new Row[] {
+ Row.of(true, LocalDate.of(2022, 4, 20)),
+ Row.of(true, LocalDate.of(1990, 10, 14)),
+ null
+ },
+ new Integer[] {null, null, 1},
+ new Integer[][] {
+ new Integer[] {1, null, 3}, new Integer[] {0}, new Integer[] {1}
+ },
+ new Map[] {
+ CollectionUtil.map(entry(1, "a"), entry(2, "b")),
+ CollectionUtil.map(entry(3, "c"), entry(4, "d")),
+ null
+ })
+ .andDataTypes(
+ DataTypes.ARRAY(DataTypes.INT()),
+ DataTypes.ARRAY(DataTypes.INT()),
+ DataTypes.ARRAY(
+ DataTypes.ROW(DataTypes.BOOLEAN(), DataTypes.DATE())),
+ DataTypes.ARRAY(DataTypes.INT()),
+ DataTypes.ARRAY(DataTypes.ARRAY(DataTypes.INT())),
+ DataTypes.ARRAY(DataTypes.MAP(DataTypes.INT(), DataTypes.STRING())))
+ // ARRAY<INT>
+ .testResult(
+ $("f0").arrayRemove(2),
+ "ARRAY_REMOVE(f0, 2)",
+ new Integer[] {1},
+ DataTypes.ARRAY(DataTypes.INT()).nullable())
+ .testResult(
+ $("f0").arrayRemove(42),
+ "ARRAY_REMOVE(f0, 42)",
+ new Integer[] {1, 2, 2},
+ DataTypes.ARRAY(DataTypes.INT()).nullable())
+ .testResult(
+ $("f0").arrayRemove(
+ lit(null, DataTypes.SMALLINT())
+ .cast(DataTypes.INT())),
+ "ARRAY_REMOVE(f0, CAST(NULL AS INT))",
+ new Integer[] {1, 2, 2},
+ DataTypes.ARRAY(DataTypes.INT()).nullable())
+ // ARRAY<INT> of NULL value
+ .testResult(
+ $("f1").arrayRemove(12),
+ "ARRAY_REMOVE(f1, 12)",
+ null,
+ DataTypes.ARRAY(DataTypes.INT()).nullable())
+ .testResult(
+ $("f1").arrayRemove(null),
+ "ARRAY_REMOVE(f1, NULL)",
+ null,
+ DataTypes.ARRAY(DataTypes.INT()).nullable())
+ // ARRAY<ROW<BOOLEAN, DATE>>
+ .testResult(
+ $("f2").arrayRemove(row(true, LocalDate.of(1990, 10, 14))),
+ "ARRAY_REMOVE(f2, (TRUE, DATE '1990-10-14'))",
+ new Row[] {Row.of(true, LocalDate.of(2022, 4, 20)), null},
+ DataTypes.ARRAY(
+ DataTypes.ROW(
+ DataTypes.BOOLEAN(), DataTypes.DATE()))
+ .nullable())
+ .testResult(
+ $("f2").arrayRemove(null),
+ "ARRAY_REMOVE(f2, NULL)",
+ new Row[] {
+ Row.of(true, LocalDate.of(2022, 4, 20)),
+ Row.of(true, LocalDate.of(1990, 10, 14)),
+ },
+ DataTypes.ARRAY(
+ DataTypes.ROW(
+ DataTypes.BOOLEAN(), DataTypes.DATE()))
+ .nullable())
+ // ARRAY<INT> with NULL elements
+ .testResult(
+ $("f3").arrayRemove(null),
+ "ARRAY_REMOVE(f3, NULL)",
+ new Integer[] {1},
+ DataTypes.ARRAY(DataTypes.INT()).nullable())
+ // ARRAY<ARRAY<INT>>
+ .testResult(
+ $("f4").arrayRemove(new Integer[] {0}),
+ "ARRAY_REMOVE(f4, ARRAY[0])",
+ new Integer[][] {new Integer[] {1, null, 3}, new Integer[] {1}},
+ DataTypes.ARRAY(DataTypes.ARRAY(DataTypes.INT()).nullable()))
+ // ARRAY<MAP<INT, STRING>> with NULL elements
+ .testResult(
+ $("f5").arrayRemove(
+ CollectionUtil.map(entry(3, "c"), entry(4, "d"))),
+ "ARRAY_REMOVE(f5, MAP[3, 'c', 4, 'd'])",
+ new Map[] {CollectionUtil.map(entry(1, "a"), entry(2, "b")), null},
+ DataTypes.ARRAY(DataTypes.MAP(DataTypes.INT(), DataTypes.STRING()))
+ .nullable())
+ // invalid signatures
+ .testSqlValidationError(
+ "ARRAY_REMOVE(f0, TRUE)",
+ "Invalid input arguments. Expected signatures are:\n"
+ + "ARRAY_REMOVE(haystack <ARRAY>, needle <ARRAY ELEMENT>)")
+ .testTableApiValidationError(
+ $("f0").arrayRemove(true),
+ "Invalid input arguments. Expected signatures are:\n"
+ + "ARRAY_REMOVE(haystack <ARRAY>, needle <ARRAY ELEMENT>)"));
}
}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ArrayRemoveFunction.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ArrayRemoveFunction.java
new file mode 100644
index 00000000000..fa05732b727
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ArrayRemoveFunction.java
@@ -0,0 +1,95 @@
+/*
+ * 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.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.CollectionDataType;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import javax.annotation.Nullable;
+
+import java.lang.invoke.MethodHandle;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#ARRAY_REMOVE}. */
+@Internal
+public class ArrayRemoveFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter elementGetter;
+ private final SpecializedFunction.ExpressionEvaluator equalityEvaluator;
+ private transient MethodHandle equalityHandle;
+
+ public ArrayRemoveFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.ARRAY_REMOVE, context);
+ final DataType elementDataType =
+ ((CollectionDataType) context.getCallContext().getArgumentDataTypes().get(0))
+ .getElementDataType();
+ final DataType needleDataType = context.getCallContext().getArgumentDataTypes().get(1);
+ elementGetter = ArrayData.createElementGetter(elementDataType.getLogicalType());
+ equalityEvaluator =
+ context.createEvaluator(
+ $("element").isEqual($("needle")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("element", elementDataType.notNull().toInternal()),
+ DataTypes.FIELD("needle", needleDataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ equalityHandle = equalityEvaluator.open(context);
+ }
+
+ public @Nullable ArrayData eval(ArrayData haystack, Object needle) {
+ try {
+ if (haystack == null) {
+ return null;
+ }
+
+ List list = new ArrayList();
+ final int size = haystack.size();
+ for (int pos = 0; pos < size; pos++) {
+ final Object element = elementGetter.getElementOrNull(haystack, pos);
+ if ((element == null && needle == null)
+ || (element != null
+ && needle != null
+ && (boolean) equalityHandle.invoke(element, needle))) {
+ continue;
+ }
+ list.add(element);
+ }
+ return new GenericArrayData(list.toArray());
+ } catch (Throwable t) {
+ throw new FlinkRuntimeException(t);
+ }
+ }
+
+ @Override
+ public void close() throws Exception {
+ equalityEvaluator.close();
+ }
+}