planner: support MaxOneRow clone (pingcap#33888)

close pingcap#33887

executor/join.go

@@ -855,6 +855,7 @@ func (e *NestedLoopApplyExec) Close() error {
 		} else {
 			runtimeStats.setCacheInfo(false, 0)
 		}
+		runtimeStats.SetConcurrencyInfo(execdetails.NewConcurrencyInfo("Concurrency", 0))
 	}
 	return e.outerExec.Close()
 }

planner/core/optimizer.go

@@ -414,6 +414,8 @@ func enableParallelApply(sctx sessionctx.Context, plan PhysicalPlan) PhysicalPla
 		supportClone := err == nil // limitation 2
 		if noOrder && supportClone {
 			apply.Concurrency = sctx.GetSessionVars().ExecutorConcurrency
+		} else {
+			sctx.GetSessionVars().StmtCtx.AppendWarning(errors.Errorf("Some apply operators can not be executed in parallel"))
 		}
 
 		// because of the limitation 3, we cannot parallelize Apply operators in this Apply's inner size,

planner/core/physical_plans.go

@@ -1297,6 +1297,17 @@ type PhysicalMaxOneRow struct {
 	basePhysicalPlan
 }
 
+// Clone implements PhysicalPlan interface.
+func (p *PhysicalMaxOneRow) Clone() (PhysicalPlan, error) {
+	cloned := new(PhysicalMaxOneRow)
+	base, err := p.basePhysicalPlan.cloneWithSelf(cloned)
+	if err != nil {
+		return nil, err
+	}
+	cloned.basePhysicalPlan = *base
+	return cloned, nil
+}
+
 // PhysicalTableDual is the physical operator of dual.
 type PhysicalTableDual struct {
 	physicalSchemaProducer

planner/core/planbuilder_test.go

@@ -287,6 +287,11 @@ func TestPhysicalPlanClone(t *testing.T) {
 	sel = sel.Init(ctx, stats, 0)
 	require.NoError(t, checkPhysicalPlanClone(sel))
 
+	// maxOneRow
+	maxOneRow := &PhysicalMaxOneRow{}
+	maxOneRow = maxOneRow.Init(ctx, stats, 0)
+	require.NoError(t, checkPhysicalPlanClone(maxOneRow))
+
 	// projection
 	proj := &PhysicalProjection{Exprs: []expression.Expression{col, cst}}
 	proj = proj.Init(ctx, stats, 0)