Fix for PERCENTILES query result in csv output (opendistro-for-elasti…

…csearch#37)

Issue: NLPchina/elastcsearch-sql#734

Details: If the query specified percentiles and the output was
requested in CSV format, the default set of percentiles was always
returned and the specified percentiles were ignored. This change fixes
that.

Test: Tested manually on local cluster + added integ tests to cover this
use case.

src/main/java/com/amazon/opendistroforelasticsearch/sql/executor/csv/CSVResultsExtractor.java

@@ -25,6 +25,7 @@
 import org.elasticsearch.search.aggregations.bucket.SingleBucketAggregation;
 import org.elasticsearch.search.aggregations.metrics.NumericMetricsAggregation;
 import org.elasticsearch.search.aggregations.metrics.geobounds.GeoBounds;
+import org.elasticsearch.search.aggregations.metrics.percentiles.Percentile;
 import org.elasticsearch.search.aggregations.metrics.percentiles.Percentiles;
 import org.elasticsearch.search.aggregations.metrics.stats.Stats;
 import org.elasticsearch.search.aggregations.metrics.stats.extended.ExtendedStats;
@@ -164,19 +165,20 @@ private  List<String> fillHeaderAndCreateLineForNumericAggregations(Aggregations
         return line;
     }
 
-    private  void handleNumericMetricAggregation(List<String> header, List<String> line, Aggregation aggregation) throws CsvExtractorException {
-        String name = aggregation.getName();
+    private  void handleNumericMetricAggregation(List<String> header, List<String> line, Aggregation aggregation)
+                                                                                    throws CsvExtractorException {
+        final String name = aggregation.getName();
 
-        if(aggregation instanceof NumericMetricsAggregation.SingleValue){
-            if(!header.contains(name)){
+        if (aggregation instanceof NumericMetricsAggregation.SingleValue){
+            if (!header.contains(name)){
                 header.add(name);
             }
             NumericMetricsAggregation.SingleValue agg = (NumericMetricsAggregation.SingleValue) aggregation;
             line.add(!Double.isInfinite(agg.value()) ? agg.getValueAsString() : "null");
         }
         //todo:Numeric MultiValue - Stats,ExtendedStats,Percentile...
-        else if(aggregation instanceof NumericMetricsAggregation.MultiValue){
-            if(aggregation instanceof Stats) {
+        else if (aggregation instanceof NumericMetricsAggregation.MultiValue){
+            if (aggregation instanceof Stats) {
                 String[] statsHeaders = new String[]{"count", "sum", "avg", "min", "max"};
                 boolean isExtendedStats = aggregation instanceof ExtendedStats;
                 if(isExtendedStats){
@@ -197,17 +199,16 @@ else if(aggregation instanceof NumericMetricsAggregation.MultiValue){
                     line.add(extendedStats.getStdDeviationAsString());
                 }
             }
-            else if( aggregation instanceof Percentiles){
-                String[] percentileHeaders = new String[]{"1.0", "5.0", "25.0", "50.0", "75.0", "95.0", "99.0"};
-                mergeHeadersWithPrefix(header, name, percentileHeaders);
-                Percentiles percentiles = (Percentiles) aggregation;
-                line.add(percentiles.percentileAsString(1.0));
-                line.add(percentiles.percentileAsString(5.0));
-                line.add(percentiles.percentileAsString(25.0));
-                line.add(percentiles.percentileAsString(50.0));
-                line.add(percentiles.percentileAsString(75));
-                line.add(percentiles.percentileAsString(95.0));
-                line.add(percentiles.percentileAsString(99.0));
+            else if (aggregation instanceof Percentiles){
+
+                final List<String> percentileHeaders = new ArrayList<>(7);
+                final Percentiles percentiles = (Percentiles)aggregation;
+
+                for (final Percentile p : percentiles) {
+                    percentileHeaders.add(String.valueOf(p.getPercent()));
+                    line.add(percentiles.percentileAsString(p.getPercent()));
+                }
+                mergeHeadersWithPrefix(header, name, percentileHeaders.toArray(new String[0]));
             }
             else {
                 throw new CsvExtractorException("unknown NumericMetricsAggregation.MultiValue:" + aggregation.getClass());

src/main/java/com/amazon/opendistroforelasticsearch/sql/query/maker/AggMaker.java

@@ -153,6 +153,8 @@ private void addSpecificPercentiles(PercentilesAggregationBuilder percentilesBui
                 BigDecimal percentile = (BigDecimal) kValue.value;
                 percentiles.add(percentile.doubleValue());
 
+            } else if (kValue.value instanceof Integer) {
+                percentiles.add(((Integer)kValue.value).doubleValue());
             }
         }
         if (percentiles.size() > 0) {

src/test/java/com/amazon/opendistroforelasticsearch/sql/esintgtest/CsvFormatResponseIT.java

@@ -0,0 +1,72 @@
+/*
+ *   Copyright 2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *   Licensed under the Apache License, Version 2.0 (the "License").
+ *   You may not use this file except in compliance with the License.
+ *   A copy of the License is located at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *   or in the "license" file accompanying this file. This file 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.amazon.opendistroforelasticsearch.sql.esintgtest;
+
+import org.elasticsearch.client.Request;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Locale;
+
+import static com.amazon.opendistroforelasticsearch.sql.esintgtest.TestsConstants.TEST_INDEX_ACCOUNT;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.not;
+
+/**
+ * Tests to cover requests with "?format=csv" parameter
+ */
+public class CsvFormatResponseIT extends SQLIntegTestCase {
+
+    @Override
+    protected void init() throws Exception {
+        loadIndex(Index.ACCOUNT);
+    }
+
+    @Override
+    protected Request getSqlRequest(String request, boolean explain) {
+
+        Request sqlRequest = super.getSqlRequest(request, explain);
+        sqlRequest.addParameter("format", "csv");
+        return sqlRequest;
+    }
+
+    @Test
+    public void allPercentilesByDefault() throws IOException  {
+
+        final String query = String.format(Locale.ROOT, "SELECT PERCENTILES(age) FROM %s", TEST_INDEX_ACCOUNT);
+        final String result = executeQueryWithStringOutput(query);
+
+        final String expectedHeaders = "PERCENTILES(age).1.0,PERCENTILES(age).5.0,PERCENTILES(age).25.0," +
+                "PERCENTILES(age).50.0,PERCENTILES(age).75.0,PERCENTILES(age).95.0,PERCENTILES(age).99.0";
+        Assert.assertThat(result, containsString(expectedHeaders));
+    }
+
+    @Test
+    public void specificPercentilesIntAndDouble() throws IOException  {
+
+        final String query = String.format(Locale.ROOT, "SELECT PERCENTILES(age,10,49.0) FROM %s",
+                TEST_INDEX_ACCOUNT);
+        final String result = executeQueryWithStringOutput(query);
+
+        final String[] unexpectedPercentiles = {"1.0", "5.0", "25.0", "50.0", "75.0", "95.0", "99.0"};
+        final String expectedHeaders = "PERCENTILES(age,10,49.0).10.0,PERCENTILES(age,10,49.0).49.0";
+        Assert.assertThat(result, containsString(expectedHeaders));
+        for (final String unexpectedPercentile : unexpectedPercentiles) {
+            Assert.assertThat(result, not(containsString("PERCENTILES(age,10,49.0)." + unexpectedPercentile)));
+        }
+    }
+}

src/test/java/com/amazon/opendistroforelasticsearch/sql/esintgtest/SQLIntegTestCase.java

@@ -111,6 +111,12 @@ protected String explainQuery(final String sqlQuery) throws IOException {
         return executeExplainRequest(requestBody);
     }
 
+    protected String executeQueryWithStringOutput(final String sqlQuery) throws IOException {
+
+        final String requestString = makeRequest(sqlQuery);
+        return executeRequest(requestString, false);
+    }
+
     protected JSONObject executeRequest(final String requestBody) throws IOException {
 
         return new JSONObject(executeRequest(requestBody, false));