Atomic read/write on DeltaMergeStore level (pingcap#172)

* Atomic read/write on DeltaMergeStore level

* Bug fix

* bugfix 2

* bug fix

* fix compile error

* remove useless code

* address comments

* fix compile error

dbms/src/Common/EventRecorder.h

@@ -3,7 +3,7 @@
 #include <Common/ProfileEvents.h>
 #include <Common/Stopwatch.h>
 
-
+/// This class is NOT multi-threads safe!
 class EventRecorder
 {
 public:
@@ -12,8 +12,14 @@ class EventRecorder
         watch.start();
     }
 
+    ~EventRecorder()
+    {
+      if (!done) submit();
+    }
+
     inline void submit()
     {
+        done = true;
         ProfileEvents::increment(event);
         ProfileEvents::increment(event_elapsed, watch.elapsed());
     }
@@ -23,4 +29,5 @@ class EventRecorder
     ProfileEvents::Event event_elapsed;
 
     Stopwatch watch;
+    bool done = false;
 };

dbms/src/Common/ProfileEvents.cpp

@@ -171,8 +171,14 @@
     \
     M(DMWriteBlock) \
     M(DMWriteBlockNS) \
-    M(DMAppendDelta) \
-    M(DMAppendDeltaNS) \
+    M(DMAppendDeltaPrepare) \
+    M(DMAppendDeltaPrepareNS) \
+    M(DMAppendDeltaCommitMemory) \
+    M(DMAppendDeltaCommitMemoryNS) \
+    M(DMAppendDeltaCommitDisk) \
+    M(DMAppendDeltaCommitDiskNS) \
+    M(DMAppendDeltaCleanUp) \
+    M(DMAppendDeltaCleanUpNS) \
     M(DMPlace) \
     M(DMPlaceNS) \
     M(DMPlaceUpsert) \

dbms/src/Storages/DeltaMerge/Chunk.cpp

@@ -143,7 +143,7 @@ void deserializeColumn(IColumn & column, const ColumnMeta & meta, const Page & p
 void readChunkData(MutableColumns &      columns,
                    const Chunk &         chunk,
                    const ColumnDefines & column_defines,
-                   PageStorage &         storage,
+                   const PageReader &    page_reader,
                    size_t                rows_offset,
                    size_t                rows_limit)
 {
@@ -177,11 +177,11 @@ void readChunkData(MutableColumns &      columns,
             col.insertRangeFrom(*tmp_col, rows_offset, rows_limit);
         }
     };
-    storage.read(page_ids, page_handler);
+    page_reader.read(page_ids, page_handler);
 }
 
 
-Block readChunk(const Chunk & chunk, const ColumnDefines & read_column_defines, PageStorage & data_storage)
+Block readChunk(const Chunk & chunk, const ColumnDefines & read_column_defines, const PageReader & page_reader)
 {
     if (read_column_defines.empty())
         return {};
@@ -196,7 +196,7 @@ Block readChunk(const Chunk & chunk, const ColumnDefines & read_column_defines,
     if (chunk.getRows())
     {
         // Read from storage
-        readChunkData(columns, chunk, read_column_defines, data_storage, 0, chunk.getRows());
+        readChunkData(columns, chunk, read_column_defines, page_reader, 0, chunk.getRows());
     }
 
     Block res;

dbms/src/Storages/DeltaMerge/Chunk.h

@@ -101,6 +101,7 @@ class Chunk
     size_t        rows = 0;
 };
 
+// TODO: use list instead of vector, so that DiskValueSpace won't need to do copy during Segment#getReadSnapshot.
 using Chunks    = std::vector<Chunk>;
 using GenPageId = std::function<PageId()>;
 
@@ -116,12 +117,12 @@ Chunk prepareChunkDataWrite(const DMContext & dm_context, const GenPageId & gen_
 void readChunkData(MutableColumns &      columns,
                    const Chunk &         chunk,
                    const ColumnDefines & column_defines,
-                   PageStorage &         storage,
+                   const PageReader &    page_reader,
                    size_t                rows_offset,
                    size_t                rows_limit);
 
 
-Block readChunk(const Chunk & chunk, const ColumnDefines & read_column_defines, PageStorage & data_storage);
+Block readChunk(const Chunk & chunk, const ColumnDefines & read_column_defines, const PageReader & page_reader);
 
 
 } // namespace DM

dbms/src/Storages/DeltaMerge/ChunkBlockInputStream.h

@@ -11,8 +11,8 @@ namespace DM
 class ChunkBlockInputStream final : public IBlockInputStream
 {
 public:
-    ChunkBlockInputStream(const Chunks & chunks_, const ColumnDefines & read_columns_, PageStorage & data_storage_)
-        : chunks(chunks_), read_columns(read_columns_), data_storage(data_storage_)
+    ChunkBlockInputStream(const Chunks & chunks_, const ColumnDefines & read_columns_, const PageReader & page_reader_)
+        : chunks(chunks_), read_columns(read_columns_), page_reader(page_reader_)
     {
     }
 
@@ -36,7 +36,7 @@ class ChunkBlockInputStream final : public IBlockInputStream
     {
         if (!hasNextBlock())
             return {};
-        return readChunk(chunks[chunk_index++], read_columns, data_storage);
+        return readChunk(chunks[chunk_index++], read_columns, page_reader);
     }
 
     bool       hasNextBlock() { return chunk_index < chunks.size(); }
@@ -48,7 +48,7 @@ class ChunkBlockInputStream final : public IBlockInputStream
     Chunks        chunks;
     size_t        chunk_index = 0;
     ColumnDefines read_columns;
-    PageStorage & data_storage;
+    PageReader    page_reader;
     Block         header;
 };
 

dbms/src/Storages/DeltaMerge/DMSegmentThreadInputStream.h

@@ -12,17 +12,19 @@ class DMSegmentThreadInputStream : public IProfilingBlockInputStream
 public:
     /// If handle_real_type_ is empty, means do not convert handle column back to real type.
     DMSegmentThreadInputStream(const SegmentReadTaskPoolPtr & task_pool_,
+                               const SegmentStreamCreator &   stream_creator_,
                                const ColumnDefines &          columns_to_read_,
                                const String &                 handle_name_,
                                const DataTypePtr &            handle_real_type_,
                                const Context &                context_)
         : task_pool(task_pool_),
+          stream_creator(stream_creator_),
           columns_to_read(columns_to_read_),
           header(createHeader(columns_to_read)),
           handle_name(handle_name_),
           handle_real_type(handle_real_type_),
           context(context_),
-          log(&Logger::get("SegmentReadTaskPool"))
+          log(&Logger::get("DMSegmentThreadInputStream"))
     {
     }
 
@@ -38,12 +40,15 @@ class DMSegmentThreadInputStream : public IProfilingBlockInputStream
         {
             if (!cur_stream)
             {
-                std::tie(cur_segment_id, cur_stream) = task_pool->nextTask();
-                if (!cur_stream) // we are done.
+                auto task = task_pool->nextTask();
+                if (!task)
                 {
                     done = true;
                     return {};
                 }
+
+                cur_segment_id = task->segment->segmentId();
+                cur_stream     = stream_creator(*task);
                 LOG_DEBUG(log, "Start to read segment [" + DB::toString(cur_segment_id) + "]");
             }
 
@@ -57,7 +62,8 @@ class DMSegmentThreadInputStream : public IProfilingBlockInputStream
             }
             else
             {
-                cur_stream = {};
+                cur_segment_id = 0;
+                cur_stream     = {};
                 LOG_DEBUG(log, "Finish reading segment [" + DB::toString(cur_segment_id) + "]");
             }
         }
@@ -80,6 +86,7 @@ class DMSegmentThreadInputStream : public IProfilingBlockInputStream
 
 private:
     SegmentReadTaskPoolPtr task_pool;
+    SegmentStreamCreator   stream_creator;
     ColumnDefines          columns_to_read;
     Block                  header;
     String                 handle_name;

dbms/src/Storages/DeltaMerge/DeltaMerge.h

@@ -41,6 +41,7 @@ class DeltaMergeBlockInputStream final : public IProfilingBlockInputStream
     ssize_t stable_skip = 0;
 
     DeltaValueSpacePtr delta_value_space;
+    size_t             delta_rows_limit;
     IndexIterator      entry_it;
     IndexIterator      entry_end;
 
@@ -71,6 +72,7 @@ class DeltaMergeBlockInputStream final : public IProfilingBlockInputStream
           stable_input_stream(stable_input_stream_),
           stable_input_stream_raw_ptr(stable_input_stream.get()),
           delta_value_space(delta_value_space_),
+          delta_rows_limit(delta_value_space->getRows()),
           entry_it(index_begin),
           entry_end(index_end),
           max_block_size(max_block_size_)
@@ -159,8 +161,11 @@ class DeltaMergeBlockInputStream final : public IProfilingBlockInputStream
             }
             else if (entry_it.getType() == DT_INS)
             {
-                writeInsertFromDelta(output_columns, entry_it.getValue());
-                --output_write_limit;
+                if (entry_it.getValue() < delta_rows_limit)
+                {
+                    writeInsertFromDelta(output_columns, entry_it.getValue());
+                    --output_write_limit;
+                }
             }
             else
             {