[fix](io) Harden Parquet reader-local cache for File Scanner V2

be/benchmark/parquet/README.md

@@ -34,8 +34,8 @@ timed region. It covers PLAIN, dictionary, byte-stream-split, and DELTA encoding
 supported fixed-width and binary physical types. Sparse selections are provided as both one
 clustered range and many alternating ranges.
 
-The decoder selection axis includes 0%, 1%, 10%, 50%, 90%, and 100% so boundary and
-high-selectivity behavior are visible.
+The decoder selection axis includes 0%, 1%, 5%, 10%, 50%, 90%, and 100% so Q28-shaped sparse,
+boundary, and high-selectivity behavior are visible.
 
 ```shell
 be/output/lib/benchmark_test \

be/benchmark/parquet/benchmark_parquet_decoder.hpp

@@ -516,8 +516,7 @@ inline DecoderDigest expected_decoder_digest(const DecoderScenario& scenario,
 
 inline Status verify_decoder_output(format::parquet::native::Decoder* decoder, Slice* encoded,
                                     const DecoderScenario& scenario, bool binary,
-                                    const ParquetSelection& selection,
-                                    const SelectionPlan& plan) {
+                                    const ParquetSelection& selection, const SelectionPlan& plan) {
     RETURN_IF_ERROR(decoder->set_data(encoded));
     DecoderDigest actual;
     if (scenario.encoding == Encoding::DICTIONARY) {
@@ -626,7 +625,7 @@ inline void run_decoder(benchmark::State& state, DecoderScenario scenario, int s
 
 inline bool register_decoder_benchmarks() {
     for (const auto& scenario : decoder_scenarios()) {
-        for (const int selectivity : {0, 1, 10, 50, 90, 100}) {
+        for (const int selectivity : {0, 1, 5, 10, 50, 90, 100}) {
             for (const auto pattern : {Pattern::CLUSTERED, Pattern::ALTERNATING}) {
                 const std::string name = "ParquetDecoder/" + to_string(scenario.encoding) + "/" +
                                          to_string(scenario.value_type) + "/sel_" +

be/src/common/config.cpp

@@ -1271,6 +1271,8 @@ DEFINE_mInt64(file_cache_background_block_lru_update_qps_limit, "1000");
 DEFINE_mInt64(file_cache_background_block_lru_update_queue_max_size, "500000");
 DEFINE_mBool(enable_file_cache_async_touch_on_get_or_set, "false");
 DEFINE_mBool(enable_reader_dryrun_when_download_file_cache, "true");
+DEFINE_mBool(enable_file_scanner_v2_reader_local_cache, "true");
+DEFINE_mInt64(file_scanner_v2_reader_local_cache_size, "67108864"); // 64MB per scanner
 DEFINE_mInt64(file_cache_background_monitor_interval_ms, "5000");
 DEFINE_mInt64(file_cache_background_ttl_gc_interval_ms, "180000");
 DEFINE_mInt64(file_cache_background_ttl_info_update_interval_ms, "180000");

be/src/common/config.h

@@ -1310,6 +1310,11 @@ DECLARE_mInt64(file_cache_background_block_lru_update_qps_limit);
 DECLARE_mInt64(file_cache_background_block_lru_update_queue_max_size);
 DECLARE_mBool(enable_file_cache_async_touch_on_get_or_set);
 DECLARE_mBool(enable_reader_dryrun_when_download_file_cache);
+// Cache File Scanner V2 file-cache blocks in reader-local memory. File Scanner V1 and internal
+// table readers never opt in to this cache.
+DECLARE_mBool(enable_file_scanner_v2_reader_local_cache);
+// Maximum reader-local cache bytes managed by one File Scanner V2 scanner.
+DECLARE_mInt64(file_scanner_v2_reader_local_cache_size);
 DECLARE_mInt64(file_cache_background_monitor_interval_ms);
 DECLARE_mInt64(file_cache_background_ttl_gc_interval_ms);
 DECLARE_mInt64(file_cache_background_ttl_info_update_interval_ms);

be/src/exec/scan/file_scanner_v2.cpp

@@ -67,6 +67,7 @@
 #include "format_v2/table_reader.h"
 #include "format_v2/wal/wal_table_reader.h"
 #include "io/cache/block_file_cache_profile.h"
+#include "io/cache/cached_remote_file_reader.h"
 #include "io/fs/file_meta_cache.h"
 #include "io/io_common.h"
 #include "runtime/descriptors.h"
@@ -281,6 +282,14 @@ Status adapt_runtime_filter_for_table_reader(VExprSPtr* expr) {
 
 } // namespace
 
+int64_t FileScannerV2::_cumulative_profile_delta(int64_t current, int64_t* reported) {
+    DORIS_CHECK(reported != nullptr);
+    DORIS_CHECK(current >= *reported);
+    const int64_t delta = current - *reported;
+    *reported = current;
+    return delta;
+}
+
 #ifdef BE_TEST
 FileScannerV2::FileScannerV2(RuntimeState* state, RuntimeProfile* profile,
                              std::unique_ptr<format::TableReader> table_reader)
@@ -956,6 +965,12 @@ Status FileScannerV2::_to_file_format(TFileFormatType::type format_type,
 
 Status FileScannerV2::_init_io_ctx() {
     _io_ctx = create_file_scan_io_context(_state);
+    if (config::enable_file_scanner_v2_reader_local_cache) {
+        const size_t capacity = cast_set<size_t>(
+                std::max<int64_t>(0, config::file_scanner_v2_reader_local_cache_size));
+        _io_ctx->reader_local_cache = std::make_shared<io::FileScannerV2ReaderLocalCache>(
+                capacity, _state->query_mem_tracker());
+    }
     return Status::OK();
 }
 
@@ -1095,7 +1110,10 @@ void FileScannerV2::update_realtime_counters() {
     _state->get_query_ctx()->resource_ctx()->io_context()->update_scan_bytes_from_remote_storage(
             deltas.scan_bytes_from_remote_storage);
 
-    COUNTER_SET(_file_read_bytes_counter, bytes_read);
+    // Scanner instances share the profile counter, so publishing an absolute value would erase
+    // bytes already reported by sibling scanners.
+    COUNTER_UPDATE(_file_read_bytes_counter,
+                   _cumulative_profile_delta(bytes_read, &_reported_file_read_bytes));
     COUNTER_SET(_file_read_calls_counter, cast_set<int64_t>(_file_reader_stats->read_calls));
     COUNTER_SET(_file_read_time_counter, cast_set<int64_t>(_file_reader_stats->read_time_ns));
 
@@ -1192,7 +1210,9 @@ void FileScannerV2::_collect_profile_before_close() {
         _reported_file_cache_statistics = *_file_cache_statistics;
     }
     if (_file_reader_stats != nullptr) {
-        COUNTER_SET(_file_read_bytes_counter, cast_set<int64_t>(_file_reader_stats->read_bytes));
+        COUNTER_UPDATE(_file_read_bytes_counter,
+                       _cumulative_profile_delta(cast_set<int64_t>(_file_reader_stats->read_bytes),
+                                                 &_reported_file_read_bytes));
         COUNTER_SET(_file_read_calls_counter, cast_set<int64_t>(_file_reader_stats->read_calls));
         COUNTER_SET(_file_read_time_counter, cast_set<int64_t>(_file_reader_stats->read_time_ns));
         const auto read_time = cast_set<int64_t>(_file_reader_stats->read_time_ns);

be/src/exec/scan/file_scanner_v2.h

@@ -87,6 +87,9 @@ class FileScannerV2 final : public Scanner {
             int64_t* last_bytes_read_from_remote);
     static void TEST_report_file_cache_profile(
             RuntimeProfile* profile, const io::FileCacheStatistics& file_cache_statistics);
+    static int64_t TEST_cumulative_profile_delta(int64_t current, int64_t* reported) {
+        return _cumulative_profile_delta(current, reported);
+    }
     static bool TEST_should_skip_not_found(const Status& status, bool ignore_not_found);
     static bool TEST_should_skip_empty(const Status& status, bool stopped);
     static Status TEST_contextualize_output_filter_status(Status status,
@@ -98,6 +101,13 @@ class FileScannerV2 final : public Scanner {
         return _should_run_adaptive_batch_size(predictor_initialized,
                                                current_split_uses_metadata_count);
     }
+    Status TEST_init_io_ctx() { return _init_io_ctx(); }
+    bool TEST_has_reader_local_cache() const {
+        return _io_ctx != nullptr && _io_ctx->reader_local_cache != nullptr;
+    }
+    const void* TEST_reader_local_cache() const {
+        return _io_ctx != nullptr ? _io_ctx->reader_local_cache.get() : nullptr;
+    }
 #endif
 
     FileScannerV2(RuntimeState* state, FileScanLocalState* parent, int64_t limit,
@@ -137,6 +147,7 @@ class FileScannerV2 final : public Scanner {
     static bool _should_skip_empty(const Status& status, bool stopped);
     static Status _contextualize_output_filter_status(Status status,
                                                       TFileFormatType::type format_type);
+    static int64_t _cumulative_profile_delta(int64_t current, int64_t* reported);
     bool _should_enable_file_meta_cache() const;
     std::optional<format::GlobalRowIdContext> _create_global_rowid_context(
             const TFileRangeDesc& range) const;
@@ -227,6 +238,7 @@ class FileScannerV2 final : public Scanner {
     int64_t _last_bytes_read_from_local = 0;
     int64_t _last_bytes_read_from_remote = 0;
     int64_t _reported_io_read_time = 0;
+    int64_t _reported_file_read_bytes = 0;
 };
 
 } // namespace doris

be/src/format_v2/file_reader.cpp

@@ -94,6 +94,12 @@ Status FileReader::init(RuntimeState* state) {
     ++_reader_statistics.open_file_num;
     io::FileReaderOptions reader_options =
             FileFactory::get_reader_options(state->query_options(), *_file_description);
+    // Parquet currently supplies the planned range reuse that amortizes a promoted block. Other
+    // V2 formats keep their existing buffering path until they opt in with equivalent semantics.
+    reader_options.reader_local_cache = _supports_reader_local_cache() && _io_ctx != nullptr
+                                                ? _io_ctx->reader_local_cache
+                                                : nullptr;
+    reader_options.enable_reader_local_cache = reader_options.reader_local_cache != nullptr;
     _file_reader = DORIS_TRY(io::DelegateReader::create_file_reader(
             _profile, *_system_properties, *_file_description, reader_options,
             io::DelegateReader::AccessMode::RANDOM, _io_ctx));

be/src/format_v2/file_reader.h

@@ -419,6 +419,7 @@ class FileReader {
 
 protected:
     virtual void _init_profile() {}
+    virtual bool _supports_reader_local_cache() const { return false; }
     void _record_scan_rows(int64_t rows) {
         DORIS_CHECK(rows >= 0);
         _reader_statistics.read_rows += rows;

be/src/format_v2/parquet/parquet_file_context.cpp

@@ -556,9 +556,18 @@ void ParquetFileContext::prefetch_ranges(const std::vector<ParquetPageCacheRange
     }
 }
 
+bool ParquetFileContext::native_file_should_defer_merge_ranges() const {
+    io::FileReaderSPtr reader = native_file;
+    if (auto tracing_reader = std::dynamic_pointer_cast<io::TracingFileReader>(reader)) {
+        reader = tracing_reader->inner_reader();
+    }
+    return dynamic_cast<io::ExactCacheReader*>(reader.get()) != nullptr ||
+           reader->get_data_dir_path() == io::FileReader::VIRTUAL_REMOTE_DATA_DIR;
+}
+
 bool ParquetFileContext::set_native_random_access_ranges(
         const std::vector<ParquetPageCacheRange>& ranges, size_t avg_io_size,
-        RuntimeProfile* profile, int64_t merge_read_slice_size) {
+        RuntimeProfile* profile, int64_t merge_read_slice_size, bool expose_ranges_immediately) {
     DORIS_CHECK(native_file != nullptr);
     if (!detail::should_use_merge_range_reader(
                 ranges, avg_io_size,
@@ -576,7 +585,9 @@ bool ParquetFileContext::set_native_random_access_ranges(
     }
     std::ranges::sort(native_ranges, {}, &io::PrefetchRange::start_offset);
     native_row_group_file = std::make_shared<io::MergeRangeFileReader>(
-            profile, native_file, native_ranges, merge_read_slice_size);
+            profile, native_file,
+            expose_ranges_immediately ? native_ranges : std::vector<io::PrefetchRange> {},
+            merge_read_slice_size);
     return true;
 }
 

be/src/format_v2/parquet/parquet_file_context.h

@@ -164,7 +164,9 @@ struct ParquetFileContext {
     // sequential projected chunk ranges consumed by MergeRangeFileReader.
     bool set_native_random_access_ranges(const std::vector<ParquetPageCacheRange>& ranges,
                                          size_t avg_io_size, RuntimeProfile* profile,
-                                         int64_t merge_read_slice_size);
+                                         int64_t merge_read_slice_size,
+                                         bool expose_ranges_immediately = true);
+    bool native_file_should_defer_merge_ranges() const;
     const io::FileReaderSPtr& native_data_file() const {
         return native_row_group_file != nullptr ? native_row_group_file : native_file;
     }

be/src/format_v2/parquet/parquet_reader.h

@@ -86,6 +86,7 @@ class ParquetReader : public format::FileReader {
 
 protected:
     void _init_profile() override;
+    bool _supports_reader_local_cache() const override { return true; }
 
 private:
     void _sync_page_cache_profile();