KYLIN-5117 Support percentile function after aggregate sub query (apa…

…che#1760)

* Support percentile function after aggregate sub query

* add test

kylin-it/src/test/resources/query/sql_percentile/query03.sql

@@ -0,0 +1,20 @@
+--
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements.  See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership.  The ASF licenses this file
+-- to you 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.
+--
+
+select percentile_approx(t.gmv, 0.5) from ( select seller_id,sum(price) as gmv from test_kylin_fact group by seller_id ) t
+;{"scanRowCount":9928,"scanBytes":0,"scanFiles":1,"cuboidId":[1310735]}

kylin-spark-project/kylin-spark-query/src/main/scala/org/apache/kylin/query/SchemaProcessor.scala

@@ -194,7 +194,7 @@ case class AggColumnInfo(
   hash: String,
   args: String*) {
   override def toString: String =
-    s"$funcName(${args.mkString("_")})_${index}_$hash"
+    s"${funcName}_${args.mkString("_")}__${index}_$hash"
 }
 
 case class TopNColumnInfo(tableName: String, columnId: Int, columnName: String)

kylin-spark-project/kylin-spark-query/src/main/scala/org/apache/kylin/query/runtime/plans/AggregatePlan.scala

@@ -18,15 +18,14 @@
 package org.apache.kylin.query.runtime.plans
 
 import org.apache.calcite.DataContext
-import org.apache.calcite.rel.core.Aggregate
-import org.apache.calcite.rel.core.AggregateCall
+import org.apache.calcite.rel.core.{Aggregate, AggregateCall}
 import org.apache.calcite.sql.SqlKind
 import org.apache.kylin.common.KylinConfig
 import org.apache.kylin.cube.CubeInstance
-import org.apache.kylin.metadata.model.{FunctionDesc, PartitionDesc, SegmentStatusEnum, TblColRef}
-import org.apache.kylin.query.relnode.{KylinAggregateCall, OLAPAggregateRel}
-import org.apache.kylin.query.runtime.RuntimeHelper
+import org.apache.kylin.metadata.model.{FunctionDesc, PartitionDesc, SegmentStatusEnum}
 import org.apache.kylin.query.SchemaProcessor
+import org.apache.kylin.query.relnode.{KylinAggregateCall, OLAPAggregateRel, OLAPRel}
+import org.apache.kylin.query.runtime.RuntimeHelper
 import org.apache.spark.sql.KylinFunctions._
 import org.apache.spark.sql._
 import org.apache.spark.sql.catalyst.expressions.{CreateArray, In}
@@ -217,6 +216,13 @@ object AggregatePlan extends LogEx {
             first(argNames.head).alias(aggName)
           case FunctionDesc.FUNC_GROUPING =>
             grouping(argNames.head).alias(aggName)
+          case FunctionDesc.FUNC_PERCENTILE => {
+            val col = argNames(0)
+            val inputColumnRowType = rel.getInput.asInstanceOf[OLAPRel].getColumnRowType
+            val percentage = inputColumnRowType.getColumnByIndex(call.getArgList.get(1)).getName
+            expr(s"approx_percentile($col, $percentage)").alias(aggName)
+          }
+
           case _ =>
             throw new IllegalArgumentException(
               s"""Unsupported function name $funcName""")