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 4b0a39590b9 [FLINK-31664][table] Implement ARRAY_INTERSECT function
4b0a39590b9 is described below
commit 4b0a39590b9159e5904290a06684c7b677e489ba
Author: Jacky Lau <[email protected]>
AuthorDate: Fri Jun 7 05:49:52 2024 +0800
[FLINK-31664][table] Implement ARRAY_INTERSECT function
---
docs/data/sql_functions.yml | 3 +
.../docs/reference/pyflink.table/expressions.rst | 2 +-
flink-python/pyflink/table/expression.py | 9 ++
.../flink/table/api/internal/BaseExpressions.java | 14 +++
.../functions/BuiltInFunctionDefinitions.java | 10 ++
.../functions/CollectionFunctionsITCase.java | 78 +++++++++++++++
.../functions/scalar/ArrayIntersectFunction.java | 105 +++++++++++++++++++++
.../flink/table/runtime/util/ObjectContainer.java | 14 ++-
8 files changed, 229 insertions(+), 6 deletions(-)
diff --git a/docs/data/sql_functions.yml b/docs/data/sql_functions.yml
index 17f03773786..6901891aeb8 100644
--- a/docs/data/sql_functions.yml
+++ b/docs/data/sql_functions.yml
@@ -688,6 +688,9 @@ collection:
- sql: ARRAY_EXCEPT(array1, array2)
table: arrayOne.arrayExcept(arrayTwo)
description: Returns an ARRAY that contains the elements from array1 that
are not in array2. If no elements remain after excluding the elements in array2
from array1, the function returns an empty ARRAY. If one or both arguments are
NULL, the function returns NULL. The order of the elements from array1 is kept.
+ - sql: ARRAY_INTERSECT(array1, array2)
+ table: array1.arrayIntersect(array2)
+ description: Returns an ARRAY that contains the elements from array1 that
are also in array2, without duplicates. If no elements that are both in array1
and array2, the function returns an empty ARRAY. If any of the array is null,
the function will return null. The order of the elements from array1 is kept.
- sql: SPLIT(string, delimiter)
table: string.split(delimiter)
description: Returns an array of substrings by splitting the input string
based on the given delimiter. If the delimiter is not found in the string, the
original string is returned as the only element in the array. If the delimiter
is empty, every character in the string is split. If the string or delimiter is
null, a null value is returned. If the delimiter is found at the beginning or
end of the string, or there are contiguous delimiters, then an empty string is
added to the array.
diff --git a/flink-python/docs/reference/pyflink.table/expressions.rst
b/flink-python/docs/reference/pyflink.table/expressions.rst
index f9c3247c5f2..3351cf9cdd2 100644
--- a/flink-python/docs/reference/pyflink.table/expressions.rst
+++ b/flink-python/docs/reference/pyflink.table/expressions.rst
@@ -243,9 +243,9 @@ advanced type helper functions
Expression.map_union
Expression.map_values
Expression.array_except
+ Expression.array_intersect
Expression.split
-
time definition functions
-------------------------
diff --git a/flink-python/pyflink/table/expression.py
b/flink-python/pyflink/table/expression.py
index f305a95f5c9..9bfcb7746bb 100644
--- a/flink-python/pyflink/table/expression.py
+++ b/flink-python/pyflink/table/expression.py
@@ -1618,6 +1618,15 @@ class Expression(Generic[T]):
"""
return _binary_op("arrayExcept")(self, array)
+ def array_intersect(self, array) -> 'Expression':
+ """
+ Returns an ARRAY that contains the elements from array1 that are also
in array2,
+ without duplicates. If no elements are both in array1 and array2, the
function
+ returns an empty ARRAY. If one or both arguments are NULL, the
function returns NULL.
+ The order of the elements from array1 is kept.
+ """
+ return _binary_op("arrayIntersect")(self, array)
+
def split(self, delimiter) -> 'Expression':
"""
Returns an array of substrings by splitting the input string based on
the given delimiter.
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 9d66a0d13fc..13fef5ad1db 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
@@ -60,6 +60,7 @@ import static
org.apache.flink.table.functions.BuiltInFunctionDefinitions.ARRAY_
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_EXCEPT;
+import static
org.apache.flink.table.functions.BuiltInFunctionDefinitions.ARRAY_INTERSECT;
import static
org.apache.flink.table.functions.BuiltInFunctionDefinitions.ARRAY_MAX;
import static
org.apache.flink.table.functions.BuiltInFunctionDefinitions.ARRAY_MIN;
import static
org.apache.flink.table.functions.BuiltInFunctionDefinitions.ARRAY_POSITION;
@@ -246,6 +247,19 @@ public abstract class BaseExpressions<InType, OutType> {
unresolvedCall(ARRAY_EXCEPT, toExpr(),
objectToExpression(array)));
}
+ /**
+ * Returns an ARRAY that contains the elements from array1 that are also
in array2, without
+ * duplicates. If no elements that are both in array1 and array2, the
function returns an empty
+ * ARRAY.
+ *
+ * <p>If one or both arguments are NULL, the function returns NULL. The
order of the elements
+ * from array1 is kept.
+ */
+ public OutType arrayIntersect(InType array) {
+ return toApiSpecificExpression(
+ unresolvedCall(ARRAY_INTERSECT, toExpr(),
objectToExpression(array)));
+ }
+
/**
* Boolean AND in three-valued logic. This is an infix notation. See also
{@link
* Expressions#and(Object, Object, Object...)} for prefix notation with
multiple arguments.
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 6f57079f460..e91c146fea8 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
@@ -471,6 +471,16 @@ public final class BuiltInFunctionDefinitions {
"org.apache.flink.table.runtime.functions.scalar.ArrayExceptFunction")
.build();
+ public static final BuiltInFunctionDefinition ARRAY_INTERSECT =
+ BuiltInFunctionDefinition.newBuilder()
+ .name("ARRAY_INTERSECT")
+ .kind(SCALAR)
+ .inputTypeStrategy(commonArrayType(2))
+ .outputTypeStrategy(nullableIfArgs(COMMON))
+ .runtimeClass(
+
"org.apache.flink.table.runtime.functions.scalar.ArrayIntersectFunction")
+ .build();
+
//
--------------------------------------------------------------------------------------------
// Logic functions
//
--------------------------------------------------------------------------------------------
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 75844eba993..4f6cb758914 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
@@ -56,6 +56,7 @@ class CollectionFunctionsITCase extends
BuiltInFunctionTestBase {
arrayMinTestCases(),
arraySortTestCases(),
arrayExceptTestCases(),
+ arrayIntersectTestCases(),
splitTestCases())
.flatMap(s -> s);
}
@@ -1723,6 +1724,83 @@ class CollectionFunctionsITCase extends
BuiltInFunctionTestBase {
+ "ARRAY_EXCEPT(<COMMON>, <COMMON>)"));
}
+ private Stream<TestSetSpec> arrayIntersectTestCases() {
+ return Stream.of(
+
TestSetSpec.forFunction(BuiltInFunctionDefinitions.ARRAY_INTERSECT)
+ .onFieldsWithData(
+ new Integer[] {1, 1, 2},
+ null,
+ new Row[] {Row.of(true, 1), Row.of(true, 2),
null},
+ new Integer[] {null, null, 1},
+ new Map[] {
+ CollectionUtil.map(entry(1, "a"), entry(2,
"b")),
+ CollectionUtil.map(entry(3, "c"), entry(4,
"d"))
+ },
+ new Integer[][] {new Integer[] {1, 2, 3}})
+ .andDataTypes(
+ DataTypes.ARRAY(DataTypes.INT()),
+ DataTypes.ARRAY(DataTypes.INT()),
+ DataTypes.ARRAY(
+ DataTypes.ROW(DataTypes.BOOLEAN(),
DataTypes.INT())),
+ DataTypes.ARRAY(DataTypes.INT()),
+ DataTypes.ARRAY(DataTypes.MAP(DataTypes.INT(),
DataTypes.STRING())),
+
DataTypes.ARRAY(DataTypes.ARRAY(DataTypes.INT())))
+ // ARRAY<INT>
+ .testResult(
+ $("f0").arrayIntersect(new Integer[] {1, null,
4}),
+ "ARRAY_INTERSECT(f0, ARRAY[1, NULL, 4])",
+ new Integer[] {1},
+ DataTypes.ARRAY(DataTypes.INT()))
+ .testResult(
+ $("f0").arrayIntersect(new Integer[] {3, 4}),
+ "ARRAY_INTERSECT(f0, ARRAY[3, 4])",
+ new Integer[] {},
+ DataTypes.ARRAY(DataTypes.INT()))
+ .testResult(
+ $("f1").arrayIntersect(new Integer[] {1, null,
4}),
+ "ARRAY_INTERSECT(f1, ARRAY[1, NULL, 4])",
+ null,
+ DataTypes.ARRAY(DataTypes.INT()))
+ // ARRAY<ROW<BOOLEAN, DATE>>
+ .testResult(
+ $("f2").arrayIntersect(
+ new Row[] {
+ null, Row.of(true, 2),
+ }),
+ "ARRAY_INTERSECT(f2, ARRAY[NULL, ROW(TRUE,
2)])",
+ new Row[] {Row.of(true, 2), null},
+ DataTypes.ARRAY(
+ DataTypes.ROW(DataTypes.BOOLEAN(),
DataTypes.INT())))
+ // arrayOne contains null elements
+ .testResult(
+ $("f3").arrayIntersect(new Integer[] {null,
42}),
+ "ARRAY_INTERSECT(f3, ARRAY[null, 42])",
+ new Integer[] {null},
+ DataTypes.ARRAY(DataTypes.INT()).nullable())
+ .testResult(
+ $("f4").arrayIntersect(
+ new Map[] {
+
CollectionUtil.map(entry(1, "a"), entry(2, "b"))
+ }),
+ "ARRAY_INTERSECT(f4, ARRAY[MAP[1, 'a', 2,
'b']])",
+ new Map[] {CollectionUtil.map(entry(1, "a"),
entry(2, "b"))},
+ DataTypes.ARRAY(DataTypes.MAP(DataTypes.INT(),
DataTypes.STRING())))
+ .testResult(
+ $("f5").arrayIntersect(new Integer[][] {new
Integer[] {1, 2, 3}}),
+ "ARRAY_INTERSECT(f5, ARRAY[ARRAY[1, 2, 3]])",
+ new Integer[][] {new Integer[] {1, 2, 3}},
+
DataTypes.ARRAY(DataTypes.ARRAY(DataTypes.INT())))
+ // invalid signatures
+ .testSqlValidationError(
+ "ARRAY_INTERSECT(f3, TRUE)",
+ "Invalid input arguments. Expected signatures
are:\n"
+ + "ARRAY_INTERSECT(<COMMON>,
<COMMON>)")
+ .testTableApiValidationError(
+ $("f3").arrayIntersect(true),
+ "Invalid input arguments. Expected signatures
are:\n"
+ + "ARRAY_INTERSECT(<COMMON>,
<COMMON>)"));
+ }
+
private Stream<TestSetSpec> splitTestCases() {
return Stream.of(
TestSetSpec.forFunction(BuiltInFunctionDefinitions.SPLIT)
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ArrayIntersectFunction.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ArrayIntersectFunction.java
new file mode 100644
index 00000000000..ba10c8975e1
--- /dev/null
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ArrayIntersectFunction.java
@@ -0,0 +1,105 @@
+/*
+ * 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.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.runtime.util.EqualityAndHashcodeProvider;
+import org.apache.flink.table.runtime.util.ObjectContainer;
+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.util.HashSet;
+import java.util.LinkedHashSet;
+import java.util.Set;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#ARRAY_INTERSECT}. */
+@Internal
+public class ArrayIntersectFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter elementGetter;
+ private final EqualityAndHashcodeProvider equalityAndHashcodeProvider;
+
+ public ArrayIntersectFunction(SpecializedFunction.SpecializedContext
context) {
+ super(BuiltInFunctionDefinitions.ARRAY_INTERSECT, context);
+ final DataType dataType =
+ ((CollectionDataType)
context.getCallContext().getArgumentDataTypes().get(0))
+ .getElementDataType()
+ .toInternal();
+ elementGetter =
ArrayData.createElementGetter(dataType.toInternal().getLogicalType());
+ this.equalityAndHashcodeProvider = new
EqualityAndHashcodeProvider(context, dataType);
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ equalityAndHashcodeProvider.open(context);
+ }
+
+ public @Nullable ArrayData eval(ArrayData arrayOne, ArrayData arrayTwo) {
+ try {
+ if (arrayOne == null || arrayTwo == null) {
+ return null;
+ }
+
+ Set<ObjectContainer> set = new HashSet<>();
+ for (int pos = 0; pos < arrayTwo.size(); pos++) {
+ final Object element =
elementGetter.getElementOrNull(arrayTwo, pos);
+ final ObjectContainer objectContainer =
createObjectContainer(element);
+ set.add(objectContainer);
+ }
+
+ Set<ObjectContainer> res = new LinkedHashSet<>();
+ for (int pos = 0; pos < arrayOne.size(); pos++) {
+ final Object element =
elementGetter.getElementOrNull(arrayOne, pos);
+ final ObjectContainer objectContainer =
createObjectContainer(element);
+ if (set.contains(objectContainer)) {
+ res.add(objectContainer);
+ }
+ }
+
+ return new GenericArrayData(
+ res.stream()
+ .map(element -> element != null ?
element.getObject() : null)
+ .toArray());
+ } catch (Throwable t) {
+ throw new FlinkRuntimeException(t);
+ }
+ }
+
+ private ObjectContainer createObjectContainer(Object element) {
+ if (element == null) {
+ return null;
+ }
+ return new ObjectContainer(
+ element,
+ equalityAndHashcodeProvider::equals,
+ equalityAndHashcodeProvider::hashCode);
+ }
+
+ @Override
+ public void close() throws Exception {
+ equalityAndHashcodeProvider.close();
+ }
+}
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/ObjectContainer.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/ObjectContainer.java
index 1a876ce0246..0c70a1f5ba8 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/ObjectContainer.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/ObjectContainer.java
@@ -31,21 +31,25 @@ import java.util.function.Function;
@Internal
public class ObjectContainer {
- private final Object o;
+ private final Object object;
private final BiFunction<Object, Object, Boolean> equalsMethod;
private final Function<Object, Integer> hashCodeMethod;
public ObjectContainer(
- Object o,
+ Object object,
BiFunction<Object, Object, Boolean> equalsMethod,
Function<Object, Integer> hashCodeMethod) {
- this.o = o;
+ this.object = object;
this.equalsMethod = equalsMethod;
this.hashCodeMethod = hashCodeMethod;
}
+ public Object getObject() {
+ return object;
+ }
+
@Override
public boolean equals(Object other) {
if (this == other) {
@@ -55,11 +59,11 @@ public class ObjectContainer {
return false;
}
ObjectContainer that = (ObjectContainer) other;
- return equalsMethod.apply(this.o, that.o);
+ return equalsMethod.apply(this.object, that.object);
}
@Override
public int hashCode() {
- return hashCodeMethod.apply(o);
+ return hashCodeMethod.apply(object);
}
}
