From c74a2330c03faceb14b0ba081a4a93716964c9af Mon Sep 17 00:00:00 2001
From: Arenatlx <314806019@qq.com>
Date: Thu, 15 Aug 2024 13:13:11 +0800
Subject: [PATCH] planner: move logical union all to logicalop pkg. (#55402)

ref pingcap/tidb#51664, ref pingcap/tidb#52714
---
 pkg/planner/cascades/implementation_rules.go  |  2 +-
 pkg/planner/cascades/transformation_rules.go  |  6 ++--
 pkg/planner/core/BUILD.bazel                  |  2 --
 pkg/planner/core/casetest/stats_test.go       |  2 +-
 .../core/collect_column_stats_usage.go        |  6 ++--
 pkg/planner/core/core_init.go                 |  2 ++
 pkg/planner/core/exhaust_physical_plans.go    |  8 +++--
 pkg/planner/core/expression_rewriter.go       |  2 +-
 pkg/planner/core/logical_plan_builder.go      |  6 ++--
 pkg/planner/core/logical_plans.go             |  3 +-
 pkg/planner/core/logical_plans_test.go        |  2 +-
 .../core/operator/logicalop/BUILD.bazel       |  2 ++
 .../logicalop}/logical_partition_union_all.go |  8 ++---
 .../logicalop}/logical_union_all.go           | 31 +++++++++++++------
 .../core/rule_aggregation_push_down.go        |  8 ++---
 pkg/planner/core/rule_eliminate_projection.go |  2 +-
 pkg/planner/core/rule_partition_processor.go  |  4 +--
 pkg/planner/core/rule_topn_push_down.go       | 11 -------
 pkg/planner/core/stringer.go                  |  4 +--
 pkg/planner/pattern/pattern.go                |  2 +-
 pkg/planner/pattern/pattern_test.go           |  2 +-
 .../util/utilfuncp/func_pointer_misc.go       |  8 +++++
 22 files changed, 68 insertions(+), 55 deletions(-)
 rename pkg/planner/core/{ => operator/logicalop}/logical_partition_union_all.go (86%)
 rename pkg/planner/core/{ => operator/logicalop}/logical_union_all.go (88%)

diff --git a/pkg/planner/cascades/implementation_rules.go b/pkg/planner/cascades/implementation_rules.go
index 42683c9a19967..53f300301c4e6 100644
--- a/pkg/planner/cascades/implementation_rules.go
+++ b/pkg/planner/cascades/implementation_rules.go
@@ -529,7 +529,7 @@ func (*ImplUnionAll) Match(_ *memo.GroupExpr, prop *property.PhysicalProperty) (
 
 // OnImplement implements ImplementationRule OnImplement interface.
 func (*ImplUnionAll) OnImplement(expr *memo.GroupExpr, reqProp *property.PhysicalProperty) ([]memo.Implementation, error) {
-	logicalUnion := expr.ExprNode.(*plannercore.LogicalUnionAll)
+	logicalUnion := expr.ExprNode.(*logicalop.LogicalUnionAll)
 	chReqProps := make([]*property.PhysicalProperty, len(expr.Children))
 	for i := range expr.Children {
 		chReqProps[i] = &property.PhysicalProperty{ExpectedCnt: reqProp.ExpectedCnt}
diff --git a/pkg/planner/cascades/transformation_rules.go b/pkg/planner/cascades/transformation_rules.go
index 69ca39762d7f7..a18cbc55ef449 100644
--- a/pkg/planner/cascades/transformation_rules.go
+++ b/pkg/planner/cascades/transformation_rules.go
@@ -826,7 +826,7 @@ func (r *PushLimitDownUnionAll) Match(expr *memo.ExprIter) bool {
 // It will transform `Limit->UnionAll->X` to `Limit->UnionAll->Limit->X`.
 func (r *PushLimitDownUnionAll) OnTransform(old *memo.ExprIter) (newExprs []*memo.GroupExpr, eraseOld bool, eraseAll bool, err error) {
 	limit := old.GetExpr().ExprNode.(*logicalop.LogicalLimit)
-	unionAll := old.Children[0].GetExpr().ExprNode.(*plannercore.LogicalUnionAll)
+	unionAll := old.Children[0].GetExpr().ExprNode.(*logicalop.LogicalUnionAll)
 	unionAllSchema := old.Children[0].Group.Prop.Schema
 
 	newLimit := logicalop.LogicalLimit{
@@ -1086,7 +1086,7 @@ func NewRulePushSelDownUnionAll() Transformation {
 // It will transform `Selection->UnionAll->x` to `UnionAll->Selection->x`.
 func (*PushSelDownUnionAll) OnTransform(old *memo.ExprIter) (newExprs []*memo.GroupExpr, eraseOld bool, eraseAll bool, err error) {
 	sel := old.GetExpr().ExprNode.(*logicalop.LogicalSelection)
-	unionAll := old.Children[0].GetExpr().ExprNode.(*plannercore.LogicalUnionAll)
+	unionAll := old.Children[0].GetExpr().ExprNode.(*logicalop.LogicalUnionAll)
 	childGroups := old.Children[0].GetExpr().Children
 
 	newUnionAllExpr := memo.NewGroupExpr(unionAll)
@@ -1365,7 +1365,7 @@ func (r *PushTopNDownUnionAll) Match(expr *memo.ExprIter) bool {
 // It will transform `TopN->UnionAll->X` to `TopN->UnionAll->TopN->X`.
 func (r *PushTopNDownUnionAll) OnTransform(old *memo.ExprIter) (newExprs []*memo.GroupExpr, eraseOld bool, eraseAll bool, err error) {
 	topN := old.GetExpr().ExprNode.(*logicalop.LogicalTopN)
-	unionAll := old.Children[0].GetExpr().ExprNode.(*plannercore.LogicalUnionAll)
+	unionAll := old.Children[0].GetExpr().ExprNode.(*logicalop.LogicalUnionAll)
 
 	newTopN := logicalop.LogicalTopN{
 		Count:   topN.Count + topN.Offset,
diff --git a/pkg/planner/core/BUILD.bazel b/pkg/planner/core/BUILD.bazel
index 27e669884b614..0be151392f4df 100644
--- a/pkg/planner/core/BUILD.bazel
+++ b/pkg/planner/core/BUILD.bazel
@@ -27,12 +27,10 @@ go_library(
         "logical_expand.go",
         "logical_index_scan.go",
         "logical_initialize.go",
-        "logical_partition_union_all.go",
         "logical_plan_builder.go",
         "logical_plans.go",
         "logical_table_scan.go",
         "logical_tikv_single_gather.go",
-        "logical_union_all.go",
         "memtable_infoschema_extractor.go",
         "memtable_predicate_extractor.go",
         "mock.go",
diff --git a/pkg/planner/core/casetest/stats_test.go b/pkg/planner/core/casetest/stats_test.go
index 88dd89a5d9e60..10a81f642d31c 100644
--- a/pkg/planner/core/casetest/stats_test.go
+++ b/pkg/planner/core/casetest/stats_test.go
@@ -84,7 +84,7 @@ func TestGroupNDVs(t *testing.T) {
 			case *logicalop.LogicalApply:
 				lp = lp.Children()[0]
 				stack = append(stack, v.Children()[1])
-			case *core.LogicalUnionAll:
+			case *logicalop.LogicalUnionAll:
 				lp = lp.Children()[0]
 				for i := 1; i < len(v.Children()); i++ {
 					stack = append(stack, v.Children()[i])
diff --git a/pkg/planner/core/collect_column_stats_usage.go b/pkg/planner/core/collect_column_stats_usage.go
index 89cb1c7388e87..f1d7462565b11 100644
--- a/pkg/planner/core/collect_column_stats_usage.go
+++ b/pkg/planner/core/collect_column_stats_usage.go
@@ -162,7 +162,7 @@ func (c *columnStatsUsageCollector) collectPredicateColumnsForJoin(p *logicalop.
 	c.addPredicateColumnsFromExpressions(exprs)
 }
 
-func (c *columnStatsUsageCollector) collectPredicateColumnsForUnionAll(p *LogicalUnionAll) {
+func (c *columnStatsUsageCollector) collectPredicateColumnsForUnionAll(p *logicalop.LogicalUnionAll) {
 	// statistics of the ith column of UnionAll come from statistics of the ith column of each child.
 	schemas := make([]*expression.Schema, 0, len(p.Children()))
 	relatedCols := make([]*expression.Column, 0, len(p.Children()))
@@ -289,9 +289,9 @@ func (c *columnStatsUsageCollector) collectFromPlan(lp base.LogicalPlan) {
 			for _, item := range x.ByItems {
 				c.addPredicateColumnsFromExpressions([]expression.Expression{item.Expr})
 			}
-		case *LogicalUnionAll:
+		case *logicalop.LogicalUnionAll:
 			c.collectPredicateColumnsForUnionAll(x)
-		case *LogicalPartitionUnionAll:
+		case *logicalop.LogicalPartitionUnionAll:
 			c.collectPredicateColumnsForUnionAll(&x.LogicalUnionAll)
 		case *LogicalCTE:
 			// Visit seedPartLogicalPlan and recursivePartLogicalPlan first.
diff --git a/pkg/planner/core/core_init.go b/pkg/planner/core/core_init.go
index 0c0e63323cd02..718a6b9437305 100644
--- a/pkg/planner/core/core_init.go
+++ b/pkg/planner/core/core_init.go
@@ -46,12 +46,14 @@ func init() {
 	utilfuncp.ExhaustPhysicalPlans4LogicalApply = exhaustPhysicalPlans4LogicalApply
 	utilfuncp.ExhaustPhysicalPlans4LogicalLimit = exhaustPhysicalPlans4LogicalLimit
 	utilfuncp.ExhaustPhysicalPlans4LogicalWindow = exhaustPhysicalPlans4LogicalWindow
+	utilfuncp.ExhaustPhysicalPlans4LogicalUnionAll = exhaustPhysicalPlans4LogicalUnionAll
 	utilfuncp.ExhaustPhysicalPlans4LogicalSequence = exhaustPhysicalPlans4LogicalSequence
 	utilfuncp.ExhaustPhysicalPlans4LogicalSelection = exhaustPhysicalPlans4LogicalSelection
 	utilfuncp.ExhaustPhysicalPlans4LogicalMaxOneRow = exhaustPhysicalPlans4LogicalMaxOneRow
 	utilfuncp.ExhaustPhysicalPlans4LogicalUnionScan = exhaustPhysicalPlans4LogicalUnionScan
 	utilfuncp.ExhaustPhysicalPlans4LogicalProjection = exhaustPhysicalPlans4LogicalProjection
 	utilfuncp.ExhaustPhysicalPlans4LogicalAggregation = exhaustPhysicalPlans4LogicalAggregation
+	utilfuncp.ExhaustPhysicalPlans4LogicalPartitionUnionAll = exhaustPhysicalPlans4LogicalPartitionUnionAll
 
 	utilfuncp.GetActualProbeCntFromProbeParents = getActualProbeCntFromProbeParents
 	utilfuncp.GetEstimatedProbeCntFromProbeParents = getEstimatedProbeCntFromProbeParents
diff --git a/pkg/planner/core/exhaust_physical_plans.go b/pkg/planner/core/exhaust_physical_plans.go
index 4916531a38877..d6ee33080748f 100644
--- a/pkg/planner/core/exhaust_physical_plans.go
+++ b/pkg/planner/core/exhaust_physical_plans.go
@@ -2446,7 +2446,7 @@ func canPushToCopImpl(lp base.LogicalPlan, storeTp kv.StoreType, considerDual bo
 				// Once aggregation is pushed to cop, the cache data can't be use any more.
 				return false
 			}
-		case *LogicalUnionAll:
+		case *logicalop.LogicalUnionAll:
 			if storeTp != kv.TiFlash {
 				return false
 			}
@@ -2951,7 +2951,8 @@ func exhaustPhysicalPlans4LogicalLock(lp base.LogicalPlan, prop *property.Physic
 	return []base.PhysicalPlan{lock}, true, nil
 }
 
-func exhaustUnionAllPhysicalPlans(p *LogicalUnionAll, prop *property.PhysicalProperty) ([]base.PhysicalPlan, bool, error) {
+func exhaustPhysicalPlans4LogicalUnionAll(lp base.LogicalPlan, prop *property.PhysicalProperty) ([]base.PhysicalPlan, bool, error) {
+	p := lp.(*logicalop.LogicalUnionAll)
 	// TODO: UnionAll can not pass any order, but we can change it to sort merge to keep order.
 	if !prop.IsSortItemEmpty() || (prop.IsFlashProp() && prop.TaskTp != property.MppTaskType) {
 		return nil, true, nil
@@ -2995,7 +2996,8 @@ func exhaustUnionAllPhysicalPlans(p *LogicalUnionAll, prop *property.PhysicalPro
 	return []base.PhysicalPlan{ua}, true, nil
 }
 
-func exhaustPartitionUnionAllPhysicalPlans(p *LogicalPartitionUnionAll, prop *property.PhysicalProperty) ([]base.PhysicalPlan, bool, error) {
+func exhaustPhysicalPlans4LogicalPartitionUnionAll(lp base.LogicalPlan, prop *property.PhysicalProperty) ([]base.PhysicalPlan, bool, error) {
+	p := lp.(*logicalop.LogicalPartitionUnionAll)
 	uas, flagHint, err := p.LogicalUnionAll.ExhaustPhysicalPlans(prop)
 	if err != nil {
 		return nil, false, err
diff --git a/pkg/planner/core/expression_rewriter.go b/pkg/planner/core/expression_rewriter.go
index 7a1b14d28a702..c1e126b19817c 100644
--- a/pkg/planner/core/expression_rewriter.go
+++ b/pkg/planner/core/expression_rewriter.go
@@ -2457,7 +2457,7 @@ func (er *expressionRewriter) toColumn(v *ast.ColumnName) {
 			return
 		}
 	}
-	if _, ok := planCtx.plan.(*LogicalUnionAll); ok && v.Table.O != "" {
+	if _, ok := planCtx.plan.(*logicalop.LogicalUnionAll); ok && v.Table.O != "" {
 		er.err = plannererrors.ErrTablenameNotAllowedHere.GenWithStackByArgs(v.Table.O, "SELECT", clauseMsg[planCtx.builder.curClause])
 		return
 	}
diff --git a/pkg/planner/core/logical_plan_builder.go b/pkg/planner/core/logical_plan_builder.go
index 5561f0ab90977..d6628099b10db 100644
--- a/pkg/planner/core/logical_plan_builder.go
+++ b/pkg/planner/core/logical_plan_builder.go
@@ -1571,7 +1571,7 @@ func (b *PlanBuilder) setUnionFlen(resultTp *types.FieldType, cols []expression.
 	}
 }
 
-func (b *PlanBuilder) buildProjection4Union(_ context.Context, u *LogicalUnionAll) error {
+func (b *PlanBuilder) buildProjection4Union(_ context.Context, u *logicalop.LogicalUnionAll) error {
 	unionCols := make([]*expression.Column, 0, u.Children()[0].Schema().Len())
 	names := make([]*types.FieldName, 0, u.Children()[0].Schema().Len())
 
@@ -1894,7 +1894,7 @@ func (b *PlanBuilder) buildUnionAll(ctx context.Context, subPlan []base.LogicalP
 	if len(subPlan) == 0 {
 		return nil, nil
 	}
-	u := LogicalUnionAll{}.Init(b.ctx, b.getSelectOffset())
+	u := logicalop.LogicalUnionAll{}.Init(b.ctx, b.getSelectOffset())
 	u.SetChildren(subPlan...)
 	err := b.buildProjection4Union(ctx, u)
 	return u, err
@@ -1921,7 +1921,7 @@ func (b *PlanBuilder) buildSort(ctx context.Context, p base.LogicalPlan, byItems
 
 func (b *PlanBuilder) buildSortWithCheck(ctx context.Context, p base.LogicalPlan, byItems []*ast.ByItem, aggMapper map[*ast.AggregateFuncExpr]int, windowMapper map[*ast.WindowFuncExpr]int,
 	projExprs []expression.Expression, oldLen int, hasDistinct bool) (*logicalop.LogicalSort, error) {
-	if _, isUnion := p.(*LogicalUnionAll); isUnion {
+	if _, isUnion := p.(*logicalop.LogicalUnionAll); isUnion {
 		b.curClause = globalOrderByClause
 	} else {
 		b.curClause = orderByClause
diff --git a/pkg/planner/core/logical_plans.go b/pkg/planner/core/logical_plans.go
index f0021cbb20b0b..d7b08397a4bbc 100644
--- a/pkg/planner/core/logical_plans.go
+++ b/pkg/planner/core/logical_plans.go
@@ -31,7 +31,8 @@ var (
 	_ base.LogicalPlan = &TiKVSingleGather{}
 	_ base.LogicalPlan = &LogicalTableScan{}
 	_ base.LogicalPlan = &LogicalIndexScan{}
-	_ base.LogicalPlan = &LogicalUnionAll{}
+	_ base.LogicalPlan = &logicalop.LogicalUnionAll{}
+	_ base.LogicalPlan = &logicalop.LogicalPartitionUnionAll{}
 	_ base.LogicalPlan = &logicalop.LogicalSort{}
 	_ base.LogicalPlan = &logicalop.LogicalLock{}
 	_ base.LogicalPlan = &logicalop.LogicalLimit{}
diff --git a/pkg/planner/core/logical_plans_test.go b/pkg/planner/core/logical_plans_test.go
index dbb9ff3a0db7c..fbb05bf3974ee 100644
--- a/pkg/planner/core/logical_plans_test.go
+++ b/pkg/planner/core/logical_plans_test.go
@@ -781,7 +781,7 @@ func TestAllocID(t *testing.T) {
 func checkDataSourceCols(p base.LogicalPlan, t *testing.T, ans map[int][]string, comment string) {
 	ectx := p.SCtx().GetExprCtx().GetEvalCtx()
 	switch v := p.(type) {
-	case *DataSource, *LogicalUnionAll, *logicalop.LogicalLimit:
+	case *DataSource, *logicalop.LogicalUnionAll, *logicalop.LogicalLimit:
 		testdata.OnRecord(func() {
 			ans[p.ID()] = make([]string, p.Schema().Len())
 		})
diff --git a/pkg/planner/core/operator/logicalop/BUILD.bazel b/pkg/planner/core/operator/logicalop/BUILD.bazel
index 86286d42dafc3..f53c23229914e 100644
--- a/pkg/planner/core/operator/logicalop/BUILD.bazel
+++ b/pkg/planner/core/operator/logicalop/BUILD.bazel
@@ -12,6 +12,7 @@ go_library(
         "logical_lock.go",
         "logical_max_one_row.go",
         "logical_mem_table.go",
+        "logical_partition_union_all.go",
         "logical_projection.go",
         "logical_schema_producer.go",
         "logical_selection.go",
@@ -21,6 +22,7 @@ go_library(
         "logical_sort.go",
         "logical_table_dual.go",
         "logical_top_n.go",
+        "logical_union_all.go",
         "logical_union_scan.go",
         "logical_window.go",
     ],
diff --git a/pkg/planner/core/logical_partition_union_all.go b/pkg/planner/core/operator/logicalop/logical_partition_union_all.go
similarity index 86%
rename from pkg/planner/core/logical_partition_union_all.go
rename to pkg/planner/core/operator/logicalop/logical_partition_union_all.go
index dc77954ff9d30..095abee2bd5b3 100644
--- a/pkg/planner/core/logical_partition_union_all.go
+++ b/pkg/planner/core/operator/logicalop/logical_partition_union_all.go
@@ -12,12 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package core
+package logicalop
 
 import (
 	"github.com/pingcap/tidb/pkg/planner/core/base"
-	"github.com/pingcap/tidb/pkg/planner/core/operator/logicalop"
 	"github.com/pingcap/tidb/pkg/planner/property"
+	"github.com/pingcap/tidb/pkg/planner/util/utilfuncp"
 	"github.com/pingcap/tidb/pkg/util/plancodec"
 )
 
@@ -28,7 +28,7 @@ type LogicalPartitionUnionAll struct {
 
 // Init initializes LogicalPartitionUnionAll.
 func (p LogicalPartitionUnionAll) Init(ctx base.PlanContext, offset int) *LogicalPartitionUnionAll {
-	p.BaseLogicalPlan = logicalop.NewBaseLogicalPlan(ctx, plancodec.TypePartitionUnion, &p, offset)
+	p.BaseLogicalPlan = NewBaseLogicalPlan(ctx, plancodec.TypePartitionUnion, &p, offset)
 	return &p
 }
 
@@ -36,7 +36,7 @@ func (p LogicalPartitionUnionAll) Init(ctx base.PlanContext, offset int) *Logica
 
 // ExhaustPhysicalPlans implements LogicalPlan interface.
 func (p *LogicalPartitionUnionAll) ExhaustPhysicalPlans(prop *property.PhysicalProperty) ([]base.PhysicalPlan, bool, error) {
-	return exhaustPartitionUnionAllPhysicalPlans(p, prop)
+	return utilfuncp.ExhaustPhysicalPlans4LogicalPartitionUnionAll(p, prop)
 }
 
 // *************************** end implementation of LogicalPlan interface ***************************
diff --git a/pkg/planner/core/logical_union_all.go b/pkg/planner/core/operator/logicalop/logical_union_all.go
similarity index 88%
rename from pkg/planner/core/logical_union_all.go
rename to pkg/planner/core/operator/logicalop/logical_union_all.go
index ddfc2c5507041..8b1c9b77dd930 100644
--- a/pkg/planner/core/logical_union_all.go
+++ b/pkg/planner/core/operator/logicalop/logical_union_all.go
@@ -12,12 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package core
+package logicalop
 
 import (
+	"fmt"
+
 	"github.com/pingcap/tidb/pkg/expression"
 	"github.com/pingcap/tidb/pkg/planner/core/base"
-	"github.com/pingcap/tidb/pkg/planner/core/operator/logicalop"
 	"github.com/pingcap/tidb/pkg/planner/property"
 	"github.com/pingcap/tidb/pkg/planner/util"
 	"github.com/pingcap/tidb/pkg/planner/util/optimizetrace"
@@ -28,12 +29,12 @@ import (
 
 // LogicalUnionAll represents LogicalUnionAll plan.
 type LogicalUnionAll struct {
-	logicalop.LogicalSchemaProducer
+	LogicalSchemaProducer
 }
 
 // Init initializes LogicalUnionAll.
 func (p LogicalUnionAll) Init(ctx base.PlanContext, offset int) *LogicalUnionAll {
-	p.BaseLogicalPlan = logicalop.NewBaseLogicalPlan(ctx, plancodec.TypeUnion, &p, offset)
+	p.BaseLogicalPlan = NewBaseLogicalPlan(ctx, plancodec.TypeUnion, &p, offset)
 	return &p
 }
 
@@ -97,7 +98,7 @@ func (p *LogicalUnionAll) PruneColumns(parentUsedCols []*expression.Column, opt
 				for j, col := range schema.Columns {
 					exprs[j] = col
 				}
-				proj := logicalop.LogicalProjection{Exprs: exprs}.Init(p.SCtx(), p.QueryBlockOffset())
+				proj := LogicalProjection{Exprs: exprs}.Init(p.SCtx(), p.QueryBlockOffset())
 				proj.SetSchema(schema)
 
 				proj.SetChildren(child)
@@ -114,14 +115,14 @@ func (p *LogicalUnionAll) PruneColumns(parentUsedCols []*expression.Column, opt
 
 // PushDownTopN implements the base.LogicalPlan.<5th> interface.
 func (p *LogicalUnionAll) PushDownTopN(topNLogicalPlan base.LogicalPlan, opt *optimizetrace.LogicalOptimizeOp) base.LogicalPlan {
-	var topN *logicalop.LogicalTopN
+	var topN *LogicalTopN
 	if topNLogicalPlan != nil {
-		topN = topNLogicalPlan.(*logicalop.LogicalTopN)
+		topN = topNLogicalPlan.(*LogicalTopN)
 	}
 	for i, child := range p.Children() {
-		var newTopN *logicalop.LogicalTopN
+		var newTopN *LogicalTopN
 		if topN != nil {
-			newTopN = logicalop.LogicalTopN{Count: topN.Count + topN.Offset, PreferLimitToCop: topN.PreferLimitToCop}.Init(p.SCtx(), topN.QueryBlockOffset())
+			newTopN = LogicalTopN{Count: topN.Count + topN.Offset, PreferLimitToCop: topN.PreferLimitToCop}.Init(p.SCtx(), topN.QueryBlockOffset())
 			for _, by := range topN.ByItems {
 				newTopN.ByItems = append(newTopN.ByItems, &util.ByItems{Expr: by.Expr, Desc: by.Desc})
 			}
@@ -169,7 +170,7 @@ func (p *LogicalUnionAll) DeriveStats(childStats []*property.StatsInfo, selfSche
 
 // ExhaustPhysicalPlans implements base.LogicalPlan.<14th> interface.
 func (p *LogicalUnionAll) ExhaustPhysicalPlans(prop *property.PhysicalProperty) ([]base.PhysicalPlan, bool, error) {
-	return exhaustUnionAllPhysicalPlans(p, prop)
+	return utilfuncp.ExhaustPhysicalPlans4LogicalUnionAll(p, prop)
 }
 
 // ExtractCorrelatedCols inherits BaseLogicalPlan.LogicalPlan.<15th> implementation.
@@ -193,3 +194,13 @@ func (p *LogicalUnionAll) ExhaustPhysicalPlans(prop *property.PhysicalProperty)
 // ConvertOuterToInnerJoin inherits BaseLogicalPlan.LogicalPlan.<24th> implementation.
 
 // *************************** end implementation of logicalPlan interface ***************************
+
+func appendNewTopNTraceStep(topN *LogicalTopN, union *LogicalUnionAll, opt *optimizetrace.LogicalOptimizeOp) {
+	reason := func() string {
+		return ""
+	}
+	action := func() string {
+		return fmt.Sprintf("%v_%v is added and pushed down across %v_%v", topN.TP(), topN.ID(), union.TP(), union.ID())
+	}
+	opt.AppendStepToCurrent(topN.ID(), topN.TP(), reason, action)
+}
diff --git a/pkg/planner/core/rule_aggregation_push_down.go b/pkg/planner/core/rule_aggregation_push_down.go
index b4221671ef4bf..e172fd967c048 100644
--- a/pkg/planner/core/rule_aggregation_push_down.go
+++ b/pkg/planner/core/rule_aggregation_push_down.go
@@ -446,7 +446,7 @@ func (a *AggregationPushDownSolver) Optimize(_ context.Context, p base.LogicalPl
 	return newLogicalPlan, planChanged, err
 }
 
-func (a *AggregationPushDownSolver) tryAggPushDownForUnion(union *LogicalUnionAll, agg *logicalop.LogicalAggregation, opt *optimizetrace.LogicalOptimizeOp) error {
+func (a *AggregationPushDownSolver) tryAggPushDownForUnion(union *logicalop.LogicalUnionAll, agg *logicalop.LogicalAggregation, opt *optimizetrace.LogicalOptimizeOp) error {
 	for _, aggFunc := range agg.AggFuncs {
 		if !a.isDecomposableWithUnion(aggFunc) {
 			return nil
@@ -661,12 +661,12 @@ func (a *AggregationPushDownSolver) aggPushDown(p base.LogicalPlan, opt *optimiz
 					appendAggPushDownAcrossProjTraceStep(agg, proj, opt)
 				}
 			}
-			if union, ok1 := child.(*LogicalUnionAll); ok1 && p.SCtx().GetSessionVars().AllowAggPushDown {
+			if union, ok1 := child.(*logicalop.LogicalUnionAll); ok1 && p.SCtx().GetSessionVars().AllowAggPushDown {
 				err := a.tryAggPushDownForUnion(union, agg, opt)
 				if err != nil {
 					return nil, err
 				}
-			} else if union, ok1 := child.(*LogicalPartitionUnionAll); ok1 {
+			} else if union, ok1 := child.(*logicalop.LogicalPartitionUnionAll); ok1 {
 				err := a.tryAggPushDownForUnion(&union.LogicalUnionAll, agg, opt)
 				if err != nil {
 					return nil, err
@@ -737,7 +737,7 @@ func appendAggPushDownAcrossProjTraceStep(agg *logicalop.LogicalAggregation, pro
 	opt.AppendStepToCurrent(agg.ID(), agg.TP(), reason, action)
 }
 
-func appendAggPushDownAcrossUnionTraceStep(union *LogicalUnionAll, agg *logicalop.LogicalAggregation, opt *optimizetrace.LogicalOptimizeOp) {
+func appendAggPushDownAcrossUnionTraceStep(union *logicalop.LogicalUnionAll, agg *logicalop.LogicalAggregation, opt *optimizetrace.LogicalOptimizeOp) {
 	evalCtx := union.SCtx().GetExprCtx().GetEvalCtx()
 	reason := func() string {
 		buffer := bytes.NewBufferString(fmt.Sprintf("%v_%v functions[", agg.TP(), agg.ID()))
diff --git a/pkg/planner/core/rule_eliminate_projection.go b/pkg/planner/core/rule_eliminate_projection.go
index d655099b0b2a0..426b46c4bbfc2 100644
--- a/pkg/planner/core/rule_eliminate_projection.go
+++ b/pkg/planner/core/rule_eliminate_projection.go
@@ -165,7 +165,7 @@ func (pe *ProjectionEliminator) eliminate(p base.LogicalPlan, replace map[string
 	}
 	proj, isProj := p.(*logicalop.LogicalProjection)
 	childFlag := canEliminate
-	if _, isUnion := p.(*LogicalUnionAll); isUnion {
+	if _, isUnion := p.(*logicalop.LogicalUnionAll); isUnion {
 		childFlag = false
 	} else if _, isAgg := p.(*logicalop.LogicalAggregation); isAgg || isProj {
 		childFlag = true
diff --git a/pkg/planner/core/rule_partition_processor.go b/pkg/planner/core/rule_partition_processor.go
index 1d9e3a0708245..8601b44e3d64d 100644
--- a/pkg/planner/core/rule_partition_processor.go
+++ b/pkg/planner/core/rule_partition_processor.go
@@ -88,7 +88,7 @@ func (s *PartitionProcessor) rewriteDataSource(lp base.LogicalPlan, opt *optimiz
 		if err != nil {
 			return nil, err
 		}
-		if ua, ok := ds.(*LogicalPartitionUnionAll); ok {
+		if ua, ok := ds.(*logicalop.LogicalPartitionUnionAll); ok {
 			// Adjust the UnionScan->Union->DataSource1, DataSource2 ... to
 			// Union->(UnionScan->DataSource1), (UnionScan->DataSource2)
 			children := make([]base.LogicalPlan, 0, len(ua.Children()))
@@ -1878,7 +1878,7 @@ func (s *PartitionProcessor) makeUnionAllChildren(ds *DataSource, pi *model.Part
 		appendMakeUnionAllChildrenTranceStep(ds, usedDefinition, children[0], children, opt)
 		return children[0], nil
 	}
-	unionAll := LogicalPartitionUnionAll{}.Init(ds.SCtx(), ds.QueryBlockOffset())
+	unionAll := logicalop.LogicalPartitionUnionAll{}.Init(ds.SCtx(), ds.QueryBlockOffset())
 	unionAll.SetChildren(children...)
 	unionAll.SetSchema(ds.Schema().Clone())
 	appendMakeUnionAllChildrenTranceStep(ds, usedDefinition, unionAll, children, opt)
diff --git a/pkg/planner/core/rule_topn_push_down.go b/pkg/planner/core/rule_topn_push_down.go
index 1cb645e8b97ec..84d076773b18b 100644
--- a/pkg/planner/core/rule_topn_push_down.go
+++ b/pkg/planner/core/rule_topn_push_down.go
@@ -16,7 +16,6 @@ package core
 
 import (
 	"context"
-	"fmt"
 
 	"github.com/pingcap/tidb/pkg/planner/core/base"
 	"github.com/pingcap/tidb/pkg/planner/core/operator/logicalop"
@@ -55,13 +54,3 @@ func pushDownTopNForBaseLogicalPlan(lp base.LogicalPlan, topNLogicalPlan base.Lo
 func (*PushDownTopNOptimizer) Name() string {
 	return "topn_push_down"
 }
-
-func appendNewTopNTraceStep(topN *logicalop.LogicalTopN, union *LogicalUnionAll, opt *optimizetrace.LogicalOptimizeOp) {
-	reason := func() string {
-		return ""
-	}
-	action := func() string {
-		return fmt.Sprintf("%v_%v is added and pushed down across %v_%v", topN.TP(), topN.ID(), union.TP(), union.ID())
-	}
-	opt.AppendStepToCurrent(topN.ID(), topN.TP(), reason, action)
-}
diff --git a/pkg/planner/core/stringer.go b/pkg/planner/core/stringer.go
index 26cc9e68f070b..ec6dcf2fd66b3 100644
--- a/pkg/planner/core/stringer.go
+++ b/pkg/planner/core/stringer.go
@@ -44,7 +44,7 @@ func FDToString(p base.LogicalPlan) string {
 
 func needIncludeChildrenString(plan base.Plan) bool {
 	switch x := plan.(type) {
-	case *LogicalUnionAll, *PhysicalUnionAll, *LogicalPartitionUnionAll:
+	case *logicalop.LogicalUnionAll, *PhysicalUnionAll, *logicalop.LogicalPartitionUnionAll:
 		// after https://github.com/pingcap/tidb/pull/25218, the union may contain less than 2 children,
 		// but we still wants to include its child plan's information when calling `toString` on union.
 		return true
@@ -201,7 +201,7 @@ func toString(in base.Plan, strs []string, idxs []int) ([]string, []int) {
 			r := eq.GetArgs()[1].StringWithCtx(ectx, perrors.RedactLogDisable)
 			str += fmt.Sprintf("(%s,%s)", l, r)
 		}
-	case *LogicalUnionAll, *PhysicalUnionAll, *LogicalPartitionUnionAll:
+	case *logicalop.LogicalUnionAll, *PhysicalUnionAll, *logicalop.LogicalPartitionUnionAll:
 		last := len(idxs) - 1
 		idx := idxs[last]
 		children := strs[idx:]
diff --git a/pkg/planner/pattern/pattern.go b/pkg/planner/pattern/pattern.go
index 1d4f2e86ed051..8488bcc856e3d 100644
--- a/pkg/planner/pattern/pattern.go
+++ b/pkg/planner/pattern/pattern.go
@@ -95,7 +95,7 @@ func GetOperand(p base.LogicalPlan) Operand {
 		return OperandDataSource
 	case *logicalop.LogicalUnionScan:
 		return OperandUnionScan
-	case *plannercore.LogicalUnionAll:
+	case *logicalop.LogicalUnionAll:
 		return OperandUnionAll
 	case *logicalop.LogicalSort:
 		return OperandSort
diff --git a/pkg/planner/pattern/pattern_test.go b/pkg/planner/pattern/pattern_test.go
index 9f17faa6236f1..c2fb8ecde54ec 100644
--- a/pkg/planner/pattern/pattern_test.go
+++ b/pkg/planner/pattern/pattern_test.go
@@ -32,7 +32,7 @@ func TestGetOperand(t *testing.T) {
 	require.Equal(t, OperandTableDual, GetOperand(&logicalop.LogicalTableDual{}))
 	require.Equal(t, OperandDataSource, GetOperand(&plannercore.DataSource{}))
 	require.Equal(t, OperandUnionScan, GetOperand(&logicalop.LogicalUnionScan{}))
-	require.Equal(t, OperandUnionAll, GetOperand(&plannercore.LogicalUnionAll{}))
+	require.Equal(t, OperandUnionAll, GetOperand(&logicalop.LogicalUnionAll{}))
 	require.Equal(t, OperandSort, GetOperand(&logicalop.LogicalSort{}))
 	require.Equal(t, OperandTopN, GetOperand(&logicalop.LogicalTopN{}))
 	require.Equal(t, OperandLock, GetOperand(&logicalop.LogicalLock{}))
diff --git a/pkg/planner/util/utilfuncp/func_pointer_misc.go b/pkg/planner/util/utilfuncp/func_pointer_misc.go
index 2912ae80c882b..c8ac26fa24f51 100644
--- a/pkg/planner/util/utilfuncp/func_pointer_misc.go
+++ b/pkg/planner/util/utilfuncp/func_pointer_misc.go
@@ -153,6 +153,14 @@ var ExhaustPhysicalPlans4LogicalAggregation func(lp base.LogicalPlan, prop *prop
 var ExhaustPhysicalPlans4LogicalApply func(lp base.LogicalPlan, prop *property.PhysicalProperty) (
 	[]base.PhysicalPlan, bool, error)
 
+// ExhaustPhysicalPlans4LogicalPartitionUnionAll will be called by LogicalPartitionUnionAll in logicalOp pkg.
+var ExhaustPhysicalPlans4LogicalPartitionUnionAll func(lp base.LogicalPlan, prop *property.PhysicalProperty) (
+	[]base.PhysicalPlan, bool, error)
+
+// ExhaustPhysicalPlans4LogicalUnionAll will be called by LogicalUnionAll in logicalOp pkg.
+var ExhaustPhysicalPlans4LogicalUnionAll func(lp base.LogicalPlan, prop *property.PhysicalProperty) (
+	[]base.PhysicalPlan, bool, error)
+
 // *************************************** physical op related *******************************************
 
 // GetEstimatedProbeCntFromProbeParents will be called by BasePhysicalPlan in physicalOp pkg.