Merge #144668 #144800

144668: changefeedccl: make quantization metamorphic r=andyyang890 a=rharding6373

This PR turns changefeed.resolved_timestamp.granularity into a metamorphic test constant.

Epic: none
Fixes: #144632

Release note: None

144800: cspann: improve TestIndexConcurrency and fix issues it found r=drewkimball a=andy-kimball

Improve the TestIndexConcurrency test and fix issues it found.

#### cspann: fix buglet in searcher

Fix a small bug, where we were not setting s.levels in the case where
we're inserting into the root partition. This was causing crdb_test runs
to fail, since it reallocates the s.levels buffer every time..

#### cspann: fix data race in memstore.GetFullVectors

The race detector found a data race in MemStore's GetFullVectors that was
caused by not locking when accessing a partition's centroid. This was
deliberate, because the centroid was thought to be immutable, but the
TryClearPartition code sets the centroid to itself, which triggers the
race detector. While we could change TryClearPartition to avoid that, it's
probably best to just do the locking.

However, that causes another problem, which is that the MemStore locking
needed for getting partition centroids is different than the locking
needed to get primary key vectors. Update the GetFullVectors API to make
it clear that we can only ask for one or the other in a single call, so
that we don't need to handle the case where partition keys and primary
keys are interleaved in the same request.

#### cspann: set correct parent partition in fallbackOnTargets

If an insert fails to insert a vector into a partition that does not allow
it, it calls fallbackOnTargets to redirect to one of the target partitions.
However, the existing code does not set the parent partition correctly for
target partitions. A split of a non-root partition should set the target
partitions' parent as the parent of the splitting partition, not the
splitting partition itself.

#### cspann: add better control over search for update retries

The retry logic in searchForUpdateHelper has a couple of problems:

- In the event of an infinite retry loop bug, it can stack overflow, since
  it recursively calls itself.
- Race conditions with the MemStore can cause it to not find a valid insert
  partition. Reads of MemStore partitions race with background fixups that
  update and delete the partitions.

This commit improves the situation by specifying the maximum number of
insert/delete attempts we'll make before giving up. The remaining attempts
are preserved when making recursive retry calls, which prevents infinite
retry loops.

#### cspann: add DeletingForSplit state to split fixup flow

Under heavy stress, a target partition can be split and deleted before
its source partition finishes its own split. This results in a situation
where the source partition is still pointing to target partitions that are
now deleted. To prevent that state, this commit adds a new DeletingForSplit
state into the split fixup flow, to be used for non-root partitions. After
vectors have been copied to target partitions, the splitting partition is
marked as DeletingForSplit. Next, the target partitions are marked Ready,
and can now safely be split or merged themselves. Finally, the splitting
partition is actually removed from the tree and deleted.

#### cspann: fix partition reload race condition

During split, we reload a partition's vectors, in case any have changed
while we created target partitions. However, it's also possible that
another worker has updated the splitting partition. In that case, we
should abort and let the racing worker take over. The existing code
was not doing that, and ended up trying to continue the split, but
with incorrect target partition keys.

#### cspann: improve logging during concurrent splits

The existing logging makes it difficult to debug bugs caused by multiple
concurrent workers racing to split a partition. This commit updates the
logging to:

- log after every key step in the split process
- only log for the worker that actually wins the race for this step

These changes reduce logging noise while still improving what does get
logged.

#### memstore: restart operation when reading a deleted partition

When the memstore comes across a deleted partition, there are two cases:

1. The txn was started before the deletion, in which case it should be
   restarted, so that any search can find a different path through the tree.
2. The txn was started after the deletion, in which case ErrPartitionNotFound
   should be returned.

#### memstore: fix race condition creating an empty partition

The memstore.TryCreateEmptyPartition method has a race condition, such
that two callers can end up creating different instances of the same
partition. This commit fixes that by checking whether the partition
already exists and then creating it if needed, all within the scope of
the same lock.

This change also uncovered an existing bug in splitPartition, in which
we weren't fetching metadata for the left and right sub-partitions when
restarting the split in the AddingLevel state.

#### cspann: simulate multiple index instances in TestIndexConcurrency

Create multiple index instances in the TestIndexConcurrency test, all hooked
up to the same store. This simulates multiple CRDB nodes, each independently
inserting, removing, searching, and splitting a shared index. Eliminate the
delay for one instance assisting another, in order to maximize the possibility
of race conditions.

Co-authored-by: rharding6373 <[email protected]>
Co-authored-by: Andrew Kimball <[email protected]>

Author: 3 people

pkg/ccl/changefeedccl/changefeedbase/settings.go

@@ -345,6 +345,6 @@ var DefaultLaggingRangesPollingInterval = 1 * time.Minute
 var Quantize = settings.RegisterDurationSettingWithExplicitUnit(
 	settings.ApplicationLevel,
 	"changefeed.resolved_timestamp.granularity",
-	"the granularity at which changefeed progress are quantized to make tracking more efficient",
-	0,
+	"the granularity at which changefeed progress is quantized to make tracking more efficient",
+	time.Duration(metamorphic.ConstantWithTestRange("changefeed.resolved_timestamp.granularity", 0, 0, 20))*time.Second,
 )

pkg/cmd/roachtest/tests/cdc.go

@@ -778,11 +778,17 @@ func newCDCTester(ctx context.Context, t test.Test, c cluster.Cluster, opts ...o
 
 	startOpts, settings := makeCDCBenchOptions(c)
 
-	// With a target_duration of 10s, we won't see slow span logs from changefeeds untils we are > 100s
+	// With a target_duration of 10s, we won't see slow span logs from changefeeds until we are > 100s
 	// behind, which is well above the 60s targetSteadyLatency we have in some tests.
 	settings.ClusterSettings["changefeed.slow_span_log_threshold"] = "30s"
 	settings.ClusterSettings["server.child_metrics.enabled"] = "true"
 
+	// Randomly set a quantization interval since metamorphic settings
+	// don't extend to roachtests.
+	quantization := fmt.Sprintf("%ds", rand.Intn(30))
+	settings.ClusterSettings["changefeed.resolved_timestamp.granularity"] = quantization
+	t.Status(fmt.Sprintf("changefeed.resolved_timestamp.granularity: %s", quantization))
+
 	settings.Env = append(settings.Env, envVars...)
 
 	c.Start(ctx, t.L(), startOpts, settings, tester.crdbNodes)

pkg/sql/vecindex/cspann/BUILD.bazel

@@ -76,6 +76,7 @@ go_test(
         "//pkg/util/log",
         "//pkg/util/num32",
         "//pkg/util/stop",
+        "//pkg/util/syncutil",
         "//pkg/util/vector",
         "@com_github_cockroachdb_crlib//crtime",
         "@com_github_cockroachdb_datadriven//:datadriven",

pkg/sql/vecindex/cspann/commontest/storetests.go

@@ -129,7 +129,7 @@ func (suite *StoreTestSuite) TestGetPartitionMetadata() {
 				suite.ctx, treeKey, cspann.RootKey, false /* forUpdate */)
 			suite.NoError(err)
 			CheckPartitionMetadata(suite.T(), metadata, cspann.LeafLevel, vector.T{0, 0},
-				cspann.MakeReadyDetails())
+				cspann.PartitionStateDetails{State: cspann.ReadyState})
 
 			// Non-root partition does not yet exist, expect error.
 			_, err = txn.GetPartitionMetadata(
@@ -145,12 +145,12 @@ func (suite *StoreTestSuite) TestGetPartitionMetadata() {
 			metadata, err = txn.GetPartitionMetadata(
 				suite.ctx, treeKey, partitionKey, true /* forUpdate */)
 			CheckPartitionMetadata(suite.T(), metadata, cspann.SecondLevel, vector.T{4, 3},
-				cspann.MakeReadyDetails())
+				cspann.PartitionStateDetails{State: cspann.ReadyState})
 		})
 
 		// Update the partition state to DrainingForSplit.
 		expected := metadata
-		metadata.StateDetails = cspann.MakeDrainingForSplitDetails(20, 30)
+		metadata.StateDetails.MakeDrainingForSplit(20, 30)
 		suite.NoError(store.TryUpdatePartitionMetadata(
 			suite.ctx, treeKey, partitionKey, metadata, expected))
 
@@ -159,15 +159,16 @@ func (suite *StoreTestSuite) TestGetPartitionMetadata() {
 			metadata, err := txn.GetPartitionMetadata(
 				suite.ctx, treeKey, partitionKey, false /* forUpdate */)
 			suite.NoError(err)
-			CheckPartitionMetadata(suite.T(), metadata, cspann.SecondLevel, vector.T{4, 3},
-				cspann.MakeDrainingForSplitDetails(20, 30))
+			details := cspann.PartitionStateDetails{
+				State: cspann.DrainingForSplitState, Target1: 20, Target2: 30}
+			CheckPartitionMetadata(suite.T(), metadata, cspann.SecondLevel, vector.T{4, 3}, details)
 
 			// If forUpdate = true, GetPartitionMetadata should error.
 			var errConditionFailed *cspann.ConditionFailedError
 			_, err = txn.GetPartitionMetadata(suite.ctx, treeKey, partitionKey, true /* forUpdate */)
 			suite.ErrorAs(err, &errConditionFailed)
 			CheckPartitionMetadata(suite.T(), errConditionFailed.Actual, cspann.SecondLevel,
-				vector.T{4, 3}, cspann.MakeDrainingForSplitDetails(20, 30))
+				vector.T{4, 3}, details)
 		})
 		suite.NoError(err)
 	}
@@ -223,7 +224,7 @@ func (suite *StoreTestSuite) TestAddToPartition() {
 		partition, err := store.TryGetPartition(suite.ctx, treeKey, cspann.RootKey)
 		suite.NoError(err)
 		CheckPartitionMetadata(suite.T(), *partition.Metadata(), cspann.LeafLevel,
-			vector.T{0, 0}, cspann.MakeReadyDetails())
+			vector.T{0, 0}, cspann.PartitionStateDetails{State: cspann.ReadyState})
 		suite.Equal([]cspann.ChildKey{primaryKey1, primaryKey2}, partition.ChildKeys())
 		suite.Equal([]cspann.ValueBytes{valueBytes1, valueBytes3}, partition.ValueBytes())
 
@@ -241,7 +242,7 @@ func (suite *StoreTestSuite) TestAddToPartition() {
 		// Update the partition state to DrainingForMerge.
 		expected := *partition.Metadata()
 		metadata := expected
-		metadata.StateDetails = cspann.MakeDrainingForMergeDetails(20)
+		metadata.StateDetails.MakeDrainingForMerge(20)
 		suite.NoError(store.TryUpdatePartitionMetadata(
 			suite.ctx, treeKey, partitionKey, metadata, expected))
 
@@ -251,8 +252,9 @@ func (suite *StoreTestSuite) TestAddToPartition() {
 			err = txn.AddToPartition(suite.ctx, treeKey, partitionKey, cspann.SecondLevel,
 				vec4, partitionKey4, valueBytes4)
 			suite.ErrorAs(err, &errConditionFailed)
+			details := cspann.PartitionStateDetails{State: cspann.DrainingForMergeState, Target1: 20}
 			CheckPartitionMetadata(suite.T(), errConditionFailed.Actual, cspann.SecondLevel,
-				vector.T{4, 3}, cspann.MakeDrainingForMergeDetails(20))
+				vector.T{4, 3}, details)
 		})
 	}
 
@@ -289,9 +291,12 @@ func (suite *StoreTestSuite) TestRemoveFromPartition() {
 			err = txn.AddToPartition(suite.ctx, treeKey, cspann.RootKey, cspann.LeafLevel,
 				vec1, primaryKey1, valueBytes1)
 			suite.NoError(err)
+		})
+		CheckPartitionCount(suite.ctx, suite.T(), store, treeKey, cspann.RootKey, 1)
 
-			// Remove that vector.
-			err = txn.RemoveFromPartition(suite.ctx, treeKey, cspann.RootKey,
+		// Remove the vector that was added to the root.
+		RunTransaction(suite.ctx, suite.T(), store, func(txn cspann.Txn) {
+			err := txn.RemoveFromPartition(suite.ctx, treeKey, cspann.RootKey,
 				cspann.LeafLevel, primaryKey1)
 			suite.NoError(err)
 		})
@@ -311,7 +316,7 @@ func (suite *StoreTestSuite) TestRemoveFromPartition() {
 		// Update the partition state to DrainingForSplit.
 		expected := *partition.Metadata()
 		metadata := expected
-		metadata.StateDetails = cspann.MakeDrainingForSplitDetails(20, 30)
+		metadata.StateDetails.MakeDrainingForSplit(20, 30)
 		suite.NoError(store.TryUpdatePartitionMetadata(
 			suite.ctx, treeKey, partitionKey, metadata, expected))
 
@@ -321,8 +326,10 @@ func (suite *StoreTestSuite) TestRemoveFromPartition() {
 			err := txn.RemoveFromPartition(suite.ctx, treeKey, partitionKey, cspann.SecondLevel,
 				partitionKey3)
 			suite.ErrorAs(err, &errConditionFailed)
+			details := cspann.PartitionStateDetails{
+				State: cspann.DrainingForSplitState, Target1: 20, Target2: 30}
 			CheckPartitionMetadata(suite.T(), errConditionFailed.Actual, cspann.SecondLevel,
-				vector.T{4, 3}, cspann.MakeDrainingForSplitDetails(20, 30))
+				vector.T{4, 3}, details)
 		})
 	}
 
@@ -351,10 +358,10 @@ func (suite *StoreTestSuite) TestSearchPartitions() {
 		// Create another partition.
 		testPartitionKey2 := cspann.PartitionKey(20)
 		metadata := cspann.PartitionMetadata{
-			Level:        cspann.SecondLevel,
-			Centroid:     vector.T{2, 4},
-			StateDetails: cspann.MakeSplittingDetails(10, 20),
+			Level:    cspann.SecondLevel,
+			Centroid: vector.T{2, 4},
 		}
+		metadata.StateDetails.MakeSplitting(10, 20)
 		suite.NoError(store.TryCreateEmptyPartition(suite.ctx, treeKey, testPartitionKey2, metadata))
 		vectors := vector.MakeSet(2)
 		vectors.Add(vec4)
@@ -423,40 +430,51 @@ func (suite *StoreTestSuite) TestGetFullVectors() {
 		// Create partitions.
 		treeKey := store.MakeTreeKey(suite.T(), treeID)
 		metadata := cspann.PartitionMetadata{
-			Level:        cspann.SecondLevel,
-			Centroid:     vector.T{0, 0},
-			StateDetails: cspann.MakeSplittingDetails(20, 30),
+			Level:    cspann.SecondLevel,
+			Centroid: vector.T{0, 0},
 		}
+		metadata.StateDetails.MakeSplitting(20, 30)
 		suite.NoError(store.TryCreateEmptyPartition(suite.ctx, treeKey, cspann.RootKey, metadata))
 		partitionKey, _ := suite.createTestPartition(store, treeKey)
 
 		RunTransaction(suite.ctx, suite.T(), store, func(txn cspann.Txn) {
+			// Empty request set.
+			err := txn.GetFullVectors(suite.ctx, treeKey, []cspann.VectorWithKey{})
+			suite.NoError(err)
+
 			// Insert some full vectors into the test store.
 			key1 := store.InsertVector(suite.T(), treeID, vec1)
 			key2 := store.InsertVector(suite.T(), treeID, vec2)
 			key3 := store.InsertVector(suite.T(), treeID, vec3)
 
-			// Include primary keys, partition keys, and keys that cannot be found.
+			// Start by fetching partition keys, both that exist and that do not.
 			results := []cspann.VectorWithKey{
+				{Key: cspann.ChildKey{PartitionKey: cspann.RootKey}},
+				{Key: cspann.ChildKey{PartitionKey: cspann.PartitionKey(99)}}, // No such partition.
+				{Key: cspann.ChildKey{PartitionKey: partitionKey}},
+			}
+			err = txn.GetFullVectors(suite.ctx, treeKey, results)
+			suite.NoError(err)
+			suite.Equal(vector.T{0, 0}, results[0].Vector)
+			suite.Nil(results[1].Vector)
+			suite.Equal(vector.T{4, 3}, results[2].Vector)
+
+			// Next fetch primary keys that reference vectors that exist and that
+			// do not exist.
+			results = []cspann.VectorWithKey{
 				{Key: cspann.ChildKey{KeyBytes: key1}},
 				{Key: cspann.ChildKey{KeyBytes: cspann.KeyBytes{0}}},
-				{Key: cspann.ChildKey{PartitionKey: cspann.RootKey}},
 				{Key: cspann.ChildKey{KeyBytes: key2}},
 				{Key: cspann.ChildKey{KeyBytes: cspann.KeyBytes{0}}},
-				{Key: cspann.ChildKey{PartitionKey: partitionKey}},
 				{Key: cspann.ChildKey{KeyBytes: key3}},
-				{Key: cspann.ChildKey{PartitionKey: cspann.PartitionKey(99)}}, // No such partition.
 			}
-			err := txn.GetFullVectors(suite.ctx, treeKey, results)
+			err = txn.GetFullVectors(suite.ctx, treeKey, results)
 			suite.NoError(err)
 			suite.Equal(vec1, results[0].Vector)
 			suite.Nil(results[1].Vector)
-			suite.Equal(vector.T{0, 0}, results[2].Vector)
-			suite.Equal(vec2, results[3].Vector)
-			suite.Nil(results[4].Vector)
-			suite.Equal(vector.T{4, 3}, results[5].Vector)
-			suite.Equal(vec3, results[6].Vector)
-			suite.Nil(results[7].Vector)
+			suite.Equal(vec2, results[2].Vector)
+			suite.Nil(results[3].Vector)
+			suite.Equal(vec3, results[4].Vector)
 
 			// Grab another set of vectors to ensure that saved state is properly reset.
 			results = []cspann.VectorWithKey{
@@ -530,10 +548,10 @@ func (suite *StoreTestSuite) TestTryCreateEmptyPartition() {
 
 		// Create empty partition.
 		metadata := cspann.PartitionMetadata{
-			Level:        cspann.SecondLevel,
-			Centroid:     centroid,
-			StateDetails: cspann.MakeSplittingDetails(20, 30),
+			Level:    cspann.SecondLevel,
+			Centroid: centroid,
 		}
+		metadata.StateDetails.MakeSplitting(20, 30)
 		suite.NoError(store.TryCreateEmptyPartition(suite.ctx, treeKey, partitionKey, metadata))
 
 		// Fetch back the partition and validate it.
@@ -622,10 +640,10 @@ func (suite *StoreTestSuite) TestTryGetPartition() {
 
 		// Create partition with some vectors in it.
 		metadata := cspann.PartitionMetadata{
-			Level:        cspann.LeafLevel,
-			Centroid:     centroid,
-			StateDetails: cspann.MakeUpdatingDetails(20),
+			Level:    cspann.LeafLevel,
+			Centroid: centroid,
 		}
+		metadata.StateDetails.MakeUpdating(20)
 		suite.NoError(store.TryCreateEmptyPartition(suite.ctx, treeKey, partitionKey, metadata))
 		vectors := vector.MakeSet(2)
 		vectors.Add(vec1)
@@ -683,7 +701,7 @@ func (suite *StoreTestSuite) TestTryGetPartitionMetadata() {
 		// Update the metadata and verify we get the updated values.
 		expected := *partition.Metadata()
 		metadata := expected
-		metadata.StateDetails = cspann.MakeUpdatingDetails(30)
+		metadata.StateDetails.MakeUpdating(30)
 		suite.NoError(store.TryUpdatePartitionMetadata(
 			suite.ctx, treeKey, partitionKey, metadata, expected))
 
@@ -761,10 +779,10 @@ func (suite *StoreTestSuite) TestTryAddToPartition() {
 
 		// Partition does not yet exist.
 		metadata := cspann.PartitionMetadata{
-			Level:        cspann.LeafLevel,
-			Centroid:     centroid,
-			StateDetails: cspann.MakeUpdatingDetails(20),
+			Level:    cspann.LeafLevel,
+			Centroid: centroid,
 		}
+		metadata.StateDetails.MakeUpdating(20)
 		addVectors := vector.MakeSet(2)
 		addVectors.Add(vec1)
 		addVectors.Add(vec2)
@@ -1006,11 +1024,7 @@ func (suite *StoreTestSuite) createTestPartition(
 	store TestStore, treeKey cspann.TreeKey,
 ) (cspann.PartitionKey, *cspann.Partition) {
 	partitionKey := cspann.PartitionKey(10)
-	metadata := cspann.PartitionMetadata{
-		Level:        cspann.SecondLevel,
-		Centroid:     vector.T{4, 3},
-		StateDetails: cspann.MakeReadyDetails(),
-	}
+	metadata := cspann.MakeReadyPartitionMetadata(cspann.SecondLevel, vector.T{4, 3})
 	suite.NoError(store.TryCreateEmptyPartition(suite.ctx, treeKey, partitionKey, metadata))
 	vectors := vector.MakeSet(2)
 	vectors.Add(vec1)