[Optimizer] Eliminate the distinct

[jackwener]

Is your feature request related to a problem or challenge? Please describe what you are trying to do.

explain verbose select max(distinct(c1)) from test;

+-------------------------------------------------------+---------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type                                             | plan                                                                                                                                        |
+-------------------------------------------------------+---------------------------------------------------------------------------------------------------------------------------------------------+
| initial_logical_plan                                  | Projection: #MAX(DISTINCT test.c1)                                                                                                          |
|                                                       |   Aggregate: groupBy=[[]], aggr=[[MAX(DISTINCT #test.c1)]]                                                                                  |
|                                                       |     TableScan: test projection=None                                                                                                         |
| logical_plan after simplify_expressions               | SAME TEXT AS ABOVE                                                                                                                          |
| logical_plan after eliminate_filter                   | SAME TEXT AS ABOVE                                                                                                                          |
| logical_plan after common_sub_expression_eliminate    | SAME TEXT AS ABOVE                                                                                                                          |
| logical_plan after eliminate_limit                    | SAME TEXT AS ABOVE                                                                                                                          |
| logical_plan after projection_push_down               | Projection: #MAX(DISTINCT test.c1)                                                                                                          |
|                                                       |   Aggregate: groupBy=[[]], aggr=[[MAX(DISTINCT #test.c1)]]                                                                                  |
|                                                       |     TableScan: test projection=Some([0])                                                                                                    |
| logical_plan after filter_push_down                   | SAME TEXT AS ABOVE                                                                                                                          |
| logical_plan after limit_push_down                    | SAME TEXT AS ABOVE                                                                                                                          |
| logical_plan after SingleDistinctAggregationToGroupBy | Projection: #MAX(DISTINCT test.c1)                                                                                                          |
|                                                       |   Projection: #MAX(alias1) AS MAX(DISTINCT test.c1)                                                                                         |
|                                                       |     Aggregate: groupBy=[[]], aggr=[[MAX(#alias1)]]                                                                                          |
|                                                       |       Aggregate: groupBy=[[#test.c1 AS alias1]], aggr=[[]]                                                                                  |
|                                                       |         TableScan: test projection=Some([0]) 

Describe the solution you'd like
I think max/min don't need the rewrite, but need to eliminate the distinct.

Describe alternatives you've considered
SingleDistinctAggregationToGroupBy rule make the plan more complex.

Additional context
None

[jackwener]

I am investigating this issue. If you know some information about it, you can provide it.

[jiangzhx]

@jackwener check this issues #1315

@ic4y contribute this pr.

[jiangzhx]

talk with @jackwener and @ic4y ;
look like max and min function can not use this optimizer.
should add more test case

[jackwener]

During I rewrite the plan for eliminating the distinct, there are some problem I can't figure out. Need to wait to I learn more and then try again.

[jackwener]

Now the target is rewrite from

| initial_logical_plan | Projection: #Max(DISTINCT test.c1)                     
|                      |   Aggregate: groupBy=[[]], aggr=[[Max(DISTINCT #test.c1)]]                      
|                      |     TableScan: test projection=None

to

| initial_logical_plan | Projection: #Max(DISTINCT test.c1)
|                      |   Projection: #Max(#test.c1) AS Max(DISTINCT test.c1)                   
|                      |     Aggregate: groupBy=[[]], aggr=[[Max(#test.c1)]]                
|                      |       TableScan: test projection=None

instead of

| logical_plan  | Projection: #Max(DISTINCT test.c1)          
|               |   Projection: #Max(alias1) AS Max(DISTINCT test.c1)     
|               |     Aggregate: groupBy=[[]], aggr=[[Max(#alias1)]]        
|               |       Aggregate: groupBy=[[#test.c1 AS alias1]], aggr=[[]]                        
|               |       Aggregate: groupBy=[[#test.c1 AS alias1]], aggr=[[]]  
|               |         TableScan: test projection=Some([0])