From 57aa8de164269b487a79b6ca6ba0e1d1a393fefe Mon Sep 17 00:00:00 2001 From: Gabriel Date: Mon, 13 Jul 2026 17:10:13 +0800 Subject: [PATCH 1/5] [fix](be) Harden FileScannerV2 schema and reader edge cases ### What problem does this PR solve? Issue Number: None Related PR: None Problem Summary: FileScannerV2 treated valid empty splits as scan failures, localized promoted nested struct predicates with mismatched operand types, retained a process-global Parquet page-cache range index, and rejected whole Parquet schemas when unprojected leaves used unsupported logical types. This change skips and counts empty splits, casts localized nested leaves back to table types, scopes range metadata to each file reader, and defers unsupported-type errors until projection. Regression fixtures and commented examples cover empty CSV input, mixed Parquet struct leaf types, repeated multi-file scans, and an unprojected TIME_MILLIS column. ### Release note Fix FileScannerV2 compatibility for empty files, evolved nested predicates, and Parquet files containing unprojected unsupported columns; reduce Parquet page-cache range-index contention and lifetime. ### Check List (For Author) - Test: Regression test / Unit Test - 55 targeted BE ASAN unit tests - test_file_scanner_v2_review_fixes regression suite - Behavior changed: Yes (valid empty splits are skipped; unsupported Parquet logical leaves fail only when projected) - Does this need documentation: No --- be/src/exec/scan/file_scanner_v2.cpp | 34 +++++ be/src/exec/scan/file_scanner_v2.h | 3 + be/src/format_v2/column_mapper.cpp | 15 +++ .../parquet/parquet_column_schema.cpp | 13 +- .../parquet/parquet_file_context.cpp | 97 ++++++-------- .../format_v2/parquet/parquet_file_context.h | 16 +++ be/src/format_v2/parquet/parquet_type.cpp | 3 +- be/src/format_v2/parquet/parquet_type.h | 3 + be/test/exec/scan/file_scanner_v2_test.cpp | 6 + be/test/format_v2/column_mapper_test.cpp | 125 ++++++++++++++++++ .../format_v2/native/native_reader_test.cpp | 5 +- .../parquet/parquet_page_cache_range_test.cpp | 23 ++++ .../parquet/parquet_reader_control_test.cpp | 14 ++ .../format_v2/parquet/parquet_schema_test.cpp | 26 ++-- .../tvf/file_scanner_v2_empty.csv | 0 .../file_scanner_v2_struct_0_bigint.parquet | Bin 0 -> 965 bytes .../tvf/file_scanner_v2_struct_1_int.parquet | Bin 0 -> 923 bytes .../file_scanner_v2_unsupported_time.parquet | Bin 0 -> 689 bytes .../tvf/test_file_scanner_v2_review_fixes.out | 14 ++ .../test_file_scanner_v2_review_fixes.groovy | 84 ++++++++++++ 20 files changed, 405 insertions(+), 76 deletions(-) create mode 100644 regression-test/data/external_table_p0/tvf/file_scanner_v2_empty.csv create mode 100644 regression-test/data/external_table_p0/tvf/file_scanner_v2_struct_0_bigint.parquet create mode 100644 regression-test/data/external_table_p0/tvf/file_scanner_v2_struct_1_int.parquet create mode 100644 regression-test/data/external_table_p0/tvf/file_scanner_v2_unsupported_time.parquet create mode 100644 regression-test/data/external_table_p0/tvf/test_file_scanner_v2_review_fixes.out create mode 100644 regression-test/suites/external_table_p0/tvf/test_file_scanner_v2_review_fixes.groovy diff --git a/be/src/exec/scan/file_scanner_v2.cpp b/be/src/exec/scan/file_scanner_v2.cpp index 6700ed71276f45..7ed46e7464d1cc 100644 --- a/be/src/exec/scan/file_scanner_v2.cpp +++ b/be/src/exec/scan/file_scanner_v2.cpp @@ -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) { + return _should_skip_empty(status); +} #endif bool FileScannerV2::is_supported(const TFileScanRangeParams& params, const TFileRangeDesc& range) { @@ -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 = @@ -384,6 +390,20 @@ Status FileScannerV2::_get_block_impl(RuntimeState* state, Block* block, bool* e *eof = false; continue; } + if (_should_skip_empty(status)) { + // 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(_projected_columns.size())); + *eof = false; + continue; + } RETURN_IF_ERROR(status); } if (*eof) { @@ -428,6 +448,16 @@ Status FileScannerV2::_prepare_next_split(bool* eos) { _state->update_num_finished_scan_range(1); continue; } + if (_should_skip_empty(status)) { + // 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); @@ -527,6 +557,10 @@ bool FileScannerV2::_should_skip_not_found(const Status& status, bool ignore_not return ignore_not_found && status.is(); } +bool FileScannerV2::_should_skip_empty(const Status& status) { + return status.is(); +} + 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; diff --git a/be/src/exec/scan/file_scanner_v2.h b/be/src/exec/scan/file_scanner_v2.h index 2bddc5d5e69e6c..b7e5743bfa572d 100644 --- a/be/src/exec/scan/file_scanner_v2.h +++ b/be/src/exec/scan/file_scanner_v2.h @@ -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); #endif FileScannerV2(RuntimeState* state, FileScanLocalState* parent, int64_t limit, @@ -121,6 +122,7 @@ class FileScannerV2 final : public Scanner { Status _prepare_table_reader_split(const TFileRangeDesc& range, std::map partition_values); static bool _should_skip_not_found(const Status& status, bool ignore_not_found); + static bool _should_skip_empty(const Status& status); bool _should_enable_file_meta_cache() const; std::optional _create_global_rowid_context( const TFileRangeDesc& range) const; @@ -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; diff --git a/be/src/format_v2/column_mapper.cpp b/be/src/format_v2/column_mapper.cpp index 572e76280a0e37..6daa256197de85 100644 --- a/be/src/format_v2/column_mapper.cpp +++ b/be/src/format_v2/column_mapper.cpp @@ -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 @@ -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 to STRUCT, 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; } diff --git a/be/src/format_v2/parquet/parquet_column_schema.cpp b/be/src/format_v2/parquet/parquet_column_schema.cpp index b42d47987a54cb..1cdfed80bd273b 100644 --- a/be/src/format_v2/parquet/parquet_column_schema.cpp +++ b/be/src/format_v2/parquet/parquet_column_schema.cpp @@ -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; + } 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()); } diff --git a/be/src/format_v2/parquet/parquet_file_context.cpp b/be/src/format_v2/parquet/parquet_file_context.cpp index d80dc58181d4f6..00b0f37b88aca9 100644 --- a/be/src/format_v2/parquet/parquet_file_context.cpp +++ b/be/src/format_v2/parquet/parquet_file_context.cpp @@ -27,7 +27,6 @@ #include #include #include -#include #include #include "common/check.h" @@ -45,6 +44,32 @@ 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 + +void ParquetPageCacheRangeIndex::insert(ParquetPageCacheRange range) { + 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.insert(it, range); + } +} + +void ParquetPageCacheRangeIndex::erase(ParquetPageCacheRange range) { + 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); + } +} + +void ParquetPageCacheRangeIndex::clear() { + std::vector().swap(_ranges); +} + std::vector plan_page_cache_range_read( int64_t position, int64_t nbytes, const std::vector& cached_ranges) { if (position < 0 || nbytes <= 0) { @@ -133,61 +158,6 @@ bool should_use_merge_range_reader(const std::vector& 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> 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 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; -} - std::string build_page_cache_file_key(const io::FileReader& file_reader, const io::FileDescription& file_description) { const int64_t mtime = @@ -226,6 +196,12 @@ 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); + // The cache payload may outlive this reader in StoragePageCache, but the auxiliary + // range directory must not. For example, after file A closes, file B must not scan A's + // ranges or retain their metadata until process shutdown. + _cached_page_range_index.clear(); + std::vector().swap(_page_cache_ranges); _closed = true; } return arrow::Status::OK(); @@ -391,7 +367,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(); @@ -404,8 +380,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; } @@ -456,7 +432,7 @@ 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; } @@ -514,6 +490,7 @@ class DorisRandomAccessFile final : public arrow::io::RandomAccessFile { std::string _page_cache_file_key; mutable std::mutex _page_cache_mutex; std::vector _page_cache_ranges; + detail::ParquetPageCacheRangeIndex _cached_page_range_index; ParquetPageCacheStats _page_cache_stats; }; diff --git a/be/src/format_v2/parquet/parquet_file_context.h b/be/src/format_v2/parquet/parquet_file_context.h index 0dca52244957d7..80a45aa0d273d0 100644 --- a/be/src/format_v2/parquet/parquet_file_context.h +++ b/be/src/format_v2/parquet/parquet_file_context.h @@ -66,6 +66,22 @@ struct ParquetPageCacheStats { namespace detail { +class ParquetPageCacheRangeIndex { +public: + void insert(ParquetPageCacheRange range); + void erase(ParquetPageCacheRange range); + void clear(); + + const std::vector& ranges() const { return _ranges; } + +private: + // This index belongs to one DorisRandomAccessFile. Thus two files reading [0, 4KB) never + // contend on or observe the same range metadata, and closing the reader releases all entries. + // StoragePageCache remains process-wide; only its lightweight exact-range directory is scoped + // to the reader that inserted and can validate those entries. + std::vector _ranges; +}; + // Build the copy plan for a ReadAt(position, nbytes) request from the range metadata of // previously cached entries. // StoragePageCache cannot do range lookup by itself; it can only lookup an exact key. The diff --git a/be/src/format_v2/parquet/parquet_type.cpp b/be/src/format_v2/parquet/parquet_type.cpp index d35181d0397178..8411d53f0fba91 100644 --- a/be/src/format_v2/parquet/parquet_type.cpp +++ b/be/src/format_v2/parquet/parquet_type.cpp @@ -293,6 +293,7 @@ ParquetTypeDescriptor resolve_parquet_type(const ::parquet::ColumnDescriptor* co result.physical_type = column->physical_type(); result.converted_type = column->converted_type(); result.fixed_length = column->type_length(); + result.physical_doris_type = physical_type_to_doris_type(column); if (auto logical_type = logical_type_to_doris_type(column, &result); logical_type != nullptr) { result.doris_type = logical_type; @@ -306,7 +307,7 @@ ParquetTypeDescriptor resolve_parquet_type(const ::parquet::ColumnDescriptor* co result.doris_type = nullptr; result.supports_record_reader = false; } else { - result.doris_type = physical_type_to_doris_type(column); + result.doris_type = result.physical_doris_type; if (result.physical_type == ::parquet::Type::INT96) { result.extra_type_info = ParquetExtraTypeInfo::IMPALA_TIMESTAMP; } diff --git a/be/src/format_v2/parquet/parquet_type.h b/be/src/format_v2/parquet/parquet_type.h index 5d21aae6bae092..a4d99abc0e982a 100644 --- a/be/src/format_v2/parquet/parquet_type.h +++ b/be/src/format_v2/parquet/parquet_type.h @@ -54,6 +54,9 @@ enum class ParquetTimeUnit { // ============================================================================ struct ParquetTypeDescriptor { DataTypePtr doris_type; + // Physical fallback used only to keep file schema construction alive when the logical type is + // unsupported. Column reader creation still rejects unsupported_reason before decoding. + DataTypePtr physical_doris_type; ParquetExtraTypeInfo extra_type_info = ParquetExtraTypeInfo::NONE; ParquetTimeUnit time_unit = ParquetTimeUnit::UNKNOWN; ::parquet::Type::type physical_type = ::parquet::Type::UNDEFINED; diff --git a/be/test/exec/scan/file_scanner_v2_test.cpp b/be/test/exec/scan/file_scanner_v2_test.cpp index abd3d1e76d7ab3..9e31d2f6e7d029 100644 --- a/be/test/exec/scan/file_scanner_v2_test.cpp +++ b/be/test/exec/scan/file_scanner_v2_test.cpp @@ -494,6 +494,12 @@ TEST(FileScannerV2Test, NotFoundIsSkippedOnlyWhenConfigured) { EXPECT_FALSE(FileScannerV2::TEST_should_skip_not_found(Status::OK(), true)); } +TEST(FileScannerV2Test, EndOfFileIsSkippedAsEmptySplit) { + EXPECT_TRUE(FileScannerV2::TEST_should_skip_empty(Status::EndOfFile("empty file"))); + EXPECT_FALSE(FileScannerV2::TEST_should_skip_empty(Status::InternalError("read failed"))); + EXPECT_FALSE(FileScannerV2::TEST_should_skip_empty(Status::OK())); +} + // 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/column_mapper_test.cpp b/be/test/format_v2/column_mapper_test.cpp index 3885deb107f072..64b16895b7b2d4 100644 --- a/be/test/format_v2/column_mapper_test.cpp +++ b/be/test/format_v2/column_mapper_test.cpp @@ -28,6 +28,7 @@ #include "common/consts.h" #include "core/assert_cast.h" #include "core/block/block.h" +#include "core/column/column_struct.h" #include "core/column/column_vector.h" #include "core/data_type/data_type_array.h" #include "core/data_type/data_type_decimal.h" @@ -38,6 +39,7 @@ #include "core/data_type/data_type_struct.h" #include "core/data_type/data_type_timestamptz.h" #include "core/data_type/data_type_varbinary.h" +#include "exprs/vectorized_fn_call.h" #include "exprs/vexpr.h" #include "exprs/vexpr_context.h" #include "exprs/vin_predicate.h" @@ -316,6 +318,38 @@ class TestFunctionExpr final : public VExpr { std::string _expr_name; }; +class ExecutableStructElementExpr final : public VExpr { +public: + explicit ExecutableStructElementExpr(DataTypePtr child_type) + : VExpr(std::move(child_type), false) { + set_node_type(TExprNodeType::FUNCTION_CALL); + TFunctionName fn_name; + fn_name.__set_function_name(_expr_name); + _fn.__set_name(fn_name); + } + + const std::string& expr_name() const override { return _expr_name; } + + Status clone_node(VExprSPtr* cloned_expr) const override { + DORIS_CHECK(cloned_expr != nullptr); + *cloned_expr = std::make_shared(data_type()); + return Status::OK(); + } + + Status execute_column_impl(VExprContext* context, const Block* block, const Selector* selector, + size_t count, ColumnPtr& result_column) const override { + ColumnPtr struct_column; + RETURN_IF_ERROR( + get_child(0)->execute_column(context, block, selector, count, struct_column)); + const auto& input = assert_cast(*struct_column); + result_column = input.get_column_ptr(0); + return Status::OK(); + } + +private: + const std::string _expr_name = "element_at"; +}; + VExprSPtr table_slot(int slot_id, int column_id, DataTypePtr type, const std::string& name) { return VSlotRef::create_shared(slot_id, column_id, -1, std::move(type), name); } @@ -340,6 +374,40 @@ VExprSPtr element_at(const VExprSPtr& parent, DataTypePtr child_type, return expr; } +VExprSPtr executable_struct_element(const VExprSPtr& parent, DataTypePtr child_type, + const std::string& child_name) { + auto expr = std::make_shared(std::move(child_type)); + expr->add_child(parent); + expr->add_child(literal(str(), Field::create_field(child_name))); + return expr; +} + +VExprSPtr executable_binary_predicate(TExprOpcode::type opcode, const VExprSPtr& left, + const VExprSPtr& right) { + const auto result_type = u8(); + TFunctionName fn_name; + fn_name.__set_function_name(opcode == TExprOpcode::GT ? "gt" : "eq"); + TFunction fn; + fn.__set_name(fn_name); + fn.__set_binary_type(TFunctionBinaryType::BUILTIN); + fn.__set_arg_types({left->data_type()->to_thrift(), right->data_type()->to_thrift()}); + fn.__set_ret_type(result_type->to_thrift()); + fn.__set_has_var_args(false); + + TExprNode node; + node.__set_node_type(TExprNodeType::BINARY_PRED); + node.__set_opcode(opcode); + node.__set_type(result_type->to_thrift()); + node.__set_fn(fn); + node.__set_num_children(2); + node.__set_is_nullable(false); + + auto expr = VectorizedFnCall::create_shared(node); + expr->add_child(left); + expr->add_child(right); + return expr; +} + VExprSPtr array_element_at(const VExprSPtr& parent, DataTypePtr child_type, int64_t ordinal) { auto expr = std::make_shared("element_at", std::move(child_type)); expr->add_child(parent); @@ -2992,6 +3060,63 @@ TEST(ColumnMapperScanRequestTest, NestedElementAtConjunctUsesFileChildTypeForRen EXPECT_EQ(localized_parent_type->get_element_name(1), "bb"); } +// Scenario: Iceberg promotes a nested struct leaf from INT to BIGINT while an old file still +// stores INT. The localized element_at must be cast back to the table leaf type so the comparison +// is prepared and executed with matching operand types. +TEST_F(ColumnMapperCastTest, NestedElementAtConjunctCastsPromotedFileLeafToTableType) { + const auto file_int_type = i32(); + const auto table_bigint_type = i64(); + + auto table_b = field_id_col("b", 11, table_bigint_type); + auto table_struct = struct_col("s", 10, {table_b}); + auto file_b = field_id_col("b", 11, file_int_type, 0); + auto file_struct = struct_col("s", 10, {file_b}, 5); + + auto table_leaf = executable_struct_element( + table_slot(0, 0, table_struct.type, table_struct.name), table_bigint_type, "b"); + auto filter_expr = executable_binary_predicate( + TExprOpcode::GT, table_leaf, + literal(table_bigint_type, Field::create_field(15))); + TableFilter filter {.conjunct = VExprContext::create_shared(filter_expr), + .global_indices = {GlobalIndex(0)}}; + + TableColumnMapper mapper({.mode = TableColumnMappingMode::BY_FIELD_ID}); + ASSERT_TRUE(mapper.create_mapping({table_struct}, {}, {file_struct}).ok()); + + FileScanRequest request; + ASSERT_TRUE(mapper.create_scan_request({filter}, {table_struct}, &request, &state).ok()); + ASSERT_EQ(request.conjuncts.size(), 1); + const auto& localized_root = request.conjuncts[0]->root(); + ASSERT_EQ(localized_root->get_num_children(), 2); + const auto& localized_cast = localized_root->children()[0]; + ASSERT_NE(dynamic_cast(localized_cast.get()), nullptr); + EXPECT_TRUE(localized_cast->data_type()->equals(*table_bigint_type)); + ASSERT_EQ(localized_cast->get_num_children(), 1); + EXPECT_EQ(localized_cast->children()[0]->expr_name(), "element_at"); + EXPECT_TRUE(localized_cast->children()[0]->data_type()->equals(*file_int_type)); + + auto values = ColumnInt32::create(); + values->insert_value(10); + values->insert_value(20); + MutableColumns children; + children.push_back(std::move(values)); + Block block; + block.insert({ColumnStruct::create(std::move(children)), mapper.mappings()[0].file_type, "s"}); + + auto* conjunct = request.conjuncts[0].get(); + auto status = conjunct->prepare(&state, RowDescriptor()); + ASSERT_TRUE(status.ok()) << status; + status = conjunct->open(&state); + ASSERT_TRUE(status.ok()) << status; + IColumn::Filter result(block.rows(), 1); + bool can_filter_all = false; + status = conjunct->execute_filter(&block, result.data(), block.rows(), false, &can_filter_all); + ASSERT_TRUE(status.ok()) << status; + EXPECT_FALSE(can_filter_all); + EXPECT_EQ(result, IColumn::Filter({0, 1})); + conjunct->close(); +} + // Scenario: output projection reads one struct child while the row filter reads a different nested // struct child. File-local conjunct rewrite must use the merged scan projection type. In the SQL // shape below, `SELECT element_at(s, 'c') WHERE element_at(element_at(s, 'b'), 'cc') LIKE ...` diff --git a/be/test/format_v2/native/native_reader_test.cpp b/be/test/format_v2/native/native_reader_test.cpp index aaa7aa90e0681e..6b0c9ab27b90a8 100644 --- a/be/test/format_v2/native/native_reader_test.cpp +++ b/be/test/format_v2/native/native_reader_test.cpp @@ -295,7 +295,7 @@ TEST(NativeV2ReaderTest, RejectsInvalidHeaderAndEmptyFile) { static_cast(io::global_local_filesystem()->delete_file(empty_path)); } -TEST(NativeV2ReaderTest, RejectsUnsupportedVersionAndHeaderOnlyFile) { +TEST(NativeV2ReaderTest, RejectsUnsupportedVersionAndReportsHeaderOnlyFileAsEmpty) { std::filesystem::create_directories("./log"); RuntimeState state; RuntimeProfile profile("native_v2_reader_header_boundary_test"); @@ -322,7 +322,8 @@ TEST(NativeV2ReaderTest, RejectsUnsupportedVersionAndHeaderOnlyFile) { auto header_only_reader = create_reader(header_only_path, &state, &profile); ASSERT_TRUE(header_only_reader->init(&state).ok()); std::vector schema; - EXPECT_FALSE(header_only_reader->get_schema(&schema).ok()); + const auto header_only_status = header_only_reader->get_schema(&schema); + EXPECT_TRUE(header_only_status.is()) << header_only_status; static_cast(io::global_local_filesystem()->delete_file(header_only_path)); } 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 fac89b31c422d8..f06c3fd4660c09 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 @@ -115,6 +115,29 @@ TEST(ParquetPageCacheRangeTest, InvalidRequestMisses) { EXPECT_TRUE(detail::plan_page_cache_range_read(100, -1, cached_ranges).empty()); } +TEST(ParquetPageCacheRangeTest, PerFileRangeIndexSortsDeduplicatesAndClears) { + detail::ParquetPageCacheRangeIndex index; + index.insert({200, 20}); + index.insert({100, 30}); + index.insert({100, 10}); + index.insert({100, 30}); + + ASSERT_EQ(index.ranges().size(), 3); + EXPECT_EQ(index.ranges()[0].offset, 100); + EXPECT_EQ(index.ranges()[0].size, 10); + EXPECT_EQ(index.ranges()[1].offset, 100); + EXPECT_EQ(index.ranges()[1].size, 30); + EXPECT_EQ(index.ranges()[2].offset, 200); + + index.erase({100, 30}); + ASSERT_EQ(index.ranges().size(), 2); + EXPECT_EQ(index.ranges()[0].size, 10); + EXPECT_EQ(index.ranges()[1].offset, 200); + + index.clear(); + EXPECT_TRUE(index.ranges().empty()); +} + TEST(ParquetPageCacheRangeTest, ValidPrefetchRangesSkipInvalidAndOverflowRanges) { const std::vector ranges = { {100, 50}, diff --git a/be/test/format_v2/parquet/parquet_reader_control_test.cpp b/be/test/format_v2/parquet/parquet_reader_control_test.cpp index a21974bced294d..63197a412c5a2d 100644 --- a/be/test/format_v2/parquet/parquet_reader_control_test.cpp +++ b/be/test/format_v2/parquet/parquet_reader_control_test.cpp @@ -1152,6 +1152,20 @@ TEST(ParquetColumnReaderFactoryTest, RejectsInvalidLeafIdBeforeCreatingRecordRea EXPECT_NE(status.to_string().find("Invalid parquet leaf column id"), std::string::npos); } +TEST(ParquetColumnReaderFactoryTest, RejectsProjectedUnsupportedLogicalType) { + ParquetColumnSchema schema = int64_schema("unsupported_time"); + schema.kind = ParquetColumnSchemaKind::PRIMITIVE; + schema.type_descriptor.unsupported_reason = + "Parquet TIME with isAdjustedToUTC=true is not supported"; + + ParquetColumnReaderFactory factory(nullptr, 1); + std::unique_ptr reader; + const auto status = factory.create(schema, &reader); + EXPECT_FALSE(status.ok()); + EXPECT_NE(status.to_string().find(schema.type_descriptor.unsupported_reason), + std::string::npos); +} + TEST(ParquetColumnReaderFactoryTest, RejectsStructInvalidAndEmptyProjection) { auto schema = struct_schema_for_projection(); ParquetColumnReaderFactory factory(nullptr, 0); diff --git a/be/test/format_v2/parquet/parquet_schema_test.cpp b/be/test/format_v2/parquet/parquet_schema_test.cpp index e620ed718efbf2..9f78735bfb560a 100644 --- a/be/test/format_v2/parquet/parquet_schema_test.cpp +++ b/be/test/format_v2/parquet/parquet_schema_test.cpp @@ -414,7 +414,7 @@ TEST(ParquetSchemaTest, BuildEntryValidatesNullPointerAndEmptyRoot) { EXPECT_TRUE(fields.empty()); } -TEST(ParquetSchemaTest, RejectInvalidListMapAndUnsupportedTime) { +TEST(ParquetSchemaTest, RejectInvalidListMapAndPreserveUnsupportedTime) { const auto bad_list = ::parquet::schema::GroupNode::Make( "bad_list", ::parquet::Repetition::OPTIONAL, {::parquet::schema::PrimitiveNode::Make("item", ::parquet::Repetition::OPTIONAL, @@ -432,9 +432,11 @@ TEST(ParquetSchemaTest, RejectInvalidListMapAndUnsupportedTime) { const auto converted_time = ::parquet::schema::PrimitiveNode::Make( "time_ms", ::parquet::Repetition::REQUIRED, ::parquet::Type::INT32, ::parquet::ConvertedType::TIME_MILLIS); - const auto status = build_status({converted_time}); - EXPECT_FALSE(status.ok()); - EXPECT_NE(status.to_string().find("Parquet TIME with isAdjustedToUTC=true is not supported"), + const auto fields = build_fields({converted_time}); + ASSERT_EQ(fields.size(), 1); + EXPECT_EQ(remove_nullable(fields[0]->type)->get_primitive_type(), TYPE_INT); + EXPECT_NE(fields[0]->type_descriptor.unsupported_reason.find( + "Parquet TIME with isAdjustedToUTC=true is not supported"), std::string::npos); } @@ -513,15 +515,21 @@ TEST(ParquetSchemaTest, RejectAdditionalInvalidListAndMapLayouts) { EXPECT_FALSE(build_status({repeated_map}).ok()); } -TEST(ParquetSchemaTest, LogicalUtcTimeIsRejected) { +TEST(ParquetSchemaTest, LogicalUtcTimeIsPreservedForProjection) { const auto adjusted_time = ::parquet::schema::PrimitiveNode::Make( "time_ms", ::parquet::Repetition::REQUIRED, ::parquet::LogicalType::Time(true, ::parquet::LogicalType::TimeUnit::MILLIS), ::parquet::Type::INT32); - const auto status = build_status({adjusted_time}); - EXPECT_FALSE(status.ok()); - EXPECT_NE(status.to_string().find("Parquet TIME with isAdjustedToUTC=true is not supported"), - std::string::npos); + const auto supported_value = ::parquet::schema::PrimitiveNode::Make( + "value", ::parquet::Repetition::REQUIRED, ::parquet::Type::INT64); + const auto row = ::parquet::schema::GroupNode::Make("row", ::parquet::Repetition::OPTIONAL, + {adjusted_time, supported_value}); + const auto fields = build_fields({row}); + ASSERT_EQ(fields.size(), 1); + ASSERT_EQ(fields[0]->children.size(), 2); + EXPECT_EQ(remove_nullable(fields[0]->children[0]->type)->get_primitive_type(), TYPE_INT); + EXPECT_FALSE(fields[0]->children[0]->type_descriptor.unsupported_reason.empty()); + EXPECT_EQ(remove_nullable(fields[0]->children[1]->type)->get_primitive_type(), TYPE_BIGINT); } } // namespace doris::format::parquet diff --git a/regression-test/data/external_table_p0/tvf/file_scanner_v2_empty.csv b/regression-test/data/external_table_p0/tvf/file_scanner_v2_empty.csv new file mode 100644 index 00000000000000..e69de29bb2d1d6 diff --git a/regression-test/data/external_table_p0/tvf/file_scanner_v2_struct_0_bigint.parquet b/regression-test/data/external_table_p0/tvf/file_scanner_v2_struct_0_bigint.parquet new file mode 100644 index 0000000000000000000000000000000000000000..22eff58b7a23895a7e1d7f2e4323d6f5b1f9529d GIT binary patch literal 965 zcmaJ=-AWrl7@Zx5B}i|C&N2ggksB8a(Q2usl=b4td~q|Z`(h<;}#o1~a_NM_FWbIzRkVB6H)(?Z8OZfM0I)M0Ft6hask@E(b+rD$V= zQz)tcMIh!hbYqEJd&*fXQ(N}{A}g{BF`SxOA73G*I29DW6!b)-Y;4m+#9G?TP*}8U&PI39T~bT`I}sbR zT~-q5v(P3&u>}YzC$xnNgTiUu2hn|yd=FByVjBOP(O>efc=R)UZ`AAe=3akFHoF+l zqhYO5i+-#{Dyi`yY;2xTxBK?}8$UhsAfIEE!;FYJ{Kq;aDk&S%5z&yELzm0A;S8ur zbb^!P>f{K}cS4AcKe6a*2KjjlF_U=ZZiim(u6pOS;6B0{<|_(@`bwgfH*FDqHT;A@Xcye(6l_82M#A z+}gUjvn%K*)FEjaaq3|Om9Zx&-E8CcXoGn RZ5j*Vf8sZPfNlPO{{Z@npC14K literal 0 HcmV?d00001 diff --git a/regression-test/data/external_table_p0/tvf/file_scanner_v2_struct_1_int.parquet b/regression-test/data/external_table_p0/tvf/file_scanner_v2_struct_1_int.parquet new file mode 100644 index 0000000000000000000000000000000000000000..5a9ca0baa47e892df900ff12a74417c3db63f91f GIT binary patch literal 923 zcmaJ=%TC)s6ulls7DC;y7;8rIA{(M00fnkmRTUQ3!3_vOlT=kV-8gBKra%+I!xBrr zpr6-|sQLvgxM!Ti&>%+ko%=fHKBQ@JXUC9y(BWC+9PXqRa{Db z9UzFbMQ}~LRJLJ)yP@b3WN4Uqts)*~>22pH|Y?UD^Idq4W;nbBm_nzCy6 zopzO7>W$83t{^KyL6;)!Vpp$HY?NCI$syf{H@b~(tIPElCwH;5D@;^h4_(y{UlP?& zFv#XycY%T2_ioDDWva3Nl%)v1tcf+ z5`3vI^~Kw{$mYpS*T>u7l@8|kh6g1+zrcGC)MNgdBKD@vR`5_$5JO-XmKX tXOoNT{?%r0Jl@PElTlu|=}#_)qhFObZ@0F$wp~3YDn9xn53D86@&QX@oumK& literal 0 HcmV?d00001 diff --git a/regression-test/data/external_table_p0/tvf/file_scanner_v2_unsupported_time.parquet b/regression-test/data/external_table_p0/tvf/file_scanner_v2_unsupported_time.parquet new file mode 100644 index 0000000000000000000000000000000000000000..fe1daf97c37e4e6cfa72a93adb57f9e3ac952a20 GIT binary patch literal 689 zcmaJ<%TB^T6rI*dW#K|hrfHKd*wAPM1f%g0Hd6vYjfxmU!p0yph=EeeDPi6a8JMj&PD z%__SBp6M?a2SpK&f~Z@@cVB& zj)O}f&27idxXcl{oi>lzNJfb9BBy^b|??vP^X>qEae}j8J$AZVK(HH*E*?ezS6xZ#4>iLdXhQ zR6oK4`GB7)`C5RKprxdmJaxX*kNt#AbiglsgFlP?D%!n@DljV*wxZXGKhCzpFx!fv f;Vd=vqfu`dqz;dA`CQ)4;D6IE-^XLC;j#Vz9r1=H literal 0 HcmV?d00001 diff --git a/regression-test/data/external_table_p0/tvf/test_file_scanner_v2_review_fixes.out b/regression-test/data/external_table_p0/tvf/test_file_scanner_v2_review_fixes.out new file mode 100644 index 00000000000000..746fd355f15392 --- /dev/null +++ b/regression-test/data/external_table_p0/tvf/test_file_scanner_v2_review_fixes.out @@ -0,0 +1,14 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !empty_csv -- + +-- !struct_leaf_promotion_cold -- +1 10 100 +3 10 300 + +-- !struct_leaf_promotion_warm -- +1 10 100 +3 10 300 + +-- !unprojected_unsupported_time -- +1 +2 diff --git a/regression-test/suites/external_table_p0/tvf/test_file_scanner_v2_review_fixes.groovy b/regression-test/suites/external_table_p0/tvf/test_file_scanner_v2_review_fixes.groovy new file mode 100644 index 00000000000000..ac8359d980a023 --- /dev/null +++ b/regression-test/suites/external_table_p0/tvf/test_file_scanner_v2_review_fixes.groovy @@ -0,0 +1,84 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +suite("test_file_scanner_v2_review_fixes", "p0,external") { + def backends = sql "show backends" + assertTrue(backends.size() > 0) + def backendId = backends[0][0] + def dataPath = context.config.dataPath + "/external_table_p0/tvf" + // A developer may point this at fixtures already visible to every BE (for example, a shared + // checkout mounted at /tmp). CI leaves it unset and exercises the normal per-BE distribution. + def predeployedFixturePath = context.config.otherConfigs.get("fileScannerV2FixturePath") + def remotePath = predeployedFixturePath ?: "/tmp/file_scanner_v2_review_fixes" + def fixtureNames = [ + "file_scanner_v2_empty.csv", + "file_scanner_v2_struct_0_bigint.parquet", + "file_scanner_v2_struct_1_int.parquet", + "file_scanner_v2_unsupported_time.parquet" + ] + + if (predeployedFixturePath == null) { + mkdirRemotePathOnAllBE("root", remotePath) + for (def backend : backends) { + for (def fixtureName : fixtureNames) { + scpFiles("root", backend[1], "${dataPath}/${fixtureName}", remotePath, false) + } + } + } + + sql "set enable_file_scanner_v2 = true" + + // A zero-byte CSV is a valid zero-row split when the schema is supplied explicitly. The EOF + // raised during reader preparation must skip this split instead of failing the query. + order_qt_empty_csv """ + SELECT id, name + FROM local( + "file_path" = "${remotePath}/file_scanner_v2_empty.csv", + "backend_id" = "${backendId}", + "format" = "csv", + "csv_schema" = "id:int;name:string") + ORDER BY id + """ + + // The first file defines STRUCT, while the second stores a as INT. The predicate is + // localized independently for each split, so the INT leaf must be cast back to BIGINT before + // comparison. Repeating the multi-file read also exercises per-reader page-cache range state: + // closing one file must not leak its range directory into the next file or the warm scan. + def evolvedStructQuery = """ + SELECT id, col.a, col.b + FROM local( + "file_path" = "${remotePath}/file_scanner_v2_struct_*.parquet", + "backend_id" = "${backendId}", + "format" = "parquet") + WHERE col.a = CAST(10 AS BIGINT) + ORDER BY id + """ + order_qt_struct_leaf_promotion_cold evolvedStructQuery + order_qt_struct_leaf_promotion_warm evolvedStructQuery + + // TIME_MILLIS is intentionally unsupported by the record reader. It must not prevent schema + // construction when only the supported id column is projected. + order_qt_unprojected_unsupported_time """ + SELECT id + FROM local( + "file_path" = "${remotePath}/file_scanner_v2_unsupported_time.parquet", + "backend_id" = "${backendId}", + "format" = "parquet") + ORDER BY id + """ + +} From 038d71d4625cd92fa9e96d052b088e183b151e32 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Tue, 14 Jul 2026 10:06:32 +0800 Subject: [PATCH 2/5] [fix](be) Address FileScannerV2 review edge cases ### What problem does this PR solve? Issue Number: None Related PR: None Problem Summary: Stopped scans could be accounted as empty files, unsupported Parquet logical leaves could reach metadata pruning before rejection, and the reader-local page-cache range index lost warm non-exact reuse while growing without a bound. Gate empty EOF handling on scanner stop state, validate requested leaves before pruning and aggregate metadata, and share bounded per-file range indexes through a bounded directory outside the ReadAt global hot path. ### Release note Preserve correct FileScannerV2 stop handling and Parquet unsupported-type errors while retaining bounded cross-reader page-cache range reuse. ### Check List (For Author) - Test: Regression test / Unit Test - 56 targeted BE ASAN unit tests - test_file_scanner_v2_review_fixes regression suite - Behavior changed: Yes (stop EOF is not counted as an empty file; unsupported requested leaves fail before metadata pruning) - Does this need documentation: No --- be/src/exec/scan/file_scanner_v2.cpp | 16 ++-- be/src/exec/scan/file_scanner_v2.h | 4 +- .../parquet/parquet_file_context.cpp | 87 +++++++++++++++--- .../format_v2/parquet/parquet_file_context.h | 32 +++++-- be/src/format_v2/parquet/parquet_reader.cpp | 73 +++++++++++++++ be/test/exec/scan/file_scanner_v2_test.cpp | 10 +- .../parquet/parquet_page_cache_range_test.cpp | 42 ++++++--- .../file_scanner_v2_unsupported_time.parquet | Bin 689 -> 673 bytes .../test_file_scanner_v2_review_fixes.groovy | 16 ++++ 9 files changed, 235 insertions(+), 45 deletions(-) diff --git a/be/src/exec/scan/file_scanner_v2.cpp b/be/src/exec/scan/file_scanner_v2.cpp index 7ed46e7464d1cc..f4213c43edebb9 100644 --- a/be/src/exec/scan/file_scanner_v2.cpp +++ b/be/src/exec/scan/file_scanner_v2.cpp @@ -263,8 +263,8 @@ bool FileScannerV2::TEST_should_skip_not_found(const Status& status, bool ignore return _should_skip_not_found(status, ignore_not_found); } -bool FileScannerV2::TEST_should_skip_empty(const Status& status) { - return _should_skip_empty(status); +bool FileScannerV2::TEST_should_skip_empty(const Status& status, bool stopped) { + return _should_skip_empty(status, stopped); } #endif @@ -390,7 +390,7 @@ Status FileScannerV2::_get_block_impl(RuntimeState* state, Block* block, bool* e *eof = false; continue; } - if (_should_skip_empty(status)) { + 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 @@ -448,7 +448,7 @@ Status FileScannerV2::_prepare_next_split(bool* eos) { _state->update_num_finished_scan_range(1); continue; } - if (_should_skip_empty(status)) { + 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 @@ -557,8 +557,12 @@ bool FileScannerV2::_should_skip_not_found(const Status& status, bool ignore_not return ignore_not_found && status.is(); } -bool FileScannerV2::_should_skip_empty(const Status& status) { - return status.is(); +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(); } bool FileScannerV2::_should_enable_file_meta_cache() const { diff --git a/be/src/exec/scan/file_scanner_v2.h b/be/src/exec/scan/file_scanner_v2.h index b7e5743bfa572d..452b1652bee461 100644 --- a/be/src/exec/scan/file_scanner_v2.h +++ b/be/src/exec/scan/file_scanner_v2.h @@ -87,7 +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); + static bool TEST_should_skip_empty(const Status& status, bool stopped); #endif FileScannerV2(RuntimeState* state, FileScanLocalState* parent, int64_t limit, @@ -122,7 +122,7 @@ class FileScannerV2 final : public Scanner { Status _prepare_table_reader_split(const TFileRangeDesc& range, std::map partition_values); static bool _should_skip_not_found(const Status& status, bool ignore_not_found); - static bool _should_skip_empty(const Status& status); + static bool _should_skip_empty(const Status& status, bool stopped); bool _should_enable_file_meta_cache() const; std::optional _create_global_rowid_context( const TFileRangeDesc& range) const; diff --git a/be/src/format_v2/parquet/parquet_file_context.cpp b/be/src/format_v2/parquet/parquet_file_context.cpp index 00b0f37b88aca9..b51fc89d7b5575 100644 --- a/be/src/format_v2/parquet/parquet_file_context.cpp +++ b/be/src/format_v2/parquet/parquet_file_context.cpp @@ -52,22 +52,65 @@ bool page_cache_range_less(const ParquetPageCacheRange& lhs, const ParquetPageCa } // namespace +ParquetPageCacheRangeIndex::ParquetPageCacheRangeIndex(size_t max_ranges) + : _max_ranges(max_ranges) { + DORIS_CHECK(_max_ranges > 0); +} + void ParquetPageCacheRangeIndex::insert(ParquetPageCacheRange range) { - 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.insert(it, range); + 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); } } -void ParquetPageCacheRangeIndex::clear() { - std::vector().swap(_ranges); +std::vector 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 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(); + _indexes.emplace(file_key, index); + return index; +} + +size_t ParquetPageCacheRangeDirectory::size() const { + std::lock_guard lock(_mutex); + return _indexes.size(); } std::vector plan_page_cache_range_read( @@ -158,6 +201,16 @@ bool should_use_merge_range_reader(const std::vector& ran namespace { +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, const io::FileDescription& file_description) { const int64_t mtime = @@ -183,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(_file_reader)) { _file_reader_stats = tracing_reader->stats(); @@ -197,10 +254,8 @@ class DorisRandomAccessFile final : public arrow::io::RandomAccessFile { if (!_closed) { collect_active_merge_range_profile(); std::lock_guard lock(_page_cache_mutex); - // The cache payload may outlive this reader in StoragePageCache, but the auxiliary - // range directory must not. For example, after file A closes, file B must not scan A's - // ranges or retain their metadata until process shutdown. - _cached_page_range_index.clear(); + // 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().swap(_page_cache_ranges); _closed = true; } @@ -339,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 { @@ -367,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)) { - _cached_page_range_index.erase(cached_range); + _cached_page_range_index->erase(cached_range); return false; } Slice cached = handle.data(); @@ -381,7 +437,7 @@ 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_range_index.ranges()); + _cached_page_range_index->ranges()); if (plan.empty()) { return false; } @@ -432,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); - _cached_page_range_index.insert(ParquetPageCacheRange {.offset = position, .size = nbytes}); + _cached_page_range_index->insert( + ParquetPageCacheRange {.offset = position, .size = nbytes}); ++_page_cache_stats.write_count; ++_page_cache_stats.compressed_write_count; } @@ -490,7 +547,7 @@ class DorisRandomAccessFile final : public arrow::io::RandomAccessFile { std::string _page_cache_file_key; mutable std::mutex _page_cache_mutex; std::vector _page_cache_ranges; - detail::ParquetPageCacheRangeIndex _cached_page_range_index; + std::shared_ptr _cached_page_range_index; ParquetPageCacheStats _page_cache_stats; }; diff --git a/be/src/format_v2/parquet/parquet_file_context.h b/be/src/format_v2/parquet/parquet_file_context.h index 80a45aa0d273d0..67e9102139dbf2 100644 --- a/be/src/format_v2/parquet/parquet_file_context.h +++ b/be/src/format_v2/parquet/parquet_file_context.h @@ -21,6 +21,9 @@ #include #include #include +#include +#include +#include #include #include "common/status.h" @@ -68,20 +71,37 @@ namespace detail { class ParquetPageCacheRangeIndex { public: + static constexpr size_t DEFAULT_MAX_RANGES = 65536; + + explicit ParquetPageCacheRangeIndex(size_t max_ranges = DEFAULT_MAX_RANGES); + void insert(ParquetPageCacheRange range); void erase(ParquetPageCacheRange range); - void clear(); - const std::vector& ranges() const { return _ranges; } + std::vector ranges() const; + size_t size() const; private: - // This index belongs to one DorisRandomAccessFile. Thus two files reading [0, 4KB) never - // contend on or observe the same range metadata, and closing the reader releases all entries. - // StoragePageCache remains process-wide; only its lightweight exact-range directory is scoped - // to the reader that inserted and can validate those entries. + const size_t _max_ranges; + mutable std::mutex _mutex; std::vector _ranges; }; +class ParquetPageCacheRangeDirectory { +public: + static constexpr size_t DEFAULT_MAX_FILES = 4096; + + explicit ParquetPageCacheRangeDirectory(size_t max_files = DEFAULT_MAX_FILES); + + std::shared_ptr get_or_create(const std::string& file_key); + size_t size() const; + +private: + const size_t _max_files; + mutable std::mutex _mutex; + std::unordered_map> _indexes; +}; + // Build the copy plan for a ReadAt(position, nbytes) request from the range metadata of // previously cached entries. // StoragePageCache cannot do range lookup by itself; it can only lookup an exact key. The diff --git a/be/src/format_v2/parquet/parquet_reader.cpp b/be/src/format_v2/parquet/parquet_reader.cpp index 753b3628bfa19b..37607b3df2b864 100644 --- a/be/src/format_v2/parquet/parquet_reader.cpp +++ b/be/src/format_v2/parquet/parquet_reader.cpp @@ -118,6 +118,60 @@ void collect_request_leaf_column_ids( } } +Status validate_all_projected_leaves_supported(const ParquetColumnSchema& column_schema) { + if (column_schema.kind == ParquetColumnSchemaKind::PRIMITIVE) { + if (!column_schema.type_descriptor.unsupported_reason.empty()) { + return Status::NotSupported("Unsupported parquet column '{}': {}", column_schema.name, + column_schema.type_descriptor.unsupported_reason); + } + return Status::OK(); + } + for (const auto& child : column_schema.children) { + DORIS_CHECK(child != nullptr); + RETURN_IF_ERROR(validate_all_projected_leaves_supported(*child)); + } + return Status::OK(); +} + +Status validate_projected_leaves_supported(const ParquetColumnSchema& column_schema, + const format::LocalColumnIndex& projection) { + if (column_schema.kind == ParquetColumnSchemaKind::PRIMITIVE || + projection.project_all_children || projection.children.empty()) { + return validate_all_projected_leaves_supported(column_schema); + } + 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()); + RETURN_IF_ERROR(validate_projected_leaves_supported(**child_it, child_projection)); + } + return Status::OK(); +} + +Status validate_requested_columns_supported( + const std::vector>& file_schema, + const format::FileScanRequest& request) { + auto validate_scan_column = [&](const format::LocalColumnIndex& projection) -> Status { + const auto local_id = projection.local_id(); + if (local_id == format::ROW_POSITION_COLUMN_ID || + local_id == format::GLOBAL_ROWID_COLUMN_ID) { + return Status::OK(); + } + DORIS_CHECK(local_id >= 0 && local_id < static_cast(file_schema.size())); + DORIS_CHECK(file_schema[local_id] != nullptr); + return validate_projected_leaves_supported(*file_schema[local_id], projection); + }; + for (const auto& column : request.predicate_columns) { + RETURN_IF_ERROR(validate_scan_column(column)); + } + for (const auto& column : request.non_predicate_columns) { + RETURN_IF_ERROR(validate_scan_column(column)); + } + return Status::OK(); +} + std::vector build_page_cache_ranges( const ::parquet::FileMetaData& metadata, const std::vector>& file_schema, @@ -454,6 +508,12 @@ Status ParquetReader::open(std::shared_ptr request) { DORIS_CHECK(local_id >= 0 && local_id < num_fields); } + // Reject requested unsupported logical leaves before row-group statistics, dictionaries, + // bloom filters or page indexes inspect their physical fallback type. For example, a predicate + // on TIME_MILLIS must fail here even when its INT32 statistics would prune every row group; + // otherwise the same unsupported SELECT could fail or silently succeed depending on data. + RETURN_IF_ERROR(validate_requested_columns_supported(_state->file_schema, *request_snapshot)); + RowGroupScanPlan row_group_plan; ParquetScanRange scan_range; scan_range.start_offset = _file_description->range_start_offset; @@ -598,6 +658,19 @@ Status ParquetReader::get_aggregate_result(const format::FileAggregateRequest& r request.agg_type); } + for (const auto& aggregate_column : request.columns) { + const auto local_id = aggregate_column.projection.local_id(); + if (local_id < 0 || local_id >= static_cast(_state->file_schema.size())) { + return Status::InvalidArgument("Invalid parquet aggregate column id {}", local_id); + } + DORIS_CHECK(_state->file_schema[local_id] != nullptr); + // Aggregate pushdown can return directly from footer statistics without constructing a + // column reader. Validate first so MIN/MAX(TIME_MILLIS), or an all-pruned COUNT request, + // cannot expose the physical INT32 fallback as a supported logical value. + RETURN_IF_ERROR(validate_projected_leaves_supported(*_state->file_schema[local_id], + aggregate_column.projection)); + } + // Aggregate row count in all selected row groups. For MIN/MAX aggregate, this is used to determine whether there is no row group selected. for (const auto& row_group_plan : _state->scan_plan.row_groups) { auto row_group_metadata = diff --git a/be/test/exec/scan/file_scanner_v2_test.cpp b/be/test/exec/scan/file_scanner_v2_test.cpp index 9e31d2f6e7d029..86ac3a10f2326e 100644 --- a/be/test/exec/scan/file_scanner_v2_test.cpp +++ b/be/test/exec/scan/file_scanner_v2_test.cpp @@ -495,9 +495,13 @@ TEST(FileScannerV2Test, NotFoundIsSkippedOnlyWhenConfigured) { } TEST(FileScannerV2Test, EndOfFileIsSkippedAsEmptySplit) { - EXPECT_TRUE(FileScannerV2::TEST_should_skip_empty(Status::EndOfFile("empty file"))); - EXPECT_FALSE(FileScannerV2::TEST_should_skip_empty(Status::InternalError("read failed"))); - EXPECT_FALSE(FileScannerV2::TEST_should_skip_empty(Status::OK())); + EXPECT_TRUE(FileScannerV2::TEST_should_skip_empty(Status::EndOfFile("empty file"), false)); + // Deletion-vector and Parquet readers also use EOF to unwind an interrupted read. Once either + // scanner stop flag is visible, the same status is no longer evidence of an empty file. + EXPECT_FALSE(FileScannerV2::TEST_should_skip_empty(Status::EndOfFile("stop read."), true)); + EXPECT_FALSE( + FileScannerV2::TEST_should_skip_empty(Status::InternalError("read failed"), false)); + EXPECT_FALSE(FileScannerV2::TEST_should_skip_empty(Status::OK(), false)); } // Scenario: partition slots are identified from the explicit FE category when present, otherwise 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 f06c3fd4660c09..940f94a373a013 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 @@ -115,27 +115,43 @@ TEST(ParquetPageCacheRangeTest, InvalidRequestMisses) { EXPECT_TRUE(detail::plan_page_cache_range_read(100, -1, cached_ranges).empty()); } -TEST(ParquetPageCacheRangeTest, PerFileRangeIndexSortsDeduplicatesAndClears) { - detail::ParquetPageCacheRangeIndex index; +TEST(ParquetPageCacheRangeTest, PerFileRangeIndexDeduplicatesAndEvictsAtCapacity) { + detail::ParquetPageCacheRangeIndex index(3); index.insert({200, 20}); index.insert({100, 30}); index.insert({100, 10}); index.insert({100, 30}); - ASSERT_EQ(index.ranges().size(), 3); - EXPECT_EQ(index.ranges()[0].offset, 100); - EXPECT_EQ(index.ranges()[0].size, 10); - EXPECT_EQ(index.ranges()[1].offset, 100); - EXPECT_EQ(index.ranges()[1].size, 30); - EXPECT_EQ(index.ranges()[2].offset, 200); + EXPECT_EQ(index.size(), 3); + index.insert({300, 40}); + const auto ranges = index.ranges(); + ASSERT_EQ(ranges.size(), 3); + EXPECT_EQ(ranges[0].offset, 100); + EXPECT_EQ(ranges[0].size, 30); + EXPECT_EQ(ranges[1].offset, 200); + EXPECT_EQ(ranges[1].size, 20); + EXPECT_EQ(ranges[2].offset, 300); index.erase({100, 30}); - ASSERT_EQ(index.ranges().size(), 2); - EXPECT_EQ(index.ranges()[0].size, 10); - EXPECT_EQ(index.ranges()[1].offset, 200); + EXPECT_EQ(index.size(), 2); +} + +TEST(ParquetPageCacheRangeTest, DirectorySharesBoundedIndexAcrossReaderLifetimes) { + detail::ParquetPageCacheRangeDirectory directory(2); + auto first_reader_index = directory.get_or_create("file-a"); + first_reader_index->insert({100, 100}); + first_reader_index.reset(); + + // The directory owns the per-file index, so reader B still discovers reader A's wider cache + // entry and can plan a subset hit after A closes. + auto second_reader_index = directory.get_or_create("file-a"); + const auto plan = detail::plan_page_cache_range_read(120, 30, second_reader_index->ranges()); + ASSERT_EQ(plan.size(), 1); + expect_plan_entry(plan[0], {100, 100}, 20, 0, 30); - index.clear(); - EXPECT_TRUE(index.ranges().empty()); + directory.get_or_create("file-b"); + directory.get_or_create("file-c"); + EXPECT_EQ(directory.size(), 2); } TEST(ParquetPageCacheRangeTest, ValidPrefetchRangesSkipInvalidAndOverflowRanges) { diff --git a/regression-test/data/external_table_p0/tvf/file_scanner_v2_unsupported_time.parquet b/regression-test/data/external_table_p0/tvf/file_scanner_v2_unsupported_time.parquet index fe1daf97c37e4e6cfa72a93adb57f9e3ac952a20..5b973181a56148d379ff1ee20612491d1121b700 100644 GIT binary patch delta 29 lcmdnUx{!6k21db&8+|6TG6rmpW%Oa>kY-|F2yhHC1OSTr2fY9Q delta 44 ucmZ3;x{-Cl21cQY8-3VqG}$E87+5C$34}2>`!M=4a%eL#Fa$UT83F(c910x( diff --git a/regression-test/suites/external_table_p0/tvf/test_file_scanner_v2_review_fixes.groovy b/regression-test/suites/external_table_p0/tvf/test_file_scanner_v2_review_fixes.groovy index ac8359d980a023..fe39450d71d0e7 100644 --- a/regression-test/suites/external_table_p0/tvf/test_file_scanner_v2_review_fixes.groovy +++ b/regression-test/suites/external_table_p0/tvf/test_file_scanner_v2_review_fixes.groovy @@ -81,4 +81,20 @@ suite("test_file_scanner_v2_review_fixes", "p0,external") { ORDER BY id """ + // A requested unsupported leaf must fail before metadata pruning. The fixture stores positive + // TIME_MILLIS values, so `unsupported_time < 0` can be eliminated from INT32 row-group + // statistics. Without the early validation, this query incorrectly returns an empty result + // instead of preserving the unsupported logical-type contract. + test { + sql """ + SELECT unsupported_time + FROM local( + "file_path" = "${remotePath}/file_scanner_v2_unsupported_time.parquet", + "backend_id" = "${backendId}", + "format" = "parquet") + WHERE unsupported_time < 0 + """ + exception "Parquet TIME with isAdjustedToUTC=true is not supported" + } + } From cc9750751fcd638dd86b621a01fc8855fb5cec69 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Tue, 14 Jul 2026 12:03:23 +0800 Subject: [PATCH 3/5] [fix](be) Address remaining FileScannerV2 review cases ### What problem does this PR solve? Issue Number: None Related PR: None Problem Summary: Primitive COUNT(column) pushdown omitted the counted projection, allowing unsupported Parquet logical types to bypass aggregate validation and return footer row counts. Explicit zero-length Native blocks also surfaced as EOF and could be accounted as valid empty files. Preserve scalar COUNT projections for file-reader validation, reuse footer counts only after validating required primitive columns, and reject explicit zero-length Native blocks as malformed input. ### Release note Reject unsupported Parquet scalar COUNT projections and malformed zero-length Native blocks consistently. ### Check List (For Author) - Test: Unit Test - 16 targeted Backend unit tests - Behavior changed: Yes (primitive COUNT projections are validated and explicit zero-length Native blocks return an error) - Does this need documentation: No --- be/src/format_v2/native/native_reader.cpp | 8 ++- be/src/format_v2/parquet/parquet_reader.cpp | 10 ++++ be/src/format_v2/table_reader.h | 13 +++-- .../format_v2/native/native_reader_test.cpp | 4 +- .../format_v2/parquet/parquet_scan_test.cpp | 53 +++++++++++++++++++ be/test/format_v2/table_reader_test.cpp | 7 +++ 6 files changed, 85 insertions(+), 10 deletions(-) diff --git a/be/src/format_v2/native/native_reader.cpp b/be/src/format_v2/native/native_reader.cpp index 0e348f3de2ccf5..867ea5c6a54c98 100644 --- a/be/src/format_v2/native/native_reader.cpp +++ b/be/src/format_v2/native/native_reader.cpp @@ -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'); diff --git a/be/src/format_v2/parquet/parquet_reader.cpp b/be/src/format_v2/parquet/parquet_reader.cpp index 37607b3df2b864..1c5d00e7d76e1c 100644 --- a/be/src/format_v2/parquet/parquet_reader.cpp +++ b/be/src/format_v2/parquet/parquet_reader.cpp @@ -687,6 +687,16 @@ Status ParquetReader::get_aggregate_result(const format::FileAggregateRequest& r } const auto& count_projection = request.columns[0].projection; const auto& root_schema = projected_root_schema(_state->file_schema, count_projection); + // A required primitive COUNT(col) still carries its projection so the unsupported-type + // validation above cannot be bypassed. Once validated, its definition level proves that + // every selected row is non-NULL, so preserve the already-computed footer row count and + // avoid reading definition levels merely to rediscover COUNT(col) == COUNT(*). Complex + // roots continue through the shape reader because their count semantics and read-row + // accounting are derived from nested levels. + if (root_schema.kind == ParquetColumnSchemaKind::PRIMITIVE && + root_schema.max_definition_level == 0) { + return Status::OK(); + } result->count = 0; for (const auto& row_group_plan : _state->scan_plan.row_groups) { std::shared_ptr<::parquet::RowGroupReader> row_group; diff --git a/be/src/format_v2/table_reader.h b/be/src/format_v2/table_reader.h index b8457ee9b4a31e..a5f7194e11db1f 100644 --- a/be/src/format_v2/table_reader.h +++ b/be/src/format_v2/table_reader.h @@ -1433,16 +1433,15 @@ class TableReader { request->columns.clear(); if (agg_type == TPushAggOp::type::COUNT) { // COUNT pushdown historically meant COUNT(*) and therefore carried no columns. For - // complex COUNT(col), materializing the full MAP/LIST/STRUCT value only to test the - // top-level NULL bit can be extremely expensive. When the scan projects exactly one - // directly-mapped complex column, pass that file column to the reader so formats such - // as Parquet can count the column shape from metadata/levels without decoding payload - // values like MAP value strings. Other COUNT cases stay on the existing row-count path - // to avoid changing count(*) semantics. + // COUNT(col), Nereids retains exactly the counted slot in the file scan. Pass that + // single direct mapping for both scalar and complex columns: besides allowing nullable + // complex values to be counted from levels, it lets the file reader validate logical + // types before returning footer row counts. For example, a required TIME_MILLIS leaf + // has the same count as COUNT(*), but it must still fail as an unsupported requested + // column rather than silently returning the row-group count. if (_data_reader.column_mapper->mappings().size() == 1) { const auto& mapping = _data_reader.column_mapper->mappings()[0]; if (mapping.file_local_id.has_value() && mapping.file_type != nullptr && - is_complex_type(remove_nullable(mapping.file_type)->get_primitive_type()) && mapping.virtual_column_type == TableVirtualColumnType::INVALID && mapping.default_expr == nullptr) { FileAggregateRequest::Column column; diff --git a/be/test/format_v2/native/native_reader_test.cpp b/be/test/format_v2/native/native_reader_test.cpp index 6b0c9ab27b90a8..422b2d5c37b2b2 100644 --- a/be/test/format_v2/native/native_reader_test.cpp +++ b/be/test/format_v2/native/native_reader_test.cpp @@ -375,7 +375,9 @@ TEST(NativeV2ReaderTest, RejectsZeroLengthBlockAndInvalidPBlock) { auto zero_len_reader = create_reader(zero_len_path, &state, &profile); ASSERT_TRUE(zero_len_reader->init(&state).ok()); std::vector schema; - EXPECT_FALSE(zero_len_reader->get_schema(&schema).ok()); + const auto zero_len_status = zero_len_reader->get_schema(&schema); + EXPECT_TRUE(zero_len_status.is()) << zero_len_status; + EXPECT_NE(zero_len_status.to_string().find("zero-length native block"), std::string::npos); static_cast(io::global_local_filesystem()->delete_file(zero_len_path)); const auto invalid_pblock_path = diff --git a/be/test/format_v2/parquet/parquet_scan_test.cpp b/be/test/format_v2/parquet/parquet_scan_test.cpp index 04d46097c98648..2051f4d20bdde0 100644 --- a/be/test/format_v2/parquet/parquet_scan_test.cpp +++ b/be/test/format_v2/parquet/parquet_scan_test.cpp @@ -21,6 +21,7 @@ #include #include #include +#include #include #include @@ -303,6 +304,32 @@ void write_table(const std::string& file_path, const std::shared_ptr out = *file_result; + + // Arrow intentionally writes time32 as local time (isAdjustedToUTC=false), so use Parquet's + // low-level writer to produce the unsupported required logical type needed by this aggregate + // regression. Required is important: Nereids may push COUNT(col) because NULL filtering cannot + // change its value, which is precisely the path where an empty aggregate request lost `col`. + const auto adjusted_time = ::parquet::schema::PrimitiveNode::Make( + "unsupported_time", ::parquet::Repetition::REQUIRED, + ::parquet::LogicalType::Time(true, ::parquet::LogicalType::TimeUnit::MILLIS), + ::parquet::Type::INT32); + const auto schema_node = ::parquet::schema::GroupNode::Make( + "schema", ::parquet::Repetition::REQUIRED, {adjusted_time}); + const auto schema = std::static_pointer_cast<::parquet::schema::GroupNode>(schema_node); + auto writer = ::parquet::ParquetFileWriter::Open(out, schema); + auto* row_group = writer->AppendRowGroup(); + auto* time_writer = static_cast<::parquet::Int32Writer*>(row_group->NextColumn()); + const int32_t values[] = {1000, 2000}; + EXPECT_EQ(time_writer->WriteBatch(2, nullptr, nullptr, values), 2); + time_writer->Close(); + row_group->Close(); + writer->Close(); +} + void write_int_pair_parquet_file(const std::string& file_path, int64_t row_group_size = 2, bool enable_statistics = true) { auto schema = arrow::schema({ @@ -727,6 +754,15 @@ TEST_F(ParquetScanTest, AggregateCountAndMinMaxUseAllSelectedRowGroups) { EXPECT_EQ(count_result.count, 6); EXPECT_TRUE(count_result.columns.empty()); + format::FileAggregateResult required_count_result; + format::FileAggregateRequest required_count_request; + required_count_request.agg_type = TPushAggOp::COUNT; + required_count_request.columns.push_back({.projection = field_projection(0)}); + ASSERT_TRUE(reader->get_aggregate_result(required_count_request, &required_count_result).ok()); + // The required scalar projection is retained for logical-type validation, but after that + // validation COUNT(id) can reuse the same selected-row-group footer count as COUNT(*). + EXPECT_EQ(required_count_result.count, 6); + format::FileAggregateResult minmax_result; format::FileAggregateRequest minmax_request; minmax_request.agg_type = TPushAggOp::MINMAX; @@ -743,6 +779,23 @@ TEST_F(ParquetScanTest, AggregateCountAndMinMaxUseAllSelectedRowGroups) { EXPECT_EQ(minmax_result.columns[1].max_value.get(), 60); } +TEST_F(ParquetScanTest, AggregateCountRejectsRequiredUnsupportedScalarProjection) { + write_required_adjusted_time_parquet_file(_file_path); + auto reader = create_reader(); + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + ASSERT_TRUE(reader->init(&state).ok()); + open_all_row_groups(reader.get()); + + format::FileAggregateRequest request; + request.agg_type = TPushAggOp::COUNT; + request.columns.push_back({.projection = field_projection(0)}); + format::FileAggregateResult result; + const auto status = reader->get_aggregate_result(request, &result); + EXPECT_TRUE(status.is()) << status; + EXPECT_NE(status.to_string().find("Parquet TIME with isAdjustedToUTC=true is not supported"), + std::string::npos); +} + TEST_F(ParquetScanTest, AggregateMinMaxRejectsInexactBinaryStatistics) { write_binary_minmax_parquet_file(_file_path); auto reader = create_reader(); diff --git a/be/test/format_v2/table_reader_test.cpp b/be/test/format_v2/table_reader_test.cpp index 0d484f60d9e5de..08338bce977bc7 100644 --- a/be/test/format_v2/table_reader_test.cpp +++ b/be/test/format_v2/table_reader_test.cpp @@ -1012,6 +1012,7 @@ struct FakeFileReaderState { bool stop_during_read = false; bool not_found_during_init = false; std::shared_ptr last_request; + std::optional last_aggregate_request; std::shared_ptr condition_cache_ctx; std::shared_ptr io_ctx; }; @@ -1127,6 +1128,7 @@ class FakeFileReader final : public FileReader { if (request.agg_type != TPushAggOp::type::COUNT) { return Status::NotSupported("fake reader only supports COUNT aggregate pushdown"); } + _state->last_aggregate_request = request; if (_state->stop_during_aggregate) { DORIS_CHECK(_state->io_ctx != nullptr); _state->io_ctx->should_stop = true; @@ -1753,6 +1755,11 @@ TEST(TableReaderTest, PushDownCountRecordsReaderRowsBeforeClosingReader) { EXPECT_EQ(block.rows(), 3); EXPECT_EQ(file_reader_stats.read_rows, 3); EXPECT_EQ(fake_state->close_count, 1); + ASSERT_TRUE(fake_state->last_aggregate_request.has_value()); + ASSERT_EQ(fake_state->last_aggregate_request->columns.size(), 1); + // A primitive COUNT(col) projection must reach the file reader just like a complex one. This + // is observable even though the required `id` values make its numeric result equal COUNT(*). + EXPECT_EQ(fake_state->last_aggregate_request->columns[0].projection.local_id(), 0); } TEST(TableReaderTest, PushDownCountStopConvertsAggregateEndOfFileToEos) { From 4f11435b6aecfc0ed600f2341f435846e3deafa0 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Tue, 14 Jul 2026 16:58:10 +0800 Subject: [PATCH 4/5] [fix](be) Preserve COUNT star semantics in file scans ### What problem does this PR solve? Issue Number: None Related PR: #65548 Problem Summary: File aggregate pushdown inferred COUNT(column) from a single post-pruning scan mapping. COUNT(*) also retains an arbitrary placeholder slot, so a nullable placeholder made Parquet and ORC readers count non-null values instead of rows. Carry the semantic COUNT arguments from Nereids through the scan thrift contract, keep true COUNT(*) column-free, and fall back when an explicit COUNT argument cannot be mapped directly. Also default split-level conjunct overrides to nullopt so existing standalone readers compile and preserve their initial predicate snapshot. ### Release note Fix COUNT(*) results for Parquet and ORC file scans when the retained placeholder column contains NULL values. ### Check List (For Author) - Test: Unit Test - 7 targeted Backend ASAN unit tests - Added Frontend rule coverage for COUNT(*) and COUNT(column); execution was blocked by unrelated untracked cloud storage sources missing vendor SDKs in the validation workspace - Behavior changed: Yes (file aggregate pushdown now distinguishes COUNT(*) from COUNT(column) explicitly) - Does this need documentation: No --- be/src/exec/operator/scan_operator.cpp | 8 +++ be/src/exec/operator/scan_operator.h | 6 ++ be/src/exec/scan/file_scanner_v2.cpp | 12 ++++ be/src/format_v2/table_reader.cpp | 1 + be/src/format_v2/table_reader.h | 58 +++++++++++++------ be/test/format_v2/table_reader_test.cpp | 45 ++++++++++++++ .../apache/doris/datasource/FileScanNode.java | 2 + .../translator/PhysicalPlanTranslator.java | 9 +++ .../translator/PlanTranslatorContext.java | 10 ++++ .../implementation/AggregateStrategies.java | 17 ++++-- .../PhysicalStorageLayerAggregate.java | 23 ++++++-- .../org/apache/doris/planner/PlanNode.java | 7 +++ .../PhysicalStorageLayerAggregateTest.java | 25 +++++++- gensrc/thrift/PlanNodes.thrift | 4 ++ 14 files changed, 198 insertions(+), 29 deletions(-) diff --git a/be/src/exec/operator/scan_operator.cpp b/be/src/exec/operator/scan_operator.cpp index a9e92d08e08613..b142c621a06a04 100644 --- a/be/src/exec/operator/scan_operator.cpp +++ b/be/src/exec/operator/scan_operator.cpp @@ -1046,6 +1046,11 @@ TPushAggOp::type ScanLocalState::get_push_down_agg_type() { return _parent->cast()._push_down_agg_type; } +template +const std::vector& ScanLocalState::get_push_down_count_slot_ids() const { + return _parent->cast()._push_down_count_slot_ids; +} + template int64_t ScanLocalState::limit_per_scanner() { return _parent->cast()._limit_per_scanner; @@ -1231,6 +1236,9 @@ Status ScanOperatorX::init(const TPlanNode& tnode, RuntimeState* } else { _push_down_agg_type = TPushAggOp::type::NONE; } + if (tnode.__isset.push_down_count_slot_ids) { + _push_down_count_slot_ids = tnode.push_down_count_slot_ids; + } if (tnode.__isset.topn_filter_source_node_ids) { _topn_filter_source_node_ids = tnode.topn_filter_source_node_ids; diff --git a/be/src/exec/operator/scan_operator.h b/be/src/exec/operator/scan_operator.h index 8b12ccf0bc1195..51bc4048606eea 100644 --- a/be/src/exec/operator/scan_operator.h +++ b/be/src/exec/operator/scan_operator.h @@ -74,6 +74,7 @@ class ScanLocalStateBase : public PipelineXLocalState<> { virtual void set_scan_ranges(RuntimeState* state, const std::vector& scan_ranges) = 0; virtual TPushAggOp::type get_push_down_agg_type() = 0; + virtual const std::vector& get_push_down_count_slot_ids() const = 0; // If scan operator is serial operator(like topn), its real parallelism is 1. // Otherwise, its real parallelism is query_parallel_instance_num. @@ -252,6 +253,7 @@ class ScanLocalState : public ScanLocalStateBase { const std::vector& scan_ranges) override {} TPushAggOp::type get_push_down_agg_type() override; + const std::vector& get_push_down_count_slot_ids() const override; std::vector execution_dependencies() override { if (_filter_dependencies.empty()) { @@ -452,6 +454,10 @@ class ScanOperatorX : public OperatorX { TPushAggOp::type _push_down_agg_type; + // Semantic arguments of a pushed-down COUNT. An empty list means COUNT(*)/COUNT(1), even + // when column pruning keeps an arbitrary physical slot as a scan placeholder. + std::vector _push_down_count_slot_ids; + // Record the value of the aggregate function 'count' from doris's be int64_t _push_down_count = -1; const int _parallel_tasks = 0; diff --git a/be/src/exec/scan/file_scanner_v2.cpp b/be/src/exec/scan/file_scanner_v2.cpp index f4213c43edebb9..1e30f4d742278c 100644 --- a/be/src/exec/scan/file_scanner_v2.cpp +++ b/be/src/exec/scan/file_scanner_v2.cpp @@ -478,6 +478,17 @@ Status FileScannerV2::_init_table_reader(const TFileRangeDesc& range) { VExprContextSPtrs table_conjuncts; RETURN_IF_ERROR(_build_table_conjuncts(&table_conjuncts)); + std::vector push_down_count_columns; + push_down_count_columns.reserve(_local_state->get_push_down_count_slot_ids().size()); + for (const auto slot_id : _local_state->get_push_down_count_slot_ids()) { + const auto global_index_it = _slot_id_to_global_index.find(slot_id); + if (global_index_it == _slot_id_to_global_index.end()) { + return Status::InternalError( + "Pushed-down COUNT argument is not a projected file scan slot, slot_id={}", + slot_id); + } + push_down_count_columns.push_back(global_index_it->second); + } RETURN_IF_ERROR(_table_reader->init({ .projected_columns = _projected_columns, .conjuncts = std::move(table_conjuncts), @@ -488,6 +499,7 @@ Status FileScannerV2::_init_table_reader(const TFileRangeDesc& range) { .scanner_profile = _local_state->scanner_profile(), .file_slot_descs = &_file_slot_descs, .push_down_agg_type = _local_state->get_push_down_agg_type(), + .push_down_count_columns = std::move(push_down_count_columns), .condition_cache_digest = _local_state->get_condition_cache_digest(), })); return Status::OK(); diff --git a/be/src/format_v2/table_reader.cpp b/be/src/format_v2/table_reader.cpp index 09c5d65e82b7fd..b3c5b7a95bb1d5 100644 --- a/be/src/format_v2/table_reader.cpp +++ b/be/src/format_v2/table_reader.cpp @@ -472,6 +472,7 @@ Status TableReader::init(TableReadOptions&& options) { _scanner_profile = options.scanner_profile; _file_slot_descs = options.file_slot_descs; _push_down_agg_type = options.push_down_agg_type; + _push_down_count_columns = options.push_down_count_columns; _condition_cache_digest = options.condition_cache_digest; _projected_columns = std::move(options.projected_columns); _system_properties = create_system_properties(_scan_params); diff --git a/be/src/format_v2/table_reader.h b/be/src/format_v2/table_reader.h index a5f7194e11db1f..a63c8f6b24cc90 100644 --- a/be/src/format_v2/table_reader.h +++ b/be/src/format_v2/table_reader.h @@ -135,6 +135,9 @@ struct TableReadOptions { const std::vector* file_slot_descs = nullptr; // Push-down aggregate type. const TPushAggOp::type push_down_agg_type = TPushAggOp::type::NONE; + // Table/global indices of explicit COUNT arguments. Empty means COUNT(*)/COUNT(1), not "infer + // the argument from projected_columns": COUNT(*) still projects one arbitrary placeholder. + const std::vector push_down_count_columns = {}; // Digest of stable pushed-down predicates. A zero digest disables condition cache. uint64_t condition_cache_digest = 0; }; @@ -145,7 +148,7 @@ struct SplitReadOptions { // Latest scanner conjuncts rewritten to table/global column indices. Runtime filters may // arrive after TableReader::init(), so scanner-driven splits replace the initial snapshot. // nullopt preserves the initial snapshot for standalone TableReader callers. - std::optional conjuncts; + std::optional conjuncts = std::nullopt; // Independent clones used for partition pruning because evaluation prepares and opens them // against a synthetic partition block before the file reader opens its row-level conjuncts. VExprContextSPtrs partition_prune_conjuncts; @@ -949,7 +952,19 @@ class TableReader { return false; } if (agg_type == TPushAggOp::type::COUNT) { - return true; + // COUNT(*) needs no column metadata. COUNT(col) currently supports one direct file + // column; multiple COUNT arguments fall back to the normal scan so every upper + // aggregate receives the original rows. + if (_push_down_count_columns.empty()) { + return true; + } + if (_push_down_count_columns.size() != 1) { + return false; + } + const auto& mapping = _push_down_count_mapping(); + return mapping.file_local_id.has_value() && mapping.file_type != nullptr && + mapping.virtual_column_type == TableVirtualColumnType::INVALID && + mapping.default_expr == nullptr; } // For MIN/MAX, only support direct file-to-table column mappings. The two emitted rows // must be enough for the upper MIN/MAX aggregate without evaluating default expressions or @@ -1432,23 +1447,16 @@ class TableReader { request->agg_type = agg_type; request->columns.clear(); if (agg_type == TPushAggOp::type::COUNT) { - // COUNT pushdown historically meant COUNT(*) and therefore carried no columns. For - // COUNT(col), Nereids retains exactly the counted slot in the file scan. Pass that - // single direct mapping for both scalar and complex columns: besides allowing nullable - // complex values to be counted from levels, it lets the file reader validate logical - // types before returning footer row counts. For example, a required TIME_MILLIS leaf - // has the same count as COUNT(*), but it must still fail as an unsupported requested - // column rather than silently returning the row-group count. - if (_data_reader.column_mapper->mappings().size() == 1) { - const auto& mapping = _data_reader.column_mapper->mappings()[0]; - if (mapping.file_local_id.has_value() && mapping.file_type != nullptr && - mapping.virtual_column_type == TableVirtualColumnType::INVALID && - mapping.default_expr == nullptr) { - FileAggregateRequest::Column column; - column.projection = - LocalColumnIndex::top_level(LocalColumnId(*mapping.file_local_id)); - request->columns.push_back(std::move(column)); - } + // An empty explicit list is the semantic signal for COUNT(*). Do not inspect the + // mapping count: `SELECT COUNT(*) FROM t` may still project one nullable column because + // the planner keeps a placeholder slot, and counting that slot would return COUNT(col). + if (!_push_down_count_columns.empty()) { + const auto& mapping = _push_down_count_mapping(); + DORIS_CHECK(mapping.file_local_id.has_value()); + FileAggregateRequest::Column column; + column.projection = + LocalColumnIndex::top_level(LocalColumnId(*mapping.file_local_id)); + request->columns.push_back(std::move(column)); } return Status::OK(); } @@ -1465,6 +1473,17 @@ class TableReader { return Status::OK(); } + const ColumnMapping& _push_down_count_mapping() const { + DORIS_CHECK(_push_down_count_columns.size() == 1); + const auto mapping_it = + std::ranges::find(_data_reader.column_mapper->mappings(), + _push_down_count_columns.front(), &ColumnMapping::global_index); + // FileScannerV2 translates FE SlotIds through the same projected-column list used to build + // the mapper, so a missing mapping is an FE/BE contract violation rather than a fallback. + DORIS_CHECK(mapping_it != _data_reader.column_mapper->mappings().end()); + return *mapping_it; + } + Status _materialize_aggregate_pushdown_rows(TPushAggOp::type agg_type, const FileAggregateResult& file_result, Block* block) { @@ -1569,6 +1588,7 @@ class TableReader { const std::vector* _file_slot_descs = nullptr; FileFormat _format; TPushAggOp::type _push_down_agg_type = TPushAggOp::type::NONE; + std::vector _push_down_count_columns; size_t _batch_size = 0; uint64_t _condition_cache_digest = 0; segment_v2::ConditionCache::ExternalCacheKey _condition_cache_key; diff --git a/be/test/format_v2/table_reader_test.cpp b/be/test/format_v2/table_reader_test.cpp index 08338bce977bc7..2c45f2f9f0e6eb 100644 --- a/be/test/format_v2/table_reader_test.cpp +++ b/be/test/format_v2/table_reader_test.cpp @@ -1741,6 +1741,7 @@ TEST(TableReaderTest, PushDownCountRecordsReaderRowsBeforeClosingReader) { .runtime_state = &state, .scanner_profile = nullptr, .push_down_agg_type = TPushAggOp::type::COUNT, + .push_down_count_columns = {GlobalIndex(0)}, }) .ok()); @@ -1762,6 +1763,50 @@ TEST(TableReaderTest, PushDownCountRecordsReaderRowsBeforeClosingReader) { EXPECT_EQ(fake_state->last_aggregate_request->columns[0].projection.local_id(), 0); } +TEST(TableReaderTest, PushDownCountStarIgnoresProjectedPlaceholderColumn) { + const auto nullable_int_type = make_nullable(std::make_shared()); + std::vector file_schema; + file_schema.push_back(make_file_column(0, "nullable_id", nullable_int_type)); + + std::vector projected_columns; + projected_columns.push_back(make_table_column(0, "nullable_id", nullable_int_type)); + set_name_identifiers(&projected_columns); + + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + auto fake_state = std::make_shared(); + fake_state->aggregate_count = 3; + FakeTableReader reader(file_schema, fake_state); + ASSERT_TRUE( + reader.init({ + .projected_columns = projected_columns, + .conjuncts = {}, + .format = FileFormat::PARQUET, + .scan_params = nullptr, + .io_ctx = nullptr, + .runtime_state = &state, + .scanner_profile = nullptr, + .push_down_agg_type = TPushAggOp::type::COUNT, + // COUNT(*) deliberately has no explicit count columns. The + // nullable_id projection is only the planner's scan placeholder. + .push_down_count_columns = {}, + }) + .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); + ASSERT_TRUE(fake_state->last_aggregate_request.has_value()); + // Passing nullable_id here would implement COUNT(nullable_id) and reproduce the external ORC + // and Parquet failures where footer row counts were reduced by null values. + EXPECT_TRUE(fake_state->last_aggregate_request->columns.empty()); +} + TEST(TableReaderTest, PushDownCountStopConvertsAggregateEndOfFileToEos) { std::vector file_schema; file_schema.push_back(make_file_column(0, "id", std::make_shared())); diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/FileScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/FileScanNode.java index 466dad11d5dc49..bfcbb24489176e 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/FileScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/FileScanNode.java @@ -88,6 +88,8 @@ public FileScanNode(PlanNodeId id, TupleDescriptor desc, String planNodeName, @Override protected void toThrift(TPlanNode planNode) { planNode.setPushDownAggTypeOpt(pushDownAggNoGroupingOp); + planNode.setPushDownCountSlotIds( + pushDownCountSlotIds.stream().map(id -> id.asInt()).collect(Collectors.toList())); planNode.setNodeType(TPlanNodeType.FILE_SCAN_NODE); TFileScanNode fileScanNode = new TFileScanNode(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java index 01d9139577578b..bc7d2936eaa1cd 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java @@ -850,6 +850,11 @@ private PlanFragment getPlanFragmentForPhysicalFileScan(PhysicalFileScan fileSca scanNode.setNereidsId(fileScan.getId()); context.getNereidsIdToPlanNodeIdMap().put(fileScan.getId(), scanNode.getId()); scanNode.setPushDownAggNoGrouping(context.getRelationPushAggOp(fileScan.getRelationId())); + scanNode.setPushDownCountSlotIds(context.getRelationPushCountArgumentExprIds(fileScan.getRelationId()) + .stream() + .map(exprId -> Objects.requireNonNull(context.findSlotRef(exprId), + "missing slot for pushed-down COUNT argument " + exprId).getSlotId()) + .collect(Collectors.toList())); scanNode.setHasPartitionPredicate(fileScan.hasPartitionPredicate()); if (fileScan.getStats() != null) { @@ -1401,6 +1406,10 @@ public PlanFragment visitPhysicalStorageLayerAggregate( context.setRelationPushAggOp( storageLayerAggregate.getRelation().getRelationId(), pushAggOp); + context.setRelationPushCountArgumentExprIds( + storageLayerAggregate.getRelation().getRelationId(), + pushAggOp == TPushAggOp.COUNT + ? storageLayerAggregate.getCountArgumentExprIds() : ImmutableList.of()); PlanFragment planFragment = storageLayerAggregate.getRelation().accept(this, context); diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PlanTranslatorContext.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PlanTranslatorContext.java index f680936b5e5ef4..936f4cbe1a9235 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PlanTranslatorContext.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PlanTranslatorContext.java @@ -52,6 +52,7 @@ import com.google.common.collect.Maps; import com.google.common.collect.Sets; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Optional; @@ -106,6 +107,7 @@ public class PlanTranslatorContext { private final Map cteScanNodeMap = Maps.newHashMap(); private final Map tablePushAggOp = Maps.newHashMap(); + private final Map> tablePushCountArgumentExprIds = Maps.newHashMap(); private final Map> statsUnknownColumnsMap = Maps.newHashMap(); @@ -430,6 +432,14 @@ public TPushAggOp getRelationPushAggOp(RelationId relationId) { return tablePushAggOp.getOrDefault(relationId, TPushAggOp.NONE); } + public void setRelationPushCountArgumentExprIds(RelationId relationId, List exprIds) { + tablePushCountArgumentExprIds.put(relationId, Lists.newArrayList(exprIds)); + } + + public List getRelationPushCountArgumentExprIds(RelationId relationId) { + return tablePushCountArgumentExprIds.getOrDefault(relationId, Collections.emptyList()); + } + public boolean isTopMaterializeNode() { return isTopMaterializeNode; } diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/AggregateStrategies.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/AggregateStrategies.java index 975cfaf973c408..8a3e32d60d63d8 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/AggregateStrategies.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/AggregateStrategies.java @@ -34,6 +34,7 @@ import org.apache.doris.nereids.trees.expressions.Alias; import org.apache.doris.nereids.trees.expressions.Cast; import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.ExprId; import org.apache.doris.nereids.trees.expressions.IsNull; import org.apache.doris.nereids.trees.expressions.Or; import org.apache.doris.nereids.trees.expressions.Slot; @@ -690,6 +691,12 @@ private LogicalAggregate storageLayerAggregate( List usedSlotInTable = (List) Project.findProject(aggUsedSlots, logicalScan.getOutput()); + // COUNT(*) has no aggregate arguments, even though later column pruning retains one + // arbitrary scan slot. Preserve the semantic arguments here so the BE never needs to infer + // COUNT(col) from the post-pruning scan shape. + List countArgumentExprIds = mergeOp == PushDownAggOp.COUNT + ? usedSlotInTable.stream().map(SlotReference::getExprId).collect(Collectors.toList()) + : ImmutableList.of(); for (SlotReference slot : usedSlotInTable) { Optional optionalColumn = slot.getOriginalColumn(); @@ -738,11 +745,12 @@ private LogicalAggregate storageLayerAggregate( if (project != null) { return aggregate.withChildren(ImmutableList.of( project.withChildren( - ImmutableList.of(new PhysicalStorageLayerAggregate(physicalScan, mergeOp))) + ImmutableList.of(new PhysicalStorageLayerAggregate( + physicalScan, mergeOp, countArgumentExprIds))) )); } else { return aggregate.withChildren(ImmutableList.of( - new PhysicalStorageLayerAggregate(physicalScan, mergeOp) + new PhysicalStorageLayerAggregate(physicalScan, mergeOp, countArgumentExprIds) )); } @@ -754,11 +762,12 @@ private LogicalAggregate storageLayerAggregate( if (project != null) { return aggregate.withChildren(ImmutableList.of( project.withChildren( - ImmutableList.of(new PhysicalStorageLayerAggregate(physicalScan, mergeOp))) + ImmutableList.of(new PhysicalStorageLayerAggregate( + physicalScan, mergeOp, countArgumentExprIds))) )); } else { return aggregate.withChildren(ImmutableList.of( - new PhysicalStorageLayerAggregate(physicalScan, mergeOp) + new PhysicalStorageLayerAggregate(physicalScan, mergeOp, countArgumentExprIds) )); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalStorageLayerAggregate.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalStorageLayerAggregate.java index 2db9c1afc27492..1d4255b6df98ef 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalStorageLayerAggregate.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalStorageLayerAggregate.java @@ -20,6 +20,7 @@ import org.apache.doris.nereids.memo.GroupExpression; import org.apache.doris.nereids.properties.LogicalProperties; import org.apache.doris.nereids.properties.PhysicalProperties; +import org.apache.doris.nereids.trees.expressions.ExprId; import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction; import org.apache.doris.nereids.trees.expressions.functions.agg.Count; import org.apache.doris.nereids.trees.expressions.functions.agg.Max; @@ -43,21 +44,30 @@ public class PhysicalStorageLayerAggregate extends PhysicalCatalogRelation { private final PhysicalCatalogRelation relation; private final PushDownAggOp aggOp; + private final List countArgumentExprIds; public PhysicalStorageLayerAggregate(PhysicalCatalogRelation relation, PushDownAggOp aggOp) { + this(relation, aggOp, ImmutableList.of()); + } + + public PhysicalStorageLayerAggregate(PhysicalCatalogRelation relation, PushDownAggOp aggOp, + List countArgumentExprIds) { super(relation.getRelationId(), relation.getType(), relation.getTable(), relation.getQualifier(), Optional.empty(), relation.getLogicalProperties(), ImmutableList.of()); this.relation = Objects.requireNonNull(relation, "relation cannot be null"); this.aggOp = Objects.requireNonNull(aggOp, "aggOp cannot be null"); + this.countArgumentExprIds = ImmutableList.copyOf(countArgumentExprIds); } public PhysicalStorageLayerAggregate(PhysicalCatalogRelation relation, PushDownAggOp aggOp, + List countArgumentExprIds, Optional groupExpression, LogicalProperties logicalProperties, PhysicalProperties physicalProperties, Statistics statistics) { super(relation.getRelationId(), relation.getType(), relation.getTable(), relation.getQualifier(), groupExpression, logicalProperties, physicalProperties, statistics, ImmutableList.of()); this.relation = Objects.requireNonNull(relation, "relation cannot be null"); this.aggOp = Objects.requireNonNull(aggOp, "aggOp cannot be null"); + this.countArgumentExprIds = ImmutableList.copyOf(countArgumentExprIds); } public PhysicalRelation getRelation() { @@ -68,6 +78,10 @@ public PushDownAggOp getAggOp() { return aggOp; } + public List getCountArgumentExprIds() { + return countArgumentExprIds; + } + @Override public R accept(PlanVisitor visitor, C context) { return visitor.visitPhysicalStorageLayerAggregate(this, context); @@ -83,20 +97,21 @@ public String toString() { } public PhysicalStorageLayerAggregate withPhysicalOlapScan(PhysicalOlapScan physicalOlapScan) { - return AbstractPlan.copyWithSameId(this, () -> new PhysicalStorageLayerAggregate(physicalOlapScan, aggOp)); + return AbstractPlan.copyWithSameId(this, () -> new PhysicalStorageLayerAggregate( + physicalOlapScan, aggOp, countArgumentExprIds)); } @Override public PhysicalStorageLayerAggregate withGroupExpression(Optional groupExpression) { return AbstractPlan.copyWithSameId(this, () -> new PhysicalStorageLayerAggregate(relation, aggOp, - groupExpression, getLogicalProperties(), physicalProperties, statistics)); + countArgumentExprIds, groupExpression, getLogicalProperties(), physicalProperties, statistics)); } @Override public Plan withGroupExprLogicalPropChildren(Optional groupExpression, Optional logicalProperties, List children) { return AbstractPlan.copyWithSameId(this, () -> new PhysicalStorageLayerAggregate(relation, aggOp, - groupExpression, logicalProperties.get(), physicalProperties, statistics)); + countArgumentExprIds, groupExpression, logicalProperties.get(), physicalProperties, statistics)); } @Override @@ -104,7 +119,7 @@ public PhysicalPlan withPhysicalPropertiesAndStats(PhysicalProperties physicalPr Statistics statistics) { return AbstractPlan.copyWithSameId(this, () -> new PhysicalStorageLayerAggregate( (PhysicalCatalogRelation) relation.withPhysicalPropertiesAndStats(null, statistics), - aggOp, groupExpression, + aggOp, countArgumentExprIds, groupExpression, getLogicalProperties(), physicalProperties, statistics)); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java index c1f97fdeb96e20..1fc55fe8f5e1c0 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java @@ -771,11 +771,18 @@ public void setCardinalityAfterFilter(long cardinalityAfterFilter) { } protected TPushAggOp pushDownAggNoGroupingOp = TPushAggOp.NONE; + // Explicit COUNT arguments. COUNT(*)/COUNT(1) intentionally keep this empty even though + // column pruning retains one placeholder scan slot. + protected List pushDownCountSlotIds = Collections.emptyList(); public void setPushDownAggNoGrouping(TPushAggOp pushDownAggNoGroupingOp) { this.pushDownAggNoGroupingOp = pushDownAggNoGroupingOp; } + public void setPushDownCountSlotIds(List pushDownCountSlotIds) { + this.pushDownCountSlotIds = Lists.newArrayList(pushDownCountSlotIds); + } + public void setChildrenDistributeExprLists(List> childrenDistributeExprLists) { this.childrenDistributeExprLists = childrenDistributeExprLists; } diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/PhysicalStorageLayerAggregateTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/PhysicalStorageLayerAggregateTest.java index bf5b74f2ea6f25..eb3a42c3868610 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/PhysicalStorageLayerAggregateTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/PhysicalStorageLayerAggregateTest.java @@ -75,7 +75,26 @@ public void testWithoutProject() { .applyImplementation(storageLayerAggregateWithoutProject()) .matches( logicalAggregate( - physicalStorageLayerAggregate().when(agg -> agg.getAggOp() == PushDownAggOp.COUNT) + physicalStorageLayerAggregate().when(agg -> agg.getAggOp() == PushDownAggOp.COUNT + && agg.getCountArgumentExprIds().equals( + ImmutableList.of(olapScan.getOutput().get(0).getExprId()))) + ) + ); + + // COUNT(*) still keeps a placeholder scan slot after column pruning, so its semantic + // argument list must remain empty instead of being inferred from the physical scan shape. + aggregate = new LogicalAggregate<>( + Collections.emptyList(), + ImmutableList.of(new Alias(new Count(), "count_star")), + true, Optional.empty(), olapScan); + context = MemoTestUtils.createCascadesContext(aggregate); + + PlanChecker.from(context) + .applyImplementation(storageLayerAggregateWithoutProject()) + .matches( + logicalAggregate( + physicalStorageLayerAggregate().when(agg -> agg.getAggOp() == PushDownAggOp.COUNT + && agg.getCountArgumentExprIds().isEmpty()) ) ); @@ -138,7 +157,9 @@ public void testWithProject() { .matches( logicalAggregate( logicalProject( - physicalStorageLayerAggregate().when(agg -> agg.getAggOp() == PushDownAggOp.COUNT) + physicalStorageLayerAggregate().when(agg -> agg.getAggOp() == PushDownAggOp.COUNT + && agg.getCountArgumentExprIds().equals( + ImmutableList.of(olapScan.getOutput().get(0).getExprId()))) ) ) ); diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift index c2dd42bd522a16..c146ea67094a92 100644 --- a/gensrc/thrift/PlanNodes.thrift +++ b/gensrc/thrift/PlanNodes.thrift @@ -1709,6 +1709,10 @@ struct TPlanNode { 52: optional TRecCTEScanNode rec_cte_scan_node 53: optional TBucketedAggregationNode bucketed_agg_node 54: optional TLocalExchangeNode local_exchange_node + // COUNT(*) and COUNT(col) share push_down_agg_type_opt=COUNT, but file readers need to know + // whether a projected scan slot is the aggregate argument or merely the placeholder retained by + // column pruning. Empty means row-count semantics; non-empty identifies explicit COUNT columns. + 55: optional list push_down_count_slot_ids // projections is final projections, which means projecting into results and materializing them into the output block. 101: optional list projections From 44270853c4261f2d52e01d98a2bb03d7ec5e95b9 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Tue, 14 Jul 2026 17:45:18 +0800 Subject: [PATCH 5/5] [fix](fe) Fix aggregate strategy import order ### What problem does this PR solve? Issue Number: None Related PR: #65548 Problem Summary: The COUNT pushdown change added ExprId after Expression, violating the FE CustomImportOrder lexicographical rule. Place ExprId before Expression so Checkstyle accepts AggregateStrategies. ### Release note None ### Check List (For Author) - Test: FE Checkstyle via ./build.sh --fe -j32 on Gabriel; Checkstyle passed and compilation later stopped on unrelated cloud storage sources missing vendor SDKs - Behavior changed: No - Does this need documentation: No --- .../doris/nereids/rules/implementation/AggregateStrategies.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/AggregateStrategies.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/AggregateStrategies.java index 8a3e32d60d63d8..a106429facc605 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/AggregateStrategies.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/AggregateStrategies.java @@ -33,8 +33,8 @@ import org.apache.doris.nereids.rules.expression.rules.FoldConstantRuleOnFE; import org.apache.doris.nereids.trees.expressions.Alias; import org.apache.doris.nereids.trees.expressions.Cast; -import org.apache.doris.nereids.trees.expressions.Expression; import org.apache.doris.nereids.trees.expressions.ExprId; +import org.apache.doris.nereids.trees.expressions.Expression; import org.apache.doris.nereids.trees.expressions.IsNull; import org.apache.doris.nereids.trees.expressions.Or; import org.apache.doris.nereids.trees.expressions.Slot;