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 bc03da163b2 [improvement](parquet) Add prefetch for file scanner v2
(#65151)
bc03da163b2 is described below
commit bc03da163b28b4e5f92b0b07254ec190bd11e790
Author: Gabriel <[email protected]>
AuthorDate: Thu Jul 2 22:57:32 2026 +0800
[improvement](parquet) Add prefetch for file scanner v2 (#65151)
## Summary
Add Parquet FileScannerV2 prefetch support and align the v2 row-group
read path with the v1 Parquet reader's `MergeRangeFileReader` behavior.
## Changes
- Keep the Arrow Parquet random-access wrapper, but make its underlying
Doris `FileReader` switchable per row group.
- Build projected Parquet leaf-column chunk ranges for the current row
group, including nested projections.
- Before opening each selected row group, configure
`MergeRangeFileReader` when the average projected column-chunk IO is
smaller than `MergeRangeFileReader::SMALL_IO`, matching the v1
small-random-IO path.
- Respect the existing `merge_read_slice_size` query option when
constructing `MergeRangeFileReader`.
- Preserve `TracingFileReader` statistics when switching between the raw
reader and merge reader.
- Keep best-effort file-cache warm-up only as a fallback when
merge-range reading is not active.
## Notes
The important v1 parity point is that subsequent Arrow `ReadAt()` calls
now go through `MergeRangeFileReader` for small projected row-group IOs.
This changes the actual read path, not just background cache warming.
---
be/src/format_v2/parquet/parquet_file_context.cpp | 153 ++++++++++++++++++-
be/src/format_v2/parquet/parquet_file_context.h | 39 +++++
be/src/format_v2/parquet/parquet_reader.cpp | 5 +
be/src/format_v2/parquet/parquet_scan.cpp | 162 ++++++++++++++++++++-
be/src/format_v2/parquet/parquet_scan.h | 28 +++-
.../parquet/parquet_page_cache_range_test.cpp | 64 ++++++++
6 files changed, 441 insertions(+), 10 deletions(-)
diff --git a/be/src/format_v2/parquet/parquet_file_context.cpp
b/be/src/format_v2/parquet/parquet_file_context.cpp
index dd9bf6aa954..dac09747408 100644
--- a/be/src/format_v2/parquet/parquet_file_context.cpp
+++ b/be/src/format_v2/parquet/parquet_file_context.cpp
@@ -24,14 +24,18 @@
#include <algorithm>
#include <cstring>
#include <exception>
+#include <limits>
#include <mutex>
#include <unordered_map>
#include <utility>
#include "common/check.h"
#include "common/config.h"
+#include "io/cache/cached_remote_file_reader.h"
#include "io/file_factory.h"
+#include "io/fs/buffered_reader.h"
#include "io/fs/file_reader.h"
+#include "io/fs/tracing_file_reader.h"
#include "storage/cache/page_cache.h"
#include "util/slice.h"
@@ -91,6 +95,38 @@ std::vector<ParquetPageCacheReadPlanEntry>
plan_page_cache_range_read(
return plan;
}
+std::vector<ParquetPageCacheRange> valid_prefetch_ranges(
+ const std::vector<ParquetPageCacheRange>& ranges) {
+ std::vector<ParquetPageCacheRange> valid_ranges;
+ valid_ranges.reserve(ranges.size());
+ for (const auto& range : ranges) {
+ if (range.offset < 0 || range.size <= 0 ||
+ range.offset > std::numeric_limits<int64_t>::max() - range.size) {
+ continue;
+ }
+ valid_ranges.push_back(range);
+ }
+ return valid_ranges;
+}
+
+size_t average_prefetch_range_size(const std::vector<ParquetPageCacheRange>&
ranges) {
+ const auto valid_ranges = valid_prefetch_ranges(ranges);
+ if (valid_ranges.empty()) {
+ return 0;
+ }
+ size_t total_size = 0;
+ for (const auto& range : valid_ranges) {
+ total_size += static_cast<size_t>(range.size);
+ }
+ return total_size / valid_ranges.size();
+}
+
+bool should_use_merge_range_reader(const std::vector<ParquetPageCacheRange>&
ranges,
+ size_t avg_io_size, bool
is_in_memory_reader) {
+ return !is_in_memory_reader && !valid_prefetch_ranges(ranges).empty() &&
+ avg_io_size < io::MergeRangeFileReader::SMALL_IO;
+}
+
} // namespace detail
namespace {
@@ -172,15 +208,24 @@ public:
DorisRandomAccessFile(io::FileReaderSPtr file_reader, io::IOContext*
io_ctx,
bool enable_page_cache, std::string
page_cache_file_key)
: _file_reader(std::move(file_reader)),
+ _base_file_reader(_file_reader),
_io_ctx(io_ctx),
_enable_page_cache(enable_page_cache),
_page_cache_file_key(std::move(page_cache_file_key)) {
DORIS_CHECK(_file_reader != nullptr);
+ if (auto tracing_reader =
std::dynamic_pointer_cast<io::TracingFileReader>(_file_reader)) {
+ _file_reader_stats = tracing_reader->stats();
+ _base_file_reader = tracing_reader->inner_reader();
+ }
+ DORIS_CHECK(_base_file_reader != nullptr);
set_mode(arrow::io::FileMode::READ);
}
arrow::Status Close() override {
- _closed = true;
+ if (!_closed) {
+ collect_active_merge_range_profile();
+ _closed = true;
+ }
return arrow::Status::OK();
}
@@ -253,6 +298,53 @@ public:
_page_cache_ranges = std::move(ranges);
}
+ void prefetch_ranges(const std::vector<ParquetPageCacheRange>& ranges,
+ const io::IOContext* io_ctx) {
+ auto cached_reader = cached_remote_file_reader();
+ if (cached_reader == nullptr) {
+ return;
+ }
+ const auto* prefetch_io_ctx = io_ctx != nullptr ? io_ctx : _io_ctx;
+ for (const auto& range : ranges) {
+ if (range.offset < 0 || range.size <= 0) {
+ continue;
+ }
+ cached_reader->prefetch_range(static_cast<size_t>(range.offset),
+ static_cast<size_t>(range.size),
prefetch_io_ctx);
+ }
+ }
+
+ bool set_random_access_ranges(const std::vector<ParquetPageCacheRange>&
ranges,
+ size_t avg_io_size, RuntimeProfile* profile,
+ int64_t merge_read_slice_size) {
+ reset_active_file_reader();
+ const auto valid_ranges = detail::valid_prefetch_ranges(ranges);
+ if (!detail::should_use_merge_range_reader(
+ valid_ranges, avg_io_size,
+
typeid_cast<io::InMemoryFileReader*>(_base_file_reader.get()) != nullptr)) {
+ return false;
+ }
+
+ std::vector<io::PrefetchRange> random_access_ranges;
+ random_access_ranges.reserve(valid_ranges.size());
+ for (const auto& range : valid_ranges) {
+
random_access_ranges.emplace_back(static_cast<size_t>(range.offset),
+
static_cast<size_t>(range.end_offset()));
+ }
+
+ // This mirrors the v1 parquet reader: when projected column chunks in
a row group are
+ // small random IOs, make the actual ReadAt path range-aware. Arrow
still drives decoding,
+ // but every page read below this point sees MergeRangeFileReader
instead of the raw remote
+ // reader, so adjacent small requests can be coalesced and served from
merge buffers.
+ // Example: a row group projects leaf chunks [1MB, 1.5MB) and [1.6MB,
2MB). Arrow later
+ // issues page reads inside those chunks; MergeRangeFileReader can
fetch a wider slice once
+ // and satisfy the following ReadAt calls from its boxes, reducing
remote request count.
+ _merge_range_active = true;
+ set_active_file_reader(std::make_shared<io::MergeRangeFileReader>(
+ profile, _base_file_reader, random_access_ranges,
merge_read_slice_size));
+ return true;
+ }
+
ParquetPageCacheStats page_cache_stats() const {
std::lock_guard lock(_page_cache_mutex);
return _page_cache_stats;
@@ -359,11 +451,56 @@ private:
++_page_cache_stats.compressed_write_count;
}
+ void set_active_file_reader(io::FileReaderSPtr reader) {
+ DORIS_CHECK(reader != nullptr);
+ _file_reader = _file_reader_stats != nullptr
+ ?
std::make_shared<io::TracingFileReader>(std::move(reader),
+
_file_reader_stats)
+ : std::move(reader);
+ }
+
+ void reset_active_file_reader() {
+ collect_active_merge_range_profile();
+ _merge_range_active = false;
+ set_active_file_reader(_base_file_reader);
+ }
+
+ void collect_active_merge_range_profile() {
+ if (_merge_range_active && _file_reader != nullptr) {
+ // MergeRangeFileReader writes its MergedSmallIO counters only from
+ // collect_profile_before_close(). v2 replaces the active reader
for every row group,
+ // so collect before overwriting it; Close() handles the final row
group. Example:
+ // RG0 installs a merge reader, RG1 calls
set_random_access_ranges() and resets the
+ // active reader first, so RG0's RequestIO/MergedIO counters must
be flushed here.
+ _file_reader->collect_profile_before_close();
+ }
+ }
+
+ std::shared_ptr<io::CachedRemoteFileReader> cached_remote_file_reader() {
+ if (_merge_range_active) {
+ return nullptr;
+ }
+ auto reader = _file_reader;
+ if (reader == nullptr) {
+ return nullptr;
+ }
+ // FileReader::init wraps the physical reader with TracingFileReader
when scan IO stats are
+ // enabled. Prefetch should target the physical cached reader below
that tracing wrapper,
+ // otherwise v2 scans with profiling would silently lose prefetch.
+ if (auto tracing_reader =
std::dynamic_pointer_cast<io::TracingFileReader>(reader)) {
+ reader = tracing_reader->inner_reader();
+ }
+ return std::dynamic_pointer_cast<io::CachedRemoteFileReader>(reader);
+ }
+
io::FileReaderSPtr _file_reader;
+ io::FileReaderSPtr _base_file_reader;
+ io::FileReaderStats* _file_reader_stats = nullptr;
io::IOContext* _io_ctx = nullptr;
int64_t _pos = 0;
bool _closed = false;
bool _enable_page_cache = false;
+ bool _merge_range_active = false;
std::string _page_cache_file_key;
mutable std::mutex _page_cache_mutex;
std::vector<ParquetPageCacheRange> _page_cache_ranges;
@@ -417,6 +554,20 @@ void
ParquetFileContext::register_page_cache_ranges(std::vector<ParquetPageCache
->register_page_cache_ranges(std::move(ranges));
}
+void ParquetFileContext::prefetch_ranges(const
std::vector<ParquetPageCacheRange>& ranges,
+ const io::IOContext* io_ctx) {
+ DORIS_CHECK(arrow_file != nullptr);
+
static_cast<DorisRandomAccessFile*>(arrow_file.get())->prefetch_ranges(ranges,
io_ctx);
+}
+
+bool ParquetFileContext::set_random_access_ranges(const
std::vector<ParquetPageCacheRange>& ranges,
+ size_t avg_io_size,
RuntimeProfile* profile,
+ int64_t
merge_read_slice_size) {
+ DORIS_CHECK(arrow_file != nullptr);
+ return static_cast<DorisRandomAccessFile*>(arrow_file.get())
+ ->set_random_access_ranges(ranges, avg_io_size, profile,
merge_read_slice_size);
+}
+
ParquetPageCacheStats ParquetFileContext::page_cache_stats() const {
if (arrow_file == nullptr) {
return {};
diff --git a/be/src/format_v2/parquet/parquet_file_context.h
b/be/src/format_v2/parquet/parquet_file_context.h
index 8dedf732c8f..e75293f470f 100644
--- a/be/src/format_v2/parquet/parquet_file_context.h
+++ b/be/src/format_v2/parquet/parquet_file_context.h
@@ -18,6 +18,7 @@
#include <arrow/io/interfaces.h>
#include <parquet/api/reader.h>
+#include <cstddef>
#include <cstdint>
#include <memory>
#include <vector>
@@ -27,8 +28,13 @@
namespace doris::io {
struct FileDescription;
+struct IOContext;
} // namespace doris::io
+namespace doris {
+class RuntimeProfile;
+} // namespace doris
+
namespace doris::format::parquet {
struct ParquetPageCacheRange {
@@ -76,6 +82,27 @@ namespace detail {
std::vector<ParquetPageCacheReadPlanEntry> plan_page_cache_range_read(
int64_t position, int64_t nbytes, const
std::vector<ParquetPageCacheRange>& cached_ranges);
+// Keep only byte ranges that are safe to hand to FileReader implementations.
Parquet metadata is
+// expected to contain non-negative offsets and positive compressed sizes, but
tests and corrupted
+// footers can still feed invalid values. Example: [100, 64) is kept, while
[-1, 64), [100, 0) and
+// an offset+size overflow are ignored.
+std::vector<ParquetPageCacheRange> valid_prefetch_ranges(
+ const std::vector<ParquetPageCacheRange>& ranges);
+
+// Average projected column-chunk size for one row group. The v1 parquet path
uses this value to
+// decide whether a row group is dominated by small random IOs; v2 uses the
same signal before
+// installing MergeRangeFileReader. Example: chunks of 512KB and 1MB average
below SMALL_IO and are
+// good merge-reader candidates, while two 8MB chunks should stay on the raw
random-access reader.
+size_t average_prefetch_range_size(const std::vector<ParquetPageCacheRange>&
ranges);
+
+// Decide whether Arrow ReadAt() should be routed through MergeRangeFileReader
for the current row
+// group. This is intentionally stricter than the background warm-up path:
+// - no valid projected chunks -> nothing to merge;
+// - in-memory file readers already avoid remote random IO;
+// - average chunk size >= MergeRangeFileReader::SMALL_IO would make merged
reading wasteful.
+bool should_use_merge_range_reader(const std::vector<ParquetPageCacheRange>&
ranges,
+ size_t avg_io_size, bool
is_in_memory_reader);
+
} // namespace detail
struct ParquetFileContext {
@@ -90,6 +117,18 @@ struct ParquetFileContext {
// decoding, so v2 caches the serialized bytes read inside these ranges
and excludes
// footer/metadata reads that happen before registration.
void register_page_cache_ranges(std::vector<ParquetPageCacheRange> ranges);
+ // Best-effort asynchronous warm-up for Parquet column chunks. This only
has an effect when
+ // the underlying Doris file reader is a CachedRemoteFileReader; other
readers keep the same
+ // random-access behavior and simply skip prefetch.
+ void prefetch_ranges(const std::vector<ParquetPageCacheRange>& ranges,
+ const io::IOContext* io_ctx);
+ // Switch the active reader used by Arrow ReadAt() to v1's
MergeRangeFileReader when the current
+ // row group's projected column chunks are small random IOs. This is the
real v1-compatible
+ // prefetch path: subsequent Arrow page reads go through the merged reader
instead of merely
+ // warming file cache in the background. Returns true when merge-range
reading is active.
+ bool set_random_access_ranges(const std::vector<ParquetPageCacheRange>&
ranges,
+ size_t avg_io_size, RuntimeProfile* profile,
+ int64_t merge_read_slice_size);
ParquetPageCacheStats page_cache_stats() const;
Status close();
};
diff --git a/be/src/format_v2/parquet/parquet_reader.cpp
b/be/src/format_v2/parquet/parquet_reader.cpp
index 3e13333e7d4..3d8a0a061fb 100644
--- a/be/src/format_v2/parquet/parquet_reader.cpp
+++ b/be/src/format_v2/parquet/parquet_reader.cpp
@@ -334,6 +334,11 @@ Status ParquetReader::init(RuntimeState* state) {
_state->scheduler.set_timezone(&state->timezone_obj());
_state->scheduler.set_enable_strict_mode(_state->enable_strict_mode);
}
+ int64_t merge_read_slice_size = -1;
+ if (state != nullptr &&
state->query_options().__isset.merge_read_slice_size) {
+ merge_read_slice_size = state->query_options().merge_read_slice_size;
+ }
+ _state->scheduler.set_merge_read_options(_profile, merge_read_slice_size);
_state->scheduler.set_batch_size(_batch_size);
// Open parquet file and parse metadata to get file schema.
RETURN_IF_ERROR(_state->file_context.open(_tracing_file_reader,
_io_ctx.get(),
diff --git a/be/src/format_v2/parquet/parquet_scan.cpp
b/be/src/format_v2/parquet/parquet_scan.cpp
index d636f3e3f9e..62dad184b1e 100644
--- a/be/src/format_v2/parquet/parquet_scan.cpp
+++ b/be/src/format_v2/parquet/parquet_scan.cpp
@@ -18,6 +18,8 @@
#include <algorithm>
#include <limits>
#include <memory>
+#include <ranges>
+#include <unordered_set>
#include <utility>
#include "common/exception.h"
@@ -40,6 +42,39 @@ int64_t column_start_offset(const
::parquet::ColumnChunkMetaData& column_metadat
: cast_set<int64_t>(column_metadata.data_page_offset());
}
+void collect_all_leaf_column_ids(const ParquetColumnSchema& column_schema,
+ std::unordered_set<int>* leaf_column_ids) {
+ DORIS_CHECK(leaf_column_ids != nullptr);
+ if (column_schema.kind == ParquetColumnSchemaKind::PRIMITIVE) {
+ if (column_schema.leaf_column_id >= 0) {
+ leaf_column_ids->insert(column_schema.leaf_column_id);
+ }
+ return;
+ }
+ for (const auto& child : column_schema.children) {
+ DORIS_CHECK(child != nullptr);
+ collect_all_leaf_column_ids(*child, leaf_column_ids);
+ }
+}
+
+void collect_projected_leaf_column_ids(const ParquetColumnSchema&
column_schema,
+ const format::LocalColumnIndex&
projection,
+ std::unordered_set<int>*
leaf_column_ids) {
+ DORIS_CHECK(leaf_column_ids != nullptr);
+ if (projection.project_all_children || projection.children.empty()) {
+ collect_all_leaf_column_ids(column_schema, leaf_column_ids);
+ return;
+ }
+ for (const auto& child_projection : projection.children) {
+ const auto child_it =
+ std::ranges::find_if(column_schema.children, [&](const auto&
child_schema) {
+ return child_schema->local_id ==
child_projection.local_id();
+ });
+ DORIS_CHECK(child_it != column_schema.children.end());
+ collect_projected_leaf_column_ids(**child_it, child_projection,
leaf_column_ids);
+ }
+}
+
bool is_row_group_outside_range(const ::parquet::FileMetaData& metadata,
const ParquetScanRange& scan_range, int
row_group_idx) {
if (scan_range.size < 0) {
@@ -69,6 +104,56 @@ bool is_row_group_outside_range(const
::parquet::FileMetaData& metadata,
return row_group_mid_offset < range_start_offset || row_group_mid_offset
>= range_end_offset;
}
+std::vector<format::LocalColumnIndex> request_scan_columns(const
format::FileScanRequest& request) {
+ std::vector<format::LocalColumnIndex> scan_columns;
+ scan_columns.reserve(request.predicate_columns.size() +
request.non_predicate_columns.size());
+ scan_columns.insert(scan_columns.end(), request.predicate_columns.begin(),
+ request.predicate_columns.end());
+ scan_columns.insert(scan_columns.end(),
request.non_predicate_columns.begin(),
+ request.non_predicate_columns.end());
+ return scan_columns;
+}
+
+std::vector<ParquetPageCacheRange> build_row_group_prefetch_ranges(
+ const ::parquet::FileMetaData& metadata,
+ const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
+ const std::vector<format::LocalColumnIndex>& scan_columns, int
row_group_idx) {
+ std::unordered_set<int> leaf_column_ids;
+ for (const auto& projection : scan_columns) {
+ const auto local_id = projection.local_id();
+ if (local_id == format::ROW_POSITION_COLUMN_ID ||
+ local_id == format::GLOBAL_ROWID_COLUMN_ID) {
+ continue;
+ }
+ DORIS_CHECK(local_id >= 0 && local_id <
static_cast<int32_t>(file_schema.size()));
+ DORIS_CHECK(file_schema[local_id] != nullptr);
+ // Prefetch and merge-reader ranges must be physical leaf chunks, not
Doris logical slots.
+ // Example: for a struct column s<a:int,b:string>, projecting only s.a
should include only
+ // the Parquet leaf chunk of a. Projecting the whole struct includes
both a and b.
+ collect_projected_leaf_column_ids(*file_schema[local_id], projection,
&leaf_column_ids);
+ }
+
+ auto row_group_metadata = metadata.RowGroup(row_group_idx);
+ DORIS_CHECK(row_group_metadata != nullptr);
+ std::vector<int> ordered_leaf_column_ids(leaf_column_ids.begin(),
leaf_column_ids.end());
+ std::ranges::sort(ordered_leaf_column_ids);
+
+ std::vector<ParquetPageCacheRange> ranges;
+ ranges.reserve(ordered_leaf_column_ids.size());
+ for (const auto leaf_column_id : ordered_leaf_column_ids) {
+ DORIS_CHECK(leaf_column_id >= 0 && leaf_column_id <
row_group_metadata->num_columns());
+ auto column_metadata = row_group_metadata->ColumnChunk(leaf_column_id);
+ DORIS_CHECK(column_metadata != nullptr);
+ const int64_t offset = column_start_offset(*column_metadata);
+ const int64_t size = column_metadata->total_compressed_size();
+ DORIS_CHECK(offset >= 0);
+ if (size > 0) {
+ ranges.push_back(ParquetPageCacheRange {.offset = offset, .size =
size});
+ }
+ }
+ return ranges;
+}
+
} // namespace
Status plan_parquet_row_groups(const ::parquet::FileMetaData& metadata,
@@ -326,11 +411,15 @@ void ParquetScanScheduler::reset_current_row_group() {
_current_predicate_columns.clear();
_current_non_predicate_columns.clear();
_current_row_group_rows = 0;
+ _current_row_group_id = -1;
_current_row_group_rows_read = 0;
_current_row_group_first_row = 0;
_current_selected_ranges.clear();
_current_range_idx = 0;
_current_range_rows_read = 0;
+ _current_predicate_prefetched = false;
+ _current_non_predicate_prefetched = false;
+ _current_merge_range_active = false;
}
Status ParquetScanScheduler::open_next_row_group(
@@ -343,6 +432,8 @@ Status ParquetScanScheduler::open_next_row_group(
}
const RowGroupReadPlan& row_group_plan =
_row_group_plans[_next_row_group_plan_idx++];
const int row_group_idx = row_group_plan.row_group_id;
+ _current_merge_range_active =
+ prepare_current_row_group_reader(file_context, file_schema,
request, row_group_idx);
try {
_current_row_group = file_context.file_reader->RowGroup(row_group_idx);
} catch (const ::parquet::ParquetException& e) {
@@ -358,6 +449,7 @@ Status ParquetScanScheduler::open_next_row_group(
_current_row_group_rows = row_group_metadata->num_rows();
DORIS_CHECK(_current_row_group_rows == row_group_plan.row_group_rows);
DORIS_CHECK(_current_row_group_rows > 0);
+ _current_row_group_id = row_group_idx;
DORIS_CHECK(!row_group_plan.selected_ranges.empty());
_current_row_group_first_row = row_group_plan.first_file_row;
_current_row_group_rows_read = 0;
@@ -394,6 +486,13 @@ Status ParquetScanScheduler::open_next_row_group(
RETURN_IF_ERROR(column_reader_factory.create(*column_schema, &col,
&column_reader));
_current_predicate_columns[local_id] = std::move(column_reader);
}
+ // Start warming filter-column chunks as soon as their row group is
selected. Parquet v2 still
+ // reads through Arrow's random-access reader; this prefetch only warms
Doris file cache blocks
+ // in the background and never changes the row/column materialization
order.
+ if (!_current_merge_range_active) {
+ prefetch_current_row_group_columns(file_context, file_schema,
request.predicate_columns,
+ &_current_predicate_prefetched);
+ }
for (const auto& col : request.non_predicate_columns) {
const auto local_id = col.local_id();
if (local_id == format::ROW_POSITION_COLUMN_ID) {
@@ -416,6 +515,14 @@ Status ParquetScanScheduler::open_next_row_group(
RETURN_IF_ERROR(column_reader_factory.create(*column_schema, &col,
&column_reader));
_current_non_predicate_columns[local_id] = std::move(column_reader);
}
+ if (!_current_merge_range_active && request.conjuncts.empty() &&
+ request.delete_conjuncts.empty()) {
+ // With no row-level filters there is no lazy-read decision to wait
for, so start warming
+ // output chunks immediately after their readers are created. Filtered
scans still defer
+ // this until at least one row survives the predicate phase.
+ prefetch_current_row_group_columns(file_context, file_schema,
request.non_predicate_columns,
+ &_current_non_predicate_prefetched);
+ }
*has_row_group = true;
return Status::OK();
}
@@ -476,10 +583,45 @@ Status ParquetScanScheduler::read_filter_columns(int64_t
batch_rows,
conjunct_filtered_rows);
}
-Status ParquetScanScheduler::read_current_row_group_batch(int64_t batch_rows,
- const
format::FileScanRequest& request,
- int64_t
batch_first_file_row,
- Block* file_block,
size_t* rows) {
+bool ParquetScanScheduler::prepare_current_row_group_reader(
+ ParquetFileContext& file_context,
+ const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
+ const format::FileScanRequest& request, int row_group_idx) {
+ if (file_context.metadata == nullptr) {
+ return false;
+ }
+ const auto ranges = build_row_group_prefetch_ranges(
+ *file_context.metadata, file_schema,
request_scan_columns(request), row_group_idx);
+ const size_t avg_io_size = detail::average_prefetch_range_size(ranges);
+ return file_context.set_random_access_ranges(ranges, avg_io_size, _profile,
+ _merge_read_slice_size);
+}
+
+void ParquetScanScheduler::prefetch_current_row_group_columns(
+ ParquetFileContext& file_context,
+ const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
+ const std::vector<format::LocalColumnIndex>& scan_columns, bool*
prefetched) {
+ DORIS_CHECK(prefetched != nullptr);
+ if (_current_merge_range_active || *prefetched || scan_columns.empty() ||
+ _current_row_group_id < 0 || file_context.metadata == nullptr) {
+ return;
+ }
+ *prefetched = true;
+ // The scanner request separates predicate and non-predicate columns so
Parquet can read
+ // predicate columns first and lazily materialize the rest. Keep the same
contract for
+ // prefetch: callers decide which side to warm, and this helper only
translates that selected
+ // projection into physical column-chunk byte ranges for the current row
group.
+ file_context.prefetch_ranges(
+ build_row_group_prefetch_ranges(*file_context.metadata,
file_schema, scan_columns,
+ _current_row_group_id),
+ nullptr);
+}
+
+Status ParquetScanScheduler::read_current_row_group_batch(
+ ParquetFileContext& file_context,
+ const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
int64_t batch_rows,
+ const format::FileScanRequest& request, int64_t batch_first_file_row,
Block* file_block,
+ size_t* rows) {
if (_scan_profile.total_batches != nullptr) {
COUNTER_UPDATE(_scan_profile.total_batches, 1);
}
@@ -527,6 +669,14 @@ Status
ParquetScanScheduler::read_current_row_group_batch(int64_t batch_rows,
.column->filter(output_filter,
selected_rows)));
}
}
+ if (!_current_merge_range_active && selected_rows > 0 &&
+ !_current_non_predicate_columns.empty()) {
+ // Do not prefetch lazy output columns until at least one row survives
filtering. This is
+ // the same decision point where the v2 reader switches from
predicate-only reads to
+ // materializing non-predicate columns, so fully filtered batches
avoid unnecessary IO.
+ prefetch_current_row_group_columns(file_context, file_schema,
request.non_predicate_columns,
+ &_current_non_predicate_prefetched);
+ }
{
SCOPED_TIMER(_scan_profile.column_read_time);
@@ -629,8 +779,8 @@ Status ParquetScanScheduler::read_next_batch(
const int64_t physical_rows_read = batch_rows;
const int64_t batch_first_file_row =
_current_row_group_first_row + _current_row_group_rows_read;
- RETURN_IF_ERROR(read_current_row_group_batch(batch_rows, request,
batch_first_file_row,
- file_block, rows));
+ RETURN_IF_ERROR(read_current_row_group_batch(file_context,
file_schema, batch_rows, request,
+ batch_first_file_row,
file_block, rows));
_current_row_group_rows_read += physical_rows_read;
_current_range_rows_read += physical_rows_read;
if (_current_range_rows_read >= current_range.length) {
diff --git a/be/src/format_v2/parquet/parquet_scan.h
b/be/src/format_v2/parquet/parquet_scan.h
index ca3c665a2e9..0253fbdd50d 100644
--- a/be/src/format_v2/parquet/parquet_scan.h
+++ b/be/src/format_v2/parquet/parquet_scan.h
@@ -109,6 +109,10 @@ public:
_page_skip_profile = page_skip_profile;
}
void set_scan_profile(ParquetScanProfile scan_profile) { _scan_profile =
scan_profile; }
+ void set_merge_read_options(RuntimeProfile* profile, int64_t
merge_read_slice_size) {
+ _profile = profile;
+ _merge_read_slice_size = merge_read_slice_size;
+ }
void set_global_rowid_context(std::optional<format::GlobalRowIdContext>
context) {
_global_rowid_context = context;
}
@@ -145,9 +149,21 @@ private:
Block* file_block, SelectionVector* selection,
uint16_t* selected_rows, int64_t*
conjunct_filtered_rows);
- Status read_current_row_group_batch(int64_t batch_rows, const
format::FileScanRequest& request,
- int64_t batch_first_file_row, Block*
file_block,
- size_t* rows);
+ void prefetch_current_row_group_columns(
+ ParquetFileContext& file_context,
+ const std::vector<std::unique_ptr<ParquetColumnSchema>>&
file_schema,
+ const std::vector<format::LocalColumnIndex>& scan_columns, bool*
prefetched);
+
+ bool prepare_current_row_group_reader(
+ ParquetFileContext& file_context,
+ const std::vector<std::unique_ptr<ParquetColumnSchema>>&
file_schema,
+ const format::FileScanRequest& request, int row_group_idx);
+
+ Status read_current_row_group_batch(
+ ParquetFileContext& file_context,
+ const std::vector<std::unique_ptr<ParquetColumnSchema>>&
file_schema,
+ int64_t batch_rows, const format::FileScanRequest& request,
+ int64_t batch_first_file_row, Block* file_block, size_t* rows);
void mark_condition_cache_granules(const SelectionVector& selection,
uint16_t selected_rows,
int64_t batch_first_file_row);
@@ -161,6 +177,7 @@ private:
std::map<ColumnId, std::unique_ptr<ParquetColumnReader>>
_current_non_predicate_columns; // non-predicate ColumnReaders
int64_t _current_row_group_rows = 0; // current row group row count
+ int _current_row_group_id = -1; // current row group id in
parquet metadata
int64_t _current_row_group_rows_read = 0; // rows read in the current row
group (cursor)
int64_t _current_row_group_first_row = 0; // first file row of the current
row group
std::vector<RowRange>
@@ -168,8 +185,13 @@ private:
size_t _current_range_idx = 0; // current selected_range index
int64_t _current_range_rows_read = 0; // rows read in the current range
+ bool _current_predicate_prefetched = false;
+ bool _current_non_predicate_prefetched = false;
+ bool _current_merge_range_active = false;
ParquetPageSkipProfile _page_skip_profile;
ParquetScanProfile _scan_profile;
+ RuntimeProfile* _profile = nullptr;
+ int64_t _merge_read_slice_size = -1;
std::optional<format::GlobalRowIdContext> _global_rowid_context;
const cctz::time_zone* _timezone = nullptr;
bool _enable_strict_mode = false;
diff --git a/be/test/format_v2/parquet/parquet_page_cache_range_test.cpp
b/be/test/format_v2/parquet/parquet_page_cache_range_test.cpp
index f8e12206bb1..fac89b31c42 100644
--- a/be/test/format_v2/parquet/parquet_page_cache_range_test.cpp
+++ b/be/test/format_v2/parquet/parquet_page_cache_range_test.cpp
@@ -17,9 +17,11 @@
#include <gtest/gtest.h>
+#include <limits>
#include <vector>
#include "format_v2/parquet/parquet_file_context.h"
+#include "io/fs/buffered_reader.h"
namespace doris::format::parquet {
namespace {
@@ -113,5 +115,67 @@ TEST(ParquetPageCacheRangeTest, InvalidRequestMisses) {
EXPECT_TRUE(detail::plan_page_cache_range_read(100, -1,
cached_ranges).empty());
}
+TEST(ParquetPageCacheRangeTest,
ValidPrefetchRangesSkipInvalidAndOverflowRanges) {
+ const std::vector<ParquetPageCacheRange> ranges = {
+ {100, 50},
+ {-1, 50},
+ {200, 0},
+ {300, -1},
+ {std::numeric_limits<int64_t>::max() - 10, 20},
+ {400, 60},
+ };
+
+ const auto valid_ranges = detail::valid_prefetch_ranges(ranges);
+
+ ASSERT_EQ(valid_ranges.size(), 2);
+ EXPECT_EQ(valid_ranges[0].offset, 100);
+ EXPECT_EQ(valid_ranges[0].size, 50);
+ EXPECT_EQ(valid_ranges[1].offset, 400);
+ EXPECT_EQ(valid_ranges[1].size, 60);
+}
+
+TEST(ParquetPageCacheRangeTest, AveragePrefetchRangeSizeUsesOnlyValidRanges) {
+ const std::vector<ParquetPageCacheRange> ranges = {
+ {0, 512},
+ {512, 1536},
+ {-1, 1024},
+ {2048, 0},
+ };
+
+ EXPECT_EQ(detail::average_prefetch_range_size(ranges), 1024);
+ EXPECT_EQ(detail::average_prefetch_range_size({{-1, 1024}, {0, 0}}), 0);
+}
+
+TEST(ParquetPageCacheRangeTest,
MergeRangeReaderDecisionMatchesV1SmallIoThreshold) {
+ const std::vector<ParquetPageCacheRange> ranges = {
+ {0, 512 * 1024},
+ {512 * 1024, 1024 * 1024},
+ };
+
+ // Two sub-2MB column chunks are the intended merge-reader case: Arrow may
issue many page-level
+ // ReadAt calls inside these chunks, so v2 should route them through
MergeRangeFileReader.
+ EXPECT_TRUE(detail::should_use_merge_range_reader(
+ ranges, detail::average_prefetch_range_size(ranges), false));
+
+ // The v1 threshold is strict: a 2MB average chunk is no longer considered
"small IO".
+ EXPECT_FALSE(detail::should_use_merge_range_reader(ranges,
io::MergeRangeFileReader::SMALL_IO,
+ false));
+}
+
+TEST(ParquetPageCacheRangeTest,
MergeRangeReaderDecisionRejectsEmptyInvalidAndInMemoryInputs) {
+ const std::vector<ParquetPageCacheRange> invalid_ranges = {
+ {-1, 128},
+ {0, 0},
+ };
+ const std::vector<ParquetPageCacheRange> valid_ranges = {
+ {0, 128 * 1024},
+ };
+
+ EXPECT_FALSE(detail::should_use_merge_range_reader({}, 0, false));
+ EXPECT_FALSE(detail::should_use_merge_range_reader(invalid_ranges, 128,
false));
+ EXPECT_FALSE(detail::should_use_merge_range_reader(
+ valid_ranges, detail::average_prefetch_range_size(valid_ranges),
true));
+}
+
} // namespace
} // namespace doris::format::parquet
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
