Repository: blevesearch/bleve
Pull request #2187 (2025-04-25 11:29:37)

[BP] MB-66163: fix checkpointing mechanism under sparse mutation scenarios
Backporting #2186 to couchbase 7.6.x release cycle


Patch:
diff --git a/index/scorch/persister.go b/index/scorch/persister.go
index 05e7254c..119daf37 100644
--- a/index/scorch/persister.go
+++ b/index/scorch/persister.go
@@ -729,7 +729,7 @@ var boltStatsKey = []byte("stats")
 var TotBytesWrittenKey = []byte("TotBytesWritten")
 
 func (s *Scorch) loadFromBolt() error {
-	return s.rootBolt.View(func(tx *bolt.Tx) error {
+	err := s.rootBolt.View(func(tx *bolt.Tx) error {
 		snapshots := tx.Bucket(boltSnapshotsBucket)
 		if snapshots == nil {
 			return nil
@@ -779,6 +779,16 @@ func (s *Scorch) loadFromBolt() error {
 		}
 		return nil
 	})
+	if err != nil {
+		return err
+	}
+
+	persistedSnapshots, err := s.rootBoltSnapshotMetaData()
+	if err != nil {
+		return err
+	}
+	s.checkPoints = persistedSnapshots
+	return nil
 }
 
 // LoadSnapshot loads the segment with the specified epoch
@@ -1166,18 +1176,20 @@ func (s *Scorch) removeOldZapFiles() error {
 // duration. This results in all of them being purged from the boltDB
 // and the next iteration of the removeOldData() would end up protecting
 // latest contiguous snapshot which is a poor pattern in the rollback checkpoints.
-// Hence we try to retain atleast retentionFactor portion worth of old snapshots
+// Hence we try to retain atmost retentionFactor portion worth of old snapshots
 // in such a scenario using the following function
 func getBoundaryCheckPoint(retentionFactor float64,
 	checkPoints []*snapshotMetaData, timeStamp time.Time) time.Time {
 	if checkPoints != nil {
 		boundary := checkPoints[int(math.Floor(float64(len(checkPoints))*
 			retentionFactor))]
-		if timeStamp.Sub(boundary.timeStamp) < 0 {
-			// too less checkPoints would be left.
+		if timeStamp.Sub(boundary.timeStamp) > 0 {
+			// return the extended boundary which will dictate the older snapshots
+			// to be retained
 			return boundary.timeStamp
 		}
 	}
+
 	return timeStamp
 }
 


Test patch:
diff --git a/index/scorch/rollback_test.go b/index/scorch/rollback_test.go
index c52b30c2..4b5406e9 100644
--- a/index/scorch/rollback_test.go
+++ b/index/scorch/rollback_test.go
@@ -512,7 +512,7 @@ func TestBackupRacingWithPurge(t *testing.T) {
 	}
 
 	// replicate the following scenario of persistence of snapshots
-	// tc, tc - d/12, tc - d/6, tc - 3d/4, tc - 5d/6, tc - 6d/5, tc - 2d
+	// tc, tc - d/12, tc - d/6, tc - 3d/4, tc - 5d/6, tc - 6d/5
 	// approximate timestamps where there's a chance that the latest snapshot
 	// might not fit into the time-series
 	indexDummyData(t, scorchi, 1)
@@ -550,3 +550,74 @@ func TestBackupRacingWithPurge(t *testing.T) {
 		t.Fatalf("error copying the index: %v", err)
 	}
 }
+
+func TestSparseMutationCheckpointing(t *testing.T) {
+	cfg := CreateConfig("TestSparseMutationCheckpointing")
+	numSnapshotsToKeepOrig := NumSnapshotsToKeep
+	NumSnapshotsToKeep = 3
+	rollbackSamplingIntervalOrig := RollbackSamplingInterval
+	RollbackSamplingInterval = 2 * time.Second
+
+	err := InitTest(cfg)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		NumSnapshotsToKeep = numSnapshotsToKeepOrig
+		RollbackSamplingInterval = rollbackSamplingIntervalOrig
+		err := DestroyTest(cfg)
+		if err != nil {
+			t.Log(err)
+		}
+	}()
+
+	// disable merger and purger
+	RegistryEventCallbacks["test"] = func(e Event) bool {
+		if e.Kind == EventKindPreMergeCheck {
+			return false
+		}
+		return true
+	}
+	cfg["eventCallbackName"] = "test"
+	analysisQueue := index.NewAnalysisQueue(1)
+	idx, err := NewScorch(Name, cfg, analysisQueue)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	scorchi, ok := idx.(*Scorch)
+	if !ok {
+		t.Fatalf("Not a scorch index?")
+	}
+
+	err = scorchi.Open()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// create 4 snapshots every 2 seconds
+	indexDummyData(t, scorchi, 1)
+	time.Sleep(RollbackSamplingInterval)
+	indexDummyData(t, scorchi, 3)
+	time.Sleep(RollbackSamplingInterval)
+	indexDummyData(t, scorchi, 5)
+	time.Sleep(RollbackSamplingInterval)
+	indexDummyData(t, scorchi, 7)
+
+	// now the another snapshot is persisted outside of the window of checkpointing
+	// and we should be able to retain some older checkpoints as well along with
+	// the latest one
+	time.Sleep(time.Duration(NumSnapshotsToKeep) * RollbackSamplingInterval)
+	indexDummyData(t, scorchi, 9)
+
+	persistedSnapshots, err := scorchi.rootBoltSnapshotMetaData()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// should have more than 1 snapshots
+	protectedSnapshots := getProtectedSnapshots(RollbackSamplingInterval, NumSnapshotsToKeep, persistedSnapshots)
+	if len(protectedSnapshots) <= 1 {
+		t.Fatalf("expected %d protected snapshots, got %d", NumSnapshotsToKeep, len(protectedSnapshots))
+	}
+}