fix: disable percentile for distributed tables (#1406)

## Rationale See #1405 ## Detailed Changes Disable percentile functions ## Test Plan CI
apache · Dec 27, 2023 · b3fd459 · b3fd459
1 parent 9974944
commit b3fd459
Show file tree

Hide file tree

Showing 3 changed files with 33 additions and 0 deletions.
diff --git a/df_engine_extensions/src/dist_sql_query/physical_plan.rs b/df_engine_extensions/src/dist_sql_query/physical_plan.rs
@@ -33,6 +33,7 @@ use datafusion::{
         coalesce_batches::CoalesceBatchesExec,
         coalesce_partitions::CoalescePartitionsExec,
         displayable,
+        expressions::{ApproxPercentileCont, ApproxPercentileContWithWeight},
         filter::FilterExec,
         metrics::{Count, MetricValue, MetricsSet},
         projection::ProjectionExec,
@@ -619,8 +620,20 @@ pub enum PushDownEvent {
 }
 
 impl PushDownEvent {
+    // Those aggregate functions can't be pushed down.
+    // https://github.com/apache/incubator-horaedb/issues/1405
+    fn blacklist_expr(expr: &dyn Any) -> bool {
+        expr.is::<ApproxPercentileCont>() || expr.is::<ApproxPercentileContWithWeight>()
+    }
+
     pub fn new(plan: Arc<dyn ExecutionPlan>) -> Self {
         if let Some(aggr) = plan.as_any().downcast_ref::<AggregateExec>() {
+            for aggr_expr in aggr.aggr_expr() {
+                if Self::blacklist_expr(aggr_expr.as_any()) {
+                    return Self::Unable;
+                }
+            }
+
             if *aggr.mode() == AggregateMode::Partial {
                 Self::Terminated(plan)
             } else {

diff --git a/integration_tests/cases/env/cluster/ddl/partition_table.result b/integration_tests/cases/env/cluster/ddl/partition_table.result
@@ -169,6 +169,18 @@ tsid,t,name,id,value,
 UInt64(12677620772014847982),Timestamp(1651737067000),String("ceresdb5"),Int32(0),Double(105.0),
 
 
+SELECT
+    time_bucket (t, "PT1M") AS ts,
+    approx_percentile_cont (value, 0.9) AS value
+FROM
+    random_partition_table_t
+GROUP BY
+    time_bucket (t, "PT1M");
+
+ts,value,
+Timestamp(1651737060000),Double(109.4),
+
+
 DROP TABLE IF EXISTS `random_partition_table_t`;
 
 affected_rows: 0

diff --git a/integration_tests/cases/env/cluster/ddl/partition_table.sql b/integration_tests/cases/env/cluster/ddl/partition_table.sql
@@ -82,6 +82,14 @@ SELECT * from random_partition_table_t where name = "ceresdb0";
 
 SELECT * from random_partition_table_t where name = "ceresdb5";
 
+SELECT
+    time_bucket (t, "PT1M") AS ts,
+    approx_percentile_cont (value, 0.9) AS value
+FROM
+    random_partition_table_t
+GROUP BY
+    time_bucket (t, "PT1M");
+
 DROP TABLE IF EXISTS `random_partition_table_t`;
 
 SHOW CREATE TABLE random_partition_table_t;