Skip to content

[fix](be) Harden FileScannerV2 schema and reader edge cases #65548

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Gabriel39 wants to merge 3 commits into
base: master
Choose a base branch
from
+680 −86
Open

[fix](be) Harden FileScannerV2 schema and reader edge cases #65548

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
57aa8de
[fix](be) Harden FileScannerV2 schema and reader edge cases
Gabriel39 Jul 13, 2026
038d71d
[fix](be) Address FileScannerV2 review edge cases
Gabriel39 Jul 14, 2026
cc97507
[fix](be) Address remaining FileScannerV2 review cases
Gabriel39 Jul 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 38 additions & 0 deletions be/src/exec/scan/file_scanner_v2.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -262,6 +262,10 @@ void FileScannerV2::TEST_report_file_cache_profile(
bool FileScannerV2::TEST_should_skip_not_found(const Status& status, bool ignore_not_found) {
return _should_skip_not_found(status, ignore_not_found);
}

bool FileScannerV2::TEST_should_skip_empty(const Status& status, bool stopped) {
return _should_skip_empty(status, stopped);
}
#endif

bool FileScannerV2::is_supported(const TFileScanRangeParams& params, const TFileRangeDesc& range) {
Expand Down Expand Up @@ -312,6 +316,8 @@ Status FileScannerV2::init(RuntimeState* state, const VExprContextSPtrs& conjunc
RETURN_IF_ERROR(Scanner::init(state, conjuncts));
_get_block_timer =
ADD_TIMER_WITH_LEVEL(_local_state->scanner_profile(), "FileScannerV2GetBlockTime", 1);
_empty_file_counter =
ADD_COUNTER_WITH_LEVEL(_local_state->scanner_profile(), "EmptyFileNum", TUnit::UNIT, 1);
_not_found_file_counter = ADD_COUNTER_WITH_LEVEL(_local_state->scanner_profile(),
"NotFoundFileNum", TUnit::UNIT, 1);
_file_counter =
Expand Down Expand Up @@ -384,6 +390,20 @@ Status FileScannerV2::_get_block_impl(RuntimeState* state, Block* block, bool* e
*eof = false;
continue;

@github-actions github-actions Bot Jul 14, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This still treats too many EOF statuses as valid empty splits. A Native file that has a valid header followed by an explicit zero-length first block is malformed: NativeReader::_read_next_pblock() consumes the block length, sets eof when block_len == 0 at file end, and _ensure_schema_loaded() turns the empty buffer into Status::EndOfFile. The existing RejectsZeroLengthBlockAndInvalidPBlock unit builds exactly header + uint64(0) and expects get_schema() to fail. When TableReader lazily opens that reader from get_block(), this branch would instead abort the split, increment EmptyFileNum, and continue as if the file were a valid header-only empty file. Please make the Native reader return a hard error for an explicit zero-length block, or have the scanner skip only a dedicated empty-file status that cannot also represent malformed input.

}
if (_should_skip_empty(status, _should_stop || _io_ctx->should_stop)) {
// END_OF_FILE here means the reader discovered a valid split with no data while
// opening or probing it, not that the Scanner has exhausted all splits. Examples
// are a zero-byte CSV with an explicit schema and a Doris Native file containing
// only its 12-byte header. Treat it like V1's empty-file path: finish this range,
// discard partial reader state, and let the loop fetch the next split.
RETURN_IF_ERROR(_table_reader->abort_split());
COUNTER_UPDATE(_empty_file_counter, 1);
_state->update_num_finished_scan_range(1);
_has_prepared_split = false;
block->clear_column_data(cast_set<int64_t>(_projected_columns.size()));
*eof = false;
continue;
}
RETURN_IF_ERROR(status);
}
if (*eof) {
Expand Down Expand Up @@ -428,6 +448,16 @@ Status FileScannerV2::_prepare_next_split(bool* eos) {
_state->update_num_finished_scan_range(1);
continue;
}
if (_should_skip_empty(status, _should_stop || _io_ctx->should_stop)) {
// Schema discovery can reach EOF before a split becomes prepared. A header-only Native
// file follows this path, while a reader that discovers emptiness on its first
// get_block() follows the symmetric branch in _get_block_impl(). Both paths must
// advance exactly one scan range and preserve later files in the same scan.
RETURN_IF_ERROR(_table_reader->abort_split());
COUNTER_UPDATE(_empty_file_counter, 1);
_state->update_num_finished_scan_range(1);
continue;
}
RETURN_IF_ERROR(status);
if (_table_reader->current_split_pruned()) {
_state->update_num_finished_scan_range(1);
Expand Down Expand Up @@ -527,6 +557,14 @@ bool FileScannerV2::_should_skip_not_found(const Status& status, bool ignore_not
return ignore_not_found && status.is<ErrorCode::NOT_FOUND>();
}

bool FileScannerV2::_should_skip_empty(const Status& status, bool stopped) {
// Several readers use END_OF_FILE both for a valid zero-row split and for an interrupted IO.
// For example, DeletionVectorReader returns END_OF_FILE("stop read.") after try_stop() marks
// the shared IOContext. That status must unwind the stopped scanner; counting it as an empty
// file would incorrectly finish the scan range and increment EmptyFileNum.
return !stopped && status.is<ErrorCode::END_OF_FILE>();
}

bool FileScannerV2::_should_enable_file_meta_cache() const {
return ExecEnv::GetInstance()->file_meta_cache()->enabled() &&
_split_source->num_scan_ranges() < config::max_external_file_meta_cache_num / 3;
Expand Down
3 changes: 3 additions & 0 deletions be/src/exec/scan/file_scanner_v2.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,7 @@ class FileScannerV2 final : public Scanner {
static void TEST_report_file_cache_profile(
RuntimeProfile* profile, const io::FileCacheStatistics& file_cache_statistics);
static bool TEST_should_skip_not_found(const Status& status, bool ignore_not_found);
static bool TEST_should_skip_empty(const Status& status, bool stopped);
#endif

FileScannerV2(RuntimeState* state, FileScanLocalState* parent, int64_t limit,
Expand Down Expand Up @@ -121,6 +122,7 @@ class FileScannerV2 final : public Scanner {
Status _prepare_table_reader_split(const TFileRangeDesc& range,
std::map<std::string, Field> partition_values);
static bool _should_skip_not_found(const Status& status, bool ignore_not_found);
static bool _should_skip_empty(const Status& status, bool stopped);
bool _should_enable_file_meta_cache() const;
std::optional<format::GlobalRowIdContext> _create_global_rowid_context(
const TFileRangeDesc& range) const;
Expand Down Expand Up @@ -181,6 +183,7 @@ class FileScannerV2 final : public Scanner {
ShardedKVCache* _kv_cache = nullptr;

RuntimeProfile::Counter* _get_block_timer = nullptr;
RuntimeProfile::Counter* _empty_file_counter = nullptr;
RuntimeProfile::Counter* _not_found_file_counter = nullptr;
RuntimeProfile::Counter* _file_counter = nullptr;
RuntimeProfile::Counter* _file_read_bytes_counter = nullptr;
Expand Down
15 changes: 15 additions & 0 deletions be/src/format_v2/column_mapper.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -870,6 +870,7 @@ static VExprSPtr rewrite_table_expr_to_file_expr(
return expr;
}
if (is_struct_element_expr(expr)) {
const auto table_leaf_type = expr->data_type();
if (!rewrite_struct_element_path_to_file_expr(expr, filter_mappings, global_to_file_slot,
rewrite_context)) {
// The scanner still evaluates the original table-level conjunct after TableReader
Expand All @@ -879,6 +880,20 @@ static VExprSPtr rewrite_table_expr_to_file_expr(
// parents such as `element_at(element_at(map_values(m), 1), 'field')`; only direct
// slot-rooted struct chains are supported here.
*can_localize = false;
return expr;
}
DORIS_CHECK(table_leaf_type != nullptr);
DORIS_CHECK(expr->data_type() != nullptr);
if (!expr->data_type()->equals(*table_leaf_type)) {
// Path localization changes the leaf to the physical file type. For example, after an
// Iceberg evolution from STRUCT<a: INT> to STRUCT<a: BIGINT>, the localized old-file
// predicate is initially `element_at(file_col, 'a')::INT = 10::BIGINT`. Cast only the
// leaf back to BIGINT so the comparison has matching operands without forcing a cast
// of the entire evolved struct (whose children may also have been added or reordered).
auto cast_expr = Cast::create_shared(table_leaf_type);
cast_expr->add_child(expr);
rewrite_context->add_created_expr(cast_expr);
return cast_expr;
}
return expr;
}
Expand Down
8 changes: 6 additions & 2 deletions be/src/format_v2/native/native_reader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -247,8 +247,12 @@ Status NativeReader::_read_next_pblock(std::string* buffer, bool* eof) const {
}
_current_offset += sizeof(block_len);
if (block_len == 0) {
*eof = (_current_offset >= _file_size);
return Status::OK();
// A header-only file reaches the `_current_offset >= _file_size` branch above and is a
// valid empty Native file. Once an explicit length prefix is present, however, zero is not
// an EOF marker in the Native format. For example, `header + uint64(0)` is truncated input,
// not a zero-row split, and must not escape as EOF for FileScannerV2 to count as empty.
return Status::InternalError("zero-length native block in file {} at offset {}",
_file_description->path, _current_offset - sizeof(block_len));
}

buffer->assign(block_len, '\0');
Expand Down
13 changes: 9 additions & 4 deletions be/src/format_v2/parquet/parquet_column_schema.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -358,11 +358,16 @@ Status build_node_schema_with_mode(const ::parquet::SchemaDescriptor& schema,
}
column_schema->type_descriptor = resolve_parquet_type(column_schema->descriptor);
column_schema->type = column_schema->type_descriptor.doris_type;
if (column_schema->type == nullptr &&
!column_schema->type_descriptor.unsupported_reason.empty()) {
// Keep unsupported logical leaves in the file schema using their physical storage
// type. For example, a file `{id: INT32, clock: TIME_MILLIS}` remains readable for
// `SELECT id`: schema mapping sees `clock` as its physical INT32 but never creates its
// reader. `SELECT clock` still fails explicitly in ParquetColumnReaderFactory before
// any physical value is decoded, preserving the unsupported-type contract.
column_schema->type = column_schema->type_descriptor.physical_doris_type;

@github-actions github-actions Bot Jul 14, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This fallback keeps unsupported TIME leaves in the schema as their physical type, but the rejection now happens only if a ParquetColumnReader is actually created. Some requested-column paths use metadata before that point: open() can finish with an empty scan plan after statistics/page-index pruning, and MIN/MAX pushdown reaches validate_minmax_aggregate_statistics()/TransformColumnStatistics() with this physical INT leaf and no unsupported_reason check. A query that references the unsupported column can therefore return an all-pruned result or physical min/max values instead of the promised NotSupported error. Please reject requested predicate/aggregate leaves whose type_descriptor.unsupported_reason is set before any metadata pruning or aggregate pushdown uses them.

}
if (column_schema->type == nullptr) {
if (!column_schema->type_descriptor.unsupported_reason.empty()) {
return Status::NotSupported("Unsupported parquet column '{}': {}", node.name(),
column_schema->type_descriptor.unsupported_reason);
}
return Status::NotSupported("Unsupported parquet column type for column {}",
node.name());
}
Expand Down
154 changes: 94 additions & 60 deletions be/src/format_v2/parquet/parquet_file_context.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,6 @@
#include <limits>
#include <mutex>
#include <string_view>
#include <unordered_map>
#include <utility>

#include "common/check.h"
Expand All @@ -45,6 +44,75 @@ namespace doris::format::parquet {

namespace detail {

namespace {

bool page_cache_range_less(const ParquetPageCacheRange& lhs, const ParquetPageCacheRange& rhs) {
return lhs.offset < rhs.offset || (lhs.offset == rhs.offset && lhs.size < rhs.size);
}

} // namespace

ParquetPageCacheRangeIndex::ParquetPageCacheRangeIndex(size_t max_ranges)
: _max_ranges(max_ranges) {
DORIS_CHECK(_max_ranges > 0);
}

void ParquetPageCacheRangeIndex::insert(ParquetPageCacheRange range) {

@github-actions github-actions Bot Jul 14, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The old global directory had a per-file range cap, but the replacement ParquetPageCacheRangeIndex is an uncapped vector. During a large Parquet scan with page cache enabled, every full in-scope ReadAt inserts another range, and later misses call plan_page_cache_range_read() over the reader-local vector, which copies/scans/sorts candidates while the page-cache mutex is held. Since StoragePageCache eviction does not prune this metadata, one large reader can grow this without bound until close and pay increasing CPU and memory overhead. Please restore a per-reader/file cap or prune the index as entries become stale.

std::lock_guard lock(_mutex);
auto it = std::lower_bound(_ranges.begin(), _ranges.end(), range, page_cache_range_less);
if (it != _ranges.end() && it->offset == range.offset && it->size == range.size) {
return;
}
if (_ranges.size() >= _max_ranges) {
_ranges.erase(_ranges.begin());
it = std::lower_bound(_ranges.begin(), _ranges.end(), range, page_cache_range_less);
}
_ranges.insert(it, range);
}

void ParquetPageCacheRangeIndex::erase(ParquetPageCacheRange range) {
std::lock_guard lock(_mutex);
const auto it = std::lower_bound(_ranges.begin(), _ranges.end(), range, page_cache_range_less);
if (it != _ranges.end() && it->offset == range.offset && it->size == range.size) {
_ranges.erase(it);
}
}

std::vector<ParquetPageCacheRange> ParquetPageCacheRangeIndex::ranges() const {
std::lock_guard lock(_mutex);
return _ranges;
}

size_t ParquetPageCacheRangeIndex::size() const {
std::lock_guard lock(_mutex);
return _ranges.size();
}

ParquetPageCacheRangeDirectory::ParquetPageCacheRangeDirectory(size_t max_files)
: _max_files(max_files) {
DORIS_CHECK(_max_files > 0);
}

std::shared_ptr<ParquetPageCacheRangeIndex> ParquetPageCacheRangeDirectory::get_or_create(
const std::string& file_key) {
DORIS_CHECK(!file_key.empty());
std::lock_guard lock(_mutex);
if (const auto it = _indexes.find(file_key); it != _indexes.end()) {
return it->second;
}
if (_indexes.size() >= _max_files) {
_indexes.erase(_indexes.begin());
}
auto index = std::make_shared<ParquetPageCacheRangeIndex>();
_indexes.emplace(file_key, index);
return index;
}

size_t ParquetPageCacheRangeDirectory::size() const {
std::lock_guard lock(_mutex);
return _indexes.size();
}

std::vector<ParquetPageCacheReadPlanEntry> plan_page_cache_range_read(
int64_t position, int64_t nbytes, const std::vector<ParquetPageCacheRange>& cached_ranges) {
if (position < 0 || nbytes <= 0) {
Expand Down Expand Up @@ -133,59 +201,14 @@ bool should_use_merge_range_reader(const std::vector<ParquetPageCacheRange>& ran

namespace {

// StoragePageCache only supports exact-key lookup. Keep lightweight range metadata here so later
// Arrow ReadAt requests can reuse cached bytes when their requested ranges are subsets of, or are
// fully covered by, previously cached ranges. Stale metadata is pruned on lookup.
std::mutex cached_page_range_index_mutex;
std::unordered_map<std::string, std::vector<ParquetPageCacheRange>> cached_page_range_index;
constexpr size_t MAX_CACHED_PAGE_RANGE_FILES = 4096;
constexpr size_t MAX_CACHED_PAGE_RANGES_PER_FILE = 65536;

void register_cached_page_range(const std::string& file_key, int64_t position, int64_t nbytes) {
DORIS_CHECK(nbytes > 0);
std::lock_guard lock(cached_page_range_index_mutex);
if (cached_page_range_index.find(file_key) == cached_page_range_index.end() &&
cached_page_range_index.size() >= MAX_CACHED_PAGE_RANGE_FILES) {
cached_page_range_index.erase(cached_page_range_index.begin());
}
auto& ranges = cached_page_range_index[file_key];
auto it = std::find_if(ranges.begin(), ranges.end(), [&](const ParquetPageCacheRange& range) {
return range.offset == position && range.size == nbytes;
});
if (it == ranges.end()) {
if (ranges.size() >= MAX_CACHED_PAGE_RANGES_PER_FILE) {
ranges.erase(ranges.begin());
}
ranges.push_back(ParquetPageCacheRange {position, nbytes});
}
}

void unregister_cached_page_range(const std::string& file_key,
const ParquetPageCacheRange& stale_range) {
std::lock_guard lock(cached_page_range_index_mutex);
auto it = cached_page_range_index.find(file_key);
if (it == cached_page_range_index.end()) {
return;
}
auto& ranges = it->second;
ranges.erase(std::remove_if(ranges.begin(), ranges.end(),
[&](const ParquetPageCacheRange& range) {
return range.offset == stale_range.offset &&
range.size == stale_range.size;
}),
ranges.end());
if (ranges.empty()) {
cached_page_range_index.erase(it);
}
}

std::vector<ParquetPageCacheRange> cached_page_ranges_for_file(const std::string& file_key) {
std::lock_guard lock(cached_page_range_index_mutex);
auto it = cached_page_range_index.find(file_key);
if (it == cached_page_range_index.end()) {
return {};
}
return it->second;
detail::ParquetPageCacheRangeDirectory& cached_page_range_directory() {
// Directory lookup is paid once when a reader opens. ReadAt() then synchronizes only on the
// shared index for this file, so unrelated Parquet files no longer serialize on a process-wide
// hot-path mutex. Strong references deliberately keep range discovery alive after reader A
// closes: reader B can reuse cached [100, 200) for a request [120, 150). The directory and each
// per-file index are independently capped, bounding stale metadata left by page-cache eviction.
static detail::ParquetPageCacheRangeDirectory directory;
return directory;
}

std::string build_page_cache_file_key(const io::FileReader& file_reader,
Expand Down Expand Up @@ -213,7 +236,11 @@ class DorisRandomAccessFile final : public arrow::io::RandomAccessFile {
_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)) {
_page_cache_file_key(std::move(page_cache_file_key)),
_cached_page_range_index(
_enable_page_cache && !_page_cache_file_key.empty()
? cached_page_range_directory().get_or_create(_page_cache_file_key)
: nullptr) {
DORIS_CHECK(_file_reader != nullptr);
if (auto tracing_reader = std::dynamic_pointer_cast<io::TracingFileReader>(_file_reader)) {
_file_reader_stats = tracing_reader->stats();
Expand All @@ -226,6 +253,10 @@ class DorisRandomAccessFile final : public arrow::io::RandomAccessFile {
arrow::Status Close() override {
if (!_closed) {
collect_active_merge_range_profile();
std::lock_guard lock(_page_cache_mutex);
// Page payloads and their bounded per-file range index intentionally outlive this
// reader for warm scans. Only reader-specific projected ranges are released here.
std::vector<ParquetPageCacheRange>().swap(_page_cache_ranges);
_closed = true;
}
return arrow::Status::OK();
Expand Down Expand Up @@ -363,7 +394,8 @@ class DorisRandomAccessFile final : public arrow::io::RandomAccessFile {
private:
bool page_cache_enabled() const {
return _enable_page_cache && !config::disable_storage_page_cache &&
StoragePageCache::instance() != nullptr && !_page_cache_file_key.empty();
StoragePageCache::instance() != nullptr && !_page_cache_file_key.empty() &&
_cached_page_range_index != nullptr;
}

bool range_in_page_cache_scope(int64_t position, int64_t nbytes) const {
Expand Down Expand Up @@ -391,7 +423,7 @@ class DorisRandomAccessFile final : public arrow::io::RandomAccessFile {
if (!StoragePageCache::instance()->lookup(
page_cache_key(cached_range.offset, cached_range.size), &handle,
segment_v2::DATA_PAGE)) {
unregister_cached_page_range(_page_cache_file_key, cached_range);
_cached_page_range_index->erase(cached_range);
return false;
}
Slice cached = handle.data();
Expand All @@ -404,8 +436,8 @@ class DorisRandomAccessFile final : public arrow::io::RandomAccessFile {
}

bool try_read_from_cached_ranges(int64_t position, int64_t nbytes, void* out) {
auto plan = detail::plan_page_cache_range_read(
position, nbytes, cached_page_ranges_for_file(_page_cache_file_key));
auto plan = detail::plan_page_cache_range_read(position, nbytes,
_cached_page_range_index->ranges());
if (plan.empty()) {
return false;
}
Expand Down Expand Up @@ -456,7 +488,8 @@ class DorisRandomAccessFile final : public arrow::io::RandomAccessFile {
PageCacheHandle handle;
StoragePageCache::instance()->insert(page_cache_key(position, nbytes), page, &handle,
segment_v2::DATA_PAGE);
register_cached_page_range(_page_cache_file_key, position, nbytes);
_cached_page_range_index->insert(
ParquetPageCacheRange {.offset = position, .size = nbytes});
++_page_cache_stats.write_count;
++_page_cache_stats.compressed_write_count;
}
Expand Down Expand Up @@ -514,6 +547,7 @@ class DorisRandomAccessFile final : public arrow::io::RandomAccessFile {
std::string _page_cache_file_key;
mutable std::mutex _page_cache_mutex;
std::vector<ParquetPageCacheRange> _page_cache_ranges;
std::shared_ptr<detail::ParquetPageCacheRangeIndex> _cached_page_range_index;
ParquetPageCacheStats _page_cache_stats;
};

Expand Down
Loading
Loading