From a67a328803bc805059d02d52a5eb7bdc7af69dee Mon Sep 17 00:00:00 2001 From: Gabriel Date: Mon, 3 Aug 2026 22:36:03 +0800 Subject: [PATCH] [opt](parquet) fuse fragmented nullable selection planning --- be/benchmark/parquet/AGENTS.md | 12 +- be/benchmark/parquet/README.md | 11 +- .../parquet/benchmark_parquet_kernels.hpp | 202 ++++++++++++++++++ .../parquet/parquet_benchmark_scenarios.h | 37 ++++ .../reader/native/column_chunk_reader.cpp | 59 +++++ .../reader/native/column_chunk_reader.h | 8 + .../parquet/reader/native/column_reader.cpp | 43 +++- .../parquet/reader/native/column_reader.h | 1 + .../parquet/reader/native/common.cpp | 128 +++++++++++ .../format_v2/parquet/reader/native/common.h | 11 + .../parquet_benchmark_scenarios_test.cpp | 26 +++ .../parquet/parquet_reader_control_test.cpp | 74 +++++++ 12 files changed, 601 insertions(+), 11 deletions(-) diff --git a/be/benchmark/parquet/AGENTS.md b/be/benchmark/parquet/AGENTS.md index 4c1d3cf4d5e197..4d8c0610f17b5d 100644 --- a/be/benchmark/parquet/AGENTS.md +++ b/be/benchmark/parquet/AGENTS.md @@ -51,7 +51,7 @@ be/output/lib/benchmark_test --benchmark_list_tests \ | grep -c '^ParquetDecoder/' # currently 228 be/output/lib/benchmark_test --benchmark_list_tests \ - | grep -c '^ParquetKernel/' # currently 92 + | grep -c '^ParquetKernel/' # currently 292 be/output/lib/benchmark_test --benchmark_list_tests \ | grep -c '^ParquetSelection/' # currently 25 @@ -146,13 +146,19 @@ cache to manufacture a cold run. | DELTA_LENGTH_BYTE_ARRAY | BYTE_ARRAY | | DELTA_BYTE_ARRAY | BYTE_ARRAY | -`ParquetKernel` contains 92 cases across six decode and selection stages: BYTE_STREAM_SPLIT, -DELTA_PREFIX_SUM, DICTIONARY_GATHER, NULLABLE_EXPAND, RAW_PREDICATE, and NESTED_SELECTION. It covers +`ParquetKernel` contains 292 cases across seven decode and selection stages: BYTE_STREAM_SPLIT, +DELTA_PREFIX_SUM, DICTIONARY_GATHER, NULLABLE_EXPAND, NULLABLE_SELECTION, RAW_PREDICATE, and +NESTED_SELECTION. It covers the applicable four- and eight-byte types, three dictionary working-set sizes, 0% through 90% null rates with both placement patterns, 0% through 100% raw-predicate selectivities, and 1%, 10%, and 50% nested parent-row selectivities with both placement patterns. Nested selection registers the legacy and fused implementations in the same binary and validates both against an independent source-level oracle before timing. +Nullable selection contributes 200 legacy/fused cases across five selectivities, five null rates, +and independent clustered or alternating selection/null placement. Each pair is validated for +identical physical ranges and null maps before timing. Treat no-NULL, low-NULL, and clustered +level-plan cases as negative controls: production fusion is gated to batches with at least 1,024 +rows, at least 10% NULLs, and materially fragmented definition-level runs. `ParquetSelection` contains 25 cases that isolate the selection-vector work used by Parquet predicate evaluation. It measures identity initialization, one raw-row filter, and two successive diff --git a/be/benchmark/parquet/README.md b/be/benchmark/parquet/README.md index 156302b24238d1..891fd02d8375ed 100644 --- a/be/benchmark/parquet/README.md +++ b/be/benchmark/parquet/README.md @@ -45,14 +45,21 @@ be/output/lib/benchmark_test \ ## SIMD kernel cases -`ParquetKernel` isolates six decode and selection stages from reader setup and virtual consumer +`ParquetKernel` isolates seven decode and selection stages from reader setup and virtual consumer overhead: byte-stream-split transpose, delta prefix sum, numeric dictionary gather, nullable -expansion, raw predicate evaluation, and repeated-level sparse selection. It covers the applicable +expansion, nullable selection planning, raw predicate evaluation, and repeated-level sparse +selection. It covers the applicable 4-byte and 8-byte integer and floating-point physical types, raw-predicate selectivities from 0% through 100%, and nullable rates from 0% through 90% with clustered and alternating placement. Nested selection covers 1%, 10%, and 50% surviving parent rows with both placement patterns. Each nested-selection scenario registers both `impl_legacy` and `impl_fused`; both paths use the same source levels and are checked against an independent oracle before timing. +Nullable selection planning registers legacy and fused pairs across five selectivities, five null +rates, and independent clustered or alternating selection/null placement. Both implementations are +checked for identical physical ranges and null maps before timing. The full matrix also acts as a +negative control: production fusion is limited to batches with at least 1,024 rows, at least 10% +NULLs, and fragmented definition-level runs; no-NULL, low-NULL, and clustered pages retain the +legacy planner. Dictionary gather uses 32-, 4,096-, and 262,144-entry working sets to separate cache-resident and cache-miss-dominated behavior. diff --git a/be/benchmark/parquet/benchmark_parquet_kernels.hpp b/be/benchmark/parquet/benchmark_parquet_kernels.hpp index 619d58fd8cf8ac..de3a15c254b8db 100644 --- a/be/benchmark/parquet/benchmark_parquet_kernels.hpp +++ b/be/benchmark/parquet/benchmark_parquet_kernels.hpp @@ -63,6 +63,191 @@ struct NestedSelectionScratch { size_t ancestor_null_count = 0; }; +struct NullableSelectionScratch { + format::parquet::native::ColumnSelectVector legacy_selection; + ParquetSelection physical_selection; + NullMap output_nulls; + NullMap selected_nulls; + size_t num_filtered = 0; +}; + +inline void append_nullable_run(std::vector* runs, bool is_null, size_t run_length, + bool* previous_is_null) { + if (runs->empty()) { + if (is_null) { + runs->push_back(0); + } + } else if (*previous_is_null == is_null) { + runs->push_back(0); + } + while (run_length > USHRT_MAX) { + runs->push_back(USHRT_MAX); + runs->push_back(0); + run_length -= USHRT_MAX; + } + runs->push_back(static_cast(run_length)); + *previous_is_null = is_null; +} + +inline std::vector build_nullable_runs(const NullMap& nulls) { + std::vector runs; + bool previous_is_null = false; + size_t row = 0; + while (row < nulls.size()) { + const bool is_null = nulls[row] != 0; + const size_t begin = row++; + while (row < nulls.size() && (nulls[row] != 0) == is_null) { + ++row; + } + append_nullable_run(&runs, is_null, row - begin, &previous_is_null); + } + return runs; +} + +inline Status run_legacy_nullable_selection(NullableSelectionScratch* scratch, + const std::vector& null_runs, + size_t num_values, + format::parquet::native::FilterMap* filter) { + using ReadType = format::parquet::native::ColumnSelectVector::DataReadType; + scratch->output_nulls.clear(); + scratch->selected_nulls.clear(); + scratch->physical_selection.ranges.clear(); + scratch->physical_selection.total_values = 0; + scratch->physical_selection.selected_values = 0; + RETURN_IF_ERROR(scratch->legacy_selection.init(null_runs, num_values, &scratch->output_nulls, + filter, 0)); + scratch->num_filtered = scratch->legacy_selection.num_filtered(); + + size_t physical_cursor = 0; + ReadType type; + while (const size_t run_length = scratch->legacy_selection.get_next_run(&type)) { + switch (type) { + case ReadType::CONTENT: + if (!scratch->physical_selection.ranges.empty() && + scratch->physical_selection.ranges.back().first + + scratch->physical_selection.ranges.back().count == + physical_cursor) { + scratch->physical_selection.ranges.back().count += run_length; + } else { + scratch->physical_selection.ranges.push_back( + {.first = physical_cursor, .count = run_length}); + } + scratch->physical_selection.selected_values += run_length; + scratch->selected_nulls.resize_fill(scratch->selected_nulls.size() + run_length, 0); + physical_cursor += run_length; + break; + case ReadType::NULL_DATA: + scratch->selected_nulls.resize_fill(scratch->selected_nulls.size() + run_length, 1); + break; + case ReadType::FILTERED_CONTENT: + physical_cursor += run_length; + break; + case ReadType::FILTERED_NULL: + break; + } + } + scratch->physical_selection.total_values = physical_cursor; + return Status::OK(); +} + +inline Status run_nullable_selection_once(NullableSelectionScratch* scratch, + const std::vector& null_runs, size_t num_values, + size_t num_nulls, + format::parquet::native::FilterMap* filter, + NullableSelectionImplementation implementation) { + if (implementation == NullableSelectionImplementation::LEGACY) { + return run_legacy_nullable_selection(scratch, null_runs, num_values, filter); + } + scratch->output_nulls.clear(); + return format::parquet::native::build_filtered_nullable_selection( + null_runs, num_values, num_nulls, &scratch->output_nulls, filter, 0, + &scratch->physical_selection, &scratch->selected_nulls, &scratch->num_filtered); +} + +inline bool equal_selection(const ParquetSelection& lhs, const ParquetSelection& rhs) { + if (lhs.total_values != rhs.total_values || lhs.selected_values != rhs.selected_values || + lhs.ranges.size() != rhs.ranges.size()) { + return false; + } + for (size_t range = 0; range < lhs.ranges.size(); ++range) { + if (lhs.ranges[range].first != rhs.ranges[range].first || + lhs.ranges[range].count != rhs.ranges[range].count) { + return false; + } + } + return true; +} + +inline void run_nullable_selection_kernel(benchmark::State& state, + const NullableSelectionScenario& scenario) { + using format::parquet::native::FilterMap; + + std::vector filter_data(KERNEL_ROWS, 0); + const auto selected = make_selection_plan(KERNEL_ROWS, scenario.selectivity_percent, + scenario.selection_pattern); + visit_selected_rows(selected, [&](size_t row) { filter_data[row] = 1; }); + FilterMap filter; + auto status = filter.init(filter_data.data(), filter_data.size(), false); + if (!status.ok()) { + state.SkipWithError(status.to_string().c_str()); + return; + } + + NullMap nulls; + nulls.resize_fill(KERNEL_ROWS, 0); + const auto null_plan = + make_selection_plan(KERNEL_ROWS, scenario.null_percent, scenario.null_pattern); + visit_selected_rows(null_plan, [&](size_t row) { nulls[row] = 1; }); + const auto null_runs = build_nullable_runs(nulls); + + NullableSelectionScratch legacy; + NullableSelectionScratch fused; + status = run_nullable_selection_once(&legacy, null_runs, KERNEL_ROWS, null_plan.selected_rows, + &filter, NullableSelectionImplementation::LEGACY); + if (status.ok()) { + status = + run_nullable_selection_once(&fused, null_runs, KERNEL_ROWS, null_plan.selected_rows, + &filter, NullableSelectionImplementation::FUSED); + } + if (!status.ok() || !equal_selection(legacy.physical_selection, fused.physical_selection) || + legacy.output_nulls != fused.output_nulls || + legacy.selected_nulls != fused.selected_nulls || + legacy.num_filtered != fused.num_filtered) { + if (status.ok()) { + state.SkipWithError("nullable selection implementations disagree"); + } else { + state.SkipWithError(status.to_string().c_str()); + } + return; + } + + NullableSelectionScratch scratch; + status = run_nullable_selection_once(&scratch, null_runs, KERNEL_ROWS, null_plan.selected_rows, + &filter, scenario.implementation); + if (!status.ok()) { + state.SkipWithError(status.to_string().c_str()); + return; + } + for (auto _ : state) { + status = run_nullable_selection_once(&scratch, null_runs, KERNEL_ROWS, + null_plan.selected_rows, &filter, + scenario.implementation); + if (!status.ok()) { + state.SkipWithError(status.to_string().c_str()); + return; + } + benchmark::DoNotOptimize(scratch.physical_selection.ranges.data()); + benchmark::DoNotOptimize(scratch.selected_nulls.data()); + benchmark::ClobberMemory(); + } + + state.SetItemsProcessed(static_cast(state.iterations()) * + static_cast(KERNEL_ROWS)); + state.counters["rows"] = static_cast(KERNEL_ROWS); + state.counters["selected_rows"] = static_cast(selected.selected_rows); + state.counters["null_rows"] = static_cast(null_plan.selected_rows); +} + inline NestedSelectionOracle build_nested_selection_oracle( const std::vector& repetition_levels, const std::vector& definition_levels, @@ -493,7 +678,24 @@ inline bool register_kernel_benchmarks() { return true; } +inline bool register_nullable_selection_benchmarks() { + for (const auto& scenario : nullable_selection_scenarios()) { + const std::string name = "ParquetKernel/nullable_selection/sel_" + + std::to_string(scenario.selectivity_percent) + "/null_" + + std::to_string(scenario.null_percent) + "/selection_" + + to_string(scenario.selection_pattern) + "/nulls_" + + to_string(scenario.null_pattern) + "/impl_" + + to_string(scenario.implementation); + benchmark::RegisterBenchmark(name.c_str(), [=](benchmark::State& state) { + run_nullable_selection_kernel(state, scenario); + })->Unit(benchmark::kNanosecond); + } + return true; +} + inline const bool KERNEL_BENCHMARKS_REGISTERED = register_kernel_benchmarks(); +inline const bool NULLABLE_SELECTION_BENCHMARKS_REGISTERED = + register_nullable_selection_benchmarks(); } // namespace detail } // namespace doris::parquet_benchmark diff --git a/be/benchmark/parquet/parquet_benchmark_scenarios.h b/be/benchmark/parquet/parquet_benchmark_scenarios.h index a9c58c15d8cff1..f7eddf0b11a5aa 100644 --- a/be/benchmark/parquet/parquet_benchmark_scenarios.h +++ b/be/benchmark/parquet/parquet_benchmark_scenarios.h @@ -55,6 +55,7 @@ enum class Kernel { NESTED_SELECTION }; enum class NestedSelectionImplementation { LEGACY, FUSED }; +enum class NullableSelectionImplementation { LEGACY, FUSED }; struct DecoderScenario { Encoding encoding; @@ -89,6 +90,14 @@ struct SelectionScenario { Pattern pattern; }; +struct NullableSelectionScenario { + int selectivity_percent; + int null_percent; + Pattern selection_pattern; + Pattern null_pattern; + NullableSelectionImplementation implementation; +}; + struct SelectionRange { size_t first; size_t count; @@ -177,6 +186,24 @@ inline std::vector selection_scenarios() { return scenarios; } +inline std::vector nullable_selection_scenarios() { + std::vector scenarios; + for (const int selectivity : {1, 10, 50, 90, 99}) { + for (const int null_percent : {0, 1, 10, 50, 90}) { + for (const auto selection_pattern : {Pattern::CLUSTERED, Pattern::ALTERNATING}) { + for (const auto null_pattern : {Pattern::CLUSTERED, Pattern::ALTERNATING}) { + for (const auto implementation : {NullableSelectionImplementation::LEGACY, + NullableSelectionImplementation::FUSED}) { + scenarios.push_back({selectivity, null_percent, selection_pattern, + null_pattern, implementation}); + } + } + } + } + } + return scenarios; +} + inline std::vector reader_scenarios() { std::vector scenarios; std::set::materialize_values( return Status::OK(); } +template +bool ColumnChunkReader::supports_fused_nullable_selection( + IColumn& column) const { + return visit_nullable_expandable_column(column, [](auto&) {}); +} + +template +Status ColumnChunkReader::materialize_fused_nullable_values( + MutableColumnPtr& doris_column, const DataTypeSerDe& serde, ParquetDecodeContext& context, + ParquetMaterializationState& state, size_t num_values, size_t num_nulls, + const NullMap& selected_nulls) { + if (num_values == 0) { + return Status::OK(); + } + SCOPED_RAW_TIMER(&_chunk_statistics.decode_value_time); + DORIS_CHECK_GT(num_nulls, 0); + const size_t physical_values = num_values - num_nulls; + DORIS_CHECK_EQ(state.selection.total_values, physical_values); + DORIS_CHECK_LE(state.selection.selected_values, selected_nulls.size()); + if (UNLIKELY(_empty_value_section && physical_values != 0)) { + return Status::Corruption( + "Parquet definition levels require {} values from an empty value section", + physical_values); + } + if (UNLIKELY((doris_column->is_column_dictionary() || context.dictionary_index_only) && + !_has_dict && physical_values != 0)) { + return Status::IOError("Not dictionary coded"); + } + if (UNLIKELY(_remaining_num_values < num_values)) { + return Status::IOError("Decode too many values in current page"); + } + RETURN_IF_ERROR(translate_value_encoding(_current_encoding, &context.encoding)); + + ++_chunk_statistics.hybrid_selection_batches; + const auto status = decode_prepared_nullable_values(*doris_column, serde, *_page_decoder, + context, state, selected_nulls, + &_chunk_statistics.materialization_time); + _chunk_statistics.hybrid_selection_ranges += state.selection.ranges.size(); + RETURN_IF_ERROR(status); + _remaining_num_values -= num_values; + return Status::OK(); +} + template bool ColumnChunkReader::can_filter_fixed_width_values( const VExprSPtrs& conjuncts, int column_id, const DataTypeSerDe* serde, diff --git a/be/src/format_v2/parquet/reader/native/column_chunk_reader.h b/be/src/format_v2/parquet/reader/native/column_chunk_reader.h index 35bf5200d7aebb..a50b7e4bc080ff 100644 --- a/be/src/format_v2/parquet/reader/native/column_chunk_reader.h +++ b/be/src/format_v2/parquet/reader/native/column_chunk_reader.h @@ -177,6 +177,14 @@ class ColumnChunkReader { ParquetDecodeContext& context, ParquetMaterializationState& state, ColumnSelectVector& select_vector); + bool supports_fused_nullable_selection(IColumn& column) const; + + Status materialize_fused_nullable_values(MutableColumnPtr& doris_column, + const DataTypeSerDe& serde, + ParquetDecodeContext& context, + ParquetMaterializationState& state, size_t num_values, + size_t num_nulls, const NullMap& selected_nulls); + static bool supports_raw_fixed_filter_encoding(tparquet::Encoding::type encoding, tparquet::Type::type physical_type) { switch (encoding) { diff --git a/be/src/format_v2/parquet/reader/native/column_reader.cpp b/be/src/format_v2/parquet/reader/native/column_reader.cpp index f762cb0f9d35f5..b5ea931b29fdeb 100644 --- a/be/src/format_v2/parquet/reader/native/column_reader.cpp +++ b/be/src/format_v2/parquet/reader/native/column_reader.cpp @@ -799,6 +799,8 @@ void ScalarColumnReader::release_batch_scratch( max_retained_bytes); release_selection |= release_vector_if_oversized(&_materialization_state.selection.ranges, max_retained_bytes); + release_selection |= + release_filter_if_oversized(&_fused_nullable_selection_nulls, max_retained_bytes); release_selection |= release_filter_if_oversized(&_fixed_width_predicate_nulls, max_retained_bytes); release_selection |= @@ -832,6 +834,7 @@ void ScalarColumnReader::release_batch_scratch( release_selection |= release_vector_for_aggregate(&_nested_filter_map_data); release_selection |= release_vector_for_aggregate(&_materialization_state.dictionary_indices); release_selection |= release_vector_for_aggregate(&_materialization_state.selection.ranges); + release_selection |= release_filter_for_aggregate(&_fused_nullable_selection_nulls); release_selection |= release_filter_for_aggregate(&_fixed_width_predicate_nulls); release_selection |= release_filter_for_aggregate(&_fixed_width_predicate_matches); release_selection |= release_filter_for_aggregate(&_fixed_width_predicate_conversion_nulls); @@ -860,7 +863,8 @@ size_t ScalarColumnReader::retained_batch_scratch_b _def_levels.capacity() * sizeof(level_t) + _null_run_lengths.capacity() * sizeof(uint16_t) + _nested_filter_map_data.capacity() * sizeof(uint8_t) + - _fixed_width_predicate_nulls.capacity() + _fixed_width_predicate_matches.capacity() + + _fused_nullable_selection_nulls.capacity() + _fixed_width_predicate_nulls.capacity() + + _fixed_width_predicate_matches.capacity() + _fixed_width_predicate_conversion_nulls.capacity() + _materialization_state.dictionary_indices.capacity() * sizeof(uint32_t) + _materialization_state.selection.ranges.capacity() * sizeof(ParquetSelectionRange) + @@ -875,7 +879,8 @@ size_t ScalarColumnReader::active_batch_scratch_byt _serde == nullptr ? 0 : _serde->active_parquet_raw_predicate_scratch_bytes(); return decoder_bytes + serde_bytes + _rep_levels.size() * sizeof(level_t) + _def_levels.size() * sizeof(level_t) + _null_run_lengths.size() * sizeof(uint16_t) + - _nested_filter_map_data.size() * sizeof(uint8_t) + _fixed_width_predicate_nulls.size() + + _nested_filter_map_data.size() * sizeof(uint8_t) + + _fused_nullable_selection_nulls.size() + _fixed_width_predicate_nulls.size() + _fixed_width_predicate_matches.size() + _fixed_width_predicate_conversion_nulls.size() + _materialization_state.dictionary_indices.size() * sizeof(uint32_t) + _materialization_state.selection.ranges.size() * sizeof(ParquetSelectionRange) + @@ -892,6 +897,7 @@ void ScalarColumnReader::reserve_batch_scratch_for_ _nested_filter_map_data.reserve(elements); _materialization_state.dictionary_indices.reserve(elements); _materialization_state.selection.ranges.reserve(elements); + _fused_nullable_selection_nulls.reserve(elements); _ancestor_null_indices.reserve(elements); } @@ -955,6 +961,7 @@ Status ScalarColumnReader::_read_values(size_t num_ } MutableColumnPtr data_column; _null_run_lengths.clear(); + size_t num_nulls = 0; NullMap* map_data_column = nullptr; doris_column = IColumn::mutate(std::move(doris_column)); if (is_column_nullable(*doris_column)) { @@ -977,6 +984,9 @@ Status ScalarColumnReader::_read_values(size_t num_ } bool is_null = def_level < _field_schema->definition_level; + if (is_null) { + num_nulls += loop_read; + } if (!(prev_is_null ^ is_null)) { _null_run_lengths.emplace_back(0); } @@ -1006,10 +1016,26 @@ Status ScalarColumnReader::_read_values(size_t num_ } _null_run_lengths.emplace_back((u_short)remaining); } + const bool use_fused_nullable_selection = + map_data_column != nullptr && filter_map.has_filter() && num_nulls > 0 && + should_use_fused_nullable_selection(num_values, num_nulls, _null_run_lengths.size()) && + _chunk_reader->supports_fused_nullable_selection(*data_column); { SCOPED_RAW_TIMER(&_decode_null_map_time); - RETURN_IF_ERROR(_select_vector.init(_null_run_lengths, num_values, map_data_column, - &filter_map, _filter_map_index)); + if (use_fused_nullable_selection) { + size_t num_filtered = 0; + // The fused path owns both the physical ranges and selected NULL layout. Restrict it + // to fragmented, materially nullable level plans: clustered, low-NULL, and no-NULL + // pages already collapse into a few cheap legacy runs, while fusing them adds planning + // branches without removing enough work to guarantee a win. + RETURN_IF_ERROR(build_filtered_nullable_selection( + _null_run_lengths, num_values, num_nulls, map_data_column, &filter_map, + _filter_map_index, &_materialization_state.selection, + &_fused_nullable_selection_nulls, &num_filtered)); + } else { + RETURN_IF_ERROR(_select_vector.init(_null_run_lengths, num_values, map_data_column, + &filter_map, _filter_map_index)); + } _filter_map_index += num_values; } DORIS_CHECK(_serde != nullptr); @@ -1020,8 +1046,13 @@ Status ScalarColumnReader::_read_values(size_t num_ conversion_failure_map(*_field_schema, type, _materialization_state.enable_strict_mode, map_data_column, &compatibility_scratch); const size_t materialization_start_row = data_column->size(); - const auto status = _chunk_reader->materialize_values(data_column, *_serde, _decode_context, - _materialization_state, _select_vector); + const auto status = + use_fused_nullable_selection + ? _chunk_reader->materialize_fused_nullable_values( + data_column, *_serde, _decode_context, _materialization_state, + num_values, num_nulls, _fused_nullable_selection_nulls) + : _chunk_reader->materialize_values(data_column, *_serde, _decode_context, + _materialization_state, _select_vector); _materialization_state.conversion_failure_null_map = nullptr; if (status.ok()) { mark_local_timestamp_defaults(*_field_schema, type, diff --git a/be/src/format_v2/parquet/reader/native/column_reader.h b/be/src/format_v2/parquet/reader/native/column_reader.h index 18bcdd866f83d7..594f0ac0921523 100644 --- a/be/src/format_v2/parquet/reader/native/column_reader.h +++ b/be/src/format_v2/parquet/reader/native/column_reader.h @@ -432,6 +432,7 @@ class ScalarColumnReader : public ColumnReader { std::vector _null_run_lengths; std::unordered_set _ancestor_null_indices; std::vector _nested_filter_map_data; + NullMap _fused_nullable_selection_nulls; NullMap _fixed_width_predicate_nulls; IColumn::Filter _fixed_width_predicate_matches; IColumn::Filter _fixed_width_predicate_conversion_nulls; diff --git a/be/src/format_v2/parquet/reader/native/common.cpp b/be/src/format_v2/parquet/reader/native/common.cpp index c50488575e4993..020f613c5d267f 100644 --- a/be/src/format_v2/parquet/reader/native/common.cpp +++ b/be/src/format_v2/parquet/reader/native/common.cpp @@ -17,6 +17,7 @@ #include "format_v2/parquet/reader/native/common.h" +#include #include #include "core/types.h" @@ -68,6 +69,133 @@ bool FilterMap::can_filter_all(size_t remaining_num_values, size_t filter_map_in remaining_num_values) == remaining_num_values; } +bool should_use_fused_nullable_selection(size_t num_values, size_t num_nulls, + size_t num_null_runs) { + constexpr size_t MIN_BATCH_VALUES = 1024; + constexpr size_t MIN_NULL_RUNS = 32; + constexpr size_t MAX_AVERAGE_NULL_RUN = 64; + constexpr size_t MIN_NULL_RATIO_DENOMINATOR = 10; + if (num_values < MIN_BATCH_VALUES || num_nulls < num_values / MIN_NULL_RATIO_DENOMINATOR) { + return false; + } + return num_null_runs >= std::max(MIN_NULL_RUNS, num_values / MAX_AVERAGE_NULL_RUN); +} + +Status build_filtered_nullable_selection(const std::vector& run_length_null_map, + size_t num_values, size_t num_nulls, + NullMap* output_null_map, FilterMap* filter_map, + size_t filter_map_index, ParquetSelection* selection, + NullMap* selected_nulls, size_t* num_filtered) { + if (output_null_map == nullptr || filter_map == nullptr || selection == nullptr || + selected_nulls == nullptr || num_filtered == nullptr) { + return Status::InvalidArgument( + "Nullable selection planning requires non-null output state"); + } + if (!filter_map->has_filter()) { + return Status::InvalidArgument("Nullable selection planning requires a row filter"); + } + if (!filter_map->filter_all() && + (filter_map->filter_map_data() == nullptr || + filter_map_index + num_values > filter_map->filter_map_size())) { + return Status::InvalidArgument("Nullable selection filter range [{}, {}) exceeds size {}", + filter_map_index, filter_map_index + num_values, + filter_map->filter_map_size()); + } + if (num_nulls > num_values) { + return Status::InvalidArgument("Nullable selection has {} nulls for {} values", num_nulls, + num_values); + } + + selection->ranges.clear(); + selection->total_values = num_values - num_nulls; + selection->selected_values = 0; + selected_nulls->clear(); + *num_filtered = 0; + if (filter_map->filter_all()) { + *num_filtered = num_values; + return Status::OK(); + } + + selected_nulls->reserve(num_values); + const uint8_t* filter = filter_map->filter_map_data() + filter_map_index; + const auto select_physical_values = [&](size_t physical_index, size_t count) { + if (!selection->ranges.empty() && + selection->ranges.back().first + selection->ranges.back().count == physical_index) { + selection->ranges.back().count += count; + } else { + selection->ranges.push_back({.first = physical_index, .count = count}); + } + selection->selected_values += count; + }; + + if (num_nulls == 0) { + size_t row = 0; + while (row < num_values) { + const bool selected = filter[row] != 0; + const size_t run_start = row++; + while (row < num_values && (filter[row] != 0) == selected) { + ++row; + } + const size_t run_length = row - run_start; + if (selected) { + select_physical_values(run_start, run_length); + } else { + *num_filtered += run_length; + } + } + selected_nulls->resize_fill(selection->selected_values, 0); + } else { + size_t logical_index = 0; + size_t physical_index = 0; + size_t observed_nulls = 0; + bool is_null = false; + for (const size_t run_length : run_length_null_map) { + if (logical_index + run_length > num_values) { + return Status::InvalidArgument("Nullable selection run lengths exceed {} values", + num_values); + } + const size_t run_end = logical_index + run_length; + while (logical_index < run_end) { + const bool selected = filter[logical_index] != 0; + const size_t filter_run_start = logical_index++; + while (logical_index < run_end && (filter[logical_index] != 0) == selected) { + ++logical_index; + } + const size_t filter_run_length = logical_index - filter_run_start; + if (selected) { + selected_nulls->resize_fill(selected_nulls->size() + filter_run_length, + static_cast(is_null)); + if (!is_null) { + select_physical_values(physical_index, filter_run_length); + } + } else { + *num_filtered += filter_run_length; + } + if (!is_null) { + physical_index += filter_run_length; + } else { + observed_nulls += filter_run_length; + } + } + is_null = !is_null; + } + if (logical_index != num_values || observed_nulls != num_nulls || + physical_index != selection->total_values) { + return Status::InvalidArgument( + "Nullable selection level plan is inconsistent: values={}, nulls={}", + logical_index, observed_nulls); + } + } + + const size_t old_null_size = output_null_map->size(); + output_null_map->resize(old_null_size + selected_nulls->size()); + if (!selected_nulls->empty()) { + memcpy(output_null_map->data() + old_null_size, selected_nulls->data(), + selected_nulls->size()); + } + return Status::OK(); +} + Status FilterMap::generate_nested_filter_map(const std::vector& rep_levels, std::vector& nested_filter_map_data, std::unique_ptr* nested_filter_map, diff --git a/be/src/format_v2/parquet/reader/native/common.h b/be/src/format_v2/parquet/reader/native/common.h index eb6848ee299f10..bd687616f226ce 100644 --- a/be/src/format_v2/parquet/reader/native/common.h +++ b/be/src/format_v2/parquet/reader/native/common.h @@ -25,6 +25,7 @@ #include "common/status.h" #include "core/column/column_nullable.h" +#include "core/data_type_serde/parquet_decode_source.h" namespace doris::format::parquet::native { @@ -116,4 +117,14 @@ class ColumnSelectVector { size_t _read_index = 0; }; +Status build_filtered_nullable_selection(const std::vector& run_length_null_map, + size_t num_values, size_t num_nulls, + NullMap* output_null_map, FilterMap* filter_map, + size_t filter_map_index, ParquetSelection* selection, + NullMap* selected_nulls, size_t* num_filtered); + +// Fusion pays for its additional planning branches only when definition levels are materially +// nullable and fragmented. Keep compact/no-NULL batches on the run-oriented legacy path. +bool should_use_fused_nullable_selection(size_t num_values, size_t num_nulls, size_t num_null_runs); + } // namespace doris::format::parquet::native diff --git a/be/test/format_v2/parquet/parquet_benchmark_scenarios_test.cpp b/be/test/format_v2/parquet/parquet_benchmark_scenarios_test.cpp index 2145b6ab60da43..596d076c5f0bca 100644 --- a/be/test/format_v2/parquet/parquet_benchmark_scenarios_test.cpp +++ b/be/test/format_v2/parquet/parquet_benchmark_scenarios_test.cpp @@ -127,6 +127,32 @@ TEST(ParquetBenchmarkScenariosTest, NestedSelectionCoversSparseParentSurvivors) } } +TEST(ParquetBenchmarkScenariosTest, NullableSelectionPairsLegacyAndFusedAcrossRowShapes) { + const auto scenarios = nullable_selection_scenarios(); + EXPECT_EQ(scenarios.size(), size_t {200}); + for (const int selectivity : {1, 10, 50, 90, 99}) { + for (const int null_percent : {0, 1, 10, 50, 90}) { + for (const auto selection_pattern : {Pattern::CLUSTERED, Pattern::ALTERNATING}) { + for (const auto null_pattern : {Pattern::CLUSTERED, Pattern::ALTERNATING}) { + for (const auto implementation : {NullableSelectionImplementation::LEGACY, + NullableSelectionImplementation::FUSED}) { + EXPECT_TRUE(std::ranges::any_of( + scenarios, + [&](const NullableSelectionScenario& scenario) { + return scenario.selectivity_percent == selectivity && + scenario.null_percent == null_percent && + scenario.selection_pattern == selection_pattern && + scenario.null_pattern == null_pattern && + scenario.implementation == implementation; + })) + << "missing nullable selection comparison shape"; + } + } + } + } + } +} + TEST(ParquetBenchmarkScenariosTest, SelectionMatrixCoversIdentityAndSuccessiveCompaction) { const auto scenarios = selection_scenarios(); EXPECT_EQ(scenarios.size(), size_t {25}); 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 e21439e885570d..37deb32d5bf37b 100644 --- a/be/test/format_v2/parquet/parquet_reader_control_test.cpp +++ b/be/test/format_v2/parquet/parquet_reader_control_test.cpp @@ -27,6 +27,7 @@ #include "core/column/column_string.h" #include "core/column/column_vector.h" #include "core/data_type/data_type_number.h" +#include "core/data_type_serde/parquet_decode_source.h" #include "format_v2/parquet/parquet_column_schema.h" #include "format_v2/parquet/parquet_scan.h" #include "format_v2/parquet/reader/column_reader.h" @@ -186,6 +187,79 @@ TEST(SelectionVectorTest, IdentitySelectionDoesNotMaterializeFilter) { EXPECT_EQ(filter, nullptr); } +TEST(NativeNullableSelectionTest, BuildsPhysicalRangesAndSelectedNullsInOnePass) { + using native::FilterMap; + + const std::vector null_runs {2, 1, 3, 2, 2}; + const std::vector filter_data {1, 0, 1, 1, 0, 1, 1, 1, 0, 1}; + FilterMap filter; + ASSERT_TRUE(filter.init(filter_data.data(), filter_data.size(), false).ok()); + ParquetSelection selection; + NullMap output_nulls {1}; + NullMap selected_nulls; + size_t num_filtered = 0; + + ASSERT_TRUE(native::build_filtered_nullable_selection(null_runs, filter_data.size(), 3, + &output_nulls, &filter, 0, &selection, + &selected_nulls, &num_filtered) + .ok()); + + EXPECT_EQ(selection.total_values, 7); + EXPECT_EQ(selection.selected_values, 4); + ASSERT_EQ(selection.ranges.size(), 4); + EXPECT_EQ(selection.ranges[0].first, 0); + EXPECT_EQ(selection.ranges[0].count, 1); + EXPECT_EQ(selection.ranges[1].first, 2); + EXPECT_EQ(selection.ranges[1].count, 1); + EXPECT_EQ(selection.ranges[2].first, 4); + EXPECT_EQ(selection.ranges[2].count, 1); + EXPECT_EQ(selection.ranges[3].first, 6); + EXPECT_EQ(selection.ranges[3].count, 1); + EXPECT_EQ(selected_nulls, (NullMap {0, 1, 0, 0, 1, 1, 0})); + EXPECT_EQ(output_nulls, (NullMap {1, 0, 1, 0, 0, 1, 1, 0})); + EXPECT_EQ(num_filtered, 3); +} + +TEST(NativeNullableSelectionTest, UsesDirectPhysicalCoordinatesWithoutNulls) { + using native::FilterMap; + + const std::vector no_nulls {10}; + const std::vector filter_data {1, 1, 0, 1, 0, 0, 1, 1, 1, 0}; + FilterMap filter; + ASSERT_TRUE(filter.init(filter_data.data(), filter_data.size(), false).ok()); + ParquetSelection selection; + NullMap output_nulls; + NullMap selected_nulls; + size_t num_filtered = 0; + + ASSERT_TRUE(native::build_filtered_nullable_selection(no_nulls, filter_data.size(), 0, + &output_nulls, &filter, 0, &selection, + &selected_nulls, &num_filtered) + .ok()); + + EXPECT_EQ(selection.total_values, 10); + EXPECT_EQ(selection.selected_values, 6); + ASSERT_EQ(selection.ranges.size(), 3); + EXPECT_EQ(selection.ranges[0].first, 0); + EXPECT_EQ(selection.ranges[0].count, 2); + EXPECT_EQ(selection.ranges[1].first, 3); + EXPECT_EQ(selection.ranges[1].count, 1); + EXPECT_EQ(selection.ranges[2].first, 6); + EXPECT_EQ(selection.ranges[2].count, 3); + EXPECT_EQ(selected_nulls, (NullMap {0, 0, 0, 0, 0, 0})); + EXPECT_EQ(output_nulls, selected_nulls); + EXPECT_EQ(num_filtered, 4); +} + +TEST(NativeNullableSelectionTest, EnablesFusionOnlyForMateriallyFragmentedNullableBatches) { + EXPECT_FALSE(native::should_use_fused_nullable_selection(65536, 0, 3)); + EXPECT_FALSE(native::should_use_fused_nullable_selection(65536, 655, 1311)); + EXPECT_FALSE(native::should_use_fused_nullable_selection(65536, 32768, 3)); + EXPECT_FALSE(native::should_use_fused_nullable_selection(512, 256, 512)); + EXPECT_TRUE(native::should_use_fused_nullable_selection(65536, 6553, 13107)); + EXPECT_TRUE(native::should_use_fused_nullable_selection(65536, 32768, 65536)); +} + TEST(NativeNestedSelectionTest, BuildsSelectionAndCompactsSurvivingParentLevels) { using native::ColumnSelectVector; using native::FilterMap;