This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 46bd26c7f6f [opt](parquet) Fuse fragmented nullable selection planning
(#66397)
46bd26c7f6f is described below
commit 46bd26c7f6f8431e4ab87cd507f3ecbfba10c2d1
Author: Gabriel <[email protected]>
AuthorDate: Tue Aug 4 09:27:15 2026 +0800
[opt](parquet) Fuse fragmented nullable selection planning (#66397)
### What
- Fuse nullable definition-level runs with the row filter in one
traversal.
- Produce physical decode ranges, the selected NULL map, and selected
value counts without first materializing and rescanning a row-wise
selection map.
- Reuse the existing selected-decoder strategies and nullable in-place
expansion.
- Restrict fusion to batches with at least 1,024 rows, at least 10%
NULLs, and materially fragmented definition-level runs. No-NULL,
low-NULL, clustered, nested, and non-expandable shapes keep the legacy
path.
### Why the guard is conservative
The full benchmark matrix includes no-NULL, low-NULL, and clustered
level plans as negative controls. Those shapes do not remove enough
legacy work to guarantee a win, so this change deliberately leaves them
unchanged. Decoder selection and encoding-specific materialization are
not modified.
### Verification
- ASAN: `NativeNullableSelectionTest.*` and benchmark scenario tests:
16/16 passed.
- ASAN: `ParquetV2NativeDecoderTest.*`: 118/118 passed.
#### Release microbenchmark
| Coverage | Legacy/fused pairs | Correctness | Regressions | Mean CPU
change | Least improvement |
|---|---:|---|---:|---:|---:|
| Full scenario matrix | 100 | Identical ranges and NULL maps | N/A
(includes negative controls) | N/A | N/A |
| Production-eligible scenarios | 30 | Identical ranges and NULL maps |
0 | -42.68% | 8.08% |
| Scenario | Repetitions | Legacy median CPU | Fused median CPU | CPU
change | Legacy CV | Fused CV |
|---|---:|---:|---:|---:|---:|---:|
| 10% selectivity / 50% NULL, fragmented | 10 | 400,220 ns | 177,905 ns
| -55.55% | 0.58% | 1.28% |
| 99% selectivity / 50% NULL, high-selectivity boundary | 10 | 687,385
ns | 460,657 ns | -32.98% | 0.60% | 1.32% |
The microbenchmark isolates nullable selection planning; it is not
presented as an end-to-end query speedup.
---
be/benchmark/parquet/AGENTS.md | 12 +-
be/benchmark/parquet/README.md | 11 +-
be/benchmark/parquet/benchmark_parquet_kernels.hpp | 202 +++++++++++++++++++++
be/benchmark/parquet/parquet_benchmark_scenarios.h | 37 ++++
.../parquet/reader/native/column_chunk_reader.cpp | 59 ++++++
.../parquet/reader/native/column_chunk_reader.h | 8 +
.../parquet/reader/native/column_reader.cpp | 43 ++++-
.../parquet/reader/native/column_reader.h | 1 +
be/src/format_v2/parquet/reader/native/common.cpp | 128 +++++++++++++
be/src/format_v2/parquet/reader/native/common.h | 11 ++
.../parquet/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 4c1d3cf4d5e..4d8c0610f17 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 156302b2423..891fd02d837 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 619d58fd8cf..de3a15c254b 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<uint16_t>* 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<uint16_t>(run_length));
+ *previous_is_null = is_null;
+}
+
+inline std::vector<uint16_t> build_nullable_runs(const NullMap& nulls) {
+ std::vector<uint16_t> 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<uint16_t>&
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<true>(&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<uint16_t>&
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<uint8_t> 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<int64_t>(state.iterations()) *
+ static_cast<int64_t>(KERNEL_ROWS));
+ state.counters["rows"] = static_cast<double>(KERNEL_ROWS);
+ state.counters["selected_rows"] =
static_cast<double>(selected.selected_rows);
+ state.counters["null_rows"] = static_cast<double>(null_plan.selected_rows);
+}
+
inline NestedSelectionOracle build_nested_selection_oracle(
const std::vector<NestedLevel>& repetition_levels,
const std::vector<NestedLevel>& 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 a9c58c15d8c..f7eddf0b11a 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<SelectionScenario>
selection_scenarios() {
return scenarios;
}
+inline std::vector<NullableSelectionScenario> nullable_selection_scenarios() {
+ std::vector<NullableSelectionScenario> 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<ReaderScenario> reader_scenarios() {
std::vector<ReaderScenario> scenarios;
std::set<std::tuple<ReaderOperation, Encoding, int, Pattern, int,
Projection, int, int,
@@ -435,4 +462,14 @@ inline std::string to_string(NestedSelectionImplementation
value) {
return "unknown";
}
+inline std::string to_string(NullableSelectionImplementation value) {
+ switch (value) {
+ case NullableSelectionImplementation::LEGACY:
+ return "legacy";
+ case NullableSelectionImplementation::FUSED:
+ return "fused";
+ }
+ return "unknown";
+}
+
} // namespace doris::parquet_benchmark
diff --git a/be/src/format_v2/parquet/reader/native/column_chunk_reader.cpp
b/be/src/format_v2/parquet/reader/native/column_chunk_reader.cpp
index f3c3af5ba99..d1677c66676 100644
--- a/be/src/format_v2/parquet/reader/native/column_chunk_reader.cpp
+++ b/be/src/format_v2/parquet/reader/native/column_chunk_reader.cpp
@@ -681,6 +681,12 @@ void remap_nullable_conversion_failures(IColumn::Filter*
conversion_failure_null
DORIS_CHECK_EQ(source, 0);
}
+Status decode_prepared_nullable_values(IColumn& column, const DataTypeSerDe&
serde,
+ Decoder& decoder, const
ParquetDecodeContext& context,
+ ParquetMaterializationState& state,
+ const NullMap& selected_nulls,
+ int64_t* materialization_time);
+
Status decode_selected_nullable_values(IColumn& column, const DataTypeSerDe&
serde,
Decoder& decoder, const
ParquetDecodeContext& context,
ParquetMaterializationState& state,
@@ -722,6 +728,16 @@ Status decode_selected_nullable_values(IColumn& column,
const DataTypeSerDe& ser
DORIS_CHECK_EQ(selection.total_values, select_vector.num_values() -
select_vector.num_nulls());
DORIS_CHECK_EQ(selected_nulls.size(),
select_vector.num_values() - select_vector.num_filtered());
+ return decode_prepared_nullable_values(column, serde, decoder, context,
state, selected_nulls,
+ materialization_time);
+}
+
+Status decode_prepared_nullable_values(IColumn& column, const DataTypeSerDe&
serde,
+ Decoder& decoder, const
ParquetDecodeContext& context,
+ ParquetMaterializationState& state,
+ const NullMap& selected_nulls,
+ int64_t* materialization_time) {
+ auto& selection = state.selection;
const size_t old_size = column.size();
SCOPED_RAW_TIMER(materialization_time);
@@ -1735,6 +1751,49 @@ Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::materialize_values(
return Status::OK();
}
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+bool ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::supports_fused_nullable_selection(
+ IColumn& column) const {
+ return visit_nullable_expandable_column(column, [](auto&) {});
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::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 IN_COLLECTION, bool OFFSET_INDEX>
bool ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::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 35bf5200d7a..a50b7e4bc08 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 @@ public:
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 f762cb0f9d3..b5ea931b29f 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<IN_COLLECTION,
OFFSET_INDEX>::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<IN_COLLECTION,
OFFSET_INDEX>::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<IN_COLLECTION,
OFFSET_INDEX>::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<IN_COLLECTION,
OFFSET_INDEX>::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<IN_COLLECTION,
OFFSET_INDEX>::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<IN_COLLECTION,
OFFSET_INDEX>::_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<IN_COLLECTION,
OFFSET_INDEX>::_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<IN_COLLECTION,
OFFSET_INDEX>::_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<IN_COLLECTION,
OFFSET_INDEX>::_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 18bcdd866f8..594f0ac0921 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 @@ private:
std::vector<uint16_t> _null_run_lengths;
std::unordered_set<size_t> _ancestor_null_indices;
std::vector<uint8_t> _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 c50488575e4..020f613c5d2 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 <algorithm>
#include <cstring>
#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<uint16_t>&
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<UInt8>(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<level_t>&
rep_levels,
std::vector<uint8_t>&
nested_filter_map_data,
std::unique_ptr<FilterMap>*
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 eb6848ee299..bd687616f22 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 @@ private:
size_t _read_index = 0;
};
+Status build_filtered_nullable_selection(const std::vector<uint16_t>&
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 2145b6ab60d..596d076c5f0 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 e21439e8855..37deb32d5bf 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<uint16_t> null_runs {2, 1, 3, 2, 2};
+ const std::vector<uint8_t> 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<uint16_t> no_nulls {10};
+ const std::vector<uint8_t> 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;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
