github-actions[bot] commented on code in PR #65218:
URL: https://github.com/apache/doris/pull/65218#discussion_r3526389887
##########
be/src/exec/scan/file_scanner_v2.cpp:
##########
@@ -329,6 +340,7 @@ Status FileScannerV2::_get_block_impl(RuntimeState* state,
Block* block, bool* e
continue;
}
_update_adaptive_batch_size(*block);
+ _read_rows += cast_set<int64_t>(block->rows());
Review Comment:
This replacement row source still undercounts `ScanRows` for pushed-down V2
filters. `block->rows()` here is the number of rows `TableReader` returns after
file-reader predicates have already run; in the V2 Parquet path
`read_current_row_group_batch()` records `RawRowsRead` from `batch_rows`,
applies conjunct/delete filtering, and then returns only `selected_rows`, while
`read_next_batch()` advances the physical row cursor by the original
`batch_rows` even when every row was filtered. Existing scan-row accounting
uses the raw/candidate rows read (`vparquet_reader` adds
`candidate_row_ranges.count()` to `file_reader_stats->read_rows`, and Olap
publishes `stats.raw_rows_read`). With this source, a scan that reads many rows
but filters them all in the file reader can publish zero scan rows to
IOContext/workload policy metrics. Please feed the raw V2 reader/scheduler row
count into the realtime delta path, keeping emitted rows separate from scan-row
accounting.
##########
be/src/exec/scan/file_scanner_v2.cpp:
##########
@@ -770,10 +782,75 @@ 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, _read_rows,
&_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
read_rows,
+ 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 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;
+ // Peer cache is a known cache source, but it is not remote object storage.
+ const bool has_cache_source_stats =
file_cache_statistics.num_local_io_total != 0 ||
+
file_cache_statistics.num_remote_io_total != 0 ||
+
file_cache_statistics.num_peer_io_total != 0 ||
+ bytes_read_from_local != 0 ||
bytes_read_from_remote != 0 ||
+
file_cache_statistics.bytes_read_from_peer != 0;
+ if (!has_cache_source_stats) {
Review Comment:
This fallback still treats local-file scans as remote object storage.
`FileScannerV2::is_supported()` accepts `NotSet`/`tvf` ranges for
Parquet/CSV/text/JSON/native, and local TVFs set `TFileType::FILE_LOCAL`; that
flows through `TableReader::create_system_properties()` into `FileFactory`,
where `FILE_LOCAL` opens a plain local reader and never populates
`FileCacheStatistics`. `TracingFileReader` still increments `read_bytes`, so
`has_cache_source_stats` remains false and the branch below charges all local
bytes to `scan_bytes_from_remote_storage`. A query over `local(...)` files can
therefore satisfy a workload policy on remote-storage bytes even though it only
read local disk. Please distinguish the scan's file type when there are no
cache stats, and only fall back to remote for remote file systems that truly
lack cache-source counters.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
