diff --git a/be/src/exec/operator/scan_operator.cpp b/be/src/exec/operator/scan_operator.cpp index a9e92d08e08613..3bb234d16b19b9 100644 --- a/be/src/exec/operator/scan_operator.cpp +++ b/be/src/exec/operator/scan_operator.cpp @@ -1046,6 +1046,12 @@ TPushAggOp::type ScanLocalState::get_push_down_agg_type() { return _parent->cast()._push_down_agg_type; } +template +const std::optional>& 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 +1237,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..ae93e155852c00 100644 --- a/be/src/exec/operator/scan_operator.h +++ b/be/src/exec/operator/scan_operator.h @@ -18,6 +18,7 @@ #pragma once #include +#include #include #include @@ -74,6 +75,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::optional>& 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 +254,7 @@ class ScanLocalState : public ScanLocalStateBase { const std::vector& scan_ranges) override {} TPushAggOp::type get_push_down_agg_type() override; + const std::optional>& get_push_down_count_slot_ids() const override; std::vector execution_dependencies() override { if (_filter_dependencies.empty()) { @@ -452,6 +455,16 @@ class ScanOperatorX : public OperatorX { TPushAggOp::type _push_down_agg_type; + // Semantic arguments of a pushed-down COUNT. This is deliberately optional because absence + // and an empty list have different meanings during a BE-first rolling upgrade: + // + // - nullopt: an old FE did not send the field, so the new BE must use the normal scan; + // - empty: the new FE explicitly planned COUNT(*)/COUNT(1); + // - non-empty: the new FE explicitly planned COUNT(col). + // + // Treating nullopt as empty would silently reinterpret an old plan as COUNT(*). + std::optional> _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.h b/be/src/exec/scan/file_scanner.h index f05f4d5903dcba..dfb99a86abc697 100644 --- a/be/src/exec/scan/file_scanner.h +++ b/be/src/exec/scan/file_scanner.h @@ -21,6 +21,7 @@ #include #include +#include #include #include #include @@ -79,6 +80,10 @@ class FileScanner : public Scanner { const VExprContextSPtrs& TEST_runtime_filter_partition_prune_ctxs() const { return _runtime_filter_partition_prune_ctxs; } + static TPushAggOp::type TEST_effective_push_down_agg_type( + TPushAggOp::type agg_type, const std::optional>& count_slot_ids) { + return _effective_push_down_agg_type(agg_type, count_slot_ids); + } #endif FileScanner(RuntimeState* state, FileScanLocalState* parent, int64_t limit, @@ -334,9 +339,25 @@ class FileScanner : public Scanner { void _update_adaptive_batch_size_before_truncate(const Block& block); void _update_adaptive_batch_size_after_truncate(const Block& block); + static TPushAggOp::type _effective_push_down_agg_type( + TPushAggOp::type agg_type, const std::optional>& count_slot_ids) { + if (agg_type != TPushAggOp::type::COUNT) { + return agg_type; + } + // V1's CountReader can only emit rows for an upper COUNT(*). It cannot evaluate the NULL + // or CAST semantics of COUNT(col), so a non-empty argument list must use the normal reader. + // nullopt is an old FE plan that predates the argument field; treating it as empty would + // silently reinterpret unknown semantics as COUNT(*). + return count_slot_ids.has_value() && count_slot_ids->empty() ? TPushAggOp::type::COUNT + : TPushAggOp::type::NONE; + } + TPushAggOp::type _get_push_down_agg_type() const { - return _local_state == nullptr ? TPushAggOp::type::NONE - : _local_state->get_push_down_agg_type(); + if (_local_state == nullptr) { + return TPushAggOp::type::NONE; + } + return _effective_push_down_agg_type(_local_state->get_push_down_agg_type(), + _local_state->get_push_down_count_slot_ids()); } // enable the file meta cache only when diff --git a/be/src/exec/scan/file_scanner_v2.cpp b/be/src/exec/scan/file_scanner_v2.cpp index 6700ed71276f45..25944d7a785ff7 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, bool stopped) { + return _should_skip_empty(status, stopped); +} #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, _should_stop || _io_ctx->should_stop)) { + // END_OF_FILE here means the reader discovered a valid split with no data while + // opening or probing it, not that the Scanner has exhausted all splits. Examples + // are a zero-byte CSV with an explicit schema and a Doris Native file containing + // only its 12-byte header. Treat it like V1's empty-file path: finish this range, + // discard partial reader state, and let the loop fetch the next split. + RETURN_IF_ERROR(_table_reader->abort_split()); + COUNTER_UPDATE(_empty_file_counter, 1); + _state->update_num_finished_scan_range(1); + _has_prepared_split = false; + block->clear_column_data(cast_set(_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, _should_stop || _io_ctx->should_stop)) { + // Schema discovery can reach EOF before a split becomes prepared. A header-only Native + // file follows this path, while a reader that discovers emptiness on its first + // get_block() follows the symmetric branch in _get_block_impl(). Both paths must + // advance exactly one scan range and preserve later files in the same scan. + RETURN_IF_ERROR(_table_reader->abort_split()); + COUNTER_UPDATE(_empty_file_counter, 1); + _state->update_num_finished_scan_range(1); + continue; + } RETURN_IF_ERROR(status); if (_table_reader->current_split_pruned()) { _state->update_num_finished_scan_range(1); @@ -448,6 +478,21 @@ Status FileScannerV2::_init_table_reader(const TFileRangeDesc& range) { VExprContextSPtrs table_conjuncts; RETURN_IF_ERROR(_build_table_conjuncts(&table_conjuncts)); + std::optional> push_down_count_columns; + const auto& push_down_count_slot_ids = _local_state->get_push_down_count_slot_ids(); + if (push_down_count_slot_ids.has_value()) { + push_down_count_columns.emplace(); + push_down_count_columns->reserve(push_down_count_slot_ids->size()); + for (const auto slot_id : *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), @@ -458,6 +503,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(); @@ -527,6 +573,14 @@ 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, 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 { 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..452b1652bee461 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, bool stopped); #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 stopped); 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/native/native_reader.cpp b/be/src/format_v2/native/native_reader.cpp index 0e348f3de2ccf5..5d0984084a6d41 100644 --- a/be/src/format_v2/native/native_reader.cpp +++ b/be/src/format_v2/native/native_reader.cpp @@ -231,6 +231,16 @@ Status NativeReader::_read_next_pblock(std::string* buffer, bool* eof) const { return Status::OK(); } + const auto remaining_prefix_bytes = _file_size - _current_offset; + if (remaining_prefix_bytes < sizeof(uint64_t)) { + // Header-only is the one valid empty-file representation and returned above. Once any + // prefix byte exists, all eight bytes are mandatory. For example, `header + 4 bytes` is a + // truncated Native file, not EOF that FileScannerV2 may skip as an empty split. + return Status::InternalError( + "truncated native block length in file {} at offset {}, expect {}, remaining {}", + _file_description->path, _current_offset, sizeof(uint64_t), remaining_prefix_bytes); + } + uint64_t block_len = 0; Slice len_slice(reinterpret_cast(&block_len), sizeof(block_len)); size_t bytes_read = 0; @@ -247,8 +257,22 @@ 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)); + } + + const auto remaining_block_bytes = cast_set(_file_size - _current_offset); + if (block_len > remaining_block_bytes) { + // Validate before allocating `block_len` bytes. Besides reporting a precise corruption + // error, this prevents a truncated file with a damaged length prefix from requesting an + // allocation much larger than the physical file. + return Status::InternalError( + "truncated native block body in file {} at offset {}, expect {}, remaining {}", + _file_description->path, _current_offset, block_len, remaining_block_bytes); } buffer->assign(block_len, '\0'); 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..b51fc89d7b5575 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,75 @@ namespace doris::format::parquet { namespace detail { +namespace { + +bool page_cache_range_less(const ParquetPageCacheRange& lhs, const ParquetPageCacheRange& rhs) { + return lhs.offset < rhs.offset || (lhs.offset == rhs.offset && lhs.size < rhs.size); +} + +} // namespace + +ParquetPageCacheRangeIndex::ParquetPageCacheRangeIndex(size_t max_ranges) + : _max_ranges(max_ranges) { + DORIS_CHECK(_max_ranges > 0); +} + +void ParquetPageCacheRangeIndex::insert(ParquetPageCacheRange range) { + std::lock_guard lock(_mutex); + auto it = std::lower_bound(_ranges.begin(), _ranges.end(), range, page_cache_range_less); + if (it != _ranges.end() && it->offset == range.offset && it->size == range.size) { + return; + } + if (_ranges.size() >= _max_ranges) { + _ranges.erase(_ranges.begin()); + it = std::lower_bound(_ranges.begin(), _ranges.end(), range, page_cache_range_less); + } + _ranges.insert(it, range); +} + +void ParquetPageCacheRangeIndex::erase(ParquetPageCacheRange range) { + std::lock_guard lock(_mutex); + const auto it = std::lower_bound(_ranges.begin(), _ranges.end(), range, page_cache_range_less); + if (it != _ranges.end() && it->offset == range.offset && it->size == range.size) { + _ranges.erase(it); + } +} + +std::vector 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( int64_t position, int64_t nbytes, const std::vector& cached_ranges) { if (position < 0 || nbytes <= 0) { @@ -133,59 +201,14 @@ 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; +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, @@ -213,7 +236,11 @@ class DorisRandomAccessFile final : public arrow::io::RandomAccessFile { _base_file_reader(_file_reader), _io_ctx(io_ctx), _enable_page_cache(enable_page_cache), - _page_cache_file_key(std::move(page_cache_file_key)) { + _page_cache_file_key(std::move(page_cache_file_key)), + _cached_page_range_index( + _enable_page_cache && !_page_cache_file_key.empty() + ? cached_page_range_directory().get_or_create(_page_cache_file_key) + : nullptr) { DORIS_CHECK(_file_reader != nullptr); if (auto tracing_reader = std::dynamic_pointer_cast(_file_reader)) { _file_reader_stats = tracing_reader->stats(); @@ -226,6 +253,10 @@ class DorisRandomAccessFile final : public arrow::io::RandomAccessFile { arrow::Status Close() override { if (!_closed) { collect_active_merge_range_profile(); + std::lock_guard lock(_page_cache_mutex); + // Page payloads and their bounded per-file range index intentionally outlive this + // reader for warm scans. Only reader-specific projected ranges are released here. + std::vector().swap(_page_cache_ranges); _closed = true; } return arrow::Status::OK(); @@ -363,7 +394,8 @@ class DorisRandomAccessFile final : public arrow::io::RandomAccessFile { private: bool page_cache_enabled() const { return _enable_page_cache && !config::disable_storage_page_cache && - StoragePageCache::instance() != nullptr && !_page_cache_file_key.empty(); + StoragePageCache::instance() != nullptr && !_page_cache_file_key.empty() && + _cached_page_range_index != nullptr; } bool range_in_page_cache_scope(int64_t position, int64_t nbytes) const { @@ -391,7 +423,7 @@ class DorisRandomAccessFile final : public arrow::io::RandomAccessFile { if (!StoragePageCache::instance()->lookup( page_cache_key(cached_range.offset, cached_range.size), &handle, segment_v2::DATA_PAGE)) { - unregister_cached_page_range(_page_cache_file_key, cached_range); + _cached_page_range_index->erase(cached_range); return false; } Slice cached = handle.data(); @@ -404,8 +436,8 @@ class DorisRandomAccessFile final : public arrow::io::RandomAccessFile { } bool try_read_from_cached_ranges(int64_t position, int64_t nbytes, void* out) { - auto plan = detail::plan_page_cache_range_read( - position, nbytes, cached_page_ranges_for_file(_page_cache_file_key)); + auto plan = detail::plan_page_cache_range_read(position, nbytes, + _cached_page_range_index->ranges()); if (plan.empty()) { return false; } @@ -456,7 +488,8 @@ class DorisRandomAccessFile final : public arrow::io::RandomAccessFile { PageCacheHandle handle; StoragePageCache::instance()->insert(page_cache_key(position, nbytes), page, &handle, segment_v2::DATA_PAGE); - register_cached_page_range(_page_cache_file_key, position, nbytes); + _cached_page_range_index->insert( + ParquetPageCacheRange {.offset = position, .size = nbytes}); ++_page_cache_stats.write_count; ++_page_cache_stats.compressed_write_count; } @@ -514,6 +547,7 @@ class DorisRandomAccessFile final : public arrow::io::RandomAccessFile { std::string _page_cache_file_key; mutable std::mutex _page_cache_mutex; std::vector _page_cache_ranges; + 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 0dca52244957d7..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" @@ -66,6 +69,39 @@ struct ParquetPageCacheStats { 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); + + std::vector ranges() const; + size_t size() const; + +private: + 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..1c5d00e7d76e1c 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 = @@ -614,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/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/src/format_v2/table_reader.cpp b/be/src/format_v2/table_reader.cpp index 09c5d65e82b7fd..bb6232d5440180 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); @@ -807,7 +808,12 @@ Status TableReader::prepare_split(const SplitReadOptions& options) { // active and no predicate can arrive later. The metadata path can return several batches for // one split; after its first synthetic batch there is no way to recover the real rows if a // runtime filter arrives before the next scheduler turn. - if (_push_down_agg_type == TPushAggOp::type::COUNT && options.all_runtime_filters_applied && + // Table-level metadata only contains the number of rows; it cannot evaluate an expression or + // the NULL state of a COUNT argument. Require the new FE's explicit empty argument list, which + // means COUNT(*)/COUNT(1). A non-empty list means COUNT(col), while nullopt comes from an old FE + // whose COUNT semantics are unknown during a BE-first rolling upgrade. + if (_push_down_agg_type == TPushAggOp::type::COUNT && _push_down_count_columns.has_value() && + _push_down_count_columns->empty() && options.all_runtime_filters_applied && _conjuncts.empty() && options.current_range.__isset.table_format_params && options.current_range.table_format_params.__isset.table_level_row_count) { DORIS_CHECK(options.current_range.table_format_params.table_level_row_count >= -1); diff --git a/be/src/format_v2/table_reader.h b/be/src/format_v2/table_reader.h index b8457ee9b4a31e..1c110a8a3c6687 100644 --- a/be/src/format_v2/table_reader.h +++ b/be/src/format_v2/table_reader.h @@ -135,6 +135,10 @@ 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. nullopt means an old FE did not send the + // semantic argument field, while an explicit empty vector means COUNT(*)/COUNT(1). Keeping + // those states separate prevents a rolling-upgrade plan from being reinterpreted by a new BE. + const std::optional> push_down_count_columns = std::nullopt; // Digest of stable pushed-down predicates. A zero digest disables condition cache. uint64_t condition_cache_digest = 0; }; @@ -145,7 +149,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 +953,31 @@ class TableReader { return false; } if (agg_type == TPushAggOp::type::COUNT) { - return true; + // Old FEs do not serialize push_down_count_slot_ids. During the supported BE-first + // rolling upgrade, nullopt therefore means "COUNT semantics are unknown", not + // COUNT(*). Fall back to reading rows until the FE explicitly sends either an empty + // list for COUNT(*) or one slot for COUNT(col). + if (!_push_down_count_columns.has_value()) { + return false; + } + // 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(); + // Metadata COUNT skips TableReader's normal materialization path. Only a trivial + // mapping is safe: for example, a nullable Parquet INT mapped to a NOT NULL table + // BIGINT normally needs both an INT->BIGINT cast and nullability validation. Counting + // footer values directly would bypass both operations and could hide invalid data. + return mapping.file_local_id.has_value() && mapping.file_type != nullptr && + mapping.table_type != nullptr && mapping.is_trivial && + 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,24 +1460,17 @@ 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 - // 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. - 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; - column.projection = - LocalColumnIndex::top_level(LocalColumnId(*mapping.file_local_id)); - request->columns.push_back(std::move(column)); - } + DORIS_CHECK(_push_down_count_columns.has_value()); + // 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(); } @@ -1466,6 +1487,18 @@ class TableReader { return Status::OK(); } + const ColumnMapping& _push_down_count_mapping() const { + DORIS_CHECK(_push_down_count_columns.has_value()); + 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) { @@ -1570,6 +1603,7 @@ class TableReader { const std::vector* _file_slot_descs = nullptr; FileFormat _format; TPushAggOp::type _push_down_agg_type = TPushAggOp::type::NONE; + std::optional> _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/exec/scan/file_scanner_v2_test.cpp b/be/test/exec/scan/file_scanner_v2_test.cpp index abd3d1e76d7ab3..50b39a113ff79b 100644 --- a/be/test/exec/scan/file_scanner_v2_test.cpp +++ b/be/test/exec/scan/file_scanner_v2_test.cpp @@ -80,6 +80,23 @@ TFileRangeDesc legacy_paimon_jni_range_without_reader_type() { return range; } +TEST(FileScannerTest, V1CountPushdownRequiresExplicitCountStarArguments) { + EXPECT_EQ(TPushAggOp::type::COUNT, FileScanner::TEST_effective_push_down_agg_type( + TPushAggOp::type::COUNT, std::vector {})); + + // A missing field is an old FE plan with unknown COUNT semantics, not COUNT(*). + EXPECT_EQ(TPushAggOp::type::NONE, FileScanner::TEST_effective_push_down_agg_type( + TPushAggOp::type::COUNT, std::nullopt)); + // V1 cannot evaluate COUNT(col) NULL/CAST semantics before replacing the reader with + // CountReader, so an explicit argument must use the normal scan path. + EXPECT_EQ(TPushAggOp::type::NONE, FileScanner::TEST_effective_push_down_agg_type( + TPushAggOp::type::COUNT, std::vector {7})); + + // The COUNT argument field must not affect other storage-layer aggregate operations. + EXPECT_EQ(TPushAggOp::type::MINMAX, FileScanner::TEST_effective_push_down_agg_type( + TPushAggOp::type::MINMAX, std::nullopt)); +} + struct RetryableCloseState { int close_calls = 0; }; @@ -494,6 +511,16 @@ 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"), 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 // 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..2745ecf852c38b 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)); } @@ -350,6 +351,34 @@ TEST(NativeV2ReaderTest, RejectsTruncatedBlockDuringSchemaProbe) { static_cast(io::global_local_filesystem()->delete_file(path)); } +TEST(NativeV2ReaderTest, RejectsPartialBlockLengthAsCorruption) { + std::filesystem::create_directories("./log"); + RuntimeState state; + RuntimeProfile profile("native_v2_reader_partial_length_test"); + + std::string header; + header.append(DORIS_NATIVE_MAGIC, sizeof(DORIS_NATIVE_MAGIC)); + uint8_t version_buffer[sizeof(uint32_t)]; + encode_fixed32_le(version_buffer, DORIS_NATIVE_FORMAT_VERSION); + header.append(reinterpret_cast(version_buffer), sizeof(version_buffer)); + + for (size_t prefix_bytes = 1; prefix_bytes < sizeof(uint64_t); ++prefix_bytes) { + const auto path = "./log/native_v2_partial_length_" + std::to_string(prefix_bytes) + "_" + + UniqueId::gen_uid().to_string() + ".native"; + auto content = header; + content.append(prefix_bytes, '\0'); + ASSERT_TRUE(write_file(path, content).ok()); + + auto reader = create_reader(path, &state, &profile); + ASSERT_TRUE(reader->init(&state).ok()); + std::vector schema; + const auto status = reader->get_schema(&schema); + EXPECT_TRUE(status.is()) << status; + EXPECT_NE(status.to_string().find("truncated native block length"), std::string::npos); + static_cast(io::global_local_filesystem()->delete_file(path)); + } +} + TEST(NativeV2ReaderTest, RejectsZeroLengthBlockAndInvalidPBlock) { std::filesystem::create_directories("./log"); RuntimeState state; @@ -374,7 +403,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_page_cache_range_test.cpp b/be/test/format_v2/parquet/parquet_page_cache_range_test.cpp index fac89b31c422d8..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,6 +115,45 @@ TEST(ParquetPageCacheRangeTest, InvalidRequestMisses) { EXPECT_TRUE(detail::plan_page_cache_range_read(100, -1, cached_ranges).empty()); } +TEST(ParquetPageCacheRangeTest, PerFileRangeIndexDeduplicatesAndEvictsAtCapacity) { + detail::ParquetPageCacheRangeIndex index(3); + index.insert({200, 20}); + index.insert({100, 30}); + index.insert({100, 10}); + index.insert({100, 30}); + + 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}); + 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); + + directory.get_or_create("file-b"); + directory.get_or_create("file-c"); + EXPECT_EQ(directory.size(), 2); +} + 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_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/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/be/test/format_v2/table_reader_test.cpp b/be/test/format_v2/table_reader_test.cpp index 0d484f60d9e5de..4e6899a3bc7860 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; @@ -1510,6 +1512,7 @@ TEST(TableReaderTest, RefreshedConjunctDisablesTableLevelCount) { .runtime_state = &state, .scanner_profile = nullptr, .push_down_agg_type = TPushAggOp::type::COUNT, + .push_down_count_columns = std::vector {}, }) .ok()); @@ -1552,6 +1555,7 @@ TEST(TableReaderTest, PendingRuntimeFilterDisablesTableLevelCount) { .runtime_state = &state, .scanner_profile = nullptr, .push_down_agg_type = TPushAggOp::type::COUNT, + .push_down_count_columns = std::vector {}, }) .ok()); @@ -1652,6 +1656,7 @@ TEST(TableReaderTest, SlotlessConjunctDisablesAggregatePushdown) { .runtime_state = &state, .scanner_profile = nullptr, .push_down_agg_type = TPushAggOp::type::COUNT, + .push_down_count_columns = std::vector {}, }) .ok()); @@ -1714,11 +1719,12 @@ TEST(TableReaderTest, AbortSplitClearsReaderAfterIgnorableNotFound) { } TEST(TableReaderTest, PushDownCountRecordsReaderRowsBeforeClosingReader) { + const auto nullable_int_type = make_nullable(std::make_shared()); std::vector file_schema; - file_schema.push_back(make_file_column(0, "id", std::make_shared())); + file_schema.push_back(make_file_column(0, "id", nullable_int_type)); std::vector projected_columns; - projected_columns.push_back(make_table_column(0, "id", std::make_shared())); + projected_columns.push_back(make_table_column(0, "id", nullable_int_type)); set_name_identifiers(&projected_columns); io::FileReaderStats file_reader_stats; @@ -1739,6 +1745,8 @@ TEST(TableReaderTest, PushDownCountRecordsReaderRowsBeforeClosingReader) { .runtime_state = &state, .scanner_profile = nullptr, .push_down_agg_type = TPushAggOp::type::COUNT, + .push_down_count_columns = + std::vector {GlobalIndex(0)}, }) .ok()); @@ -1753,6 +1761,179 @@ 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. + 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 = std::vector {}, + }) + .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, PushDownCountFallsBackWhenSemanticArgumentsAreAbsent) { + const auto nullable_int_type = make_nullable(std::make_shared()); + std::vector file_schema; + file_schema.push_back(make_file_column(0, "id", nullable_int_type)); + + std::vector projected_columns; + projected_columns.push_back(make_table_column(0, "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, + // Simulate an old FE: it can request COUNT pushdown but cannot + // serialize push_down_count_slot_ids. + .push_down_count_columns = std::nullopt, + }) + .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(), 2); + // The explicit aggregate_count=3 would be returned if absence were confused with COUNT(*). + EXPECT_FALSE(fake_state->last_aggregate_request.has_value()); +} + +TEST(TableReaderTest, PushDownCountFallsBackForNullableToRequiredMapping) { + const auto nullable_int_type = make_nullable(std::make_shared()); + std::vector file_schema; + file_schema.push_back(make_file_column(0, "id", nullable_int_type)); + + std::vector projected_columns; + projected_columns.push_back(make_table_column(0, "id", std::make_shared())); + // make_table_column models the usual nullable external-table descriptor. Override it here to + // reproduce an evolved table contract that declares the mapped physical column required. + projected_columns[0].type = std::make_shared(); + 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, + .push_down_count_columns = + std::vector {GlobalIndex(0)}, + }) + .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; + // The normal scan rejects the nullable physical column because it cannot satisfy the required + // table contract. Footer COUNT would bypass that validation and incorrectly return 3. + EXPECT_FALSE(reader.get_block(&block, &eos).ok()); + EXPECT_FALSE(fake_state->last_aggregate_request.has_value()); +} + +TEST(TableReaderTest, PushDownCountFallsBackForCastMapping) { + const auto nullable_int_type = make_nullable(std::make_shared()); + const auto nullable_bigint_type = make_nullable(std::make_shared()); + std::vector file_schema; + file_schema.push_back(make_file_column(0, "id", nullable_int_type)); + + std::vector projected_columns; + projected_columns.push_back(make_table_column(0, "id", nullable_bigint_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, + .push_down_count_columns = + std::vector {GlobalIndex(0)}, + }) + .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(), 2); + EXPECT_TRUE(block.get_by_position(0).type->equals(*nullable_bigint_type)); + // INT->BIGINT is a non-trivial mapping, so COUNT must not skip the cast/materialization path. + EXPECT_FALSE(fake_state->last_aggregate_request.has_value()); } TEST(TableReaderTest, PushDownCountStopConvertsAggregateEndOfFileToEos) { @@ -1780,6 +1961,7 @@ TEST(TableReaderTest, PushDownCountStopConvertsAggregateEndOfFileToEos) { .runtime_state = &state, .scanner_profile = nullptr, .push_down_agg_type = TPushAggOp::type::COUNT, + .push_down_count_columns = std::vector {}, }) .ok()); @@ -1820,6 +2002,7 @@ TEST(TableReaderTest, DebugStringCoversReaderStateAndEnumNames) { .runtime_state = &state, .scanner_profile = nullptr, .push_down_agg_type = TPushAggOp::type::COUNT, + .push_down_count_columns = std::vector {}, }) .ok()); @@ -2411,6 +2594,7 @@ TEST(TableReaderTest, PushDownCountFromNewParquetReader) { .runtime_state = &state, .scanner_profile = nullptr, .push_down_agg_type = TPushAggOp::type::COUNT, + .push_down_count_columns = std::vector {}, }) .ok()); ASSERT_TRUE(reader.prepare_split(build_split_options(file_path)).ok()); @@ -2452,6 +2636,7 @@ TEST(TableReaderTest, TableLevelCountUsesAssignedRowCount) { .runtime_state = &state, .scanner_profile = nullptr, .push_down_agg_type = TPushAggOp::type::COUNT, + .push_down_count_columns = std::vector {}, }) .ok()); auto split_options = build_split_options(file_path); @@ -2483,6 +2668,56 @@ TEST(TableReaderTest, TableLevelCountUsesAssignedRowCount) { std::filesystem::remove_all(test_dir); } +TEST(TableReaderTest, TableLevelCountRequiresExplicitCountStarArguments) { + const auto test_dir = std::filesystem::temp_directory_path() / + "doris_table_reader_table_count_arguments_test"; + std::filesystem::remove_all(test_dir); + std::filesystem::create_directories(test_dir); + + const auto file_path = (test_dir / "split.parquet").string(); + write_int_pair_parquet_file(file_path, {1, 2, 3}, {10, 20, 30}, {"one", "two", "three"}); + + std::vector projected_columns; + projected_columns.push_back(make_table_column(0, "id", std::make_shared())); + set_name_identifiers(&projected_columns); + + const std::vector>> unsafe_count_arguments { + std::nullopt, std::vector {GlobalIndex(0)}}; + for (const auto& count_arguments : unsafe_count_arguments) { + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + TableReader reader; + 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, + .push_down_count_columns = count_arguments, + }) + .ok()); + auto split_options = build_split_options(file_path); + // Five metadata rows deliberately disagree with the three physical rows. nullopt models an + // old FE, while the non-empty vector models COUNT(id); neither may be treated as COUNT(*). + set_table_level_row_count(&split_options, 5); + ASSERT_TRUE(reader.prepare_split(split_options).ok()); + + size_t total_rows = 0; + bool eos = false; + while (!eos) { + Block block = build_table_block(projected_columns); + ASSERT_TRUE(reader.get_block(&block, &eos).ok()); + total_rows += block.rows(); + } + EXPECT_EQ(3, total_rows); + ASSERT_TRUE(reader.close().ok()); + } + + std::filesystem::remove_all(test_dir); +} + TEST(TableReaderTest, PushDownMinMaxFromNewParquetReader) { const auto test_dir = std::filesystem::temp_directory_path() / "doris_table_reader_minmax_test"; std::filesystem::remove_all(test_dir); @@ -3299,6 +3534,7 @@ TEST(TableReaderTest, PushDownCountOnlyUsesSelectedRowGroupInFileRange) { .runtime_state = &state, .scanner_profile = nullptr, .push_down_agg_type = TPushAggOp::type::COUNT, + .push_down_count_columns = std::vector {}, }) .ok()); ASSERT_TRUE(reader.prepare_split(build_split_options_for_row_group_mid(file_path, 2)).ok()); @@ -3338,6 +3574,7 @@ TEST(TableReaderTest, PushDownCountFallsBackWithTableConjunct) { .runtime_state = &state, .scanner_profile = nullptr, .push_down_agg_type = TPushAggOp::type::COUNT, + .push_down_count_columns = std::vector {}, }) .ok()); ASSERT_TRUE(reader.prepare_split(build_split_options(file_path)).ok()); @@ -3379,6 +3616,7 @@ TEST(TableReaderTest, PushDownCountFallsBackWithFilter) { .runtime_state = &state, .scanner_profile = nullptr, .push_down_agg_type = TPushAggOp::type::COUNT, + .push_down_count_columns = std::vector {}, }) .ok()); ASSERT_TRUE(reader.prepare_split(build_split_options(file_path)).ok()); 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..e91d458a098c81 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,6 +33,7 @@ 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.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; @@ -570,6 +571,7 @@ private LogicalAggregate storageLayerAggregate( Map, PushDownAggOp> supportedAgg = PushDownAggOp.supportedFunctions(); boolean containsCount = false; + boolean countHasCastArgument = false; Set checkNullSlots = new HashSet<>(); Set expressionAfterProject = new HashSet<>(); @@ -591,6 +593,7 @@ private LogicalAggregate storageLayerAggregate( checkNullSlots.add((SlotReference) arg0); expressionAfterProject.add(arg0); } else if (arg0 instanceof Cast) { + countHasCastArgument = true; Expression child0 = arg0.child(0); if (child0 instanceof SlotReference) { checkNullSlots.add((SlotReference) child0); @@ -667,6 +670,7 @@ private LogicalAggregate storageLayerAggregate( return canNotPush; } else { if (needCheckSlotNull) { + countHasCastArgument = true; checkNullSlots.add((SlotReference) argument.child(0)); } } @@ -677,6 +681,16 @@ private LogicalAggregate storageLayerAggregate( argumentsOfAggregateFunction = processedExpressions; } + // File aggregate metadata can describe COUNT(*) or COUNT(file_column), but it cannot + // describe the CAST wrapped around a COUNT argument. Dropping that CAST is incorrect even + // when the source column is NOT NULL. For example, a non-null DOUBLE value outside the INT + // range becomes NULL for CAST(double_col AS INT), so COUNT(CAST(double_col AS INT)) must + // exclude it while a footer-level COUNT(double_col) would include it. Keep OLAP's existing + // storage-layer behavior unchanged, and make external files evaluate the CAST normally. + if (logicalScan instanceof LogicalFileScan && countHasCastArgument) { + return canNotPush; + } + Set pushDownAggOps = functionClasses.stream() .map(supportedAgg::get) .collect(Collectors.toSet()); @@ -690,6 +704,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 +758,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 +775,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 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 00000000000000..22eff58b7a2389 Binary files /dev/null and b/regression-test/data/external_table_p0/tvf/file_scanner_v2_struct_0_bigint.parquet differ 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 00000000000000..5a9ca0baa47e89 Binary files /dev/null and b/regression-test/data/external_table_p0/tvf/file_scanner_v2_struct_1_int.parquet differ 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 00000000000000..5b973181a56148 Binary files /dev/null and b/regression-test/data/external_table_p0/tvf/file_scanner_v2_unsupported_time.parquet differ 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..c8312571c56e36 --- /dev/null +++ b/regression-test/data/external_table_p0/tvf/test_file_scanner_v2_review_fixes.out @@ -0,0 +1,17 @@ +-- 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 + +-- !count_evolved_struct -- +4 + +-- !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..d62435ac1c38f2 --- /dev/null +++ b/regression-test/suites/external_table_p0/tvf/test_file_scanner_v2_review_fixes.groovy @@ -0,0 +1,111 @@ +// 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 + + // COUNT(col) sees a non-trivial mapping for the file whose STRUCT leaf is INT while the merged + // table type is BIGINT. It must fall back to the normal scan so schema-evolution casts and + // nullability checks run instead of counting footer values directly. + order_qt_count_evolved_struct """ + SELECT COUNT(col.a) + FROM local( + "file_path" = "${remotePath}/file_scanner_v2_struct_*.parquet", + "backend_id" = "${backendId}", + "format" = "parquet") + """ + + // 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 + """ + + // 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" + } + +}