[GitHub] [flink] twalthr commented on a diff in pull request #22030: [FLINK-31237][table] Fix possible bug of array_distinct.

["twalthr (via GitHub)" <gi...@apache.org>]

twalthr commented on code in PR #22030:
URL: https://github.com/apache/flink/pull/22030#discussion_r1118535783


##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ArrayDistinctFunction.java:
##########
@@ -19,46 +19,88 @@
 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.util.LinkedHashSet;
-import java.util.Set;
+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_DISTINCT}. */
 @Internal
 public class ArrayDistinctFunction extends BuiltInScalarFunction {
     private final ArrayData.ElementGetter elementGetter;
+    private final SpecializedFunction.ExpressionEvaluator equalityEvaluator;
+    private transient MethodHandle equalityHandle;
 
     public ArrayDistinctFunction(SpecializedFunction.SpecializedContext context) {
         super(BuiltInFunctionDefinitions.ARRAY_DISTINCT, context);
         final DataType dataType =
                 ((CollectionDataType) context.getCallContext().getArgumentDataTypes().get(0))
                         .getElementDataType();
         elementGetter = ArrayData.createElementGetter(dataType.getLogicalType());
+        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 {
+        equalityHandle = equalityEvaluator.open(context);
     }
 
     public @Nullable ArrayData eval(ArrayData haystack) {
         try {
             if (haystack == null) {
                 return null;
             }
-            Set set = new LinkedHashSet<>();
-            final int size = haystack.size();
-            for (int pos = 0; pos < size; pos++) {
-                final Object element = elementGetter.getElementOrNull(haystack, pos);
-                set.add(element);
+
+            List list = new ArrayList();
+            boolean alreadyStoredNull = false;
+
+            for (int i = 0; i < haystack.size(); i++) {
+                final Object element1 = elementGetter.getElementOrNull(haystack, i);
+                if (element1 != null) {
+                    boolean found = false;
+                    for (int j = 0; !found && j < list.size(); j++) {
+                        Object element2 = list.get(j);
+                        if (element2 != null) {
+                            found = (boolean) equalityHandle.invoke(element1, element2);
+                        }
+                    }
+                    if (!found) {
+                        list.add(element1);
+                    }
+                } else {
+                    // De-duplicate the null values.
+                    if (!alreadyStoredNull) {
+                        list.add(element1);
+                        alreadyStoredNull = true;
+                    }
+                }

Review Comment:
   agg functions are currently not supported by the expression evaluator, only scalar functions



-- 
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