feat: Add RemoveInsensitiveAggregateDistinct optimizer rule

Author: xin-zhang2

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;