added comments

Author: darkcoderrises

posting/index.go

@@ -148,8 +148,6 @@ func (mp *MutationPipeline) InsertTokenizerIndexes(ctx context.Context, pipeline
 		}
 	}
 
-	wg := &sync.WaitGroup{}
-
 	strings := make([]string, 0, len(values))
 	for i := range values {
 		strings = append(strings, i)
@@ -158,6 +156,8 @@ func (mp *MutationPipeline) InsertTokenizerIndexes(ctx context.Context, pipeline
 	//fmt.Println("START")
 
 	f := func(numGo int) *types.LockedShardedMap[string, *pb.PostingList] {
+		wg := &sync.WaitGroup{}
+
 		globalMap := types.NewLockedShardedMap[string, *pb.PostingList]()
 		process := func(start int) {
 			defer wg.Done()
@@ -220,7 +220,10 @@ func (mp *MutationPipeline) InsertTokenizerIndexes(ctx context.Context, pipeline
 		return globalMap
 	}
 
-	globalMapI := f(1)
+	globalMapI := f(100)
+	mp.txn.cache.Lock()
+	defer mp.txn.cache.Unlock()
+	mp.txn.cache.globalMap[pipeline.attr] = globalMapI
 	// parallelGlobalMap := f(100)
 
 	// parallelGlobalMap.ParallelIterate(func (key string, val *pb.PostingList) error {
@@ -240,13 +243,13 @@ func (mp *MutationPipeline) InsertTokenizerIndexes(ctx context.Context, pipeline
 	// 	return nil
 	// })
 
-	globalMapI.ParallelIterate(func(key string, val *pb.PostingList) error {
-		if _, err := mp.txn.AddDelta(key, *val); err != nil {
-			pipeline.errCh <- err
-			return err
-		}
-		return nil
-	})
+	// globalMapI.ParallelIterate(func(key string, val *pb.PostingList) error {
+	// 	if _, err := mp.txn.AddDelta(key, *val); err != nil {
+	// 		pipeline.errCh <- err
+	// 		return err
+	// 	}
+	// 	return nil
+	// })
 }
 
 func (mp *MutationPipeline) ProcessList(ctx context.Context, pipeline *PredicatePipeline, index bool, reverse bool, count bool) {

posting/lists.go

@@ -75,6 +75,8 @@ type LocalCache struct {
 	// during commit.
 	deltas *types.LockedShardedMap[string, []byte]
 
+	globalMap map[string]*types.LockedShardedMap[string, *pb.PostingList]
+
 	// max committed timestamp of the read posting lists.
 	maxVersions map[string]uint64
 

posting/mvcc.go

@@ -292,6 +292,35 @@ func (txn *Txn) CommitToDisk(writer *TxnWriter, commitTs uint64) error {
 		}
 	}()
 
+	for _, i := range cache.globalMap {
+		i.Iterate(func(key string, data *pb.PostingList) error {
+			return writer.update(commitTs, func(btxn *badger.Txn) error {
+				if len(data.Postings) == 0{
+					return nil
+				}
+				dataBytes, err := proto.Marshal(data)
+				if err != nil {
+					return err
+				}
+				if ts := cache.maxVersions[key]; ts >= commitTs {
+					// Skip write because we already have a write at a higher ts.
+					// Logging here can cause a lot of output when doing Raft log replay. So, let's
+					// not output anything here.
+					return nil
+				}
+				err = btxn.SetEntry(&badger.Entry{
+					Key:      []byte(key),
+					Value:    dataBytes,
+					UserMeta: BitDeltaPosting,
+				})
+				if err != nil {
+					return err
+				}
+				return nil
+			})
+		})
+	}
+
 	var idx int
 	for idx < len(keys) {
 		// writer.update can return early from the loop in case we encounter badger.ErrTxnTooBig. On