save add block_wait

JaySon-Huang · JaySon-Huang · commit c8fe853b339f · 2025-10-23T15:52:37.000+08:00
Signed-off-by: JaySon-Huang &lt;tshent@qq.com&gt;
diff --git a/dbms/src/Interpreters/Settings.h b/dbms/src/Interpreters/Settings.h
@@ -249,6 +249,7 @@ struct Settings
     M(SettingDouble, dt_filecache_max_downloading_count_scale, 1.0, "Max downloading task count of FileCache = io thread count * dt_filecache_max_downloading_count_scale.")                                                            \
     M(SettingUInt64, dt_filecache_min_age_seconds, 1800, "Files of the same priority can only be evicted from files that were not accessed within `dt_filecache_min_age_seconds` seconds.")                                             \
     M(SettingBool, dt_enable_fetch_memtableset, true, "Whether fetching delta cache in FetchDisaggPages")                                                                                                                               \
+    M(SettingBool, dt_enable_block_wait, false, "Whether block wait until remote file is cached in compute node")                                                                                                                       \
     M(SettingUInt64, dt_fetch_pages_packet_limit_size, 512 * 1024, "Response packet bytes limit of FetchDisaggPages, 0 means one page per packet")                                                                                      \
     M(SettingDouble, dt_fetch_page_concurrency_scale, 4.0, "Concurrency of fetching pages of one query equals to num_streams * dt_fetch_page_concurrency_scale.")                                                                       \
     M(SettingDouble, dt_prepare_stream_concurrency_scale, 2.0, "Concurrency of preparing streams of one query equals to num_streams * dt_prepare_stream_concurrency_scale.")                                                            \
diff --git a/dbms/src/Server/DTTool/DTToolInspect.cpp b/dbms/src/Server/DTTool/DTToolInspect.cpp
@@ -288,7 +288,8 @@ int inspectServiceMain(DB::Context & context, const InspectArgs & args)
                     *dmfile,
                     fp,
                     nullptr,
-                    false,
+                    /*block_wait=*/false,
+                    /*set_cache_if_miss=*/false,
                     col_id,
                     nullptr,
                     nullptr);
diff --git a/dbms/src/Storages/DeltaMerge/File/ColumnStream.cpp b/dbms/src/Storages/DeltaMerge/File/ColumnStream.cpp
@@ -35,6 +35,7 @@ class MarkLoader
         auto mark_guard = S3::S3RandomAccessFile::setReadFileInfo({
             .size = reader.dmfile->getReadFileSize(col_id, colMarkFileName(file_name_base)),
             .scan_context = reader.scan_context,
+            .block_wait = block_wait,
         });
 
         if (likely(reader.dmfile->useMetaV2()))
@@ -59,17 +60,20 @@ class MarkLoader
         DMFileReader & reader_,
         ColId col_id_,
         const String & file_name_base_,
+        bool block_wait_,
         const ReadLimiterPtr & read_limiter_)
         : reader(reader_)
         , col_id(col_id_)
         , file_name_base(file_name_base_)
         , read_limiter(read_limiter_)
+        , block_wait(block_wait_)
     {}
 
     DMFileReader & reader;
     ColId col_id;
     const String & file_name_base;
     ReadLimiterPtr read_limiter;
+    bool block_wait;
 
 private:
     MarksInCompressedFilePtr loadRawColMarkTo(const MarksInCompressedFilePtr & res, size_t bytes_size)
@@ -288,6 +292,7 @@ ColumnReadStream::ColumnReadStream(
     ColId col_id,
     const String & file_name_base,
     size_t max_read_buffer_size,
+    bool block_wait,
     const LoggerPtr & log,
     const ReadLimiterPtr & read_limiter)
     : avg_size_hint(reader.dmfile->getColumnStat(col_id).avg_size)
@@ -297,9 +302,9 @@ ColumnReadStream::ColumnReadStream(
     if (reader.mark_cache)
         marks = reader.mark_cache->getOrSet(
             reader.dmfile->colMarkCacheKey(file_name_base),
-            MarkLoader(reader, col_id, file_name_base, read_limiter));
+            MarkLoader(reader, col_id, file_name_base, block_wait, read_limiter));
     else
-        marks = MarkLoader(reader, col_id, file_name_base, read_limiter)();
+        marks = MarkLoader(reader, col_id, file_name_base, block_wait, read_limiter)();
 
     // skip empty dmfile
     size_t packs = reader.dmfile->getPacks();
@@ -309,6 +314,7 @@ ColumnReadStream::ColumnReadStream(
     auto data_guard = S3::S3RandomAccessFile::setReadFileInfo({
         .size = reader.dmfile->getReadFileSize(col_id, colDataFileName(file_name_base)),
         .scan_context = reader.scan_context,
+        .block_wait = block_wait,
     });
 
     // load column data read buffer
diff --git a/dbms/src/Storages/DeltaMerge/File/ColumnStream.h b/dbms/src/Storages/DeltaMerge/File/ColumnStream.h
@@ -36,6 +36,7 @@ class ColumnReadStream
         ColId col_id,
         const String & file_name_base,
         size_t max_read_buffer_size,
+        bool block_wait,
         const LoggerPtr & log,
         const ReadLimiterPtr & read_limiter);
 
diff --git a/dbms/src/Storages/DeltaMerge/File/DMFileBlockInputStream.cpp b/dbms/src/Storages/DeltaMerge/File/DMFileBlockInputStream.cpp
@@ -95,6 +95,7 @@ DMFileBlockInputStreamPtr DMFileBlockInputStreamBuilder::buildNoLocalIndex(
         dmfile,
         read_columns,
         is_common_handle,
+        block_wait,
         enable_handle_clean_read,
         enable_del_clean_read,
         is_fast_scan,
@@ -136,6 +137,7 @@ SkippableBlockInputStreamPtr createSimpleBlockInputStream(
 
 DMFileBlockInputStreamBuilder & DMFileBlockInputStreamBuilder::setFromSettings(const Settings & settings)
 {
+    block_wait = settings.dt_enable_block_wait;
     enable_column_cache = settings.dt_enable_stable_column_cache;
     max_read_buffer_size = settings.max_read_buffer_size;
     max_sharing_column_bytes_for_all = settings.dt_max_sharing_column_bytes_for_all;
@@ -217,6 +219,7 @@ SkippableBlockInputStreamPtr DMFileBlockInputStreamBuilder::buildForVectorIndex(
         dmfile,
         *vec_index_ctx->rest_col_defs,
         is_common_handle,
+        block_wait,
         enable_handle_clean_read,
         enable_del_clean_read,
         is_fast_scan,
diff --git a/dbms/src/Storages/DeltaMerge/File/DMFileBlockInputStream.h b/dbms/src/Storages/DeltaMerge/File/DMFileBlockInputStream.h
@@ -244,6 +244,7 @@ class DMFileBlockInputStreamBuilder
     IdSetPtr read_packs;
     MarkCachePtr mark_cache;
     MinMaxIndexCachePtr index_cache;
+    bool block_wait = false;
     // column cache
     bool enable_column_cache = false;
     ColumnCachePtr column_cache;
diff --git a/dbms/src/Storages/DeltaMerge/File/DMFilePackFilter.cpp b/dbms/src/Storages/DeltaMerge/File/DMFilePackFilter.cpp
@@ -58,7 +58,7 @@ DMFilePackFilterResultPtr DMFilePackFilter::load()
     auto read_all_packs = (rowkey_ranges.size() == 1 && rowkey_ranges[0].all()) || rowkey_ranges.empty();
     if (!read_all_packs)
     {
-        tryLoadIndex(result.param, MutSup::extra_handle_id);
+        tryLoadIndex(result.param, MutSup::extra_handle_id, block_wait);
         std::vector<RSOperatorPtr> handle_filters;
         for (auto & rowkey_range : rowkey_ranges)
             handle_filters.emplace_back(toFilter(rowkey_range));
@@ -129,7 +129,7 @@ DMFilePackFilterResultPtr DMFilePackFilter::load()
         ColIds ids = filter->getColumnIDs();
         for (const auto & id : ids)
         {
-            tryLoadIndex(result.param, id);
+            tryLoadIndex(result.param, id, block_wait);
         }
 
         const auto check_results = filter->roughCheck(0, pack_count, result.param);
@@ -189,13 +189,21 @@ void DMFilePackFilter::loadIndex(
     const DMFilePtr & dmfile,
     const FileProviderPtr & file_provider,
     const MinMaxIndexCachePtr & index_cache,
+    bool block_wait,
     bool set_cache_if_miss,
     ColId col_id,
     const ReadLimiterPtr & read_limiter,
     const ScanContextPtr & scan_context)
 {
-    auto [type, minmax_index]
-        = loadIndex(*dmfile, file_provider, index_cache, set_cache_if_miss, col_id, read_limiter, scan_context);
+    auto [type, minmax_index] = loadIndex(
+        *dmfile,
+        file_provider,
+        index_cache,
+        block_wait,
+        set_cache_if_miss,
+        col_id,
+        read_limiter,
+        scan_context);
     indexes.emplace(col_id, RSIndex(type, minmax_index));
 }
 
@@ -214,6 +222,7 @@ class MinMaxIndexLoader
         auto index_guard = S3::S3RandomAccessFile::setReadFileInfo({
             .size = dmfile.getReadFileSize(col_id, colIndexFileName(file_name_base)),
             .scan_context = scan_context,
+            .block_wait = block_wait,
         });
 
         if (likely(dmfile.useMetaV2()))
@@ -238,6 +247,7 @@ class MinMaxIndexLoader
         const DMFile & dmfile_,
         const FileProviderPtr & file_provider_,
         ColId col_id_,
+        bool block_wait_,
         const ReadLimiterPtr & read_limiter_,
         const ScanContextPtr & scan_context_)
         : dmfile(dmfile_)
@@ -246,6 +256,7 @@ class MinMaxIndexLoader
         , file_provider(file_provider_)
         , read_limiter(read_limiter_)
         , scan_context(scan_context_)
+        , block_wait(block_wait_)
     {}
 
     const DMFile & dmfile;
@@ -254,6 +265,7 @@ class MinMaxIndexLoader
     FileProviderPtr file_provider;
     ReadLimiterPtr read_limiter;
     ScanContextPtr scan_context;
+    bool block_wait = false;
 
 private:
     MinMaxIndexPtr loadRawMinMaxIndex(const DataTypePtr & type, size_t index_file_size) const
@@ -339,6 +351,7 @@ std::pair<DataTypePtr, MinMaxIndexPtr> DMFilePackFilter::loadIndex(
     const DMFile & dmfile,
     const FileProviderPtr & file_provider,
     const MinMaxIndexCachePtr & index_cache,
+    bool block_wait,
     bool set_cache_if_miss,
     ColId col_id,
     const ReadLimiterPtr & read_limiter,
@@ -350,7 +363,7 @@ std::pair<DataTypePtr, MinMaxIndexPtr> DMFilePackFilter::loadIndex(
     MinMaxIndexPtr minmax_index;
     if (index_cache && set_cache_if_miss)
     {
-        auto loader = MinMaxIndexLoader(dmfile, file_provider, col_id, read_limiter, scan_context);
+        auto loader = MinMaxIndexLoader(dmfile, file_provider, col_id, block_wait, read_limiter, scan_context);
         minmax_index = index_cache->getOrSet(dmfile.colIndexCacheKey(file_name_base), loader);
     }
     else
@@ -359,12 +372,12 @@ std::pair<DataTypePtr, MinMaxIndexPtr> DMFilePackFilter::loadIndex(
         if (index_cache)
             minmax_index = index_cache->get(dmfile.colIndexCacheKey(file_name_base));
         if (minmax_index == nullptr)
-            minmax_index = MinMaxIndexLoader(dmfile, file_provider, col_id, read_limiter, scan_context)();
+            minmax_index = MinMaxIndexLoader(dmfile, file_provider, col_id, block_wait, read_limiter, scan_context)();
     }
     return {type, minmax_index};
 }
 
-void DMFilePackFilter::tryLoadIndex(RSCheckParam & param, ColId col_id)
+void DMFilePackFilter::tryLoadIndex(RSCheckParam & param, ColId col_id, bool block_wait)
 {
     if (param.indexes.count(col_id))
         return;
@@ -373,7 +386,16 @@ void DMFilePackFilter::tryLoadIndex(RSCheckParam & param, ColId col_id)
         return;
 
     Stopwatch watch;
-    loadIndex(param.indexes, dmfile, file_provider, index_cache, set_cache_if_miss, col_id, read_limiter, scan_context);
+    loadIndex(
+        param.indexes,
+        dmfile,
+        file_provider,
+        index_cache,
+        block_wait,
+        set_cache_if_miss,
+        col_id,
+        read_limiter,
+        scan_context);
 }
 
 std::pair<std::vector<DMFilePackFilter::Range>, DMFilePackFilterResults> DMFilePackFilter::getSkippedRangeAndFilter(
diff --git a/dbms/src/Storages/DeltaMerge/File/DMFilePackFilter.h b/dbms/src/Storages/DeltaMerge/File/DMFilePackFilter.h
@@ -154,6 +154,7 @@ class DMFilePackFilter
         const DMFile & dmfile,
         const FileProviderPtr & file_provider,
         const MinMaxIndexCachePtr & index_cache,
+        bool block_wait,
         bool set_cache_if_miss,
         ColId col_id,
         const ReadLimiterPtr & read_limiter,
@@ -190,17 +191,19 @@ class DMFilePackFilter
         const DMFilePtr & dmfile,
         const FileProviderPtr & file_provider,
         const MinMaxIndexCachePtr & index_cache,
+        bool block_wait,
         bool set_cache_if_miss,
         ColId col_id,
         const ReadLimiterPtr & read_limiter,
         const ScanContextPtr & scan_context);
 
-    void tryLoadIndex(RSCheckParam & param, ColId col_id);
+    void tryLoadIndex(RSCheckParam & param, ColId col_id, bool block_wait);
 
 private:
     DMFilePtr dmfile;
 
     MinMaxIndexCachePtr index_cache;
+    bool block_wait;
     bool set_cache_if_miss;
     RowKeyRanges rowkey_ranges;
     RSOperatorPtr filter;
diff --git a/dbms/src/Storages/DeltaMerge/File/DMFilePackFilterResult.cpp b/dbms/src/Storages/DeltaMerge/File/DMFilePackFilterResult.cpp
@@ -78,6 +78,7 @@ void DMFilePackFilterResult::tryLoadIndex(
         dmfile,
         file_provider,
         index_cache,
+        /*block_wait=*/false,
         /*set_cache_if_miss=*/true,
         col_id,
         read_limiter,
diff --git a/dbms/src/Storages/DeltaMerge/File/DMFileReader.cpp b/dbms/src/Storages/DeltaMerge/File/DMFileReader.cpp
@@ -54,6 +54,7 @@ DMFileReader::DMFileReader(
     const DMFilePtr & dmfile_,
     const ColumnDefines & read_columns_,
     bool is_common_handle_,
+    bool block_wait,
     // clean read
     bool enable_handle_clean_read_,
     bool enable_del_clean_read_,
@@ -114,6 +115,7 @@ DMFileReader::DMFileReader(
                 cd.id,
                 stream_name,
                 max_read_buffer_size,
+                block_wait,
                 log,
                 read_limiter);
             column_streams.emplace(stream_name, std::move(stream));
diff --git a/dbms/src/Storages/DeltaMerge/File/DMFileReader.h b/dbms/src/Storages/DeltaMerge/File/DMFileReader.h
@@ -46,6 +46,7 @@ class DMFileReader
         const DMFilePtr & dmfile_,
         const ColumnDefines & read_columns_,
         bool is_common_handle_,
+        bool block_wait,
         // Only set this param to true when
         // 1. There is no delta.
         // 2. You don't need pk, version and delete_tag columns
diff --git a/dbms/src/Storages/DeltaMerge/ScanContext.cpp b/dbms/src/Storages/DeltaMerge/ScanContext.cpp
@@ -135,6 +135,8 @@ String ScanContext::toJson() const
         json->set("disagg_page_miss_count", disagg_page_miss_count.load());
         json->set("disagg_s3file_hit_count", disagg_s3file_hit_count.load());
         json->set("disagg_s3file_miss_count", disagg_s3file_miss_count.load());
+        json->set("disagg_s3file_block_wait_count", disagg_s3file_block_wait_count.load());
+        json->set("disagg_s3file_block_wait_ms", disagg_s3file_block_wait_ms.load());
         json->set("disagg_s3file_get_object_ms", disagg_s3file_get_object_ms.load());
         json->set("disagg_s3file_get_object_count", disagg_s3file_get_object_count.load());
         json->set("disagg_s3file_read_time_ms", disagg_s3file_read_time_ms.load());
diff --git a/dbms/src/Storages/DeltaMerge/ScanContext.h b/dbms/src/Storages/DeltaMerge/ScanContext.h
@@ -72,6 +72,8 @@ class ScanContext
     std::atomic<uint64_t> disagg_page_miss_count{0};
     std::atomic<uint64_t> disagg_s3file_hit_count{0};
     std::atomic<uint64_t> disagg_s3file_miss_count{0};
+    std::atomic<uint64_t> disagg_s3file_block_wait_count{0};
+    std::atomic<uint64_t> disagg_s3file_block_wait_ms{0};
     std::atomic<uint64_t> disagg_s3file_get_object_count{0};
     std::atomic<uint64_t> disagg_s3file_get_object_ms{0};
     std::atomic<uint64_t> disagg_s3file_read_count{0};
diff --git a/dbms/src/Storages/DeltaMerge/VersionChain/Common.cpp b/dbms/src/Storages/DeltaMerge/VersionChain/Common.cpp
@@ -99,6 +99,7 @@ std::vector<T> loadPackMaxValue(const DMContext & dm_context, const DMFile & dmf
         dmfile,
         global_context.getFileProvider(),
         global_context.getMinMaxIndexCache(),
+        false,
         /*set_cache_if_miss*/ true,
         col_id,
         global_context.getReadLimiter(),
@@ -138,6 +139,7 @@ std::optional<std::pair<HandleType, HandleType>> loadDMFileHandleRange(
         dmfile,
         global_context.getFileProvider(),
         global_context.getMinMaxIndexCache(),
+        false,
         /*set_cache_if_miss*/ true,
         MutSup::extra_handle_id,
         global_context.getReadLimiter(),
diff --git a/dbms/src/Storages/DeltaMerge/VersionChain/DMFileHandleIndex.h b/dbms/src/Storages/DeltaMerge/VersionChain/DMFileHandleIndex.h
@@ -154,6 +154,7 @@ class DMFileHandleIndex
             dmfile,
             {getHandleColumnDefine<HandleType>()},
             isCommonHandle<HandleType>(),
+            /*block_wait*/ false,
             /*enable_handle_clean_read*/ false,
             /*enable_del_clean_read*/ false,
             /*is_fast_scan*/ false,
diff --git a/dbms/src/Storages/S3/S3RandomAccessFile.cpp b/dbms/src/Storages/S3/S3RandomAccessFile.cpp
diff --git a/dbms/src/Storages/S3/S3RandomAccessFile.h b/dbms/src/Storages/S3/S3RandomAccessFile.h
diff --git a/dbms/src/Storages/StorageDisaggregated.cpp b/dbms/src/Storages/StorageDisaggregated.cpp
diff --git a/dbms/src/Storages/StorageDisaggregatedRemote.cpp b/dbms/src/Storages/StorageDisaggregatedRemote.cpp
