Fix approx_percentile to have constant pct for input rows (prestodb#24600)

Author: yuandagits

presto-main/src/main/java/com/facebook/presto/operator/aggregation/ApproximateLongPercentileAggregations.java

@@ -92,16 +92,15 @@ private static void addInput(
             checkAccuracy(accuracy);
             digest = new QuantileDigest(accuracy);
             state.setDigest(digest);
+            state.setPercentile(percentile);
         }
         else {
             state.addMemoryUsage(-digest.estimatedInMemorySizeInBytes());
         }
 
+        checkPercentile(percentile, state.getPercentile());
         digest.add(value, weight);
         state.addMemoryUsage(digest.estimatedInMemorySizeInBytes());
-
-        // use last percentile
-        state.setPercentile(percentile);
     }
 
     @CombineFunction
@@ -137,6 +136,11 @@ public static void output(@AggregationState DigestAndPercentileState state, Bloc
         }
     }
 
+    static void checkPercentile(double percentile, double statePercentile)
+    {
+        checkCondition(percentile == statePercentile, INVALID_FUNCTION_ARGUMENT, "Percentile argument must be constant for all input rows: %s vs. %s", percentile, statePercentile);
+    }
+
     static void checkAccuracy(double accuracy)
     {
         checkCondition(0 < accuracy && accuracy < 1, INVALID_FUNCTION_ARGUMENT, "Percentile accuracy must be strictly between 0 and 1");

presto-main/src/test/java/com/facebook/presto/operator/aggregation/TestApproximatePercentileAggregation.java

@@ -19,6 +19,7 @@
 import com.facebook.presto.common.type.Type;
 import com.facebook.presto.metadata.FunctionAndTypeManager;
 import com.facebook.presto.metadata.MetadataManager;
+import com.facebook.presto.spi.PrestoException;
 import com.facebook.presto.spi.function.JavaAggregationFunctionImplementation;
 import com.google.common.collect.ImmutableList;
 import org.testng.annotations.Test;
@@ -495,6 +496,16 @@ public void testDoublePartialStep()
                 createRLEBlock(ImmutableList.of(0.5, 0.8), 3));
     }
 
+    @Test(expectedExceptions = PrestoException.class, expectedExceptionsMessageRegExp = "Percentile argument must be constant for all input rows: 0.3 vs. 0.1")
+    public void testNonConstantPercentile()
+    {
+        assertAggregation(
+                DOUBLE_APPROXIMATE_PERCENTILE_AGGREGATION,
+                null,
+                createDoublesBlock(1.0, 2.0, 3.0),
+                createDoublesBlock(0.1, 0.3, 0.5));
+    }
+
     private static JavaAggregationFunctionImplementation getAggregation(Type... arguments)
     {
         return FUNCTION_AND_TYPE_MANAGER.getJavaAggregateFunctionImplementation(FUNCTION_AND_TYPE_MANAGER.lookupFunction("approx_percentile", fromTypes(arguments)));