diff --git a/pkg/planner/cascades/implementation_rules.go b/pkg/planner/cascades/implementation_rules.go
index 5f74b68205c4f..42683c9a19967 100644
--- a/pkg/planner/cascades/implementation_rules.go
+++ b/pkg/planner/cascades/implementation_rules.go
@@ -555,7 +555,7 @@ func (*ImplApply) Match(expr *memo.GroupExpr, prop *property.PhysicalProperty) (
 
 // OnImplement implements ImplementationRule OnImplement interface
 func (*ImplApply) OnImplement(expr *memo.GroupExpr, reqProp *property.PhysicalProperty) ([]memo.Implementation, error) {
-	la := expr.ExprNode.(*plannercore.LogicalApply)
+	la := expr.ExprNode.(*logicalop.LogicalApply)
 	join := plannercore.GetHashJoin(la, reqProp)
 	physicalApply := plannercore.PhysicalApply{
 		PhysicalHashJoin: *join,
diff --git a/pkg/planner/cascades/transformation_rules.go b/pkg/planner/cascades/transformation_rules.go
index a8273c035c98b..69ca39762d7f7 100644
--- a/pkg/planner/cascades/transformation_rules.go
+++ b/pkg/planner/cascades/transformation_rules.go
@@ -2444,7 +2444,7 @@ func NewRuleTransformApplyToJoin() Transformation {
 
 // OnTransform implements Transformation interface.
 func (r *TransformApplyToJoin) OnTransform(old *memo.ExprIter) (newExprs []*memo.GroupExpr, eraseOld bool, eraseAll bool, err error) {
-	apply := old.GetExpr().ExprNode.(*plannercore.LogicalApply)
+	apply := old.GetExpr().ExprNode.(*logicalop.LogicalApply)
 	groupExpr := old.GetExpr()
 	// It's safe to use the old apply instead of creating a new LogicalApply here,
 	// Because apply.CorCols will only be used and updated by this rule during Transformation.
@@ -2502,7 +2502,7 @@ func NewRulePullSelectionUpApply() Transformation {
 // This rule tries to pull up the inner side Selection, and add these conditions
 // to Join condition inside the Apply.
 func (*PullSelectionUpApply) OnTransform(old *memo.ExprIter) (newExprs []*memo.GroupExpr, eraseOld bool, eraseAll bool, err error) {
-	apply := old.GetExpr().ExprNode.(*plannercore.LogicalApply)
+	apply := old.GetExpr().ExprNode.(*logicalop.LogicalApply)
 	outerChildGroup := old.Children[0].Group
 	innerChildGroup := old.Children[1].Group
 	sel := old.Children[1].GetExpr().ExprNode.(*logicalop.LogicalSelection)
@@ -2510,7 +2510,7 @@ func (*PullSelectionUpApply) OnTransform(old *memo.ExprIter) (newExprs []*memo.G
 	for _, cond := range sel.Conditions {
 		newConds = append(newConds, cond.Clone().Decorrelate(outerChildGroup.Prop.Schema))
 	}
-	newApply := plannercore.LogicalApply{
+	newApply := logicalop.LogicalApply{
 		LogicalJoin: *(apply.LogicalJoin.Shallow()),
 		CorCols:     apply.CorCols,
 	}.Init(apply.SCtx(), apply.QueryBlockOffset())
diff --git a/pkg/planner/core/BUILD.bazel b/pkg/planner/core/BUILD.bazel
index 273e6ae569c11..27e669884b614 100644
--- a/pkg/planner/core/BUILD.bazel
+++ b/pkg/planner/core/BUILD.bazel
@@ -22,7 +22,6 @@ go_library(
         "indexmerge_path.go",
         "indexmerge_unfinished_path.go",
         "initialize.go",
-        "logical_apply.go",
         "logical_cte.go",
         "logical_datasource.go",
         "logical_expand.go",
diff --git a/pkg/planner/core/casetest/stats_test.go b/pkg/planner/core/casetest/stats_test.go
index f4a816991cb2b..88dd89a5d9e60 100644
--- a/pkg/planner/core/casetest/stats_test.go
+++ b/pkg/planner/core/casetest/stats_test.go
@@ -81,7 +81,7 @@ func TestGroupNDVs(t *testing.T) {
 				join = v
 				lp = v.Children()[0]
 				stack = append(stack, v.Children()[1])
-			case *core.LogicalApply:
+			case *logicalop.LogicalApply:
 				lp = lp.Children()[0]
 				stack = append(stack, v.Children()[1])
 			case *core.LogicalUnionAll:
diff --git a/pkg/planner/core/collect_column_stats_usage.go b/pkg/planner/core/collect_column_stats_usage.go
index 28a2661e09a43..89cb1c7388e87 100644
--- a/pkg/planner/core/collect_column_stats_usage.go
+++ b/pkg/planner/core/collect_column_stats_usage.go
@@ -271,7 +271,7 @@ func (c *columnStatsUsageCollector) collectFromPlan(lp base.LogicalPlan) {
 			}
 		case *logicalop.LogicalJoin:
 			c.collectPredicateColumnsForJoin(x)
-		case *LogicalApply:
+		case *logicalop.LogicalApply:
 			c.collectPredicateColumnsForJoin(&x.LogicalJoin)
 			// Assume statistics of correlated columns are needed.
 			// Correlated columns can be found in LogicalApply.Children()[0].Schema(). Since we already visit LogicalApply.Children()[0],
diff --git a/pkg/planner/core/core_init.go b/pkg/planner/core/core_init.go
index cdb35778df1dc..0c0e63323cd02 100644
--- a/pkg/planner/core/core_init.go
+++ b/pkg/planner/core/core_init.go
@@ -43,6 +43,7 @@ func init() {
 	utilfuncp.ExhaustPhysicalPlans4LogicalTopN = exhaustPhysicalPlans4LogicalTopN
 	utilfuncp.ExhaustPhysicalPlans4LogicalLock = exhaustPhysicalPlans4LogicalLock
 	utilfuncp.ExhaustPhysicalPlans4LogicalJoin = exhaustPhysicalPlans4LogicalJoin
+	utilfuncp.ExhaustPhysicalPlans4LogicalApply = exhaustPhysicalPlans4LogicalApply
 	utilfuncp.ExhaustPhysicalPlans4LogicalLimit = exhaustPhysicalPlans4LogicalLimit
 	utilfuncp.ExhaustPhysicalPlans4LogicalWindow = exhaustPhysicalPlans4LogicalWindow
 	utilfuncp.ExhaustPhysicalPlans4LogicalSequence = exhaustPhysicalPlans4LogicalSequence
diff --git a/pkg/planner/core/exhaust_physical_plans.go b/pkg/planner/core/exhaust_physical_plans.go
index f0bd7f98c8c7c..4916531a38877 100644
--- a/pkg/planner/core/exhaust_physical_plans.go
+++ b/pkg/planner/core/exhaust_physical_plans.go
@@ -2224,12 +2224,13 @@ func MatchItems(p *property.PhysicalProperty, items []*util.ByItems) bool {
 }
 
 // GetHashJoin is public for cascades planner.
-func GetHashJoin(la *LogicalApply, prop *property.PhysicalProperty) *PhysicalHashJoin {
+func GetHashJoin(la *logicalop.LogicalApply, prop *property.PhysicalProperty) *PhysicalHashJoin {
 	return getHashJoin(&la.LogicalJoin, prop, 1, false)
 }
 
-// ExhaustPhysicalPlans4LogicalApply generates the physical plan for a logical apply.
-func ExhaustPhysicalPlans4LogicalApply(la *LogicalApply, prop *property.PhysicalProperty) ([]base.PhysicalPlan, bool, error) {
+// exhaustPhysicalPlans4LogicalApply generates the physical plan for a logical apply.
+func exhaustPhysicalPlans4LogicalApply(lp base.LogicalPlan, prop *property.PhysicalProperty) ([]base.PhysicalPlan, bool, error) {
+	la := lp.(*logicalop.LogicalApply)
 	if !prop.AllColsFromSchema(la.Children()[0].Schema()) || prop.IsFlashProp() { // for convenient, we don't pass through any prop
 		la.SCtx().GetSessionVars().RaiseWarningWhenMPPEnforced(
 			"MPP mode may be blocked because operator `Apply` is not supported now.")
@@ -2243,7 +2244,9 @@ func ExhaustPhysicalPlans4LogicalApply(la *LogicalApply, prop *property.Physical
 	join := GetHashJoin(la, prop)
 	var columns = make([]*expression.Column, 0, len(la.CorCols))
 	for _, colColumn := range la.CorCols {
-		columns = append(columns, &colColumn.Column)
+		// fix the liner warning.
+		tmp := colColumn
+		columns = append(columns, &tmp.Column)
 	}
 	cacheHitRatio := 0.0
 	if la.StatsInfo().RowCount != 0 {
diff --git a/pkg/planner/core/logical_plan_builder.go b/pkg/planner/core/logical_plan_builder.go
index cf1a04448419b..1c5b5915e8137 100644
--- a/pkg/planner/core/logical_plan_builder.go
+++ b/pkg/planner/core/logical_plan_builder.go
@@ -5063,7 +5063,7 @@ func (b *PlanBuilder) buildProjUponView(_ context.Context, dbName model.CIStr, t
 // every row from outerPlan and the whole innerPlan.
 func (b *PlanBuilder) buildApplyWithJoinType(outerPlan, innerPlan base.LogicalPlan, tp logicalop.JoinType, markNoDecorrelate bool) base.LogicalPlan {
 	b.optFlag = b.optFlag | flagPredicatePushDown | flagBuildKeyInfo | flagDecorrelate | flagConvertOuterToInnerJoin
-	ap := LogicalApply{LogicalJoin: logicalop.LogicalJoin{JoinType: tp}, NoDecorrelate: markNoDecorrelate}.Init(b.ctx, b.getSelectOffset())
+	ap := logicalop.LogicalApply{LogicalJoin: logicalop.LogicalJoin{JoinType: tp}, NoDecorrelate: markNoDecorrelate}.Init(b.ctx, b.getSelectOffset())
 	ap.SetChildren(outerPlan, innerPlan)
 	ap.SetOutputNames(make([]*types.FieldName, outerPlan.Schema().Len()+innerPlan.Schema().Len()))
 	copy(ap.OutputNames(), outerPlan.OutputNames())
@@ -5092,7 +5092,7 @@ func (b *PlanBuilder) buildSemiApply(outerPlan, innerPlan base.LogicalPlan, cond
 	}
 
 	setIsInApplyForCTE(innerPlan, join.Schema())
-	ap := &LogicalApply{LogicalJoin: *join, NoDecorrelate: markNoDecorrelate}
+	ap := &logicalop.LogicalApply{LogicalJoin: *join, NoDecorrelate: markNoDecorrelate}
 	ap.SetTP(plancodec.TypeApply)
 	ap.SetSelf(ap)
 	return ap, nil
diff --git a/pkg/planner/core/logical_plans.go b/pkg/planner/core/logical_plans.go
index f529aaebef64f..f0021cbb20b0b 100644
--- a/pkg/planner/core/logical_plans.go
+++ b/pkg/planner/core/logical_plans.go
@@ -24,7 +24,7 @@ var (
 	_ base.LogicalPlan = &logicalop.LogicalAggregation{}
 	_ base.LogicalPlan = &logicalop.LogicalProjection{}
 	_ base.LogicalPlan = &logicalop.LogicalSelection{}
-	_ base.LogicalPlan = &LogicalApply{}
+	_ base.LogicalPlan = &logicalop.LogicalApply{}
 	_ base.LogicalPlan = &logicalop.LogicalMaxOneRow{}
 	_ base.LogicalPlan = &logicalop.LogicalTableDual{}
 	_ base.LogicalPlan = &DataSource{}
diff --git a/pkg/planner/core/operator/logicalop/BUILD.bazel b/pkg/planner/core/operator/logicalop/BUILD.bazel
index 10b454e4d2a4c..86286d42dafc3 100644
--- a/pkg/planner/core/operator/logicalop/BUILD.bazel
+++ b/pkg/planner/core/operator/logicalop/BUILD.bazel
@@ -5,6 +5,7 @@ go_library(
     srcs = [
         "base_logical_plan.go",
         "logical_aggregation.go",
+        "logical_apply.go",
         "logical_cte_table.go",
         "logical_join.go",
         "logical_limit.go",
@@ -43,6 +44,8 @@ go_library(
         "//pkg/planner/funcdep",
         "//pkg/planner/property",
         "//pkg/planner/util",
+        "//pkg/planner/util/coreusage",
+        "//pkg/planner/util/fixcontrol",
         "//pkg/planner/util/optimizetrace",
         "//pkg/planner/util/optimizetrace/logicaltrace",
         "//pkg/planner/util/utilfuncp",
diff --git a/pkg/planner/core/logical_apply.go b/pkg/planner/core/operator/logicalop/logical_apply.go
similarity index 92%
rename from pkg/planner/core/logical_apply.go
rename to pkg/planner/core/operator/logicalop/logical_apply.go
index 100b636c3f362..7ba12a78189f0 100644
--- a/pkg/planner/core/logical_apply.go
+++ b/pkg/planner/core/operator/logicalop/logical_apply.go
@@ -12,27 +12,27 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package core
+package logicalop
 
 import (
 	"github.com/pingcap/tidb/pkg/expression"
 	"github.com/pingcap/tidb/pkg/parser/ast"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
 	"github.com/pingcap/tidb/pkg/planner/core/base"
-	"github.com/pingcap/tidb/pkg/planner/core/operator/logicalop"
 	fd "github.com/pingcap/tidb/pkg/planner/funcdep"
 	"github.com/pingcap/tidb/pkg/planner/property"
 	"github.com/pingcap/tidb/pkg/planner/util/coreusage"
 	"github.com/pingcap/tidb/pkg/planner/util/fixcontrol"
 	"github.com/pingcap/tidb/pkg/planner/util/optimizetrace"
 	"github.com/pingcap/tidb/pkg/planner/util/optimizetrace/logicaltrace"
+	"github.com/pingcap/tidb/pkg/planner/util/utilfuncp"
 	"github.com/pingcap/tidb/pkg/types"
 	"github.com/pingcap/tidb/pkg/util/plancodec"
 )
 
 // LogicalApply gets one row from outer executor and gets one row from inner executor according to outer row.
 type LogicalApply struct {
-	logicalop.LogicalJoin
+	LogicalJoin
 
 	CorCols []*expression.CorrelatedColumn
 	// NoDecorrelate is from /*+ no_decorrelate() */ hint.
@@ -41,7 +41,7 @@ type LogicalApply struct {
 
 // Init initializes LogicalApply.
 func (la LogicalApply) Init(ctx base.PlanContext, offset int) *LogicalApply {
-	la.BaseLogicalPlan = logicalop.NewBaseLogicalPlan(ctx, plancodec.TypeApply, &la, offset)
+	la.BaseLogicalPlan = NewBaseLogicalPlan(ctx, plancodec.TypeApply, &la, offset)
 	return &la
 }
 
@@ -76,7 +76,7 @@ func (la *LogicalApply) PruneColumns(parentUsedCols []*expression.Column, opt *o
 	leftCols, rightCols := la.ExtractUsedCols(parentUsedCols)
 	allowEliminateApply := fixcontrol.GetBoolWithDefault(la.SCtx().GetSessionVars().GetOptimizerFixControlMap(), fixcontrol.Fix45822, true)
 	var err error
-	if allowEliminateApply && rightCols == nil && la.JoinType == logicalop.LeftOuterJoin {
+	if allowEliminateApply && rightCols == nil && la.JoinType == LeftOuterJoin {
 		logicaltrace.ApplyEliminateTraceStep(la.Children()[1], opt)
 		resultPlan := la.Children()[0]
 		// reEnter the new child's column pruning, returning child[0] as a new child here.
@@ -134,7 +134,7 @@ func (la *LogicalApply) DeriveStats(childStats []*property.StatsInfo, selfSchema
 	for id, c := range leftProfile.ColNDVs {
 		la.StatsInfo().ColNDVs[id] = c
 	}
-	if la.JoinType == logicalop.LeftOuterSemiJoin || la.JoinType == logicalop.AntiLeftOuterSemiJoin {
+	if la.JoinType == LeftOuterSemiJoin || la.JoinType == AntiLeftOuterSemiJoin {
 		la.StatsInfo().ColNDVs[selfSchema.Columns[selfSchema.Len()-1].UniqueID] = 2.0
 	} else {
 		for i := childSchema[0].Len(); i < selfSchema.Len(); i++ {
@@ -149,7 +149,7 @@ func (la *LogicalApply) DeriveStats(childStats []*property.StatsInfo, selfSchema
 func (la *LogicalApply) ExtractColGroups(colGroups [][]*expression.Column) [][]*expression.Column {
 	var outerSchema *expression.Schema
 	// Apply doesn't have RightOuterJoin.
-	if la.JoinType == logicalop.LeftOuterJoin || la.JoinType == logicalop.LeftOuterSemiJoin || la.JoinType == logicalop.AntiLeftOuterSemiJoin {
+	if la.JoinType == LeftOuterJoin || la.JoinType == LeftOuterSemiJoin || la.JoinType == AntiLeftOuterSemiJoin {
 		outerSchema = la.Children()[0].Schema()
 	}
 	if len(colGroups) == 0 || outerSchema == nil {
@@ -170,7 +170,7 @@ func (la *LogicalApply) ExtractColGroups(colGroups [][]*expression.Column) [][]*
 
 // ExhaustPhysicalPlans implements base.LogicalPlan.<14th> interface.
 func (la *LogicalApply) ExhaustPhysicalPlans(prop *property.PhysicalProperty) ([]base.PhysicalPlan, bool, error) {
-	return ExhaustPhysicalPlans4LogicalApply(la, prop)
+	return utilfuncp.ExhaustPhysicalPlans4LogicalApply(la, prop)
 }
 
 // ExtractCorrelatedCols implements base.LogicalPlan.<15th> interface.
@@ -221,11 +221,11 @@ func (la *LogicalApply) ExtractFD() *fd.FDSet {
 		}
 	}
 	switch la.JoinType {
-	case logicalop.InnerJoin:
+	case InnerJoin:
 		return la.ExtractFDForInnerJoin(eqCond)
-	case logicalop.LeftOuterJoin, logicalop.RightOuterJoin:
+	case LeftOuterJoin, RightOuterJoin:
 		return la.ExtractFDForOuterJoin(eqCond)
-	case logicalop.SemiJoin:
+	case SemiJoin:
 		return la.ExtractFDForSemiJoin(eqCond)
 	default:
 		return &fd.FDSet{HashCodeToUniqueID: make(map[string]int)}
@@ -240,7 +240,7 @@ func (la *LogicalApply) ExtractFD() *fd.FDSet {
 
 // CanPullUpAgg checks if an apply can pull an aggregation up.
 func (la *LogicalApply) CanPullUpAgg() bool {
-	if la.JoinType != logicalop.InnerJoin && la.JoinType != logicalop.LeftOuterJoin {
+	if la.JoinType != InnerJoin && la.JoinType != LeftOuterJoin {
 		return false
 	}
 	if len(la.EqualConditions)+len(la.LeftConditions)+len(la.RightConditions)+len(la.OtherConditions) > 0 {
@@ -280,7 +280,7 @@ func (la *LogicalApply) DeCorColFromEqExpr(expr expression.Expression) expressio
 }
 
 func (la *LogicalApply) getGroupNDVs(colGroups [][]*expression.Column, childStats []*property.StatsInfo) []property.GroupNDV {
-	if len(colGroups) > 0 && (la.JoinType == logicalop.LeftOuterSemiJoin || la.JoinType == logicalop.AntiLeftOuterSemiJoin || la.JoinType == logicalop.LeftOuterJoin) {
+	if len(colGroups) > 0 && (la.JoinType == LeftOuterSemiJoin || la.JoinType == AntiLeftOuterSemiJoin || la.JoinType == LeftOuterJoin) {
 		return childStats[0].GroupNDVs
 	}
 	return nil
diff --git a/pkg/planner/core/plan.go b/pkg/planner/core/plan.go
index 054a38f6eec2a..dde4cba6e8f40 100644
--- a/pkg/planner/core/plan.go
+++ b/pkg/planner/core/plan.go
@@ -252,7 +252,7 @@ func HasMaxOneRow(p base.LogicalPlan, childMaxOneRow []bool) bool {
 	}
 	switch x := p.(type) {
 	case *logicalop.LogicalLock, *logicalop.LogicalLimit, *logicalop.LogicalSort, *logicalop.LogicalSelection,
-		*LogicalApply, *logicalop.LogicalProjection, *logicalop.LogicalWindow, *logicalop.LogicalAggregation:
+		*logicalop.LogicalApply, *logicalop.LogicalProjection, *logicalop.LogicalWindow, *logicalop.LogicalAggregation:
 		return childMaxOneRow[0]
 	case *logicalop.LogicalMaxOneRow:
 		return true
diff --git a/pkg/planner/core/rule_decorrelate.go b/pkg/planner/core/rule_decorrelate.go
index 9fac608f1bfa2..962632c362d30 100644
--- a/pkg/planner/core/rule_decorrelate.go
+++ b/pkg/planner/core/rule_decorrelate.go
@@ -124,7 +124,7 @@ func (*DecorrelateSolver) aggDefaultValueMap(agg *logicalop.LogicalAggregation)
 // Optimize implements base.LogicalOptRule.<0th> interface.
 func (s *DecorrelateSolver) Optimize(ctx context.Context, p base.LogicalPlan, opt *optimizetrace.LogicalOptimizeOp) (base.LogicalPlan, bool, error) {
 	planChanged := false
-	if apply, ok := p.(*LogicalApply); ok {
+	if apply, ok := p.(*logicalop.LogicalApply); ok {
 		outerPlan := apply.Children()[0]
 		innerPlan := apply.Children()[1]
 		apply.CorCols = coreusage.ExtractCorColumnsBySchema4LogicalPlan(apply.Children()[1], apply.Children()[0].Schema())
@@ -393,7 +393,7 @@ func (*DecorrelateSolver) Name() string {
 	return "decorrelate"
 }
 
-func appendApplySimplifiedTraceStep(p *LogicalApply, j *logicalop.LogicalJoin, opt *optimizetrace.LogicalOptimizeOp) {
+func appendApplySimplifiedTraceStep(p *logicalop.LogicalApply, j *logicalop.LogicalJoin, opt *optimizetrace.LogicalOptimizeOp) {
 	action := func() string {
 		return fmt.Sprintf("%v_%v simplified into %v_%v", plancodec.TypeApply, p.ID(), plancodec.TypeJoin, j.ID())
 	}
@@ -433,7 +433,7 @@ func appendRemoveLimitTraceStep(limit *logicalop.LogicalLimit, opt *optimizetrac
 	opt.AppendStepToCurrent(limit.ID(), limit.TP(), reason, action)
 }
 
-func appendRemoveProjTraceStep(p *LogicalApply, proj *logicalop.LogicalProjection, opt *optimizetrace.LogicalOptimizeOp) {
+func appendRemoveProjTraceStep(p *logicalop.LogicalApply, proj *logicalop.LogicalProjection, opt *optimizetrace.LogicalOptimizeOp) {
 	action := func() string {
 		return fmt.Sprintf("%v_%v removed from plan tree", proj.TP(), proj.ID())
 	}
@@ -443,7 +443,7 @@ func appendRemoveProjTraceStep(p *LogicalApply, proj *logicalop.LogicalProjectio
 	opt.AppendStepToCurrent(proj.ID(), proj.TP(), reason, action)
 }
 
-func appendMoveProjTraceStep(p *LogicalApply, np base.LogicalPlan, proj *logicalop.LogicalProjection, opt *optimizetrace.LogicalOptimizeOp) {
+func appendMoveProjTraceStep(p *logicalop.LogicalApply, np base.LogicalPlan, proj *logicalop.LogicalProjection, opt *optimizetrace.LogicalOptimizeOp) {
 	action := func() string {
 		return fmt.Sprintf("%v_%v is moved as %v_%v's parent", proj.TP(), proj.ID(), np.TP(), np.ID())
 	}
@@ -463,7 +463,7 @@ func appendRemoveSortTraceStep(sort *logicalop.LogicalSort, opt *optimizetrace.L
 	opt.AppendStepToCurrent(sort.ID(), sort.TP(), reason, action)
 }
 
-func appendPullUpAggTraceStep(p *LogicalApply, np base.LogicalPlan, agg *logicalop.LogicalAggregation, opt *optimizetrace.LogicalOptimizeOp) {
+func appendPullUpAggTraceStep(p *logicalop.LogicalApply, np base.LogicalPlan, agg *logicalop.LogicalAggregation, opt *optimizetrace.LogicalOptimizeOp) {
 	action := func() string {
 		return fmt.Sprintf("%v_%v pulled up as %v_%v's parent, and %v_%v's join type becomes %v",
 			agg.TP(), agg.ID(), np.TP(), np.ID(), p.TP(), p.ID(), p.JoinType.String())
@@ -475,7 +475,7 @@ func appendPullUpAggTraceStep(p *LogicalApply, np base.LogicalPlan, agg *logical
 	opt.AppendStepToCurrent(agg.ID(), agg.TP(), reason, action)
 }
 
-func appendAddProjTraceStep(p *LogicalApply, proj *logicalop.LogicalProjection, opt *optimizetrace.LogicalOptimizeOp) {
+func appendAddProjTraceStep(p *logicalop.LogicalApply, proj *logicalop.LogicalProjection, opt *optimizetrace.LogicalOptimizeOp) {
 	action := func() string {
 		return fmt.Sprintf("%v_%v is added as %v_%v's parent", proj.TP(), proj.ID(), p.TP(), p.ID())
 	}
@@ -485,7 +485,7 @@ func appendAddProjTraceStep(p *LogicalApply, proj *logicalop.LogicalProjection,
 	opt.AppendStepToCurrent(proj.ID(), proj.TP(), reason, action)
 }
 
-func appendModifyAggTraceStep(outerPlan base.LogicalPlan, p *LogicalApply, agg *logicalop.LogicalAggregation, sel *logicalop.LogicalSelection,
+func appendModifyAggTraceStep(outerPlan base.LogicalPlan, p *logicalop.LogicalApply, agg *logicalop.LogicalAggregation, sel *logicalop.LogicalSelection,
 	appendedGroupByCols *expression.Schema, appendedAggFuncs []*aggregation.AggFuncDesc,
 	eqCondWithCorCol []*expression.ScalarFunction, opt *optimizetrace.LogicalOptimizeOp) {
 	evalCtx := outerPlan.SCtx().GetExprCtx().GetEvalCtx()
diff --git a/pkg/planner/core/rule_eliminate_projection.go b/pkg/planner/core/rule_eliminate_projection.go
index f9fb992358e16..d655099b0b2a0 100644
--- a/pkg/planner/core/rule_eliminate_projection.go
+++ b/pkg/planner/core/rule_eliminate_projection.go
@@ -180,7 +180,7 @@ func (pe *ProjectionEliminator) eliminate(p base.LogicalPlan, replace map[string
 	switch x := p.(type) {
 	case *logicalop.LogicalJoin:
 		x.SetSchema(logicalop.BuildLogicalJoinSchema(x.JoinType, x))
-	case *LogicalApply:
+	case *logicalop.LogicalApply:
 		x.SetSchema(logicalop.BuildLogicalJoinSchema(x.JoinType, x))
 	default:
 		for _, dst := range p.Schema().Columns {
diff --git a/pkg/planner/core/stringer.go b/pkg/planner/core/stringer.go
index 5a0191b3cecfd..26cc9e68f070b 100644
--- a/pkg/planner/core/stringer.go
+++ b/pkg/planner/core/stringer.go
@@ -71,7 +71,7 @@ func fdToString(in base.LogicalPlan, strs []string, idxs []int) ([]string, []int
 		}
 	case *DataSource:
 		strs = append(strs, "{"+x.FDs().String()+"}")
-	case *LogicalApply:
+	case *logicalop.LogicalApply:
 		strs = append(strs, "{"+x.FDs().String()+"}")
 	case *logicalop.LogicalJoin:
 		strs = append(strs, "{"+x.FDs().String()+"}")
@@ -160,7 +160,7 @@ func toString(in base.Plan, strs []string, idxs []int) ([]string, []int) {
 			r := x.RightJoinKeys[i].StringWithCtx(ectx, perrors.RedactLogDisable)
 			str += fmt.Sprintf("(%s,%s)", l, r)
 		}
-	case *LogicalApply, *PhysicalApply:
+	case *logicalop.LogicalApply, *PhysicalApply:
 		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 e2469c585a97e..1d4f2e86ed051 100644
--- a/pkg/planner/pattern/pattern.go
+++ b/pkg/planner/pattern/pattern.go
@@ -77,7 +77,7 @@ const (
 // GetOperand maps logical plan operator to Operand.
 func GetOperand(p base.LogicalPlan) Operand {
 	switch p.(type) {
-	case *plannercore.LogicalApply:
+	case *logicalop.LogicalApply:
 		return OperandApply
 	case *logicalop.LogicalJoin:
 		return OperandJoin
diff --git a/pkg/planner/pattern/pattern_test.go b/pkg/planner/pattern/pattern_test.go
index 0f8b0fbf2bbb0..9f17faa6236f1 100644
--- a/pkg/planner/pattern/pattern_test.go
+++ b/pkg/planner/pattern/pattern_test.go
@@ -27,7 +27,7 @@ func TestGetOperand(t *testing.T) {
 	require.Equal(t, OperandAggregation, GetOperand(&logicalop.LogicalAggregation{}))
 	require.Equal(t, OperandProjection, GetOperand(&logicalop.LogicalProjection{}))
 	require.Equal(t, OperandSelection, GetOperand(&logicalop.LogicalSelection{}))
-	require.Equal(t, OperandApply, GetOperand(&plannercore.LogicalApply{}))
+	require.Equal(t, OperandApply, GetOperand(&logicalop.LogicalApply{}))
 	require.Equal(t, OperandMaxOneRow, GetOperand(&logicalop.LogicalMaxOneRow{}))
 	require.Equal(t, OperandTableDual, GetOperand(&logicalop.LogicalTableDual{}))
 	require.Equal(t, OperandDataSource, GetOperand(&plannercore.DataSource{}))
diff --git a/pkg/planner/util/utilfuncp/func_pointer_misc.go b/pkg/planner/util/utilfuncp/func_pointer_misc.go
index abcf1117c9821..2912ae80c882b 100644
--- a/pkg/planner/util/utilfuncp/func_pointer_misc.go
+++ b/pkg/planner/util/utilfuncp/func_pointer_misc.go
@@ -149,6 +149,10 @@ var ExhaustPhysicalPlans4LogicalJoin func(lp base.LogicalPlan, prop *property.Ph
 var ExhaustPhysicalPlans4LogicalAggregation func(lp base.LogicalPlan, prop *property.PhysicalProperty) (
 	[]base.PhysicalPlan, bool, error)
 
+// ExhaustPhysicalPlans4LogicalApply will be called by LogicalApply in logicalOp pkg.
+var ExhaustPhysicalPlans4LogicalApply func(lp base.LogicalPlan, prop *property.PhysicalProperty) (
+	[]base.PhysicalPlan, bool, error)
+
 // *************************************** physical op related *******************************************
 
 // GetEstimatedProbeCntFromProbeParents will be called by BasePhysicalPlan in physicalOp pkg.