Fix for --allow_concurrent_memtable_write with batching

Summary:
Concurrent memtable adds were incorrectly computing
the last sequence number for a write batch group when the
write batches were not solitary.  This is the cause of
https://github.com/facebook/mysql-5.6/issues/155

Test Plan:
1. unit tests
2. new unit test
3. parallel db_bench stress tests with batch size of 10 and asserts enabled

Reviewers: igor, sdong

Reviewed By: sdong

Subscribers: IslamAbdelRahman, MarkCallaghan, dhruba

Differential Revision: https://reviews.facebook.net/D53595

db/db_impl.cc

@@ -4102,7 +4102,9 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options,
 
     if (write_thread_.CompleteParallelWorker(&w)) {
       // we're responsible for early exit
-      auto last_sequence = w.parallel_group->last_writer->sequence;
+      auto last_sequence =
+          w.parallel_group->last_writer->sequence +
+          WriteBatchInternal::Count(w.parallel_group->last_writer->batch) - 1;
       SetTickerCount(stats_, SEQUENCE_NUMBER, last_sequence);
       versions_->SetLastSequence(last_sequence);
       write_thread_.EarlyExitParallelGroup(&w);
@@ -4401,7 +4403,9 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options,
             this, true /*dont_filter_deletes*/,
             true /*concurrent_memtable_writes*/);
 
-        assert(last_writer->sequence == last_sequence);
+        assert(last_writer->sequence +
+                   WriteBatchInternal::Count(last_writer->batch) - 1 ==
+               last_sequence);
         // CompleteParallelWorker returns true if this thread should
         // handle exit, false means somebody else did
         exit_completed_early = !write_thread_.CompleteParallelWorker(&w);

db/db_test_util.cc

@@ -89,7 +89,8 @@ bool DBTestBase::ChangeOptions(int skip_mask) {
         option_config_ == kUniversalCompaction ||
         option_config_ == kUniversalCompactionMultiLevel ||
         option_config_ == kUniversalSubcompactions ||
-        option_config_ == kFIFOCompaction) {
+        option_config_ == kFIFOCompaction) ||
+        option_config == kConcurrentSkipList) {
       continue;
     }
 #endif
@@ -356,6 +357,11 @@ Options DBTestBase::CurrentOptions(
       options.max_subcompactions = 4;
       break;
     }
+    case kConcurrentSkipList: {
+      options.allow_concurrent_memtable_write = true;
+      options.enable_write_thread_adaptive_yield = true;
+      break;
+    }
 
     default:
       break;

db/db_test_util.h

@@ -525,9 +525,10 @@ class DBTestBase : public testing::Test {
     kOptimizeFiltersForHits = 27,
     kRowCache = 28,
     kRecycleLogFiles = 29,
-    kLevelSubcompactions = 30,
-    kUniversalSubcompactions = 31,
-    kEnd = 30
+    kConcurrentSkipList = 30,
+    kEnd = 31,
+    kLevelSubcompactions = 31,
+    kUniversalSubcompactions = 32,
   };
   int option_config_;