Separate storage delta layer to multiple layers

dbms/src/Interpreters/Settings.h

@@ -253,8 +253,10 @@ struct Settings
     M(SettingUInt64, dt_segment_stop_write_delta_size, 2147483648, "Delta size before stop new writes. 2 GB by default.")                                                                                                               \
     M(SettingUInt64, dt_segment_delta_cache_limit_rows, 4096, "Max rows of cache in segment delta in DeltaTree Engine.")                                                                                                                \
     M(SettingUInt64, dt_segment_delta_cache_limit_size, 4194304, "Max size of cache in segment delta in DeltaTree Engine. 4 MB by default.")                                                                                            \
-    M(SettingUInt64, dt_segment_delta_small_pack_rows, 2048, "Determine whether a pack in delta is small or not.")                                                                                                                       \
-    M(SettingUInt64, dt_segment_delta_small_pack_size, 8388608, "Determine whether a pack in delta is small or not. 8MB by default.")                                                                                                   \
+    M(SettingUInt64, dt_segment_delta_small_pack_rows, 2048, "Deprecated. Reserved for backward compatibility. Use dt_segment_delta_small_column_file_rows instead")                                                                    \
+    M(SettingUInt64, dt_segment_delta_small_pack_size, 8388608, "Deprecated. Reserved for backward compatibility. Use dt_segment_delta_small_column_file_size instead")                                                                 \
+    M(SettingUInt64, dt_segment_delta_small_column_file_rows, 2048, "Determine whether a column file in delta is small or not. 8MB by default.")                                                                                        \
+    M(SettingUInt64, dt_segment_delta_small_column_file_size, 8388608, "Determine whether a column file in delta is small or not. 8MB by default.")                                                                                     \
     M(SettingUInt64, dt_segment_stable_pack_rows, DEFAULT_MERGE_BLOCK_SIZE, "Expected stable pack rows in DeltaTree Engine.")                                                                                                           \
     M(SettingFloat, dt_segment_wait_duration_factor, 1, "The factor of wait duration in a write stall.")                                                                                                                                \
     M(SettingUInt64, dt_bg_gc_check_interval, 600, "Background gc thread check interval, the unit is second.")                                                                                                                          \

dbms/src/Server/tests/gtest_server_config.cpp

@@ -175,8 +175,8 @@ dt_storage_pool_data_gc_max_valid_rate = 0.5
         ASSERT_EQ(global_ctx.getSettingsRef().dt_storage_pool_data_gc_min_file_num, 8);
         ASSERT_EQ(global_ctx.getSettingsRef().dt_storage_pool_data_gc_min_legacy_num, 2);
         ASSERT_EQ(global_ctx.getSettingsRef().dt_storage_pool_data_gc_min_bytes, 256);
-        ASSERT_EQ(global_ctx.getSettingsRef().dt_segment_delta_small_pack_size, 8388608);
-        ASSERT_EQ(global_ctx.getSettingsRef().dt_segment_delta_small_pack_rows, 2048);
+        ASSERT_EQ(global_ctx.getSettingsRef().dt_segment_delta_small_column_file_size, 8388608);
+        ASSERT_EQ(global_ctx.getSettingsRef().dt_segment_delta_small_column_file_rows, 2048);
         ASSERT_EQ(global_ctx.getSettingsRef().dt_segment_limit_size, 536870912);
         ASSERT_EQ(global_ctx.getSettingsRef().dt_segment_delta_limit_size, 42991616);
         ASSERT_EQ(global_ctx.getSettingsRef().dt_segment_force_merge_delta_size, 1073741824);

dbms/src/Storages/DeltaMerge/ColumnFile/ColumnFile.cpp

@@ -3,6 +3,7 @@
 #include <Storages/DeltaMerge/ColumnFile/ColumnFileBig.h>
 #include <Storages/DeltaMerge/ColumnFile/ColumnFileDeleteRange.h>
 #include <Storages/DeltaMerge/ColumnFile/ColumnFileInMemory.h>
+#include <Storages/DeltaMerge/ColumnFile/ColumnFilePersisted.h>
 #include <Storages/DeltaMerge/ColumnFile/ColumnFileTiny.h>
 #include <Storages/DeltaMerge/RowKeyFilter.h>
 
@@ -82,20 +83,20 @@ ColumnFileBig * ColumnFile::tryToBigFile()
     return !isBigFile() ? nullptr : static_cast<ColumnFileBig *>(this);
 }
 
-String columnFilesToString(const ColumnFiles & column_files)
+template <class T>
+String columnFilesToString(const T & column_files)
 {
     String column_files_info = "[";
     for (const auto & f : column_files)
     {
         if (f->isInMemoryFile())
-            column_files_info += "B_" + DB::toString(f->getRows());
+            column_files_info += "M_" + DB::toString(f->getRows()) + ",";
         else if (f->isTinyFile())
-            column_files_info += "B_" + DB::toString(f->getRows());
+            column_files_info += "T_" + DB::toString(f->getRows()) + ",";
         else if (f->isBigFile())
-            column_files_info += "F_" + DB::toString(f->getRows());
+            column_files_info += "F_" + DB::toString(f->getRows()) + ",";
         else if (auto * f_delete = f->tryToDeleteRange(); f_delete)
-            column_files_info += "D_" + f_delete->getDeleteRange().toString();
-        column_files_info += (f->isSaved() ? "_S," : "_N,");
+            column_files_info += "D_" + f_delete->getDeleteRange().toString() + ",";
     }
 
     if (!column_files.empty())
@@ -104,5 +105,8 @@ String columnFilesToString(const ColumnFiles & column_files)
     return column_files_info;
 }
 
+template String columnFilesToString<ColumnFiles>(const ColumnFiles & column_files);
+template String columnFilesToString<ColumnFilePersisteds>(const ColumnFilePersisteds & column_files);
+
 } // namespace DM
 } // namespace DB

dbms/src/Storages/DeltaMerge/ColumnFile/ColumnFile.h

@@ -75,10 +75,6 @@ class ColumnFile
 public:
     /// This id is only used to to do equal check in DeltaValueSpace::checkHeadAndCloneTail.
     UInt64 getId() const { return id; }
-    /// This column file is already saved to disk or not. Only saved packs can be recovered after reboot.
-    /// "saved" can only be true, after the content data and the metadata are all written to disk.
-    bool isSaved() const { return saved; }
-    void setSaved() { saved = true; }
 
     virtual size_t getRows() const { return 0; }
     virtual size_t getBytes() const { return 0; };
@@ -111,12 +107,6 @@ class ColumnFile
         throw Exception("Unsupported operation", ErrorCodes::LOGICAL_ERROR);
     }
 
-    /// Put the data's page id into the corresponding WriteBatch.
-    /// The actual remove will be done later.
-    virtual void removeData(WriteBatches &) const {};
-
-    virtual void serializeMetadata(WriteBuffer & buf, bool save_schema) const = 0;
-
     virtual String toString() const = 0;
 };
 
@@ -152,6 +142,7 @@ size_t copyColumnsData(
 
 
 /// Debugging string
-String columnFilesToString(const ColumnFiles & column_files);
+template <typename T>
+String columnFilesToString(const T & column_files);
 } // namespace DM
 } // namespace DB

dbms/src/Storages/DeltaMerge/ColumnFile/ColumnFileBig.cpp

@@ -30,7 +30,7 @@ void ColumnFileBig::calculateStat(const DMContext & context)
 ColumnFileReaderPtr
 ColumnFileBig::getReader(const DMContext & context, const StorageSnapshotPtr & /*storage_snap*/, const ColumnDefinesPtr & col_defs) const
 {
-    return std::make_shared<ColumnBigFileReader>(context, *this, col_defs);
+    return std::make_shared<ColumnFileBigReader>(context, *this, col_defs);
 }
 
 void ColumnFileBig::serializeMetadata(WriteBuffer & buf, bool /*save_schema*/) const
@@ -40,9 +40,9 @@ void ColumnFileBig::serializeMetadata(WriteBuffer & buf, bool /*save_schema*/) c
     writeIntBinary(valid_bytes, buf);
 }
 
-ColumnFilePtr ColumnFileBig::deserializeMetadata(DMContext & context, //
-                                                 const RowKeyRange & segment_range,
-                                                 ReadBuffer & buf)
+ColumnFilePersistedPtr ColumnFileBig::deserializeMetadata(DMContext & context, //
+                                                          const RowKeyRange & segment_range,
+                                                          ReadBuffer & buf)
 {
     UInt64 file_ref_id;
     size_t valid_rows, valid_bytes;
@@ -56,11 +56,11 @@ ColumnFilePtr ColumnFileBig::deserializeMetadata(DMContext & context, //
 
     auto dmfile = DMFile::restore(context.db_context.getFileProvider(), file_id, file_ref_id, file_parent_path, DMFile::ReadMetaMode::all());
 
-    auto dp_file = new ColumnFileBig(dmfile, valid_rows, valid_bytes, segment_range);
+    auto * dp_file = new ColumnFileBig(dmfile, valid_rows, valid_bytes, segment_range);
     return std::shared_ptr<ColumnFileBig>(dp_file);
 }
 
-void ColumnBigFileReader::initStream()
+void ColumnFileBigReader::initStream()
 {
     if (file_stream)
         return;
@@ -94,7 +94,7 @@ void ColumnBigFileReader::initStream()
     }
 }
 
-size_t ColumnBigFileReader::readRowsRepeatedly(MutableColumns & output_cols, size_t rows_offset, size_t rows_limit, const RowKeyRange * range)
+size_t ColumnFileBigReader::readRowsRepeatedly(MutableColumns & output_cols, size_t rows_offset, size_t rows_limit, const RowKeyRange * range)
 {
     if (unlikely(rows_offset + rows_limit > column_file.valid_rows))
         throw Exception("Try to read more rows", ErrorCodes::LOGICAL_ERROR);
@@ -126,13 +126,13 @@ size_t ColumnBigFileReader::readRowsRepeatedly(MutableColumns & output_cols, siz
     return actual_read;
 }
 
-size_t ColumnBigFileReader::readRowsOnce(MutableColumns & output_cols, //
+size_t ColumnFileBigReader::readRowsOnce(MutableColumns & output_cols, //
                                          size_t rows_offset,
                                          size_t rows_limit,
                                          const RowKeyRange * range)
 {
     auto read_next_block = [&, this]() -> bool {
-        rows_before_cur_block += ((bool)cur_block) ? cur_block.rows() : 0;
+        rows_before_cur_block += (static_cast<bool>(cur_block)) ? cur_block.rows() : 0;
         cur_block_data.clear();
 
         cur_block = file_stream->read();
@@ -186,7 +186,7 @@ size_t ColumnBigFileReader::readRowsOnce(MutableColumns & output_cols, //
     return actual_read;
 }
 
-size_t ColumnBigFileReader::readRows(MutableColumns & output_cols, size_t rows_offset, size_t rows_limit, const RowKeyRange * range)
+size_t ColumnFileBigReader::readRows(MutableColumns & output_cols, size_t rows_offset, size_t rows_limit, const RowKeyRange * range)
 {
     initStream();
 
@@ -204,17 +204,17 @@ size_t ColumnBigFileReader::readRows(MutableColumns & output_cols, size_t rows_o
     }
 }
 
-Block ColumnBigFileReader::readNextBlock()
+Block ColumnFileBigReader::readNextBlock()
 {
     initStream();
 
     return file_stream->read();
 }
 
-ColumnFileReaderPtr ColumnBigFileReader::createNewReader(const ColumnDefinesPtr & new_col_defs)
+ColumnFileReaderPtr ColumnFileBigReader::createNewReader(const ColumnDefinesPtr & new_col_defs)
 {
     // Currently we don't reuse the cache data.
-    return std::make_shared<ColumnBigFileReader>(context, column_file, new_col_defs);
+    return std::make_shared<ColumnFileBigReader>(context, column_file, new_col_defs);
 }
 
 } // namespace DM

dbms/src/Storages/DeltaMerge/ColumnFile/ColumnFileBig.h

@@ -15,7 +15,7 @@ using ColumnBigFilePtr = std::shared_ptr<ColumnFileBig>;
 /// A column file which contains a DMFile. The DMFile could have many Blocks.
 class ColumnFileBig : public ColumnFilePersisted
 {
-    friend class ColumnBigFileReader;
+    friend class ColumnFileBigReader;
 
 private:
     DMFilePtr file;
@@ -71,20 +71,19 @@ class ColumnFileBig : public ColumnFilePersisted
 
     void serializeMetadata(WriteBuffer & buf, bool save_schema) const override;
 
-    static ColumnFilePtr deserializeMetadata(DMContext & context, //
-                                             const RowKeyRange & segment_range,
-                                             ReadBuffer & buf);
+    static ColumnFilePersistedPtr deserializeMetadata(DMContext & context, //
+                                                      const RowKeyRange & segment_range,
+                                                      ReadBuffer & buf);
 
     String toString() const override
     {
         String s = "{big_file,rows:" + DB::toString(getRows()) //
             + ",bytes:" + DB::toString(getBytes()) + "}"; //
-        +",saved:" + DB::toString(saved) + "}"; //
         return s;
     }
 };
 
-class ColumnBigFileReader : public ColumnFileReader
+class ColumnFileBigReader : public ColumnFileReader
 {
 private:
     const DMContext & context;
@@ -113,7 +112,7 @@ class ColumnBigFileReader : public ColumnFileReader
     size_t readRowsOnce(MutableColumns & output_cols, size_t rows_offset, size_t rows_limit, const RowKeyRange * range);
 
 public:
-    ColumnBigFileReader(const DMContext & context_, const ColumnFileBig & column_file_, const ColumnDefinesPtr & col_defs_)
+    ColumnFileBigReader(const DMContext & context_, const ColumnFileBig & column_file_, const ColumnDefinesPtr & col_defs_)
         : context(context_)
         , column_file(column_file_)
         , col_defs(col_defs_)

dbms/src/Storages/DeltaMerge/ColumnFile/ColumnFileDeleteRange.cpp

@@ -18,7 +18,7 @@ void ColumnFileDeleteRange::serializeMetadata(WriteBuffer & buf, bool /*save_sch
     delete_range.serialize(buf);
 }
 
-ColumnFilePtr ColumnFileDeleteRange::deserializeMetadata(ReadBuffer & buf)
+ColumnFilePersistedPtr ColumnFileDeleteRange::deserializeMetadata(ReadBuffer & buf)
 {
     return std::make_shared<ColumnFileDeleteRange>(RowKeyRange::deserialize(buf));
 }

dbms/src/Storages/DeltaMerge/ColumnFile/ColumnFileDeleteRange.h

@@ -42,9 +42,9 @@ class ColumnFileDeleteRange : public ColumnFilePersisted
 
     void serializeMetadata(WriteBuffer & buf, bool save_schema) const override;
 
-    static ColumnFilePtr deserializeMetadata(ReadBuffer & buf);
+    static ColumnFilePersistedPtr deserializeMetadata(ReadBuffer & buf);
 
-    String toString() const override { return "{delete_range:" + delete_range.toString() + ", saved: " + DB::toString(saved) + "}"; }
+    String toString() const override { return "{delete_range:" + delete_range.toString() + "}"; }
 };
 
 class ColumnFileEmptyReader : public ColumnFileReader

dbms/src/Storages/DeltaMerge/ColumnFile/ColumnFileInMemory.cpp

@@ -44,7 +44,7 @@ void ColumnFileInMemory::fillColumns(const ColumnDefines & col_defs, size_t col_
 ColumnFileReaderPtr
 ColumnFileInMemory::getReader(const DMContext & /*context*/, const StorageSnapshotPtr & /*storage_snap*/, const ColumnDefinesPtr & col_defs) const
 {
-    return std::make_shared<ColumnInMemoryFileReader>(*this, col_defs);
+    return std::make_shared<ColumnFileInMemoryReader>(*this, col_defs);
 }
 
 bool ColumnFileInMemory::append(DMContext & context, const Block & data, size_t offset, size_t limit, size_t data_bytes)
@@ -62,8 +62,8 @@ bool ColumnFileInMemory::append(DMContext & context, const Block & data, size_t
 
     for (size_t i = 0; i < cache->block.columns(); ++i)
     {
-        auto & col = data.getByPosition(i).column;
-        auto & cache_col = *cache->block.getByPosition(i).column;
+        const auto & col = data.getByPosition(i).column;
+        const auto & cache_col = *cache->block.getByPosition(i).column;
         auto * mutable_cache_col = const_cast<IColumn *>(&cache_col);
         mutable_cache_col->insertRangeFrom(*col, offset, limit);
     }
@@ -85,27 +85,27 @@ Block ColumnFileInMemory::readDataForFlush() const
 }
 
 
-ColumnPtr ColumnInMemoryFileReader::getPKColumn()
+ColumnPtr ColumnFileInMemoryReader::getPKColumn()
 {
     memory_file.fillColumns(*col_defs, 1, cols_data_cache);
     return cols_data_cache[0];
 }
 
-ColumnPtr ColumnInMemoryFileReader::getVersionColumn()
+ColumnPtr ColumnFileInMemoryReader::getVersionColumn()
 {
     memory_file.fillColumns(*col_defs, 2, cols_data_cache);
     return cols_data_cache[1];
 }
 
-size_t ColumnInMemoryFileReader::readRows(MutableColumns & output_cols, size_t rows_offset, size_t rows_limit, const RowKeyRange * range)
+size_t ColumnFileInMemoryReader::readRows(MutableColumns & output_cols, size_t rows_offset, size_t rows_limit, const RowKeyRange * range)
 {
     memory_file.fillColumns(*col_defs, output_cols.size(), cols_data_cache);
 
     auto & pk_col = cols_data_cache[0];
     return copyColumnsData(cols_data_cache, pk_col, output_cols, rows_offset, rows_limit, range);
 }
 
-Block ColumnInMemoryFileReader::readNextBlock()
+Block ColumnFileInMemoryReader::readNextBlock()
 {
     if (read_done)
         return {};
@@ -118,10 +118,10 @@ Block ColumnInMemoryFileReader::readNextBlock()
     return genBlock(*col_defs, columns);
 }
 
-ColumnFileReaderPtr ColumnInMemoryFileReader::createNewReader(const ColumnDefinesPtr & new_col_defs)
+ColumnFileReaderPtr ColumnFileInMemoryReader::createNewReader(const ColumnDefinesPtr & new_col_defs)
 {
     // Reuse the cache data.
-    return std::make_shared<ColumnInMemoryFileReader>(memory_file, new_col_defs, cols_data_cache);
+    return std::make_shared<ColumnFileInMemoryReader>(memory_file, new_col_defs, cols_data_cache);
 }
 
 } // namespace DM

dbms/src/Storages/DeltaMerge/ColumnFile/ColumnFileInMemory.h

@@ -12,7 +12,7 @@ using ColumnInMemoryFilePtr = std::shared_ptr<ColumnFileInMemory>;
 /// A column file which is only resides in memory
 class ColumnFileInMemory : public ColumnFile
 {
-    friend class ColumnInMemoryFileReader;
+    friend class ColumnFileInMemoryReader;
 
 private:
     BlockPtr schema;
@@ -80,11 +80,6 @@ class ColumnFileInMemory : public ColumnFile
 
     Block readDataForFlush() const;
 
-    void serializeMetadata(WriteBuffer & /*buf*/, bool /*save_schema*/) const override
-    {
-        throw Exception("Unsupported operation", ErrorCodes::LOGICAL_ERROR);
-    }
-
     String toString() const override
     {
         String s = "{in_memory_file,rows:" + DB::toString(rows) //
@@ -97,7 +92,7 @@ class ColumnFileInMemory : public ColumnFile
 };
 
 
-class ColumnInMemoryFileReader : public ColumnFileReader
+class ColumnFileInMemoryReader : public ColumnFileReader
 {
 private:
     const ColumnFileInMemory & memory_file;
@@ -107,7 +102,7 @@ class ColumnInMemoryFileReader : public ColumnFileReader
     bool read_done = false;
 
 public:
-    ColumnInMemoryFileReader(const ColumnFileInMemory & memory_file_,
+    ColumnFileInMemoryReader(const ColumnFileInMemory & memory_file_,
                              const ColumnDefinesPtr & col_defs_,
                              const Columns & cols_data_cache_)
         : memory_file(memory_file_)
@@ -116,7 +111,7 @@ class ColumnInMemoryFileReader : public ColumnFileReader
     {
     }
 
-    ColumnInMemoryFileReader(const ColumnFileInMemory & memory_file_, const ColumnDefinesPtr & col_defs_)
+    ColumnFileInMemoryReader(const ColumnFileInMemory & memory_file_, const ColumnDefinesPtr & col_defs_)
         : memory_file(memory_file_)
         , col_defs(col_defs_)
     {