Storage: Refactor DMFileReader

dbms/src/Storages/DeltaMerge/BitmapFilter/BitmapFilter.cpp

@@ -73,10 +73,10 @@ void BitmapFilter::set(const UInt32 * data, UInt32 size, const FilterPtr & f)
     }
 }
 
-void BitmapFilter::set(UInt32 start, UInt32 limit)
+void BitmapFilter::set(UInt32 start, UInt32 limit, bool value)
 {
     RUNTIME_CHECK(start + limit <= filter.size(), start, limit, filter.size());
-    std::fill(filter.begin() + start, filter.begin() + start + limit, true);
+    std::fill_n(filter.begin() + start, limit, value);
 }
 
 bool BitmapFilter::get(IColumn::Filter & f, UInt32 start, UInt32 limit) const

dbms/src/Storages/DeltaMerge/BitmapFilter/BitmapFilter.h

@@ -28,7 +28,7 @@ class BitmapFilter
     void set(BlockInputStreamPtr & stream);
     void set(const ColumnPtr & col, const FilterPtr & f);
     void set(const UInt32 * data, UInt32 size, const FilterPtr & f);
-    void set(UInt32 start, UInt32 limit);
+    void set(UInt32 start, UInt32 limit, bool value = true);
     // If return true, all data is match and do not fill the filter.
     bool get(IColumn::Filter & f, UInt32 start, UInt32 limit) const;
     // filter[start, limit] & f -> f

dbms/src/Storages/DeltaMerge/File/ColumnCache.cpp

@@ -18,15 +18,15 @@ namespace DB
 {
 namespace DM
 {
-RangeWithStrategys ColumnCache::getReadStrategy(size_t pack_id, size_t pack_count, ColId column_id)
+RangeWithStrategys ColumnCache::getReadStrategy(size_t start_pack_id, size_t pack_count, ColId column_id)
 {
-    PackRange target_range{pack_id, pack_id + pack_count};
+    PackRange target_range{start_pack_id, start_pack_id + pack_count};
 
     RangeWithStrategys range_and_strategys;
-
-    Strategy strategy = Strategy::Unknown;
+    range_and_strategys.reserve(pack_count);
+    auto strategy = Strategy::Unknown;
     size_t range_start = 0;
-    for (size_t cursor = target_range.first; cursor < target_range.second; cursor++)
+    for (size_t cursor = target_range.first; cursor < target_range.second; ++cursor)
     {
         if (isPackInCache(cursor, column_id))
         {
@@ -36,7 +36,7 @@ RangeWithStrategys ColumnCache::getReadStrategy(size_t pack_id, size_t pack_coun
             }
             else if (strategy == Strategy::Disk)
             {
-                range_and_strategys.emplace_back(std::make_pair(PackRange{range_start, cursor}, Strategy::Disk));
+                range_and_strategys.emplace_back(PackRange{range_start, cursor}, Strategy::Disk);
             }
             range_start = cursor;
             strategy = Strategy::Memory;
@@ -45,7 +45,7 @@ RangeWithStrategys ColumnCache::getReadStrategy(size_t pack_id, size_t pack_coun
         {
             if (strategy == Strategy::Memory)
             {
-                range_and_strategys.emplace_back(std::make_pair(PackRange{range_start, cursor}, Strategy::Memory));
+                range_and_strategys.emplace_back(PackRange{range_start, cursor}, Strategy::Memory);
             }
             else if (strategy == Strategy::Disk)
             {
@@ -55,8 +55,53 @@ RangeWithStrategys ColumnCache::getReadStrategy(size_t pack_id, size_t pack_coun
             strategy = Strategy::Disk;
         }
     }
-    range_and_strategys.emplace_back(std::make_pair(PackRange{range_start, target_range.second}, strategy));
+    range_and_strategys.emplace_back(PackRange{range_start, target_range.second}, strategy);
+    range_and_strategys.shrink_to_fit();
+    return range_and_strategys;
+}
+
+RangeWithStrategys ColumnCache::getReadStrategy(
+    size_t start_pack_id,
+    size_t pack_count,
+    const std::unordered_set<size_t> & memory_pack_ids)
+{
+    PackRange target_range{start_pack_id, start_pack_id + pack_count};
 
+    RangeWithStrategys range_and_strategys;
+    range_and_strategys.reserve(pack_count);
+    auto strategy = Strategy::Unknown;
+    size_t range_start = 0;
+    for (size_t cursor = target_range.first; cursor < target_range.second; ++cursor)
+    {
+        if (memory_pack_ids.contains(cursor))
+        {
+            if (strategy == Strategy::Memory)
+            {
+                continue;
+            }
+            else if (strategy == Strategy::Disk)
+            {
+                range_and_strategys.emplace_back(PackRange{range_start, cursor}, Strategy::Disk);
+            }
+            range_start = cursor;
+            strategy = Strategy::Memory;
+        }
+        else
+        {
+            if (strategy == Strategy::Memory)
+            {
+                range_and_strategys.emplace_back(PackRange{range_start, cursor}, Strategy::Memory);
+            }
+            else if (strategy == Strategy::Disk)
+            {
+                continue;
+            }
+            range_start = cursor;
+            strategy = Strategy::Disk;
+        }
+    }
+    range_and_strategys.emplace_back(PackRange{range_start, target_range.second}, strategy);
+    range_and_strategys.shrink_to_fit();
     return range_and_strategys;
 }
 

dbms/src/Storages/DeltaMerge/File/ColumnCache.h

@@ -46,7 +46,11 @@ class ColumnCache
 
     using RangeWithStrategy = std::pair<PackRange, ColumnCache::Strategy>;
     using RangeWithStrategys = std::vector<RangeWithStrategy>;
-    RangeWithStrategys getReadStrategy(size_t pack_id, size_t pack_count, ColId column_id);
+    RangeWithStrategys getReadStrategy(size_t start_pack_id, size_t pack_count, ColId column_id);
+    static RangeWithStrategys getReadStrategy(
+        size_t start_pack_id,
+        size_t pack_count,
+        const std::unordered_set<size_t> & memory_pack_ids);
 
     void tryPutColumn(size_t pack_id, ColId column_id, const ColumnPtr & column, size_t rows_offset, size_t rows_count);