You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "hanyuzheng7 (via GitHub)" <gi...@apache.org> on 2023/08/08 20:34:53 UTC
[GitHub] [flink] hanyuzheng7 opened a new pull request, #23171: [FLINK-31664]Add ARRAY_INTERSECT function
hanyuzheng7 opened a new pull request, #23171:
URL: https://github.com/apache/flink/pull/23171
## What is the purpose of the change
This is an implementation of ARRAY_INTERSECT function
Returns an array of the elements in the intersection of array1 and array2, without duplicates.
## Brief change log
ARRAY_INTERSECT for Table API and SQL
## Syntax:
`ARRAY_INTERSECT(array1, array2)`
## Arguments:
array: An ARRAY to be handled.
## Returns:
Returns an array of the elements in the intersection of array1 and array2, without duplicates..
## Examples:
`SELECT array_intersect(array(1, 2, 3), array(1, 3, 5));
[1,3]
`
## Verifying this change
CollectionFunctionsITCase
## see also:
spark https://spark.apache.org/docs/latest/api/sql/index.html#array_intersect
presto https://prestodb.io/docs/current/functions/array.html
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / no)
- The serializers: (yes / no / don't know)
- The runtime per-record code paths (performance sensitive): (yes / no / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / no / don't know)
- The S3 file system connector: (yes / no / don't know)
## Documentation
- Does this pull request introduce a new feature? (yes / no)
- If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
--
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
[GitHub] [flink] flinkbot commented on pull request #23171: [FLINK-31664][table]Add ARRAY_INTERSECT function
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #23171:
URL: https://github.com/apache/flink/pull/23171#issuecomment-1670275688
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "8fccf4099b98ee0907de7527586255627b8cf653",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "8fccf4099b98ee0907de7527586255627b8cf653",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8fccf4099b98ee0907de7527586255627b8cf653 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-31664][table]Add ARRAY_INTERSECT function [flink]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #23171:
URL: https://github.com/apache/flink/pull/23171#discussion_r1491954798
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ArrayIntersectFunction.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.api.Expressions;
+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.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#ARRAY_INTERSECT}. */
+@Internal
+public class ArrayIntersectFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter elementGetter;
+ private final SpecializedFunction.ExpressionEvaluator hashcodeEvaluator;
+ private final SpecializedFunction.ExpressionEvaluator equalityEvaluator;
+ private transient MethodHandle hashcodeHandle;
+ private transient MethodHandle equalityHandle;
+
+ 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());
+ hashcodeEvaluator =
+ context.createEvaluator(
+ Expressions.call("$HASHCODE$1", $("element1")),
+ DataTypes.INT(),
+ DataTypes.FIELD("element1", dataType.notNull().toInternal()));
+ equalityEvaluator =
+ context.createEvaluator(
+ $("element1").isEqual($("element2")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("element1", dataType.notNull().toInternal()),
+ DataTypes.FIELD("element2", dataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ hashcodeHandle = hashcodeEvaluator.open(context);
+ equalityHandle = equalityEvaluator.open(context);
+ }
+
+ public @Nullable ArrayData eval(ArrayData array1, ArrayData array2) {
+ try {
+ if (array1 == null || array2 == null) {
+ return null;
+ }
+ List<Object> list = new ArrayList<>();
+ Set<ObjectContainer> seen = new HashSet<>();
+
+ boolean isNullPresentInArrayTwo = false;
+ for (int pos = 0; pos < array2.size(); pos++) {
+ final Object element = elementGetter.getElementOrNull(array2, pos);
+ if (element == null) {
+ isNullPresentInArrayTwo = true;
+ } else {
+ ObjectContainer objectContainer = new ObjectContainer(element);
+ seen.add(objectContainer);
+ }
+ }
+ boolean isNullPresentInArrayOne = false;
+ for (int pos = 0; pos < array1.size(); pos++) {
+ final Object element = elementGetter.getElementOrNull(array1, pos);
+ if (element == null) {
+ isNullPresentInArrayOne = true;
Review Comment:
Do you mean NULL? We need to check whether both array1 and array2 contain NULL. If both contain NULL, we can include it in the list.
The description of array_intersect is like this
`Returns an array of the elements in the intersection of array1 and array2, without duplicates.`
--
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-31664][table]Add ARRAY_INTERSECT function [flink]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #23171:
URL: https://github.com/apache/flink/pull/23171#issuecomment-1938091186
@flinkbot run azure
--
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-31664][table]Add ARRAY_INTERSECT function [flink]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #23171:
URL: https://github.com/apache/flink/pull/23171#discussion_r1491941807
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ArrayIntersectFunction.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.api.Expressions;
+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.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#ARRAY_INTERSECT}. */
+@Internal
+public class ArrayIntersectFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter elementGetter;
+ private final SpecializedFunction.ExpressionEvaluator hashcodeEvaluator;
+ private final SpecializedFunction.ExpressionEvaluator equalityEvaluator;
+ private transient MethodHandle hashcodeHandle;
+ private transient MethodHandle equalityHandle;
+
+ 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());
+ hashcodeEvaluator =
+ context.createEvaluator(
+ Expressions.call("$HASHCODE$1", $("element1")),
+ DataTypes.INT(),
+ DataTypes.FIELD("element1", dataType.notNull().toInternal()));
+ equalityEvaluator =
+ context.createEvaluator(
+ $("element1").isEqual($("element2")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("element1", dataType.notNull().toInternal()),
+ DataTypes.FIELD("element2", dataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ hashcodeHandle = hashcodeEvaluator.open(context);
+ equalityHandle = equalityEvaluator.open(context);
+ }
+
+ public @Nullable ArrayData eval(ArrayData array1, ArrayData array2) {
+ try {
+ if (array1 == null || array2 == null) {
+ return null;
+ }
+ List<Object> list = new ArrayList<>();
+ Set<ObjectContainer> seen = new HashSet<>();
+
+ boolean isNullPresentInArrayTwo = false;
+ for (int pos = 0; pos < array2.size(); pos++) {
+ final Object element = elementGetter.getElementOrNull(array2, pos);
+ if (element == null) {
+ isNullPresentInArrayTwo = true;
+ } else {
+ ObjectContainer objectContainer = new ObjectContainer(element);
+ seen.add(objectContainer);
+ }
+ }
+ boolean isNullPresentInArrayOne = false;
+ for (int pos = 0; pos < array1.size(); pos++) {
+ final Object element = elementGetter.getElementOrNull(array1, pos);
+ if (element == null) {
+ isNullPresentInArrayOne = true;
+ } else {
+ ObjectContainer objectContainer = new ObjectContainer(element);
+ if (seen.contains(objectContainer)) {
+ list.add(element);
+ }
+ }
+ }
+ if (isNullPresentInArrayTwo && isNullPresentInArrayOne) {
Review Comment:
Yes, we can care about the order. I will change it after array_except merged.
--
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-31664][table]Add ARRAY_INTERSECT function [flink]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 closed pull request #23171: [FLINK-31664][table]Add ARRAY_INTERSECT function
URL: https://github.com/apache/flink/pull/23171
--
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-31664][table]Add ARRAY_INTERSECT function [flink]
Posted by "liuyongvs (via GitHub)" <gi...@apache.org>.
liuyongvs commented on PR #23171:
URL: https://github.com/apache/flink/pull/23171#issuecomment-2011038228
hi @dawidwys will you help review this https://github.com/apache/flink/pull/24526
--
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-31664][table]Add ARRAY_INTERSECT function [flink]
Posted by "liuyongvs (via GitHub)" <gi...@apache.org>.
liuyongvs commented on PR #23171:
URL: https://github.com/apache/flink/pull/23171#issuecomment-1956587783
@dawidwys great plan +1. because my prs do not have people to review. will you have time to help review them?
--
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-31664][table]Add ARRAY_INTERSECT function [flink]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #23171:
URL: https://github.com/apache/flink/pull/23171#issuecomment-1938842552
@flinkbot run azure
--
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-31664][table]Add ARRAY_INTERSECT function [flink]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #23171:
URL: https://github.com/apache/flink/pull/23171#discussion_r1491941807
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ArrayIntersectFunction.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.api.Expressions;
+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.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#ARRAY_INTERSECT}. */
+@Internal
+public class ArrayIntersectFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter elementGetter;
+ private final SpecializedFunction.ExpressionEvaluator hashcodeEvaluator;
+ private final SpecializedFunction.ExpressionEvaluator equalityEvaluator;
+ private transient MethodHandle hashcodeHandle;
+ private transient MethodHandle equalityHandle;
+
+ 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());
+ hashcodeEvaluator =
+ context.createEvaluator(
+ Expressions.call("$HASHCODE$1", $("element1")),
+ DataTypes.INT(),
+ DataTypes.FIELD("element1", dataType.notNull().toInternal()));
+ equalityEvaluator =
+ context.createEvaluator(
+ $("element1").isEqual($("element2")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("element1", dataType.notNull().toInternal()),
+ DataTypes.FIELD("element2", dataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ hashcodeHandle = hashcodeEvaluator.open(context);
+ equalityHandle = equalityEvaluator.open(context);
+ }
+
+ public @Nullable ArrayData eval(ArrayData array1, ArrayData array2) {
+ try {
+ if (array1 == null || array2 == null) {
+ return null;
+ }
+ List<Object> list = new ArrayList<>();
+ Set<ObjectContainer> seen = new HashSet<>();
+
+ boolean isNullPresentInArrayTwo = false;
+ for (int pos = 0; pos < array2.size(); pos++) {
+ final Object element = elementGetter.getElementOrNull(array2, pos);
+ if (element == null) {
+ isNullPresentInArrayTwo = true;
+ } else {
+ ObjectContainer objectContainer = new ObjectContainer(element);
+ seen.add(objectContainer);
+ }
+ }
+ boolean isNullPresentInArrayOne = false;
+ for (int pos = 0; pos < array1.size(); pos++) {
+ final Object element = elementGetter.getElementOrNull(array1, pos);
+ if (element == null) {
+ isNullPresentInArrayOne = true;
+ } else {
+ ObjectContainer objectContainer = new ObjectContainer(element);
+ if (seen.contains(objectContainer)) {
+ list.add(element);
+ }
+ }
+ }
+ if (isNullPresentInArrayTwo && isNullPresentInArrayOne) {
Review Comment:
Yes, we place the NULL values at the end of the List, similar to what the Array_Except function does. The order of the other elements in the List remains the same as in array1.
--
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-31664][table]Add ARRAY_INTERSECT function [flink]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #23171:
URL: https://github.com/apache/flink/pull/23171#discussion_r1491940565
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java:
##########
@@ -388,7 +388,16 @@ ANY, and(logical(LogicalTypeRoot.BOOLEAN), LITERAL)
.inputTypeStrategy(sequence(ANY))
.outputTypeStrategy(nullableIfArgs(explicit(DataTypes.INT().notNull())))
.runtimeProvided()
- .internal()
Review Comment:
It seems that I removed it during solve conflict. I will add it back.
--
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-31664][table]Add ARRAY_INTERSECT function [flink]
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on code in PR #23171:
URL: https://github.com/apache/flink/pull/23171#discussion_r1491059814
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ArrayIntersectFunction.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.api.Expressions;
+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.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#ARRAY_INTERSECT}. */
+@Internal
+public class ArrayIntersectFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter elementGetter;
+ private final SpecializedFunction.ExpressionEvaluator hashcodeEvaluator;
+ private final SpecializedFunction.ExpressionEvaluator equalityEvaluator;
+ private transient MethodHandle hashcodeHandle;
+ private transient MethodHandle equalityHandle;
+
+ 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());
+ hashcodeEvaluator =
+ context.createEvaluator(
+ Expressions.call("$HASHCODE$1", $("element1")),
+ DataTypes.INT(),
+ DataTypes.FIELD("element1", dataType.notNull().toInternal()));
+ equalityEvaluator =
+ context.createEvaluator(
+ $("element1").isEqual($("element2")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("element1", dataType.notNull().toInternal()),
+ DataTypes.FIELD("element2", dataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ hashcodeHandle = hashcodeEvaluator.open(context);
+ equalityHandle = equalityEvaluator.open(context);
+ }
+
+ public @Nullable ArrayData eval(ArrayData array1, ArrayData array2) {
+ try {
+ if (array1 == null || array2 == null) {
+ return null;
+ }
+ List<Object> list = new ArrayList<>();
+ Set<ObjectContainer> seen = new HashSet<>();
+
+ boolean isNullPresentInArrayTwo = false;
+ for (int pos = 0; pos < array2.size(); pos++) {
+ final Object element = elementGetter.getElementOrNull(array2, pos);
+ if (element == null) {
+ isNullPresentInArrayTwo = true;
+ } else {
+ ObjectContainer objectContainer = new ObjectContainer(element);
+ seen.add(objectContainer);
+ }
+ }
+ boolean isNullPresentInArrayOne = false;
+ for (int pos = 0; pos < array1.size(); pos++) {
+ final Object element = elementGetter.getElementOrNull(array1, pos);
+ if (element == null) {
+ isNullPresentInArrayOne = true;
Review Comment:
Can't we just check if it was present in the second array and add it to the `list`?
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java:
##########
@@ -388,7 +388,16 @@ ANY, and(logical(LogicalTypeRoot.BOOLEAN), LITERAL)
.inputTypeStrategy(sequence(ANY))
.outputTypeStrategy(nullableIfArgs(explicit(DataTypes.INT().notNull())))
.runtimeProvided()
- .internal()
Review Comment:
Why did you remove the `internal` here?
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ArrayIntersectFunction.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.api.Expressions;
+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.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#ARRAY_INTERSECT}. */
+@Internal
+public class ArrayIntersectFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter elementGetter;
+ private final SpecializedFunction.ExpressionEvaluator hashcodeEvaluator;
+ private final SpecializedFunction.ExpressionEvaluator equalityEvaluator;
+ private transient MethodHandle hashcodeHandle;
+ private transient MethodHandle equalityHandle;
+
+ 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());
+ hashcodeEvaluator =
+ context.createEvaluator(
+ Expressions.call("$HASHCODE$1", $("element1")),
+ DataTypes.INT(),
+ DataTypes.FIELD("element1", dataType.notNull().toInternal()));
+ equalityEvaluator =
+ context.createEvaluator(
+ $("element1").isEqual($("element2")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("element1", dataType.notNull().toInternal()),
+ DataTypes.FIELD("element2", dataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ hashcodeHandle = hashcodeEvaluator.open(context);
+ equalityHandle = equalityEvaluator.open(context);
+ }
+
+ public @Nullable ArrayData eval(ArrayData array1, ArrayData array2) {
+ try {
+ if (array1 == null || array2 == null) {
+ return null;
+ }
+ List<Object> list = new ArrayList<>();
+ Set<ObjectContainer> seen = new HashSet<>();
+
+ boolean isNullPresentInArrayTwo = false;
+ for (int pos = 0; pos < array2.size(); pos++) {
+ final Object element = elementGetter.getElementOrNull(array2, pos);
+ if (element == null) {
+ isNullPresentInArrayTwo = true;
+ } else {
+ ObjectContainer objectContainer = new ObjectContainer(element);
+ seen.add(objectContainer);
+ }
+ }
+ boolean isNullPresentInArrayOne = false;
+ for (int pos = 0; pos < array1.size(); pos++) {
+ final Object element = elementGetter.getElementOrNull(array1, pos);
+ if (element == null) {
+ isNullPresentInArrayOne = true;
+ } else {
+ ObjectContainer objectContainer = new ObjectContainer(element);
+ if (seen.contains(objectContainer)) {
+ list.add(element);
+ }
+ }
+ }
+ if (isNullPresentInArrayTwo && isNullPresentInArrayOne) {
Review Comment:
The same question as in the other PR. Shouldn't we care about the order within the first array?
--
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-31664][table]Add ARRAY_INTERSECT function [flink]
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on PR #23171:
URL: https://github.com/apache/flink/pull/23171#issuecomment-1956501651
Hey @liuyongvs I am sorry for the situation we ended up here. You're right we should've checked with you. I admit I have not been careful enough here. Please accept my apologies.
Let's try to solve it now somehow @liuyongvs @hanyuzheng7 @zzzzzzzs
1. How do you all feel if @liuyongvs takes care of `ARRAY_INTERSECT`? Both his and @hanyuzheng7 PRs are closest to be completed.
2. @hanyuzheng7 continues with `ARRAY_EXCEPT` which can reuse parts of the solution from `ARRAY_INTERSECT`
Do all fell okish with such a plan? I know it's not a perfect solution, but I want to find a way out of 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-31664][table]Add ARRAY_INTERSECT function [flink]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #23171:
URL: https://github.com/apache/flink/pull/23171#issuecomment-1957585378
@dawidwys @liuyongvs Ok
--
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-31664][table]Add ARRAY_INTERSECT function [flink]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #23171:
URL: https://github.com/apache/flink/pull/23171#discussion_r1491941807
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ArrayIntersectFunction.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.api.Expressions;
+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.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#ARRAY_INTERSECT}. */
+@Internal
+public class ArrayIntersectFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter elementGetter;
+ private final SpecializedFunction.ExpressionEvaluator hashcodeEvaluator;
+ private final SpecializedFunction.ExpressionEvaluator equalityEvaluator;
+ private transient MethodHandle hashcodeHandle;
+ private transient MethodHandle equalityHandle;
+
+ 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());
+ hashcodeEvaluator =
+ context.createEvaluator(
+ Expressions.call("$HASHCODE$1", $("element1")),
+ DataTypes.INT(),
+ DataTypes.FIELD("element1", dataType.notNull().toInternal()));
+ equalityEvaluator =
+ context.createEvaluator(
+ $("element1").isEqual($("element2")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("element1", dataType.notNull().toInternal()),
+ DataTypes.FIELD("element2", dataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ hashcodeHandle = hashcodeEvaluator.open(context);
+ equalityHandle = equalityEvaluator.open(context);
+ }
+
+ public @Nullable ArrayData eval(ArrayData array1, ArrayData array2) {
+ try {
+ if (array1 == null || array2 == null) {
+ return null;
+ }
+ List<Object> list = new ArrayList<>();
+ Set<ObjectContainer> seen = new HashSet<>();
+
+ boolean isNullPresentInArrayTwo = false;
+ for (int pos = 0; pos < array2.size(); pos++) {
+ final Object element = elementGetter.getElementOrNull(array2, pos);
+ if (element == null) {
+ isNullPresentInArrayTwo = true;
+ } else {
+ ObjectContainer objectContainer = new ObjectContainer(element);
+ seen.add(objectContainer);
+ }
+ }
+ boolean isNullPresentInArrayOne = false;
+ for (int pos = 0; pos < array1.size(); pos++) {
+ final Object element = elementGetter.getElementOrNull(array1, pos);
+ if (element == null) {
+ isNullPresentInArrayOne = true;
+ } else {
+ ObjectContainer objectContainer = new ObjectContainer(element);
+ if (seen.contains(objectContainer)) {
+ list.add(element);
+ }
+ }
+ }
+ if (isNullPresentInArrayTwo && isNullPresentInArrayOne) {
Review Comment:
Yes, we put the NULL at the end of the List like the Array_Except function does.
Other elements' order in the List are same as array1
--
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-31664][table]Add ARRAY_INTERSECT function [flink]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #23171:
URL: https://github.com/apache/flink/pull/23171#discussion_r1491954798
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ArrayIntersectFunction.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.api.Expressions;
+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.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#ARRAY_INTERSECT}. */
+@Internal
+public class ArrayIntersectFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter elementGetter;
+ private final SpecializedFunction.ExpressionEvaluator hashcodeEvaluator;
+ private final SpecializedFunction.ExpressionEvaluator equalityEvaluator;
+ private transient MethodHandle hashcodeHandle;
+ private transient MethodHandle equalityHandle;
+
+ 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());
+ hashcodeEvaluator =
+ context.createEvaluator(
+ Expressions.call("$HASHCODE$1", $("element1")),
+ DataTypes.INT(),
+ DataTypes.FIELD("element1", dataType.notNull().toInternal()));
+ equalityEvaluator =
+ context.createEvaluator(
+ $("element1").isEqual($("element2")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("element1", dataType.notNull().toInternal()),
+ DataTypes.FIELD("element2", dataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ hashcodeHandle = hashcodeEvaluator.open(context);
+ equalityHandle = equalityEvaluator.open(context);
+ }
+
+ public @Nullable ArrayData eval(ArrayData array1, ArrayData array2) {
+ try {
+ if (array1 == null || array2 == null) {
+ return null;
+ }
+ List<Object> list = new ArrayList<>();
+ Set<ObjectContainer> seen = new HashSet<>();
+
+ boolean isNullPresentInArrayTwo = false;
+ for (int pos = 0; pos < array2.size(); pos++) {
+ final Object element = elementGetter.getElementOrNull(array2, pos);
+ if (element == null) {
+ isNullPresentInArrayTwo = true;
+ } else {
+ ObjectContainer objectContainer = new ObjectContainer(element);
+ seen.add(objectContainer);
+ }
+ }
+ boolean isNullPresentInArrayOne = false;
+ for (int pos = 0; pos < array1.size(); pos++) {
+ final Object element = elementGetter.getElementOrNull(array1, pos);
+ if (element == null) {
+ isNullPresentInArrayOne = true;
Review Comment:
Do you mean NULL? We need to check whether both array1 and array2 contain NULL. If both contain NULL, we can include it in the list.
If we only verify the presence of NULL in array2, we would need to iterate through array2 again, incurring an additional O(n) in time complexity. Currently, we simply note whether NULL exists in both array1 and array2. If they both contain NULL, we append NULL to the end of the list.
--
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-31664][table]Add ARRAY_INTERSECT function [flink]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #23171:
URL: https://github.com/apache/flink/pull/23171#issuecomment-1938995805
@flinkbot run azure
--
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-31664][table]Add ARRAY_INTERSECT function [flink]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #23171:
URL: https://github.com/apache/flink/pull/23171#discussion_r1491941807
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ArrayIntersectFunction.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.api.Expressions;
+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.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#ARRAY_INTERSECT}. */
+@Internal
+public class ArrayIntersectFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter elementGetter;
+ private final SpecializedFunction.ExpressionEvaluator hashcodeEvaluator;
+ private final SpecializedFunction.ExpressionEvaluator equalityEvaluator;
+ private transient MethodHandle hashcodeHandle;
+ private transient MethodHandle equalityHandle;
+
+ 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());
+ hashcodeEvaluator =
+ context.createEvaluator(
+ Expressions.call("$HASHCODE$1", $("element1")),
+ DataTypes.INT(),
+ DataTypes.FIELD("element1", dataType.notNull().toInternal()));
+ equalityEvaluator =
+ context.createEvaluator(
+ $("element1").isEqual($("element2")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("element1", dataType.notNull().toInternal()),
+ DataTypes.FIELD("element2", dataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ hashcodeHandle = hashcodeEvaluator.open(context);
+ equalityHandle = equalityEvaluator.open(context);
+ }
+
+ public @Nullable ArrayData eval(ArrayData array1, ArrayData array2) {
+ try {
+ if (array1 == null || array2 == null) {
+ return null;
+ }
+ List<Object> list = new ArrayList<>();
+ Set<ObjectContainer> seen = new HashSet<>();
+
+ boolean isNullPresentInArrayTwo = false;
+ for (int pos = 0; pos < array2.size(); pos++) {
+ final Object element = elementGetter.getElementOrNull(array2, pos);
+ if (element == null) {
+ isNullPresentInArrayTwo = true;
+ } else {
+ ObjectContainer objectContainer = new ObjectContainer(element);
+ seen.add(objectContainer);
+ }
+ }
+ boolean isNullPresentInArrayOne = false;
+ for (int pos = 0; pos < array1.size(); pos++) {
+ final Object element = elementGetter.getElementOrNull(array1, pos);
+ if (element == null) {
+ isNullPresentInArrayOne = true;
+ } else {
+ ObjectContainer objectContainer = new ObjectContainer(element);
+ if (seen.contains(objectContainer)) {
+ list.add(element);
+ }
+ }
+ }
+ if (isNullPresentInArrayTwo && isNullPresentInArrayOne) {
Review Comment:
Yes, we place the NULL at the end of the List, similar to what the Array_Except function does. The order of the other elements in the List remains the same as in array1.
--
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-31664][table]Add ARRAY_INTERSECT function [flink]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #23171:
URL: https://github.com/apache/flink/pull/23171#discussion_r1491941807
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ArrayIntersectFunction.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.api.Expressions;
+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.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#ARRAY_INTERSECT}. */
+@Internal
+public class ArrayIntersectFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter elementGetter;
+ private final SpecializedFunction.ExpressionEvaluator hashcodeEvaluator;
+ private final SpecializedFunction.ExpressionEvaluator equalityEvaluator;
+ private transient MethodHandle hashcodeHandle;
+ private transient MethodHandle equalityHandle;
+
+ 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());
+ hashcodeEvaluator =
+ context.createEvaluator(
+ Expressions.call("$HASHCODE$1", $("element1")),
+ DataTypes.INT(),
+ DataTypes.FIELD("element1", dataType.notNull().toInternal()));
+ equalityEvaluator =
+ context.createEvaluator(
+ $("element1").isEqual($("element2")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("element1", dataType.notNull().toInternal()),
+ DataTypes.FIELD("element2", dataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ hashcodeHandle = hashcodeEvaluator.open(context);
+ equalityHandle = equalityEvaluator.open(context);
+ }
+
+ public @Nullable ArrayData eval(ArrayData array1, ArrayData array2) {
+ try {
+ if (array1 == null || array2 == null) {
+ return null;
+ }
+ List<Object> list = new ArrayList<>();
+ Set<ObjectContainer> seen = new HashSet<>();
+
+ boolean isNullPresentInArrayTwo = false;
+ for (int pos = 0; pos < array2.size(); pos++) {
+ final Object element = elementGetter.getElementOrNull(array2, pos);
+ if (element == null) {
+ isNullPresentInArrayTwo = true;
+ } else {
+ ObjectContainer objectContainer = new ObjectContainer(element);
+ seen.add(objectContainer);
+ }
+ }
+ boolean isNullPresentInArrayOne = false;
+ for (int pos = 0; pos < array1.size(); pos++) {
+ final Object element = elementGetter.getElementOrNull(array1, pos);
+ if (element == null) {
+ isNullPresentInArrayOne = true;
+ } else {
+ ObjectContainer objectContainer = new ObjectContainer(element);
+ if (seen.contains(objectContainer)) {
+ list.add(element);
+ }
+ }
+ }
+ if (isNullPresentInArrayTwo && isNullPresentInArrayOne) {
Review Comment:
Yes, we put the NULL at the end of the List like the Array_Except function does.
--
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-31664][table]Add ARRAY_INTERSECT function [flink]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #23171:
URL: https://github.com/apache/flink/pull/23171#discussion_r1491940565
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java:
##########
@@ -388,7 +388,16 @@ ANY, and(logical(LogicalTypeRoot.BOOLEAN), LITERAL)
.inputTypeStrategy(sequence(ANY))
.outputTypeStrategy(nullableIfArgs(explicit(DataTypes.INT().notNull())))
.runtimeProvided()
- .internal()
Review Comment:
It seems that I removed it during solving conflict. I will add it back.
--
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-31664][table]Add ARRAY_INTERSECT function [flink]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on PR #23171:
URL: https://github.com/apache/flink/pull/23171#issuecomment-1939193051
@flinkbot run azure
--
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-31664][table]Add ARRAY_INTERSECT function [flink]
Posted by "hanyuzheng7 (via GitHub)" <gi...@apache.org>.
hanyuzheng7 commented on code in PR #23171:
URL: https://github.com/apache/flink/pull/23171#discussion_r1491954798
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ArrayIntersectFunction.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.api.Expressions;
+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.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#ARRAY_INTERSECT}. */
+@Internal
+public class ArrayIntersectFunction extends BuiltInScalarFunction {
+ private final ArrayData.ElementGetter elementGetter;
+ private final SpecializedFunction.ExpressionEvaluator hashcodeEvaluator;
+ private final SpecializedFunction.ExpressionEvaluator equalityEvaluator;
+ private transient MethodHandle hashcodeHandle;
+ private transient MethodHandle equalityHandle;
+
+ 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());
+ hashcodeEvaluator =
+ context.createEvaluator(
+ Expressions.call("$HASHCODE$1", $("element1")),
+ DataTypes.INT(),
+ DataTypes.FIELD("element1", dataType.notNull().toInternal()));
+ equalityEvaluator =
+ context.createEvaluator(
+ $("element1").isEqual($("element2")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("element1", dataType.notNull().toInternal()),
+ DataTypes.FIELD("element2", dataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ hashcodeHandle = hashcodeEvaluator.open(context);
+ equalityHandle = equalityEvaluator.open(context);
+ }
+
+ public @Nullable ArrayData eval(ArrayData array1, ArrayData array2) {
+ try {
+ if (array1 == null || array2 == null) {
+ return null;
+ }
+ List<Object> list = new ArrayList<>();
+ Set<ObjectContainer> seen = new HashSet<>();
+
+ boolean isNullPresentInArrayTwo = false;
+ for (int pos = 0; pos < array2.size(); pos++) {
+ final Object element = elementGetter.getElementOrNull(array2, pos);
+ if (element == null) {
+ isNullPresentInArrayTwo = true;
+ } else {
+ ObjectContainer objectContainer = new ObjectContainer(element);
+ seen.add(objectContainer);
+ }
+ }
+ boolean isNullPresentInArrayOne = false;
+ for (int pos = 0; pos < array1.size(); pos++) {
+ final Object element = elementGetter.getElementOrNull(array1, pos);
+ if (element == null) {
+ isNullPresentInArrayOne = true;
Review Comment:
Do you mean NULL? We need to check whether both array1 and array2 contain NULL. If both contain NULL, we can include it in the list.
If we just check whether null exist in the array2, we have iterate array2 again, it will spend extra O(n).
Now, we just record whether array1 and array2 contain null, if they contain null, we add null in the `list `in the end.
--
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