diff --git a/be/src/exec/scan/file_scanner_v2.cpp b/be/src/exec/scan/file_scanner_v2.cpp index 10b5f850ea36f7..519237ea85c252 100644 --- a/be/src/exec/scan/file_scanner_v2.cpp +++ b/be/src/exec/scan/file_scanner_v2.cpp @@ -492,13 +492,19 @@ Status FileScannerV2::_prepare_table_reader_split(const TFileRangeDesc& range, std::map partition_values) { format::FileFormat current_split_format; RETURN_IF_ERROR(_to_file_format(get_range_format_type(*_params, range), ¤t_split_format)); + VExprContextSPtrs conjuncts; + RETURN_IF_ERROR(_build_table_conjuncts(&conjuncts)); VExprContextSPtrs partition_prune_conjuncts; if (_state->query_options().enable_runtime_filter_partition_prune) { RETURN_IF_ERROR(_build_table_conjuncts(&partition_prune_conjuncts)); } RETURN_IF_ERROR(_table_reader->prepare_split({ .partition_values = std::move(partition_values), + .conjuncts = std::move(conjuncts), .partition_prune_conjuncts = std::move(partition_prune_conjuncts), + // A metadata COUNT split may span scheduler turns. Do not enter that irreversible + // synthetic-row path while a runtime filter can still arrive between batches. + .all_runtime_filters_applied = _applied_rf_num == _total_rf_num, .cache = _kv_cache, .current_range = range, .current_split_format = current_split_format, diff --git a/be/src/format_v2/jni/jni_table_reader.cpp b/be/src/format_v2/jni/jni_table_reader.cpp index 7cec22a7c5999f..9416659b8f5bfc 100644 --- a/be/src/format_v2/jni/jni_table_reader.cpp +++ b/be/src/format_v2/jni/jni_table_reader.cpp @@ -32,14 +32,6 @@ namespace doris::format { Status JniTableReader::init(TableReadOptions&& options) { RETURN_IF_ERROR(TableReader::init(std::move(options))); _init_profile(); - - // JNI readers do not go through TableReader::open_reader(), where file-local filters are - // prepared for file readers. They execute table-level conjuncts directly on the JNI block. - RowDescriptor row_desc; - for (const auto& conjunct : _conjuncts) { - RETURN_IF_ERROR(conjunct->prepare(_runtime_state, row_desc)); - RETURN_IF_ERROR(conjunct->open(_runtime_state)); - } return Status::OK(); } @@ -55,6 +47,13 @@ Status JniTableReader::prepare_split(const SplitReadOptions& options) { if (_is_table_level_count_active()) { return Status::OK(); } + // JNI readers do not go through TableReader::open_reader(), where native readers prepare + // file-local filters. Prepare the fresh per-split snapshot before it filters JNI blocks. + RowDescriptor row_desc; + for (const auto& conjunct : _conjuncts) { + RETURN_IF_ERROR(conjunct->prepare(_runtime_state, row_desc)); + RETURN_IF_ERROR(conjunct->open(_runtime_state)); + } // Subclasses populate split-specific scanner params before calling this method, so the Java // scanner can be opened here instead of being lazily opened by the first get_block() call. return _open_jni_scanner(); diff --git a/be/src/format_v2/parquet/reader/column_reader.cpp b/be/src/format_v2/parquet/reader/column_reader.cpp index 1b6e66beefe860..352fbbd7c3d215 100644 --- a/be/src/format_v2/parquet/reader/column_reader.cpp +++ b/be/src/format_v2/parquet/reader/column_reader.cpp @@ -606,14 +606,34 @@ Status ParquetColumnReader::build_nested_column(int64_t, MutableColumnPtr&, int6 _name); } -Status ParquetColumnReader::skip_nested_column(int64_t rows) { - auto scratch_column = _type->create_column(); - int64_t values_read = 0; - RETURN_IF_ERROR(build_nested_column(rows, scratch_column, &values_read)); - if (values_read != rows) { - return Status::Corruption("Failed to skip nested parquet column {}: skipped {} of {} rows", - _name, values_read, rows); +Status ParquetColumnReader::consume_nested_column(int64_t, int64_t*) { + return Status::NotSupported("Parquet nested column consume is not supported for column {}", + _name); +} + +Status ParquetColumnReader::skip_nested_rows(int64_t rows) { + if (rows <= 0) { + return Status::OK(); + } + + // A nested parent row may expand to many child values. Capping the number of parent rows per + // loaded batch bounds that amplification for large holes. The consume interface advances the + // loaded definition/repetition levels recursively without constructing a discarded Column. + constexpr int64_t MAX_NESTED_SKIP_BATCH_SIZE = 4096; + int64_t remaining_rows = rows; + while (remaining_rows > 0) { + const int64_t batch_rows = std::min(remaining_rows, MAX_NESTED_SKIP_BATCH_SIZE); + RETURN_IF_ERROR(load_nested_levels_batch(batch_rows)); + int64_t rows_consumed = 0; + RETURN_IF_ERROR(consume_nested_column(batch_rows, &rows_consumed)); + if (rows_consumed != batch_rows) { + return Status::Corruption( + "Failed to skip nested parquet column {}: skipped {} of {} rows in batch", + _name, rows_consumed, batch_rows); + } + remaining_rows -= batch_rows; } + update_reader_skip_rows(rows); return Status::OK(); } diff --git a/be/src/format_v2/parquet/reader/column_reader.h b/be/src/format_v2/parquet/reader/column_reader.h index 1d63e03ca9cf43..51dbd44c11c226 100644 --- a/be/src/format_v2/parquet/reader/column_reader.h +++ b/be/src/format_v2/parquet/reader/column_reader.h @@ -80,17 +80,33 @@ class ParquetColumnReader { virtual Status load_nested_batch(int64_t rows); - // Shape-only load interface for COUNT(col). Implementations only guarantee that - // nested_definition_levels(), nested_repetition_levels(), and nested_levels_written() are available; - // value_indices and values_column are not guaranteed, so callers must not call build_nested_column() afterwards. - // This protocol lets the V2 aggregation path avoid Doris-side value materialization even when - // the representative ARRAY/STRUCT leaf is STRING/BINARY; normal scans still use load_nested_batch(). + // Shape-only load interface for COUNT(col) and skip. Implementations guarantee only that + // nested_definition_levels(), nested_repetition_levels(), and nested_levels_written() are + // available; value indices and payload columns may be absent. Callers may inspect the levels or + // call consume_nested_column(), but must not call build_nested_column() afterwards. For example, + // skipping ARRAY uses this method to find ARRAY boundaries without constructing a + // ColumnString. The underlying Arrow reader may still decode a page because it has no public + // levels-only API. Normal scans that need output values use load_nested_batch() instead. virtual Status load_nested_levels_batch(int64_t rows); virtual Status build_nested_column(int64_t length_upper_bound, MutableColumnPtr& column, int64_t* values_read); - virtual Status skip_nested_column(int64_t rows); + // Consume logical values from a batch previously loaded by load_nested_batch() or + // load_nested_levels_batch() without appending them to an output Column. Implementations must + // advance exactly the same nested level cursors and perform the same shape/null/alignment + // validation as build_nested_column(). The levels-only form is preferred for skip paths because + // it avoids transferring leaf payloads into Doris Columns when they will be discarded. + // + // `length_upper_bound` is expressed at this reader's logical level, not in physical leaf + // values. For example, consuming two rows from ARRAY [[1, 2], []] consumes two parent ARRAY + // rows but only two element values. A MAP implementation must also consume key/value streams + // in lockstep, while a nullable STRUCT consumes no child value for a null parent. + // + // Callers must not use the ordinary skip() after either load call: the leaf stream has already + // advanced into an in-memory nested batch, and doing so would advance it twice. + // `values_consumed` may be smaller than the requested bound only when the loaded batch ends. + virtual Status consume_nested_column(int64_t length_upper_bound, int64_t* values_consumed); virtual const std::vector& nested_definition_levels() const; virtual const std::vector& nested_repetition_levels() const; @@ -109,6 +125,10 @@ class ParquetColumnReader { ParquetColumnReader(const ParquetColumnSchema& schema, const DataTypePtr type, ParquetColumnReaderProfile profile = {}); ParquetColumnReader() = default; + // Load shape levels and consume skipped parent rows in bounded batches. The bound limits level + // memory when a parent expands to many children; the levels-only load plus + // consume_nested_column() avoids payload materialization and output Columns. + Status skip_nested_rows(int64_t rows); void update_reader_read_rows(int64_t rows) const; void update_reader_skip_rows(int64_t rows) const; diff --git a/be/src/format_v2/parquet/reader/list_column_reader.cpp b/be/src/format_v2/parquet/reader/list_column_reader.cpp index aaf8f6635f1af0..c042fc99b512aa 100644 --- a/be/src/format_v2/parquet/reader/list_column_reader.cpp +++ b/be/src/format_v2/parquet/reader/list_column_reader.cpp @@ -47,19 +47,7 @@ Status ListColumnReader::read(int64_t rows, MutableColumnPtr& column, int64_t* r } Status ListColumnReader::skip(int64_t rows) { - if (rows <= 0) { - return Status::OK(); - } - auto scratch_column = _type->create_column(); - RETURN_IF_ERROR(load_nested_batch(rows)); - int64_t rows_read = 0; - RETURN_IF_ERROR(build_nested_column(rows, scratch_column, &rows_read)); - if (rows_read != rows) { - return Status::Corruption("Failed to skip parquet LIST column {}: skipped {} of {} rows", - _name, rows_read, rows); - } - update_reader_skip_rows(rows); - return Status::OK(); + return skip_nested_rows(rows); } Status ListColumnReader::load_nested_batch(int64_t rows) { @@ -68,27 +56,53 @@ Status ListColumnReader::load_nested_batch(int64_t rows) { return _element_reader->load_nested_batch(rows); } +Status ListColumnReader::load_nested_levels_batch(int64_t rows) { + DORIS_CHECK(_element_reader != nullptr); + reset_nested_build_level_cursor(); + return _element_reader->load_nested_levels_batch(rows); +} + Status ListColumnReader::build_nested_column(int64_t length_upper_bound, MutableColumnPtr& column, int64_t* values_read) { - if (column.get() == nullptr || values_read == nullptr) { + if (column.get() == nullptr) { return Status::InvalidArgument("Invalid parquet list build result pointer for column {}", _name); } + return _consume_or_build_nested_column(length_upper_bound, &column, values_read); +} + +Status ListColumnReader::consume_nested_column(int64_t length_upper_bound, + int64_t* values_consumed) { + return _consume_or_build_nested_column(length_upper_bound, nullptr, values_consumed); +} + +Status ListColumnReader::_consume_or_build_nested_column(int64_t length_upper_bound, + MutableColumnPtr* column, + int64_t* values_processed) { + if (values_processed == nullptr) { + return Status::InvalidArgument("Invalid parquet list process result pointer for column {}", + _name); + } DORIS_CHECK(_element_reader != nullptr); - auto* array_column = array_column_from_output(column); - DORIS_CHECK(array_column != nullptr); - auto* parent_null_map = null_map_from_nullable_output(column); - auto nested_column = array_column->get_data_ptr()->assert_mutable(); - const auto& element_output_type = - assert_cast(*remove_nullable(_type)).get_nested_type(); - remove_nullable_wrapper_if_not_expected(element_output_type, &nested_column); + ColumnArray* array_column = nullptr; + NullMap* parent_null_map = nullptr; + MutableColumnPtr nested_column; + if (column != nullptr) { + array_column = array_column_from_output(*column); + DORIS_CHECK(array_column != nullptr); + parent_null_map = null_map_from_nullable_output(*column); + nested_column = array_column->get_data_ptr()->assert_mutable(); + const auto& element_output_type = + assert_cast(*remove_nullable(_type)).get_nested_type(); + remove_nullable_wrapper_if_not_expected(element_output_type, &nested_column); + } const auto& def_levels = _element_reader->nested_definition_levels(); const auto& rep_levels = _element_reader->nested_repetition_levels(); const int64_t levels_written = _element_reader->nested_levels_written(); std::vector entry_counts; NullMap parent_nulls; - *values_read = 0; + *values_processed = 0; int64_t level_idx = nested_build_level_cursor(); const int16_t min_parent_definition_level = static_cast(_definition_level - 1 - (_type->is_nullable() ? 1 : 0)); @@ -96,7 +110,7 @@ Status ListColumnReader::build_nested_column(int64_t length_upper_bound, Mutable const int16_t def_level = def_levels[level_idx]; const int16_t rep_level = rep_levels[level_idx]; const bool starts_parent = rep_level < _repetition_level; - if (starts_parent && *values_read >= length_upper_bound) { + if (starts_parent && *values_processed >= length_upper_bound) { break; } ++level_idx; @@ -116,13 +130,13 @@ Status ListColumnReader::build_nested_column(int64_t length_upper_bound, Mutable } const bool parent_is_null = def_level < _definition_level - 1; - if (parent_is_null && parent_null_map == nullptr) { + if (parent_is_null && !_type->is_nullable()) { return Status::Corruption("Parquet LIST column {} contains null for non-nullable LIST", _name); } parent_nulls.push_back(parent_is_null); entry_counts.push_back(def_level >= _definition_level ? 1 : 0); - ++*values_read; + ++*values_processed; } set_nested_build_level_cursor(level_idx); @@ -132,8 +146,13 @@ Status ListColumnReader::build_nested_column(int64_t length_upper_bound, Mutable for (const auto entry_count : entry_counts) { total_entries += entry_count; } - RETURN_IF_ERROR(_element_reader->build_nested_column(static_cast(total_entries), - nested_column, &child_value_count)); + if (column != nullptr) { + RETURN_IF_ERROR(_element_reader->build_nested_column( + static_cast(total_entries), nested_column, &child_value_count)); + } else { + RETURN_IF_ERROR(_element_reader->consume_nested_column( + static_cast(total_entries), &child_value_count)); + } } else { uint64_t pending_entries = 0; auto flush_pending_entries = [&]() -> Status { @@ -141,8 +160,14 @@ Status ListColumnReader::build_nested_column(int64_t length_upper_bound, Mutable return Status::OK(); } int64_t span_child_value_count = 0; - RETURN_IF_ERROR(_element_reader->build_nested_column( - static_cast(pending_entries), nested_column, &span_child_value_count)); + if (column != nullptr) { + RETURN_IF_ERROR(_element_reader->build_nested_column( + static_cast(pending_entries), nested_column, + &span_child_value_count)); + } else { + RETURN_IF_ERROR(_element_reader->consume_nested_column( + static_cast(pending_entries), &span_child_value_count)); + } if (span_child_value_count != static_cast(pending_entries)) { return Status::Corruption( "Parquet LIST column {} built {} child values, expected {}", _name, @@ -168,9 +193,11 @@ Status ListColumnReader::build_nested_column(int64_t length_upper_bound, Mutable return Status::Corruption("Parquet LIST column {} built {} child values, expected {}", _name, child_value_count, total_entries); } - array_column->get_data_ptr() = std::move(nested_column); - append_offsets(array_column->get_offsets(), entry_counts); - append_parent_nulls(parent_null_map, parent_nulls); + if (column != nullptr) { + array_column->get_data_ptr() = std::move(nested_column); + append_offsets(array_column->get_offsets(), entry_counts); + append_parent_nulls(parent_null_map, parent_nulls); + } return Status::OK(); } diff --git a/be/src/format_v2/parquet/reader/list_column_reader.h b/be/src/format_v2/parquet/reader/list_column_reader.h index 5a60eecacb0e3e..d64be546e394d7 100644 --- a/be/src/format_v2/parquet/reader/list_column_reader.h +++ b/be/src/format_v2/parquet/reader/list_column_reader.h @@ -36,8 +36,10 @@ class ListColumnReader final : public ParquetColumnReader { Status read(int64_t rows, MutableColumnPtr& column, int64_t* rows_read) override; Status skip(int64_t rows) override; Status load_nested_batch(int64_t rows) override; + Status load_nested_levels_batch(int64_t rows) override; Status build_nested_column(int64_t length_upper_bound, MutableColumnPtr& column, int64_t* values_read) override; + Status consume_nested_column(int64_t length_upper_bound, int64_t* values_consumed) override; const std::vector& nested_definition_levels() const override; const std::vector& nested_repetition_levels() const override; int64_t nested_levels_written() const override; @@ -45,6 +47,9 @@ class ListColumnReader final : public ParquetColumnReader { void advance_nested_build_level_cursor_past_parent(int16_t parent_repetition_level) override; private: + Status _consume_or_build_nested_column(int64_t length_upper_bound, MutableColumnPtr* column, + int64_t* values_processed); + std::unique_ptr _element_reader; // element reader (recursive; may be Scalar/Struct/List/Map) }; diff --git a/be/src/format_v2/parquet/reader/map_column_reader.cpp b/be/src/format_v2/parquet/reader/map_column_reader.cpp index 90d4a867331190..8217d0c013abc0 100644 --- a/be/src/format_v2/parquet/reader/map_column_reader.cpp +++ b/be/src/format_v2/parquet/reader/map_column_reader.cpp @@ -49,19 +49,7 @@ Status MapColumnReader::read(int64_t rows, MutableColumnPtr& column, int64_t* ro } Status MapColumnReader::skip(int64_t rows) { - if (rows <= 0) { - return Status::OK(); - } - auto scratch_column = _type->create_column(); - RETURN_IF_ERROR(load_nested_batch(rows)); - int64_t rows_read = 0; - RETURN_IF_ERROR(build_nested_column(rows, scratch_column, &rows_read)); - if (rows_read != rows) { - return Status::Corruption("Failed to skip parquet MAP column {}: skipped {} of {} rows", - _name, rows_read, rows); - } - update_reader_skip_rows(rows); - return Status::OK(); + return skip_nested_rows(rows); } Status MapColumnReader::load_nested_batch(int64_t rows) { @@ -72,22 +60,51 @@ Status MapColumnReader::load_nested_batch(int64_t rows) { return _value_reader->load_nested_batch(rows); } +Status MapColumnReader::load_nested_levels_batch(int64_t rows) { + DORIS_CHECK(_key_reader != nullptr); + DORIS_CHECK(_value_reader != nullptr); + reset_nested_build_level_cursor(); + RETURN_IF_ERROR(_key_reader->load_nested_levels_batch(rows)); + return _value_reader->load_nested_levels_batch(rows); +} + Status MapColumnReader::build_nested_column(int64_t length_upper_bound, MutableColumnPtr& column, int64_t* values_read) { - if (column.get() == nullptr || values_read == nullptr) { + if (column.get() == nullptr) { return Status::InvalidArgument("Invalid parquet map build result pointer for column {}", _name); } + return _consume_or_build_nested_column(length_upper_bound, &column, values_read); +} + +Status MapColumnReader::consume_nested_column(int64_t length_upper_bound, + int64_t* values_consumed) { + return _consume_or_build_nested_column(length_upper_bound, nullptr, values_consumed); +} + +Status MapColumnReader::_consume_or_build_nested_column(int64_t length_upper_bound, + MutableColumnPtr* column, + int64_t* values_processed) { + if (values_processed == nullptr) { + return Status::InvalidArgument("Invalid parquet map process result pointer for column {}", + _name); + } DORIS_CHECK(_key_reader != nullptr); DORIS_CHECK(_value_reader != nullptr); - auto* map_column = map_column_from_output(column); - DORIS_CHECK(map_column != nullptr); - auto* parent_null_map = null_map_from_nullable_output(column); - auto key_column = map_column->get_keys_ptr()->assert_mutable(); - auto value_column = map_column->get_values_ptr()->assert_mutable(); - const auto& map_output_type = assert_cast(*remove_nullable(_type)); - remove_nullable_wrapper_if_not_expected(map_output_type.get_key_type(), &key_column); - remove_nullable_wrapper_if_not_expected(map_output_type.get_value_type(), &value_column); + ColumnMap* map_column = nullptr; + NullMap* parent_null_map = nullptr; + MutableColumnPtr key_column; + MutableColumnPtr value_column; + if (column != nullptr) { + map_column = map_column_from_output(*column); + DORIS_CHECK(map_column != nullptr); + parent_null_map = null_map_from_nullable_output(*column); + key_column = map_column->get_keys_ptr()->assert_mutable(); + value_column = map_column->get_values_ptr()->assert_mutable(); + const auto& map_output_type = assert_cast(*remove_nullable(_type)); + remove_nullable_wrapper_if_not_expected(map_output_type.get_key_type(), &key_column); + remove_nullable_wrapper_if_not_expected(map_output_type.get_value_type(), &value_column); + } const auto& def_levels = _key_reader->nested_definition_levels(); const auto& rep_levels = _key_reader->nested_repetition_levels(); @@ -96,7 +113,7 @@ Status MapColumnReader::build_nested_column(int64_t length_upper_bound, MutableC std::vector entry_counts; std::vector map_level_indices; NullMap parent_nulls; - *values_read = 0; + *values_processed = 0; int64_t level_idx = nested_build_level_cursor(); const int16_t min_parent_definition_level = static_cast(_definition_level - 1 - (_type->is_nullable() ? 1 : 0)); @@ -104,7 +121,7 @@ Status MapColumnReader::build_nested_column(int64_t length_upper_bound, MutableC const int16_t def_level = def_levels[level_idx]; const int16_t rep_level = rep_levels[level_idx]; const bool starts_parent = rep_level < _repetition_level; - if (starts_parent && *values_read >= length_upper_bound) { + if (starts_parent && *values_processed >= length_upper_bound) { break; } const int64_t current_level_idx = level_idx; @@ -126,13 +143,13 @@ Status MapColumnReader::build_nested_column(int64_t length_upper_bound, MutableC } const bool parent_is_null = def_level < _definition_level - 1; - if (parent_is_null && parent_null_map == nullptr) { + if (parent_is_null && !_type->is_nullable()) { return Status::Corruption("Parquet MAP column {} contains null for non-nullable MAP", _name); } parent_nulls.push_back(parent_is_null); entry_counts.push_back(def_level >= _definition_level ? 1 : 0); - ++*values_read; + ++*values_processed; } set_nested_build_level_cursor(level_idx); @@ -140,18 +157,36 @@ Status MapColumnReader::build_nested_column(int64_t length_upper_bound, MutableC for (const auto entry_count : entry_counts) { total_entries += entry_count; } - const size_t key_start = key_column->size(); int64_t key_value_count = 0; - RETURN_IF_ERROR(_key_reader->build_nested_column(static_cast(total_entries), - key_column, &key_value_count)); + size_t key_start = 0; + if (column != nullptr) { + key_start = key_column->size(); + RETURN_IF_ERROR(_key_reader->build_nested_column(static_cast(total_entries), + key_column, &key_value_count)); + } else if (auto* scalar_key_reader = dynamic_cast(_key_reader.get())) { + // MAP keys are required even if a projected Doris key type is nullable. Validate each + // actual entry directly from the key level stream while advancing past empty/null maps. + for (const int64_t key_level_idx : map_level_indices) { + if (def_levels[key_level_idx] >= _definition_level) { + RETURN_IF_ERROR(scalar_key_reader->validate_nested_value(key_level_idx, true)); + ++key_value_count; + } + } + scalar_key_reader->set_nested_build_level_cursor(level_idx); + } else { + RETURN_IF_ERROR(_key_reader->consume_nested_column(static_cast(total_entries), + &key_value_count)); + } if (key_value_count != static_cast(total_entries)) { return Status::Corruption("Parquet MAP column {} built {} keys, expected {}", _name, key_value_count, total_entries); } - if (const auto* nullable_key_column = check_and_get_column(*key_column); - nullable_key_column != nullptr && - nullable_key_column->has_null(key_start, nullable_key_column->size())) { - return Status::Corruption("Parquet MAP column {} contains null key", _name); + if (column != nullptr) { + if (const auto* nullable_key_column = check_and_get_column(*key_column); + nullable_key_column != nullptr && + nullable_key_column->has_null(key_start, nullable_key_column->size())) { + return Status::Corruption("Parquet MAP column {} contains null key", _name); + } } int64_t value_count = 0; if (auto* scalar_value_reader = dynamic_cast(_value_reader.get())) { @@ -182,28 +217,40 @@ Status MapColumnReader::build_nested_column(int64_t length_upper_bound, MutableC _name); } if (def_levels[key_level_idx] >= _definition_level) { - RETURN_IF_ERROR( - scalar_value_reader->append_nested_value(value_level_idx, value_column)); + if (column != nullptr) { + RETURN_IF_ERROR(scalar_value_reader->append_nested_value(value_level_idx, + value_column)); + } else { + RETURN_IF_ERROR( + scalar_value_reader->validate_nested_value(value_level_idx, false)); + } ++value_count; } ++value_level_idx; } scalar_value_reader->set_nested_build_level_cursor(value_level_idx); } else { - // Complex MAP values own their nested shape below the entry slot, so they can recursively - // materialize exactly one child value for each MAP entry. - RETURN_IF_ERROR(_value_reader->build_nested_column(static_cast(total_entries), - value_column, &value_count)); + // Complex MAP values own their nested shape below the entry slot, so they recursively + // process exactly one child value for each MAP entry. + if (column != nullptr) { + RETURN_IF_ERROR(_value_reader->build_nested_column(static_cast(total_entries), + value_column, &value_count)); + } else { + RETURN_IF_ERROR(_value_reader->consume_nested_column( + static_cast(total_entries), &value_count)); + } } if (value_count != static_cast(total_entries)) { return Status::Corruption("Parquet MAP column {} built {} values, expected {}", _name, value_count, total_entries); } - map_column->get_keys_ptr() = std::move(key_column); - map_column->get_values_ptr() = std::move(value_column); - append_offsets(map_column->get_offsets(), entry_counts); - append_parent_nulls(parent_null_map, parent_nulls); + if (column != nullptr) { + map_column->get_keys_ptr() = std::move(key_column); + map_column->get_values_ptr() = std::move(value_column); + append_offsets(map_column->get_offsets(), entry_counts); + append_parent_nulls(parent_null_map, parent_nulls); + } return Status::OK(); } diff --git a/be/src/format_v2/parquet/reader/map_column_reader.h b/be/src/format_v2/parquet/reader/map_column_reader.h index 3e26a7a480a2a5..1a8ca9c70d8c5b 100644 --- a/be/src/format_v2/parquet/reader/map_column_reader.h +++ b/be/src/format_v2/parquet/reader/map_column_reader.h @@ -39,8 +39,10 @@ class MapColumnReader final : public ParquetColumnReader { Status read(int64_t rows, MutableColumnPtr& column, int64_t* rows_read) override; Status skip(int64_t rows) override; Status load_nested_batch(int64_t rows) override; + Status load_nested_levels_batch(int64_t rows) override; Status build_nested_column(int64_t length_upper_bound, MutableColumnPtr& column, int64_t* values_read) override; + Status consume_nested_column(int64_t length_upper_bound, int64_t* values_consumed) override; const std::vector& nested_definition_levels() const override; const std::vector& nested_repetition_levels() const override; int64_t nested_levels_written() const override; @@ -48,6 +50,9 @@ class MapColumnReader final : public ParquetColumnReader { void advance_nested_build_level_cursor_past_parent(int16_t parent_repetition_level) override; private: + Status _consume_or_build_nested_column(int64_t length_upper_bound, MutableColumnPtr* column, + int64_t* values_processed); + std::unique_ptr _key_reader; // key column reader (always read fully) std::unique_ptr _value_reader; // value column reader (can be pruned by projection) diff --git a/be/src/format_v2/parquet/reader/parquet_leaf_reader.cpp b/be/src/format_v2/parquet/reader/parquet_leaf_reader.cpp index cb59571d14123f..fd261ef5219d27 100644 --- a/be/src/format_v2/parquet/reader/parquet_leaf_reader.cpp +++ b/be/src/format_v2/parquet/reader/parquet_leaf_reader.cpp @@ -340,10 +340,22 @@ Status ParquetLeafReader::collect_levels_batch(::parquet::internal::RecordReader } batch->_read_dense_for_nullable = record_reader.read_dense_for_nullable(); - // Deliberately ignore values_written(), values() and BinaryRecordReader::GetBuilderChunks(). - // COUNT(col) only needs top-level shape. Pulling binary chunks transfers Arrow builder - // ownership into Doris arrays and later into ColumnString, which is exactly the OOM-prone - // materialization path for huge MAP/ARRAY/STRUCT string payloads. + // Arrow's RecordReader::Reset() does not reset ByteArray/FLBA builders. GetBuilderChunks() + // (or DictionaryRecordReader::GetResult()) is the documented reset operation and must be + // called before the next ReadRecords(). Otherwise a levels-only skip followed by a normal read + // observes values from both batches; for example, skipping ARRAY ["a", "b"] and then + // reading ["c"] would report one current level but three values. Release the chunks here and + // let the temporary vector destroy them immediately. We deliberately do not inspect or copy + // their payload into a Doris Column, so the levels-only contract still avoids Doris-side value + // materialization. + if (batch->is_binary_value()) { + std::vector> discarded_chunks; + RETURN_IF_ERROR(get_binary_chunks(_name, record_reader, &discarded_chunks)); + } + + // COUNT(col) and nested skip only need top-level shape. Fixed-width values remain owned by the + // RecordReader and are cleared by Reset(); binary values were released above solely to reset + // the Arrow builder. batch->_values_written = 0; return Status::OK(); } diff --git a/be/src/format_v2/parquet/reader/parquet_leaf_reader.h b/be/src/format_v2/parquet/reader/parquet_leaf_reader.h index 7d97d0ad6985f9..b396b35fd1f32c 100644 --- a/be/src/format_v2/parquet/reader/parquet_leaf_reader.h +++ b/be/src/format_v2/parquet/reader/parquet_leaf_reader.h @@ -119,15 +119,16 @@ class ParquetLeafReader { ParquetNestedScalarBatch* batch, int16_t value_slot_repetition_level = std::numeric_limits::max()) const; - // COUNT(col) shape-only read path. It still calls Arrow RecordReader::ReadRecords() - // to advance the Parquet cursor and obtain def/rep levels, but Doris only copies levels: - // - it does not call BinaryRecordReader::GetBuilderChunks() + // COUNT(col) and nested-skip shape-only read path. It still calls Arrow + // RecordReader::ReadRecords() to advance the Parquet cursor and obtain def/rep levels, but + // Doris only copies levels: // - it does not build value_indices or values_column // - it does not enter DataTypeSerde::read_column_from_decoded_values() - // This lets COUNT(col) on MAP/ARRAY/STRUCT evaluate top-level NULL state while avoiding - // materializing representative leaf STRING/BINARY payloads into Doris Columns. Arrow RecordReader - // does not expose a public levels-only API, so ReadRecords may still perform required page decoding; - // this API guarantees that the V2 reader does not take ownership of or copy value payloads. + // - for Binary/FLBA, it releases and immediately discards Arrow builder chunks because that is + // the RecordReader's required reset operation; it never copies them into a Doris Column + // This lets COUNT(col) on MAP/ARRAY/STRUCT evaluate top-level NULL state and lets skip advance + // nested shape without Doris-side STRING/BINARY materialization. Arrow RecordReader does not + // expose a public levels-only API, so ReadRecords may still perform required page decoding. Status read_nested_levels_batch(int64_t batch_rows, ParquetNestedScalarBatch* batch) const; private: @@ -136,8 +137,9 @@ class ParquetLeafReader { Status collect_batch(::parquet::internal::RecordReader& record_reader, ParquetLeafBatch* batch) const; - // Levels-only variant of collect_batch(). It snapshots only def/rep level state and does not take - // binary chunks or expose fixed-width value buffers. Used by the COUNT(col) aggregation fast path. + // Levels-only variant of collect_batch(). It snapshots only def/rep level state and does not + // expose value buffers. Binary chunks are released only to reset Arrow's builder and are + // immediately discarded. Used by COUNT(col) and nested skip. Status collect_levels_batch(::parquet::internal::RecordReader& record_reader, ParquetLeafBatch* batch) const; diff --git a/be/src/format_v2/parquet/reader/scalar_column_reader.cpp b/be/src/format_v2/parquet/reader/scalar_column_reader.cpp index 6e3b1c7f4d5d20..784e4cdc900fb7 100644 --- a/be/src/format_v2/parquet/reader/scalar_column_reader.cpp +++ b/be/src/format_v2/parquet/reader/scalar_column_reader.cpp @@ -449,7 +449,11 @@ Status ScalarColumnReader::build_nested_column(int64_t length_upper_bound, Mutab } DORIS_CHECK(_nested_batch != nullptr); ParquetNestedScalarValueCursor value_cursor(_nested_batch.get()); - const int16_t materialized_slot_definition_level = _nested_batch->value_slot_definition_level; + // The levels-only loader intentionally does not populate value-slot metadata or payload + // buffers. Derive the logical slot threshold from the schema, exactly as load_nested_batch() + // does, so this consumer works for both loaded batch forms. + const int16_t materialized_slot_definition_level = + static_cast(_definition_level - (_type->is_nullable() ? 1 : 0)); *values_read = 0; int64_t level_idx = nested_build_level_cursor(); while (level_idx < _nested_batch->levels_written && *values_read < length_upper_bound) { @@ -476,6 +480,36 @@ Status ScalarColumnReader::build_nested_column(int64_t length_upper_bound, Mutab return Status::OK(); } +Status ScalarColumnReader::consume_nested_column(int64_t length_upper_bound, + int64_t* values_consumed) { + if (values_consumed == nullptr) { + return Status::InvalidArgument("Invalid parquet nested scalar consume result for column {}", + _name); + } + DORIS_CHECK(_nested_batch != nullptr); + // A levels-only batch intentionally has no value-slot metadata. Reconstruct the same logical + // slot threshold used by load_nested_batch(): a nullable leaf owns a slot at one definition + // level below a non-null value, while a required leaf owns a slot only at its full definition + // level. For example, an empty ARRAY boundary must not be consumed as a STRING value. + const int16_t materialized_slot_definition_level = + static_cast(_definition_level - (_type->is_nullable() ? 1 : 0)); + *values_consumed = 0; + int64_t level_idx = nested_build_level_cursor(); + while (level_idx < _nested_batch->levels_written && *values_consumed < length_upper_bound) { + const int64_t current_level_idx = level_idx; + const int16_t def_level = _nested_batch->def_levels[current_level_idx]; + const int16_t rep_level = _nested_batch->rep_levels[current_level_idx]; + ++level_idx; + if (def_level < materialized_slot_definition_level || rep_level > _repetition_level) { + continue; + } + RETURN_IF_ERROR(validate_nested_value(current_level_idx, false)); + ++*values_consumed; + } + set_nested_build_level_cursor(level_idx); + return Status::OK(); +} + Status ScalarColumnReader::append_nested_value(int64_t level_idx, MutableColumnPtr& column) const { if (column.get() == nullptr) { return Status::InvalidArgument("Invalid parquet nested scalar append result for column {}", @@ -497,6 +531,21 @@ Status ScalarColumnReader::append_nested_value(int64_t level_idx, MutableColumnP return Status::OK(); } +Status ScalarColumnReader::validate_nested_value(int64_t level_idx, bool require_non_null) const { + DORIS_CHECK(_nested_batch != nullptr); + DORIS_CHECK(level_idx >= 0); + DORIS_CHECK(level_idx < _nested_batch->levels_written); + const int16_t def_level = _nested_batch->def_levels[level_idx]; + if (def_level == _definition_level) { + return Status::OK(); + } + if (require_non_null || !_type->is_nullable()) { + return Status::Corruption("Parquet scalar column {} contains null for non-nullable field", + _name); + } + return Status::OK(); +} + const std::vector& ScalarColumnReader::nested_definition_levels() const { DORIS_CHECK(_nested_batch != nullptr); return _nested_batch->def_levels; diff --git a/be/src/format_v2/parquet/reader/scalar_column_reader.h b/be/src/format_v2/parquet/reader/scalar_column_reader.h index 99fec69fa3ab8a..5342baa803eca0 100644 --- a/be/src/format_v2/parquet/reader/scalar_column_reader.h +++ b/be/src/format_v2/parquet/reader/scalar_column_reader.h @@ -65,6 +65,7 @@ class ScalarColumnReader final : public ParquetColumnReader { Status load_nested_levels_batch(int64_t rows) override; Status build_nested_column(int64_t length_upper_bound, MutableColumnPtr& column, int64_t* values_read) override; + Status consume_nested_column(int64_t length_upper_bound, int64_t* values_consumed) override; const std::vector& nested_definition_levels() const override; const std::vector& nested_repetition_levels() const override; int64_t nested_levels_written() const override; @@ -72,6 +73,7 @@ class ScalarColumnReader final : public ParquetColumnReader { private: Status append_nested_value(int64_t level_idx, MutableColumnPtr& column) const; + Status validate_nested_value(int64_t level_idx, bool require_non_null) const; Status read_range_with_dictionary_filter(int64_t rows, const IColumn::Filter& dictionary_filter, MutableColumnPtr& column, IColumn::Filter* row_filter, int64_t* rows_read, bool* used_filter); diff --git a/be/src/format_v2/parquet/reader/struct_column_reader.cpp b/be/src/format_v2/parquet/reader/struct_column_reader.cpp index 66e450c567133a..5abe7abe75e9a2 100644 --- a/be/src/format_v2/parquet/reader/struct_column_reader.cpp +++ b/be/src/format_v2/parquet/reader/struct_column_reader.cpp @@ -90,19 +90,7 @@ Status StructColumnReader::read(int64_t rows, MutableColumnPtr& column, int64_t* } Status StructColumnReader::skip(int64_t rows) { - if (rows <= 0) { - return Status::OK(); - } - auto scratch_column = _type->create_column(); - RETURN_IF_ERROR(load_nested_batch(rows)); - int64_t rows_read = 0; - RETURN_IF_ERROR(build_nested_column(rows, scratch_column, &rows_read)); - if (rows_read != rows) { - return Status::Corruption("Failed to skip parquet STRUCT column {}: skipped {} of {} rows", - _name, rows_read, rows); - } - update_reader_skip_rows(rows); - return Status::OK(); + return skip_nested_rows(rows); } Status StructColumnReader::load_nested_batch(int64_t rows) { @@ -114,20 +102,50 @@ Status StructColumnReader::load_nested_batch(int64_t rows) { return Status::OK(); } +Status StructColumnReader::load_nested_levels_batch(int64_t rows) { + reset_nested_build_level_cursor(); + for (auto& child_reader : _children) { + DORIS_CHECK(child_reader != nullptr); + RETURN_IF_ERROR(child_reader->load_nested_levels_batch(rows)); + } + return Status::OK(); +} + Status StructColumnReader::build_nested_column(int64_t length_upper_bound, MutableColumnPtr& column, int64_t* values_read) { - if (column.get() == nullptr || values_read == nullptr) { + if (column.get() == nullptr) { return Status::InvalidArgument("Invalid parquet struct build result pointer for column {}", _name); } + return _consume_or_build_nested_column(length_upper_bound, &column, values_read); +} + +Status StructColumnReader::consume_nested_column(int64_t length_upper_bound, + int64_t* values_consumed) { + return _consume_or_build_nested_column(length_upper_bound, nullptr, values_consumed); +} + +Status StructColumnReader::_consume_or_build_nested_column(int64_t length_upper_bound, + MutableColumnPtr* column, + int64_t* values_processed) { + if (values_processed == nullptr) { + return Status::InvalidArgument( + "Invalid parquet struct process result pointer for column {}", _name); + } if (_children.empty()) { - column->resize(column->size() + static_cast(length_upper_bound)); - *values_read = length_upper_bound; + if (column != nullptr) { + (*column)->resize((*column)->size() + static_cast(length_upper_bound)); + } + *values_processed = length_upper_bound; return Status::OK(); } - auto* struct_column = struct_column_from_output(column); - DORIS_CHECK(struct_column != nullptr); - auto* parent_null_map = null_map_from_nullable_output(column); + ColumnStruct* struct_column = nullptr; + NullMap* parent_null_map = nullptr; + if (column != nullptr) { + struct_column = struct_column_from_output(*column); + DORIS_CHECK(struct_column != nullptr); + parent_null_map = null_map_from_nullable_output(*column); + } auto* shape_reader = shape_source_reader(); DORIS_CHECK(shape_reader != nullptr); const auto& def_levels = shape_reader->nested_definition_levels(); @@ -136,7 +154,7 @@ Status StructColumnReader::build_nested_column(int64_t length_upper_bound, Mutab NullMap parent_nulls; std::vector parent_level_indices; - *values_read = 0; + *values_processed = 0; int64_t level_idx = nested_build_level_cursor(); while (level_idx < levels_written) { const int64_t current_level_idx = level_idx; @@ -144,7 +162,7 @@ Status StructColumnReader::build_nested_column(int64_t length_upper_bound, Mutab const int16_t rep_level = rep_levels[level_idx]; const bool starts_parent = !shape_reader->is_or_has_repeated_child() || rep_level <= _repetition_level; - if (starts_parent && *values_read >= length_upper_bound) { + if (starts_parent && *values_processed >= length_upper_bound) { break; } ++level_idx; @@ -155,24 +173,26 @@ Status StructColumnReader::build_nested_column(int64_t length_upper_bound, Mutab continue; } const bool parent_is_null = def_level < _nullable_definition_level; - if (parent_is_null && parent_null_map == nullptr) { + if (parent_is_null && !_type->is_nullable()) { return Status::Corruption( "Parquet STRUCT column {} contains null for non-nullable struct", _name); } parent_nulls.push_back(parent_is_null); parent_level_indices.push_back(current_level_idx); - ++*values_read; + ++*values_processed; } set_nested_build_level_cursor(level_idx); std::vector child_columns; - child_columns.reserve(struct_column->get_columns().size()); - for (size_t child_idx = 0; child_idx < struct_column->get_columns().size(); ++child_idx) { - child_columns.push_back(struct_column->get_column_ptr(child_idx)->assert_mutable()); + if (column != nullptr) { + child_columns.reserve(struct_column->get_columns().size()); + for (size_t child_idx = 0; child_idx < struct_column->get_columns().size(); ++child_idx) { + child_columns.push_back(struct_column->get_column_ptr(child_idx)->assert_mutable()); + } } for (size_t child_idx = 0; child_idx < _children.size(); ++child_idx) { const int output_idx = _child_output_indices[child_idx]; - if (output_idx < 0) { + if (column != nullptr && output_idx < 0) { continue; } // STRUCT owns row alignment. Child readers consume only present parent rows from their @@ -186,8 +206,13 @@ Status StructColumnReader::build_nested_column(int64_t length_upper_bound, Mutab return Status::OK(); } int64_t child_rows = 0; - RETURN_IF_ERROR(_children[child_idx]->build_nested_column( - pending_present_rows, child_columns[output_idx], &child_rows)); + if (column != nullptr) { + RETURN_IF_ERROR(_children[child_idx]->build_nested_column( + pending_present_rows, child_columns[output_idx], &child_rows)); + } else { + RETURN_IF_ERROR(_children[child_idx]->consume_nested_column(pending_present_rows, + &child_rows)); + } if (child_rows != pending_present_rows) { return Status::Corruption( "Parquet STRUCT child {} built {} rows, expected {} for column {}", @@ -204,22 +229,26 @@ Status StructColumnReader::build_nested_column(int64_t length_upper_bound, Mutab continue; } RETURN_IF_ERROR(flush_present_rows()); - child_columns[output_idx]->insert_default(); + if (column != nullptr) { + child_columns[output_idx]->insert_default(); + } RETURN_IF_ERROR(advance_child_past_null_parent(_children[child_idx].get(), parent_level_indices[parent_idx])); ++total_child_rows; } RETURN_IF_ERROR(flush_present_rows()); - if (total_child_rows != *values_read) { + if (total_child_rows != *values_processed) { return Status::Corruption( "Parquet STRUCT child {} built {} rows, expected {} for column {}", - _children[child_idx]->name(), total_child_rows, *values_read, _name); + _children[child_idx]->name(), total_child_rows, *values_processed, _name); } } - for (size_t child_idx = 0; child_idx < child_columns.size(); ++child_idx) { - struct_column->get_column_ptr(child_idx) = std::move(child_columns[child_idx]); + if (column != nullptr) { + for (size_t child_idx = 0; child_idx < child_columns.size(); ++child_idx) { + struct_column->get_column_ptr(child_idx) = std::move(child_columns[child_idx]); + } + append_parent_nulls(parent_null_map, parent_nulls); } - append_parent_nulls(parent_null_map, parent_nulls); return Status::OK(); } diff --git a/be/src/format_v2/parquet/reader/struct_column_reader.h b/be/src/format_v2/parquet/reader/struct_column_reader.h index 3e88b75cede3d9..3c2d6904cb36f4 100644 --- a/be/src/format_v2/parquet/reader/struct_column_reader.h +++ b/be/src/format_v2/parquet/reader/struct_column_reader.h @@ -41,8 +41,10 @@ class StructColumnReader final : public ParquetColumnReader { Status read(int64_t rows, MutableColumnPtr& column, int64_t* rows_read) override; Status skip(int64_t rows) override; Status load_nested_batch(int64_t rows) override; + Status load_nested_levels_batch(int64_t rows) override; Status build_nested_column(int64_t length_upper_bound, MutableColumnPtr& column, int64_t* values_read) override; + Status consume_nested_column(int64_t length_upper_bound, int64_t* values_consumed) override; const std::vector& nested_definition_levels() const override; const std::vector& nested_repetition_levels() const override; int64_t nested_levels_written() const override; @@ -50,6 +52,8 @@ class StructColumnReader final : public ParquetColumnReader { void advance_nested_build_level_cursor_past_parent(int16_t parent_repetition_level) override; private: + Status _consume_or_build_nested_column(int64_t length_upper_bound, MutableColumnPtr* column, + int64_t* values_processed); ParquetColumnReader* shape_source_reader() const; Status advance_child_past_null_parent(ParquetColumnReader* child_reader, int64_t parent_level_idx) const; diff --git a/be/src/format_v2/table_reader.cpp b/be/src/format_v2/table_reader.cpp index 11e2b30df6de23..468a649f8d6e74 100644 --- a/be/src/format_v2/table_reader.cpp +++ b/be/src/format_v2/table_reader.cpp @@ -750,6 +750,10 @@ std::unique_ptr create_file_description(const TFileRangeDes Status TableReader::prepare_split(const SplitReadOptions& options) { SCOPED_TIMER(_profile.prepare_split_timer); _current_split_pruned = false; + _all_runtime_filters_applied_for_split = options.all_runtime_filters_applied; + if (options.conjuncts.has_value()) { + _conjuncts = *options.conjuncts; + } // Update to current split format to handle ORC/PARQUET files in one table. _format = options.current_split_format; _partition_values = std::move(options.partition_values); @@ -777,8 +781,12 @@ Status TableReader::prepare_split(const SplitReadOptions& options) { _current_task = std::make_unique(); _current_task->data_file = create_file_description(options.current_range); _current_file_description = *_current_task->data_file; - if (_push_down_agg_type == TPushAggOp::type::COUNT && - options.current_range.__isset.table_format_params && + // A table-level row count is only equivalent to scanning the split when no row predicate is + // 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 && + _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); _remaining_table_level_count = diff --git a/be/src/format_v2/table_reader.h b/be/src/format_v2/table_reader.h index da2cb351ff68ce..c27f64f7f54a80 100644 --- a/be/src/format_v2/table_reader.h +++ b/be/src/format_v2/table_reader.h @@ -142,9 +142,18 @@ struct TableReadOptions { struct SplitReadOptions { // Split-level information for reader initialization, which may include file path, partition values, delete file info, etc. The content is table format specific and opaque to table reader base class; it's the responsibility of the concrete table reader implementation to parse necessary information for reader initialization and filter pushdown. std::map partition_values; - // Latest scanner conjuncts rewritten to table/global column indices. Runtime filters can - // arrive after TableReader::init(), so split preparation must receive a fresh snapshot. + // 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; + // 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; + // Table-level COUNT may emit one metadata-derived batch and resume on a later scheduler turn. + // It is safe only after every runtime filter assigned to the scanner has arrived; otherwise a + // filter could arrive after synthetic rows have already been returned and those rows cannot be + // retracted. Standalone TableReader callers have no scanner runtime-filter lifecycle. + bool all_runtime_filters_applied = true; ShardedKVCache* cache; TFileRangeDesc current_range; FileFormat current_split_format = FileFormat::PARQUET; @@ -900,6 +909,14 @@ class TableReader { if (agg_type != TPushAggOp::type::COUNT && agg_type != TPushAggOp::type::MINMAX) { return false; } + // Aggregate pushdown returns reduced synthetic rows and may close the physical reader + // before the next scheduler turn. If a runtime filter is still pending, those rows could + // escape before the filter arrives and cannot later be reconstructed from real file rows. + // This is the same irreversibility constraint as table-level metadata COUNT, and applies + // to COUNT and MIN/MAX for Parquet/ORC as well as COUNT for text readers. + if (!_all_runtime_filters_applied_for_split) { + return false; + } // Only support aggregate pushdown when there is no delete or filter, so // the reduced rows consumed by the upper aggregate remain semantically equivalent to a // normal scan. @@ -1536,6 +1553,9 @@ class TableReader { int64_t _condition_cache_hit_count = 0; bool _current_reader_reached_eof = false; int64_t _remaining_table_level_count = -1; + // Snapshot supplied by FileScannerV2 for the active split. It gates every shortcut that emits + // irreversible aggregate rows, not only the table-level row-count shortcut in prepare_split(). + bool _all_runtime_filters_applied_for_split = true; std::optional _global_rowid_context; bool _aggregate_pushdown_tried = false; bool _current_split_pruned = false; diff --git a/be/test/format_v2/parquet/parquet_column_reader_test.cpp b/be/test/format_v2/parquet/parquet_column_reader_test.cpp index 91382203c5cea9..fb4cd129e1b03d 100644 --- a/be/test/format_v2/parquet/parquet_column_reader_test.cpp +++ b/be/test/format_v2/parquet/parquet_column_reader_test.cpp @@ -121,13 +121,8 @@ class NestedSkipReader final : public ParquetColumnReader { Status read(int64_t, MutableColumnPtr&, int64_t*) override { return Status::OK(); } - Status build_nested_column(int64_t length_upper_bound, MutableColumnPtr& column, - int64_t* values_read) override { - auto& values = assert_cast(*column); - for (int64_t row = 0; row < length_upper_bound; ++row) { - values.insert_value(static_cast(row)); - } - *values_read = length_upper_bound; + Status consume_nested_column(int64_t length_upper_bound, int64_t* values_consumed) override { + *values_consumed = length_upper_bound; return Status::OK(); } }; @@ -139,6 +134,7 @@ class ParquetColumnReaderTest : public testing::Test { std::filesystem::remove_all(_test_dir); std::filesystem::create_directories(_test_dir); _file_path = (_test_dir / "reader.parquet").string(); + _plain_file_path = (_test_dir / "plain_reader.parquet").string(); write_parquet_file(); _file_reader = ::parquet::ParquetFileReader::OpenFile(_file_path, false); auto metadata = _file_reader->metadata(); @@ -1824,6 +1820,38 @@ class ParquetColumnReaderTest : public testing::Test { return reader; } + std::unique_ptr create_plain_reader(size_t field_idx) { + // Keep the normal fixture dictionary encoded. This one test writes a plain-encoded copy + // because Arrow BinaryRecordReader has a stricter reset contract than + // DictionaryRecordReader. + auto schema = arrow::schema(_arrow_fields); + auto table = arrow::Table::Make(schema, _arrays); + auto plain_file_result = arrow::io::FileOutputStream::Open(_plain_file_path); + DORIS_CHECK(plain_file_result.ok()); + std::shared_ptr plain_out = *plain_file_result; + ::parquet::WriterProperties::Builder plain_builder; + plain_builder.version(::parquet::ParquetVersion::PARQUET_2_6); + plain_builder.data_page_version(::parquet::ParquetDataPageVersion::V2); + plain_builder.compression(::parquet::Compression::UNCOMPRESSED); + plain_builder.disable_dictionary(); + PARQUET_THROW_NOT_OK(::parquet::arrow::WriteTable( + *table, arrow::default_memory_pool(), plain_out, ROW_COUNT, plain_builder.build())); + DORIS_CHECK(plain_out->Close().ok()); + + _plain_file_reader = ::parquet::ParquetFileReader::OpenFile(_plain_file_path, false); + auto metadata = _plain_file_reader->metadata(); + DORIS_CHECK(metadata != nullptr); + DORIS_CHECK(metadata->num_row_groups() == 1); + _plain_row_group = _plain_file_reader->RowGroup(0); + DORIS_CHECK(_plain_row_group != nullptr); + + ParquetColumnReaderFactory factory(_plain_row_group, metadata->num_columns()); + std::unique_ptr reader; + auto st = factory.create(*_fields[field_idx], &reader); + EXPECT_TRUE(st.ok()) << st; + return reader; + } + std::unique_ptr create_projected_child_reader(size_t field_idx, size_t child_idx) const { const auto& struct_schema = *_fields[field_idx]; @@ -1892,8 +1920,11 @@ class ParquetColumnReaderTest : public testing::Test { std::filesystem::path _test_dir; std::string _file_path; + std::string _plain_file_path; std::unique_ptr<::parquet::ParquetFileReader> _file_reader; + std::unique_ptr<::parquet::ParquetFileReader> _plain_file_reader; std::shared_ptr<::parquet::RowGroupReader> _row_group; + std::shared_ptr<::parquet::RowGroupReader> _plain_row_group; std::vector> _fields; std::vector> _arrow_fields; std::vector> _arrays; @@ -1929,7 +1960,7 @@ TEST(ParquetColumnReaderBaseTest, SelectionVectorRangesAndValidation) { EXPECT_FALSE(identity.verify(1, -1).ok()); } -TEST(ParquetColumnReaderBaseTest, DefaultSelectUsesSkipReadRangesAndSkipNestedUsesBuild) { +TEST(ParquetColumnReaderBaseTest, DefaultSelectUsesSkipReadRangesAndNestedConsumeIsExplicit) { DefaultSelectReader reader; std::array selected = {1, 3, 4}; SelectionVector selection(selected.data(), selected.size()); @@ -1953,10 +1984,12 @@ TEST(ParquetColumnReaderBaseTest, DefaultSelectUsesSkipReadRangesAndSkipNestedUs EXPECT_FALSE(unsupported_reader.load_nested_batch(1).ok()); int64_t values_read = 0; EXPECT_FALSE(unsupported_reader.build_nested_column(1, mutable_column, &values_read).ok()); + EXPECT_FALSE(unsupported_reader.consume_nested_column(1, &values_read).ok()); NestedSkipReader nested_reader; - auto nested_status = nested_reader.skip_nested_column(3); + auto nested_status = nested_reader.consume_nested_column(3, &values_read); ASSERT_TRUE(nested_status.ok()) << nested_status; + EXPECT_EQ(values_read, 3); } TEST_F(ParquetColumnReaderTest, ScalarReadCoversRequiredNullableAllNullAndMultipleBatches) { @@ -3195,6 +3228,35 @@ TEST_F(ParquetColumnReaderTest, SkipMapWithOverflowThenRead) { EXPECT_EQ(offsets[2], 1); } +TEST_F(ParquetColumnReaderTest, SkipPlainBinaryMapThenReadResetsArrowBuilder) { + const auto field_idx = find_field_idx("nullable_map_int_string_col"); + auto reader = create_plain_reader(field_idx); + + // Row 0 contains two STRING values. The levels-only skip must release (and discard) those + // Arrow BinaryRecordReader builder chunks before the next normal read. If they leak into the + // next batch, ParquetLeafReader observes more values than current definition/repetition levels. + auto st = reader->skip(1); + ASSERT_TRUE(st.ok()) << st; + + MutableColumnPtr column = reader->type()->create_column(); + int64_t rows_read = 0; + st = reader->read(3, column, &rows_read); + ASSERT_TRUE(st.ok()) << st; + ASSERT_EQ(rows_read, 3); + + const auto& nullable_column = assert_cast(*column); + ASSERT_EQ(nullable_column.size(), 3); + EXPECT_TRUE(nullable_column.is_null_at(0)); + const auto& map_column = assert_cast(nullable_column.get_nested_column()); + ASSERT_EQ(map_column.get_offsets().size(), 3); + EXPECT_EQ(map_column.get_offsets()[0], 0); + EXPECT_EQ(map_column.get_offsets()[1], 0); + EXPECT_EQ(map_column.get_offsets()[2], 1); + const auto& values = get_nullable_nested_column(map_column.get_values()); + ASSERT_EQ(values.size(), 1); + EXPECT_EQ(values.get_data_at(0).to_string(), "cc"); +} + TEST_F(ParquetColumnReaderTest, SelectMapWithOverflow) { const auto field_idx = find_field_idx("nullable_map_int_string_col"); auto reader = create_reader(field_idx); 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 c7d430350d1b26..a21974bced294d 100644 --- a/be/test/format_v2/parquet/parquet_reader_control_test.cpp +++ b/be/test/format_v2/parquet/parquet_reader_control_test.cpp @@ -17,6 +17,7 @@ #include +#include #include #include #include @@ -218,37 +219,6 @@ class CursorColumnReader final : public ParquetColumnReader { std::vector _read_lengths; }; -class NestedBuildReader final : public ParquetColumnReader { -public: - explicit NestedBuildReader(int64_t values_to_build) - : ParquetColumnReader(int64_schema("nested"), std::make_shared()), - _values_to_build(values_to_build) {} - - Status read(int64_t, MutableColumnPtr&, int64_t*) override { - return Status::NotSupported("unused"); - } - - Status build_nested_column(int64_t length_upper_bound, MutableColumnPtr& column, - int64_t* values_read) override { - if (column.get() == nullptr || values_read == nullptr) { - return Status::InvalidArgument("invalid mock nested build arguments"); - } - _last_length_upper_bound = length_upper_bound; - auto* values = assert_cast(column.get()); - for (int64_t value = 0; value < _values_to_build; ++value) { - values->insert_value(value); - } - *values_read = _values_to_build; - return Status::OK(); - } - - int64_t last_length_upper_bound() const { return _last_length_upper_bound; } - -private: - int64_t _values_to_build = 0; - int64_t _last_length_upper_bound = 0; -}; - class ScriptedNestedReader final : public ParquetColumnReader { public: ScriptedNestedReader(ParquetColumnSchema schema, DataTypePtr type, @@ -269,6 +239,11 @@ class ScriptedNestedReader final : public ParquetColumnReader { return Status::OK(); } + Status load_nested_levels_batch(int64_t rows) override { + _level_load_lengths.push_back(rows); + return Status::OK(); + } + Status build_nested_column(int64_t length_upper_bound, MutableColumnPtr& column, int64_t* values_read) override { _build_lengths.push_back(length_upper_bound); @@ -282,6 +257,15 @@ class ScriptedNestedReader final : public ParquetColumnReader { return Status::OK(); } + Status consume_nested_column(int64_t length_upper_bound, int64_t* values_consumed) override { + DORIS_CHECK(values_consumed != nullptr); + _consume_lengths.push_back(length_upper_bound); + set_nested_build_level_cursor(std::min(nested_build_level_cursor() + length_upper_bound, + static_cast(_def_levels.size()))); + *values_consumed = length_upper_bound; + return Status::OK(); + } + const std::vector& nested_definition_levels() const override { return _def_levels; } const std::vector& nested_repetition_levels() const override { return _rep_levels; } int64_t nested_levels_written() const override { @@ -290,6 +274,8 @@ class ScriptedNestedReader final : public ParquetColumnReader { bool is_or_has_repeated_child() const override { return _has_repeated_child; } const std::vector& build_lengths() const { return _build_lengths; } + const std::vector& consume_lengths() const { return _consume_lengths; } + const std::vector& level_load_lengths() const { return _level_load_lengths; } private: static void insert_value(MutableColumnPtr& column, int64_t value, bool is_null) { @@ -312,7 +298,85 @@ class ScriptedNestedReader final : public ParquetColumnReader { bool _build_nulls = false; int64_t _next_value = 0; std::vector _load_lengths; + std::vector _level_load_lengths; std::vector _build_lengths; + std::vector _consume_lengths; +}; + +class ChunkedNestedLeafReader final : public ParquetColumnReader { +public: + ChunkedNestedLeafReader() + : ParquetColumnReader(nested_int64_schema("element", 0, 1, 1, 1), + std::make_shared()) {} + + Status read(int64_t, MutableColumnPtr&, int64_t*) override { + return Status::NotSupported("unused"); + } + + Status load_nested_batch(int64_t rows) override { + _load_lengths.push_back(rows); + _def_levels.assign(static_cast(rows), 1); + _rep_levels.assign(static_cast(rows), 0); + return Status::OK(); + } + + Status load_nested_levels_batch(int64_t rows) override { + _level_load_lengths.push_back(rows); + _def_levels.assign(static_cast(rows), 1); + _rep_levels.assign(static_cast(rows), 0); + return Status::OK(); + } + + Status build_nested_column(int64_t length_upper_bound, MutableColumnPtr& column, + int64_t* values_read) override { + DORIS_CHECK(column.get() != nullptr); + DORIS_CHECK(values_read != nullptr); + _initial_column_sizes.push_back(column->size()); + _build_lengths.push_back(length_upper_bound); + if (auto* nullable = check_and_get_column(*column); nullable != nullptr) { + auto& values = assert_cast(nullable->get_nested_column()); + for (int64_t row = 0; row < length_upper_bound; ++row) { + values.insert_value(row); + nullable->get_null_map_data().push_back(0); + } + } else { + auto* values = assert_cast(column.get()); + for (int64_t row = 0; row < length_upper_bound; ++row) { + values->insert_value(row); + } + } + *values_read = length_upper_bound; + return Status::OK(); + } + + Status consume_nested_column(int64_t length_upper_bound, int64_t* values_consumed) override { + DORIS_CHECK(values_consumed != nullptr); + _consume_lengths.push_back(length_upper_bound); + *values_consumed = length_upper_bound; + return Status::OK(); + } + + const std::vector& nested_definition_levels() const override { return _def_levels; } + const std::vector& nested_repetition_levels() const override { return _rep_levels; } + int64_t nested_levels_written() const override { + return static_cast(_def_levels.size()); + } + bool is_or_has_repeated_child() const override { return true; } + + const std::vector& load_lengths() const { return _load_lengths; } + const std::vector& build_lengths() const { return _build_lengths; } + const std::vector& consume_lengths() const { return _consume_lengths; } + const std::vector& level_load_lengths() const { return _level_load_lengths; } + const std::vector& initial_column_sizes() const { return _initial_column_sizes; } + +private: + std::vector _def_levels; + std::vector _rep_levels; + std::vector _load_lengths; + std::vector _level_load_lengths; + std::vector _build_lengths; + std::vector _consume_lengths; + std::vector _initial_column_sizes; }; } // namespace @@ -456,12 +520,88 @@ TEST(ParquetColumnReaderControlTest, BaseNestedDefaultsAndSkipNested) { int64_t values_read = 0; EXPECT_FALSE(base_reader.build_nested_column(1, column, &values_read).ok()); - NestedBuildReader ok_reader(3); - ASSERT_TRUE(ok_reader.skip_nested_column(3).ok()); - EXPECT_EQ(ok_reader.last_length_upper_bound(), 3); + int64_t values_consumed = 0; + EXPECT_FALSE(base_reader.consume_nested_column(1, &values_consumed).ok()); +} + +TEST(ParquetColumnReaderControlTest, NestedSkipConsumesBoundedBatchesWithoutMaterializing) { + auto element_reader = std::make_unique(); + auto* element_reader_ptr = element_reader.get(); + ListColumnReader reader(bare_repeated_int64_list_schema(), + bare_repeated_int64_list_schema().type, std::move(element_reader)); + + ASSERT_TRUE(reader.skip(8193).ok()); + EXPECT_TRUE(element_reader_ptr->load_lengths().empty()); + EXPECT_EQ(element_reader_ptr->level_load_lengths(), std::vector({4096, 4096, 1})); + EXPECT_EQ(element_reader_ptr->consume_lengths(), std::vector({4096, 4096, 1})); + EXPECT_TRUE(element_reader_ptr->build_lengths().empty()); + EXPECT_TRUE(element_reader_ptr->initial_column_sizes().empty()); +} + +TEST(ParquetColumnReaderControlTest, MapSkipConsumesBothStreamsWithoutMaterializing) { + const std::vector def_levels {3, 3, 1, 3}; + const std::vector rep_levels {0, 1, 0, 0}; + auto key_reader = std::make_unique( + nested_int64_schema("key", 2, 3, 1, 2), + make_nullable(std::make_shared()), def_levels, rep_levels); + auto* key_reader_ptr = key_reader.get(); + auto value_reader = std::make_unique( + nested_int64_schema("value", 2, 3, 1, 2), + make_nullable(std::make_shared()), def_levels, rep_levels); + auto* value_reader_ptr = value_reader.get(); + MapColumnReader reader(nested_map_schema(), nested_map_schema().type, std::move(key_reader), + std::move(value_reader)); + + ASSERT_TRUE(reader.skip(3).ok()); + EXPECT_EQ(key_reader_ptr->level_load_lengths(), std::vector({3})); + EXPECT_EQ(value_reader_ptr->level_load_lengths(), std::vector({3})); + EXPECT_EQ(key_reader_ptr->consume_lengths(), std::vector({3})); + EXPECT_EQ(value_reader_ptr->consume_lengths(), std::vector({3})); + EXPECT_TRUE(key_reader_ptr->build_lengths().empty()); + EXPECT_TRUE(value_reader_ptr->build_lengths().empty()); +} + +TEST(ParquetColumnReaderControlTest, StructSkipConsumesNullSeparatedChildSpans) { + const std::vector def_levels {2, 0, 2}; + const std::vector rep_levels {0, 0, 0}; + auto shape_reader = std::make_unique( + nested_int64_schema("shape", 1, 2), make_nullable(std::make_shared()), + def_levels, rep_levels); + auto* shape_reader_ptr = shape_reader.get(); + auto child_reader = std::make_unique( + nested_int64_schema("child", 1, 2), make_nullable(std::make_shared()), + def_levels, rep_levels); + auto* child_reader_ptr = child_reader.get(); + std::vector> children; + children.push_back(std::move(shape_reader)); + children.push_back(std::move(child_reader)); + StructColumnReader reader(nested_struct_schema(), nested_struct_schema().type, + std::move(children), {-1, 0}); + + ASSERT_TRUE(reader.skip(3).ok()); + EXPECT_EQ(shape_reader_ptr->level_load_lengths(), std::vector({3})); + EXPECT_EQ(child_reader_ptr->level_load_lengths(), std::vector({3})); + EXPECT_EQ(shape_reader_ptr->consume_lengths(), std::vector({1, 1})); + EXPECT_EQ(child_reader_ptr->consume_lengths(), std::vector({1, 1})); + EXPECT_TRUE(shape_reader_ptr->build_lengths().empty()); + EXPECT_TRUE(child_reader_ptr->build_lengths().empty()); +} - NestedBuildReader short_reader(2); - EXPECT_FALSE(short_reader.skip_nested_column(3).ok()); +TEST(ParquetColumnReaderControlTest, NestedListSkipConsumesRecursivelyWithoutMaterializing) { + auto leaf_reader = std::make_unique( + nested_int64_schema("leaf", 0, 1, 1, 1), std::make_shared(), + std::vector {1, 1}, std::vector {0, 0}); + auto* leaf_reader_ptr = leaf_reader.get(); + const auto inner_type = std::make_shared(std::make_shared()); + auto inner_reader = std::make_unique(bare_repeated_int64_list_schema(), + inner_type, std::move(leaf_reader)); + const auto outer_type = std::make_shared(inner_type); + ListColumnReader reader(bare_repeated_int64_list_schema(), outer_type, std::move(inner_reader)); + + ASSERT_TRUE(reader.skip(2).ok()); + EXPECT_EQ(leaf_reader_ptr->level_load_lengths(), std::vector({2})); + EXPECT_EQ(leaf_reader_ptr->consume_lengths(), std::vector({2})); + EXPECT_TRUE(leaf_reader_ptr->build_lengths().empty()); } TEST(ParquetColumnReaderControlTest, NestedMaterializerHelpersAppendOffsetsAndParentNulls) { @@ -865,6 +1005,34 @@ TEST(ParquetColumnReaderControlTest, MapRejectsNullKeysAndMisalignedScalarValueR EXPECT_NE(status.to_string().find("value repetition level is not aligned"), std::string::npos); } +TEST(ParquetColumnReaderControlTest, MapConsumePreservesKeyAndValueCorruptionChecks) { + auto null_key_reader = + make_scripted_scalar_reader(nested_int64_schema("key", 2, 3, 1, 2), + scalar_batch({2}, {0}, {-1}, std::vector {})); + auto value_reader = std::make_unique( + nested_int64_schema("value", 2, 3, 1, 2), + make_nullable(std::make_shared()), std::vector {2}, + std::vector {0}); + MapColumnReader null_key_reader_map(nested_map_schema(), nested_map_schema().type, + std::move(null_key_reader), std::move(value_reader)); + int64_t values_consumed = 0; + auto status = null_key_reader_map.consume_nested_column(1, &values_consumed); + EXPECT_FALSE(status.ok()); + EXPECT_NE(status.to_string().find("contains null"), std::string::npos); + + auto key_reader = std::make_unique( + nested_int64_schema("key", 1, 2, 1), make_nullable(std::make_shared()), + std::vector {2, 2}, std::vector {0, 1}); + auto misaligned_value_reader = + make_scripted_scalar_reader(nested_int64_schema("value", 2, 3, 1), + scalar_batch({3, 3}, {0, 0}, {0, 1}, {100, 200})); + MapColumnReader misaligned_reader(nested_map_schema(), nested_map_schema().type, + std::move(key_reader), std::move(misaligned_value_reader)); + status = misaligned_reader.consume_nested_column(1, &values_consumed); + EXPECT_FALSE(status.ok()); + EXPECT_NE(status.to_string().find("value repetition level is not aligned"), std::string::npos); +} + TEST(ParquetColumnReaderControlTest, MapBuildsScalarAndComplexValuePaths) { auto key_reader = std::make_unique( nested_int64_schema("key", 1, 2, 1), make_nullable(std::make_shared()), diff --git a/be/test/format_v2/table_reader_test.cpp b/be/test/format_v2/table_reader_test.cpp index 252bfa5c04d73f..65070cfed784b8 100644 --- a/be/test/format_v2/table_reader_test.cpp +++ b/be/test/format_v2/table_reader_test.cpp @@ -1273,6 +1273,181 @@ TEST(TableReaderTest, CanUseInjectedFileReaderForStandaloneUnitTest) { EXPECT_TRUE(eos); } +TEST(TableReaderTest, PrepareSplitReplacesInitialConjunctSnapshot) { + std::vector file_schema; + file_schema.push_back(make_file_column(0, "id", std::make_shared())); + + std::vector projected_columns; + projected_columns.push_back(make_table_column(0, "id", std::make_shared())); + set_name_identifiers(&projected_columns); + + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + auto fake_state = std::make_shared(); + FakeTableReader reader(file_schema, fake_state); + ASSERT_TRUE(reader.init({ + .projected_columns = projected_columns, + .conjuncts = {VExprContext::create_shared( + table_int32_greater_than_expr(0, 0, 0))}, + .format = FileFormat::PARQUET, + .scan_params = nullptr, + .io_ctx = nullptr, + .runtime_state = &state, + .scanner_profile = nullptr, + }) + .ok()); + + SplitReadOptions split_options; + split_options.current_range.__set_path("fake-table-reader-input"); + split_options.conjuncts = VExprContextSPtrs {VExprContext::create_shared( + runtime_filter_wrapper_expr(table_int32_greater_than_expr(0, 0, 1)))}; + 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()); + ASSERT_NE(fake_state->last_request, nullptr); + ASSERT_EQ(fake_state->last_request->conjuncts.size(), 1); + EXPECT_TRUE(fake_state->last_request->conjuncts.front()->root()->is_rf_wrapper()); + ASSERT_TRUE(reader.close().ok()); +} + +TEST(TableReaderTest, RefreshedConjunctDisablesTableLevelCount) { + std::vector file_schema; + file_schema.push_back(make_file_column(0, "id", std::make_shared())); + + std::vector projected_columns; + projected_columns.push_back(make_table_column(0, "id", std::make_shared())); + set_name_identifiers(&projected_columns); + + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + auto fake_state = std::make_shared(); + 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, + }) + .ok()); + + SplitReadOptions split_options; + split_options.current_range.__set_path("fake-table-reader-input"); + split_options.conjuncts = VExprContextSPtrs {VExprContext::create_shared( + runtime_filter_wrapper_expr(table_int32_greater_than_expr(0, 0, 1)))}; + set_table_level_row_count(&split_options, 5); + 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()); + // The metadata count advertises five rows, while the fake reader contains two. Opening the + // reader and returning its rows proves the fresh runtime filter did not take the synthetic + // table-level COUNT path that would bypass all row predicates. + EXPECT_EQ(fake_state->open_count, 1); + EXPECT_EQ(block.rows(), 2); + ASSERT_TRUE(reader.close().ok()); +} + +TEST(TableReaderTest, PendingRuntimeFilterDisablesTableLevelCount) { + std::vector file_schema; + file_schema.push_back(make_file_column(0, "id", std::make_shared())); + + std::vector projected_columns; + projected_columns.push_back(make_table_column(0, "id", std::make_shared())); + set_name_identifiers(&projected_columns); + + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + auto fake_state = std::make_shared(); + fake_state->aggregate_count = state.batch_size() + 5; + 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, + }) + .ok()); + + SplitReadOptions split_options; + split_options.current_range.__set_path("fake-table-reader-input"); + // A pending runtime filter makes metadata COUNT ineligible before its first synthetic batch. + // This prevents the filter from arriving between scheduler reads after unfiltered rows have + // already escaped. + split_options.all_runtime_filters_applied = false; + set_table_level_row_count(&split_options, state.batch_size() + 5); + 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_EQ(fake_state->open_count, 1); + EXPECT_EQ(block.rows(), 2); + ASSERT_TRUE(reader.close().ok()); +} + +TEST(TableReaderTest, PendingRuntimeFilterDisablesMinMaxPushdown) { + const auto test_dir = + std::filesystem::temp_directory_path() / "doris_table_reader_pending_rf_minmax_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, {3, 1, 5, 2}, {30, 10, 50, 20}, + {"three", "one", "five", "two"}, 2); + + std::vector projected_columns; + projected_columns.push_back(make_table_column(0, "id", std::make_shared())); + projected_columns.push_back(make_table_column(1, "score", std::make_shared())); + set_name_identifiers(&projected_columns); + + 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::MINMAX, + }) + .ok()); + auto split_options = build_split_options(file_path); + split_options.all_runtime_filters_applied = false; + ASSERT_TRUE(reader.prepare_split(split_options).ok()); + + bool eos = false; + size_t total_rows = 0; + bool checked_first_batch = false; + while (!eos) { + Block block = build_table_block(projected_columns); + ASSERT_TRUE(reader.get_block(&block, &eos).ok()); + total_rows += block.rows(); + if (!checked_first_batch && block.rows() > 0) { + const auto& ids = + assert_cast(expect_not_null_table_column(block, 0)); + ASSERT_EQ(ids.size(), 2); + EXPECT_EQ(ids.get_element(0), 3); + EXPECT_EQ(ids.get_element(1), 1); + checked_first_batch = true; + } + } + // MIN/MAX pushdown would return the two synthetic extrema [1, 5]. Reading the original first + // row group [3, 1] and all four rows proves a pending RF kept the physical reader active. + EXPECT_TRUE(checked_first_batch); + EXPECT_EQ(total_rows, 4); + ASSERT_TRUE(reader.close().ok()); + std::filesystem::remove_all(test_dir); +} + TEST(TableReaderTest, AbortSplitClearsReaderAfterIgnorableNotFound) { std::vector file_schema; file_schema.push_back(make_file_column(0, "id", std::make_shared()));