Merge #143947 #143951 #143955

143947: explain: fix rare flake in recently added test r=yuzefovich a=yuzefovich

Recently added `TestMaximumMemoryUsage` can encounter a rare flake when some metamorphic variables are set - the ComponentStats proto might be dropped from the trace, so we won't have "maximum memory usage" stat that the test expects. Fix this by adding a retry loop.

Fixes: #143833.

Release note: None

143951: kvcoord: fix CPut handling for intra-batch conflicts r=yuzefovich a=yuzefovich

Previously, for condition evaluation part of CPuts, we would ignore writes that were performed within the same KV batch. That was the case since we performed the read from the buffer when applying the transformation but would actually buffer writes after the KV server response comes back, so previous writes within the same KV batch would be invisible. This is now fixed by buffering the write when applying the transformation.

Also avoid flushing any buffered writes when rolling back the txn.

Epic: None
Release note: None

143955: sql/stats: remove extended sentry reporting for an error r=yuzefovich a=yuzefovich

This commit reverts 85500f0. I think we fixed the root cause of the "first bucket should have NumRange=0" in 10a2ffe, so we no longer need the extended sentry reporting (which also didn't quite work in all cases).

Informs: #93892.
Epic: None
Release note: None

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

Author: craig[bot]

pkg/kv/kvclient/kvcoord/txn_interceptor_write_buffer.go

@@ -131,10 +131,12 @@ func (twb *txnWriteBuffer) SendLocked(
 		return twb.wrapped.SendLocked(ctx, ba)
 	}
 
-	if _, ok := ba.GetArg(kvpb.EndTxn); ok {
-		// TODO(arul): should we only flush if the transaction is being committed?
-		// If the transaction is being rolled back, we shouldn't needlessly flush
-		// writes.
+	if etArg, ok := ba.GetArg(kvpb.EndTxn); ok {
+		if !etArg.(*kvpb.EndTxnRequest).Commit {
+			// We're performing a rollback, so there is no point in flushing
+			// anything.
+			return twb.wrapped.SendLocked(ctx, ba)
+		}
 		return twb.flushWithEndTxn(ctx, ba)
 	}
 
@@ -375,6 +377,9 @@ func (twb *txnWriteBuffer) applyTransformations(
 			// Send a locking Get request to the KV layer; we'll evaluate the
 			// condition locally based on the response.
 			baRemote.Requests = append(baRemote.Requests, getReqU)
+			// Buffer a Put under the optimistic assumption that the condition
+			// will be satisfied.
+			twb.addToBuffer(t.Key, t.Value, t.Sequence)
 
 		case *kvpb.PutRequest:
 			// If the MustAcquireExclusiveLock flag is set on the Put, then we need to
@@ -838,13 +843,14 @@ func (t transformation) toResp(
 			req.AllowIfDoesNotExist,
 		)
 		if condFailedErr != nil {
+			// TODO(yuzefovich): consider "poisoning" the txnWriteBuffer when we
+			// hit a condition failed error to avoid mistaken usages (e.g. an
+			// attempt to flush with the EndTxn request with Commit=true).
 			pErr := kvpb.NewErrorWithTxn(condFailedErr, txn)
 			pErr.SetErrorIndex(int32(t.index))
 			return kvpb.ResponseUnion{}, pErr
 		}
-		// The condition was satisfied; buffer a Put, and return a synthesized
-		// response.
-		twb.addToBuffer(req.Key, req.Value, req.Sequence)
+		// The condition was satisfied - return a synthesized response.
 		ru.MustSetInner(&kvpb.ConditionalPutResponse{})
 
 	case *kvpb.PutRequest:
@@ -897,12 +903,6 @@ func (t transformation) toResp(
 		panic("unimplemented")
 	}
 
-	// TODO(arul): in the future, when we'll evaluate CPuts locally, we'll have
-	// this function take in the result of the KVGet, save the CPut function
-	// locally on the transformation, and use these two things to evaluate the
-	// condition here, on the client. We'll then construct and return the
-	// appropriate response.
-
 	return ru, nil
 }
 

pkg/kv/kvclient/kvcoord/txn_interceptor_write_buffer_test.go

@@ -1109,11 +1109,11 @@ func TestTxnWriteBufferDecomposesConditionalPuts(t *testing.T) {
 			require.IsType(t, &kvpb.ConditionFailedError{}, pErr.GoError())
 		}
 
-		// Lastly, commit the transaction. A put should only be flushed if the condition
-		// evaluated successfully.
+		// Lastly, commit or rollback the transaction. A put should only be
+		// flushed if the condition evaluated successfully.
 		ba = &kvpb.BatchRequest{}
 		ba.Header = kvpb.Header{Txn: &txn}
-		ba.Add(&kvpb.EndTxnRequest{Commit: true})
+		ba.Add(&kvpb.EndTxnRequest{Commit: condEvalSuccessful})
 
 		mockSender.MockSend(func(ba *kvpb.BatchRequest) (*kvpb.BatchResponse, *kvpb.Error) {
 			if condEvalSuccessful {

pkg/kv/kvclient/kvcoord/txn_test.go

@@ -2011,6 +2011,7 @@ func TestTxnBufferedWritesConditionalPuts(t *testing.T) {
 				if expErr {
 					require.Error(t, err)
 					require.IsType(t, &kvpb.ConditionFailedError{}, err)
+					return err
 				} else {
 					require.NoError(t, err)
 				}
@@ -2022,11 +2023,10 @@ func TestTxnBufferedWritesConditionalPuts(t *testing.T) {
 				}
 			})
 
-			if commit {
+			if commit && !expErr {
 				require.NoError(t, err)
 			} else {
 				require.Error(t, err)
-				testutils.IsError(err, "abort")
 			}
 
 			// Verify the values are visible only if the transaction commited

pkg/sql/logictest/testdata/logic_test/buffered_writes

@@ -82,3 +82,47 @@ query II
 SELECT * FROM t1
 ----
 1 1
+
+statement ok
+CREATE TABLE t2 (k INT PRIMARY KEY);
+
+statement ok
+BEGIN;
+
+statement error pgcode 23505 duplicate key value violates unique constraint "t2_pkey"
+INSERT INTO t2 VALUES (1), (1);
+
+statement ok
+ROLLBACK;
+
+statement ok
+BEGIN;
+
+statement ok
+INSERT INTO t2 VALUES (1);
+
+statement error pgcode 23505 duplicate key value violates unique constraint "t2_pkey"
+INSERT INTO t2 VALUES (1);
+
+statement ok
+ROLLBACK;
+
+statement ok
+BEGIN;
+
+statement ok
+INSERT INTO t2 VALUES (1);
+
+statement ok
+DELETE FROM t2 WHERE k = 1;
+
+statement ok
+INSERT INTO t2 VALUES (1);
+
+statement ok
+COMMIT;
+
+query I
+SELECT * FROM t2
+----
+1

pkg/sql/opt/exec/explain/BUILD.bazel

@@ -89,6 +89,7 @@ go_test(
         "//pkg/util/log",
         "//pkg/util/treeprinter",
         "@com_github_cockroachdb_datadriven//:datadriven",
+        "@com_github_cockroachdb_errors//:errors",
         "@com_github_stretchr_testify//assert",
         "@com_github_stretchr_testify//require",
         "@in_gopkg_yaml_v2//:yaml_v2",

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

@@ -30,6 +30,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/datadriven"
+	"github.com/cockroachdb/errors"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	yaml "gopkg.in/yaml.v2"
@@ -417,19 +418,29 @@ func TestMaximumMemoryUsage(t *testing.T) {
 		return err
 	})
 
-	rows := db.QueryStr(t, "EXPLAIN ANALYZE SELECT max(v) FROM t GROUP BY bucket;")
-	var output strings.Builder
-	maxMemoryRE := regexp.MustCompile(`maximum memory usage: ([\d\.]+) MiB`)
-	var maxMemoryUsage float64
-	for _, row := range rows {
-		output.WriteString(row[0])
-		output.WriteString("\n")
-		s := strings.TrimSpace(row[0])
-		if matches := maxMemoryRE.FindStringSubmatch(s); len(matches) > 0 {
-			var err error
-			maxMemoryUsage, err = strconv.ParseFloat(matches[1], 64)
-			require.NoError(t, err)
+	// In rare cases (due to metamorphic randomization) we might drop the
+	// ComponentStats proto that powers "maximum memory usage" stat from the
+	// trace, so we add a retry loop around this.
+	testutils.SucceedsSoon(t, func() error {
+		rows := db.QueryStr(t, "EXPLAIN ANALYZE SELECT max(v) FROM t GROUP BY bucket;")
+		var output strings.Builder
+		maxMemoryRE := regexp.MustCompile(`maximum memory usage: ([\d\.]+) MiB`)
+		var maxMemoryUsage float64
+		for _, row := range rows {
+			output.WriteString(row[0])
+			output.WriteString("\n")
+			s := strings.TrimSpace(row[0])
+			if matches := maxMemoryRE.FindStringSubmatch(s); len(matches) > 0 {
+				var err error
+				maxMemoryUsage, err = strconv.ParseFloat(matches[1], 64)
+				if err != nil {
+					return err
+				}
+			}
 		}
-	}
-	require.Greaterf(t, maxMemoryUsage, 5.0, "expected maximum memory usage to be at least 5 MiB, full output:\n\n%s", output.String())
+		if maxMemoryUsage < 5.0 {
+			return errors.Newf("expected maximum memory usage to be at least 5 MiB, full output:\n\n%s", output.String())
+		}
+		return nil
+	})
 }

pkg/sql/stats/BUILD.bazel

@@ -57,7 +57,6 @@ go_library(
         "//pkg/util/log",
         "//pkg/util/mon",
         "//pkg/util/protoutil",
-        "//pkg/util/sentryutil",
         "//pkg/util/stop",
         "//pkg/util/syncutil",
         "//pkg/util/timeutil",

pkg/sql/stats/forecast.go

@@ -7,24 +7,17 @@ package stats
 
 import (
 	"context"
-	"crypto/md5"
-	"encoding/json"
-	"fmt"
 	"math"
 	"slices"
-	"strconv"
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/settings"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql/opt/cat"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/eval"
-	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
-	"github.com/cockroachdb/cockroach/pkg/util/errorutil"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
-	"github.com/cockroachdb/cockroach/pkg/util/sentryutil"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/redact"
 )
@@ -336,61 +329,6 @@ func forecastColumnStatistics(
 		}
 		forecast.HistogramData = &histData
 		forecast.setHistogramBuckets(hist)
-
-		// Verify that the first two buckets (the initial NULL bucket and the first
-		// non-NULL bucket) both have NumRange=0 and DistinctRange=0. (We must check
-		// this after calling setHistogramBuckets to account for rounding.) See
-		// #93892.
-		for _, bucket := range forecast.Histogram {
-			if bucket.NumRange != 0 || bucket.DistinctRange != 0 {
-				// Build a JSON representation of the first several buckets in each
-				// observed histogram so that we can figure out what happened.
-				const debugBucketCount = 5
-				jsonStats := make([]*JSONStatistic, 0, len(observed))
-
-				addStat := func(stat *TableStatistic) {
-					jsonStat := &JSONStatistic{
-						Name:          stat.Name,
-						CreatedAt:     stat.CreatedAt.String(),
-						Columns:       []string{strconv.FormatInt(int64(stat.ColumnIDs[0]), 10)},
-						RowCount:      stat.RowCount,
-						DistinctCount: stat.DistinctCount,
-						NullCount:     stat.NullCount,
-						AvgSize:       stat.AvgSize,
-					}
-					if err := jsonStat.SetHistogram(stat.HistogramData); err == nil &&
-						len(jsonStat.HistogramBuckets) > debugBucketCount {
-						// Limit the histogram to the first several buckets.
-						jsonStat.HistogramBuckets = jsonStat.HistogramBuckets[0:debugBucketCount]
-					}
-					// Replace UpperBounds with a hash.
-					for i := range jsonStat.HistogramBuckets {
-						hash := md5.Sum([]byte(jsonStat.HistogramBuckets[i].UpperBound))
-						jsonStat.HistogramBuckets[i].UpperBound = fmt.Sprintf("_%x", hash)
-					}
-					jsonStats = append(jsonStats, jsonStat)
-				}
-				addStat(forecast)
-				for i := range observed {
-					addStat(observed[i])
-				}
-				var debugging redact.SafeString
-				if j, err := json.Marshal(jsonStats); err == nil {
-					debugging = redact.SafeString(j)
-				}
-				err := errorutil.UnexpectedWithIssueErrorf(
-					93892,
-					"forecasted histogram had first bucket with non-zero NumRange or DistinctRange: %s",
-					debugging,
-				)
-				sentryutil.SendReport(ctx, &st.SV, err)
-				return nil, err
-			}
-			if bucket.UpperBound != tree.DNull {
-				// Stop checking after the first non-NULL bucket.
-				break
-			}
-		}
 	}
 
 	return forecast, nil