Backport #58739 to 23.8: Fix stream partitioning in parallel window functions

src/Interpreters/InterpreterSelectQuery.cpp

@@ -2837,7 +2837,15 @@ void InterpreterSelectQuery::executeWindow(QueryPlan & query_plan)
         // has suitable sorting. Also don't create sort steps when there are no
         // columns to sort by, because the sort nodes are confused by this. It
         // happens in case of `over ()`.
-        if (!window.full_sort_description.empty() && (i == 0 || !sortIsPrefix(window, *windows_sorted[i - 1])))
+        // Even if full_sort_description of both windows match, in case of different
+        // partitioning we need to add a SortingStep to reshuffle data in the streams.
+        bool need_sort = !window.full_sort_description.empty();
+        if (need_sort && i != 0)
+        {
+            need_sort = !sortIsPrefix(window, *windows_sorted[i - 1])
+                        || (settings.max_threads != 1 && window.partition_by.size() != windows_sorted[i - 1]->partition_by.size());
+        }
+        if (need_sort)
         {
             SortingStep::Settings sort_settings(*context);
 

src/Planner/Planner.cpp

@@ -803,9 +803,17 @@ void addWindowSteps(QueryPlan & query_plan,
           * has suitable sorting. Also don't create sort steps when there are no
           * columns to sort by, because the sort nodes are confused by this. It
           * happens in case of `over ()`.
+          * Even if full_sort_description of both windows match, in case of different
+          * partitioning we need to add a SortingStep to reshuffle data in the streams.
           */
-        if (!window_description.full_sort_description.empty() &&
-            (i == 0 || !sortDescriptionIsPrefix(window_description.full_sort_description, window_descriptions[i - 1].full_sort_description)))
+
+        bool need_sort = !window_description.full_sort_description.empty();
+        if (need_sort && i != 0)
+        {
+            need_sort = !sortDescriptionIsPrefix(window_description.full_sort_description, window_descriptions[i - 1].full_sort_description)
+                        || (settings.max_threads != 1 && window_description.partition_by.size() != window_descriptions[i - 1].partition_by.size());
+        }
+        if (need_sort)
         {
             SortingStep::Settings sort_settings(*query_context);
 

tests/queries/0_stateless/02962_parallel_window_functions_different_partitioning.reference

@@ -0,0 +1,18 @@
+sales	15000
+sales	15000
+sales	15000
+sales	29400
+sales	29400
+sales	29400
+sales	43800
+sales	43800
+sales	43800
+sales	15000	5000
+sales	15000	5000
+sales	15000	5000
+sales	29400	4800
+sales	29400	4800
+sales	29400	4800
+sales	43800	4800
+sales	43800	4800
+sales	43800	4800

tests/queries/0_stateless/02962_parallel_window_functions_different_partitioning.sql

@@ -0,0 +1,32 @@
+CREATE TABLE empsalary 
+(
+    `depname` LowCardinality(String),
+    `empno` UInt64,
+    `salary` Int32,
+    `enroll_date` Date
+)
+ENGINE = Memory;
+
+insert into empsalary values ('sales',3,4800,'2007-08-01'), ('sales',1,5000,'2006-10-01'), ('sales',4,4800,'2007-08-08');
+
+
+insert into empsalary values ('sales',3,4800,'2007-08-01'), ('sales',1,5000,'2006-10-01'), ('sales',4,4800,'2007-08-08');
+
+insert into empsalary values ('sales',3,4800,'2007-08-01'), ('sales',1,5000,'2006-10-01'), ('sales',4,4800,'2007-08-08');
+
+-- 1 window function
+
+SELECT depname,
+   sum(salary) OVER (PARTITION BY depname order by empno) AS depsalary
+FROM  empsalary
+order by depsalary;
+
+
+-- 2 window functions with different window,
+-- but result should be the same for depsalary
+
+SELECT depname,
+   sum(salary) OVER (PARTITION BY depname order by empno) AS depsalary,
+   min(salary) OVER (PARTITION BY depname, empno order by enroll_date) AS depminsalary
+FROM  empsalary
+order by depsalary;