This is an automated email from the ASF dual-hosted git repository. Gabriel39 pushed a commit to branch codex/fix-file-scanner-v2-remote-scan-bytes in repository https://gitbox.apache.org/repos/asf/doris.git
commit 01eea212c897aedbb57be5a43ddb20d4b17a74d6 Author: Gabriel <[email protected]> AuthorDate: Sun Jul 5 09:35:35 2026 +0800 [fix](be) Publish FileScannerV2 remote scan bytes ### What problem does this PR solve? Issue Number: None Related PR: #65046 Problem Summary: FileScannerV2 only refreshed profile file-read counters in update_realtime_counters(), so the query IOContext counters stayed at zero for external file scans. Workload policies using be_scan_bytes_from_remote_storage could therefore miss remote Hive/Parquet reads and fail to cancel queries. This change publishes FileScannerV2 scan rows, scan bytes, local bytes, and remote bytes to the query IOContext using deltas from cumulative reader/cache stats, and falls back to t [...] ### Release note None ### Check List (For Author) - Test: Unit Test / Manual test - Ran build-support/clang-format.sh - Ran targeted run_clang_format.py check for modified BE files - Ran git diff --check - Attempted ./run-be-ut.sh --run --filter='FileScannerV2Test.*' (blocked by missing Snappy in thirdparty/installed in this worktree) - Behavior changed: No - Does this need documentation: No --- be/src/exec/scan/file_scanner_v2.cpp | 73 +++++++++++++++++++++++++++++- be/src/exec/scan/file_scanner_v2.h | 21 +++++++++ be/test/exec/scan/file_scanner_v2_test.cpp | 70 ++++++++++++++++++++++++++++ 3 files changed, 163 insertions(+), 1 deletion(-) diff --git a/be/src/exec/scan/file_scanner_v2.cpp b/be/src/exec/scan/file_scanner_v2.cpp index 57791cb2d85..36e40c6e78a 100644 --- a/be/src/exec/scan/file_scanner_v2.cpp +++ b/be/src/exec/scan/file_scanner_v2.cpp @@ -30,6 +30,7 @@ #include "common/cast_set.h" #include "common/config.h" #include "common/consts.h" +#include "common/metrics/doris_metrics.h" #include "common/status.h" #include "core/assert_cast.h" #include "core/block/column_with_type_and_name.h" @@ -230,6 +231,16 @@ Status FileScannerV2::TEST_rewrite_slot_refs_to_global_index( const std::unordered_map<int32_t, format::GlobalIndex>& slot_id_to_global_index) { return rewrite_slot_refs_to_global_index(expr, slot_id_to_global_index); } + +FileScannerV2::RealtimeCounterDeltas FileScannerV2::TEST_collect_realtime_counter_deltas( + const io::FileReaderStats& file_reader_stats, + const io::FileCacheStatistics& file_cache_statistics, int64_t* last_read_bytes, + int64_t* last_read_rows, int64_t* last_bytes_read_from_local, + int64_t* last_bytes_read_from_remote) { + return _collect_realtime_counter_deltas( + file_reader_stats, file_cache_statistics, last_read_bytes, last_read_rows, + last_bytes_read_from_local, last_bytes_read_from_remote); +} #endif bool FileScannerV2::is_supported(const TFileScanRangeParams& params, const TFileRangeDesc& range) { @@ -770,10 +781,70 @@ void FileScannerV2::update_realtime_counters() { if (_file_reader_stats == nullptr) { return; } - const int64_t bytes_read = _file_reader_stats->read_bytes; + DORIS_CHECK(_file_cache_statistics != nullptr); + const int64_t bytes_read = cast_set<int64_t>(_file_reader_stats->read_bytes); + auto* local_state = static_cast<FileScanLocalState*>(_local_state); + const auto deltas = _collect_realtime_counter_deltas( + *_file_reader_stats, *_file_cache_statistics, &_last_read_bytes, &_last_read_rows, + &_last_bytes_read_from_local, &_last_bytes_read_from_remote); + + COUNTER_UPDATE(local_state->_scan_bytes, deltas.scan_bytes); + COUNTER_UPDATE(local_state->_scan_rows, deltas.scan_rows); + + _state->get_query_ctx()->resource_ctx()->io_context()->update_scan_rows(deltas.scan_rows); + _state->get_query_ctx()->resource_ctx()->io_context()->update_scan_bytes(deltas.scan_bytes); + _state->get_query_ctx()->resource_ctx()->io_context()->update_scan_bytes_from_local_storage( + deltas.scan_bytes_from_local_storage); + _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); 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)); + + DorisMetrics::instance()->query_scan_bytes->increment(deltas.scan_bytes); + DorisMetrics::instance()->query_scan_rows->increment(deltas.scan_rows); + DorisMetrics::instance()->query_scan_bytes_from_local->increment( + deltas.scan_bytes_from_local_storage); + DorisMetrics::instance()->query_scan_bytes_from_remote->increment( + deltas.scan_bytes_from_remote_storage); +} + +FileScannerV2::RealtimeCounterDeltas FileScannerV2::_collect_realtime_counter_deltas( + const io::FileReaderStats& file_reader_stats, + const io::FileCacheStatistics& file_cache_statistics, int64_t* last_read_bytes, + int64_t* last_read_rows, int64_t* last_bytes_read_from_local, + int64_t* last_bytes_read_from_remote) { + DORIS_CHECK(last_read_bytes != nullptr); + DORIS_CHECK(last_read_rows != nullptr); + DORIS_CHECK(last_bytes_read_from_local != nullptr); + DORIS_CHECK(last_bytes_read_from_remote != nullptr); + + const int64_t read_bytes = cast_set<int64_t>(file_reader_stats.read_bytes); + const int64_t read_rows = cast_set<int64_t>(file_reader_stats.read_rows); + const int64_t bytes_read_from_local = file_cache_statistics.bytes_read_from_local; + const int64_t bytes_read_from_remote = file_cache_statistics.bytes_read_from_remote; + DORIS_CHECK(read_bytes >= *last_read_bytes); + DORIS_CHECK(read_rows >= *last_read_rows); + DORIS_CHECK(bytes_read_from_local >= *last_bytes_read_from_local); + DORIS_CHECK(bytes_read_from_remote >= *last_bytes_read_from_remote); + + RealtimeCounterDeltas deltas; + deltas.scan_rows = read_rows - *last_read_rows; + deltas.scan_bytes = read_bytes - *last_read_bytes; + if (bytes_read_from_local == 0 && bytes_read_from_remote == 0) { + deltas.scan_bytes_from_remote_storage = deltas.scan_bytes; + } else { + deltas.scan_bytes_from_local_storage = bytes_read_from_local - *last_bytes_read_from_local; + deltas.scan_bytes_from_remote_storage = + bytes_read_from_remote - *last_bytes_read_from_remote; + } + + *last_read_bytes = read_bytes; + *last_read_rows = read_rows; + *last_bytes_read_from_local = bytes_read_from_local; + *last_bytes_read_from_remote = bytes_read_from_remote; + return deltas; } void FileScannerV2::_collect_profile_before_close() { diff --git a/be/src/exec/scan/file_scanner_v2.h b/be/src/exec/scan/file_scanner_v2.h index bc493bfbd85..13d890e5a1f 100644 --- a/be/src/exec/scan/file_scanner_v2.h +++ b/be/src/exec/scan/file_scanner_v2.h @@ -54,6 +54,13 @@ public: static constexpr const char* NAME = "FileScannerV2"; static constexpr size_t ADAPTIVE_BATCH_INITIAL_PROBE_ROWS = 32; + struct RealtimeCounterDeltas { + int64_t scan_rows = 0; + int64_t scan_bytes = 0; + int64_t scan_bytes_from_local_storage = 0; + int64_t scan_bytes_from_remote_storage = 0; + }; + static bool is_supported(const TFileScanRangeParams& params, const TFileRangeDesc& range); #ifdef BE_TEST static Status TEST_to_file_format(TFileFormatType::type format_type, @@ -65,6 +72,11 @@ public: static Status TEST_rewrite_slot_refs_to_global_index( VExprSPtr* expr, const std::unordered_map<int32_t, format::GlobalIndex>& slot_id_to_global_index); + static RealtimeCounterDeltas TEST_collect_realtime_counter_deltas( + const io::FileReaderStats& file_reader_stats, + const io::FileCacheStatistics& file_cache_statistics, int64_t* last_read_bytes, + int64_t* last_read_rows, int64_t* last_bytes_read_from_local, + int64_t* last_bytes_read_from_remote); #endif FileScannerV2(RuntimeState* state, FileScanLocalState* parent, int64_t limit, @@ -114,6 +126,11 @@ private: bool _should_run_adaptive_batch_size() const; size_t _predict_reader_batch_rows(); void _update_adaptive_batch_size(const Block& block); + static RealtimeCounterDeltas _collect_realtime_counter_deltas( + const io::FileReaderStats& file_reader_stats, + const io::FileCacheStatistics& file_cache_statistics, int64_t* last_read_bytes, + int64_t* last_read_rows, int64_t* last_bytes_read_from_local, + int64_t* last_bytes_read_from_remote); void _report_file_reader_predicate_filtered_rows(); void _report_condition_cache_profile(); @@ -157,6 +174,10 @@ private: int64_t _reported_predicate_filtered_rows = 0; int64_t _reported_condition_cache_hit_count = 0; int64_t _reported_condition_cache_filtered_rows = 0; + int64_t _last_read_bytes = 0; + int64_t _last_read_rows = 0; + int64_t _last_bytes_read_from_local = 0; + int64_t _last_bytes_read_from_remote = 0; }; } // namespace doris diff --git a/be/test/exec/scan/file_scanner_v2_test.cpp b/be/test/exec/scan/file_scanner_v2_test.cpp index 436a18c66de..9bcc83b3a9b 100644 --- a/be/test/exec/scan/file_scanner_v2_test.cpp +++ b/be/test/exec/scan/file_scanner_v2_test.cpp @@ -222,6 +222,76 @@ TEST(FileScannerV2Test, FileFormatConversionMatrix) { } } +TEST(FileScannerV2Test, RealtimeCounterDeltasUseReaderBytesAsRemoteWithoutCacheStats) { + io::FileReaderStats file_reader_stats; + io::FileCacheStatistics file_cache_statistics; + int64_t last_read_bytes = 0; + int64_t last_read_rows = 0; + int64_t last_bytes_read_from_local = 0; + int64_t last_bytes_read_from_remote = 0; + + file_reader_stats.read_bytes = 100; + file_reader_stats.read_rows = 7; + auto deltas = FileScannerV2::TEST_collect_realtime_counter_deltas( + file_reader_stats, file_cache_statistics, &last_read_bytes, &last_read_rows, + &last_bytes_read_from_local, &last_bytes_read_from_remote); + EXPECT_EQ(7, deltas.scan_rows); + EXPECT_EQ(100, deltas.scan_bytes); + EXPECT_EQ(0, deltas.scan_bytes_from_local_storage); + EXPECT_EQ(100, deltas.scan_bytes_from_remote_storage); + + deltas = FileScannerV2::TEST_collect_realtime_counter_deltas( + file_reader_stats, file_cache_statistics, &last_read_bytes, &last_read_rows, + &last_bytes_read_from_local, &last_bytes_read_from_remote); + EXPECT_EQ(0, deltas.scan_rows); + EXPECT_EQ(0, deltas.scan_bytes); + EXPECT_EQ(0, deltas.scan_bytes_from_local_storage); + EXPECT_EQ(0, deltas.scan_bytes_from_remote_storage); + + file_reader_stats.read_bytes = 160; + file_reader_stats.read_rows = 9; + deltas = FileScannerV2::TEST_collect_realtime_counter_deltas( + file_reader_stats, file_cache_statistics, &last_read_bytes, &last_read_rows, + &last_bytes_read_from_local, &last_bytes_read_from_remote); + EXPECT_EQ(2, deltas.scan_rows); + EXPECT_EQ(60, deltas.scan_bytes); + EXPECT_EQ(0, deltas.scan_bytes_from_local_storage); + EXPECT_EQ(60, deltas.scan_bytes_from_remote_storage); +} + +TEST(FileScannerV2Test, RealtimeCounterDeltasUseFileCacheDeltasWhenAvailable) { + io::FileReaderStats file_reader_stats; + io::FileCacheStatistics file_cache_statistics; + int64_t last_read_bytes = 0; + int64_t last_read_rows = 0; + int64_t last_bytes_read_from_local = 0; + int64_t last_bytes_read_from_remote = 0; + + file_reader_stats.read_bytes = 100; + file_reader_stats.read_rows = 7; + file_cache_statistics.bytes_read_from_local = 30; + file_cache_statistics.bytes_read_from_remote = 70; + auto deltas = FileScannerV2::TEST_collect_realtime_counter_deltas( + file_reader_stats, file_cache_statistics, &last_read_bytes, &last_read_rows, + &last_bytes_read_from_local, &last_bytes_read_from_remote); + EXPECT_EQ(7, deltas.scan_rows); + EXPECT_EQ(100, deltas.scan_bytes); + EXPECT_EQ(30, deltas.scan_bytes_from_local_storage); + EXPECT_EQ(70, deltas.scan_bytes_from_remote_storage); + + file_reader_stats.read_bytes = 125; + file_reader_stats.read_rows = 10; + file_cache_statistics.bytes_read_from_local = 35; + file_cache_statistics.bytes_read_from_remote = 90; + deltas = FileScannerV2::TEST_collect_realtime_counter_deltas( + file_reader_stats, file_cache_statistics, &last_read_bytes, &last_read_rows, + &last_bytes_read_from_local, &last_bytes_read_from_remote); + EXPECT_EQ(3, deltas.scan_rows); + EXPECT_EQ(25, deltas.scan_bytes); + EXPECT_EQ(5, deltas.scan_bytes_from_local_storage); + EXPECT_EQ(20, deltas.scan_bytes_from_remote_storage); +} + // Scenario: partition slots are identified from the explicit FE category when present, otherwise // from the legacy is_file_slot flag. Scanner-generated rowid columns must never be treated as // partition columns even if FE marks them as non-file slots. --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
