feat: Add RemoveInsensitiveAggregateDistinct rule

presto-built-in-worker-function-tools/src/main/java/com/facebook/presto/builtin/tools/WorkerFunctionUtil.java

@@ -62,7 +62,7 @@ public static synchronized SqlInvokedFunction createSqlInvokedFunction(String fu
                 jsonBasedUdfFunctionMetaData.getAggregateMetadata()
                         .map(metadata -> new AggregationFunctionMetadata(
                                 convertApplicableTypeToVariable(metadata.getIntermediateType()),
-                                metadata.isOrderSensitive()));
+                                metadata.isOrderSensitive(), metadata.isDistinctSensitive()));
 
         return new SqlInvokedFunction(
                 qualifiedFunctionName,

presto-function-namespace-managers-common/src/main/java/com/facebook/presto/functionNamespace/AbstractSqlInvokedFunctionNamespaceManager.java

@@ -367,6 +367,7 @@ protected AggregationFunctionImplementation sqlInvokedFunctionToAggregationImple
                         typeManager.getType(aggregationMetadata.getIntermediateType()),
                         typeManager.getType(function.getSignature().getReturnType()),
                         aggregationMetadata.isOrderSensitive(),
+                        aggregationMetadata.isDistinctSensitive(),
                         parameters);
             default:
                 throw new IllegalStateException(format("Unknown function implementation type: %s", implementationType));

presto-hive-function-namespace/src/main/java/com/facebook/presto/hive/functions/aggregation/HiveAggregationFunctionDescription.java

@@ -29,21 +29,24 @@ public class HiveAggregationFunctionDescription
     private final Type finalType;
     private final boolean decomposable;
     private final boolean orderSensitive;
+    private final boolean distinctSensitive;
 
     public HiveAggregationFunctionDescription(
             QualifiedObjectName name,
             List<Type> parameterTypes,
             List<Type> intermediateTypes,
             Type finalType,
             boolean decomposable,
-            boolean orderSensitive)
+            boolean orderSensitive,
+            boolean distinctSensitive)
     {
         this.name = requireNonNull(name);
         this.parameterTypes = requireNonNull(parameterTypes);
         this.intermediateTypes = requireNonNull(intermediateTypes);
         this.finalType = requireNonNull(finalType);
         this.decomposable = decomposable;
         this.orderSensitive = orderSensitive;
+        this.distinctSensitive = distinctSensitive;
     }
 
     public String getName()
@@ -75,4 +78,9 @@ public boolean isOrderSensitive()
     {
         return orderSensitive;
     }
+
+    public boolean isDistinctSensitive()
+    {
+        return distinctSensitive;
+    }
 }

presto-hive-function-namespace/src/main/java/com/facebook/presto/hive/functions/aggregation/HiveAggregationFunctionImplementation.java

@@ -107,6 +107,12 @@ public boolean isOrderSensitive()
         return aggregationFunctionDescription.isOrderSensitive();
     }
 
+    @Override
+    public boolean isDistinctSensitive()
+    {
+        return aggregationFunctionDescription.isDistinctSensitive();
+    }
+
     public AggregationMetadata getAggregationMetadata()
     {
         return aggregationMetadata;

presto-hive-function-namespace/src/main/java/com/facebook/presto/hive/functions/aggregation/HiveAggregationFunctionImplementationFactory.java

@@ -76,7 +76,8 @@ public HiveAggregationFunctionImplementation create()
                 ImmutableList.of(intermediateType),
                 outputType,
                 true,
-                false);
+                false,
+                true);
 
         HiveAccumulatorInvoker invocationContext = new HiveAccumulatorInvoker(
                 partialEvaluatorSupplier,

presto-main-base/src/main/java/com/facebook/presto/metadata/BuiltInSpecialFunctionNamespaceManager.java

@@ -253,6 +253,7 @@ private AggregationFunctionImplementation sqlInvokedFunctionToAggregationImpleme
                 typeManager.getType(aggregationMetadata.getIntermediateType()),
                 typeManager.getType(function.getSignature().getReturnType()),
                 aggregationMetadata.isOrderSensitive(),
+                aggregationMetadata.isDistinctSensitive(),
                 parameters);
     }
 

presto-main-base/src/main/java/com/facebook/presto/operator/aggregation/AbstractMinMaxAggregationFunction.java

@@ -163,7 +163,7 @@ else if (type.getJavaType() == boolean.class) {
                 metadata,
                 classLoader);
         return new BuiltInAggregationFunctionImplementation(getSignature().getNameSuffix(), inputTypes, ImmutableList.of(intermediateType),
-                type, true, false, metadata, accumulatorClass, groupedAccumulatorClass);
+                type, true, false, false, metadata, accumulatorClass, groupedAccumulatorClass);
     }
 
     protected AccumulatorStateSerializer<?> getStateSerializer(Class<? extends AccumulatorState> stateInterface, DynamicClassLoader classLoader)

presto-main-base/src/main/java/com/facebook/presto/operator/aggregation/AbstractMinMaxNAggregationFunction.java

@@ -117,7 +117,7 @@ protected BuiltInAggregationFunctionImplementation generateAggregation(Type type
                 metadata,
                 classLoader);
         return new BuiltInAggregationFunctionImplementation(getSignature().getNameSuffix(), inputTypes, ImmutableList.of(intermediateType), outputType,
-                true, false, metadata, accumulatorClass, groupedAccumulatorClass);
+                true, false, true, metadata, accumulatorClass, groupedAccumulatorClass);
     }
 
     public static void input(BlockComparator comparator, Type type, MinMaxNState state, Block block, long n, int blockIndex)

presto-main-base/src/main/java/com/facebook/presto/operator/aggregation/AggregationFromAnnotationsParser.java

@@ -142,6 +142,7 @@ private static AggregationHeader parseHeader(AnnotatedElement aggregationDefinit
                 parseDescription(aggregationDefinition),
                 aggregationAnnotation.decomposable(),
                 aggregationAnnotation.isOrderSensitive(),
+                aggregationAnnotation.isDistinctSensitive(),
                 aggregationAnnotation.visibility(),
                 aggregationAnnotation.isCalledOnNullInput());
     }
@@ -157,6 +158,7 @@ private static List<AggregationHeader> parseHeaders(AnnotatedElement aggregation
                                 parseDescription(aggregationDefinition, toParse),
                                 aggregationAnnotation.decomposable(),
                                 aggregationAnnotation.isOrderSensitive(),
+                                aggregationAnnotation.isDistinctSensitive(),
                                 aggregationAnnotation.visibility(),
                                 aggregationAnnotation.isCalledOnNullInput()))
                 .collect(toImmutableList());

presto-main-base/src/main/java/com/facebook/presto/operator/aggregation/AggregationHeader.java

@@ -25,15 +25,17 @@ public class AggregationHeader
     private final Optional<String> description;
     private final boolean decomposable;
     private final boolean orderSensitive;
+    private final boolean distinctSensitive;
     private final SqlFunctionVisibility visibility;
     private final boolean isCalledOnNullInput;
 
-    public AggregationHeader(String name, Optional<String> description, boolean decomposable, boolean orderSensitive, SqlFunctionVisibility visibility, boolean isCalledOnNullInput)
+    public AggregationHeader(String name, Optional<String> description, boolean decomposable, boolean orderSensitive, boolean distinctSensitive, SqlFunctionVisibility visibility, boolean isCalledOnNullInput)
     {
         this.name = requireNonNull(name, "name cannot be null");
         this.description = requireNonNull(description, "description cannot be null");
         this.decomposable = decomposable;
         this.orderSensitive = orderSensitive;
+        this.distinctSensitive = distinctSensitive;
         this.visibility = visibility;
         this.isCalledOnNullInput = isCalledOnNullInput;
     }
@@ -58,6 +60,11 @@ public boolean isOrderSensitive()
         return orderSensitive;
     }
 
+    public boolean isDistinctSensitive()
+    {
+        return distinctSensitive;
+    }
+
     public SqlFunctionVisibility getVisibility()
     {
         return visibility;

presto-main-base/src/main/java/com/facebook/presto/operator/aggregation/AlternativeApproxPercentile.java

@@ -120,7 +120,7 @@ public BuiltInAggregationFunctionImplementation specialize(
                             new RowType.Field(Optional.of("levels"), new ArrayType(IntegerType.INTEGER))));
             return new BuiltInAggregationFunctionImplementation(
                     NAME, inputTypes, ImmutableList.of(intermediateType), outputType,
-                    true, false, null, null, null);
+                    true, false, true, null, null, null);
         }
     }
 }

presto-main-base/src/main/java/com/facebook/presto/operator/aggregation/ArbitraryAggregationFunction.java

@@ -169,7 +169,7 @@ else if (type.getJavaType() == boolean.class) {
                 metadata,
                 classLoader);
         return new BuiltInAggregationFunctionImplementation(NAME, inputTypes, ImmutableList.of(intermediateType), type,
-                true, false, metadata, accumulatorClass, groupedAccumulatorClass);
+                true, false, false, metadata, accumulatorClass, groupedAccumulatorClass);
     }
 
     protected Type overrideIntermediateType(Type inputType, Type defaultIntermediateType)

presto-main-base/src/main/java/com/facebook/presto/operator/aggregation/BitwiseAndAggregation.java

@@ -24,7 +24,7 @@
 import com.facebook.presto.spi.function.OutputFunction;
 import com.facebook.presto.spi.function.SqlType;
 
-@AggregationFunction("bitwise_and_agg")
+@AggregationFunction(value = "bitwise_and_agg", isDistinctSensitive = false)
 public final class BitwiseAndAggregation
 {
     private BitwiseAndAggregation() {}

presto-main-base/src/main/java/com/facebook/presto/operator/aggregation/BitwiseOrAggregation.java

@@ -24,7 +24,7 @@
 import com.facebook.presto.spi.function.OutputFunction;
 import com.facebook.presto.spi.function.SqlType;
 
-@AggregationFunction("bitwise_or_agg")
+@AggregationFunction(value = "bitwise_or_agg", isDistinctSensitive = false)
 public class BitwiseOrAggregation
 {
     private BitwiseOrAggregation() {}

presto-main-base/src/main/java/com/facebook/presto/operator/aggregation/BooleanAndAggregation.java

@@ -28,7 +28,7 @@
 import static com.facebook.presto.operator.aggregation.state.TriStateBooleanState.NULL_VALUE;
 import static com.facebook.presto.operator.aggregation.state.TriStateBooleanState.TRUE_VALUE;
 
-@AggregationFunction(value = "bool_and", alias = "every")
+@AggregationFunction(value = "bool_and", alias = "every", isDistinctSensitive = false)
 public final class BooleanAndAggregation
 {
     private BooleanAndAggregation() {}

presto-main-base/src/main/java/com/facebook/presto/operator/aggregation/BooleanOrAggregation.java

@@ -28,7 +28,7 @@
 import static com.facebook.presto.operator.aggregation.state.TriStateBooleanState.NULL_VALUE;
 import static com.facebook.presto.operator.aggregation.state.TriStateBooleanState.TRUE_VALUE;
 
-@AggregationFunction(value = "bool_or")
+@AggregationFunction(value = "bool_or", isDistinctSensitive = false)
 public final class BooleanOrAggregation
 {
     private BooleanOrAggregation() {}

presto-main-base/src/main/java/com/facebook/presto/operator/aggregation/BuiltInAggregationFunctionImplementation.java

@@ -37,6 +37,7 @@ public final class BuiltInAggregationFunctionImplementation
     private final List<Class> lambdaInterfaces;
     private final boolean decomposable;
     private final boolean orderSensitive;
+    private final boolean distinctSensitive;
 
     private final AggregationMetadata aggregationMetadata;
 
@@ -51,6 +52,7 @@ public BuiltInAggregationFunctionImplementation(
             Type finalType,
             boolean decomposable,
             boolean orderSensitive,
+            boolean distinctSensitive,
             AggregationMetadata aggregationMetadata,
             Class<? extends Accumulator> accumulatorClass,
             Class<? extends GroupedAccumulator> groupedAccumulatorClass)
@@ -62,6 +64,7 @@ public BuiltInAggregationFunctionImplementation(
                 finalType,
                 decomposable,
                 orderSensitive,
+                distinctSensitive,
                 aggregationMetadata,
                 accumulatorClass,
                 groupedAccumulatorClass,
@@ -75,6 +78,7 @@ public BuiltInAggregationFunctionImplementation(
             Type finalType,
             boolean decomposable,
             boolean orderSensitive,
+            boolean distinctSensitive,
             AggregationMetadata aggregationMetadata,
             Class<? extends Accumulator> accumulatorClass,
             Class<? extends GroupedAccumulator> groupedAccumulatorClass,
@@ -87,6 +91,7 @@ public BuiltInAggregationFunctionImplementation(
         this.finalType = requireNonNull(finalType, "finalType is null");
         this.decomposable = decomposable;
         this.orderSensitive = orderSensitive;
+        this.distinctSensitive = distinctSensitive;
         this.aggregationMetadata = aggregationMetadata;
         this.accumulatorClass = accumulatorClass;
         this.groupedAccumulatorClass = groupedAccumulatorClass;
@@ -139,6 +144,11 @@ public boolean isOrderSensitive()
         return orderSensitive;
     }
 
+    public boolean isDistinctSensitive()
+    {
+        return distinctSensitive;
+    }
+
     public AggregationMetadata getAggregationMetadata()
     {
         return aggregationMetadata;

presto-main-base/src/main/java/com/facebook/presto/operator/aggregation/ChecksumAggregationFunction.java

@@ -106,7 +106,7 @@ private static BuiltInAggregationFunctionImplementation generateAggregation(Type
                 metadata,
                 classLoader);
         return new BuiltInAggregationFunctionImplementation(NAME, inputTypes, ImmutableList.of(BIGINT), VARBINARY,
-                true, false, metadata, accumulatorClass, groupedAccumulatorClass);
+                true, false, true, metadata, accumulatorClass, groupedAccumulatorClass);
     }
 
     private static List<ParameterMetadata> createInputParameterMetadata(Type type)

presto-main-base/src/main/java/com/facebook/presto/operator/aggregation/CountColumn.java

@@ -107,7 +107,7 @@ private static BuiltInAggregationFunctionImplementation generateAggregation(Type
                 metadata,
                 classLoader);
         return new BuiltInAggregationFunctionImplementation(NAME, inputTypes, ImmutableList.of(intermediateType), BIGINT,
-                true, false, metadata, accumulatorClass, groupedAccumulatorClass);
+                true, false, true, metadata, accumulatorClass, groupedAccumulatorClass);
     }
 
     private static List<ParameterMetadata> createInputParameterMetadata(Type type)

presto-main-base/src/main/java/com/facebook/presto/operator/aggregation/DecimalAverageAggregation.java

@@ -146,7 +146,7 @@ private static BuiltInAggregationFunctionImplementation generateAggregation(Type
                 metadata,
                 classLoader);
         return new BuiltInAggregationFunctionImplementation(NAME, inputTypes, ImmutableList.of(intermediateType), type,
-                true, false, metadata, accumulatorClass, groupedAccumulatorClass);
+                true, false, true, metadata, accumulatorClass, groupedAccumulatorClass);
     }
 
     private static List<ParameterMetadata> createInputParameterMetadata(Type type)

presto-main-base/src/main/java/com/facebook/presto/operator/aggregation/DecimalSumAggregation.java

@@ -135,7 +135,7 @@ private static BuiltInAggregationFunctionImplementation generateAggregation(Type
                 metadata,
                 classLoader);
         return new BuiltInAggregationFunctionImplementation(NAME, inputTypes, ImmutableList.of(intermediateType), outputType,
-                true, false, metadata, accumulatorClass, groupedAccumulatorClass);
+                true, false, true, metadata, accumulatorClass, groupedAccumulatorClass);
     }
 
     private static List<ParameterMetadata> createInputParameterMetadata(Type type)

presto-main-base/src/main/java/com/facebook/presto/operator/aggregation/MapAggregationFunction.java

@@ -111,7 +111,7 @@ private static BuiltInAggregationFunctionImplementation generateAggregation(Type
                 metadata,
                 classLoader);
         return new BuiltInAggregationFunctionImplementation(NAME, inputTypes, ImmutableList.of(intermediateType), outputType,
-                true, true, metadata, accumulatorClass, groupedAccumulatorClass);
+                true, true, true, metadata, accumulatorClass, groupedAccumulatorClass);
     }
 
     private static List<ParameterMetadata> createInputParameterMetadata(Type keyType, Type valueType)

presto-main-base/src/main/java/com/facebook/presto/operator/aggregation/MapUnionAggregation.java

@@ -104,7 +104,7 @@ private static BuiltInAggregationFunctionImplementation generateAggregation(Type
                 metadata,
                 classLoader);
         return new BuiltInAggregationFunctionImplementation(NAME, inputTypes, ImmutableList.of(intermediateType), outputType,
-                true, true, metadata, accumulatorClass, groupedAccumulatorClass);
+                true, true, true, metadata, accumulatorClass, groupedAccumulatorClass);
     }
 
     private static List<ParameterMetadata> createInputParameterMetadata(Type inputType)

presto-main-base/src/main/java/com/facebook/presto/operator/aggregation/MapUnionSumAggregation.java

@@ -110,7 +110,7 @@ private static BuiltInAggregationFunctionImplementation generateAggregation(Type
                 metadata,
                 classLoader);
         return new BuiltInAggregationFunctionImplementation(NAME, inputTypes, ImmutableList.of(intermediateType), outputType,
-                true, false, metadata, accumulatorClass, groupedAccumulatorClass);
+                true, false, true, metadata, accumulatorClass, groupedAccumulatorClass);
     }
 
     private static List<ParameterMetadata> createInputParameterMetadata(Type inputType)

presto-main-base/src/main/java/com/facebook/presto/operator/aggregation/MergeStatisticalDigestFunction.java

@@ -116,7 +116,7 @@ private BuiltInAggregationFunctionImplementation generateAggregation(Type type)
                 metadata,
                 classLoader);
         return new BuiltInAggregationFunctionImplementation(name, ImmutableList.of(type), ImmutableList.of(intermediateType), type,
-                true, true, metadata, accumulatorClass, groupedAccumulatorClass);
+                true, true, true, metadata, accumulatorClass, groupedAccumulatorClass);
     }
 
     private static List<ParameterMetadata> createInputParameterMetadata(Type valueType)

presto-main-base/src/main/java/com/facebook/presto/operator/aggregation/ParametricAggregation.java

@@ -127,6 +127,7 @@ public BuiltInAggregationFunctionImplementation specialize(BoundVariables variab
                 outputType,
                 details.isDecomposable(),
                 details.isOrderSensitive(),
+                details.isDistinctSensitive(),
                 metadata,
                 accumulatorClass,
                 groupedAccumulatorClass);

presto-main-base/src/main/java/com/facebook/presto/operator/aggregation/RealAverageAggregation.java

@@ -114,6 +114,7 @@ public BuiltInAggregationFunctionImplementation specialize(BoundVariables boundV
                 REAL,
                 true,
                 false,
+                true,
                 metadata,
                 accumulatorClass,
                 groupedAccumulatorClass);

presto-main-base/src/main/java/com/facebook/presto/operator/aggregation/ReduceAggregationFunction.java

@@ -143,6 +143,7 @@ private BuiltInAggregationFunctionImplementation generateAggregation(Type inputT
                 stateType,
                 true,
                 false,
+                true,
                 metadata,
                 accumulatorClass,
                 groupedAccumulatorClass,

presto-main-base/src/main/java/com/facebook/presto/operator/aggregation/StatisticalDigestAggregationFunction.java

@@ -161,7 +161,7 @@ private BuiltInAggregationFunctionImplementation generateAggregation(String name
                 metadata,
                 classLoader);
         return new BuiltInAggregationFunctionImplementation(name, inputTypes, ImmutableList.of(intermediateType), outputType,
-                true, true, metadata, accumulatorClass, groupedAccumulatorClass);
+                true, true, true, metadata, accumulatorClass, groupedAccumulatorClass);
     }
 
     private static List<Type> getInputTypes(Type valueType, int arity)

presto-main-base/src/main/java/com/facebook/presto/operator/aggregation/arrayagg/ArrayAggregationFunction.java

@@ -121,7 +121,7 @@ private static BuiltInAggregationFunctionImplementation generateAggregation(Type
                 classLoader);
 
         return new BuiltInAggregationFunctionImplementation(NAME, inputTypes, ImmutableList.of(intermediateType), outputType,
-                true, true, metadata, accumulatorClass, groupedAccumulatorClass);
+                true, true, true, metadata, accumulatorClass, groupedAccumulatorClass);
     }
 
     private static List<ParameterMetadata> createInputParameterMetadata(Type value, boolean legacyArrayAgg)

presto-main-base/src/main/java/com/facebook/presto/operator/aggregation/arrayagg/SetAggregationFunction.java

@@ -114,10 +114,8 @@ private static BuiltInAggregationFunctionImplementation generateAggregation(Type
                 GroupedAccumulator.class,
                 metadata,
                 classLoader);
-        return new
-
-                BuiltInAggregationFunctionImplementation(NAME, inputTypes, ImmutableList.of(intermediateType), outputType,
-                true, true, metadata, accumulatorClass, groupedAccumulatorClass);
+        return new BuiltInAggregationFunctionImplementation(NAME, inputTypes, ImmutableList.of(intermediateType), outputType,
+                true, true, false, metadata, accumulatorClass, groupedAccumulatorClass);
     }
 
     private static List<AggregationMetadata.ParameterMetadata> createInputParameterMetadata(Type valueType)

presto-main-base/src/main/java/com/facebook/presto/operator/aggregation/arrayagg/SetUnionFunction.java

@@ -115,10 +115,8 @@ private static BuiltInAggregationFunctionImplementation generateAggregation(Type
                 GroupedAccumulator.class,
                 metadata,
                 classLoader);
-        return new
-
-                BuiltInAggregationFunctionImplementation(NAME, inputTypes, ImmutableList.of(intermediateType), outputType,
-                true, true, metadata, accumulatorClass, groupedAccumulatorClass);
+        return new BuiltInAggregationFunctionImplementation(NAME, inputTypes, ImmutableList.of(intermediateType), outputType,
+                true, true, false, metadata, accumulatorClass, groupedAccumulatorClass);
     }
 
     private static List<AggregationMetadata.ParameterMetadata> createInputParameterMetadata(Type valueType)