sql: move notifying StatsRefresh of new table to post-commit hook

Prior to this change, the StatsRefresher was being notified of a new table
during the execution of the createTable node, before the creating transaction
had committed. Prior to cockroachdb#46170, the StatsRefresher was likely to block on the
intent of the creating transaction. After that change, the StatsRefresher might
not discover the new table because the creating transaction is much more likely
to get pushed further into the future, past the read of the stats refresher.

Release note (bug fix): Fix rare bug where stats were not automatically
generated for a new table.

pkg/sql/conn_executor.go

@@ -2152,6 +2152,8 @@ func (ex *connExecutor) txnStateTransitionsApplyWrapper(
 				res.SetError(newErr)
 			}
 		}
+		ex.notifyStatsRefresherOfNewTables(ex.Ctx())
+
 		if err := ex.server.cfg.JobRegistry.Run(
 			ex.ctxHolder.connCtx,
 			ex.server.cfg.InternalExecutor,
@@ -2320,6 +2322,23 @@ func (ex *connExecutor) sessionEventf(ctx context.Context, format string, args .
 	}
 }
 
+// notifyStatsRefresherOfNewTables is called on txn commit to inform
+// the stats refresher that new tables exist and should have their stats
+// collected now.
+func (ex *connExecutor) notifyStatsRefresherOfNewTables(ctx context.Context) {
+	for _, desc := range ex.extraTxnState.tables.getNewTables() {
+		// The CREATE STATISTICS run for an async CTAS query is initiated by the
+		// SchemaChanger, so we don't do it here.
+		if desc.IsTable() && !desc.IsAs() {
+			// Initiate a run of CREATE STATISTICS. We use a large number
+			// for rowsAffected because we want to make sure that stats always get
+			// created/refreshed here.
+			ex.planner.execCfg.StatsRefresher.
+				NotifyMutation(desc.ID, math.MaxInt32 /* rowsAffected */)
+		}
+	}
+}
+
 // StatementCounters groups metrics for counting different types of
 // statements.
 type StatementCounters struct {

pkg/sql/create_table.go

@@ -15,7 +15,6 @@ import (
 	"context"
 	"fmt"
 	"go/constant"
-	"math"
 	"sort"
 	"strconv"
 	"strings"
@@ -472,16 +471,6 @@ func (n *createTableNode) startExec(params runParams) error {
 		}
 	}
 
-	// The CREATE STATISTICS run for an async CTAS query is initiated by the
-	// SchemaChanger.
-	if n.n.As() && params.p.autoCommit {
-		return nil
-	}
-
-	// Initiate a run of CREATE STATISTICS. We use a large number
-	// for rowsAffected because we want to make sure that stats always get
-	// created/refreshed here.
-	params.ExecCfg().StatsRefresher.NotifyMutation(desc.ID, math.MaxInt32 /* rowsAffected */)
 	return nil
 }
 

pkg/sql/table.go

@@ -604,6 +604,15 @@ func (tc *TableCollection) getTablesWithNewVersion() []IDVersion {
 	return tables
 }
 
+func (tc *TableCollection) getNewTables() (newTables []*ImmutableTableDescriptor) {
+	for _, table := range tc.uncommittedTables {
+		if mut := table.MutableTableDescriptor; mut.IsNewTable() {
+			newTables = append(newTables, table.ImmutableTableDescriptor)
+		}
+	}
+	return newTables
+}
+
 type dbAction bool
 
 const (