Merge #142248 #142671

142248: sql: unify "distribution" in EXPLAIN and EXPLAIN ANALYZE r=yuzefovich a=yuzefovich

Previously, it was possible to observe different `distribution` property between EXPLAIN and EXPLAIN ANALYZE output of the same query. The reason for that was that in the former we looked at the physical plan's distribution which indicates whether the physical plan is actually local or not, whereas for EXPLAIN ANALYZE we looked at presence of some plan flags which indicate the _recommendation_ about the physical plan, and the actual physical plan depends on the range placement. Concretely, it was possible to show `distribution: full` in the output of EXPLAIN ANALYZE when we only had a single flow on the gateway. This commit fixes that and unifies two EXPLAIN variants.

It also removes `planFlagNotDistributed` plan flag since it's not used anywhere.

Fixes: #128623.

Release note (bug fix): EXPLAIN ANALYZE output could previously incorrectly show `distribution: full` when the physical plan only ran on the gateway node in some cases (meaning it should've shown `distribution: local`), and this is now fixed. Note that "vanilla" EXPLAIN showed the correct information. The bug has been present since before 23.1 version.

142671: logictest: fix rare flake around partial stats r=yuzefovich a=yuzefovich

This commit applies the same fix as we had in 5707cf1 to be done after each attempt to collect partial stats. We were missing clearing of the stats cache in 3 places, and that is now fixed. Additionally, this commit moves the call to the builtin to be done right before collecting the partial stat to make it easier to see.

Fixes: #141979.

Release note: None

Co-authored-by: Yahor Yuzefovich <[email protected]>

Author: craig[bot]

pkg/sql/conn_executor_exec.go

@@ -2966,8 +2966,6 @@ func (ex *connExecutor) dispatchToExecutionEngine(
 		planner.curPlan.flags.Set(planFlagFullyDistributed)
 	case physicalplan.PartiallyDistributedPlan:
 		planner.curPlan.flags.Set(planFlagPartiallyDistributed)
-	default:
-		planner.curPlan.flags.Set(planFlagNotDistributed)
 	}
 
 	ex.sessionTracing.TraceRetryInformation(ctx, int(ex.state.mu.autoRetryCounter), ex.state.mu.autoRetryReason)

pkg/sql/distsql_physical_planner.go

@@ -1115,7 +1115,7 @@ func getDefaultSaveFlowsFunc(
 		var explainVecVerbose []string
 		if planner.instrumentation.collectBundle && vectorized {
 			flowCtx := newFlowCtxForExplainPurposes(ctx, planner)
-			flowCtx.Local = !planner.curPlan.flags.IsDistributed()
+			flowCtx.Local = !planner.curPlan.flags.ShouldBeDistributed()
 			getExplain := func(verbose bool) []string {
 				gatewaySQLInstanceID := planner.extendedEvalCtx.DistSQLPlanner.gatewaySQLInstanceID
 				// When we're collecting the bundle, we're always recording the

pkg/sql/executor_statement_metrics.go

@@ -153,7 +153,7 @@ func (ex *connExecutor) recordStatementSummary(
 	recordedStmtStatsKey := appstatspb.StatementStatisticsKey{
 		Query:        stmt.StmtNoConstants,
 		QuerySummary: stmt.StmtSummary,
-		DistSQL:      flags.IsDistributed(),
+		DistSQL:      flags.ShouldBeDistributed(),
 		Vec:          flags.IsSet(planFlagVectorized),
 		ImplicitTxn:  flags.IsSet(planFlagImplicitTxn),
 		FullScan:     fullScan,
@@ -299,7 +299,7 @@ func (ex *connExecutor) recordStatementLatencyMetrics(
 
 		m.StatementFingerprintCount.Add([]byte(stmt.StmtNoConstants))
 
-		if flags.IsDistributed() {
+		if flags.ShouldBeDistributed() {
 			if _, ok := stmt.AST.(*tree.Select); ok {
 				m.DistSQLSelectCount.Inc(1)
 				if flags.IsSet(planFlagDistributedExecution) {

pkg/sql/explain_test.go

@@ -349,7 +349,7 @@ func TestExplainKVInfo(t *testing.T) {
 			}
 
 			scanQuery := "SELECT count(*) FROM ab"
-			info := getKVInfo(t, r, scanQuery)
+			info, fullOutput := getKVInfo(t, r, scanQuery)
 
 			assert.Equal(t, 1, info.counters[kvNodes])
 			assert.Equal(t, 1000, info.counters[rowsRead])
@@ -358,9 +358,14 @@ func TestExplainKVInfo(t *testing.T) {
 			assert.Equal(t, 1, info.counters[gRPCCalls])
 			assert.Equal(t, 1000, info.counters[stepCount])
 			assert.Equal(t, 1, info.counters[seekCount])
+			// Additionally assert that correct distribution is shown.
+			assert.Truef(
+				t, strings.Contains(fullOutput, "distribution: local"),
+				"expected local distribution, found\n\n%s", fullOutput,
+			)
 
 			lookupJoinQuery := "SELECT count(*) FROM ab INNER LOOKUP JOIN bc ON ab.b = bc.b"
-			info = getKVInfo(t, r, lookupJoinQuery)
+			info, fullOutput = getKVInfo(t, r, lookupJoinQuery)
 
 			assert.Equal(t, 1, info.counters[kvNodes])
 			assert.Equal(t, 1000, info.counters[rowsRead])
@@ -369,6 +374,10 @@ func TestExplainKVInfo(t *testing.T) {
 			assert.Equal(t, 1, info.counters[gRPCCalls])
 			assert.Equal(t, 0, info.counters[stepCount])
 			assert.Equal(t, 1000, info.counters[seekCount])
+			assert.Truef(
+				t, strings.Contains(fullOutput, "distribution: local"),
+				"expected local distribution\n\n%s", fullOutput,
+			)
 		}
 	}
 }
@@ -406,7 +415,10 @@ func init() {
 // of the given query from the top-most operator in the plan (i.e. if there are
 // multiple operators exposing the scan stats, then the first info that appears
 // in the EXPLAIN output is used).
-func getKVInfo(t *testing.T, r *sqlutils.SQLRunner, query string) kvInfo {
+//
+// It additionally returns the full output of EXPLAIN ANALYZE of the given
+// query.
+func getKVInfo(t *testing.T, r *sqlutils.SQLRunner, query string) (_ kvInfo, fullOutput string) {
 	rows := r.Query(t, "EXPLAIN ANALYZE (VERBOSE) "+query)
 	var output strings.Builder
 	var str string
@@ -442,7 +454,7 @@ func getKVInfo(t *testing.T, r *sqlutils.SQLRunner, query string) kvInfo {
 		fmt.Println("Explain output:")
 		fmt.Println(output.String())
 	}
-	return info
+	return info, output.String()
 }
 
 // TestExplainAnalyzeWarnings verifies that warnings are printed whenever the

pkg/sql/logictest/testdata/logic_test/distsql_stats

@@ -2424,6 +2424,11 @@ INSERT INTO abcd VALUES
 statement ok
 INSERT INTO xy VALUES (-1, 9), (-2, 8), (5, 15), (6, 16)
 
+# Clear the stat cache so that creating partial statistics has access to the
+# latest full statistic.
+statement ok
+SELECT crdb_internal.clear_table_stats_cache();
+
 statement error pgcode 0A000 creating partial statistics with a WHERE clause is not yet supported
 CREATE STATISTICS abcd_a_partial ON a FROM abcd WHERE a > 1;
 
@@ -2539,14 +2544,14 @@ INSERT INTO a_null VALUES (NULL), (1), (2);
 statement ok
 CREATE STATISTICS a_null_stat ON a FROM a_null;
 
+statement ok
+INSERT INTO a_null VALUES (NULL), (NULL), (NULL);
+
 # Clear the stat cache so that creating partial statistics has access to the
 # latest full statistic.
 statement ok
 SELECT crdb_internal.clear_table_stats_cache();
 
-statement ok
-INSERT INTO a_null VALUES (NULL), (NULL), (NULL);
-
 statement ok
 CREATE STATISTICS a_null_stat_partial ON a FROM a_null USING EXTREMES;
 
@@ -2577,14 +2582,14 @@ INSERT INTO d_desc VALUES (1, 10), (2, 20), (3, 30), (4, 40);
 statement ok
 CREATE STATISTICS sd ON a FROM d_desc;
 
+statement ok
+INSERT INTO d_desc VALUES (0, 0), (5, 50);
+
 # Clear the stat cache so that creating partial statistics has access to the
 # latest full statistic.
 statement ok
 SELECT crdb_internal.clear_table_stats_cache();
 
-statement ok
-INSERT INTO d_desc VALUES (0, 0), (5, 50);
-
 statement ok
 CREATE STATISTICS sdp ON a FROM d_desc USING EXTREMES;
 
@@ -3016,6 +3021,11 @@ ANALYZE t130817;
 statement ok
 INSERT INTO t68254 (a, b, c) VALUES (5, '5', '{"foo": {"bar": {"baz": 5}}}')
 
+# Clear the stat cache so that creating partial statistics has access to the
+# latest full statistic.
+statement ok
+SELECT crdb_internal.clear_table_stats_cache();
+
 statement ok
 CREATE STATISTICS j6 ON d FROM t68254 USING EXTREMES
 
@@ -3081,10 +3091,12 @@ ALTER TABLE int_outer_buckets INJECT STATISTICS '[
 ]'
 
 statement ok
-SELECT crdb_internal.clear_table_stats_cache();
+INSERT INTO int_outer_buckets SELECT generate_series(-10, -1) UNION ALL SELECT generate_series(10000, 10009);
 
+# Clear the stat cache so that creating partial statistics has access to the
+# latest full statistic.
 statement ok
-INSERT INTO int_outer_buckets SELECT generate_series(-10, -1) UNION ALL SELECT generate_series(10000, 10009);
+SELECT crdb_internal.clear_table_stats_cache();
 
 statement ok
 CREATE STATISTICS int_outer_buckets_partial ON a FROM int_outer_buckets USING EXTREMES;
@@ -3143,11 +3155,6 @@ INSERT INTO timestamp_outer_buckets VALUES
 statement ok
 CREATE STATISTICS timestamp_outer_buckets_full ON a FROM timestamp_outer_buckets;
 
-# Clear the stat cache so that creating partial statistics has access to the
-# latest full statistic.
-statement ok
-SELECT crdb_internal.clear_table_stats_cache();
-
 let $hist_id_timestamp_outer_buckets_full
 SELECT histogram_id FROM [SHOW STATISTICS FOR TABLE timestamp_outer_buckets] WHERE statistics_name = 'timestamp_outer_buckets_full'
 
@@ -3162,6 +3169,11 @@ INSERT INTO timestamp_outer_buckets VALUES
   ('2024-06-26 00:00:00'),
   ('2024-06-27 03:30:00');
 
+# Clear the stat cache so that creating partial statistics has access to the
+# latest full statistic.
+statement ok
+SELECT crdb_internal.clear_table_stats_cache();
+
 statement ok
 CREATE STATISTICS timestamp_outer_buckets_partial ON a FROM timestamp_outer_buckets USING EXTREMES;
 
@@ -3233,6 +3245,11 @@ ALTER TABLE timestamp_outer_buckets INJECT STATISTICS '[
 statement ok
 INSERT INTO timestamp_outer_buckets VALUES ('2024-06-28 01:00:00');
 
+# Clear the stat cache so that creating partial statistics has access to the
+# latest full statistic.
+statement ok
+SELECT crdb_internal.clear_table_stats_cache();
+
 statement ok
 CREATE STATISTICS timestamp_outer_buckets_partial ON a FROM timestamp_outer_buckets USING EXTREMES;
 
@@ -3615,6 +3632,11 @@ INSERT INTO pstat_allindex VALUES
   (6, 6, 6, 6, 6, 6, '{"6": "6"}'),
   (7, 7, 7, 7, 7, 7, '{"7": "7"}');
 
+# Clear the stat cache so that creating partial statistics has access to the
+# latest full statistic.
+statement ok
+SELECT crdb_internal.clear_table_stats_cache();
+
 statement ok
 CREATE STATISTICS pstat_allindex_partial FROM pstat_allindex USING EXTREMES;
 

pkg/sql/plan.go

@@ -510,10 +510,16 @@ func (p *planTop) init(stmt *Statement, instrumentation *instrumentationHelper)
 func (p *planTop) savePlanInfo() {
 	vectorized := p.flags.IsSet(planFlagVectorized)
 	distribution := physicalplan.LocalPlan
-	if p.flags.IsSet(planFlagFullyDistributed) {
-		distribution = physicalplan.FullyDistributedPlan
-	} else if p.flags.IsSet(planFlagPartiallyDistributed) {
-		distribution = physicalplan.PartiallyDistributedPlan
+	if p.flags.IsSet(planFlagDistributedExecution) {
+		// Only show that the plan was distributed if we actually had
+		// distributed execution. This matches the logic from explainPlanNode
+		// where we use the actual physical plan's distribution rather than the
+		// physical planning heuristic.
+		if p.flags.IsSet(planFlagFullyDistributed) {
+			distribution = physicalplan.FullyDistributedPlan
+		} else if p.flags.IsSet(planFlagPartiallyDistributed) {
+			distribution = physicalplan.PartiallyDistributedPlan
+		}
 	}
 	containsMutation := p.flags.IsSet(planFlagContainsMutation)
 	generic := p.flags.IsSet(planFlagGeneric)
@@ -610,17 +616,18 @@ const (
 	planFlagOptCacheMiss
 
 	// planFlagFullyDistributed is set if the query is planned to use full
-	// distribution.
+	// distribution. This flag indicates that the query is such that it can be
+	// distributed, and we think it's worth doing so; however, due to range
+	// placement and other physical planning decisions, the plan might still end
+	// up being local. See planFlagDistributedExecution if interested in whether
+	// the physical plan actually ends up being distributed.
 	planFlagFullyDistributed
 
-	// planFlagPartiallyDistributed is set if the query is planned to use partial
-	// distribution (see physicalplan.PartiallyDistributedPlan).
+	// planFlagPartiallyDistributed is set if the query is planned to use
+	// partial distribution (see physicalplan.PartiallyDistributedPlan). Same
+	// caveats apply as for planFlagFullyDistributed.
 	planFlagPartiallyDistributed
 
-	// planFlagNotDistributed is set if the query is planned to not use
-	// distribution.
-	planFlagNotDistributed
-
 	// planFlagImplicitTxn marks that the plan was run inside of an implicit
 	// transaction.
 	planFlagImplicitTxn
@@ -707,8 +714,10 @@ func (pf *planFlags) Unset(flags planFlags) {
 	*pf &^= flags
 }
 
-// IsDistributed returns true if either the fully or the partially distributed
-// flags is set.
-func (pf planFlags) IsDistributed() bool {
+// ShouldBeDistributed returns true if either fully distributed or partially
+// distributed flag is set. In other words, it returns whether the plan should
+// be distributed (we might end up not distributing it though due to range
+// placement or moving the single flow to the gateway).
+func (pf planFlags) ShouldBeDistributed() bool {
 	return pf&(planFlagFullyDistributed|planFlagPartiallyDistributed) != 0
 }

pkg/sql/testdata/telemetry_logging/logging/force_logging

@@ -31,7 +31,7 @@ SELECT * FROM t LIMIT 1;
 {
 	"ApplicationName": "telemetry-logging-datadriven",
 	"Database": "defaultdb",
-	"Distribution": "full",
+	"Distribution": "local",
 	"EventType": "sampled_query",
 	"PlanGist": "AgHQAQIAAAAAAg==",
 	"ScanCount": 1,
@@ -164,7 +164,7 @@ SELECT * FROM t LIMIT 1;
 {
 	"ApplicationName": "$ internal-console-app",
 	"Database": "defaultdb",
-	"Distribution": "full",
+	"Distribution": "local",
 	"EventType": "sampled_query",
 	"PlanGist": "AgHQAQIAAAAAAg==",
 	"ScanCount": 1,

pkg/sql/testdata/telemetry_logging/logging/transaction_mode

@@ -208,7 +208,7 @@ SELECT * FROM t LIMIT 2
 {
 	"ApplicationName": "telemetry-logging-datadriven",
 	"Database": "defaultdb",
-	"Distribution": "full",
+	"Distribution": "local",
 	"EventType": "sampled_query",
 	"PlanGist": "AgHQAQQAAAAAAg==",
 	"ScanCount": 1,
@@ -246,7 +246,7 @@ BEGIN; SELECT * FROM t LIMIT 4; SELECT * FROM t LIMIT 5; COMMIT
 {
 	"ApplicationName": "telemetry-logging-datadriven",
 	"Database": "defaultdb",
-	"Distribution": "full",
+	"Distribution": "local",
 	"EventType": "sampled_query",
 	"PlanGist": "AgHQAQQAAAAAAg==",
 	"ScanCount": 1,
@@ -260,7 +260,7 @@ BEGIN; SELECT * FROM t LIMIT 4; SELECT * FROM t LIMIT 5; COMMIT
 {
 	"ApplicationName": "telemetry-logging-datadriven",
 	"Database": "defaultdb",
-	"Distribution": "full",
+	"Distribution": "local",
 	"EventType": "sampled_query",
 	"PlanGist": "AgHQAQQAAAAAAg==",
 	"ScanCount": 1,