Always put those empty pages at the front of BlobFile

dbms/src/Storages/Page/V3/Blob/BlobStat.cpp

@@ -216,8 +216,6 @@ std::pair<BlobStats::BlobStatPtr, BlobFileId> BlobStats::chooseStat(
     PageType page_type,
     const std::lock_guard<std::mutex> &)
 {
-    BlobStatPtr stat_ptr = nullptr;
-
     // No stats exist
     if (stats_map.empty())
     {
@@ -301,6 +299,12 @@ BlobStats::StatsMap BlobStats::getStats() const NO_THREAD_SAFETY_ANALYSIS
 
 BlobFileOffset BlobStats::BlobStat::getPosFromStat(size_t buf_size, const std::unique_lock<std::mutex> &)
 {
+    // A shortcut for empty page. All empty pages will be stored
+    // at the beginning of the BlobFile. It should not affects the
+    // sm_max_caps or other fields by adding these empty pages.
+    if (unlikely(buf_size == 0))
+        return 0;
+
     BlobFileOffset offset = 0;
     UInt64 max_cap = 0;
     bool expansion = true;

dbms/src/Storages/Page/V3/BlobStore.cpp

@@ -729,6 +729,7 @@ std::pair<BlobFileId, BlobFileOffset> BlobStore<Trait>::getPosFromStats(size_t s
     Stopwatch watch;
     BlobStatPtr stat;
 
+    // TODO: make this lambda as a function of BlobStats to simplify code
     auto lock_stat = [size, this, &stat, &page_type]() NO_THREAD_SAFETY_ANALYSIS {
         auto lock_stats = blob_stats.lock();
         BlobFileId blob_file_id = INVALID_BLOBFILE_ID;

dbms/src/Storages/Page/V3/spacemap/SpaceMap.h

@@ -82,7 +82,7 @@ class SpaceMap
      * If such span is found.
      * It will mark that span to be used and also return a hint of the max capacity available in this SpaceMap. 
      * 
-     * return value is <insert_offset, max_cap>:
+     * return value is <insert_offset, max_cap, is_expansion>:
      *  insert_offset: start offset for the inserted space
      *  max_cap: A hint of the largest available space this SpaceMap can hold. 
      *  is_expansion: Whether it is an expansion span

dbms/src/Storages/Page/V3/spacemap/SpaceMapSTDMap.h

@@ -20,6 +20,7 @@
 
 #include <ext/shared_ptr_helper.h>
 #include <map>
+#include <tuple>
 
 namespace DB
 {
@@ -222,8 +223,15 @@ class STDMapSpaceMap
         return true;
     }
 
+    // return value is <insert_offset, max_cap, is_expansion>
     std::tuple<UInt64, UInt64, bool> searchInsertOffset(size_t size) override
     {
+        if (unlikely(size == 0))
+        {
+            // The returned `max_cap` is 0 under this case, user should not use it.
+            return std::make_tuple(0, 0, false);
+        }
+
         if (unlikely(free_map.empty()))
         {
             LOG_ERROR(Logger::get(), "Current space map is full");

dbms/src/Storages/Page/V3/tests/gtest_blob_stat.cpp

@@ -391,14 +391,21 @@ TEST_F(BlobStoreStatsTest, StatWithEmptyBlob)
     ASSERT_EQ(offset, 0);
 
     offset = stat->getPosFromStat(0, stat->lock()); // empty
-    ASSERT_EQ(offset, 10);
+    ASSERT_EQ(offset, 0); // empty page always "stored" to the beginning of the space
+    offset = stat->getPosFromStat(0, stat->lock()); // empty
+    ASSERT_EQ(offset, 0); // empty page always "stored" to the beginning of the space
+    offset = stat->getPosFromStat(0, stat->lock()); // empty
+    ASSERT_EQ(offset, 0); // empty page always "stored" to the beginning of the space
 
     offset = stat->getPosFromStat(20, stat->lock());
     ASSERT_EQ(offset, 10);
 
     offset = stat->getPosFromStat(100, stat->lock());
     ASSERT_EQ(offset, 30);
 
+    offset = stat->getPosFromStat(0, stat->lock()); // empty
+    ASSERT_EQ(offset, 0); // empty page always "stored" to the beginning of the space
+
     ASSERT_EQ(stat->sm_total_size, 10 + 20 + 100);
     ASSERT_EQ(stat->sm_valid_size, 10 + 20 + 100);
     ASSERT_EQ(stat->sm_valid_rate, 1);
@@ -419,9 +426,8 @@ TEST_F(BlobStoreStatsTest, testFullStats)
     BlobStats stats(logger, delegator, config);
 
     {
-        std::lock_guard lock(stats.lock_stats);
         BlobFileId file_id = 10;
-        BlobStats::BlobStatPtr stat = stats.createStat(file_id, config.file_limit_size, lock);
+        BlobStats::BlobStatPtr stat = stats.createStat(file_id, config.file_limit_size, stats.lock());
         auto offset = stat->getPosFromStat(BLOBFILE_LIMIT_SIZE - 1, stat->lock());
         ASSERT_EQ(offset, 0);
         stats.cur_max_id = file_id;