From 5bf0694064519f27d9bba3a3785b4451be3aad47 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Sun, 12 Jul 2026 15:06:40 +0800 Subject: [PATCH 1/8] [fix](be) Refresh V2 file predicates for each split ### What problem does this PR solve? Issue Number: None Related PR: None Problem Summary: FileScannerV2 cloned its table-level conjuncts only when TableReader was initialized. Runtime filters arriving after scanner open were visible to the scanner but never reached subsequent split readers, so Parquet and ORC metadata pruning, native file filtering, and JNI block filtering could not benefit from late runtime filters. Pass a freshly cloned conjunct snapshot into every split, replace TableReader's initial snapshot during split preparation, and prepare JNI conjuncts per split. ### Release note Enable late runtime filters to participate in FileScannerV2 file-level filtering for subsequent splits. ### Check List (For Author) - Test: Unit Test - `TableReaderTest.PrepareSplitReplacesInitialConjunctSnapshot` - `FileScannerV2Test.RewriteSlotRefsToGlobalIndexMatrix` - Behavior changed: Yes. Subsequent FileScannerV2 splits use the latest runtime-filter snapshot for native and JNI filtering. - Does this need documentation: No --- be/src/exec/scan/file_scanner_v2.cpp | 3 ++ be/src/format_v2/jni/jni_table_reader.cpp | 15 +++++---- be/src/format_v2/table_reader.cpp | 3 ++ be/src/format_v2/table_reader.h | 8 +++-- be/test/format_v2/table_reader_test.cpp | 38 +++++++++++++++++++++++ 5 files changed, 57 insertions(+), 10 deletions(-) diff --git a/be/src/exec/scan/file_scanner_v2.cpp b/be/src/exec/scan/file_scanner_v2.cpp index 10b5f850ea36f7..f3231ecd7d420b 100644 --- a/be/src/exec/scan/file_scanner_v2.cpp +++ b/be/src/exec/scan/file_scanner_v2.cpp @@ -492,12 +492,15 @@ 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), .cache = _kv_cache, .current_range = range, 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/table_reader.cpp b/be/src/format_v2/table_reader.cpp index 11e2b30df6de23..1c0fef2e74be06 100644 --- a/be/src/format_v2/table_reader.cpp +++ b/be/src/format_v2/table_reader.cpp @@ -750,6 +750,9 @@ 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; + 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); diff --git a/be/src/format_v2/table_reader.h b/be/src/format_v2/table_reader.h index da2cb351ff68ce..adb75dfe22b314 100644 --- a/be/src/format_v2/table_reader.h +++ b/be/src/format_v2/table_reader.h @@ -142,8 +142,12 @@ 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; ShardedKVCache* cache; TFileRangeDesc current_range; diff --git a/be/test/format_v2/table_reader_test.cpp b/be/test/format_v2/table_reader_test.cpp index 252bfa5c04d73f..96cb8000d5c221 100644 --- a/be/test/format_v2/table_reader_test.cpp +++ b/be/test/format_v2/table_reader_test.cpp @@ -1273,6 +1273,44 @@ 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, AbortSplitClearsReaderAfterIgnorableNotFound) { std::vector file_schema; file_schema.push_back(make_file_column(0, "id", std::make_shared())); From 7f06a4677d9e12cabc91f164e16814d1123fb2bc Mon Sep 17 00:00:00 2001 From: Gabriel Date: Sun, 12 Jul 2026 20:43:28 +0800 Subject: [PATCH 2/8] [fix](be) Bound nested Parquet skip memory ### What problem does this PR solve? Issue Number: None Related PR: None Problem Summary: Complex Parquet column readers materialized every discarded row in a selection gap into one scratch column, so memory and decoding work grew with the entire gap. In addition, a refreshed split predicate could be bypassed by the table-level COUNT shortcut. Materialize skipped nested rows in bounded batches with per-batch scratch columns, and only use metadata row counts when the split has no active conjuncts. ### Release note Bound memory usage while skipping nested Parquet columns and preserve late runtime-filter semantics for table-level COUNT. ### Check List (For Author) - Test: Unit Test - `ParquetColumnReaderControlTest.*` (16 tests) - `TableReaderTest.PrepareSplitReplacesInitialConjunctSnapshot` - `TableReaderTest.RefreshedConjunctDisablesTableLevelCount` - Behavior changed: Yes. Nested Parquet skip work is processed in bounded batches, and table-level COUNT no longer bypasses active split predicates. - Does this need documentation: No --- .../parquet/reader/column_reader.cpp | 28 ++++++++ .../format_v2/parquet/reader/column_reader.h | 4 ++ .../parquet/reader/list_column_reader.cpp | 14 +--- .../parquet/reader/map_column_reader.cpp | 14 +--- .../parquet/reader/struct_column_reader.cpp | 14 +--- be/src/format_v2/table_reader.cpp | 5 +- .../parquet/parquet_reader_control_test.cpp | 72 +++++++++++++++++++ be/test/format_v2/table_reader_test.cpp | 41 +++++++++++ 8 files changed, 152 insertions(+), 40 deletions(-) diff --git a/be/src/format_v2/parquet/reader/column_reader.cpp b/be/src/format_v2/parquet/reader/column_reader.cpp index 1b6e66beefe860..80a95cf17ee171 100644 --- a/be/src/format_v2/parquet/reader/column_reader.cpp +++ b/be/src/format_v2/parquet/reader/column_reader.cpp @@ -617,6 +617,34 @@ Status ParquetColumnReader::skip_nested_column(int64_t rows) { return Status::OK(); } +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 + // scratch column bounds the amplification for large holes while preserving the same nested + // level decoding and validation path as a normal read. Recreate the scratch column per batch + // so its child buffers are released before decoding the next batch. + 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); + auto scratch_column = _type->create_column(); + RETURN_IF_ERROR(load_nested_batch(batch_rows)); + int64_t rows_read = 0; + RETURN_IF_ERROR(build_nested_column(batch_rows, scratch_column, &rows_read)); + if (rows_read != batch_rows) { + return Status::Corruption( + "Failed to skip nested parquet column {}: skipped {} of {} rows in batch", + _name, rows_read, batch_rows); + } + remaining_rows -= batch_rows; + } + update_reader_skip_rows(rows); + return Status::OK(); +} + const std::vector& ParquetColumnReader::nested_definition_levels() const { static const std::vector empty; return empty; diff --git a/be/src/format_v2/parquet/reader/column_reader.h b/be/src/format_v2/parquet/reader/column_reader.h index 1d63e03ca9cf43..7072a8c7f28bad 100644 --- a/be/src/format_v2/parquet/reader/column_reader.h +++ b/be/src/format_v2/parquet/reader/column_reader.h @@ -109,6 +109,10 @@ class ParquetColumnReader { ParquetColumnReader(const ParquetColumnSchema& schema, const DataTypePtr type, ParquetColumnReaderProfile profile = {}); ParquetColumnReader() = default; + // Complex readers cannot advance their child streams without rebuilding parent boundaries + // from definition/repetition levels. Materialize that discarded shape in bounded batches so + // a large selection gap does not allocate a scratch column for the entire gap at once. + 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..3caac5b2ac6d6a 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) { 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..c5a7e0e47f9e13 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) { 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..8d69f75f8d3431 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) { diff --git a/be/src/format_v2/table_reader.cpp b/be/src/format_v2/table_reader.cpp index 1c0fef2e74be06..304b46db6085e0 100644 --- a/be/src/format_v2/table_reader.cpp +++ b/be/src/format_v2/table_reader.cpp @@ -780,7 +780,10 @@ 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 && + // A table-level row count is only equivalent to scanning the split when no row predicate is + // active. In particular, a runtime filter may arrive after init() and replace `_conjuncts` + // above. Returning synthetic rows here would bypass that fresh split snapshot completely. + if (_push_down_agg_type == TPushAggOp::type::COUNT && _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/test/format_v2/parquet/parquet_reader_control_test.cpp b/be/test/format_v2/parquet/parquet_reader_control_test.cpp index c7d430350d1b26..ca19f096924e90 100644 --- a/be/test/format_v2/parquet/parquet_reader_control_test.cpp +++ b/be/test/format_v2/parquet/parquet_reader_control_test.cpp @@ -315,6 +315,64 @@ class ScriptedNestedReader final : public ParquetColumnReader { std::vector _build_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 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(); + } + + 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& initial_column_sizes() const { return _initial_column_sizes; } + +private: + std::vector _def_levels; + std::vector _rep_levels; + std::vector _load_lengths; + std::vector _build_lengths; + std::vector _initial_column_sizes; +}; + } // namespace struct ScalarColumnReaderTestAccess { @@ -464,6 +522,20 @@ TEST(ParquetColumnReaderControlTest, BaseNestedDefaultsAndSkipNested) { EXPECT_FALSE(short_reader.skip_nested_column(3).ok()); } +TEST(ParquetColumnReaderControlTest, NestedSkipMaterializesBoundedBatches) { + 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_EQ(element_reader_ptr->load_lengths(), std::vector({4096, 4096, 1})); + EXPECT_EQ(element_reader_ptr->build_lengths(), std::vector({4096, 4096, 1})); + // Each child build sees an empty destination, proving the parent scratch column from the + // previous batch was destroyed instead of retaining all skipped nested values. + EXPECT_EQ(element_reader_ptr->initial_column_sizes(), std::vector({0, 0, 0})); +} + TEST(ParquetColumnReaderControlTest, NestedMaterializerHelpersAppendOffsetsAndParentNulls) { ColumnArray::Offsets64 offsets; append_offsets(offsets, {3, 0, 2}); diff --git a/be/test/format_v2/table_reader_test.cpp b/be/test/format_v2/table_reader_test.cpp index 96cb8000d5c221..2bca11fc95a9c7 100644 --- a/be/test/format_v2/table_reader_test.cpp +++ b/be/test/format_v2/table_reader_test.cpp @@ -1311,6 +1311,47 @@ TEST(TableReaderTest, PrepareSplitReplacesInitialConjunctSnapshot) { 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, AbortSplitClearsReaderAfterIgnorableNotFound) { std::vector file_schema; file_schema.push_back(make_file_column(0, "id", std::make_shared())); From 1735f1e54e76ae4948721bafc4399be9bd2820c4 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Sun, 12 Jul 2026 21:31:31 +0800 Subject: [PATCH 3/8] [fix](be) Guard metadata count against pending filters ### What problem does this PR solve? Issue Number: None Related PR: #65498 Problem Summary: A table-level metadata COUNT split can emit several synthetic batches across scheduler turns. If a runtime filter was still pending when the shortcut started, it could arrive after an unfiltered batch had already escaped, and the active split could not recover the corresponding real file rows. Pass the scanner runtime-filter completion state into split preparation and allow metadata COUNT only after every assigned runtime filter has arrived. ### Release note Table-level metadata COUNT now falls back to reading file rows while runtime filters are pending. ### Check List (For Author) - Test: Unit Test - TableReaderTest.PrepareSplitReplacesInitialConjunctSnapshot - TableReaderTest.RefreshedConjunctDisablesTableLevelCount - TableReaderTest.PendingRuntimeFilterDisablesTableLevelCount - Remote BE build - Behavior changed: Yes (metadata COUNT waits for all scanner runtime filters before using synthetic rows) - Does this need documentation: No --- be/src/exec/scan/file_scanner_v2.cpp | 3 ++ be/src/format_v2/table_reader.cpp | 9 +++--- be/src/format_v2/table_reader.h | 5 ++++ be/test/format_v2/table_reader_test.cpp | 40 +++++++++++++++++++++++++ 4 files changed, 53 insertions(+), 4 deletions(-) diff --git a/be/src/exec/scan/file_scanner_v2.cpp b/be/src/exec/scan/file_scanner_v2.cpp index f3231ecd7d420b..519237ea85c252 100644 --- a/be/src/exec/scan/file_scanner_v2.cpp +++ b/be/src/exec/scan/file_scanner_v2.cpp @@ -502,6 +502,9 @@ Status FileScannerV2::_prepare_table_reader_split(const TFileRangeDesc& range, .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/table_reader.cpp b/be/src/format_v2/table_reader.cpp index 304b46db6085e0..a7eb5ba35e891d 100644 --- a/be/src/format_v2/table_reader.cpp +++ b/be/src/format_v2/table_reader.cpp @@ -781,10 +781,11 @@ Status TableReader::prepare_split(const SplitReadOptions& options) { _current_task->data_file = create_file_description(options.current_range); _current_file_description = *_current_task->data_file; // A table-level row count is only equivalent to scanning the split when no row predicate is - // active. In particular, a runtime filter may arrive after init() and replace `_conjuncts` - // above. Returning synthetic rows here would bypass that fresh split snapshot completely. - if (_push_down_agg_type == TPushAggOp::type::COUNT && _conjuncts.empty() && - options.current_range.__isset.table_format_params && + // 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 adb75dfe22b314..30d6a9c3d24f4d 100644 --- a/be/src/format_v2/table_reader.h +++ b/be/src/format_v2/table_reader.h @@ -149,6 +149,11 @@ struct SplitReadOptions { // 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; diff --git a/be/test/format_v2/table_reader_test.cpp b/be/test/format_v2/table_reader_test.cpp index 2bca11fc95a9c7..d47da71962f1e8 100644 --- a/be/test/format_v2/table_reader_test.cpp +++ b/be/test/format_v2/table_reader_test.cpp @@ -1352,6 +1352,46 @@ TEST(TableReaderTest, RefreshedConjunctDisablesTableLevelCount) { 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(); + 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, AbortSplitClearsReaderAfterIgnorableNotFound) { std::vector file_schema; file_schema.push_back(make_file_column(0, "id", std::make_shared())); From 4c870ef0bd9700dd24218293254878c38f20f1a2 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Sun, 12 Jul 2026 23:22:24 +0800 Subject: [PATCH 4/8] [improvement](be) Avoid materializing skipped nested values ### What problem does this PR solve? Issue Number: close #65498 Related PR: #65498 Problem Summary: Nested Parquet skips rebuilt parent boundaries into scratch Columns and decoded leaf payloads that were immediately discarded. In addition, pending runtime filters disabled table-level metadata COUNT but could still allow normal COUNT or MIN/MAX aggregate pushdown to emit irreversible synthetic rows before the filter arrived. Add a levels-only nested consume protocol for scalar, list, map, and struct readers, preserving nullability and stream-alignment validation without payload materialization. Gate every aggregate pushdown path on the split runtime-filter snapshot. ### Release note Improve nested Parquet skip efficiency and preserve runtime-filter correctness for aggregate pushdown. ### Check List (For Author) - Test: Regression test / Unit Test - ParquetColumnReaderTest, ParquetColumnReaderBaseTest, ParquetColumnReaderControlTest, and targeted TableReader tests - Remote ./build.sh --be - Behavior changed: Yes (nested skip avoids decoding discarded payloads; aggregate pushdown waits for pending runtime filters) - Does this need documentation: No --- .../parquet/reader/column_reader.cpp | 28 +-- .../format_v2/parquet/reader/column_reader.h | 33 +++- .../parquet/reader/list_column_reader.cpp | 77 ++++++-- .../parquet/reader/list_column_reader.h | 5 + .../parquet/reader/map_column_reader.cpp | 119 +++++++++--- .../parquet/reader/map_column_reader.h | 5 + .../parquet/reader/scalar_column_reader.cpp | 46 ++++- .../parquet/reader/scalar_column_reader.h | 2 + .../parquet/reader/struct_column_reader.cpp | 85 +++++--- .../parquet/reader/struct_column_reader.h | 4 + be/src/format_v2/table_reader.cpp | 1 + be/src/format_v2/table_reader.h | 11 ++ .../parquet/parquet_column_reader_test.cpp | 15 +- .../parquet/parquet_reader_control_test.cpp | 182 +++++++++++++----- be/test/format_v2/table_reader_test.cpp | 56 ++++++ 15 files changed, 518 insertions(+), 151 deletions(-) diff --git a/be/src/format_v2/parquet/reader/column_reader.cpp b/be/src/format_v2/parquet/reader/column_reader.cpp index 80a95cf17ee171..352fbbd7c3d215 100644 --- a/be/src/format_v2/parquet/reader/column_reader.cpp +++ b/be/src/format_v2/parquet/reader/column_reader.cpp @@ -606,15 +606,9 @@ 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); - } - return Status::OK(); +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) { @@ -623,21 +617,19 @@ Status ParquetColumnReader::skip_nested_rows(int64_t rows) { } // A nested parent row may expand to many child values. Capping the number of parent rows per - // scratch column bounds the amplification for large holes while preserving the same nested - // level decoding and validation path as a normal read. Recreate the scratch column per batch - // so its child buffers are released before decoding the next batch. + // 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); - auto scratch_column = _type->create_column(); - RETURN_IF_ERROR(load_nested_batch(batch_rows)); - int64_t rows_read = 0; - RETURN_IF_ERROR(build_nested_column(batch_rows, scratch_column, &rows_read)); - if (rows_read != batch_rows) { + 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_read, batch_rows); + _name, rows_consumed, batch_rows); } remaining_rows -= batch_rows; } diff --git a/be/src/format_v2/parquet/reader/column_reader.h b/be/src/format_v2/parquet/reader/column_reader.h index 7072a8c7f28bad..9416c4eb22739f 100644 --- a/be/src/format_v2/parquet/reader/column_reader.h +++ b/be/src/format_v2/parquet/reader/column_reader.h @@ -80,17 +80,32 @@ 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 decoding strings. + // 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 also avoids decoding leaf payloads that 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,9 +124,9 @@ class ParquetColumnReader { ParquetColumnReader(const ParquetColumnSchema& schema, const DataTypePtr type, ParquetColumnReaderProfile profile = {}); ParquetColumnReader() = default; - // Complex readers cannot advance their child streams without rebuilding parent boundaries - // from definition/repetition levels. Materialize that discarded shape in bounded batches so - // a large selection gap does not allocate a scratch column for the entire gap at once. + // 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 both payload decoding 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 3caac5b2ac6d6a..c042fc99b512aa 100644 --- a/be/src/format_v2/parquet/reader/list_column_reader.cpp +++ b/be/src/format_v2/parquet/reader/list_column_reader.cpp @@ -56,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)); @@ -84,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; @@ -104,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); @@ -120,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 { @@ -129,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, @@ -156,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 c5a7e0e47f9e13..8217d0c013abc0 100644 --- a/be/src/format_v2/parquet/reader/map_column_reader.cpp +++ b/be/src/format_v2/parquet/reader/map_column_reader.cpp @@ -60,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(); @@ -84,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)); @@ -92,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; @@ -114,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); @@ -128,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())) { @@ -170,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/scalar_column_reader.cpp b/be/src/format_v2/parquet/reader/scalar_column_reader.cpp index 6e3b1c7f4d5d20..1e21c71b211acd 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,31 @@ 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); + const int16_t materialized_slot_definition_level = _nested_batch->value_slot_definition_level; + *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 +526,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 8d69f75f8d3431..5abe7abe75e9a2 100644 --- a/be/src/format_v2/parquet/reader/struct_column_reader.cpp +++ b/be/src/format_v2/parquet/reader/struct_column_reader.cpp @@ -102,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(); @@ -124,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; @@ -132,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; @@ -143,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 @@ -174,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 {}", @@ -192,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 a7eb5ba35e891d..468a649f8d6e74 100644 --- a/be/src/format_v2/table_reader.cpp +++ b/be/src/format_v2/table_reader.cpp @@ -750,6 +750,7 @@ 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; } diff --git a/be/src/format_v2/table_reader.h b/be/src/format_v2/table_reader.h index 30d6a9c3d24f4d..c27f64f7f54a80 100644 --- a/be/src/format_v2/table_reader.h +++ b/be/src/format_v2/table_reader.h @@ -909,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. @@ -1545,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..f9bfe137cf74eb 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(); } }; @@ -1929,7 +1924,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 +1948,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) { 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 ca19f096924e90..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,9 @@ 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 { @@ -332,6 +320,13 @@ class ChunkedNestedLeafReader final : public ParquetColumnReader { 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); @@ -354,6 +349,13 @@ class ChunkedNestedLeafReader 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); + *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 { @@ -363,13 +365,17 @@ class ChunkedNestedLeafReader final : public ParquetColumnReader { 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; }; @@ -514,26 +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); - - NestedBuildReader short_reader(2); - EXPECT_FALSE(short_reader.skip_nested_column(3).ok()); + int64_t values_consumed = 0; + EXPECT_FALSE(base_reader.consume_nested_column(1, &values_consumed).ok()); } -TEST(ParquetColumnReaderControlTest, NestedSkipMaterializesBoundedBatches) { +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_EQ(element_reader_ptr->load_lengths(), std::vector({4096, 4096, 1})); - EXPECT_EQ(element_reader_ptr->build_lengths(), std::vector({4096, 4096, 1})); - // Each child build sees an empty destination, proving the parent scratch column from the - // previous batch was destroyed instead of retaining all skipped nested values. - EXPECT_EQ(element_reader_ptr->initial_column_sizes(), std::vector({0, 0, 0})); + 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()); +} + +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) { @@ -937,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 d47da71962f1e8..65070cfed784b8 100644 --- a/be/test/format_v2/table_reader_test.cpp +++ b/be/test/format_v2/table_reader_test.cpp @@ -1362,6 +1362,7 @@ TEST(TableReaderTest, PendingRuntimeFilterDisablesTableLevelCount) { 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, @@ -1392,6 +1393,61 @@ TEST(TableReaderTest, PendingRuntimeFilterDisablesTableLevelCount) { 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())); From 19de063a24de0b17430f0f8670082bdb39f9d260 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Mon, 13 Jul 2026 09:11:00 +0800 Subject: [PATCH 5/8] [fix](be) Reset binary builders after nested level reads ### What problem does this PR solve? Issue Number: close #65498 Related PR: #65498 Problem Summary: The Parquet levels-only nested path called Arrow RecordReader::Reset() without first releasing ByteArray and FLBA builder chunks. Arrow requires GetBuilderChunks() to reset those builders, so a nested skip followed by a normal read mixed values from both batches and failed with values greater than the current definition/repetition levels. Release and immediately discard binary chunks after copying levels, without copying payloads into Doris Columns. Also derive the scalar consume slot threshold from schema metadata because levels-only batches intentionally omit value-slot metadata. ### Release note Fix nested Parquet queries that skip filtered rows before reading plain-encoded string values. ### Check List (For Author) - Test: Unit Test - 95 targeted Parquet reader, levels-only COUNT, and TableReader tests - Plain-encoded nested MAP skip-then-read regression test - Remote ./build.sh --be - Behavior changed: Yes (levels-only nested reads correctly reset Arrow binary builders) - Does this need documentation: No --- .../format_v2/parquet/reader/column_reader.h | 9 +-- .../parquet/reader/parquet_leaf_reader.cpp | 20 ++++-- .../parquet/reader/parquet_leaf_reader.h | 20 +++--- .../parquet/reader/scalar_column_reader.cpp | 7 +- .../parquet/parquet_column_reader_test.cpp | 65 +++++++++++++++++++ 5 files changed, 103 insertions(+), 18 deletions(-) diff --git a/be/src/format_v2/parquet/reader/column_reader.h b/be/src/format_v2/parquet/reader/column_reader.h index 9416c4eb22739f..51dbd44c11c226 100644 --- a/be/src/format_v2/parquet/reader/column_reader.h +++ b/be/src/format_v2/parquet/reader/column_reader.h @@ -84,8 +84,9 @@ class ParquetColumnReader { // 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 decoding strings. - // Normal scans that need output values use load_nested_batch() instead. + // 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, @@ -95,7 +96,7 @@ class ParquetColumnReader { // 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 also avoids decoding leaf payloads that will be discarded. + // 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 @@ -126,7 +127,7 @@ class ParquetColumnReader { 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 both payload decoding and output Columns. + // 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/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 1e21c71b211acd..784e4cdc900fb7 100644 --- a/be/src/format_v2/parquet/reader/scalar_column_reader.cpp +++ b/be/src/format_v2/parquet/reader/scalar_column_reader.cpp @@ -487,7 +487,12 @@ Status ScalarColumnReader::consume_nested_column(int64_t length_upper_bound, _name); } DORIS_CHECK(_nested_batch != nullptr); - const int16_t materialized_slot_definition_level = _nested_batch->value_slot_definition_level; + // 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) { 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 f9bfe137cf74eb..fb4cd129e1b03d 100644 --- a/be/test/format_v2/parquet/parquet_column_reader_test.cpp +++ b/be/test/format_v2/parquet/parquet_column_reader_test.cpp @@ -134,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(); @@ -1819,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]; @@ -1887,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; @@ -3192,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); From 19ce544e1a611ced612f4f49a9d6f05a0076f017 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Mon, 13 Jul 2026 09:27:55 +0800 Subject: [PATCH 6/8] [fix](be) Keep VARBINARY filters above file readers ### What problem does this PR solve? Issue Number: None Related PR: #65498 Problem Summary: Split-level conjunct refresh can expose late runtime filters to TableReader. VARBINARY predicates are intentionally unsupported for external-reader pushdown, but ColumnMapper treated direct VARBINARY mappings as COPY_DIRECTLY and localized late filters into file readers. This caused Iceberg BINARY partition runtime-filter queries to drop matching rows. Mark VARBINARY mappings FINALIZE_ONLY so Scanner evaluates them after materialization while the refreshed snapshot still gates unsafe aggregate shortcuts. ### Release note Fix incorrect results for runtime filters on external VARBINARY columns, including Iceberg BINARY partition columns. ### Check List (For Author) - Test: Unit Test - Remote ASAN BE unit tests: ColumnMapperLocalizeFiltersTest.* and TableReaderTest.PrepareSplitReplacesInitialConjunctSnapshot (6 passed) - Iceberg regression attempted but skipped because remote config disables Iceberg external tests - Behavior changed: Yes (VARBINARY runtime filters stay on scanner-level evaluation path) - Does this need documentation: No --- be/src/format_v2/column_mapper.cpp | 15 ++++++++++-- be/test/format_v2/column_mapper_test.cpp | 31 ++++++++++++++++++++++++ 2 files changed, 44 insertions(+), 2 deletions(-) diff --git a/be/src/format_v2/column_mapper.cpp b/be/src/format_v2/column_mapper.cpp index e60e22b85e7fdf..eefdf6f28a2764 100644 --- a/be/src/format_v2/column_mapper.cpp +++ b/be/src/format_v2/column_mapper.cpp @@ -990,6 +990,18 @@ static bool mapping_can_use_file_column_directly(const ColumnMapping& mapping) { return !needs_complex_rematerialize(mapping); } +static FilterConversionType direct_filter_conversion(const ColumnMapping& mapping) { + DORIS_CHECK(mapping.table_type != nullptr); + // FileScanOperator deliberately keeps VARBINARY predicates above external readers. Their + // physical binary representations are not uniformly supported by reader-side expression and + // metadata filtering, so localizing a late runtime filter here can incorrectly reject rows. + if (remove_nullable(mapping.table_type)->get_primitive_type() == TYPE_VARBINARY) { + return FilterConversionType::FINALIZE_ONLY; + } + return mapping.is_trivial ? FilterConversionType::COPY_DIRECTLY + : FilterConversionType::CAST_FILTER; +} + static const ColumnDefinition* find_file_child_for_mapping(const ColumnDefinition& table_child, const ColumnDefinition& file_parent, TableColumnMappingMode mode, @@ -1870,8 +1882,7 @@ Status TableColumnMapper::_create_direct_mapping(const ColumnDefinition& table_c mapping->projected_file_children = file_field.children; mapping->file_type = file_field.type; mapping->is_trivial = mapping_can_use_file_column_directly(*mapping); - mapping->filter_conversion = mapping->is_trivial ? FilterConversionType::COPY_DIRECTLY - : FilterConversionType::CAST_FILTER; + mapping->filter_conversion = direct_filter_conversion(*mapping); mapping->child_mappings.clear(); auto table_children = table_column.children; diff --git a/be/test/format_v2/column_mapper_test.cpp b/be/test/format_v2/column_mapper_test.cpp index 4e815d6cbf3c5e..715d9eb512fac1 100644 --- a/be/test/format_v2/column_mapper_test.cpp +++ b/be/test/format_v2/column_mapper_test.cpp @@ -37,6 +37,7 @@ #include "core/data_type/data_type_string.h" #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/vexpr.h" #include "exprs/vexpr_context.h" #include "exprs/vin_predicate.h" @@ -78,6 +79,10 @@ DataTypePtr str() { return std::make_shared(); } +DataTypePtr varbinary() { + return std::make_shared(); +} + DataTypePtr timestamptz(uint32_t scale) { return std::make_shared(scale); } @@ -2157,6 +2162,32 @@ TEST(ColumnMapperLocalizeFiltersTest, VisibleLocalFilterAddsPredicateColumnAndCo EXPECT_TRUE(localized_slot->data_type()->equals(*int_type)); } +TEST(ColumnMapperLocalizeFiltersTest, VarbinaryFilterStaysAboveFileReader) { + const auto binary_type = varbinary(); + const auto table_column = name_col("partition_key", binary_type); + const auto file_column = name_col("partition_key", binary_type, 7); + + TableColumnMapper mapper({.mode = TableColumnMappingMode::BY_NAME}); + ASSERT_TRUE(mapper.create_mapping({table_column}, {}, {file_column}).ok()); + ASSERT_EQ(mapper.mappings().size(), 1); + EXPECT_TRUE(mapper.mappings()[0].is_trivial); + EXPECT_EQ(mapper.mappings()[0].filter_conversion, FilterConversionType::FINALIZE_ONLY); + + const auto value = Field::create_field(StringView("binary-value")); + TableFilter filter { + .conjunct = VExprContext::create_shared(binary_predicate( + TExprOpcode::EQ, table_slot(0, 0, binary_type, "partition_key"), + literal(binary_type, value))), + .global_indices = {GlobalIndex(0)}}; + + FileScanRequest request; + ASSERT_TRUE(mapper.create_scan_request({filter}, {table_column}, &request).ok()); + EXPECT_TRUE(request.predicate_columns.empty()); + ASSERT_EQ(request.non_predicate_columns.size(), 1); + EXPECT_EQ(request.non_predicate_columns[0].column_id(), LocalColumnId(7)); + EXPECT_TRUE(request.conjuncts.empty()); +} + TEST(ColumnMapperLocalizeFiltersTest, ConstantFilterBuildsEntryWithoutFileScanColumn) { auto partition_column = name_col("part", i32()); partition_column.is_partition_key = true; From 5b3c674ef08ee9d554f285b57e17aca4513f8669 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Mon, 13 Jul 2026 09:32:16 +0800 Subject: [PATCH 7/8] update --- be/test/format_v2/column_mapper_test.cpp | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/be/test/format_v2/column_mapper_test.cpp b/be/test/format_v2/column_mapper_test.cpp index 715d9eb512fac1..496eb00ad0af96 100644 --- a/be/test/format_v2/column_mapper_test.cpp +++ b/be/test/format_v2/column_mapper_test.cpp @@ -2174,11 +2174,10 @@ TEST(ColumnMapperLocalizeFiltersTest, VarbinaryFilterStaysAboveFileReader) { EXPECT_EQ(mapper.mappings()[0].filter_conversion, FilterConversionType::FINALIZE_ONLY); const auto value = Field::create_field(StringView("binary-value")); - TableFilter filter { - .conjunct = VExprContext::create_shared(binary_predicate( - TExprOpcode::EQ, table_slot(0, 0, binary_type, "partition_key"), - literal(binary_type, value))), - .global_indices = {GlobalIndex(0)}}; + TableFilter filter {.conjunct = VExprContext::create_shared(binary_predicate( + TExprOpcode::EQ, table_slot(0, 0, binary_type, "partition_key"), + literal(binary_type, value))), + .global_indices = {GlobalIndex(0)}}; FileScanRequest request; ASSERT_TRUE(mapper.create_scan_request({filter}, {table_column}, &request).ok()); From 77a59cb6a9c1389944c88df52c096721afe8e0eb Mon Sep 17 00:00:00 2001 From: Gabriel Date: Mon, 13 Jul 2026 09:57:29 +0800 Subject: [PATCH 8/8] [fix](be) Revert misplaced VARBINARY filter change ### What problem does this PR solve? Issue Number: None Related PR: #65498 Problem Summary: The VARBINARY external-reader filter fix and its unit test were committed to this PR by mistake. Restore the two affected files to their state before the misplaced commits so this PR contains only its intended changes. ### Release note None ### Check List (For Author) - Test: No need to test (the commit only restores the exact pre-change file contents) - Behavior changed: No - Does this need documentation: No --- be/src/format_v2/column_mapper.cpp | 15 ++---------- be/test/format_v2/column_mapper_test.cpp | 30 ------------------------ 2 files changed, 2 insertions(+), 43 deletions(-) diff --git a/be/src/format_v2/column_mapper.cpp b/be/src/format_v2/column_mapper.cpp index eefdf6f28a2764..e60e22b85e7fdf 100644 --- a/be/src/format_v2/column_mapper.cpp +++ b/be/src/format_v2/column_mapper.cpp @@ -990,18 +990,6 @@ static bool mapping_can_use_file_column_directly(const ColumnMapping& mapping) { return !needs_complex_rematerialize(mapping); } -static FilterConversionType direct_filter_conversion(const ColumnMapping& mapping) { - DORIS_CHECK(mapping.table_type != nullptr); - // FileScanOperator deliberately keeps VARBINARY predicates above external readers. Their - // physical binary representations are not uniformly supported by reader-side expression and - // metadata filtering, so localizing a late runtime filter here can incorrectly reject rows. - if (remove_nullable(mapping.table_type)->get_primitive_type() == TYPE_VARBINARY) { - return FilterConversionType::FINALIZE_ONLY; - } - return mapping.is_trivial ? FilterConversionType::COPY_DIRECTLY - : FilterConversionType::CAST_FILTER; -} - static const ColumnDefinition* find_file_child_for_mapping(const ColumnDefinition& table_child, const ColumnDefinition& file_parent, TableColumnMappingMode mode, @@ -1882,7 +1870,8 @@ Status TableColumnMapper::_create_direct_mapping(const ColumnDefinition& table_c mapping->projected_file_children = file_field.children; mapping->file_type = file_field.type; mapping->is_trivial = mapping_can_use_file_column_directly(*mapping); - mapping->filter_conversion = direct_filter_conversion(*mapping); + mapping->filter_conversion = mapping->is_trivial ? FilterConversionType::COPY_DIRECTLY + : FilterConversionType::CAST_FILTER; mapping->child_mappings.clear(); auto table_children = table_column.children; diff --git a/be/test/format_v2/column_mapper_test.cpp b/be/test/format_v2/column_mapper_test.cpp index 496eb00ad0af96..4e815d6cbf3c5e 100644 --- a/be/test/format_v2/column_mapper_test.cpp +++ b/be/test/format_v2/column_mapper_test.cpp @@ -37,7 +37,6 @@ #include "core/data_type/data_type_string.h" #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/vexpr.h" #include "exprs/vexpr_context.h" #include "exprs/vin_predicate.h" @@ -79,10 +78,6 @@ DataTypePtr str() { return std::make_shared(); } -DataTypePtr varbinary() { - return std::make_shared(); -} - DataTypePtr timestamptz(uint32_t scale) { return std::make_shared(scale); } @@ -2162,31 +2157,6 @@ TEST(ColumnMapperLocalizeFiltersTest, VisibleLocalFilterAddsPredicateColumnAndCo EXPECT_TRUE(localized_slot->data_type()->equals(*int_type)); } -TEST(ColumnMapperLocalizeFiltersTest, VarbinaryFilterStaysAboveFileReader) { - const auto binary_type = varbinary(); - const auto table_column = name_col("partition_key", binary_type); - const auto file_column = name_col("partition_key", binary_type, 7); - - TableColumnMapper mapper({.mode = TableColumnMappingMode::BY_NAME}); - ASSERT_TRUE(mapper.create_mapping({table_column}, {}, {file_column}).ok()); - ASSERT_EQ(mapper.mappings().size(), 1); - EXPECT_TRUE(mapper.mappings()[0].is_trivial); - EXPECT_EQ(mapper.mappings()[0].filter_conversion, FilterConversionType::FINALIZE_ONLY); - - const auto value = Field::create_field(StringView("binary-value")); - TableFilter filter {.conjunct = VExprContext::create_shared(binary_predicate( - TExprOpcode::EQ, table_slot(0, 0, binary_type, "partition_key"), - literal(binary_type, value))), - .global_indices = {GlobalIndex(0)}}; - - FileScanRequest request; - ASSERT_TRUE(mapper.create_scan_request({filter}, {table_column}, &request).ok()); - EXPECT_TRUE(request.predicate_columns.empty()); - ASSERT_EQ(request.non_predicate_columns.size(), 1); - EXPECT_EQ(request.non_predicate_columns[0].column_id(), LocalColumnId(7)); - EXPECT_TRUE(request.conjuncts.empty()); -} - TEST(ColumnMapperLocalizeFiltersTest, ConstantFilterBuildsEntryWithoutFileScanColumn) { auto partition_column = name_col("part", i32()); partition_column.is_partition_key = true;