From aec11ff713f6fde328e5b4bec9dcba97a067c3d1 Mon Sep 17 00:00:00 2001
From: Seebs <seebs@seebs.net>
Date: Mon, 14 Sep 2020 15:40:19 -0500
Subject: [PATCH] Update cache entries when setting a row even in shards with
 no new data

The logic assumes that the lack of a corresponding rowSegment means that
there's no changes, but that's not true -- we just deleted all the
existing data! Update to match clearRow behavior better.

Also, add a corresponding test case for this.

Also, change references to 'defaultSnapshotQueue' to use
[fragment].holder.SnapshotQueue, because defaultSnapshotQueue was
the queueless queue, but holders were getting a snapshot queue,
meaning that "awaiting" a snapshot could result in moving on
and closing the holder before the actual snapshot queue finished
snapshotting.
---
 fragment.go               | 36 +++++++++++++++++++-----------------
 fragment_internal_test.go | 29 +++++++++++++++++++++--------
 2 files changed, 40 insertions(+), 25 deletions(-)

diff --git a/fragment.go b/fragment.go
index 122688ec5..33d0f3bfd 100644
--- a/fragment.go
+++ b/fragment.go
@@ -798,26 +798,28 @@ func (f *fragment) unprotectedSetRow(tx Tx, row *Row, rowID uint64) (changed boo
 
 	// From the given row, get the rowSegment for this shard.
 	seg := row.segment(f.shard)
-	if seg == nil {
-		return changed, nil
-	}
-
-	// Put each container from rowSegment to fragment storage.
-	citer, _ := seg.data.Containers.Iterator(f.shard << shardVsContainerExponent)
-	for citer.Next() {
-		k, c := citer.Value()
-		if err := tx.PutContainer(f.index, f.field, f.view, f.shard, headContainerKey+(k%(1<<shardVsContainerExponent)), c); err != nil {
-			return changed, err
+	if seg != nil {
+		// Put each container from rowSegment to fragment storage.
+		citer, _ := seg.data.Containers.Iterator(f.shard << shardVsContainerExponent)
+		for citer.Next() {
+			k, c := citer.Value()
+			if err := tx.PutContainer(f.index, f.field, f.view, f.shard, headContainerKey+(k%(1<<shardVsContainerExponent)), c); err != nil {
+				return changed, err
+			}
 		}
-	}
 
-	// Update the row in cache.
-	if f.CacheType != CacheTypeNone {
-		n, err := tx.CountRange(f.index, f.field, f.view, f.shard, rowID*ShardWidth, (rowID+1)*ShardWidth)
-		if err != nil {
-			return changed, err
+		// Update the row in cache.
+		if f.CacheType != CacheTypeNone {
+			n, err := tx.CountRange(f.index, f.field, f.view, f.shard, rowID*ShardWidth, (rowID+1)*ShardWidth)
+			if err != nil {
+				return changed, err
+			}
+			f.cache.BulkAdd(rowID, n)
+		}
+	} else {
+		if f.CacheType != CacheTypeNone {
+			f.cache.BulkAdd(rowID, 0)
 		}
-		f.cache.BulkAdd(rowID, n)
 	}
 
 	// invalidate rowCache for this row.
diff --git a/fragment_internal_test.go b/fragment_internal_test.go
index d2e2b57a2..b33862d54 100644
--- a/fragment_internal_test.go
+++ b/fragment_internal_test.go
@@ -268,6 +268,19 @@ func TestFragment_SetRow(t *testing.T) {
 	} else if n := f.mustRow(tx, rowID).Count(); n != 3 {
 		t.Fatalf("unexpected count (reopen): %d", n)
 	}
+
+	// verify that setting something from a row which lacks a segment for
+	// this fragment's shard still clears this fragment correctly.
+	notOurs := NewRow(8*ShardWidth + 1024)
+	if changed, err := f.unprotectedSetRow(tx, notOurs, rowID); err != nil {
+		t.Fatal(err)
+	} else if !changed {
+		t.Fatalf("setRow didn't report a change")
+	}
+	if cols := f.mustRow(tx, rowID).Columns(); len(cols) != 0 {
+		t.Fatalf("expected setting a row with no entries to clear the cache")
+	}
+	panicOn(tx.Commit())
 }
 
 // Ensure a fragment can set & read a value.
@@ -2853,7 +2866,7 @@ func BenchmarkImportRoaring(b *testing.B) {
 						// care whether this succeeds,
 						// but if it's happening we want
 						// it to be done.
-						_ = defaultSnapshotQueue.Await(f)
+						_ = f.holder.SnapshotQueue.Await(f)
 						f.Clean(b)
 						b.Fatalf("import error: %v", err)
 					}
@@ -2897,7 +2910,7 @@ func BenchmarkImportRoaringConcurrent(b *testing.B) {
 								err := frags[j].importRoaringT(txs[j], data[j], false)
 								// error unimportant if it happened, but we want
 								// any snapshots to have finished.
-								_ = defaultSnapshotQueue.Await(frags[j])
+								_ = frags[j].holder.SnapshotQueue.Await(frags[j])
 								return err
 							})
 						}
@@ -2942,7 +2955,7 @@ func BenchmarkImportRoaringUpdateConcurrent(b *testing.B) {
 								if err != nil {
 									b.Fatalf("importing roaring: %v", err)
 								}
-								err = defaultSnapshotQueue.Immediate(frags[j])
+								err = frags[j].holder.SnapshotQueue.Immediate(frags[j])
 								if err != nil {
 									b.Fatalf("snapshot after import: %v", err)
 								}
@@ -2955,7 +2968,7 @@ func BenchmarkImportRoaringUpdateConcurrent(b *testing.B) {
 									defer txs[j].Rollback()
 
 									err := frags[j].importRoaringT(txs[j], updata, false)
-									err2 := defaultSnapshotQueue.Await(frags[j])
+									err2 := frags[j].holder.SnapshotQueue.Await(frags[j])
 									if err == nil {
 										err = err2
 									}
@@ -3029,7 +3042,7 @@ func BenchmarkImportRoaringUpdate(b *testing.B) {
 						if err != nil {
 							b.Errorf("import error: %v", err)
 						}
-						err = defaultSnapshotQueue.Immediate(f)
+						err = f.holder.SnapshotQueue.Immediate(f)
 						if err != nil {
 							b.Errorf("snapshot after import error: %v", err)
 						}
@@ -3039,7 +3052,7 @@ func BenchmarkImportRoaringUpdate(b *testing.B) {
 							f.Clean(b)
 							b.Errorf("import error: %v", err)
 						}
-						err = defaultSnapshotQueue.Await(f)
+						err = f.holder.SnapshotQueue.Await(f)
 						if err != nil {
 							b.Errorf("snapshot after import error: %v", err)
 						}
@@ -3402,7 +3415,7 @@ func (f *fragment) Clean(t testing.TB) {
 		// badger doesn't need snapshot, so this stuff is skipped.
 		// The snapshot queue stuff doesn't work under badger.
 		if f.idx.NeedsSnapshot() {
-			err := defaultSnapshotQueue.Await(f)
+			err := f.holder.SnapshotQueue.Await(f)
 			if err != nil {
 				t.Fatalf("snapshot failed before sanity check: %v", err)
 			}
@@ -4228,7 +4241,7 @@ func TestUnionInPlaceMapped(t *testing.T) {
 	// it's used only in computation of things that usually don't go to
 	// disk, which is why we handle this specially in testing and not
 	// generically.
-	err = defaultSnapshotQueue.Immediate(f)
+	err = f.holder.SnapshotQueue.Immediate(f)
 	if err != nil {
 		t.Fatalf("snapshot after union-in-place: %v", err)
 	}