planner: lateral join quality updates

pkg/planner/core/logical_plan_builder.go

@@ -495,35 +495,46 @@ func (b *PlanBuilder) buildResultSetNode(ctx context.Context, node ast.ResultSet
 			return nil, err
 		}
 
-		// Clone output names before modifying to avoid mutating shared structs
 		if x.AsName.L != "" {
-			clonedNames := make([]*types.FieldName, len(p.OutputNames()))
-			for i, name := range p.OutputNames() {
-				if name.Hidden {
-					clonedNames[i] = name
-					continue
-				}
-				// Clone the field name and update table name.
-				// For derived tables, clear DBName so that error messages (e.g. only_full_group_by)
-				// show "alias.col" not "db.alias.col".  The current-database qualifier needed for
-				// hint generation (leading()) is set separately on plannerSelectBlockAsName below.
-				// For base-table aliases (isTableName), inherit DBName for DEFAULT() resolution.
-				dbName := ast.NewCIStr("")
-				if isTableName {
-					dbName = name.DBName
+			if x.Lateral {
+				// LATERAL derived tables: clone output names to avoid mutating shared
+				// structs that may be referenced by the outer scope's schema.
+				clonedNames := make([]*types.FieldName, len(p.OutputNames()))
+				for i, name := range p.OutputNames() {
+					if name.Hidden {
+						clonedNames[i] = name
+						continue
+					}
+					// Clone the field name and update table name.
+					// For derived tables, clear DBName so that error messages (e.g. only_full_group_by)
+					// show "alias.col" not "db.alias.col".  The current-database qualifier needed for
+					// hint generation (leading()) is set separately on plannerSelectBlockAsName below.
+					// For base-table aliases (isTableName), inherit DBName for DEFAULT() resolution.
+					dbName := ast.NewCIStr("")
+					if isTableName {
+						dbName = name.DBName
+					}
+					clonedNames[i] = &types.FieldName{
+						DBName:            dbName,
+						OrigTblName:       name.OrigTblName,
+						OrigColName:       name.OrigColName,
+						TblName:           x.AsName,
+						ColName:           name.ColName,
+						NotExplicitUsable: name.NotExplicitUsable,
+						Redundant:         name.Redundant,
+						Hidden:            name.Hidden,
+					}
 				}
-				clonedNames[i] = &types.FieldName{
-					DBName:            dbName,
-					OrigTblName:       name.OrigTblName,
-					OrigColName:       name.OrigColName,
-					TblName:           x.AsName,
-					ColName:           name.ColName,
-					NotExplicitUsable: name.NotExplicitUsable,
-					Redundant:         name.Redundant,
-					Hidden:            name.Hidden,
+				p.SetOutputNames(clonedNames)
+			} else {
+				// Non-LATERAL: update TblName in place (original behavior).
+				for _, name := range p.OutputNames() {
+					if name.Hidden {
+						continue
+					}
+					name.TblName = x.AsName
 				}
 			}
-			p.SetOutputNames(clonedNames)
 		}
 		// Apply column alias list from AS dt(c1, c2, ...) syntax.
 		// Only valid for derived tables (not table names); parser enforces this.
@@ -674,7 +685,15 @@ func findJoinFullSchema(p base.LogicalPlan) (*expression.Schema, types.NameSlice
 func containsLateralTableSource(node ast.ResultSetNode) bool {
 	switch n := node.(type) {
 	case *ast.TableSource:
-		return n.Lateral
+		if n.Lateral {
+			return true
+		}
+		// Descend into the inner source (derived table / set-op) so nested
+		// LATERAL inside a subquery or set-op used as a table source is detected.
+		if inner, ok := n.Source.(ast.ResultSetNode); ok {
+			return containsLateralTableSource(inner)
+		}
+		return false
 	case *ast.Join:
 		// For parenthesized single table refs, the parser creates Join{Left: TableSource, Right: nil}
 		if n.Right == nil {
@@ -982,7 +1001,6 @@ func (b *PlanBuilder) buildLateralJoin(ctx context.Context, leftPlan, rightPlan
 
 	ap.SetChildren(leftPlan, rightPlan)
 	ap.SetSchema(expression.MergeSchema(leftPlan.Schema(), rightPlan.Schema()))
-	setIsInApplyForCTE(rightPlan, ap.Schema())
 
 	// Note: nullability adjustment is not needed for InnerJoin (the only type supported currently).
 	// When LEFT/RIGHT JOIN support is added, ResetNotNullFlag must be called here.
@@ -1040,6 +1058,10 @@ func (b *PlanBuilder) buildLateralJoin(ctx context.Context, leftPlan, rightPlan
 		ap.FullNames = append(ap.FullNames, &name)
 	}
 
+	// Mark inner CTEs against FullSchema so correlations via USING/NATURAL
+	// merged columns are detected and the CTE storage is reset per outer row.
+	setIsInApplyForCTE(rightPlan, ap.FullSchema)
+
 	// Handle ON conditions if present
 	if joinNode.On != nil {
 		b.curClause = onClause

pkg/planner/core/operator/logicalop/logical_apply.go

@@ -214,10 +214,11 @@ func (la *LogicalApply) DeriveStats(childStats []*property.StatsInfo, selfSchema
 		if la.JoinType == base.LeftOuterJoin {
 			rowCount = max(rowCount, leftProfile.RowCount)
 		}
-	} else if la.JoinType == base.SemiJoin || la.JoinType == base.AntiSemiJoin {
-		// For SemiJoin and AntiSemiJoin Apply operators (EXISTS / NOT EXISTS
-		// subqueries that cannot be decorrelated), apply SelectionFactor to
-		// the row count estimate, consistent with LogicalJoin.DeriveStats.
+	} else if la.IsLateral && (la.JoinType == base.SemiJoin || la.JoinType == base.AntiSemiJoin) {
+		// For LATERAL SemiJoin/AntiSemiJoin Apply operators, apply SelectionFactor
+		// to the row count estimate, consistent with LogicalJoin.DeriveStats.
+		// Non-lateral Apply (correlated subqueries) keeps the original left row count
+		// to avoid changing existing plan estimates.
 		rowCount *= cost.SelectionFactor
 	}
 	la.SetStats(&property.StatsInfo{

pkg/planner/core/operator/logicalop/logical_cte.go

@@ -218,8 +218,8 @@ func (p *LogicalCTE) DeriveStats(_ []*property.StatsInfo, selfSchema *expression
 			vars := p.SCtx().GetSessionVars()
 			savedParallelApply := vars.EnableParallelApply
 			vars.EnableParallelApply = false
+			defer func() { vars.EnableParallelApply = savedParallelApply }()
 			_, p.Cte.RecursivePartPhysicalPlan, _, err = utilfuncp.DoOptimize(context.TODO(), p.SCtx(), p.Cte.OptFlag, p.Cte.RecursivePartLogicalPlan)
-			vars.EnableParallelApply = savedParallelApply
 			if err != nil {
 				return nil, false, err
 			}

pkg/planner/core/rule_decorrelate.go

@@ -237,8 +237,10 @@ func (s *DecorrelateSolver) optimize(ctx context.Context, p base.LogicalPlan, gr
 		// to find the underlying LogicalJoin, matching the schema used for name
 		// resolution in LATERAL subqueries (see logical_plan_builder.go buildJoin).
 		outerSchema := outerPlan.Schema()
-		if fullSchema, _ := findJoinFullSchema(outerPlan); fullSchema != nil {
-			outerSchema = fullSchema
+		if apply.IsLateral {
+			if fullSchema, _ := findJoinFullSchema(outerPlan); fullSchema != nil {
+				outerSchema = fullSchema
+			}
 		}
 		apply.CorCols = coreusage.ExtractCorColumnsBySchema4LogicalPlan(innerPlan, outerSchema)
 		if len(apply.CorCols) == 0 {
@@ -258,7 +260,7 @@ func (s *DecorrelateSolver) optimize(ctx context.Context, p base.LogicalPlan, gr
 			// Notice that no matter what kind of join is, it's always right.
 			newConds := make([]expression.Expression, 0, len(sel.Conditions))
 			for _, cond := range sel.Conditions {
-				newConds = append(newConds, cond.Decorrelate(outerPlan.Schema()))
+				newConds = append(newConds, cond.Decorrelate(outerSchema))
 			}
 			apply.AttachOnConds(newConds)
 			innerPlan = sel.Children()[0]
@@ -296,9 +298,9 @@ func (s *DecorrelateSolver) optimize(ctx context.Context, p base.LogicalPlan, gr
 			}
 			// step2: when it can be substituted all, we then just do the de-correlation (apply conditions included).
 			for i, expr := range proj.Exprs {
-				proj.Exprs[i] = expr.Decorrelate(outerPlan.Schema())
+				proj.Exprs[i] = expr.Decorrelate(outerSchema)
 			}
-			apply.Decorrelate(outerPlan.Schema())
+			apply.Decorrelate(outerSchema)
 
 			innerPlan = proj.Children()[0]
 			apply.SetChildren(outerPlan, innerPlan)