This is an automated cherry-pick of pingcap#9283

Signed-off-by: ti-chi-bot <ti-community-prow-bot@tidb.io>

dbms/src/Common/UniThreadPool.h

@@ -29,7 +29,6 @@
 #include <memory>
 #include <mutex>
 #include <optional>
-#include <queue>
 #include <thread>
 
 namespace DB

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

@@ -15,6 +15,7 @@
 #include <Common/ProfileEvents.h>
 #include <Storages/Page/V3/Blob/BlobFile.h>
 #include <Storages/Page/V3/Blob/BlobStat.h>
+#include <Storages/Page/V3/PageDefines.h>
 #include <Storages/PathPool.h>
 #include <boost_wrapper/string_split.h>
 #include <common/logger_useful.h>
@@ -216,8 +217,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 +300,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

@@ -36,7 +36,6 @@
 #include <fiu.h>
 
 #include <ext/scope_guard.h>
-#include <iterator>
 #include <magic_enum.hpp>
 #include <unordered_map>
 
@@ -730,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;
@@ -911,8 +911,7 @@ typename BlobStore<Trait>::PageMap BlobStore<Trait>::read(FieldReadInfos & to_re
             // TODO: Continuously fields can read by one system call.
             const auto [beg_offset, end_offset] = entry.getFieldOffsets(field_index);
             const auto size_to_read = end_offset - beg_offset;
-            auto blob_file
-                = read(page_id_v3, entry.file_id, entry.offset + beg_offset, write_offset, size_to_read, read_limiter);
+            read(page_id_v3, entry.file_id, entry.offset + beg_offset, write_offset, size_to_read, read_limiter);
             fields_offset_in_page.emplace(field_index, read_size_this_entry);
 
             if constexpr (BLOBSTORE_CHECKSUM_ON_READ)
@@ -926,17 +925,22 @@ typename BlobStore<Trait>::PageMap BlobStore<Trait>::read(FieldReadInfos & to_re
                     throw Exception(
                         ErrorCodes::CHECKSUM_DOESNT_MATCH,
                         "Reading with fields meet checksum not match "
+<<<<<<< HEAD
                         "[page_id={}] [expected=0x{:X}] [actual=0x{:X}] "
                         "[field_index={}] [field_offset={}] [field_size={}] "
                         "[entry={}] [file={}]",
+=======
+                        "page_id={} expected=0x{:X} actual=0x{:X} "
+                        "field_index={} field_offset={} field_size={} "
+                        "entry={}",
+>>>>>>> dc20fe919f (PageStorage: Fix empty page cause TiFlash failed to start (#9283))
                         page_id_v3,
                         expect_checksum,
                         field_checksum,
                         field_index,
                         beg_offset,
                         size_to_read,
-                        entry,
-                        blob_file->getPath());
+                        entry);
                 }
             }
 
@@ -1008,8 +1012,8 @@ typename BlobStore<Trait>::PageMap BlobStore<Trait>::read(
         for (const auto & [page_id_v3, entry] : entries)
         {
             // Unexpected behavior but do no harm
-            LOG_INFO(log, "Read entry without entry size, page_id={} entry={}", page_id_v3, entry);
             Page page(Trait::PageIdTrait::getU64ID(page_id_v3));
+            page.data = std::string_view(nullptr, 0);
             page_map.emplace(Trait::PageIdTrait::getPageMapKey(page_id_v3), page);
         }
         return page_map;
@@ -1022,7 +1026,7 @@ typename BlobStore<Trait>::PageMap BlobStore<Trait>::read(
     PageMap page_map;
     for (const auto & [page_id_v3, entry] : entries)
     {
-        auto blob_file = read(page_id_v3, entry.file_id, entry.offset, pos, entry.size, read_limiter);
+        read(page_id_v3, entry.file_id, entry.offset, pos, entry.size, read_limiter);
 
         if constexpr (BLOBSTORE_CHECKSUM_ON_READ)
         {
@@ -1032,6 +1036,7 @@ typename BlobStore<Trait>::PageMap BlobStore<Trait>::read(
             if (unlikely(entry.size != 0 && checksum != entry.checksum))
             {
                 throw Exception(
+<<<<<<< HEAD
                     fmt::format(
                         "Reading with entries meet checksum not match [page_id={}] [expected=0x{:X}] [actual=0x{:X}] "
                         "[entry={}] [file={}]",
@@ -1041,6 +1046,15 @@ typename BlobStore<Trait>::PageMap BlobStore<Trait>::read(
                         entry,
                         blob_file->getPath()),
                     ErrorCodes::CHECKSUM_DOESNT_MATCH);
+=======
+                    ErrorCodes::CHECKSUM_DOESNT_MATCH,
+                    "Reading with entries meet checksum not match page_id={} expected=0x{:X} actual=0x{:X} "
+                    "entry={}",
+                    page_id_v3,
+                    entry.checksum,
+                    checksum,
+                    entry);
+>>>>>>> dc20fe919f (PageStorage: Fix empty page cause TiFlash failed to start (#9283))
             }
         }
 
@@ -1098,15 +1112,15 @@ Page BlobStore<Trait>::read(const PageIdAndEntry & id_entry, const ReadLimiterPt
     if (buf_size == 0)
     {
         // Unexpected behavior but do no harm
-        LOG_INFO(log, "Read entry without entry size, page_id={} entry={}", page_id_v3, entry);
         Page page(Trait::PageIdTrait::getU64ID(page_id_v3));
+        page.data = std::string_view(nullptr, 0);
         return page;
     }
 
     char * data_buf = static_cast<char *>(alloc(buf_size));
     MemHolder mem_holder = createMemHolder(data_buf, [&, buf_size](char * p) { free(p, buf_size); });
 
-    auto blob_file = read(page_id_v3, entry.file_id, entry.offset, data_buf, buf_size, read_limiter);
+    read(page_id_v3, entry.file_id, entry.offset, data_buf, buf_size, read_limiter);
     if constexpr (BLOBSTORE_CHECKSUM_ON_READ)
     {
         ChecksumClass digest;
@@ -1115,15 +1129,13 @@ Page BlobStore<Trait>::read(const PageIdAndEntry & id_entry, const ReadLimiterPt
         if (unlikely(entry.size != 0 && checksum != entry.checksum))
         {
             throw Exception(
-                fmt::format(
-                    "Reading with entries meet checksum not match [page_id={}] [expected=0x{:X}] [actual=0x{:X}] "
-                    "[entry={}] [file={}]",
-                    page_id_v3,
-                    entry.checksum,
-                    checksum,
-                    entry,
-                    blob_file->getPath()),
-                ErrorCodes::CHECKSUM_DOESNT_MATCH);
+                ErrorCodes::CHECKSUM_DOESNT_MATCH,
+                "Reading with entries meet checksum not match page_id={} expected=0x{:X} actual=0x{:X} "
+                "entry={}",
+                page_id_v3,
+                entry.checksum,
+                checksum,
+                entry);
         }
     }
 
@@ -1142,7 +1154,7 @@ Page BlobStore<Trait>::read(const PageIdAndEntry & id_entry, const ReadLimiterPt
 }
 
 template <typename Trait>
-BlobFilePtr BlobStore<Trait>::read(
+void BlobStore<Trait>::read(
     const typename BlobStore<Trait>::PageId & page_id_v3,
     BlobFileId blob_id,
     BlobFileOffset offset,
@@ -1152,6 +1164,12 @@ BlobFilePtr BlobStore<Trait>::read(
     bool background)
 {
     GET_METRIC(tiflash_storage_page_command_count, type_read_blob).Increment();
+
+    // A shortcut to avoid unnecessary locks / system call when "reading an empty page"
+    // Reading an empty page should not create a BlobFile if it has already removed.
+    if (unlikely(size == 0))
+        return;
+
     assert(buffers != nullptr);
     BlobFilePtr blob_file = getBlobFile(blob_id);
     try
@@ -1170,7 +1188,6 @@ BlobFilePtr BlobStore<Trait>::read(
             background));
         e.rethrow();
     }
-    return blob_file;
 }
 
 

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

@@ -126,7 +126,7 @@ class BlobStore : private Allocator<false>
         PageType page_type,
         const WriteLimiterPtr & write_limiter = nullptr);
 
-    BlobFilePtr read(
+    void read(
         const PageId & page_id_v3,
         BlobFileId blob_id,
         BlobFileOffset offset,

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

@@ -17,9 +17,6 @@
 #include <Storages/Page/V3/spacemap/SpaceMap.h>
 #include <Storages/Page/V3/spacemap/SpaceMapSTDMap.h>
 #include <common/likely.h>
-#include <limits.h>
-#include <stdio.h>
-#include <stdlib.h>
 
 namespace DB
 {
@@ -52,14 +49,17 @@ SpaceMapPtr SpaceMap::createSpaceMap(SpaceMapType type, UInt64 start, UInt64 end
     return smap;
 }
 
-bool SpaceMap::checkSpace(UInt64 offset, size_t size) const
+bool SpaceMap::isInvalidRange(UInt64 offset, size_t size) const
 {
-    return (offset < start) || (offset > end) || (offset + size - 1 > end);
+    return (offset < start)
+        || (offset > end)
+        // check whether it can be changed to `(offset + size > end)`
+        || ((size != 0) && (offset + size - 1 > end));
 }
 
 bool SpaceMap::markFree(UInt64 offset, size_t length)
 {
-    if (checkSpace(offset, length))
+    if (isInvalidRange(offset, length))
     {
         throw Exception(
             fmt::format(
@@ -75,7 +75,7 @@ bool SpaceMap::markFree(UInt64 offset, size_t length)
 
 bool SpaceMap::markUsed(UInt64 offset, size_t length)
 {
-    if (checkSpace(offset, length))
+    if (isInvalidRange(offset, length))
     {
         throw Exception(
             fmt::format(
@@ -91,7 +91,7 @@ bool SpaceMap::markUsed(UInt64 offset, size_t length)
 
 bool SpaceMap::isMarkUsed(UInt64 offset, size_t length)
 {
-    if (checkSpace(offset, length))
+    if (isInvalidRange(offset, length))
     {
         throw Exception(
             fmt::format(

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
@@ -145,7 +145,7 @@ class SpaceMap
 
 private:
     /* Check the range */
-    bool checkSpace(UInt64 offset, size_t size) const;
+    bool isInvalidRange(UInt64 offset, size_t size) const;
 
 #ifndef DBMS_PUBLIC_GTEST
 protected:

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

@@ -20,6 +20,7 @@
 
 #include <ext/shared_ptr_helper.h>
 #include <map>
+#include <tuple>
 
 namespace DB
 {
@@ -153,6 +154,11 @@ class STDMapSpaceMap
 
     bool markUsedImpl(UInt64 offset, size_t length) override
     {
+        // An empty data, we can simply consider it is stored and return true
+        // Do not let it split the space into smaller pieces.
+        if (length == 0)
+            return true;
+
         auto it = MapUtils::findLessEQ(free_map, offset); // first free block <= `offset`
         if (it == free_map.end())
         {
@@ -217,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");
@@ -255,24 +268,23 @@ class STDMapSpaceMap
 
     bool markFreeImpl(UInt64 offset, size_t length) override
     {
-        auto it = free_map.find(offset);
+        // for an empty blob, no new free block is created, just skip
+        if (length == 0)
+        {
+            return true;
+        }
 
         /**
          * already unmarked.
          * The `offset` won't be mid of free space.
          * Because we alloc space from left to right.
          */
+        auto it = free_map.find(offset);
         if (it != free_map.end())
         {
             return true;
         }
 
-        // for an empty blob, no new free block is created, just skip
-        if (length == 0)
-        {
-            return true;
-        }
-
         bool meanless = false;
         std::tie(it, meanless) = free_map.insert({offset, length});
         insertIntoInvertIndex(length, offset);