Repository: petermattis/pebble
Pull request #218 (2019-08-12 14:58:41)

Assign distinct seqnums to ingested sstables
When two or more files are ingested atomically, assign distinct global
sequence numbers to each sstable. This avoids an error that would
previously occur when two L0 sstables shared the same global sequence
number which violated the invariant about the ordering of L0
sstables. Rather than relaxing the error, assigning distinct sequence
numbers avoids it and opens up the possibility of ingesting overlapping
sstables in the future.

petermattis: 

<!-- Reviewable:start -->
This change is [<img src="https://reviewable.io/review_button.png" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/petermattis/pebble/218)
<!-- Reviewable:end -->

petermattis: I've adjusted this PR to use a different global sequence number per ingested sstable. PTAL.

Patch:
diff --git a/commit.go b/commit.go
index 109c4a0..32fcfd5 100644
--- a/commit.go
+++ b/commit.go
@@ -266,14 +266,14 @@ func (p *commitPipeline) Commit(b *Batch, syncWAL bool) error {
 	return nil
 }
 
-// AllocateSeqNum allocates a sequence number, invokes the prepare callback,
-// then the apply callback, and then publishes the sequence
-// number. AllocateSeqNum does not write to the WAL or add entries to the
+// AllocateSeqNum allocates count sequence numbers, invokes the prepare
+// callback, then the apply callback, and then publishes the sequence
+// numbers. AllocateSeqNum does not write to the WAL or add entries to the
 // memtable. AllocateSeqNum can be used to sequence an operation such as
 // sstable ingestion within the commit pipeline. The prepare callback is
 // invoked with commitPipeline.mu held, but note that DB.mu is not held and
 // must be locked if necessary.
-func (p *commitPipeline) AllocateSeqNum(prepare func(), apply func(seqNum uint64)) {
+func (p *commitPipeline) AllocateSeqNum(count int, prepare func(), apply func(seqNum uint64)) {
 	// This method is similar to Commit and prepare. Be careful about trying to
 	// share additional code with those methods because Commit and prepare are
 	// performance critical code paths.
@@ -284,7 +284,7 @@ func (p *commitPipeline) AllocateSeqNum(prepare func(), apply func(seqNum uint64
 	// Give the batch a count of 1 so that the log and visible sequence number
 	// are incremented correctly.
 	b.storage.data = make([]byte, batchHeaderLen)
-	b.setCount(1)
+	b.setCount(uint32(count))
 	b.commit.Add(1)
 
 	p.sem <- struct{}{}
@@ -296,14 +296,16 @@ func (p *commitPipeline) AllocateSeqNum(prepare func(), apply func(seqNum uint64
 	// number order.
 	p.pending.enqueue(b)
 
-	// Assign the batch a sequence number. Note that logSeqNum is not protected
-	// by commitPipeline.mu or DB.mu and thus needs to be incremented atomically.
-	seqNum := atomic.AddUint64(p.env.logSeqNum, 1) - 1
+	// Assign the batch a sequence number. Note that we use atomic operations
+	// here to handle concurrent reads of logSeqNum. commitPipeline.mu provides
+	// mutual exclusion for other goroutines writing to logSeqNum.
+	seqNum := atomic.AddUint64(p.env.logSeqNum, uint64(count)) - uint64(count)
 	if seqNum == 0 {
 		// We can't use the value 0 for the global seqnum during ingestion, because
-		// 0 indicates no global seqnum. So allocate another seqnum.
-		seqNum = atomic.AddUint64(p.env.logSeqNum, 1) - 1
-		b.setCount(2)
+		// 0 indicates no global seqnum. So allocate one more seqnum.
+		atomic.AddUint64(p.env.logSeqNum, 1)
+		seqNum++
+		b.setCount(1 + uint32(count))
 	}
 	b.setSeqNum(seqNum)
 
@@ -346,8 +348,9 @@ func (p *commitPipeline) prepare(b *Batch, syncWAL bool) (*memTable, error) {
 	// number order.
 	p.pending.enqueue(b)
 
-	// Assign the batch a sequence number. Note that logSeqNum is not protected
-	// by commitPipeline.mu or DB.mu and thus needs to be incremented atomically.
+	// Assign the batch a sequence number. Note that we use atomic operations
+	// here to handle concurrent reads of logSeqNum. commitPipeline.mu provides
+	// mutual exclusion for other goroutines writing to logSeqNum.
 	b.setSeqNum(atomic.AddUint64(p.env.logSeqNum, n) - n)
 
 	// Write the data to the WAL.
diff --git a/ingest.go b/ingest.go
index 0388577..2f16523 100644
--- a/ingest.go
+++ b/ingest.go
@@ -199,6 +199,7 @@ func ingestUpdateSeqNum(opts *Options, dirname string, seqNum uint64, meta []*fi
 		// Properties.GlobalSeqNum when an sstable is loaded.
 		m.smallestSeqNum = seqNum
 		m.largestSeqNum = seqNum
+		seqNum++
 
 		// TODO(peter): Update the global sequence number property. This is only
 		// necessary for compatibility with RocksDB.
@@ -357,7 +358,7 @@ func (d *DB) Ingest(paths []string) error {
 		ve, err = d.ingestApply(jobID, meta)
 	}
 
-	d.commit.AllocateSeqNum(prepare, apply)
+	d.commit.AllocateSeqNum(len(meta), prepare, apply)
 
 	if err != nil {
 		if err2 := ingestCleanup(d.opts.FS, d.dirname, meta); err2 != nil {


Test patch:
diff --git a/commit_test.go b/commit_test.go
index 975ad94..5d0a71d 100644
--- a/commit_test.go
+++ b/commit_test.go
@@ -117,13 +117,13 @@ func TestCommitPipelineAllocateSeqNum(t *testing.T) {
 	wg.Add(n)
 	var prepareCount uint64
 	var applyCount uint64
-	for i := 0; i < n; i++ {
+	for i := 1; i <= n; i++ {
 		go func(i int) {
 			defer wg.Done()
-			p.AllocateSeqNum(func() {
-				atomic.AddUint64(&prepareCount, 1)
+			p.AllocateSeqNum(i, func() {
+				atomic.AddUint64(&prepareCount, uint64(1))
 			}, func(seqNum uint64) {
-				atomic.AddUint64(&applyCount, 1)
+				atomic.AddUint64(&applyCount, uint64(1))
 			})
 		}(i)
 	}
@@ -137,11 +137,12 @@ func TestCommitPipelineAllocateSeqNum(t *testing.T) {
 	}
 	// AllocateSeqNum always returns a non-zero sequence number causing the
 	// values we see to be offset from 1.
-	if s := atomic.LoadUint64(&e.logSeqNum); n+1 != s {
-		t.Fatalf("expected %d, but found %d", n+1, s)
+	const total = 1 + 1 + 2 + 3 + 4 + 5 + 6 + 7 + 8 + 9 + 10
+	if s := atomic.LoadUint64(&e.logSeqNum); total != s {
+		t.Fatalf("expected %d, but found %d", total, s)
 	}
-	if s := atomic.LoadUint64(&e.visibleSeqNum); n+1 != s {
-		t.Fatalf("expected %d, but found %d", n+1, s)
+	if s := atomic.LoadUint64(&e.visibleSeqNum); total != s {
+		t.Fatalf("expected %d, but found %d", total, s)
 	}
 }
 
diff --git a/get_iter_test.go b/get_iter_test.go
index 3b27374..cd4ac56 100644
--- a/get_iter_test.go
+++ b/get_iter_test.go
@@ -372,45 +372,73 @@ func TestGetIter(t *testing.T) {
 		},
 
 		{
-			description: "broken invariants 0: non-increasing level 0 file numbers",
+			description: "broken invariants 0: non-increasing level 0 sequence numbers",
 			badOrdering: true,
 			tables: []testTable{
 				{
 					level:   0,
-					fileNum: 20,
+					fileNum: 19,
+					data: []string{
+						"a.SET.101 a",
+						"b.SET.102 b",
+					},
 				},
 				{
 					level:   0,
-					fileNum: 19,
+					fileNum: 20,
+					data: []string{
+						"c.SET.101 c",
+					},
 				},
 			},
 		},
 
 		{
-			description: "broken invariants 1: non-increasing level 0 file numbers",
+			description: "broken invariants 1: non-increasing level 0 sequence numbers",
 			badOrdering: true,
 			tables: []testTable{
 				{
 					level:   0,
-					fileNum: 20,
+					fileNum: 19,
+					data: []string{
+						"a.SET.101 a",
+						"b.SET.102 b",
+					},
 				},
 				{
 					level:   0,
-					fileNum: 21,
+					fileNum: 20,
+					data: []string{
+						"c.SET.100 c",
+						"d.SET.101 d",
+					},
 				},
+			},
+		},
+
+		{
+			description: "broken invariants 2: non-increasing level 0 sequence numbers",
+			badOrdering: true,
+			tables: []testTable{
 				{
 					level:   0,
-					fileNum: 21,
+					fileNum: 19,
+					data: []string{
+						"a.SET.101 a",
+					},
 				},
 				{
 					level:   0,
-					fileNum: 22,
+					fileNum: 20,
+					data: []string{
+						"a.SET.101 a",
+					},
 				},
 			},
 		},
 
 		{
-			description: "broken invariants 2: level non-0 overlapping internal key ranges",
+			description: "broken invariants 3: level non-0 overlapping internal key ranges",
 			badOrdering: true,
 			tables: []testTable{
 				{
diff --git a/ingest_test.go b/ingest_test.go
index d16be1a..f98d551 100644
--- a/ingest_test.go
+++ b/ingest_test.go
@@ -463,15 +463,20 @@ func TestIngest(t *testing.T) {
 
 	datadriven.RunTest(t, "testdata/ingest", func(td *datadriven.TestData) string {
 		switch td.Cmd {
-		case "ingest", "batch":
+		case "build", "batch":
 			b := d.NewIndexedBatch()
 			if err := runBatchDefineCmd(td, b); err != nil {
 				return err.Error()
 			}
 
 			switch td.Cmd {
-			case "ingest":
-				f, err := mem.Create("ext/0")
+			case "build":
+				if len(td.CmdArgs) != 1 {
+					return "build <path>: argument missing\n"
+				}
+				path := td.CmdArgs[0].String()
+
+				f, err := mem.Create(path)
 				if err != nil {
 					return err.Error()
 				}
@@ -499,15 +504,27 @@ func TestIngest(t *testing.T) {
 					return err.Error()
 				}
 
-				if err := d.Ingest([]string{"ext/0"}); err != nil {
-					return err.Error()
-				}
-				if err := mem.Remove("ext/0"); err != nil {
+			case "batch":
+				if err := b.Commit(nil); err != nil {
 					return err.Error()
 				}
+			}
+			return ""
 
-			case "batch":
-				if err := b.Commit(nil); err != nil {
+		case "ingest":
+			if len(td.CmdArgs) == 0 {
+				return "ingest <paths>: path arguments missing\n"
+			}
+			var paths []string
+			for _, arg := range td.CmdArgs {
+				paths = append(paths, arg.String())
+			}
+
+			if err := d.Ingest(paths); err != nil {
+				return err.Error()
+			}
+			for _, path := range paths {
+				if err := mem.Remove(path); err != nil {
 					return err.Error()
 				}
 			}
diff --git a/testdata/ingest b/testdata/ingest
index 0be5140..fb2ec7e 100644
--- a/testdata/ingest
+++ b/testdata/ingest
@@ -1,8 +1,11 @@
-ingest
+build ext0
 set a 1
 set b 2
 ----
 
+ingest ext0
+----
+
 lsm
 ----
 6: a-b
@@ -23,11 +26,14 @@ b
 a:1
 b:2
 
-ingest
+build ext1
 set a 3
 del b
 ----
 
+ingest ext1
+----
+
 lsm
 ----
 5: a-b
@@ -47,12 +53,15 @@ b
 a:3
 b: pebble: not found
 
-ingest
+build ext2
 set a 4
 set b 5
 set c 6
 ----
 
+ingest ext2
+----
+
 lsm
 ----
 4: a-c
@@ -77,11 +86,14 @@ a:4
 b:5
 c:6
 
-ingest
+build ext3
 merge b 5
 del c
 ----
 
+ingest ext3
+----
+
 lsm
 ----
 3: b-c
@@ -107,11 +119,14 @@ a:4
 b:55
 c: pebble: not found
 
-ingest
+build ext4
 set x 7
 set y 8
 ----
 
+ingest ext4
+----
+
 lsm
 ----
 3: b-c
@@ -140,10 +155,13 @@ set j 9
 set k 10
 ----
 
-ingest
+build ext5
 set k 11
 ----
 
+ingest ext5
+----
+
 lsm
 ----
 0: j-k k-k
@@ -170,10 +188,13 @@ batch
 set m 12
 ----
 
-ingest
+build ext6
 set n 13
 ----
 
+ingest ext6
+----
+
 lsm
 ----
 0: j-k k-k
@@ -189,11 +210,14 @@ n
 m:12
 n:13
 
-ingest
+build ext7
 del-range a c
 del-range x z
 ----
 
+ingest ext7
+----
+
 lsm
 ----
 0: j-k k-k m-m a-z
@@ -226,12 +250,15 @@ y: pebble: not found
 # A set operation takes precedence over a range deletion at the same
 # sequence number as can occur during ingestion.
 
-ingest
+build ext8
 set j 20
 del-range j k
 set m 30
 ----
 
+ingest ext8
+----
+
 get
 j
 k
@@ -240,3 +267,35 @@ m
 j:20
 k:11
 m:30
+
+build ext9
+del-range a x
+----
+
+build ext10
+set y 40
+----
+
+# Ingesting two files into L0 is allowed.
+
+ingest ext9 ext10
+----
+
+get
+j
+k
+m
+y
+----
+j: pebble: not found
+k: pebble: not found
+m: pebble: not found
+y:40
+
+lsm
+----
+0: j-k k-k m-m a-z j-m a-x y-y
+3: b-c
+4: a-c
+5: a-b
+6: a-b n-n x-y