Storage: Refactor DMFileReader

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

@@ -14,19 +14,18 @@
 
 #include <Storages/DeltaMerge/File/ColumnCache.h>
 
-namespace DB
+namespace DB::DM
 {
-namespace DM
-{
-RangeWithStrategys ColumnCache::getReadStrategy(size_t pack_id, size_t pack_count, ColId column_id)
+
+RangeWithStrategys ColumnCache::getReadStrategy(size_t start_pack_idx, size_t pack_count, ColId column_id)
 {
-    PackRange target_range{pack_id, pack_id + pack_count};
+    PackRange target_range{start_pack_idx, start_pack_idx + 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 +35,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 +44,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 +54,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_idx,
+    size_t pack_count,
+    const std::vector<size_t> & clean_read_pack_idx)
+{
+    PackRange target_range{start_pack_idx, start_pack_idx + 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 (std::find(clean_read_pack_idx.cbegin(), clean_read_pack_idx.cend(), cursor) != clean_read_pack_idx.cend())
+        {
+            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;
 }
 
@@ -125,5 +169,4 @@ bool ColumnCache::isPackInCache(PackId pack_id, ColId column_id)
     return false;
 }
 
-} // namespace DM
-} // namespace DB
+} // namespace DB::DM

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

@@ -14,18 +14,15 @@
 
 #pragma once
 
-#include <Core/Block.h>
 #include <Storages/DeltaMerge/DeltaMergeDefines.h>
 #include <Storages/KVStore/Types.h>
-#include <common/logger_useful.h>
 
 #include <cstddef>
 #include <memory>
 
-namespace DB
-{
-namespace DM
+namespace DB::DM
 {
+
 using ColId = DB::ColumnID;
 using PackId = size_t;
 using PackRange = std::pair<PackId, PackId>;
@@ -46,7 +43,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_idx, size_t pack_count, ColId column_id);
+    static RangeWithStrategys getReadStrategy(
+        size_t start_pack_idx,
+        size_t pack_count,
+        const std::vector<size_t> & clean_read_pack_idx);
 
     void tryPutColumn(size_t pack_id, ColId column_id, const ColumnPtr & column, size_t rows_offset, size_t rows_count);
 
@@ -74,5 +75,5 @@ using ColumnCachePtrs = std::vector<ColumnCachePtr>;
 using RangeWithStrategy = ColumnCache::RangeWithStrategy;
 using RangeWithStrategys = ColumnCache::RangeWithStrategys;
 using ColumnCacheElement = ColumnCache::ColumnCacheElement;
-} // namespace DM
-} // namespace DB
+
+} // namespace DB::DM