From b4ce0b0412fed1554528cf8758744c1204a2885d Mon Sep 17 00:00:00 2001
From: Evgenii Stratonikov <e.stratonikov@yadro.com>
Date: Wed, 26 Apr 2023 15:06:16 +0300
Subject: [PATCH] [#447] pilorama: Do not undo log for create ops
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

```
goos: linux
goarch: amd64
cpu: 11th Gen Intel(R) Core(TM) i5-1135G7 @ 2.40GHz
         │     old      │                 new                  │
         │    sec/op    │    sec/op     vs base                │
Create-8   36.48µ ± 11%   30.34µ ± 14%  -16.84% (p=0.000 n=10)

         │     old      │                 new                  │
         │     B/op     │     B/op      vs base                │
Create-8   43.01Ki ± 4%   37.78Ki ± 5%  -12.15% (p=0.000 n=10)

         │    old     │                new                 │
         │ allocs/op  │ allocs/op   vs base                │
Create-8   166.0 ± 3%   146.0 ± 3%  -12.05% (p=0.000 n=10)
```

Signed-off-by: Evgenii Stratonikov <e.stratonikov@yadro.com>
---
 pkg/local_object_storage/pilorama/batch.go | 64 +++++++++++++++++++++-
 1 file changed, 61 insertions(+), 3 deletions(-)

diff --git a/pkg/local_object_storage/pilorama/batch.go b/pkg/local_object_storage/pilorama/batch.go
index 5722c68a..bccd640f 100644
--- a/pkg/local_object_storage/pilorama/batch.go
+++ b/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