Improve Transaction: deduplicate quads, allow adding/removing a quad in the same tx

graph/transaction.go

@@ -14,30 +14,52 @@
 
 package graph
 
-import (
+import "github.com/google/cayley/quad"
-	"github.com/google/cayley/quad"
-)
 
+// Transaction stores a bunch of Deltas to apply atomatically on the database
 type Transaction struct {
-	Deltas []Delta
+	Deltas map[Delta]struct{}
 }
 
+// NewTransaction initialize a new transaction
 func NewTransaction() *Transaction {
-	return &Transaction{make([]Delta, 0, 5)}
+	return &Transaction{Deltas: make(map[Delta]struct{}, 100)}
 }
 
+// AddQuad adds a new quad to the transaction if it is not already present in it
+// If there is a 'remove' delta for that quad, it will remove that delta from
+// the transaction instead of actually addind the quad
 func (t *Transaction) AddQuad(q quad.Quad) {
-	t.Deltas = append(t.Deltas,
+	ad := Delta{
-		Delta{
+		Quad:   q,
-			Quad:   q,
+		Action: Add,
-			Action: Add,
+	}
-		})
+	rd := Delta{
+		Quad:   q,
+		Action: Delete,
+	}
+
+	if _, adExists := t.Deltas[ad]; !adExists {
+		if _, rdExists := t.Deltas[rd]; rdExists {
+			delete(t.Deltas, rd)
+		} else {
+			t.Deltas[ad] = struct{}{}
+		}
+	}
 }
 
+// RemoveQuad adds a quad to remove to the transaction
+// The quad will be removed from the database if it is not present in the
+// transaction, otherwise it simply remove it from the transaction
 func (t *Transaction) RemoveQuad(q quad.Quad) {
-	t.Deltas = append(t.Deltas,
+	ad := Delta{
-		Delta{
+		Quad:   q,
-			Quad:   q,
+		Action: Add,
-			Action: Delete,
+	}
-		})
+
+	if _, adExists := t.Deltas[ad]; adExists {
+		delete(t.Deltas, ad)
+	} else {
+		t.Deltas[Delta{Quad: q, Action: Delete}] = struct{}{}
+	}
 }

graph/transaction_test.go

@@ -0,0 +1,62 @@
+package graph
+
+import (
+	"testing"
+
+	"github.com/google/cayley/quad"
+)
+
+func TestTransaction(t *testing.T) {
+	var tx *Transaction
+
+	// simples adds / removes
+	tx = NewTransaction()
+
+	tx.AddQuad(quad.Quad{Subject: "E", Predicate: "follows", Object: "F", Label: ""})
+	tx.AddQuad(quad.Quad{Subject: "F", Predicate: "follows", Object: "G", Label: ""})
+	tx.RemoveQuad(quad.Quad{Subject: "A", Predicate: "follows", Object: "Z", Label: ""})
+	if len(tx.Deltas) != 3 {
+		t.Errorf("Expected 3 Deltas, have %d delta(s)", len(tx.Deltas))
+	}
+
+	// add, remove -> nothing
+	tx = NewTransaction()
+	tx.AddQuad(quad.Quad{Subject: "E", Predicate: "follows", Object: "G", Label: ""})
+	tx.RemoveQuad(quad.Quad{Subject: "E", Predicate: "follows", Object: "G", Label: ""})
+	if len(tx.Deltas) != 0 {
+		t.Errorf("Expected [add, remove]->[], have %d Deltas", len(tx.Deltas))
+	}
+
+	// remove, add -> nothing
+	tx = NewTransaction()
+	tx.AddQuad(quad.Quad{Subject: "E", Predicate: "follows", Object: "G", Label: ""})
+	tx.RemoveQuad(quad.Quad{Subject: "E", Predicate: "follows", Object: "G", Label: ""})
+	if len(tx.Deltas) != 0 {
+		t.Errorf("Expected [add, remove]->[], have %d delta(s)", len(tx.Deltas))
+	}
+
+	// add x2 -> add x1
+	tx = NewTransaction()
+	tx.AddQuad(quad.Quad{Subject: "E", Predicate: "follows", Object: "G", Label: ""})
+	tx.AddQuad(quad.Quad{Subject: "E", Predicate: "follows", Object: "G", Label: ""})
+	if len(tx.Deltas) != 1 {
+		t.Errorf("Expected [add, add]->[add], have %d delta(s)", len(tx.Deltas))
+	}
+
+	// remove x2 -> remove x1
+	tx = NewTransaction()
+	tx.RemoveQuad(quad.Quad{Subject: "E", Predicate: "follows", Object: "G", Label: ""})
+	tx.RemoveQuad(quad.Quad{Subject: "E", Predicate: "follows", Object: "G", Label: ""})
+	if len(tx.Deltas) != 1 {
+		t.Errorf("Expected [remove, remove]->[remove], have %d delta(s)", len(tx.Deltas))
+	}
+
+	// add, remove x2 -> remove x1
+	tx = NewTransaction()
+	tx.AddQuad(quad.Quad{Subject: "E", Predicate: "follows", Object: "G", Label: ""})
+	tx.RemoveQuad(quad.Quad{Subject: "E", Predicate: "follows", Object: "G", Label: ""})
+	tx.RemoveQuad(quad.Quad{Subject: "E", Predicate: "follows", Object: "G", Label: ""})
+	if len(tx.Deltas) != 1 {
+		t.Errorf("Expected [add, remove, remove]->[remove], have %d delta(s)", len(tx.Deltas))
+	}
+}

writer/single.go

@@ -109,9 +109,11 @@ func (s *Single) Close() error {
 
 func (s *Single) ApplyTransaction(t *graph.Transaction) error {
 	ts := time.Now()
-	for i := 0; i < len(t.Deltas); i++ {
+	deltas := make([]graph.Delta, 0, len(t.Deltas))
-		t.Deltas[i].ID = s.currentID.Next()
+	for d := range t.Deltas {
-		t.Deltas[i].Timestamp = ts
+		d.ID = s.currentID.Next()
+		d.Timestamp = ts
+		deltas = append(deltas, d)
 	}
-	return s.qs.ApplyDeltas(t.Deltas, s.ignoreOpts)
+	return s.qs.ApplyDeltas(deltas, s.ignoreOpts)
 }