[FLASH-426] Eliminate data copy of Segment split/merge (#258)

* Use ref page to do fast segment split.
* Segment merge is done by simply merge the chunks' meta.
* Improve the logic of checking delta merge, split and merge, after DeltaMergeStore#write or delateRange.

* Optimize split/merge: update meta data only, instead of moving data

* fix get valid normal page ids

* use ref page for both side when split

* Use `std::optional<PageEntry>` instead of `PageEntry *` for PageEntries::find
* Fix bug of PageEntriesForDelta::merge

* fix bug: PageStorage GC Normal page not found

* fix bug: shrink read range while doing DeltaMerge

* Add a test case for multiple split and DeltaMerge

* fix data error after ref split

Author: flowbehappy

dbms/src/Storages/DeltaMerge/Chunk.cpp

@@ -9,6 +9,7 @@ namespace DB
 {
 namespace DM
 {
+
 void Chunk::serialize(WriteBuffer & buf) const
 {
     writeIntBinary(handle_start, buf);
@@ -71,6 +72,39 @@ Chunk Chunk::deserialize(ReadBuffer & buf)
     return chunk;
 }
 
+Chunk createRefChunk(const Chunk & chunk, const GenPageId & gen_data_page_id, WriteBatch & wb)
+{
+    if (chunk.isDeleteRange())
+        return Chunk(chunk.getDeleteRange());
+
+    auto [handle_first, handle_end] = chunk.getHandleFirstLast();
+    Chunk ref_chunk(handle_first, handle_end);
+    for (auto && [col_id, col_meta] : chunk.getMetas())
+    {
+        ColumnMeta m;
+
+        m.col_id  = col_id;
+        m.page_id = gen_data_page_id();
+        m.rows    = col_meta.rows;
+        m.bytes   = col_meta.bytes;
+        m.type    = col_meta.type;
+        m.minmax  = col_meta.minmax;
+
+        wb.putRefPage(m.page_id, col_meta.page_id);
+        ref_chunk.insert(m);
+    }
+    return ref_chunk;
+}
+
+Chunks createRefChunks(const Chunks & chunks, const GenPageId & gen_data_page_id, WriteBatch & wb)
+{
+    Chunks ref_chunks;
+    ref_chunks.reserve(chunks.size());
+    for (auto & chunk : chunks)
+        ref_chunks.push_back(createRefChunk(chunk, gen_data_page_id, wb));
+    return ref_chunks;
+}
+
 void serializeChunks(
     WriteBuffer & buf, Chunks::const_iterator begin, Chunks ::const_iterator end, const Chunk * extra1, const Chunk * extra2)
 {
@@ -241,7 +275,7 @@ void readChunkData(MutableColumns &      columns,
                                 ErrorCodes::NOT_IMPLEMENTED);
             }
 
-            // Read from disk according as chunk meta
+            // Read from disk according to chunk meta
             MutableColumnPtr disk_col = disk_meta.type->createColumn();
             deserializeColumn(*disk_col, disk_meta, page, rows_offset + rows_limit);
 

dbms/src/Storages/DeltaMerge/Chunk.h

@@ -115,6 +115,9 @@ class Chunk
 using Chunks    = std::vector<Chunk>;
 using GenPageId = std::function<PageId()>;
 
+Chunk  createRefChunk(const Chunk & chunk, const GenPageId & gen_data_page_id, WriteBatch & wb);
+Chunks createRefChunks(const Chunks & chunks, const GenPageId & gen_data_page_id, WriteBatch & wb);
+
 void   serializeChunks(WriteBuffer &           buf,
                        Chunks::const_iterator  begin,
                        Chunks ::const_iterator end,

dbms/src/Storages/DeltaMerge/ChunkBlockInputStream.h

@@ -4,35 +4,46 @@
 
 #include <Storages/DeltaMerge/Chunk.h>
 #include <Storages/DeltaMerge/Filter/RSOperator.h>
+#include <Storages/DeltaMerge/HandleFilter.h>
 
 namespace DB
 {
 namespace DM
 {
-/// Read `chunks` as blocks according to `read_columns`
+/// Read `chunks` as blocks.
+/// We can use `handle_range` param to filter out rows, and use `filter` to ignore some chunks roughly.
+///
+/// Note that `handle_range` param assumes that data in chunks are in order of handle. If not, please use handle range of {MIN, MAX}.
+///
+/// For example:
+///     size_t skip_rows = 0;
+///     while(stream.hasNext())
+///     {
+///         if(stream.shouldSkipNext())
+///         {
+///             skip_rows += stream.nextRows();
+///             stream.skipNext();
+///             continue;
+///         }
+///         auto block = stream.read();
+///         ...
+///     }
 class ChunkBlockInputStream final : public IBlockInputStream
 {
 public:
     ChunkBlockInputStream(const Chunks &        chunks_,
-                          const RSOperatorPtr & filter,
+                          size_t                handle_col_pos_,
+                          const HandleRange &   handle_range_,
                           const ColumnDefines & read_columns_,
-                          const PageReader &    page_reader_)
-        : chunks(chunks_), skip_chunks(chunks.size()), read_columns(read_columns_), page_reader(page_reader_)
+                          const PageReader &    page_reader_,
+                          const RSOperatorPtr & filter_)
+        : chunks(chunks_),
+          handle_col_pos(handle_col_pos_),
+          handle_range(handle_range_),
+          read_columns(read_columns_),
+          page_reader(page_reader_),
+          filter(filter_)
     {
-        for (size_t i = 0; i < chunks.size(); ++i)
-        {
-            if (!filter)
-            {
-                skip_chunks[i] = 0;
-                continue;
-            }
-            auto &       chunk = chunks[i];
-            RSCheckParam param;
-            for (auto & [col_id, meta] : chunk.getMetas())
-                param.indexes.emplace(col_id, RSIndex(meta.type, meta.minmax));
-
-            skip_chunks[i] = filter->roughCheck(param) == None;
-        }
     }
 
     String getName() const override { return "Chunk"; }
@@ -42,22 +53,120 @@ class ChunkBlockInputStream final : public IBlockInputStream
     {
         if (!hasNext())
             return {};
-        return readChunk(chunks[chunk_index++], read_columns, page_reader);
+        Block tmp;
+        if (!cur_chunk_data)
+            // It means user ignore the skipNext() result and insist to read data.
+            tmp = readCurChunkData();
+        else
+            tmp.swap(cur_chunk_data);
+
+        ++cur_chunk_index;
+        cur_chunk_skip = false;
+
+        return tmp;
+    }
+
+    bool hasNext()
+    {
+        if (cur_chunk_index >= chunks.size())
+            return false;
+        // Filter out those rows not fit for handle_range.
+        for (; cur_chunk_index < chunks.size(); ++cur_chunk_index)
+        {
+            auto [first, last] = chunks[cur_chunk_index].getHandleFirstLast();
+            if (handle_range.intersect(first, last))
+                break;
+        }
+
+        if (cur_chunk_index >= chunks.size())
+            return false;
+
+        if (!cur_chunk_data)
+        {
+            if (filter)
+            {
+                auto &       chunk = chunks[cur_chunk_index];
+                RSCheckParam param;
+                for (auto & [col_id, meta] : chunk.getMetas())
+                    param.indexes.emplace(col_id, RSIndex(meta.type, meta.minmax));
+
+                cur_chunk_skip = filter->roughCheck(param) == None;
+            }
+            if (!cur_chunk_skip)
+            {
+                cur_chunk_data = readCurChunkData();
+            }
+        }
+
+        return true;
     }
 
-    bool   hasNext() { return chunk_index < chunks.size(); }
-    size_t nextRows() { return chunks[chunk_index].getRows(); }
-    bool   shouldSkipNext() { return skip_chunks[chunk_index]; }
-    void   skipNext() { ++chunk_index; }
+    size_t nextRows()
+    {
+        auto & chunk = chunks[cur_chunk_index];
+        if (isCurChunkCompleted(chunk))
+            return chunk.getRows();
+
+        // Otherwise, some rows of current chunk are filtered out by handle_range.
+
+        if (cur_chunk_data)
+        {
+            return cur_chunk_data.rows();
+        }
+        else
+        {
+            // Current chunk is ignored by `filter`,
+            // but we still need to get the row count which their handles are included by handle_range.
+            auto block = readChunk(chunk, {read_columns[handle_col_pos]}, page_reader);
+            auto offset_limit
+                = HandleFilter::getPosRangeOfSorted(handle_range, block.getByPosition(handle_col_pos).column, 0, block.rows());
+            return offset_limit.second;
+        }
+    }
+
+    bool shouldSkipNext() { return cur_chunk_skip; }
+
+    void skipNext()
+    {
+        ++cur_chunk_index;
+
+        cur_chunk_data = {};
+        cur_chunk_skip = false;
+    }
 
 private:
-    Chunks             chunks;
-    std::vector<UInt8> skip_chunks;
+    inline bool isCurChunkCompleted(const Chunk & chunk)
+    {
+        auto [first, last] = chunk.getHandleFirstLast();
+        return handle_range.include(first, last);
+    }
+
+    inline Block readCurChunkData()
+    {
+        auto & chunk = chunks[cur_chunk_index];
+        if (isCurChunkCompleted(chunk))
+        {
+            return readChunk(chunk, read_columns, page_reader);
+        }
+        else
+        {
+            auto block = readChunk(chunk, read_columns, page_reader);
+            return HandleFilter::filterSorted(handle_range, std::move(block), handle_col_pos);
+        }
+    }
+
+private:
+    Chunks      chunks;
+    size_t      handle_col_pos;
+    HandleRange handle_range;
 
-    size_t        chunk_index = 0;
     ColumnDefines read_columns;
     PageReader    page_reader;
-    Block         header;
+    RSOperatorPtr filter;
+
+    size_t cur_chunk_index = 0;
+    bool   cur_chunk_skip  = false;
+    Block  cur_chunk_data;
 };
 
 using ChunkBlockInputStreamPtr = std::shared_ptr<ChunkBlockInputStream>;

dbms/src/Storages/DeltaMerge/DMVersionFilterBlockInputStream.h

@@ -11,7 +11,8 @@ namespace DM
 
 /// Use the latest rows. For rows with the same handle, only take the rows with biggest version and version <= version_limit.
 static constexpr int DM_VERSION_FILTER_MODE_MVCC = 0;
-/// Remove the outdated rows. For rows with the same handle, take all rows with version >= version_limit. And if all of them are smaller than version_limit, then take the biggest one, if it is not deleted.
+/// Remove the outdated rows. For rows with the same handle, take all rows with version >= version_limit.
+/// And if all of them are smaller than version_limit, then take the biggest one, if it is not deleted.
 static constexpr int DM_VERSION_FILTER_MODE_COMPACT = 1;
 
 template <int MODE>

dbms/src/Storages/DeltaMerge/DeltaMergeDefines.h

@@ -131,5 +131,6 @@ static_assert(static_cast<Int64>(static_cast<UInt64>(MAX_INT64)) == MAX_INT64, "
 
 static constexpr UInt64 DEL_RANGE_POS_MARK = (1ULL << 63);
 
+
 } // namespace DM
 } // namespace DB

dbms/src/Storages/DeltaMerge/DeltaMergeHelpers.h

@@ -13,8 +13,8 @@
 #include <Interpreters/sortBlock.h>
 #include <Storages/ColumnsDescription.h>
 #include <Storages/DeltaMerge/DeltaMergeDefines.h>
-#include <Storages/Transaction/TiDB.h>
 #include <Storages/DeltaMerge/Filter/RSOperator.h>
+#include <Storages/Transaction/TiDB.h>
 
 namespace DB
 {
@@ -123,7 +123,7 @@ inline PaddedPODArray<T> const * getColumnVectorDataPtr(const Block & block, siz
     return toColumnVectorDataPtr<T>(block.getByPosition(pos).column);
 }
 
-inline void addColumnToBlock(Block & block, ColId col_id, const String &col_name, const DataTypePtr & col_type, const ColumnPtr & col)
+inline void addColumnToBlock(Block & block, ColId col_id, const String & col_name, const DataTypePtr & col_type, const ColumnPtr & col)
 {
     ColumnWithTypeAndName column(col, col_type, col_name, col_id);
     block.insert(std::move(column));
@@ -209,15 +209,21 @@ inline void appendIntoHandleColumn(ColumnVector<Handle>::Container & handle_colu
         auto & data_vector = typeid_cast<const ColumnVector<Int32> &>(*data).getData();
         APPEND(32, 0xFFFFFFFF, data_vector)
     }
-    else if (checkDataType<DataTypeInt64>(type_ptr) || checkDataType<DataTypeDateTime>(type_ptr))
+    else if (checkDataType<DataTypeInt64>(type_ptr))
     {
         auto & data_vector = typeid_cast<const ColumnVector<Int64> &>(*data).getData();
         for (size_t i = 0; i < size; ++i)
             handle_column[i] |= data_vector[i];
     }
+    else if (checkDataType<DataTypeDateTime>(type_ptr))
+    {
+        auto & data_vector = typeid_cast<const ColumnVector<typename DataTypeDateTime::FieldType> &>(*data).getData();
+        for (size_t i = 0; i < size; ++i)
+            handle_column[i] |= data_vector[i];
+    }
     else if (checkDataType<DataTypeDate>(type_ptr))
     {
-        auto & data_vector = typeid_cast<const ColumnVector<UInt32> &>(*data).getData();
+        auto & data_vector = typeid_cast<const ColumnVector<typename DataTypeDate::FieldType> &>(*data).getData();
         APPEND(32, 0xFFFFFFFF, data_vector)
     }
     else