sql: introduce canary full stats rollout and stats_as_of session var

pkg/sql/catalog/descriptor.go

@@ -7,6 +7,7 @@ package catalog
 
 import (
 	"context"
+	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
@@ -878,6 +879,9 @@ type TableDescriptor interface {
 	// security for the table and false if it is no force. When forced is
 	// set the table's RLS policies are enforced even on the table owner.
 	IsRowLevelSecurityForced() bool
+	// GetStatsCanaryWindow returns the canary statistics rollout duration.
+	// See TableDescriptor.StatsCanaryWindow for details.
+	GetStatsCanaryWindow() time.Duration
 }
 
 // MutableTableDescriptor is both a MutableDescriptor and a TableDescriptor.

pkg/sql/catalog/tabledesc/table_desc.go

@@ -7,6 +7,8 @@
 package tabledesc
 
 import (
+	"time"
+
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catenumpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catpb"
@@ -740,3 +742,8 @@ func (desc *wrapper) IsRowLevelSecurityEnabled() bool {
 func (desc *wrapper) IsRowLevelSecurityForced() bool {
 	return desc.RowLevelSecurityForced
 }
+
+// GetStatsCanaryWindow implements the TableDescriptor interface.
+func (desc *wrapper) GetStatsCanaryWindow() time.Duration {
+	return desc.StatsCanaryWindow
+}

pkg/sql/logictest/testdata/logic_test/information_schema

@@ -4258,6 +4258,7 @@ sql_safe_updates                                                 off
 ssl                                                              on
 standard_conforming_strings                                      on
 statement_timeout                                                0
+stats_as_of                                                      ·
 streamer_always_maintain_ordering                                off
 streamer_enabled                                                 on
 streamer_head_of_line_only_fraction                              0.8

pkg/sql/logictest/testdata/logic_test/pg_catalog

@@ -3207,6 +3207,7 @@ sql_safe_updates                                                 off
 ssl                                                              on                  NULL      NULL        NULL        string
 standard_conforming_strings                                      on                  NULL      NULL        NULL        string
 statement_timeout                                                0                   NULL      NULL        NULL        string
+stats_as_of                                                      ·                   NULL      NULL        NULL        string
 streamer_always_maintain_ordering                                off                 NULL      NULL        NULL        string
 streamer_enabled                                                 on                  NULL      NULL        NULL        string
 streamer_head_of_line_only_fraction                              0.8                 NULL      NULL        NULL        string
@@ -3456,6 +3457,7 @@ sql_safe_updates                                                 off
 ssl                                                              on                  NULL  user     NULL      on                  on
 standard_conforming_strings                                      on                  NULL  user     NULL      on                  on
 statement_timeout                                                0                   ms    user     NULL      0s                  0s
+stats_as_of                                                      ·                   NULL  user     NULL      ·                   ·
 streamer_always_maintain_ordering                                off                 NULL  user     NULL      off                 off
 streamer_enabled                                                 on                  NULL  user     NULL      on                  on
 streamer_head_of_line_only_fraction                              0.8                 NULL  user     NULL      0.8                 0.8
@@ -3697,6 +3699,7 @@ sql_safe_updates                                                 NULL    NULL
 ssl                                                              NULL    NULL     NULL     NULL        NULL
 standard_conforming_strings                                      NULL    NULL     NULL     NULL        NULL
 statement_timeout                                                NULL    NULL     NULL     NULL        NULL
+stats_as_of                                                      NULL    NULL     NULL     NULL        NULL
 streamer_always_maintain_ordering                                NULL    NULL     NULL     NULL        NULL
 streamer_enabled                                                 NULL    NULL     NULL     NULL        NULL
 streamer_head_of_line_only_fraction                              NULL    NULL     NULL     NULL        NULL

pkg/sql/logictest/testdata/logic_test/show_source

@@ -224,6 +224,7 @@ sql_safe_updates                                                 off
 ssl                                                              on
 standard_conforming_strings                                      on
 statement_timeout                                                0
+stats_as_of                                                      ·
 streamer_always_maintain_ordering                                off
 streamer_enabled                                                 on
 streamer_head_of_line_only_fraction                              0.8

pkg/sql/opt/cat/table.go

@@ -205,6 +205,9 @@ type Table interface {
 
 	// Policies returns all the policies defined for this table.
 	Policies() *Policies
+
+	// StatsCanaryWindow returns the canary window size for the table.
+	StatsCanaryWindow() time.Duration
 }
 
 // CheckConstraint represents a check constraint on a table. Check constraints

pkg/sql/opt/exec/explain/plan_gist_factory.go

@@ -10,6 +10,7 @@ import (
 	"context"
 	b64 "encoding/base64"
 	"encoding/binary"
+	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/geo/geopb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
@@ -682,6 +683,9 @@ func (u *unknownTable) IsRowLevelSecurityEnabled() bool { return false }
 // IsRowLevelSecurityForced is part of the cat.Table interface
 func (u *unknownTable) IsRowLevelSecurityForced() bool { return false }
 
+// StatsCanaryWindow is part of the cat.Table interface
+func (u *unknownTable) StatsCanaryWindow() time.Duration { return 0 }
+
 // Policies is part of the cat.Table interface.
 func (u *unknownTable) Policies() *cat.Policies { return nil }
 

pkg/sql/opt/memo/BUILD.bazel

@@ -47,11 +47,13 @@ go_library(
         "//pkg/sql/sem/tree",
         "//pkg/sql/sem/tree/treewindow",
         "//pkg/sql/sem/volatility",
+        "//pkg/sql/sessiondata",
         "//pkg/sql/stats",
         "//pkg/sql/types",
         "//pkg/util/buildutil",
         "//pkg/util/duration",
         "//pkg/util/encoding",
+        "//pkg/util/hlc",
         "//pkg/util/intsets",
         "//pkg/util/iterutil",
         "//pkg/util/json",

pkg/sql/opt/memo/statistics_builder.go

@@ -9,6 +9,7 @@ import (
 	"context"
 	"math"
 	"reflect"
+	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/geo/geoindex"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/colinfo"
@@ -20,9 +21,11 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/eval"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/idxtype"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
+	"github.com/cockroachdb/cockroach/pkg/sql/sessiondata"
 	"github.com/cockroachdb/cockroach/pkg/sql/stats"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
 	"github.com/cockroachdb/cockroach/pkg/util/buildutil"
+	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/json"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/errors"
@@ -647,6 +650,32 @@ func GetTableStats(md *opt.Metadata, tabID opt.TableID) (*props.Statistics, bool
 	return stats, ok
 }
 
+func isEligibleFullStats(
+	stat cat.TableStatistic, sd *sessiondata.SessionData, asOf hlc.Timestamp,
+) bool {
+	createdAtTS := hlc.Timestamp{WallTime: stat.CreatedAt().UnixNano()}
+	if stat.IsPartial() ||
+		stat.IsMerged() && !sd.OptimizerUseMergedPartialStatistics ||
+		stat.IsForecast() && !sd.OptimizerUseForecasts || createdAtTS.After(asOf) {
+		return false
+	}
+	return true
+}
+
+// getNextFullStats returns the index of the next eligible full statistics.
+func getNextFullStats(
+	start int, tab cat.Table, sd *sessiondata.SessionData, asOf hlc.Timestamp,
+) int {
+	for i := start; i < tab.StatisticCount(); i++ {
+		stat := tab.Statistic(i)
+		if !isEligibleFullStats(stat, sd, asOf) {
+			continue
+		}
+		return i
+	}
+	return -1
+}
+
 // makeTableStatistics returns the available statistics for the given table.
 // Statistics are derived lazily and are cached in the metadata, since they may
 // be accessed multiple times during query optimization. For more details, see
@@ -659,6 +688,7 @@ func (sb *statisticsBuilder) makeTableStatistics(tabID opt.TableID) *props.Stati
 	}
 
 	tab := sb.md.Table(tabID)
+	tabMeta := sb.md.TableMeta(tabID)
 	// Create a mapping from table column ordinals to inverted index column
 	// ordinals. This allows us to do a fast lookup while iterating over all
 	// stats from a statistic's column to any associated inverted columns.
@@ -680,16 +710,55 @@ func (sb *statisticsBuilder) makeTableStatistics(tabID opt.TableID) *props.Stati
 	// Make now and annotate the metadata table with it for next time.
 	stats = &props.Statistics{}
 
+	useCanary := sb.evalCtx.UseCanaryStats
+	var skippedStatsCreationTimestamp time.Time
+	statsCanaryWindow := tabMeta.StatsCanaryWindow
+
+	// TODO(janexing): should we use clock.Now() or the StmtTimestamp as
+	// the default? Or avoid this issue by only setting it if the session var
+	// is set?
+	asOfTs := hlc.Timestamp{WallTime: sb.evalCtx.StmtTimestamp.UnixNano()}
+	if asOf := sb.evalCtx.SessionData().StatsAsOf; !asOf.IsEmpty() {
+		asOfTs = asOf
+	}
 	// Find the most recent full statistic. (Stats are ordered with most recent first.)
 	var first int
-	for first < tab.StatisticCount() &&
-		(tab.Statistic(first).IsPartial() ||
-			tab.Statistic(first).IsMerged() && !sb.evalCtx.SessionData().OptimizerUseMergedPartialStatistics ||
-			tab.Statistic(first).IsForecast() && !sb.evalCtx.SessionData().OptimizerUseForecasts) {
-		first++
+	sd := sb.evalCtx.SessionData()
+	for first < tab.StatisticCount() {
+		first = getNextFullStats(first, tab, sd, asOfTs)
+		if first < 0 {
+			break
+		}
+		if statsCanaryWindow > 0 && !useCanary {
+			// We should use stable stats, which is defined as the second most recent full stats.
+			stat := tab.Statistic(first)
+			createdAtTS := hlc.Timestamp{WallTime: stat.CreatedAt().UnixNano()}
+			// Check if the current full stats is the last available one. If yes,
+			// use it as both canary and stable stats.
+			nextFullStatsIdx := getNextFullStats(first+1, tab, sd, asOfTs)
+			if nextFullStatsIdx != -1 {
+				// The following, we are getting full statistics for the stable stats,
+				// in contrast to the canary stats. The canary stats is skipped.
+				// If there remains only one full statistics, don't skip.
+				if stat.CreatedAt() == skippedStatsCreationTimestamp && !skippedStatsCreationTimestamp.IsZero() {
+					// We've already seen this canary stat, so skip it.
+					first++
+					continue
+				}
+				// Found a canary stats (defined as creation time within the canary window size). Register the
+				// creation timestamp and move on the next older one.
+				// If there is already a canary stats skipped, we don't skip again.
+				if createdAtTS.AddDuration(statsCanaryWindow).After(asOfTs) && skippedStatsCreationTimestamp.IsZero() {
+					skippedStatsCreationTimestamp = stat.CreatedAt()
+					first++
+					continue
+				}
+			}
+		}
+		break
 	}
 
-	if first >= tab.StatisticCount() {
+	if first >= tab.StatisticCount() || first < 0 {
 		// No statistics.
 		stats.Available = false
 		stats.RowCount = unknownRowCount
@@ -707,13 +776,7 @@ func (sb *statisticsBuilder) makeTableStatistics(tabID opt.TableID) *props.Stati
 	EachStat:
 		for i := first; i < tab.StatisticCount(); i++ {
 			stat := tab.Statistic(i)
-			if stat.IsPartial() {
-				continue
-			}
-			if stat.IsMerged() && !sb.evalCtx.SessionData().OptimizerUseMergedPartialStatistics {
-				continue
-			}
-			if stat.IsForecast() && !sb.evalCtx.SessionData().OptimizerUseForecasts {
+			if !isEligibleFullStats(stat, sd, asOfTs) {
 				continue
 			}
 			if stat.ColumnCount() > 1 && !sb.evalCtx.SessionData().OptimizerUseMultiColStats {