fix(frontend): publish ANALYZE stats by table generation

Makefile

@@ -1315,11 +1315,13 @@ install-static-check-tools:
 	@go install github.com/apache/skywalking-eyes/cmd/license-eye@v0.4.0
 
 .PHONY: static-check
+GOLANGCI_LINT_CONCURRENCY ?=
+GOLANGCI_LINT_CONCURRENCY_FLAG := $(if $(strip $(GOLANGCI_LINT_CONCURRENCY)),--concurrency $(strip $(GOLANGCI_LINT_CONCURRENCY)))
 static-check: config err-check
 	$(CGO_OPTS) go vet $(GO_MODULE_MODE) -vettool=`which molint` ./...
 	$(CGO_OPTS) license-eye -c .licenserc.yml header check
 	$(CGO_OPTS) license-eye -c .licenserc.yml dep check
-	$(CGO_OPTS) golangci-lint run -v -c .golangci.yml ./...
+	$(CGO_OPTS) golangci-lint run -v $(GOLANGCI_LINT_CONCURRENCY_FLAG) -c .golangci.yml ./...
 
 fmtErrs := $(shell grep -onr 'fmt.Errorf' pkg/ --exclude-dir=.git --exclude-dir=vendor \
 				--exclude=*.pb.go --exclude=*_test.go --exclude=system_vars.go --exclude=Makefile)

pkg/frontend/compiler_context.go

@@ -162,6 +162,23 @@ func (tcc *TxnCompilerContext) GetStatsCache() *plan2.StatsCache {
 	return tcc.execCtx.ses.GetStatsCache()
 }
 
+func (tcc *TxnCompilerContext) getStatsCacheVersion(
+	key optimizerStatsTableKey,
+) (*Session, *plan2.StatsCache, uint64) {
+	tcc.mu.Lock()
+	feSes := tcc.execCtx.ses
+	txnWrapper, _ := tcc.tcw.(*TxnComputationWrapper)
+	tcc.mu.Unlock()
+	if ses, ok := feSes.(*Session); ok {
+		cache, version := ses.getStatsCacheWithVersion(key)
+		if txnWrapper != nil {
+			txnWrapper.recordOptimizerStatsVersion(key, version)
+		}
+		return ses, cache, version
+	}
+	return nil, feSes.GetStatsCache(), 0
+}
+
 func InitTxnCompilerContext(db string) *TxnCompilerContext {
 	return &TxnCompilerContext{dbName: db}
 }
@@ -252,32 +269,61 @@ func (tcc *TxnCompilerContext) GetAccountId() (uint32, error) {
 
 // ResolveViewDependencyAccount returns the account whose catalog namespace was
 // used to resolve a View dependency. Keep the override order aligned with
-// getRelation: snapshot tenant, subscription publisher, then relations that
-// are always read from the system account.
+// getRelation: snapshot tenant, cluster-table name override, subscription
+// publisher, then relations that are always read from the system account.
 func (tcc *TxnCompilerContext) ResolveViewDependencyAccount(
 	obj *plan2.ObjectRef,
 	tableDef *plan2.TableDef,
 	snapshot *plan2.Snapshot,
 ) (uint32, error) {
+	return tcc.resolvePhysicalObjectAccount(obj, tableDef, snapshot), nil
+}
+
+// resolvePhysicalObjectAccount keeps statistics and view dependencies aligned
+// with the account context used by getRelation. The identity must be resolved
+// before consulting any cache so cached data and its generation share one key.
+func (tcc *TxnCompilerContext) resolvePhysicalObjectAccount(
+	obj *plan2.ObjectRef,
+	tableDef *plan2.TableDef,
+	snapshot *plan2.Snapshot,
+) uint32 {
 	accountID := tcc.execCtx.ses.GetAccountId()
 	if snapshot != nil && snapshot.Tenant != nil {
 		accountID = snapshot.Tenant.TenantID
 	}
-	if obj.PubInfo != nil {
-		accountID = uint32(obj.PubInfo.TenantId)
-	}
 
-	dbName, tableName := obj.SchemaName, obj.ObjName
-	if dbName == "" {
+	var dbName, tableName string
+	if obj != nil {
+		dbName, tableName = obj.SchemaName, obj.ObjName
+	}
+	if dbName == "" && tableDef != nil {
 		dbName = tableDef.DbName
 	}
-	if tableName == "" {
+	if tableName == "" && tableDef != nil {
 		tableName = tableDef.Name
 	}
-	if isClusterTable(dbName, tableName) || ShouldSwitchToSysAccount(dbName, tableName) {
+	if isClusterTable(dbName, tableName) {
 		accountID = sysAccountID
 	}
-	return accountID, nil
+	// getRelation applies publication ownership after the generic cluster-table
+	// name rule, so the publisher remains the physical owner in that overlap.
+	if obj != nil && obj.PubInfo != nil {
+		accountID = uint32(obj.PubInfo.TenantId)
+	}
+	if ShouldSwitchToSysAccount(dbName, tableName) {
+		accountID = sysAccountID
+	}
+	return accountID
+}
+
+func (tcc *TxnCompilerContext) optimizerStatsKey(
+	obj *plan2.ObjectRef,
+	snapshot *plan2.Snapshot,
+) optimizerStatsTableKey {
+	return optimizerStatsTableKey{
+		accountID: tcc.resolvePhysicalObjectAccount(obj, nil, snapshot),
+		tableID:   uint64(obj.Obj),
+	}
 }
 
 func (tcc *TxnCompilerContext) GetAccountName() string {
@@ -1125,10 +1171,12 @@ func (tcc *TxnCompilerContext) Stats(obj *plan2.ObjectRef, snapshot *plan2.Snaps
 	}()
 
 	tableID := uint64(obj.Obj)
+	statsKey := tcc.optimizerStatsKey(obj, snapshot)
+	ses, statsCache, statsVersion := tcc.getStatsCacheVersion(statsKey)
 
 	// Fast path: return cached result if visited within 3 seconds AND stats is valid
 	// Stats is valid if AccurateObjectNumber > 0 (meaning we have real data)
-	if w := tcc.GetStatsCache().Get(tableID); w.Exists() {
+	if w := statsCache.Get(tableID); w.Exists() {
 		if time.Now().Unix()-w.GetLastVisit() < 3 {
 			s := w.GetStats()
 			if s != nil && s.AccurateObjectNumber > 0 {
@@ -1144,8 +1192,13 @@ func (tcc *TxnCompilerContext) Stats(obj *plan2.ObjectRef, snapshot *plan2.Snaps
 		return nil, err
 	}
 
-	// Cache the result
-	tcc.GetStatsCache().Set(tableID, result)
+	// A refresh may have completed while the slow path was reading storage. Do
+	// not let work from the old generation repopulate the new session cache.
+	if ses == nil {
+		statsCache.Set(tableID, result)
+	} else {
+		ses.cacheStatsIfCurrent(statsKey, statsVersion, result)
+	}
 
 	return result, nil
 }

pkg/frontend/compiler_context_test.go

@@ -103,6 +103,7 @@ func TestResolveViewDependencyAccount(t *testing.T) {
 	for _, test := range []struct {
 		name     string
 		obj      *pbplan.ObjectRef
+		tableDef *pbplan.TableDef
 		snapshot *pbplan.Snapshot
 		want     uint32
 	}{
@@ -115,14 +116,23 @@ func TestResolveViewDependencyAccount(t *testing.T) {
 			PubInfo: &pbplan.PubInfo{TenantId: 9}},
 			snapshot: &pbplan.Snapshot{Tenant: &pbplan.SnapshotTenant{TenantID: 8}}, want: 9},
 		{name: "cluster table", obj: &pbplan.ObjectRef{SchemaName: catalog.MO_CATALOG, ObjName: "cluster_table"}, want: 0},
+		{name: "relation kind alone keeps tenant context", obj: &pbplan.ObjectRef{SchemaName: "db", ObjName: "cluster_table"},
+			tableDef: &pbplan.TableDef{TableType: catalog.SystemClusterRel}, want: 7},
+		{name: "publication overrides generic cluster name", obj: &pbplan.ObjectRef{
+			SchemaName: catalog.MO_CATALOG, ObjName: "cluster_table",
+			PubInfo: &pbplan.PubInfo{TenantId: 9}}, want: 9},
 		{name: "statement info", obj: &pbplan.ObjectRef{SchemaName: catalog.MO_SYSTEM, ObjName: catalog.MO_STATEMENT}, want: 0},
 		{name: "system relation overrides publisher", obj: &pbplan.ObjectRef{SchemaName: catalog.MO_SYSTEM,
 			ObjName: catalog.MO_STATEMENT, PubInfo: &pbplan.PubInfo{TenantId: 9}}, want: 0},
 		{name: "metric", obj: &pbplan.ObjectRef{SchemaName: catalog.MO_SYSTEM_METRICS, ObjName: catalog.MO_METRIC}, want: 0},
 		{name: "sql statement cu", obj: &pbplan.ObjectRef{SchemaName: catalog.MO_SYSTEM_METRICS, ObjName: catalog.MO_SQL_STMT_CU}, want: 0},
 	} {
 		t.Run(test.name, func(t *testing.T) {
-			got, err := tcc.ResolveViewDependencyAccount(test.obj, &pbplan.TableDef{}, test.snapshot)
+			tableDef := test.tableDef
+			if tableDef == nil {
+				tableDef = &pbplan.TableDef{}
+			}
+			got, err := tcc.ResolveViewDependencyAccount(test.obj, tableDef, test.snapshot)
 			require.NoError(t, err)
 			require.Equal(t, test.want, got)
 		})

pkg/frontend/computation_wrapper.go

@@ -116,9 +116,11 @@ type TxnComputationWrapper struct {
 	hasPreparedSchedulingSQLMode bool
 	preparedSchedulingSQL        string
 
-	// protocolVersion is captured when plan is built. The session plan cache
-	// uses it instead of the version observed later when execution completes.
-	protocolVersion int64
+	// protocolVersion and optimizerStatsVersions are captured when the plan is
+	// built. The session plan cache uses them instead of values observed later
+	// when execution completes.
+	protocolVersion        int64
+	optimizerStatsVersions map[optimizerStatsTableKey]uint64
 
 	// A reusable logical plan and its generation snapshot are one immutable
 	// binding. cachedPlan* identifies the session-cache slot so a definition
@@ -245,6 +247,7 @@ func (cwft *TxnComputationWrapper) Clear() {
 	cwft.preparedSchedulingSQLMode = ""
 	cwft.hasPreparedSchedulingSQLMode = false
 	cwft.preparedSchedulingSQL = ""
+	cwft.optimizerStatsVersions = nil
 	cwft.planSnapshotTS = timestamp.Timestamp{}
 	cwft.hasPlanSnapshotTS = false
 	cwft.planGenerationReused = false
@@ -254,6 +257,17 @@ func (cwft *TxnComputationWrapper) Clear() {
 	cwft.schedulingTrace.Reset()
 }
 
+func (cwft *TxnComputationWrapper) recordOptimizerStatsVersion(key optimizerStatsTableKey, version uint64) {
+	if cwft.optimizerStatsVersions == nil {
+		cwft.optimizerStatsVersions = make(map[optimizerStatsTableKey]uint64)
+	}
+	// Keep the first observed version. If publication happens between repeated
+	// reads, admission against the newer current version will reject the plan.
+	if _, exists := cwft.optimizerStatsVersions[key]; !exists {
+		cwft.optimizerStatsVersions[key] = version
+	}
+}
+
 func (cwft *TxnComputationWrapper) ParamVals() []any {
 	return cwft.paramVals
 }
@@ -363,6 +377,7 @@ func (cwft *TxnComputationWrapper) Compile(any any, fill func(*batch.Batch, *per
 	cacheHit := cwft.plan != nil
 	if !cacheHit {
 		cwft.protocolVersion = currentProtocolVersion(cwft.proc)
+		clear(cwft.optimizerStatsVersions)
 		cwft.plan, err = buildPlanWithPrepareMode(
 			execCtx.reqCtx,
 			cwft.ses,
@@ -725,6 +740,7 @@ func (cwft *TxnComputationWrapper) completeCompileExecution(
 					cwft.cachedPlanGeneration,
 					cwft.plan,
 					cwft.planSnapshotTS,
+					cwft.optimizerStatsVersions,
 				)
 			}
 			if !updated {