This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 2eaf314fc0d [fix](be) Rework FileScannerV2 scan accounting and stop
handling (#65218)
2eaf314fc0d is described below
commit 2eaf314fc0d2ad81995271da806222f6ae4ee35c
Author: Gabriel <[email protected]>
AuthorDate: Tue Jul 7 14:22:50 2026 +0800
[fix](be) Rework FileScannerV2 scan accounting and stop handling (#65218)
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 stop queries. This PR publishes
FileScannerV2 scan rows, scan bytes, local bytes, and remote bytes to
the query IOContext and Doris metrics using deltas from cumulative
reader/cache stats.
The row-accounting implementation was reworked after review. The old
approach tried to synchronize concrete `FileReader` private row
statistics from `TableReader` after `get_block()` or `close()`. That
abstraction was wrong because `FORMAT_JNI` overrides
`TableReader::get_block()`, materialized readers could expose
post-filter rows, and Parquet `COUNT(complex_col)` aggregate reads
nested levels without going through the normal block row path.
The current implementation records scan rows at the point each reader
actually reads or materializes source rows, before file-local filters:
- Parquet normal scan records the scheduler raw-row delta.
- Parquet `COUNT(complex_col)` records the selected batch rows loaded
through nested levels.
- CSV/TEXT records `rows_before_filter`.
- JSON, Native, and RemoteDoris record rows after materializing the file
block and before applying filters.
- JNI table readers record the `current_rows` returned by the Java
scanner before `finalize_jni_block()`.
This PR also handles `io_ctx->should_stop` in the FileScannerV2 reader
path. `TableReader` short-circuits when the scanner stop flag is set,
stop-time reader init/open EOF is converted to normal EOS, the Parquet
v2 reader / Arrow `RandomAccessFile` path checks the stop flag before
file size/read operations, and Parquet aggregate pushdown converts
stop-time nested-level read EOF into normal scanner EOS instead of a
scan failure.
---
be/src/exec/scan/file_scanner_v2.cpp | 115 +++++++++++++++++++-
be/src/exec/scan/file_scanner_v2.h | 26 +++++
.../delimited_text/delimited_text_reader.cpp | 4 +-
be/src/format_v2/file_reader.h | 8 ++
be/src/format_v2/jni/jni_table_reader.cpp | 1 +
be/src/format_v2/json/json_reader.cpp | 2 +-
be/src/format_v2/native/native_reader.cpp | 2 +-
be/src/format_v2/parquet/parquet_file_context.cpp | 16 +++
be/src/format_v2/parquet/parquet_reader.cpp | 52 ++++++++-
be/src/format_v2/parquet/parquet_reader.h | 2 +
be/src/format_v2/parquet/parquet_scan.cpp | 2 +
be/src/format_v2/parquet/parquet_scan.h | 2 +
be/src/format_v2/table/remote_doris_reader.cpp | 2 +-
be/src/format_v2/table_reader.cpp | 20 +++-
be/src/format_v2/table_reader.h | 21 +++-
be/test/exec/scan/file_scanner_v2_test.cpp | 117 +++++++++++++++++++++
.../format_v2/delimited_text/csv_reader_test.cpp | 3 +
be/test/format_v2/parquet/parquet_scan_test.cpp | 47 ++++++++-
be/test/format_v2/table_reader_test.cpp | 108 ++++++++++++++++++-
19 files changed, 533 insertions(+), 17 deletions(-)
diff --git a/be/src/exec/scan/file_scanner_v2.cpp
b/be/src/exec/scan/file_scanner_v2.cpp
index e3c42679a29..a65f9a74d6a 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,18 @@ 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,
+ UncachedReaderBytesStorage uncached_reader_bytes_storage, 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,
+ uncached_reader_bytes_storage,
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) {
@@ -748,10 +761,110 @@ 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 file_type =
+ _current_range.__isset.file_type
+ ? _current_range.file_type
+ : (_params != nullptr && _params->__isset.file_type ?
_params->file_type
+ :
TFileType::FILE_LOCAL);
+ const auto deltas = _collect_realtime_counter_deltas(
+ *_file_reader_stats, *_file_cache_statistics,
_uncached_reader_bytes_storage(file_type),
+ &_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,
+ UncachedReaderBytesStorage uncached_reader_bytes_storage, 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;
+ // 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) {
+ switch (uncached_reader_bytes_storage) {
+ case UncachedReaderBytesStorage::LOCAL:
+ deltas.scan_bytes_from_local_storage = deltas.scan_bytes;
+ break;
+ case UncachedReaderBytesStorage::REMOTE:
+ deltas.scan_bytes_from_remote_storage = deltas.scan_bytes;
+ break;
+ case UncachedReaderBytesStorage::NONE:
+ break;
+ }
+ } 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;
+}
+
+FileScannerV2::UncachedReaderBytesStorage
FileScannerV2::_uncached_reader_bytes_storage(
+ TFileType::type file_type) {
+ switch (file_type) {
+ case TFileType::FILE_LOCAL:
+ return UncachedReaderBytesStorage::LOCAL;
+ case TFileType::FILE_STREAM:
+ return UncachedReaderBytesStorage::NONE;
+ case TFileType::FILE_BROKER:
+ case TFileType::FILE_S3:
+ case TFileType::FILE_HDFS:
+ case TFileType::FILE_NET:
+ case TFileType::FILE_HTTP:
+ return UncachedReaderBytesStorage::REMOTE;
+ }
+ DORIS_CHECK(false) << "unknown file type: " << file_type;
+ return UncachedReaderBytesStorage::NONE;
}
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 7764d095885..a911d42c3ad 100644
--- a/be/src/exec/scan/file_scanner_v2.h
+++ b/be/src/exec/scan/file_scanner_v2.h
@@ -54,6 +54,15 @@ 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;
+ };
+
+ enum class UncachedReaderBytesStorage { LOCAL, REMOTE, NONE };
+
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 +74,12 @@ 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,
+ UncachedReaderBytesStorage uncached_reader_bytes_storage, 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,
@@ -113,6 +128,13 @@ 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,
+ UncachedReaderBytesStorage uncached_reader_bytes_storage, int64_t*
last_read_bytes,
+ int64_t* last_read_rows, int64_t* last_bytes_read_from_local,
+ int64_t* last_bytes_read_from_remote);
+ static UncachedReaderBytesStorage
_uncached_reader_bytes_storage(TFileType::type file_type);
void _report_file_reader_predicate_filtered_rows();
void _report_condition_cache_profile();
@@ -156,6 +178,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/src/format_v2/delimited_text/delimited_text_reader.cpp
b/be/src/format_v2/delimited_text/delimited_text_reader.cpp
index f6e84b4aa77..b26e0d44456 100644
--- a/be/src/format_v2/delimited_text/delimited_text_reader.cpp
+++ b/be/src/format_v2/delimited_text/delimited_text_reader.cpp
@@ -340,6 +340,7 @@ Status DelimitedTextReader::get_block(Block* file_block,
size_t* rows, bool* eof
const size_t rows_before_filter = *rows;
update_counter(_text_profile.rows_read_before_filter, rows_before_filter);
+ _record_scan_rows(cast_set<int64_t>(rows_before_filter));
MaterializedReaderFilterProfile filter_profile;
filter_profile.delete_conjunct_filter_time =
_text_profile.delete_conjunct_filter_time;
@@ -350,7 +351,6 @@ Status DelimitedTextReader::get_block(Block* file_block,
size_t* rows, bool* eof
RETURN_IF_ERROR(apply_materialized_reader_filters(_request.get(),
_io_ctx.get(), file_block,
rows, &filter_profile));
update_counter(_text_profile.rows_returned, *rows);
- _reader_statistics.read_rows += *rows;
*eof = _line_reader_eof && *rows == 0;
_eof = *eof;
return Status::OK();
@@ -390,7 +390,7 @@ Status DelimitedTextReader::get_aggregate_result(const
FileAggregateRequest& req
result->columns.clear();
update_counter(_text_profile.rows_read_before_filter, count);
update_counter(_text_profile.rows_returned, count);
- _reader_statistics.read_rows += count;
+ _record_scan_rows(count);
_eof = true;
return Status::OK();
}
diff --git a/be/src/format_v2/file_reader.h b/be/src/format_v2/file_reader.h
index 57cfe3aa5ea..135df282014 100644
--- a/be/src/format_v2/file_reader.h
+++ b/be/src/format_v2/file_reader.h
@@ -24,6 +24,7 @@
#include <utility>
#include <vector>
+#include "common/cast_set.h"
#include "common/status.h"
#include "core/data_type/data_type.h"
#include "core/field.h"
@@ -319,6 +320,13 @@ public:
protected:
virtual void _init_profile() {}
+ void _record_scan_rows(int64_t rows) {
+ DORIS_CHECK(rows >= 0);
+ _reader_statistics.read_rows += rows;
+ if (_io_ctx != nullptr && _io_ctx->file_reader_stats != nullptr) {
+ _io_ctx->file_reader_stats->read_rows += cast_set<size_t>(rows);
+ }
+ }
io::FileReaderSPtr _file_reader;
// _tracing_file_reader wraps _file_reader.
diff --git a/be/src/format_v2/jni/jni_table_reader.cpp
b/be/src/format_v2/jni/jni_table_reader.cpp
index d43a22e632b..01c2dd6ecd0 100644
--- a/be/src/format_v2/jni/jni_table_reader.cpp
+++ b/be/src/format_v2/jni/jni_table_reader.cpp
@@ -83,6 +83,7 @@ Status JniTableReader::get_block(Block* output_block, bool*
eos) {
return Status::OK();
}
+ _record_scan_rows(current_rows);
RETURN_IF_ERROR(finalize_jni_block(&_jni_block_template, output_block,
¤t_rows));
if (current_rows == 0) {
output_block->clear_column_data(_projected_columns.size());
diff --git a/be/src/format_v2/json/json_reader.cpp
b/be/src/format_v2/json/json_reader.cpp
index f0219bb7d85..313115e82ce 100644
--- a/be/src/format_v2/json/json_reader.cpp
+++ b/be/src/format_v2/json/json_reader.cpp
@@ -323,8 +323,8 @@ Status JsonReader::get_block(Block* file_block, size_t*
rows, bool* eof) {
}
*rows = file_block->rows();
+ _record_scan_rows(cast_set<int64_t>(*rows));
RETURN_IF_ERROR(_apply_filters(file_block, rows));
- _reader_statistics.read_rows += *rows;
*eof = _reader_eof && *rows == 0;
_eof = *eof;
return Status::OK();
diff --git a/be/src/format_v2/native/native_reader.cpp
b/be/src/format_v2/native/native_reader.cpp
index 2a0a89f80ad..0e348f3de2c 100644
--- a/be/src/format_v2/native/native_reader.cpp
+++ b/be/src/format_v2/native/native_reader.cpp
@@ -134,8 +134,8 @@ Status NativeReader::get_block(Block* file_block, size_t*
rows, bool* eof) {
RETURN_IF_ERROR(source_block.deserialize(pblock, &uncompressed_bytes,
&decompress_time));
RETURN_IF_ERROR(_materialize_requested_columns(source_block, file_block));
*rows = file_block->rows();
+ _record_scan_rows(cast_set<int64_t>(*rows));
RETURN_IF_ERROR(_apply_filters(file_block, rows));
- _reader_statistics.read_rows += *rows;
if (_first_block_loaded && !_first_block_consumed) {
_first_block_consumed = true;
diff --git a/be/src/format_v2/parquet/parquet_file_context.cpp
b/be/src/format_v2/parquet/parquet_file_context.cpp
index dac09747408..1df3d7b353d 100644
--- a/be/src/format_v2/parquet/parquet_file_context.cpp
+++ b/be/src/format_v2/parquet/parquet_file_context.cpp
@@ -26,6 +26,7 @@
#include <exception>
#include <limits>
#include <mutex>
+#include <string_view>
#include <unordered_map>
#include <utility>
@@ -36,6 +37,7 @@
#include "io/fs/buffered_reader.h"
#include "io/fs/file_reader.h"
#include "io/fs/tracing_file_reader.h"
+#include "io/io_common.h"
#include "storage/cache/page_cache.h"
#include "util/slice.h"
@@ -245,6 +247,9 @@ public:
if (!_file_reader) {
return arrow::Status::IOError("Doris file reader is not open");
}
+ if (_io_ctx != nullptr && _io_ctx->should_stop) {
+ return arrow::Status::IOError("stop");
+ }
return static_cast<int64_t>(_file_reader->size());
}
@@ -266,6 +271,9 @@ public:
if (!_file_reader) {
return arrow::Status::IOError("Doris file reader is not open");
}
+ if (_io_ctx != nullptr && _io_ctx->should_stop) {
+ return arrow::Status::IOError("stop");
+ }
if (position < 0 || nbytes < 0) {
return arrow::Status::Invalid("negative read position or length");
}
@@ -537,8 +545,16 @@ Status ParquetFileContext::open(io::FileReaderSPtr
input_file_reader, io::IOCont
metadata = this->file_reader->metadata();
schema = metadata != nullptr ? metadata->schema() : nullptr;
} catch (const ::parquet::ParquetException& e) {
+ if (io_ctx != nullptr && io_ctx->should_stop &&
+ std::string_view(e.what()).find("stop") != std::string_view::npos)
{
+ return Status::EndOfFile("stop");
+ }
return Status::Corruption("Failed to open parquet file: {}", e.what());
} catch (const std::exception& e) {
+ if (io_ctx != nullptr && io_ctx->should_stop &&
+ std::string_view(e.what()).find("stop") != std::string_view::npos)
{
+ return Status::EndOfFile("stop");
+ }
return Status::InternalError("Failed to open parquet file: {}",
e.what());
}
diff --git a/be/src/format_v2/parquet/parquet_reader.cpp
b/be/src/format_v2/parquet/parquet_reader.cpp
index 9c3363ae41a..82ad3f78891 100644
--- a/be/src/format_v2/parquet/parquet_reader.cpp
+++ b/be/src/format_v2/parquet/parquet_reader.cpp
@@ -26,6 +26,7 @@
#include <utility>
#include <vector>
+#include "common/cast_set.h"
#include "core/assert_cast.h"
#include "core/block/block.h"
#include "core/data_type/data_type_array.h"
@@ -39,6 +40,7 @@
#include "format_v2/parquet/parquet_scan.h"
#include "format_v2/parquet/parquet_statistics.h"
#include "format_v2/parquet/reader/column_reader.h"
+#include "io/io_common.h"
#include "runtime/runtime_state.h"
namespace doris::format::parquet {
@@ -318,6 +320,9 @@
ParquetReader::ParquetReader(std::shared_ptr<io::FileSystemProperties>& system_p
ParquetReader::~ParquetReader() = default;
Status ParquetReader::init(RuntimeState* state) {
+ if (_io_ctx != nullptr && _io_ctx->should_stop) {
+ return Status::EndOfFile("stop");
+ }
RETURN_IF_ERROR(format::FileReader::init(state));
if (_profile != nullptr) {
COUNTER_UPDATE(_parquet_profile.file_reader_create_time,
@@ -466,6 +471,10 @@ Status ParquetReader::get_block(Block* file_block, size_t*
rows, bool* eof) {
return Status::Uninitialized("ParquetReader is not open");
}
*rows = 0;
+ if (_io_ctx != nullptr && _io_ctx->should_stop) {
+ *eof = true;
+ return Status::OK();
+ }
if (_eof) {
*eof = true;
return Status::OK();
@@ -476,17 +485,40 @@ Status ParquetReader::get_block(Block* file_block,
size_t* rows, bool* eof) {
}
const auto predicate_filtered_rows_before =
_state->scheduler.predicate_filtered_rows();
- RETURN_IF_ERROR(_state->scheduler.read_next_batch(_state->file_context,
_state->file_schema,
- *request_snapshot,
file_block, rows, eof));
+ const auto raw_rows_read_before = _state->scheduler.raw_rows_read();
+ Status st = _state->scheduler.read_next_batch(_state->file_context,
_state->file_schema,
+ *request_snapshot,
file_block, rows, eof);
+ if (!st.ok()) {
+ if (_io_ctx != nullptr && _io_ctx->should_stop) {
+ *rows = 0;
+ *eof = true;
+ return Status::OK();
+ }
+ return st;
+ }
_sync_page_cache_profile();
if (_io_ctx != nullptr) {
_io_ctx->predicate_filtered_rows +=
_state->scheduler.predicate_filtered_rows() -
predicate_filtered_rows_before;
}
+ const auto raw_rows_read = _state->scheduler.raw_rows_read();
+ DORIS_CHECK(raw_rows_read >= raw_rows_read_before);
+ _record_scan_rows(raw_rows_read - raw_rows_read_before);
_eof = *eof;
return Status::OK();
}
+bool ParquetReader::_should_stop() const {
+ return _io_ctx != nullptr && _io_ctx->should_stop;
+}
+
+Status ParquetReader::_stop_status_if_requested(const Status& status) const {
+ if (!status.ok() && _should_stop()) {
+ return Status::EndOfFile("stop");
+ }
+ return status;
+}
+
void ParquetReader::_sync_page_cache_profile() {
if (_profile == nullptr || _state == nullptr) {
return;
@@ -539,6 +571,9 @@ Status ParquetReader::get_aggregate_result(const
format::FileAggregateRequest& r
_state->file_context.schema == nullptr) {
return Status::Uninitialized("ParquetReader is not open");
}
+ if (_should_stop()) {
+ return Status::EndOfFile("stop");
+ }
result->count = 0;
result->columns.clear();
if (request.agg_type != TPushAggOp::type::COUNT &&
@@ -569,9 +604,15 @@ Status ParquetReader::get_aggregate_result(const
format::FileAggregateRequest& r
try {
row_group =
_state->file_context.file_reader->RowGroup(row_group_plan.row_group_id);
} catch (const ::parquet::ParquetException& e) {
+ if (_should_stop()) {
+ return Status::EndOfFile("stop");
+ }
return Status::Corruption("Failed to open parquet row group
{}: {}",
row_group_plan.row_group_id,
e.what());
} catch (const std::exception& e) {
+ if (_should_stop()) {
+ return Status::EndOfFile("stop");
+ }
return Status::InternalError("Failed to open parquet row group
{}: {}",
row_group_plan.row_group_id,
e.what());
}
@@ -589,7 +630,8 @@ Status ParquetReader::get_aggregate_result(const
format::FileAggregateRequest& r
int64_t row_group_cursor = 0;
for (const auto& selected_range : row_group_plan.selected_ranges) {
DORIS_CHECK(selected_range.start >= row_group_cursor);
- RETURN_IF_ERROR(shape_reader->skip(selected_range.start -
row_group_cursor));
+ RETURN_IF_ERROR(_stop_status_if_requested(
+ shape_reader->skip(selected_range.start -
row_group_cursor)));
row_group_cursor = selected_range.start;
int64_t range_rows_read = 0;
@@ -601,7 +643,9 @@ Status ParquetReader::get_aggregate_result(const
format::FileAggregateRequest& r
// or values_column. MAP chooses the key leaf;
ARRAY/STRUCT may choose a string
// leaf, but the levels-only protocol still avoids
Doris-side string
// materialization for that leaf.
-
RETURN_IF_ERROR(shape_reader->load_nested_levels_batch(batch_rows));
+ RETURN_IF_ERROR(_stop_status_if_requested(
+
shape_reader->load_nested_levels_batch(batch_rows)));
+ _record_scan_rows(batch_rows);
result->count +=
count_loaded_non_null_values(root_schema,
*shape_reader, batch_rows);
range_rows_read += batch_rows;
diff --git a/be/src/format_v2/parquet/parquet_reader.h
b/be/src/format_v2/parquet/parquet_reader.h
index ff74b97a26e..6c8e88cc27a 100644
--- a/be/src/format_v2/parquet/parquet_reader.h
+++ b/be/src/format_v2/parquet/parquet_reader.h
@@ -76,6 +76,8 @@ protected:
private:
void _sync_page_cache_profile();
+ bool _should_stop() const;
+ Status _stop_status_if_requested(const Status& status) const;
void _fill_column_definition(const ParquetColumnSchema& column_schema,
format::ColumnDefinition* field) const;
diff --git a/be/src/format_v2/parquet/parquet_scan.cpp
b/be/src/format_v2/parquet/parquet_scan.cpp
index 0e1b546e5c6..86de8be9f92 100644
--- a/be/src/format_v2/parquet/parquet_scan.cpp
+++ b/be/src/format_v2/parquet/parquet_scan.cpp
@@ -454,6 +454,7 @@ void
ParquetScanScheduler::set_condition_cache_context(std::shared_ptr<Condition
void ParquetScanScheduler::reset() {
_next_row_group_plan_idx = 0;
+ _raw_rows_read = 0;
reset_current_row_group();
}
@@ -679,6 +680,7 @@ Status ParquetScanScheduler::read_current_row_group_batch(
if (_scan_profile.raw_rows_read != nullptr) {
COUNTER_UPDATE(_scan_profile.raw_rows_read, batch_rows);
}
+ _raw_rows_read += batch_rows;
if (_current_predicate_columns.empty() &&
_current_non_predicate_columns.empty()) {
*rows = static_cast<size_t>(batch_rows);
if (_scan_profile.selected_rows != nullptr) {
diff --git a/be/src/format_v2/parquet/parquet_scan.h
b/be/src/format_v2/parquet/parquet_scan.h
index e8e12ea3f66..1fb313d4f75 100644
--- a/be/src/format_v2/parquet/parquet_scan.h
+++ b/be/src/format_v2/parquet/parquet_scan.h
@@ -132,6 +132,7 @@ public:
bool empty() const { return _row_group_plans.empty(); }
int64_t condition_cache_filtered_rows() const { return
_condition_cache_filtered_rows; }
int64_t predicate_filtered_rows() const { return _predicate_filtered_rows;
}
+ int64_t raw_rows_read() const { return _raw_rows_read; }
Status read_next_batch(ParquetFileContext& file_context,
const
std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
@@ -201,6 +202,7 @@ private:
std::shared_ptr<ConditionCacheContext> _condition_cache_ctx;
int64_t _condition_cache_filtered_rows = 0;
int64_t _predicate_filtered_rows = 0;
+ int64_t _raw_rows_read = 0;
};
} // namespace doris::format::parquet
diff --git a/be/src/format_v2/table/remote_doris_reader.cpp
b/be/src/format_v2/table/remote_doris_reader.cpp
index 39580fd2561..c67cece6e05 100644
--- a/be/src/format_v2/table/remote_doris_reader.cpp
+++ b/be/src/format_v2/table/remote_doris_reader.cpp
@@ -231,9 +231,9 @@ Status RemoteDorisFileReader::get_block(Block* file_block,
size_t* rows, bool* e
}
RETURN_IF_ERROR(_materialize_record_batch(*batch, file_block, rows));
+ _record_scan_rows(cast_set<int64_t>(*rows));
RETURN_IF_ERROR(
apply_materialized_reader_filters(_request.get(), _io_ctx.get(),
file_block, rows));
- _reader_statistics.read_rows += *rows;
return Status::OK();
}
diff --git a/be/src/format_v2/table_reader.cpp
b/be/src/format_v2/table_reader.cpp
index cee93f5eb30..11317c5fd7b 100644
--- a/be/src/format_v2/table_reader.cpp
+++ b/be/src/format_v2/table_reader.cpp
@@ -672,8 +672,24 @@ Status TableReader::create_next_reader(bool* eos) {
if (_batch_size > 0) {
_data_reader.reader->set_batch_size(_batch_size);
}
- RETURN_IF_ERROR(_data_reader.reader->init(_runtime_state));
- RETURN_IF_ERROR(open_reader());
+ Status st = _data_reader.reader->init(_runtime_state);
+ if (!st.ok()) {
+ if (_io_ctx != nullptr && _io_ctx->should_stop &&
st.is<ErrorCode::END_OF_FILE>()) {
+ *eos = true;
+ _data_reader.reader.reset();
+ return Status::OK();
+ }
+ return st;
+ }
+ st = open_reader();
+ if (!st.ok()) {
+ if (_io_ctx != nullptr && _io_ctx->should_stop &&
st.is<ErrorCode::END_OF_FILE>()) {
+ *eos = true;
+ _data_reader.reader.reset();
+ return Status::OK();
+ }
+ return st;
+ }
if (_data_reader.reader == nullptr) {
*eos = _current_task == nullptr;
return Status::OK();
diff --git a/be/src/format_v2/table_reader.h b/be/src/format_v2/table_reader.h
index 4f117632277..66b7eddf720 100644
--- a/be/src/format_v2/table_reader.h
+++ b/be/src/format_v2/table_reader.h
@@ -62,6 +62,7 @@
#include "format_v2/parquet/reader/column_reader.h"
#include "format_v2/schema_projection.h"
#include "gen_cpp/PlanNodes_types.h"
+#include "io/io_common.h"
#include "runtime/descriptors.h"
#include "storage/segment/condition_cache.h"
@@ -179,6 +180,10 @@ public:
if (*eos) {
return Status::OK();
}
+ if (_io_ctx != nullptr && _io_ctx->should_stop) {
+ *eos = true;
+ return Status::OK();
+ }
if (!_data_reader.reader) {
if (_is_table_level_count_active()) {
RETURN_IF_ERROR(_read_table_level_count(block, eos));
@@ -198,7 +203,15 @@ public:
if (!_aggregate_pushdown_tried) {
SCOPED_TIMER(_profile.pushdown_agg_timer);
bool pushed_down = false;
-
RETURN_IF_ERROR(_try_materialize_aggregate_pushdown_rows(block, &pushed_down));
+ const auto status =
_try_materialize_aggregate_pushdown_rows(block, &pushed_down);
+ if (!status.ok()) {
+ if (_io_ctx != nullptr && _io_ctx->should_stop &&
+ status.is<ErrorCode::END_OF_FILE>()) {
+ *eos = true;
+ return Status::OK();
+ }
+ return status;
+ }
if (pushed_down) {
return Status::OK();
}
@@ -566,6 +579,12 @@ protected:
return Status::OK();
}
+ void _record_scan_rows(size_t rows) {
+ if (_io_ctx != nullptr && _io_ctx->file_reader_stats != nullptr) {
+ _io_ctx->file_reader_stats->read_rows += rows;
+ }
+ }
+
// Finalize file-local block to table/global schema block.
Status finalize_chunk(Block* block, const size_t rows) {
SCOPED_TIMER(_profile.finalize_timer);
diff --git a/be/test/exec/scan/file_scanner_v2_test.cpp
b/be/test/exec/scan/file_scanner_v2_test.cpp
index 436a18c66de..9eb71301801 100644
--- a/be/test/exec/scan/file_scanner_v2_test.cpp
+++ b/be/test/exec/scan/file_scanner_v2_test.cpp
@@ -222,6 +222,123 @@ 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,
+ FileScannerV2::UncachedReaderBytesStorage::REMOTE,
&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,
+ FileScannerV2::UncachedReaderBytesStorage::REMOTE,
&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,
+ FileScannerV2::UncachedReaderBytesStorage::REMOTE,
&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,
+ FileScannerV2::UncachedReaderBytesStorage::REMOTE,
&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,
+ FileScannerV2::UncachedReaderBytesStorage::REMOTE,
&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);
+}
+
+TEST(FileScannerV2Test,
RealtimeCounterDeltasDoNotChargePeerCacheAsRemoteStorage) {
+ 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.num_peer_io_total = 1;
+ file_cache_statistics.bytes_read_from_peer = 100;
+ auto deltas = FileScannerV2::TEST_collect_realtime_counter_deltas(
+ file_reader_stats, file_cache_statistics,
+ FileScannerV2::UncachedReaderBytesStorage::REMOTE,
&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(0, deltas.scan_bytes_from_remote_storage);
+}
+
+TEST(FileScannerV2Test,
RealtimeCounterDeltasDoNotChargeLocalFileFallbackAsRemoteStorage) {
+ 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,
+ FileScannerV2::UncachedReaderBytesStorage::LOCAL,
&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(100, deltas.scan_bytes_from_local_storage);
+ EXPECT_EQ(0, 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.
diff --git a/be/test/format_v2/delimited_text/csv_reader_test.cpp
b/be/test/format_v2/delimited_text/csv_reader_test.cpp
index 7c787de7f8c..d5eafec7142 100644
--- a/be/test/format_v2/delimited_text/csv_reader_test.cpp
+++ b/be/test/format_v2/delimited_text/csv_reader_test.cpp
@@ -410,7 +410,9 @@ TEST_F(CsvV2ReaderTest,
ProfileCountersTrackReadParseDeserializeAndFilter) {
output.close();
_state._query_options.__set_read_csv_empty_line_as_null(true);
+ io::FileReaderStats file_reader_stats;
auto io_ctx = std::make_shared<io::IOContext>();
+ io_ctx->file_reader_stats = &file_reader_stats;
auto reader = create_reader(profile_path, &_params, _slots, &_state,
&_profile, 0, -1,
TFileCompressType::UNKNOWN, io_ctx);
std::vector<ColumnDefinition> schema;
@@ -443,6 +445,7 @@ TEST_F(CsvV2ReaderTest,
ProfileCountersTrackReadParseDeserializeAndFilter) {
EXPECT_EQ(counter_value(&_profile, "RowsReadBeforeFilter"), 3);
EXPECT_EQ(counter_value(&_profile, "RowsFilteredByConjunct"), 2);
EXPECT_EQ(io_ctx->predicate_filtered_rows, 2);
+ EXPECT_EQ(file_reader_stats.read_rows, 3);
EXPECT_EQ(counter_value(&_profile, "RowsFilteredByDeleteConjunct"), 0);
EXPECT_EQ(counter_value(&_profile, "RowsReturned"), 1);
EXPECT_EQ(counter_value(&_profile, "EmptyLinesRead"), 1);
diff --git a/be/test/format_v2/parquet/parquet_scan_test.cpp
b/be/test/format_v2/parquet/parquet_scan_test.cpp
index 339ccfb27fc..8c1ac915755 100644
--- a/be/test/format_v2/parquet/parquet_scan_test.cpp
+++ b/be/test/format_v2/parquet/parquet_scan_test.cpp
@@ -355,7 +355,8 @@ protected:
std::unique_ptr<format::parquet::ParquetReader> create_reader(
int64_t range_start_offset = 0, int64_t range_size = -1,
RuntimeProfile* profile = nullptr,
- std::optional<format::GlobalRowIdContext> global_rowid_context =
std::nullopt) const {
+ std::optional<format::GlobalRowIdContext> global_rowid_context =
std::nullopt,
+ std::shared_ptr<io::IOContext> io_ctx = nullptr) const {
auto system_properties = std::make_shared<io::FileSystemProperties>();
system_properties->system_type = TFileType::FILE_LOCAL;
auto file_description = std::make_unique<io::FileDescription>();
@@ -363,8 +364,9 @@ protected:
file_description->file_size =
static_cast<int64_t>(std::filesystem::file_size(_file_path));
file_description->range_start_offset = range_start_offset;
file_description->range_size = range_size;
- return std::make_unique<format::parquet::ParquetReader>(
- system_properties, file_description, nullptr, profile,
global_rowid_context);
+ return
std::make_unique<format::parquet::ParquetReader>(system_properties,
file_description,
+
std::move(io_ctx), profile,
+
global_rowid_context);
}
std::shared_ptr<format::FileScanRequest> open_all_row_groups(
@@ -676,6 +678,45 @@ TEST_F(ParquetScanTest,
AggregateMinMaxSupportsNestedSingleLeafProjection) {
EXPECT_EQ(result.columns[0].max_value.get<TYPE_INT>(), 11);
}
+TEST_F(ParquetScanTest, AggregateCountOnStructRecordsSelectedRowsRead) {
+ write_struct_parquet_file(_file_path);
+ io::FileReaderStats file_reader_stats;
+ auto io_ctx = std::make_shared<io::IOContext>();
+ io_ctx->file_reader_stats = &file_reader_stats;
+ auto reader = create_reader(0, -1, nullptr, std::nullopt, io_ctx);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+ open_all_row_groups(reader.get());
+
+ format::FileAggregateRequest aggregate_request;
+ aggregate_request.agg_type = TPushAggOp::COUNT;
+ aggregate_request.columns.push_back({.projection = field_projection(0)});
+ format::FileAggregateResult result;
+ ASSERT_TRUE(reader->get_aggregate_result(aggregate_request, &result).ok());
+ EXPECT_EQ(result.count, 4);
+ EXPECT_EQ(file_reader_stats.read_rows, 4);
+}
+
+TEST_F(ParquetScanTest, AggregateCountOnStructReturnsEndOfFileWhenStopped) {
+ write_struct_parquet_file(_file_path);
+ io::FileReaderStats file_reader_stats;
+ auto io_ctx = std::make_shared<io::IOContext>();
+ io_ctx->file_reader_stats = &file_reader_stats;
+ auto reader = create_reader(0, -1, nullptr, std::nullopt, io_ctx);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+ open_all_row_groups(reader.get());
+ io_ctx->should_stop = true;
+
+ format::FileAggregateRequest aggregate_request;
+ aggregate_request.agg_type = TPushAggOp::COUNT;
+ aggregate_request.columns.push_back({.projection = field_projection(0)});
+ format::FileAggregateResult result;
+ const auto status = reader->get_aggregate_result(aggregate_request,
&result);
+ EXPECT_TRUE(status.is<ErrorCode::END_OF_FILE>()) << status;
+ EXPECT_EQ(file_reader_stats.read_rows, 0);
+}
+
TEST_F(ParquetScanTest,
AggregateRejectsRepeatedMissingStatisticsAndInvalidRequests) {
write_list_parquet_file(_file_path);
auto repeated_reader = create_reader();
diff --git a/be/test/format_v2/table_reader_test.cpp
b/be/test/format_v2/table_reader_test.cpp
index e2015768e41..b1db95b135e 100644
--- a/be/test/format_v2/table_reader_test.cpp
+++ b/be/test/format_v2/table_reader_test.cpp
@@ -969,10 +969,13 @@ struct FakeFileReaderState {
int open_count = 0;
int close_count = 0;
int64_t total_rows = 2;
+ int64_t aggregate_count = -1;
bool eof_with_first_batch = true;
bool inject_delete_conjunct = false;
+ bool stop_during_aggregate = false;
std::shared_ptr<FileScanRequest> last_request;
std::shared_ptr<ConditionCacheContext> condition_cache_ctx;
+ std::shared_ptr<io::IOContext> io_ctx;
};
class FakeFileReader final : public FileReader {
@@ -980,7 +983,7 @@ public:
FakeFileReader(std::shared_ptr<io::FileSystemProperties>&
system_properties,
std::unique_ptr<io::FileDescription>& file_description,
std::vector<ColumnDefinition> schema,
std::shared_ptr<FakeFileReaderState> state)
- : FileReader(system_properties, file_description, nullptr,
nullptr),
+ : FileReader(system_properties, file_description, state->io_ctx,
nullptr),
_schema(std::move(schema)),
_state(std::move(state)) {}
@@ -1070,6 +1073,27 @@ public:
return Status::OK();
}
+ Status get_aggregate_result(const FileAggregateRequest& request,
+ FileAggregateResult* result) override {
+ DORIS_CHECK(result != nullptr);
+ if (_state->aggregate_count < 0) {
+ return FileReader::get_aggregate_result(request, result);
+ }
+ if (request.agg_type != TPushAggOp::type::COUNT) {
+ return Status::NotSupported("fake reader only supports COUNT
aggregate pushdown");
+ }
+ if (_state->stop_during_aggregate) {
+ DORIS_CHECK(_state->io_ctx != nullptr);
+ _state->io_ctx->should_stop = true;
+ return Status::EndOfFile("stop");
+ }
+ result->count = _state->aggregate_count;
+ result->columns.clear();
+ _record_scan_rows(_state->aggregate_count);
+ _eof = true;
+ return Status::OK();
+ }
+
void set_condition_cache_context(std::shared_ptr<ConditionCacheContext>
ctx) override {
_state->condition_cache_ctx = std::move(ctx);
}
@@ -1200,6 +1224,88 @@ TEST(TableReaderTest,
CanUseInjectedFileReaderForStandaloneUnitTest) {
EXPECT_TRUE(eos);
}
+TEST(TableReaderTest, PushDownCountRecordsReaderRowsBeforeClosingReader) {
+ std::vector<ColumnDefinition> file_schema;
+ file_schema.push_back(make_file_column(0, "id",
std::make_shared<DataTypeInt32>()));
+
+ std::vector<ColumnDefinition> projected_columns;
+ projected_columns.push_back(make_table_column(0, "id",
std::make_shared<DataTypeInt32>()));
+ set_name_identifiers(&projected_columns);
+
+ io::FileReaderStats file_reader_stats;
+ auto io_ctx = std::make_shared<io::IOContext>();
+ io_ctx->file_reader_stats = &file_reader_stats;
+
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ auto fake_state = std::make_shared<FakeFileReaderState>();
+ fake_state->aggregate_count = 3;
+ fake_state->io_ctx = io_ctx;
+ FakeTableReader reader(file_schema, fake_state);
+ ASSERT_TRUE(reader.init({
+ .projected_columns = projected_columns,
+ .conjuncts = {},
+ .format = FileFormat::PARQUET,
+ .scan_params = nullptr,
+ .io_ctx = io_ctx,
+ .runtime_state = &state,
+ .scanner_profile = nullptr,
+ .push_down_agg_type =
TPushAggOp::type::COUNT,
+ })
+ .ok());
+
+ SplitReadOptions split_options;
+ split_options.current_range.__set_path("fake-table-reader-input");
+ ASSERT_TRUE(reader.prepare_split(split_options).ok());
+
+ Block block = build_table_block(projected_columns);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ EXPECT_FALSE(eos);
+ EXPECT_EQ(block.rows(), 3);
+ EXPECT_EQ(file_reader_stats.read_rows, 3);
+ EXPECT_EQ(fake_state->close_count, 1);
+}
+
+TEST(TableReaderTest, PushDownCountStopConvertsAggregateEndOfFileToEos) {
+ std::vector<ColumnDefinition> file_schema;
+ file_schema.push_back(make_file_column(0, "id",
std::make_shared<DataTypeInt32>()));
+
+ std::vector<ColumnDefinition> projected_columns;
+ projected_columns.push_back(make_table_column(0, "id",
std::make_shared<DataTypeInt32>()));
+ set_name_identifiers(&projected_columns);
+
+ auto io_ctx = std::make_shared<io::IOContext>();
+
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ auto fake_state = std::make_shared<FakeFileReaderState>();
+ fake_state->aggregate_count = 3;
+ fake_state->io_ctx = io_ctx;
+ fake_state->stop_during_aggregate = true;
+ FakeTableReader reader(file_schema, fake_state);
+ ASSERT_TRUE(reader.init({
+ .projected_columns = projected_columns,
+ .conjuncts = {},
+ .format = FileFormat::PARQUET,
+ .scan_params = nullptr,
+ .io_ctx = io_ctx,
+ .runtime_state = &state,
+ .scanner_profile = nullptr,
+ .push_down_agg_type =
TPushAggOp::type::COUNT,
+ })
+ .ok());
+
+ SplitReadOptions split_options;
+ split_options.current_range.__set_path("fake-table-reader-input");
+ ASSERT_TRUE(reader.prepare_split(split_options).ok());
+
+ Block block = build_table_block(projected_columns);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ EXPECT_TRUE(eos);
+ EXPECT_EQ(block.rows(), 0);
+ EXPECT_EQ(fake_state->close_count, 0);
+}
+
TEST(TableReaderTest, DebugStringCoversReaderStateAndEnumNames) {
std::vector<ColumnDefinition> file_schema;
file_schema.push_back(make_file_column(0, "id",
std::make_shared<DataTypeInt32>()));
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
