fix(query): fix distinct set-returning function (#16883)

* fix(query): fix distinct set-returning function

* add tests

* fix

* fix tests

---------

Co-authored-by: sundyli <[email protected]>

Author: b41sh

src/query/sql/src/planner/binder/bind_query/bind_select.rs

@@ -225,7 +225,7 @@ impl Binder {
         if stmt.distinct {
             s_expr = self.bind_distinct(
                 stmt.span,
-                &from_context,
+                &mut from_context,
                 &projections,
                 &mut scalar_items,
                 s_expr,

src/query/sql/src/planner/binder/distinct.rs

@@ -18,6 +18,7 @@ use std::sync::Arc;
 use databend_common_ast::Span;
 use databend_common_exception::Result;
 
+use crate::binder::project_set::SetReturningRewriter;
 use crate::binder::Binder;
 use crate::binder::ColumnBinding;
 use crate::optimizer::SExpr;
@@ -37,11 +38,19 @@ impl Binder {
     pub fn bind_distinct(
         &self,
         span: Span,
-        bind_context: &BindContext,
+        bind_context: &mut BindContext,
         projections: &[ColumnBinding],
         scalar_items: &mut HashMap<IndexType, ScalarItem>,
         child: SExpr,
     ) -> Result<SExpr> {
+        if !bind_context.srf_info.srfs.is_empty() {
+            // Rewrite the Set-returning functions as columns.
+            let mut srf_rewriter = SetReturningRewriter::new(bind_context, false);
+            for (_, item) in scalar_items.iter_mut() {
+                srf_rewriter.visit(&mut item.scalar)?;
+            }
+        }
+
         let scalar_items: Vec<ScalarItem> = scalar_items
             .drain()
             .map(|(_, item)| {

src/query/sql/src/planner/binder/select.rs

@@ -296,10 +296,11 @@ impl Binder {
         );
 
         if distinct {
+            let columns = new_bind_context.all_column_bindings().to_vec();
             new_expr = self.bind_distinct(
                 left_span,
-                &new_bind_context,
-                new_bind_context.all_column_bindings(),
+                &mut new_bind_context,
+                &columns,
                 &mut HashMap::new(),
                 new_expr,
             )?;
@@ -359,10 +360,11 @@ impl Binder {
         right_expr: SExpr,
         join_type: JoinType,
     ) -> Result<(SExpr, BindContext)> {
+        let columns = left_context.all_column_bindings().to_vec();
         let left_expr = self.bind_distinct(
             left_span,
-            &left_context,
-            left_context.all_column_bindings(),
+            &mut left_context,
+            &columns,
             &mut HashMap::new(),
             left_expr,
         )?;

src/query/sql/src/planner/dataframe.rs

@@ -313,7 +313,7 @@ impl Dataframe {
         )?;
         self.s_expr = self.binder.bind_distinct(
             None,
-            &self.bind_context,
+            &mut self.bind_context,
             &projections,
             &mut scalar_items,
             self.s_expr.clone(),

tests/sqllogictests/suites/base/03_common/03_0003_select_group_by.test

@@ -268,6 +268,13 @@ a1
 a2
 a3
 
+query T
+SELECT distinct unnest(split(col2, ',')) AS col3 FROM t_str ORDER BY col3;
+----
+a1
+a2
+a3
+
 statement ok
 DROP TABLE t_str
 

tests/sqllogictests/suites/query/functions/02_0062_function_unnest.test

@@ -503,6 +503,12 @@ select max(unnest([11,12]))
 ----
 12
 
+query T
+SELECT distinct unnest(split(coalesce(NULL, 'a,b'), ',')) AS c1
+----
+a
+b
+
 statement error 1065
 select unnest(first_value('aa') OVER (PARTITION BY 'bb'))