/*
 * Licensed 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 com.facebook.presto.operator.scalar;

import com.facebook.presto.metadata.FunctionInfo;
import com.facebook.presto.metadata.FunctionRegistry;
import com.facebook.presto.metadata.ParametricScalar;
import com.facebook.presto.metadata.Signature;
import com.facebook.presto.operator.GroupByHash;
import com.facebook.presto.spi.Page;
import com.facebook.presto.spi.PageBuilder;
import com.facebook.presto.spi.block.Block;
import com.facebook.presto.spi.block.BlockBuilder;
import com.facebook.presto.spi.block.BlockBuilderStatus;
import com.facebook.presto.spi.block.VariableWidthBlockBuilder;
import com.facebook.presto.spi.type.Type;
import com.facebook.presto.spi.type.TypeManager;
import com.google.common.collect.ImmutableList;
import io.airlift.slice.Slice;

import java.lang.invoke.MethodHandle;
import java.util.Map;
import java.util.Optional;

import static com.facebook.presto.metadata.Signature.comparableTypeParameter;
import static com.facebook.presto.operator.GroupByHash.createGroupByHash;
import static com.facebook.presto.type.TypeUtils.buildStructuralSlice;
import static com.facebook.presto.type.TypeUtils.parameterizedTypeName;
import static com.facebook.presto.type.TypeUtils.readStructuralBlock;
import static com.facebook.presto.util.Reflection.methodHandle;
import static com.google.common.base.Preconditions.checkArgument;
import static java.lang.String.format;

public final class ArrayDistinctFunction
        extends ParametricScalar
{
    public static final ArrayDistinctFunction ARRAY_DISTINCT_FUNCTION = new ArrayDistinctFunction();
    private static final String FUNCTION_NAME = "array_distinct";
    private static final Signature SIGNATURE = new Signature(FUNCTION_NAME, ImmutableList.of(comparableTypeParameter("E")), "array<E>", ImmutableList.of("array<E>"), false, false);
    private static final MethodHandle METHOD_HANDLE = methodHandle(ArrayDistinctFunction.class, "distinct", Type.class, Slice.class);

    @Override
    public Signature getSignature()
    {
        return SIGNATURE;
    }

    @Override
    public boolean isHidden()
    {
        return false;
    }

    @Override
    public boolean isDeterministic()
    {
        return true;
    }

    @Override
    public String getDescription()
    {
        return "Remove duplicate values from the given array";
    }

    @Override
    public FunctionInfo specialize(Map<String, Type> types, int arity, TypeManager typeManager, FunctionRegistry functionRegistry)
    {
        checkArgument(types.size() == 1, format("%s expects only one argument", FUNCTION_NAME));
        Type type = types.get("E");
        MethodHandle methodHandle = METHOD_HANDLE.bindTo(type);
        Signature signature = new Signature(FUNCTION_NAME,
                parameterizedTypeName("array", type.getTypeSignature()),
                parameterizedTypeName("array", type.getTypeSignature()));
        return new FunctionInfo(signature, getDescription(), isHidden(), methodHandle, isDeterministic(), false, ImmutableList.of(false));
    }

    public static Slice distinct(Type type, Slice array)
    {
        Block elementsBlock = readStructuralBlock(array);

        if (elementsBlock.getPositionCount() == 0) {
            return array;
        }

        GroupByHash groupByHash = createGroupByHash(ImmutableList.of(type), new int[] {0}, Optional.<Integer>empty(), Optional.empty(), elementsBlock.getPositionCount());
        groupByHash.getGroupIds(new Page(elementsBlock));

        PageBuilder pageBuilder = new PageBuilder(groupByHash.getTypes());
        for (int i = 0; i < groupByHash.getGroupCount(); i++) {
            pageBuilder.declarePosition();
            groupByHash.appendValuesTo(i, pageBuilder, 0);
        }
        BlockBuilder fixedWidthResultBlockBuilder = pageBuilder.getBlockBuilder(0);
        Block fixedWidthResultBlock = fixedWidthResultBlockBuilder.build();

        // Convert the fixed width block to a variable width block because of the limitations of array representation
        BlockBuilder variableWidthBlockBuilder = new VariableWidthBlockBuilder(new BlockBuilderStatus(), fixedWidthResultBlock.getSizeInBytes());
        for (int i = 0; i < fixedWidthResultBlock.getPositionCount(); i++) {
            type.appendTo(fixedWidthResultBlock, i, variableWidthBlockBuilder);
        }

        return buildStructuralSlice(variableWidthBlockBuilder);
    }
}

@@ -842,27 +842,42 @@
 perator.
-GroupByHash
+aggregation.SimpleTypedSet
 ;%0Aimport
@@ -901,60 +901,37 @@
 sto.
-spi.Page;%0Aimport com.facebook.presto.spi.PageBuilder
+operator.aggregation.TypedSet
 ;%0Aim
@@ -1378,35 +1378,8 @@
 Map;
-%0Aimport java.util.Optional;
 %0A%0Aim
@@ -1458,82 +1458,8 @@
 er;%0A
-import static com.facebook.presto.operator.GroupByHash.createGroupByHash;%0A
 impo
@@ -3595,17 +3595,16 @@
 array);%0A
-%0A
         
@@ -3697,1048 +3697,466 @@
     
-GroupByHash groupByHash = createGroupByHash(ImmutableList.of(type), new int%5B%5D %7B0%7D, Optional.%3CInteger%3Eempty(), Optional.empty(), elementsBlock.getPositionCount());%0A        groupByHash.getGroupIds(new Page(elementsBlock));%0A%0A        PageBuilder pageBuilder = new PageBuilder(groupByHash.getTypes());%0A        for (int i = 0; i %3C groupByHash.getGroupCount(); i++) %7B%0A            pageBuilder.declarePosition();%0A            groupByHash.appendValuesTo(i, pageBuilder, 0);%0A        %7D%0A        BlockBuilder fixedWidthResultBlockBuilder = pageBuilder.getBlockBuilder(0);%0A        Block fixedWidthResultBlock = fixedWidthResultBlockBuilder.build();%0A%0A        // Convert the fixed width block to a variable width block because of the limitations of array representation%0A        BlockBuilder variableWidthBlockBuilder = new VariableWidthBlockBuilder(new BlockBuilderStatus(), fixedWidthResultBlock.getSizeInBytes());%0A        for (int i = 0; i %3C fixedWidthResultBlock.getPositionCount(); i++) %7B%0A            type.appendTo(fixedWidthResultBlock, i, variableWidth
+TypedSet typedSet = new SimpleTypedSet(type, elementsBlock.getPositionCount());%0A        BlockBuilder distinctElementBlockBuilder = new VariableWidthBlockBuilder(new BlockBuilderStatus(), elementsBlock.getPositionCount());%0A        for (int i = 0; i %3C elementsBlock.getPositionCount(); i++) %7B%0A            if (!typedSet.contains(elementsBlock, i)) %7B%0A                typedSet.add(elementsBlock, i);%0A                type.appendTo(elementsBlock, i, distinctElement
 Bloc
@@ -4162,24 +4162,38 @@
 ckBuilder);%0A
+            %7D%0A
         %7D%0A%0A 
@@ -4231,21 +4231,23 @@
 ice(
-variableWidth
+distinctElement
 Bloc

Optimize array_distinct memory usage