pkg/local_object_storage/pilorama/batch.go

@@ -1,6 +1,7 @@
 package pilorama
 
 import (
+	"encoding/binary"
 	"sort"
 	"sync"
 	"time"
@@ -49,10 +50,67 @@ func (b *batch) run() {
 		sort.Slice(b.operations, func(i, j int) bool {
 			return b.operations[i].Time < b.operations[j].Time
 		})
-
 		b.operations = removeDuplicatesInPlace(b.operations)
-		var lm Move
-		return b.forest.applyOperation(bLog, bTree, b.operations, &lm)
+
+		// Our main use-case is addition of new items. In this case,
+		// we do not need to perform undo()/redo(), just do().
+		// https://github.com/trvedata/move-op/blob/6c23447c12a7862ff31b7fc2205f6c90fbdb9dc0/proof/Move_Create.thy#L259
+		//
+		// For this optimization to work we need to ensure three things:
+		// 1. The node itself is not yet in tree.
+		// 2. The node is not a parent. This case is not mentioned in the article, because
+		//    they consider a "static order" (perform all CREATE operations before MOVE).
+		//    We need this because if node _is_ a parent, we could violate (3) for some late operation.
+		//    See TestForest_ApplySameOperation for details.
+		// 3. Parent of each operation is already in tree.
+		var parents map[uint64]struct{}
+		var cKey [17]byte
+		var slow bool
+		for i := range b.operations {
+			_, _, _, inTree := b.forest.getState(bTree, stateKey(cKey[:], b.operations[i].Child))
+			if inTree {
+				slow = true
+				break
+			}
+
+			key := childrenKey(cKey[:], b.operations[i].Child, 0)
+			k, _ := bTree.Cursor().Seek(key)
+			if len(k) == 17 && binary.LittleEndian.Uint64(k[1:]) == b.operations[i].Child {
+				slow = true
+				break
+			}
+
+			if b.operations[i].Parent == RootID {
+				continue
+			} else if parents == nil {
+				// Attaching key only to root is done frequently,
+				// no allocations are performed unless necessary.
+				parents = make(map[uint64]struct{})
+			} else if _, ok := parents[b.operations[i].Parent]; ok {
+				continue
+			}
+
+			p := b.operations[i].Parent
+			_, ts, _, inTree := b.forest.getState(bTree, stateKey(cKey[:], p))
+			if !inTree || b.operations[0].Time < ts {
+				slow = true
+				break
+			}
+			parents[b.operations[i].Parent] = struct{}{}
+		}
+
+		if slow {
+			var lm Move
+			return b.forest.applyOperation(bLog, bTree, b.operations, &lm)
+		}
+
+		var key [17]byte
+		for i := range b.operations {
+			if err := b.forest.do(bLog, bTree, key[:], b.operations[i]); err != nil {
+				return err
+			}
+		}
+		return nil
 	})
 	for i := range b.results {
 		b.results[i] <- err