sql: make getPlanDistribution a method on planner

This will simplify the following commit a bit.

Release note: None

Author: yuzefovich

pkg/sql/apply_join.go

@@ -315,10 +315,7 @@ func runPlanInsidePlan(
 		}
 	}
 
-	distributePlan, distSQLProhibitedErr := getPlanDistribution(
-		ctx, plannerCopy.Descriptors().HasUncommittedTypes(),
-		plannerCopy.SessionData(), plan.main, &plannerCopy.distSQLVisitor,
-	)
+	distributePlan, distSQLProhibitedErr := plannerCopy.getPlanDistribution(ctx, plan.main)
 	distributeType := DistributionType(LocalDistribution)
 	if distributePlan.WillDistribute() {
 		distributeType = FullDistribution

pkg/sql/conn_executor_exec.go

@@ -3028,10 +3028,7 @@ func (ex *connExecutor) dispatchToExecutionEngine(
 			}
 		}
 	}
-	distributePlan, distSQLProhibitedErr := getPlanDistribution(
-		ctx, planner.Descriptors().HasUncommittedTypes(),
-		ex.sessionData(), planner.curPlan.main, &planner.distSQLVisitor,
-	)
+	distributePlan, distSQLProhibitedErr := planner.getPlanDistribution(ctx, planner.curPlan.main)
 	if afterGetPlanDistribution != nil {
 		afterGetPlanDistribution()
 	}

pkg/sql/distsql_running.go

@@ -1896,10 +1896,7 @@ func (dsp *DistSQLPlanner) planAndRunSubquery(
 	skipDistSQLDiagramGeneration bool,
 	mustUseLeafTxn bool,
 ) error {
-	subqueryDistribution, distSQLProhibitedErr := getPlanDistribution(
-		ctx, planner.Descriptors().HasUncommittedTypes(),
-		planner.SessionData(), subqueryPlan.plan, &planner.distSQLVisitor,
-	)
+	subqueryDistribution, distSQLProhibitedErr := planner.getPlanDistribution(ctx, subqueryPlan.plan)
 	distribute := DistributionType(LocalDistribution)
 	if subqueryDistribution.WillDistribute() {
 		distribute = FullDistribution
@@ -2507,10 +2504,7 @@ func (dsp *DistSQLPlanner) planAndRunPostquery(
 	associateNodeWithComponents func(exec.Node, execComponents),
 	addTopLevelQueryStats func(stats *topLevelQueryStats),
 ) error {
-	postqueryDistribution, distSQLProhibitedErr := getPlanDistribution(
-		ctx, planner.Descriptors().HasUncommittedTypes(),
-		planner.SessionData(), postqueryPlan, &planner.distSQLVisitor,
-	)
+	postqueryDistribution, distSQLProhibitedErr := planner.getPlanDistribution(ctx, postqueryPlan)
 	distribute := DistributionType(LocalDistribution)
 	if postqueryDistribution.WillDistribute() {
 		distribute = FullDistribution

pkg/sql/exec_util.go

@@ -2079,12 +2079,8 @@ func shouldDistributeGivenRecAndMode(
 // remote node to the gateway.
 // TODO(yuzefovich): this will be easy to solve once the DistSQL spec factory is
 // completed but is quite annoying to do at the moment.
-func getPlanDistribution(
-	ctx context.Context,
-	txnHasUncommittedTypes bool,
-	sd *sessiondata.SessionData,
-	plan planMaybePhysical,
-	distSQLVisitor *distSQLExprCheckVisitor,
+func (p *planner) getPlanDistribution(
+	ctx context.Context, plan planMaybePhysical,
 ) (_ physicalplan.PlanDistribution, distSQLProhibitedErr error) {
 	if plan.isPhysicalPlan() {
 		// TODO(#47473): store the distSQLProhibitedErr for DistSQL spec factory
@@ -2095,10 +2091,11 @@ func getPlanDistribution(
 	// If this transaction has modified or created any types, it is not safe to
 	// distribute due to limitations around leasing descriptors modified in the
 	// current transaction.
-	if txnHasUncommittedTypes {
+	if p.Descriptors().HasUncommittedDescriptors() {
 		return physicalplan.LocalPlan, nil
 	}
 
+	sd := p.SessionData()
 	if sd.DistSQLMode == sessiondatapb.DistSQLOff {
 		return physicalplan.LocalPlan, nil
 	}
@@ -2108,7 +2105,7 @@ func getPlanDistribution(
 		return physicalplan.LocalPlan, nil
 	}
 
-	rec, err := checkSupportForPlanNode(ctx, plan.planNode, distSQLVisitor, sd)
+	rec, err := checkSupportForPlanNode(ctx, plan.planNode, &p.distSQLVisitor, sd)
 	if err != nil {
 		// Don't use distSQL for this request.
 		log.VEventf(ctx, 1, "query not supported for distSQL: %s", err)

pkg/sql/explain_plan.go

@@ -58,10 +58,7 @@ func (e *explainPlanNode) startExec(params runParams) error {
 		// Note that we delay adding the annotation about the distribution until
 		// after the plan is finalized (when the physical plan is successfully
 		// created).
-		distribution, _ := getPlanDistribution(
-			params.ctx, params.p.Descriptors().HasUncommittedTypes(),
-			params.extendedEvalCtx.SessionData(), plan.main, &params.p.distSQLVisitor,
-		)
+		distribution, _ := params.p.getPlanDistribution(params.ctx, plan.main)
 
 		outerSubqueries := params.p.curPlan.subqueryPlans
 		distSQLPlanner := params.extendedEvalCtx.DistSQLPlanner

pkg/sql/explain_vec.go

@@ -36,10 +36,7 @@ type explainVecNode struct {
 func (n *explainVecNode) startExec(params runParams) error {
 	n.run.values = make(tree.Datums, 1)
 	distSQLPlanner := params.extendedEvalCtx.DistSQLPlanner
-	distribution, _ := getPlanDistribution(
-		params.ctx, params.p.Descriptors().HasUncommittedTypes(),
-		params.extendedEvalCtx.SessionData(), n.plan.main, &params.p.distSQLVisitor,
-	)
+	distribution, _ := params.p.getPlanDistribution(params.ctx, n.plan.main)
 	outerSubqueries := params.p.curPlan.subqueryPlans
 	planCtx := newPlanningCtxForExplainPurposes(distSQLPlanner, params, n.plan.subqueryPlans, distribution)
 	defer func() {

pkg/sql/schema_changer.go

@@ -477,12 +477,7 @@ func (sc *SchemaChanger) backfillQueryIntoTable(
 					}
 				}
 			}
-
-			planDistribution, _ := getPlanDistribution(
-				ctx, localPlanner.Descriptors().HasUncommittedTypes(),
-				localPlanner.extendedEvalCtx.SessionData(),
-				localPlanner.curPlan.main, &localPlanner.distSQLVisitor,
-			)
+			planDistribution, _ := localPlanner.getPlanDistribution(ctx, localPlanner.curPlan.main)
 			isLocal := !planDistribution.WillDistribute()
 			out := execinfrapb.ProcessorCoreUnion{BulkRowWriter: &execinfrapb.BulkRowWriterSpec{
 				Table: *table.TableDesc(),