Merge #143930 #143963

143930: backup: fix elide skipped layers to account for new incremental path suffix r=msbutler a=kev-cao

Due to the changes introduced in #143226, the order of the backup manifests passed to `ElideSkippedLayers` is now changed. Whereas previously a compacted backup would always appear *after* any backup with the same end time, compacted backups now appear *before*. This meant that during restore, compacted backups ended up always being skipped and never used.

To fix this, prior to running the `ElideSkippedLayers` algorithm, we now remove duplicate backups (i.e. same end time) and keep the backup with the earliest start time (aka the most compacted backup).

Epic: None

Release note: None

143963: cspann: fix race conditions in index split operation r=drewkimball a=andy-kimball

This commit fixes two race conditions in the index split operation:

1. After setting the state of the left sub-partition to Ready, say that the split unexpectedly fails. Now say that the left sub- partition itself splits and is deleted. When the original split resumes, it will not be able to get the centroid for the left sub-partition, which is needed to run the K-means clustering algorithm.

2. As described by 1, it's possible that a splitting partition references target sub-partitions that are now missing from the index. This will trigger PartitionNotFound errors in insert code paths.

The fixes are:

1. Update the logic so that vectors are first copied to the left and right sub-partitions before either sub-partition's state is updated from Updating to Ready. Only Ready sub-partitions can be split, so this should prevent race condition 1.

2. Update the insert logic so that searches of non-root partitions return multiple results, to make it extremely likely that a suitable insert partition will be found. For root partitions, check split target sub-partitions instead, since the splitting partition does not share a parent with its sub-partitions.

Epic: CRDB-42943

Release note: None

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

Author: 3 people

pkg/backup/backupinfo/manifest_handling.go

@@ -887,11 +887,19 @@ func ValidateEndTimeAndTruncate(
 	endTime hlc.Timestamp,
 	includeSkipped bool,
 ) ([]string, []backuppb.BackupManifest, []jobspb.RestoreDetails_BackupLocalityInfo, error) {
+
 	if endTime.IsEmpty() {
 		if includeSkipped {
 			return defaultURIs, mainBackupManifests, localityInfo, nil
 		}
-		return ElideSkippedLayers(defaultURIs, mainBackupManifests, localityInfo)
+		uris, manifests, locality, err := ElideSkippedLayers(defaultURIs, mainBackupManifests, localityInfo)
+		if err != nil {
+			return nil, nil, nil, err
+		}
+		if err := validateContinuity(manifests, endTime); err != nil {
+			return nil, nil, nil, err
+		}
+		return uris, manifests, locality, nil
 	}
 	for i, b := range mainBackupManifests {
 		// Find the backup that covers the requested time.
@@ -930,7 +938,16 @@ func ValidateEndTimeAndTruncate(
 		if includeSkipped {
 			return defaultURIs[:i+1], mainBackupManifests[:i+1], localityInfo[:i+1], nil
 		}
-		return ElideSkippedLayers(defaultURIs[:i+1], mainBackupManifests[:i+1], localityInfo[:i+1])
+		uris, manifests, locality, err := ElideSkippedLayers(
+			defaultURIs[:i+1], mainBackupManifests[:i+1], localityInfo[:i+1],
+		)
+		if err != nil {
+			return nil, nil, nil, err
+		}
+		if err := validateContinuity(manifests, endTime); err != nil {
+			return nil, nil, nil, err
+		}
+		return uris, manifests, locality, nil
 
 	}
 
@@ -939,10 +956,42 @@ func ValidateEndTimeAndTruncate(
 	)
 }
 
-// ElideSkippedLayers removes backups that are skipped in the backup chain.
+// ValidateContinuity checks that the backups are continuous and cover the
+// requested end time.
+func validateContinuity(manifests []backuppb.BackupManifest, endTime hlc.Timestamp) error {
+	if len(manifests) == 0 {
+		return errors.AssertionFailedf("an empty chain of backups cannot cover an end time")
+	}
+	for i := range len(manifests) - 1 {
+		if !manifests[i].EndTime.Equal(manifests[i+1].StartTime) {
+			return errors.AssertionFailedf(
+				"backups are not continuous: %dth backup ends at %+v, %dth backup starts at %+v",
+				i, manifests[i].EndTime,
+				i+1, manifests[i+1].StartTime,
+			)
+		}
+	}
+	if !endTime.IsEmpty() {
+		lastManifest := manifests[len(manifests)-1]
+		if !lastManifest.StartTime.Less(endTime) || !endTime.LessEq(lastManifest.EndTime) {
+			return errors.AssertionFailedf(
+				"requested time %s is not covered by the last backup",
+				endTime,
+			)
+		}
+	}
+	return nil
+}
+
+// ElideSkippedLayers removes backups that are skipped in the backup chain and
+// ensures only backups that will be used in the restore are returned.
+//
+// Note: This assumes that the provided backups are sorted in increasing order
+// by end time, and then sorted in increasing order by start time to break ties.
 func ElideSkippedLayers(
 	uris []string, backups []backuppb.BackupManifest, loc []jobspb.RestoreDetails_BackupLocalityInfo,
 ) ([]string, []backuppb.BackupManifest, []jobspb.RestoreDetails_BackupLocalityInfo, error) {
+	uris, backups, loc = elideDuplicateEndTimes(uris, backups, loc)
 	i := len(backups) - 1
 	for i > 0 {
 		// Find j such that backups[j] is parent of backups[i].
@@ -959,10 +1008,37 @@ func ElideSkippedLayers(
 		// Move up to check the chain from j now.
 		i = j
 	}
-
 	return uris, backups, loc, nil
 }
 
+// elideDuplicateEndTimes ensures that backups in a list of backups are
+// functionally unique by removing any duplicates that have the same end time,
+// choosing backups with earlier start times and eliding the rest.
+//
+// Note: This assumes that the provided backups are sorted in increasing order
+// by end time, and then sorted in increasing order by start time to break ties.
+// This is the case for backups being returned by storage clients due to us
+// encoding backup paths in a way that ensures this order.
+func elideDuplicateEndTimes(
+	uris []string, backups []backuppb.BackupManifest, loc []jobspb.RestoreDetails_BackupLocalityInfo,
+) ([]string, []backuppb.BackupManifest, []jobspb.RestoreDetails_BackupLocalityInfo) {
+	for i := range len(backups) - 1 {
+		j := i + 1
+		// Find j such that backups[j] no longer shares the same end time as
+		// backups[i].
+		for j < len(backups) && backups[i].EndTime.Equal(backups[j].EndTime) {
+			j++
+		}
+		// If there exists backups between i and j, remove them.
+		if j > i+1 {
+			uris = slices.Delete(uris, i+1, j)
+			backups = slices.Delete(backups, i+1, j)
+			loc = slices.Delete(loc, i+1, j)
+		}
+	}
+	return uris, backups, loc
+}
+
 // GetBackupIndexAtTime returns the index of the latest backup in
 // `backupManifests` with a StartTime >= asOf.
 func GetBackupIndexAtTime(

pkg/backup/backupinfo/manifest_handling_test.go

@@ -379,18 +379,46 @@ func TestMakeBackupCodec(t *testing.T) {
 func TestElideSkippedLayers(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 
+	// Note: The tests here work under the assumption that the input lists are
+	// always sorted in ascending order by end time, and then sorted in ascending
+	// order by start time.
 	for _, tc := range []struct {
 		name     string
 		times    [][]int // len 2 slices of start and end time.
-		expected []int   // expected end times.
+		expected [][]int // expected start and end times
 	}{
-		{"single", [][]int{{0, 1}}, []int{1}},
-		{"double", [][]int{{0, 1}, {1, 2}}, []int{1, 2}},
-		{"simple chain", [][]int{{0, 1}, {1, 2}, {2, 3}, {3, 5}, {5, 8}}, []int{1, 2, 3, 5, 8}},
-		{"skip one", [][]int{{0, 1}, {1, 2}, {1, 3}, {3, 5}, {5, 8}}, []int{1, 3, 5, 8}},
-		{"skip all", [][]int{{0, 1}, {1, 2}, {1, 3}, {3, 5}, {1, 8}}, []int{1, 8}},
-		{"skip twice to first", [][]int{{0, 1}, {1, 2}, {1, 3}, {3, 5}, {3, 8}}, []int{1, 3, 8}},
-		{"skip twice to second", [][]int{{0, 1}, {1, 2}, {1, 3}, {3, 5}, {2, 8}}, []int{1, 2, 8}},
+		{"single", [][]int{{0, 1}}, [][]int{{0, 1}}},
+		{"double", [][]int{{0, 1}, {1, 2}}, [][]int{{0, 1}, {1, 2}}},
+		{
+			"simple chain, no skips",
+			[][]int{{0, 1}, {1, 2}, {2, 3}, {3, 5}, {5, 8}},
+			[][]int{{0, 1}, {1, 2}, {2, 3}, {3, 5}, {5, 8}},
+		},
+		{
+			"compaction of two backups",
+			[][]int{{0, 1}, {1, 2}, {1, 3}, {2, 3}, {3, 5}, {5, 8}},
+			[][]int{{0, 1}, {1, 3}, {3, 5}, {5, 8}},
+		},
+		{
+			"compaction of entire chain",
+			[][]int{{0, 1}, {1, 2}, {2, 3}, {3, 5}, {0, 8}, {5, 8}},
+			[][]int{{0, 8}},
+		},
+		{
+			"two compactions of two backups",
+			[][]int{{0, 1}, {1, 2}, {1, 3}, {2, 3}, {3, 5}, {3, 8}, {5, 8}},
+			[][]int{{0, 1}, {1, 3}, {3, 8}},
+		},
+		{
+			"compaction includes a compacted backup in the middle",
+			[][]int{{0, 1}, {1, 2}, {1, 3}, {2, 3}, {3, 5}, {1, 8}, {5, 8}},
+			[][]int{{0, 1}, {1, 8}},
+		},
+		{
+			"two compactions with the same end time",
+			[][]int{{0, 1}, {1, 2}, {2, 3}, {3, 5}, {1, 8}, {3, 8}, {5, 8}},
+			[][]int{{0, 1}, {1, 8}},
+		},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			chain := make([]backuppb.BackupManifest, len(tc.times))
@@ -408,7 +436,8 @@ func TestElideSkippedLayers(t *testing.T) {
 			require.Equal(t, len(tc.expected), len(locs))
 			require.Equal(t, len(tc.expected), len(res))
 			for i := range tc.expected {
-				require.Equal(t, tc.expected[i], int(res[i].EndTime.WallTime), "expected %q\ngot: %q")
+				actual := []int{int(res[i].StartTime.WallTime), int(res[i].EndTime.WallTime)}
+				require.Equal(t, tc.expected[i], actual)
 			}
 		})
 	}

pkg/sql/vecindex/cspann/fixup_split.go

@@ -39,11 +39,11 @@ import (
 //     but instead are "forwarded" to the closest target sub-partition.
 //  7. Reload the splitting partition's vectors and copy the "left" subset to
 //     the left sub-partition.
-//  8. Update the left sub-partition's state from Updating to Ready.
-//  9. Copy the "right" subset of vectors to the right sub-partition.
-//  10. Update the right sub-partition's state from Updating to Ready. At this
+//  8. Copy the "right" subset of vectors to the right sub-partition. At this
 //     point, the splitting vectors are duplicated in the index. Any searches
 //     will filter out duplicates.
+//  9. Update the left sub-partition's state from Updating to Ready.
+//  10. Update the right sub-partition's state from Updating to Ready.
 //  11. Remove the splitting partition from its parent. The duplicates are no
 //     longer visible to searches.
 //  12. Delete the splitting partition from the index.
@@ -227,10 +227,30 @@ func (fw *fixupWorker) splitPartition(
 			return err
 		}
 
-		// Add vectors to nearest partition.
-		err = fw.copyToSplitSubPartitions(ctx, partition, vectors, leftMetadata, rightMetadata)
-		if err != nil {
-			return err
+		// If still updating the sub-partitions, then distribute vectors among them.
+		if leftMetadata.StateDetails.State == UpdatingState {
+			err = fw.copyToSplitSubPartitions(ctx, partition, vectors, leftMetadata, rightMetadata)
+			if err != nil {
+				return err
+			}
+		}
+
+		// Update sub-partition states from Updating to Ready.
+		if leftMetadata.StateDetails.State == UpdatingState {
+			expected := leftMetadata
+			leftMetadata.StateDetails = MakeReadyDetails()
+			err = fw.updateMetadata(ctx, leftPartitionKey, leftMetadata, expected)
+			if err != nil {
+				return err
+			}
+		}
+		if rightMetadata.StateDetails.State == UpdatingState {
+			expected := rightMetadata
+			rightMetadata.StateDetails = MakeReadyDetails()
+			err = fw.updateMetadata(ctx, rightPartitionKey, rightMetadata, expected)
+			if err != nil {
+				return err
+			}
 		}
 
 		// Check whether the splitting partition is the root.
@@ -615,8 +635,7 @@ func (fw *fixupWorker) deletePartition(
 }
 
 // copyToSplitSubPartitions copies the given set of vectors to left and right
-// sub-partitions, based on which centroid they're closer to. It also updates
-// the state of each sub-partition from Updating to Ready.
+// sub-partitions, based on which centroid they're closer to.
 func (fw *fixupWorker) copyToSplitSubPartitions(
 	ctx context.Context,
 	sourcePartition *Partition,
@@ -659,57 +678,25 @@ func (fw *fixupWorker) copyToSplitSubPartitions(
 	// transactional; if an error occurs, any vectors already added may not be
 	// rolled back. This is OK, since the vectors are still present in the
 	// source partition.
-	if leftMetadata.StateDetails.State == UpdatingState {
-		leftPartitionKey := sourceState.Target1
-		err = fw.copyVectorsToSubPartition(ctx,
-			leftPartitionKey, leftMetadata, leftVectors, leftChildKeys, leftValueBytes)
-		if err != nil {
-			return err
-		}
-	}
-	if rightMetadata.StateDetails.State == UpdatingState {
-		if sourcePartition.Level() != LeafLevel && vectors.Count == 1 {
-			// This should have been a merge, not a split, but we're too far into the
-			// split operation to back out now, so avoid an empty non-root partition by
-			// duplicating the last remaining vector in both partitions.
-			rightVectors = leftVectors
-			rightChildKeys = leftChildKeys
-			rightValueBytes = leftValueBytes
-		}
-
-		rightPartitionKey := sourceState.Target2
-		err = fw.copyVectorsToSubPartition(ctx,
-			rightPartitionKey, rightMetadata, rightVectors, rightChildKeys, rightValueBytes)
-		if err != nil {
-			return err
-		}
-	}
-
-	return nil
-}
-
-// copyVectorsToSubPartition copies the given set of vectors, along with
-// associated keys and values, to a split sub-partition with the given key. The
-// vectors will only be added if the partition's metadata matches the expected
-// value. It also updates the state of the partition from Updating to Ready.
-func (fw *fixupWorker) copyVectorsToSubPartition(
-	ctx context.Context,
-	partitionKey PartitionKey,
-	metadata PartitionMetadata,
-	vectors vector.Set,
-	childKeys []ChildKey,
-	valueBytes []ValueBytes,
-) error {
-	// Add vectors to sub-partition, as long as metadata matches.
-	err := fw.addToPartition(ctx, partitionKey, vectors, childKeys, valueBytes, metadata)
+	leftPartitionKey := sourceState.Target1
+	err = fw.addToPartition(ctx,
+		leftPartitionKey, leftVectors, leftChildKeys, leftValueBytes, leftMetadata)
 	if err != nil {
 		return err
 	}
 
-	// Update partition state from Updating to Ready.
-	expected := metadata
-	metadata.StateDetails = MakeReadyDetails()
-	err = fw.updateMetadata(ctx, partitionKey, metadata, expected)
+	if sourcePartition.Level() != LeafLevel && vectors.Count == 1 {
+		// This should have been a merge, not a split, but we're too far into the
+		// split operation to back out now, so avoid an empty non-root partition by
+		// duplicating the last remaining vector in both partitions.
+		rightVectors = leftVectors
+		rightChildKeys = leftChildKeys
+		rightValueBytes = leftValueBytes
+	}
+
+	rightPartitionKey := sourceState.Target2
+	err = fw.addToPartition(ctx,
+		rightPartitionKey, rightVectors, rightChildKeys, rightValueBytes, rightMetadata)
 	if err != nil {
 		return err
 	}