planner: add missing exchange between hash-partition window and single-partition window

planner/core/exhaust_physical_plans.go

@@ -2291,7 +2291,7 @@ func (lw *LogicalWindow) tryToGetMppWindows(prop *property.PhysicalProperty) []P
 
 	if len(lw.PartitionBy) > 0 {
 		partitionCols := lw.GetPartitionKeys()
-		// trying to match the required parititions.
+		// trying to match the required partitions.
 		if prop.MPPPartitionTp == property.HashType {
 			if matches := prop.IsSubsetOf(partitionCols); len(matches) != 0 {
 				partitionCols = choosePartitionKeys(partitionCols, matches)
@@ -2306,6 +2306,10 @@ func (lw *LogicalWindow) tryToGetMppWindows(prop *property.PhysicalProperty) []P
 		childProperty.MPPPartitionTp = property.SinglePartitionType
 	}
 
+	if prop.MPPPartitionTp == property.SinglePartitionType && childProperty.MPPPartitionTp != property.SinglePartitionType {
+		return nil
+	}
+
 	window := PhysicalWindow{
 		WindowFuncDescs: lw.WindowFuncDescs,
 		PartitionBy:     lw.PartitionBy,
@@ -2596,7 +2600,7 @@ func (la *LogicalAggregation) tryToGetMppHashAggs(prop *property.PhysicalPropert
 
 	if len(la.GroupByItems) > 0 {
 		partitionCols := la.GetPotentialPartitionKeys()
-		// trying to match the required parititions.
+		// trying to match the required partitions.
 		if prop.MPPPartitionTp == property.HashType {
 			if matches := prop.IsSubsetOf(partitionCols); len(matches) != 0 {
 				partitionCols = choosePartitionKeys(partitionCols, matches)

planner/core/testdata/window_push_down_suite_in.json

@@ -46,7 +46,8 @@
       "explain select count(distinct empid) from (select *, row_number() over () from employee) t",
       "explain select count(distinct empid) from (select *, row_number() over (partition by deptid) from employee) t",
       "explain select count(empid) from (select *, row_number() over () a from employee) t group by a",
-      "explain select count(empid) from (select *, row_number() over (partition by deptid) a from employee) t group by a"
+      "explain select count(empid) from (select *, row_number() over (partition by deptid) a from employee) t group by a",
+      "explain select row_number() over w2, row_number() over w1 from employee window w2 as (order by deptid), w1 as (partition by deptid);"
     ]
   }
 ]

planner/core/testdata/window_push_down_suite_out.json

@@ -443,6 +443,24 @@
           "                  └─TableFullScan_18 10000.00 mpp[tiflash] table:employee keep order:false, stats:pseudo"
         ],
         "Warn": null
+      },
+      {
+        "SQL": "explain select row_number() over w2, row_number() over w1 from employee window w2 as (order by deptid), w1 as (partition by deptid);",
+        "Plan": [
+          "TableReader_37 10000.00 root  data:ExchangeSender_36",
+          "└─ExchangeSender_36 10000.00 mpp[tiflash]  ExchangeType: PassThrough",
+          "  └─Projection_10 10000.00 mpp[tiflash]  Column#8, Column#7",
+          "    └─Window_35 10000.00 mpp[tiflash]  row_number()->Column#8 over(order by test.employee.deptid rows between current row and current row)",
+          "      └─Sort_20 10000.00 mpp[tiflash]  test.employee.deptid",
+          "        └─ExchangeReceiver_19 10000.00 mpp[tiflash]  ",
+          "          └─ExchangeSender_18 10000.00 mpp[tiflash]  ExchangeType: PassThrough",
+          "            └─Window_13 10000.00 mpp[tiflash]  row_number()->Column#7 over(partition by test.employee.deptid rows between current row and current row), stream_count: 8",
+          "              └─Sort_17 10000.00 mpp[tiflash]  test.employee.deptid, stream_count: 8",
+          "                └─ExchangeReceiver_16 10000.00 mpp[tiflash]  stream_count: 8",
+          "                  └─ExchangeSender_15 10000.00 mpp[tiflash]  ExchangeType: HashPartition, Hash Cols: [name: test.employee.deptid, collate: binary], stream_count: 8",
+          "                    └─TableFullScan_14 10000.00 mpp[tiflash] table:employee keep order:false, stats:pseudo"
+        ],
+        "Warn": null
       }
     ]
   }