sql,roachpb: add plan hash correct value to persisted stats

Previously, we didn't have the plan hash/gist values, so
a dummy value was being used instead. Now that we have the value,
this commit uses those values to be corrected stored.
The Plan Hash is saved on its own column and is part of a
statement key. A plan gist is a string saved in the metadata
and can later on converted back into a logical plan.
The combined statements groups the value no using the
plan hash as a key, and creates a list of gist executed by
the same fingerprint id.

Partially addresses cockroachdb#72129

Release note (sql change): Saving plan hash/gist to the Statements
persisted stats inside Statistics column.

pkg/roachpb/app_stats.go

@@ -164,6 +164,7 @@ func (s *StatementStatistics) Add(other *StatementStatistics) {
 	s.RowsRead.Add(other.RowsRead, s.Count, other.Count)
 	s.RowsWritten.Add(other.RowsWritten, s.Count, other.Count)
 	s.Nodes = util.CombineUniqueInt64(s.Nodes, other.Nodes)
+	s.PlanGists = util.CombineUniqueString(s.PlanGists, other.PlanGists)
 
 	s.ExecStats.Add(other.ExecStats)
 

pkg/roachpb/app_stats.proto

@@ -104,6 +104,12 @@ message StatementStatistics {
   // Nodes is the ordered list of nodes ids on which the statement was executed.
   repeated int64 nodes = 24;
 
+  // plan_gists is list of a compressed version of plan that can be converted (lossily)
+  // back into a logical plan.
+  // Each statement contain only one plan gist, but the same statement fingerprint id
+  // can contain more than one value.
+  repeated string plan_gists = 26;
+
   // Note: be sure to update `sql/app_stats.go` when adding/removing fields here!
 
   reserved 13, 14, 17, 18, 19, 20;

pkg/server/combined_statement_stats.go

@@ -68,8 +68,8 @@ func getCombinedStatementStats(
 	startTime := getTimeFromSeconds(req.Start)
 	endTime := getTimeFromSeconds(req.End)
 	limit := SQLStatsResponseMax.Get(&settings.SV)
-	whereClause, args := getFilterAndParams(startTime, endTime, limit, testingKnobs)
-	statements, err := collectCombinedStatements(ctx, ie, whereClause, args)
+	whereClause, args := getFilterAndParams(startTime, endTime)
+	statements, err := collectCombinedStatements(ctx, ie, whereClause, args, limit, testingKnobs)
 	if err != nil {
 		return nil, err
 	}
@@ -89,12 +89,9 @@ func getCombinedStatementStats(
 	return response, nil
 }
 
-func getFilterAndParams(
-	start, end *time.Time, limit int64, testingKnobs *sqlstats.TestingKnobs,
-) (string, []interface{}) {
+func getFilterAndParams(start, end *time.Time) (string, []interface{}) {
 	var args []interface{}
 	var buffer strings.Builder
-	buffer.WriteString(testingKnobs.GetAOSTClause())
 
 	// Filter out internal statements by app name.
 	buffer.WriteString(" WHERE app_name NOT LIKE '$ internal%'")
@@ -109,17 +106,16 @@ func getFilterAndParams(
 		args = append(args, *end)
 	}
 
-	// Retrieve the top rows ordered by aggregation time and service latency.
-	buffer.WriteString(fmt.Sprintf(`
-ORDER BY aggregated_ts DESC,(statistics -> 'statistics' -> 'svcLat' ->> 'mean')::float DESC
-LIMIT $%d`, len(args)+1))
-	args = append(args, limit)
-
 	return buffer.String(), args
 }
 
 func collectCombinedStatements(
-	ctx context.Context, ie *sql.InternalExecutor, whereClause string, qargs []interface{},
+	ctx context.Context,
+	ie *sql.InternalExecutor,
+	whereClause string,
+	qargs []interface{},
+	limit int64,
+	testingKnobs *sqlstats.TestingKnobs,
 ) ([]serverpb.StatementsResponse_CollectedStatementStatistics, error) {
 
 	query := fmt.Sprintf(
@@ -132,8 +128,29 @@ func collectCombinedStatements(
 				statistics,
 				sampled_plan,
         aggregation_interval
-			FROM crdb_internal.statement_statistics
-			%s`, whereClause)
+				FROM (
+					SELECT
+					fingerprint_id,
+					transaction_fingerprint_id,
+					app_name,
+					aggregated_ts,
+					max(metadata) AS metadata,
+					crdb_internal.merge_statement_stats(array_agg(statistics)) AS statistics,
+					max(sampled_plan) AS sampled_plan,
+					aggregation_interval
+					FROM crdb_internal.statement_statistics
+					%s
+					GROUP BY
+					fingerprint_id,
+					transaction_fingerprint_id,
+					app_name,
+          aggregated_ts,
+					aggregation_interval)
+				%s
+				ORDER BY (statistics -> 'statistics' -> 'svcLat' ->> 'mean')::float DESC
+				LIMIT $%d`, whereClause, testingKnobs.GetAOSTClause(), len(qargs)+1)
+
+	qargs = append(qargs, limit)
 
 	const expectedNumDatums = 8
 

pkg/sql/crdb_internal.go

@@ -5118,10 +5118,8 @@ CREATE TABLE crdb_internal.cluster_statement_statistics (
 				transactionFingerprintID := tree.NewDBytes(
 					tree.DBytes(sqlstatsutil.EncodeUint64ToBytes(uint64(statistics.Key.TransactionFingerprintID))))
 
-				// TODO(azhng): properly update plan_hash value once we can expose it
-				//  from the optimizer.
 				planHash := tree.NewDBytes(
-					tree.DBytes(sqlstatsutil.EncodeUint64ToBytes(0)))
+					tree.DBytes(sqlstatsutil.EncodeUint64ToBytes(statistics.Key.PlanHash)))
 
 				metadataJSON, err := sqlstatsutil.BuildStmtMetadataJSON(statistics)
 				if err != nil {

pkg/sql/executor_statement_metrics.go

@@ -163,6 +163,7 @@ func (ex *connExecutor) recordStatementSummary(
 		FullScan:     flags.IsSet(planFlagContainsFullIndexScan) || flags.IsSet(planFlagContainsFullTableScan),
 		Failed:       stmtErr != nil,
 		Database:     planner.SessionData().Database,
+		PlanHash:     planner.instrumentation.planGist.Hash(),
 	}
 
 	// We only populate the transaction fingerprint ID field if we are in an
@@ -204,6 +205,7 @@ func (ex *connExecutor) recordStatementSummary(
 		Nodes:           getNodesFromPlanner(planner),
 		StatementType:   stmt.AST.StatementType(),
 		Plan:            planner.instrumentation.PlanForStats(ctx),
+		PlanGist:        planner.instrumentation.planGist.String(),
 		StatementError:  stmtErr,
 	}
 

pkg/sql/instrumentation.go

@@ -296,6 +296,7 @@ func (ih *instrumentationHelper) Finish(
 			ImplicitTxn: ih.implicitTxn,
 			Database:    p.SessionData().Database,
 			Failed:      retErr != nil,
+			PlanHash:    ih.planGist.Hash(),
 		}
 		// We populate transaction fingerprint ID if this is an implicit transaction.
 		// See executor_statement_metrics.go:recordStatementSummary() for further

pkg/sql/instrumentation_test.go

@@ -73,6 +73,7 @@ func TestSampledStatsCollection(t *testing.T) {
 					},
 				))
 			require.NotNil(t, stats)
+			require.NotZero(t, stats.Key.PlanHash)
 			return stats
 		}
 

pkg/sql/sqlstats/persistedsqlstats/flush.go

@@ -147,7 +147,7 @@ func (s *PersistedSQLStats) doFlushSingleStmtStats(
 
 		serializedFingerprintID := sqlstatsutil.EncodeUint64ToBytes(uint64(scopedStats.ID))
 		serializedTransactionFingerprintID := sqlstatsutil.EncodeUint64ToBytes(uint64(scopedStats.Key.TransactionFingerprintID))
-		serializedPlanHash := sqlstatsutil.EncodeUint64ToBytes(uint64(dummyPlanHash))
+		serializedPlanHash := sqlstatsutil.EncodeUint64ToBytes(scopedStats.Key.PlanHash)
 
 		insertFn := func(ctx context.Context, txn *kv.Txn) (alreadyExists bool, err error) {
 			rowsAffected, err := s.insertStatementStats(
@@ -417,7 +417,7 @@ WHERE fingerprint_id = $2
 			"plan_hash: %d, "+
 			"node_id: %d",
 			serializedFingerprintID, serializedTransactionFingerprintID, stats.Key.App,
-			aggregatedTs, dummyPlanHash, s.cfg.SQLIDContainer.SQLInstanceID())
+			aggregatedTs, serializedPlanHash, s.cfg.SQLIDContainer.SQLInstanceID())
 	}
 
 	return nil
@@ -581,12 +581,12 @@ FOR UPDATE
 	if row == nil {
 		return errors.AssertionFailedf(
 			"statement statistics not found fingerprint_id: %s, app: %s, aggregated_ts: %s, plan_hash: %d, node_id: %d",
-			serializedFingerprintID, key.App, aggregatedTs, dummyPlanHash, s.cfg.SQLIDContainer.SQLInstanceID())
+			serializedFingerprintID, key.App, aggregatedTs, serializedPlanHash, s.cfg.SQLIDContainer.SQLInstanceID())
 	}
 
 	if len(row) != 1 {
 		return errors.AssertionFailedf("unexpectedly found %d returning columns for fingerprint_id: %s, app: %s, aggregated_ts: %s, plan_hash %d, node_id: %d",
-			len(row), serializedFingerprintID, key.App, aggregatedTs, dummyPlanHash,
+			len(row), serializedFingerprintID, key.App, aggregatedTs, serializedPlanHash,
 			s.cfg.SQLIDContainer.SQLInstanceID())
 	}
 

pkg/sql/sqlstats/persistedsqlstats/provider.go

@@ -34,14 +34,6 @@ import (
 	"github.com/cockroachdb/errors"
 )
 
-const (
-	// TODO(azhng): currently we do not have the ability to compute a hash for
-	//  query plan. This is currently being worked on by the SQL Queries team.
-	//  Once we are able get consistent hash value from a query plan, we should
-	//  update this.
-	dummyPlanHash = int64(0)
-)
-
 // ErrConcurrentSQLStatsCompaction is reported when two sql stats compaction
 // jobs are issued concurrently. This is a sentinel error.
 var ErrConcurrentSQLStatsCompaction = errors.New("another sql stats compaction job is already running")

pkg/sql/sqlstats/persistedsqlstats/sqlstatsutil/json_encoding_test.go

@@ -96,7 +96,8 @@ func TestSQLStatsJsonEncoding(t *testing.T) {
            "mean": {{.Float}},
            "sqDiff": {{.Float}}
          },
-         "nodes": [{{joinInts .IntArray}}]
+         "nodes": [{{joinInts .IntArray}}],
+         "planGists": [{{joinStrings .StringArray}}]
        },
        "execution_statistics": {
          "cnt": {{.Int64}},
@@ -211,6 +212,7 @@ func TestSQLStatsJsonEncoding(t *testing.T) {
            "sqDiff": {{.Float}}
          },
          "nodes": [{{joinInts .IntArray}}]
+         "planGists": [{{joinStrings .StringArray}}]
        },
        "execution_statistics": {
          "cnt": {{.Int64}},

pkg/sql/sqlstats/persistedsqlstats/sqlstatsutil/json_impl.go

@@ -135,6 +135,39 @@ func (a *int64Array) encodeJSON() (json.JSON, error) {
 	return builder.Build(), nil
 }
 
+type stringArray []string
+
+func (a *stringArray) decodeJSON(js json.JSON) error {
+	arrLen := js.Len()
+	for i := 0; i < arrLen; i++ {
+		var value jsonString
+		valJSON, err := js.FetchValIdx(i)
+		if err != nil {
+			return err
+		}
+		if err := value.decodeJSON(valJSON); err != nil {
+			return err
+		}
+		*a = append(*a, string(value))
+	}
+
+	return nil
+}
+
+func (a *stringArray) encodeJSON() (json.JSON, error) {
+	builder := json.NewArrayBuilder(len(*a))
+
+	for _, value := range *a {
+		jsVal, err := (*jsonString)(&value).encodeJSON()
+		if err != nil {
+			return nil, err
+		}
+		builder.Add(jsVal)
+	}
+
+	return builder.Build(), nil
+}
+
 type stmtFingerprintIDArray []roachpb.StmtFingerprintID
 
 func (s *stmtFingerprintIDArray) decodeJSON(js json.JSON) error {
@@ -237,6 +270,7 @@ func (s *innerStmtStats) jsonFields() jsonFields {
 		{"rowsRead", (*numericStats)(&s.RowsRead)},
 		{"rowsWritten", (*numericStats)(&s.RowsWritten)},
 		{"nodes", (*int64Array)(&s.Nodes)},
+		{"planGists", (*stringArray)(&s.PlanGists)},
 	}
 }
 

pkg/sql/sqlstats/persistedsqlstats/sqlstatsutil/testutils.go

@@ -11,12 +11,13 @@
 package sqlstatsutil
 
 import (
-	"html/template"
+	"fmt"
 	"math/rand"
 	"reflect"
 	"strconv"
 	"strings"
 	"testing"
+	"text/template"
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
@@ -36,12 +37,13 @@ func GetRandomizedCollectedStatementStatisticsForTest(
 }
 
 type randomData struct {
-	Bool     bool
-	String   string
-	Int64    int64
-	Float    float64
-	IntArray []int64
-	Time     time.Time
+	Bool        bool
+	String      string
+	Int64       int64
+	Float       float64
+	IntArray    []int64
+	StringArray []string
+	Time        time.Time
 }
 
 var alphabet = []rune("abcdefghijklmkopqrstuvwxyz")
@@ -57,11 +59,22 @@ func genRandomData() randomData {
 	}
 	r.String = b.String()
 
+	// Generate a randomized array of length 5.
+	arrLen := 5
+	r.StringArray = make([]string, arrLen)
+	for i := 0; i < arrLen; i++ {
+		// Randomly generating 10-character string.
+		b := strings.Builder{}
+		for j := 0; j < 10; j++ {
+			b.WriteRune(alphabet[rand.Intn(26)])
+		}
+		r.StringArray[i] = b.String()
+	}
+
 	r.Int64 = rand.Int63()
 	r.Float = rand.Float64()
 
 	// Generate a randomized array of length 5.
-	arrLen := 5
 	r.IntArray = make([]int64, arrLen)
 	for i := 0; i < arrLen; i++ {
 		r.IntArray[i] = rand.Int63()
@@ -79,6 +92,13 @@ func fillTemplate(t *testing.T, tmplStr string, data randomData) string {
 		}
 		return strings.Join(strArr, ",")
 	}
+	joinStrings := func(arr []string) string {
+		strArr := make([]string, len(arr))
+		for i, val := range arr {
+			strArr[i] = fmt.Sprintf("%q", val)
+		}
+		return strings.Join(strArr, ",")
+	}
 	stringifyTime := func(tm time.Time) string {
 		s, err := tm.MarshalText()
 		require.NoError(t, err)
@@ -88,6 +108,7 @@ func fillTemplate(t *testing.T, tmplStr string, data randomData) string {
 		New("").
 		Funcs(template.FuncMap{
 			"joinInts":      joinInts,
+			"joinStrings":   joinStrings,
 			"stringifyTime": stringifyTime,
 		}).
 		Parse(tmplStr)
@@ -133,8 +154,15 @@ func fillObject(t *testing.T, val reflect.Value, data *randomData) {
 	case reflect.Bool:
 		val.SetBool(data.Bool)
 	case reflect.Slice:
-		for _, randInt := range data.IntArray {
-			val.Set(reflect.Append(val, reflect.ValueOf(randInt)))
+		switch val.Type().String() {
+		case "[]string":
+			for _, randString := range data.StringArray {
+				val.Set(reflect.Append(val, reflect.ValueOf(randString)))
+			}
+		case "[]int64":
+			for _, randInt := range data.IntArray {
+				val.Set(reflect.Append(val, reflect.ValueOf(randInt)))
+			}
 		}
 	case reflect.Struct:
 		switch val.Type().Name() {

pkg/sql/sqlstats/persistedsqlstats/testdata/logical_plan_sampling_for_explicit_txn

@@ -28,7 +28,7 @@ SELECT 1
 # logical plan sampling.
 should-sample-logical-plan db=defaultdb implicitTxn=true fingerprint=SELECT%_
 ----
-false
+true
 
 # However, if we are to execute the same statement but under explicit
 # transaction, the plan will still need to be sampled.
@@ -48,7 +48,7 @@ COMMIT
 # collection.
 should-sample-logical-plan db=defaultdb implicitTxn=false fingerprint=SELECT%_
 ----
-false
+true
 
 # Set the time to the future and ensure we will resample the logical plan.
 set-time time=2021-09-20T15:05:01Z
@@ -67,7 +67,7 @@ SELECT 1
 
 should-sample-logical-plan db=defaultdb implicitTxn=true fingerprint=SELECT%_
 ----
-false
+true
 
 should-sample-logical-plan db=defaultdb implicitTxn=false fingerprint=SELECT%_
 ----
@@ -82,4 +82,4 @@ COMMIT
 
 should-sample-logical-plan db=defaultdb implicitTxn=false fingerprint=SELECT%_
 ----
-false
+true

pkg/sql/sqlstats/ssmemstorage/ss_mem_iterator.go

@@ -94,6 +94,7 @@ func (s *StmtStatsIterator) Next() bool {
 			Failed:                   stmtKey.failed,
 			App:                      s.container.appName,
 			Database:                 database,
+			PlanHash:                 stmtKey.planHash,
 			TransactionFingerprintID: stmtKey.transactionFingerprintID,
 		},
 		ID:    stmtFingerprintID,