Merge #150178 #150643 #150876

150178: kvpb: avoid over redaction of prev_err field r=rafiss a=rafiss

When logging a kvpb.TxnAlreadyEncounteredErrorError, the previous error was getting redacted too aggressively. This patch makes the error use a redactable string instead.

fixes #146588
Release note: None

150643: crosscluster/physical: send recompute stats requests on range key flush r=jeffswenson a=msbutler

This patch ensures that destination cluster mvcc stats are up to date after a rangekey flush. Since range key flushes are assumed to be rare, we recomuptes stats over each affected range.

Fixes [#142481](#142481)

Release note: none

150876: sqlccl: bump engflow worker size r=yuzefovich a=yuzefovich

We just saw a failure that looks like an OOM, so let's bump the worker size.

Fixes: #150648.

Release note: None

Co-authored-by: Rafi Shamim <[email protected]>
Co-authored-by: Michael Butler <[email protected]>
Co-authored-by: Yahor Yuzefovich <[email protected]>

Author: 4 people

pkg/ccl/testccl/sqlccl/BUILD.bazel

@@ -19,6 +19,9 @@ go_test(
     data = [
         "//c-deps:libgeos",  # keep
     ],
+    exec_properties = select({
+        "//conditions:default": {"test.Pool": "large"},
+    }),
     shard_count = 16,
     deps = [
         "//pkg/base",

pkg/crosscluster/physical/BUILD.bazel

@@ -84,6 +84,7 @@ go_library(
         "//pkg/util/log",
         "//pkg/util/metric",
         "//pkg/util/protoutil",
+        "//pkg/util/rangedesc",
         "//pkg/util/retry",
         "//pkg/util/span",
         "//pkg/util/syncutil",
@@ -144,6 +145,7 @@ go_test(
         "//pkg/jobs",
         "//pkg/jobs/jobspb",
         "//pkg/keys",
+        "//pkg/kv",
         "//pkg/kv/kvclient/rangefeed/rangefeedcache",
         "//pkg/kv/kvpb",
         "//pkg/kv/kvserver",
@@ -185,6 +187,7 @@ go_test(
         "//pkg/util",
         "//pkg/util/ctxgroup",
         "//pkg/util/duration",
+        "//pkg/util/encoding",
         "//pkg/util/hlc",
         "//pkg/util/httputil",
         "//pkg/util/leaktest",

pkg/crosscluster/physical/replication_stream_e2e_test.go

@@ -24,6 +24,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/jobs"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/keys"
+	"github.com/cockroachdb/cockroach/pkg/kv"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/protectedts"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/protectedts/ptpb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
@@ -44,6 +45,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/storageutils"
 	"github.com/cockroachdb/cockroach/pkg/util/duration"
+	"github.com/cockroachdb/cockroach/pkg/util/encoding"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
@@ -1674,7 +1676,7 @@ func TestComputeStatsDiff(t *testing.T) {
 	var tableID int
 	c.SrcTenantSQL.QueryRow(t, "SELECT id FROM system.namespace WHERE name = 'x'").Scan(&tableID)
 	tenantID := c.Args.DestTenantID
-	liveCountOverPKQuery := fmt.Sprintf(`SELECT stats->'approximate_total_stats'->'live_count' FROM crdb_internal.tenant_span_stats(ARRAY(SELECT(crdb_internal.index_span(%d,%d,1)[1],crdb_internal.index_span(%d,%d,1)[2])))`, tenantID.ToUint64(), tableID, tenantID.ToUint64(), tableID)
+	liveCountOverPKQuery := makeLiveCountOverPKQuery(tenantID, tableID)
 
 	var liveCount int64
 	c.DestSysSQL.QueryRow(t, liveCountOverPKQuery).Scan(&liveCount)
@@ -1690,11 +1692,7 @@ func TestComputeStatsDiff(t *testing.T) {
 	// Split out the index span we will gather stats so that
 	// crdb_internal.tenant_span_stats() uses the stats that hang off of the
 	// descriptor, instead of computing the stats manually.
-	codec := keys.MakeSQLCodec(tenantID)
-	pkStartKey := codec.IndexPrefix(uint32(tableID), 1)
-	pkEndKey := pkStartKey.PrefixEnd()
-	require.NoError(t, c.DestSysServer.DB().AdminSplit(ctx, pkStartKey, hlc.MaxTimestamp))
-	require.NoError(t, c.DestSysServer.DB().AdminSplit(ctx, pkEndKey, hlc.MaxTimestamp))
+	splitPrimaryKeyIndexSpan(ctx, t, c.DestSysServer.DB(), tenantID, tableID)
 
 	c.DestSysSQL.QueryRow(t, liveCountOverPKQuery).Scan(&liveCount)
 	require.Equal(t, int64(2), liveCount)
@@ -1710,3 +1708,95 @@ func TestComputeStatsDiff(t *testing.T) {
 	c.DestSysSQL.QueryRow(t, liveCountOverPKQuery).Scan(&liveCount)
 	require.Equal(t, int64(2), liveCount)
 }
+
+// TestDelRangeStatsUpdates is an end to end test that ensures that mvcc stats are
+// computed correctly after replicating range key deletes.
+func TestDelRangeStatsUpdates(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	ctx := context.Background()
+	args := replicationtestutils.DefaultTenantStreamingClustersArgs
+	c, cleanup := replicationtestutils.CreateTenantStreamingClusters(ctx, t, args)
+	defer cleanup()
+
+	c.DestSysSQL.Exec(t, "SET CLUSTER SETTING server.debug.default_vmodule = 'stream_ingestion_processor=2'; ")
+
+	c.SrcTenantSQL.Exec(t, "CREATE DATABASE test")
+	c.SrcTenantSQL.Exec(t, "CREATE TABLE test.x (id INT PRIMARY KEY, n INT)")
+
+	var tableID int
+	c.SrcTenantSQL.QueryRow(t, "SELECT id FROM system.namespace WHERE name = 'x'").Scan(&tableID)
+	tenantID := c.Args.DestTenantID
+	liveCountOverPKQuery := makeLiveCountOverPKQuery(tenantID, tableID)
+
+	var liveCount int64
+	c.DestSysSQL.QueryRow(t, liveCountOverPKQuery).Scan(&liveCount)
+	require.Equal(t, int64(0), liveCount)
+
+	producerJobID, ingestionJobID := c.StartStreamReplication(ctx)
+
+	jobutils.WaitForJobToRun(c.T, c.SrcSysSQL, jobspb.JobID(producerJobID))
+	jobutils.WaitForJobToRun(c.T, c.DestSysSQL, jobspb.JobID(ingestionJobID))
+
+	c.WaitUntilStartTimeReached(jobspb.JobID(ingestionJobID))
+
+	// Split out the index span we will gather stats so that
+	// crdb_internal.tenant_span_stats() uses the stats that hang off of the
+	// descriptor, instead of computing the stats manually.
+	splitPrimaryKeyIndexSpan(ctx, t, c.DestSysServer.DB(), tenantID, tableID)
+
+	for i := 1; i <= 50; i++ {
+		c.SrcTenantSQL.Exec(t, fmt.Sprintf("INSERT INTO test.x VALUES (%d, %d)", i, i))
+	}
+
+	srcTime := c.SrcCluster.Server(0).Clock().Now()
+	c.WaitUntilReplicatedTime(srcTime, jobspb.JobID(ingestionJobID))
+
+	c.DestSysSQL.QueryRow(t, liveCountOverPKQuery).Scan(&liveCount)
+	require.Equal(t, int64(50), liveCount)
+
+	// Add 5 split points to the replicated table on destination cluster, to test
+	// that the recompute stats logic works on multiple ranges.
+	destCodec := keys.MakeSQLCodec(c.Args.DestTenantID)
+	for i := 1; i <= 5; i++ {
+		splitValue := i * 5
+		splitKey := destCodec.IndexPrefix(uint32(tableID), 1)
+		splitKey = encoding.EncodeVarintAscending(splitKey, int64(splitValue))
+		require.NoError(t, c.DestSysServer.DB().AdminSplit(ctx, splitKey, hlc.MaxTimestamp))
+	}
+
+	// Write a range key over table test.x's key span instead of dropping the
+	// table to guarantee that the srcTime below is after the range key write.
+	srcCodec := keys.MakeSQLCodec(c.Args.SrcTenantID)
+	tableSpan := roachpb.Span{
+		Key:    srcCodec.TablePrefix(uint32(tableID)),
+		EndKey: srcCodec.TablePrefix(uint32(tableID)).PrefixEnd(),
+	}
+	require.NoError(t, c.SrcSysServer.DB().DelRangeUsingTombstone(ctx, tableSpan.Key, tableSpan.EndKey))
+
+	srcTime = c.SrcCluster.Server(0).Clock().Now()
+	c.WaitUntilReplicatedTime(srcTime, jobspb.JobID(ingestionJobID))
+	c.RequireFingerprintMatchAtTimestamp(srcTime.AsOfSystemTime())
+
+	c.DestSysSQL.QueryRow(t, liveCountOverPKQuery).Scan(&liveCount)
+	require.Equal(t, int64(0), liveCount)
+}
+
+// makeLiveCountOverPKQuery constructs a query to get live count stats for a table's primary key.
+func makeLiveCountOverPKQuery(tenantID roachpb.TenantID, tableID int) string {
+	return fmt.Sprintf(`SELECT stats->'approximate_total_stats'->'live_count' FROM crdb_internal.tenant_span_stats(ARRAY(SELECT(crdb_internal.index_span(%d,%d,1)[1],crdb_internal.index_span(%d,%d,1)[2])))`,
+		tenantID.ToUint64(), tableID, tenantID.ToUint64(), tableID)
+}
+
+// splitPrimaryKeyIndexSpan splits the primary key index span for a table to ensure
+// crdb_internal.tenant_span_stats() uses the stats that hang off of the descriptor.
+func splitPrimaryKeyIndexSpan(
+	ctx context.Context, t *testing.T, db *kv.DB, tenantID roachpb.TenantID, tableID int,
+) {
+	codec := keys.MakeSQLCodec(tenantID)
+	pkStartKey := codec.IndexPrefix(uint32(tableID), 1)
+	pkEndKey := pkStartKey.PrefixEnd()
+	require.NoError(t, db.AdminSplit(ctx, pkStartKey, hlc.MaxTimestamp))
+	require.NoError(t, db.AdminSplit(ctx, pkEndKey, hlc.MaxTimestamp))
+}

pkg/crosscluster/physical/stream_ingestion_processor.go

@@ -40,6 +40,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/protoutil"
+	"github.com/cockroachdb/cockroach/pkg/util/rangedesc"
 	"github.com/cockroachdb/cockroach/pkg/util/span"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
@@ -199,21 +200,27 @@ func releaseBuffer(b *streamIngestionBuffer) {
 
 // Specialized SST batcher that is responsible for ingesting range tombstones.
 type rangeKeyBatcher struct {
-	db       *kv.DB
-	settings *cluster.Settings
+	db                   *kv.DB
+	settings             *cluster.Settings
+	rangeDescIterFactory rangedesc.IteratorFactory
 
 	// onFlush is the callback called after the current batch has been
 	// successfully ingested.
 	onFlush func(kvpb.BulkOpSummary)
 }
 
 func newRangeKeyBatcher(
-	ctx context.Context, cs *cluster.Settings, db *kv.DB, onFlush func(summary kvpb.BulkOpSummary),
+	ctx context.Context,
+	cs *cluster.Settings,
+	db *kv.DB,
+	ranges rangedesc.IteratorFactory,
+	onFlush func(summary kvpb.BulkOpSummary),
 ) *rangeKeyBatcher {
 	batcher := &rangeKeyBatcher{
-		db:       db,
-		settings: cs,
-		onFlush:  onFlush,
+		db:                   db,
+		rangeDescIterFactory: ranges,
+		settings:             cs,
+		onFlush:              onFlush,
 	}
 	return batcher
 }
@@ -414,7 +421,8 @@ func (sip *streamIngestionProcessor) Start(ctx context.Context) {
 		return
 	}
 
-	sip.rangeBatcher = newRangeKeyBatcher(ctx, st, db.KV(), sip.onFlushUpdateMetricUpdate)
+	execCfg := sip.FlowCtx.Cfg.ExecutorConfig.(*sql.ExecutorConfig)
+	sip.rangeBatcher = newRangeKeyBatcher(ctx, st, db.KV(), execCfg.RangeDescIteratorFactory, sip.onFlushUpdateMetricUpdate)
 
 	var subscriptionCtx context.Context
 	subscriptionCtx, sip.subscriptionCancel = context.WithCancel(sip.Ctx())
@@ -979,6 +987,7 @@ func (r *rangeKeyBatcher) flush(ctx context.Context, toFlush mvccRangeKeyValues)
 
 	batchSummary := kvpb.BulkOpSummary{}
 	start, end := keys.MaxKey, keys.MinKey
+	var spanGroup roachpb.SpanGroup
 	for _, rangeKeyVal := range toFlush {
 		if err := sstWriter.PutRawMVCCRangeKey(rangeKeyVal.RangeKey, rangeKeyVal.Value); err != nil {
 			return err
@@ -991,6 +1000,7 @@ func (r *rangeKeyBatcher) flush(ctx context.Context, toFlush mvccRangeKeyValues)
 			end = rangeKeyVal.RangeKey.EndKey
 		}
 		batchSummary.DataSize += int64(rangeKeyVal.RangeKey.EncodedSize() + len(rangeKeyVal.Value))
+		spanGroup.Add(roachpb.Span{Key: rangeKeyVal.RangeKey.StartKey, EndKey: rangeKeyVal.RangeKey.EndKey})
 	}
 
 	// Finish the current batch.
@@ -1056,6 +1066,7 @@ func (r *rangeKeyBatcher) flush(ctx context.Context, toFlush mvccRangeKeyValues)
 			batchSummary.SSTDataSize += int64(len(data))
 		}
 	}
+	r.recomputeStats(ctx, spanGroup.Slice())
 
 	if r.onFlush != nil {
 		r.onFlush(batchSummary)
@@ -1064,6 +1075,52 @@ func (r *rangeKeyBatcher) flush(ctx context.Context, toFlush mvccRangeKeyValues)
 	return nil
 }
 
+func (r *rangeKeyBatcher) recomputeStats(ctx context.Context, rangekeySpans roachpb.Spans) {
+
+	rangeStartKeys := getRangeStartkeys(ctx, r.rangeDescIterFactory, rangekeySpans)
+	var b kv.Batch
+	// Sending RecomputeStatsRequests in one batch, allowing DistSender to
+	// parallelize the requests across ranges.
+	//
+	// TODO(msbutler): send this request asynchronously to prevent checkpoint
+	// delay.
+	for i := range rangeStartKeys {
+		b.AddRawRequest(&kvpb.RecomputeStatsRequest{
+			RequestHeader: kvpb.RequestHeader{Key: roachpb.Key(rangeStartKeys[i])},
+		})
+	}
+	if err := r.db.Run(ctx, &b); err != nil {
+		log.Warningf(ctx, "recomputes stats bath failed with error: %v", err)
+	}
+}
+
+func getRangeStartkeys(
+	ctx context.Context, rangeDescIterFactory rangedesc.IteratorFactory, rangeKeySpans roachpb.Spans,
+) []roachpb.RKey {
+	rangeStartKeysFound := make(map[string]struct{})
+	rangeStartKeys := make([]roachpb.RKey, 0)
+
+	for i := range rangeKeySpans {
+		iter, err := rangeDescIterFactory.NewLazyIterator(ctx, rangeKeySpans[i], 100)
+		if err != nil {
+			log.Warningf(ctx, "could not create range descriptor iterator for %s: %v", rangeKeySpans[i], err)
+			continue
+		}
+		for ; iter.Valid(); iter.Next() {
+			desc := iter.CurRangeDescriptor()
+			startKeyStr := desc.StartKey.String()
+			if _, ok := rangeStartKeysFound[startKeyStr]; !ok {
+				rangeStartKeysFound[startKeyStr] = struct{}{}
+				rangeStartKeys = append(rangeStartKeys, desc.StartKey)
+			}
+		}
+		if err := iter.Error(); err != nil {
+			log.Warningf(ctx, "error iterating range descriptors for %s: %v", rangeKeySpans[i], err)
+		}
+	}
+	return rangeStartKeys
+}
+
 // splitRangeKeySSTAtKey splits the given SST (passed as bytes) at the
 // given split key.
 //