This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 93941b10cca [improvement](parquet) Fuse nested sparse selection
planning (#66263)
93941b10cca is described below
commit 93941b10cca7bf911d7ff09baee345f57ac606c6
Author: Gabriel <[email protected]>
AuthorDate: Thu Jul 30 20:45:28 2026 +0800
[improvement](parquet) Fuse nested sparse selection planning (#66263)
Parquet V2 repeated leaves currently rebuild a parent-row filter into a
level-entry filter, scan definition levels into null runs plus an ancestor-null
hash set, build the leaf selection, and scan the levels again to compact
surviving parent shapes. This fixed O(level entries) work is especially visible
when only 1% or 10% of parent rows survive.
This change adds a single-pass nested selection builder. It maps parent
filters to repeated entries, classifies ancestor and leaf nulls, builds the
decoder selection, appends selected nulls, and compacts repetition/definition
levels in place. Ancestor-null placeholders remain in the surviving parent
shape but do not consume leaf selection entries. Unfiltered and fully filtered
paths keep their existing behavior.
---
be/benchmark/parquet/AGENTS.md | 19 +-
be/benchmark/parquet/README.md | 34 ++-
be/benchmark/parquet/benchmark_parquet_kernels.hpp | 268 ++++++++++++++++++++-
be/benchmark/parquet/parquet_benchmark_scenarios.h | 26 +-
.../parquet/reader/native/column_reader.cpp | 39 +--
be/src/format_v2/parquet/reader/native/common.cpp | 84 +++++++
be/src/format_v2/parquet/reader/native/common.h | 8 +
.../parquet/parquet_benchmark_scenarios_test.cpp | 20 +-
.../parquet/parquet_reader_control_test.cpp | 75 ++++++
be/test/format_v2/parquet/parquet_reader_test.cpp | 252 +++++++++++++++++++
10 files changed, 789 insertions(+), 36 deletions(-)
diff --git a/be/benchmark/parquet/AGENTS.md b/be/benchmark/parquet/AGENTS.md
index ba52658a966..6c2fb8c85df 100644
--- a/be/benchmark/parquet/AGENTS.md
+++ b/be/benchmark/parquet/AGENTS.md
@@ -43,10 +43,10 @@ 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 80
+ | grep -c '^ParquetKernel/' # currently 92
be/output/lib/benchmark_test --benchmark_list_tests \
- | grep -c '^ParquetReader/' # currently 152
+ | grep -c '^ParquetReader/' # currently 167
```
When running the binary directly from `be/build_RELEASE/bin`, make sure the
JVM and third-party
@@ -120,13 +120,16 @@ cache to manufacture a cold run.
| DELTA_LENGTH_BYTE_ARRAY | BYTE_ARRAY |
| DELTA_BYTE_ARRAY | BYTE_ARRAY |
-`ParquetKernel` contains 80 cases across five SIMD-sensitive stages:
BYTE_STREAM_SPLIT,
-DELTA_PREFIX_SUM, DICTIONARY_GATHER, NULLABLE_EXPAND, and RAW_PREDICATE. It
covers the applicable
-four- and eight-byte types, three dictionary working-set sizes, 0% through 90%
null rates with both
-placement patterns, and 0% through 100% raw-predicate selectivities.
+`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
+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.
`ParquetReader` deliberately uses a single-variable matrix rather than a
Cartesian product. After
-deduplication it contains 152 cases covering:
+deduplication it contains 167 cases covering:
- operations: open-to-first-block, full scan, predicate scan, complex residual
scan, limit 1, and
limit 1000;
@@ -299,7 +302,7 @@ be simulated by silently changing the local reader
benchmark.
## Current validation record
-The current expected registration counts are 228 decoder, 80 kernel, and 152
reader cases. A smoke
+The current expected registration counts are 228 decoder, 92 kernel, and 167
reader cases. A smoke
run is an execution record only, not a reviewed performance baseline, because
repetitions, host
isolation, warmups, cache control, `perf` data, variance, and before/after
comparison are not
collected.
diff --git a/be/benchmark/parquet/README.md b/be/benchmark/parquet/README.md
index e6eab6362ce..fcbed4a548b 100644
--- a/be/benchmark/parquet/README.md
+++ b/be/benchmark/parquet/README.md
@@ -39,13 +39,16 @@ be/output/lib/benchmark_test \
## SIMD kernel cases
-`ParquetKernel` isolates the five SIMD-sensitive stages from reader setup and
virtual consumer
+`ParquetKernel` isolates six decode and selection stages from reader setup and
virtual consumer
overhead: byte-stream-split transpose, delta prefix sum, numeric dictionary
gather, nullable
-expansion, and raw predicate evaluation. 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. Dictionary
gather uses 32-,
-4,096-, and 262,144-entry working sets to separate cache-resident and
cache-miss-dominated
-behavior.
+expansion, 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.
+Dictionary gather uses 32-, 4,096-, and 262,144-entry working sets to separate
cache-resident and
+cache-miss-dominated behavior.
```shell
be/output/lib/benchmark_test \
@@ -53,6 +56,25 @@ be/output/lib/benchmark_test \
--benchmark_min_time=0.1s
```
+For a reproducible nested-selection comparison, build once and run the two
implementations from
+that same binary in ABBA order. Pin every command to the same otherwise-idle
CPU:
+
+```shell
+taskset -c 8 be/output/lib/benchmark_test \
+ --benchmark_filter='^ParquetKernel/nested_selection/.*/impl_legacy$' \
+ --benchmark_min_time=1s --benchmark_repetitions=10 \
+ --benchmark_report_aggregates_only=true \
+ --benchmark_out=nested-legacy-a1.json --benchmark_out_format=json
+
+taskset -c 8 be/output/lib/benchmark_test \
+ --benchmark_filter='^ParquetKernel/nested_selection/.*/impl_fused$' \
+ --benchmark_min_time=1s --benchmark_repetitions=10 \
+ --benchmark_report_aggregates_only=true \
+ --benchmark_out=nested-fused-b1.json --benchmark_out_format=json
+
+# Repeat fused as B2, then legacy as A2, changing only --benchmark_out.
+```
+
## Local reader cases
`ParquetReader` measures local open-to-first-block, full scan, predicate scan,
complex residual
diff --git a/be/benchmark/parquet/benchmark_parquet_kernels.hpp
b/be/benchmark/parquet/benchmark_parquet_kernels.hpp
index e064e69f64f..619d58fd8cf 100644
--- a/be/benchmark/parquet/benchmark_parquet_kernels.hpp
+++ b/be/benchmark/parquet/benchmark_parquet_kernels.hpp
@@ -20,12 +20,15 @@
#include <benchmark/benchmark.h>
#include <algorithm>
+#include <climits>
#include <cstddef>
#include <cstdint>
#include <string>
#include <type_traits>
+#include <unordered_set>
#include <vector>
+#include "format_v2/parquet/reader/native/common.h"
#include "parquet_benchmark_scenarios.h"
#include "util/byte_stream_split.h"
#include "util/simd/parquet_kernels.h"
@@ -34,6 +37,247 @@ namespace doris::parquet_benchmark {
namespace detail {
constexpr size_t KERNEL_ROWS = 1UL << 16;
+constexpr size_t NESTED_VALUES_PER_ROW = 8;
+
+using NestedReadType =
format::parquet::native::ColumnSelectVector::DataReadType;
+using NestedLevel = format::parquet::native::level_t;
+
+struct NestedSelectionOracle {
+ std::vector<NestedLevel> repetition_levels;
+ std::vector<NestedLevel> definition_levels;
+ NullMap selected_nulls;
+ std::vector<NestedReadType> reads;
+ size_t ancestor_null_count = 0;
+ size_t filtered_count = 0;
+};
+
+struct NestedSelectionScratch {
+ std::vector<NestedLevel> repetition_levels;
+ std::vector<NestedLevel> definition_levels;
+ std::vector<uint8_t> nested_filter_data;
+ std::vector<uint16_t> null_runs;
+ std::unordered_set<size_t> ancestor_null_indices;
+ format::parquet::native::FilterMap nested_filter;
+ format::parquet::native::ColumnSelectVector selection;
+ NullMap selected_nulls;
+ size_t ancestor_null_count = 0;
+};
+
+inline NestedSelectionOracle build_nested_selection_oracle(
+ const std::vector<NestedLevel>& repetition_levels,
+ const std::vector<NestedLevel>& definition_levels,
+ const std::vector<uint8_t>& parent_filter_data) {
+ // Derive expectations from source levels so validation cannot inherit a
mistake from either
+ // measured implementation.
+ NestedSelectionOracle oracle;
+ size_t parent = 0;
+ for (size_t level = 0; level < repetition_levels.size(); ++level) {
+ if (level != 0 && repetition_levels[level] == 0) {
+ ++parent;
+ }
+ const bool selected = parent_filter_data[parent] != 0;
+ if (selected) {
+ oracle.repetition_levels.push_back(repetition_levels[level]);
+ oracle.definition_levels.push_back(definition_levels[level]);
+ }
+ if (definition_levels[level] < 2) {
+ ++oracle.ancestor_null_count;
+ continue;
+ }
+ const bool is_null = definition_levels[level] < 3;
+ if (selected) {
+ oracle.selected_nulls.push_back(static_cast<UInt8>(is_null));
+ oracle.reads.push_back(is_null ? NestedReadType::NULL_DATA :
NestedReadType::CONTENT);
+ } else {
+ ++oracle.filtered_count;
+ oracle.reads.push_back(is_null ? NestedReadType::FILTERED_NULL
+ : NestedReadType::FILTERED_CONTENT);
+ }
+ }
+ return oracle;
+}
+
+inline void append_nested_null_run(std::vector<uint16_t>* null_runs, bool
is_null,
+ size_t run_length, bool* previous_is_null) {
+ if (*previous_is_null == is_null && USHRT_MAX - null_runs->back() >=
run_length) {
+ null_runs->back() += static_cast<uint16_t>(run_length);
+ return;
+ }
+ if (!(*previous_is_null ^ is_null)) {
+ null_runs->push_back(0);
+ }
+ while (run_length > USHRT_MAX) {
+ null_runs->push_back(USHRT_MAX);
+ null_runs->push_back(0);
+ run_length -= USHRT_MAX;
+ }
+ null_runs->push_back(static_cast<uint16_t>(run_length));
+ *previous_is_null = is_null;
+}
+
+inline Status run_legacy_nested_selection(NestedSelectionScratch* scratch,
+ format::parquet::native::FilterMap*
parent_filter) {
+ // Keep the pre-fusion passes selectable in the same binary so comparisons
share compiler,
+ // fixtures, and process state.
+ scratch->nested_filter_data.resize(scratch->repetition_levels.size());
+ size_t parent = 0;
+ for (size_t level = 0; level < scratch->repetition_levels.size(); ++level)
{
+ if (level != 0 && scratch->repetition_levels[level] == 0) {
+ ++parent;
+ }
+ scratch->nested_filter_data[level] =
parent_filter->filter_map_data()[parent];
+ }
+
RETURN_IF_ERROR(scratch->nested_filter.init(scratch->nested_filter_data.data(),
+
scratch->nested_filter_data.size(), false));
+
+ scratch->null_runs.clear();
+ scratch->null_runs.push_back(0);
+ scratch->ancestor_null_indices.clear();
+ bool previous_is_null = false;
+ size_t level = 0;
+ while (level < scratch->definition_levels.size()) {
+ const NestedLevel definition_level = scratch->definition_levels[level];
+ const size_t run_start = level++;
+ while (level < scratch->definition_levels.size() &&
+ scratch->definition_levels[level] == definition_level) {
+ ++level;
+ }
+ const size_t run_length = level - run_start;
+ if (definition_level < 2) {
+ for (size_t index = run_start; index < level; ++index) {
+ scratch->ancestor_null_indices.insert(index);
+ }
+ continue;
+ }
+ append_nested_null_run(&scratch->null_runs, definition_level < 3,
run_length,
+ &previous_is_null);
+ }
+ scratch->ancestor_null_count = scratch->ancestor_null_indices.size();
+ RETURN_IF_ERROR(scratch->selection.init(
+ scratch->null_runs, scratch->repetition_levels.size() -
scratch->ancestor_null_count,
+ &scratch->selected_nulls, &scratch->nested_filter, 0,
&scratch->ancestor_null_indices));
+
+ size_t output_level = 0;
+ for (size_t input_level = 0; input_level <
scratch->repetition_levels.size(); ++input_level) {
+ if (scratch->nested_filter_data[input_level] != 0) {
+ scratch->repetition_levels[output_level] =
scratch->repetition_levels[input_level];
+ scratch->definition_levels[output_level] =
scratch->definition_levels[input_level];
+ ++output_level;
+ }
+ }
+ scratch->repetition_levels.resize(output_level);
+ scratch->definition_levels.resize(output_level);
+ return Status::OK();
+}
+
+inline Status run_nested_selection_once(NestedSelectionScratch* scratch,
+ format::parquet::native::FilterMap*
parent_filter,
+ NestedSelectionImplementation
implementation) {
+ if (implementation == NestedSelectionImplementation::LEGACY) {
+ return run_legacy_nested_selection(scratch, parent_filter);
+ }
+ return scratch->selection.init_nested(
+ &scratch->repetition_levels, &scratch->definition_levels, 0,
+ /*repeated_parent_def_level=*/2, /*definition_level=*/3,
&scratch->selected_nulls,
+ parent_filter, 0, &scratch->ancestor_null_count);
+}
+
+inline Status validate_nested_selection(NestedSelectionScratch& scratch,
+ const NestedSelectionOracle& oracle) {
+ if (scratch.repetition_levels != oracle.repetition_levels ||
+ scratch.definition_levels != oracle.definition_levels ||
+ scratch.selected_nulls != oracle.selected_nulls ||
+ scratch.ancestor_null_count != oracle.ancestor_null_count ||
+ scratch.selection.num_filtered() != oracle.filtered_count) {
+ return Status::InternalError("nested selection differs from
independent oracle");
+ }
+ std::vector<NestedReadType> actual_reads;
+ NestedReadType type;
+ size_t run_length = 0;
+ while ((run_length = scratch.selection.get_next_run<true>(&type)) != 0) {
+ actual_reads.insert(actual_reads.end(), run_length, type);
+ }
+ if (actual_reads != oracle.reads) {
+ return Status::InternalError("nested selection read sequence differs
from oracle");
+ }
+ return Status::OK();
+}
+
+inline void run_nested_selection_kernel(benchmark::State& state, const
KernelScenario& scenario) {
+ using format::parquet::native::ColumnSelectVector;
+ using format::parquet::native::FilterMap;
+ using format::parquet::native::level_t;
+
+ std::vector<level_t> source_repetition_levels;
+ std::vector<level_t> source_definition_levels;
+ source_repetition_levels.reserve(KERNEL_ROWS * NESTED_VALUES_PER_ROW);
+ source_definition_levels.reserve(KERNEL_ROWS * NESTED_VALUES_PER_ROW);
+ for (size_t row = 0; row < KERNEL_ROWS; ++row) {
+ if (row % 10 == 0) {
+ source_repetition_levels.push_back(0);
+ source_definition_levels.push_back(0);
+ continue;
+ }
+ for (size_t value = 0; value < NESTED_VALUES_PER_ROW; ++value) {
+ source_repetition_levels.push_back(value == 0 ? 0 : 1);
+ source_definition_levels.push_back((row + value) % 10 == 0 ? 2 :
3);
+ }
+ }
+
+ const auto parent_selection =
+ make_selection_plan(KERNEL_ROWS, scenario.selectivity_percent,
scenario.pattern);
+ std::vector<uint8_t> parent_filter_data(KERNEL_ROWS, 0);
+ visit_selected_rows(parent_selection,
+ [&](size_t row) { parent_filter_data[row] = uint8_t
{1}; });
+ FilterMap parent_filter;
+ auto status = parent_filter.init(parent_filter_data.data(),
parent_filter_data.size(), false);
+ if (!status.ok()) {
+ state.SkipWithError(status.to_string().c_str());
+ return;
+ }
+
+ const auto oracle = build_nested_selection_oracle(source_repetition_levels,
+
source_definition_levels, parent_filter_data);
+ NestedSelectionScratch scratch;
+ scratch.repetition_levels = source_repetition_levels;
+ scratch.definition_levels = source_definition_levels;
+ status = run_nested_selection_once(&scratch, &parent_filter,
scenario.nested_implementation);
+ if (status.ok()) {
+ status = validate_nested_selection(scratch, oracle);
+ }
+ if (!status.ok()) {
+ state.SkipWithError(status.to_string().c_str());
+ return;
+ }
+
+ for (auto _ : state) {
+ state.PauseTiming();
+ scratch.repetition_levels = source_repetition_levels;
+ scratch.definition_levels = source_definition_levels;
+ scratch.selected_nulls.clear();
+ state.ResumeTiming();
+ status =
+ run_nested_selection_once(&scratch, &parent_filter,
scenario.nested_implementation);
+ if (!status.ok()) {
+ state.SkipWithError(status.to_string().c_str());
+ return;
+ }
+ auto* compacted_levels = scratch.repetition_levels.data();
+ size_t filtered_values = scratch.selection.num_filtered();
+ benchmark::DoNotOptimize(compacted_levels);
+ benchmark::DoNotOptimize(filtered_values);
+ benchmark::ClobberMemory();
+ }
+
+ state.SetItemsProcessed(static_cast<int64_t>(state.iterations()) *
+
static_cast<int64_t>(source_repetition_levels.size()));
+ state.SetBytesProcessed(
+ static_cast<int64_t>(state.iterations()) *
+ static_cast<int64_t>(source_repetition_levels.size() * 2 *
sizeof(level_t)));
+ state.counters["parent_rows"] = static_cast<double>(KERNEL_ROWS);
+ state.counters["selected_parent_rows"] =
static_cast<double>(parent_selection.selected_rows);
+ state.counters["level_entries"] =
static_cast<double>(source_repetition_levels.size());
+}
inline void decode_byte_stream_split(const uint8_t* src, size_t width, size_t
offset,
size_t num_values, size_t stride,
uint8_t* dest) {
@@ -155,6 +399,9 @@ void run_kernel(benchmark::State& state, const
KernelScenario& scenario) {
}
}
break;
+ case Kernel::NESTED_SELECTION:
+ state.SkipWithError("nested selection uses its dedicated level
kernel");
+ return;
}
for (auto _ : state) {
@@ -193,6 +440,8 @@ void run_kernel(benchmark::State& state, const
KernelScenario& scenario) {
simd::raw_compare(reinterpret_cast<const uint8_t*>(input.data()),
input.size(), literal,
simd::RawComparisonOp::LT, matches.data());
break;
+ case Kernel::NESTED_SELECTION:
+ break;
}
benchmark::ClobberMemory();
}
@@ -207,13 +456,20 @@ void run_kernel(benchmark::State& state, const
KernelScenario& scenario) {
inline bool register_kernel_benchmarks() {
for (const auto& scenario : kernel_scenarios()) {
- const std::string name = "ParquetKernel/" + to_string(scenario.kernel)
+ "/" +
- to_string(scenario.value_type) + "/sel_" +
- std::to_string(scenario.selectivity_percent)
+ "/null_" +
- std::to_string(scenario.null_percent) + "/" +
- to_string(scenario.pattern) + "/dict_" +
- std::to_string(scenario.dictionary_entries);
+ std::string name = "ParquetKernel/" + to_string(scenario.kernel) + "/"
+
+ to_string(scenario.value_type) + "/sel_" +
+ std::to_string(scenario.selectivity_percent) +
"/null_" +
+ std::to_string(scenario.null_percent) + "/" +
+ to_string(scenario.pattern) + "/dict_" +
+ std::to_string(scenario.dictionary_entries);
+ if (scenario.kernel == Kernel::NESTED_SELECTION) {
+ name += "/impl_" + to_string(scenario.nested_implementation);
+ }
benchmark::RegisterBenchmark(name.c_str(), [=](benchmark::State&
state) {
+ if (scenario.kernel == Kernel::NESTED_SELECTION) {
+ run_nested_selection_kernel(state, scenario);
+ return;
+ }
switch (scenario.value_type) {
case ValueType::INT32:
run_kernel<int32_t>(state, scenario);
diff --git a/be/benchmark/parquet/parquet_benchmark_scenarios.h
b/be/benchmark/parquet/parquet_benchmark_scenarios.h
index 900cb29a583..d2655b3380e 100644
--- a/be/benchmark/parquet/parquet_benchmark_scenarios.h
+++ b/be/benchmark/parquet/parquet_benchmark_scenarios.h
@@ -50,8 +50,10 @@ enum class Kernel {
DELTA_PREFIX_SUM,
DICTIONARY_GATHER,
NULLABLE_EXPAND,
- RAW_PREDICATE
+ RAW_PREDICATE,
+ NESTED_SELECTION
};
+enum class NestedSelectionImplementation { LEGACY, FUSED };
struct DecoderScenario {
Encoding encoding;
@@ -77,6 +79,7 @@ struct KernelScenario {
int null_percent;
Pattern pattern;
size_t dictionary_entries;
+ NestedSelectionImplementation nested_implementation =
NestedSelectionImplementation::FUSED;
};
struct SelectionRange {
@@ -141,6 +144,15 @@ inline std::vector<KernelScenario> kernel_scenarios() {
{Kernel::RAW_PREDICATE, value_type, selectivity, 0,
Pattern::ALTERNATING, 256});
}
}
+ for (const int selectivity : {1, 10, 50}) {
+ for (const auto pattern : {Pattern::CLUSTERED, Pattern::ALTERNATING}) {
+ for (const auto implementation :
+ {NestedSelectionImplementation::LEGACY,
NestedSelectionImplementation::FUSED}) {
+ scenarios.push_back({Kernel::NESTED_SELECTION,
ValueType::INT32, selectivity, 10,
+ pattern, 256, implementation});
+ }
+ }
+ }
return scenarios;
}
@@ -374,6 +386,18 @@ inline std::string to_string(Kernel value) {
return "nullable_expand";
case Kernel::RAW_PREDICATE:
return "raw_predicate";
+ case Kernel::NESTED_SELECTION:
+ return "nested_selection";
+ }
+ return "unknown";
+}
+
+inline std::string to_string(NestedSelectionImplementation value) {
+ switch (value) {
+ case NestedSelectionImplementation::LEGACY:
+ return "legacy";
+ case NestedSelectionImplementation::FUSED:
+ return "fused";
}
return "unknown";
}
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 e66495dfd7c..f762cb0f9d3 100644
--- a/be/src/format_v2/parquet/reader/native/column_reader.cpp
+++ b/be/src/format_v2/parquet/reader/native/column_reader.cpp
@@ -1066,7 +1066,15 @@ Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::_read_nested_column(
auto read_and_fill_data = [&](size_t before_rep_level_sz, size_t
filter_map_index) {
RETURN_IF_ERROR(_chunk_reader->fill_def(_def_levels));
- if (filter_map.has_filter()) {
+ const bool fuse_nested_selection = filter_map.has_filter() &&
!filter_map.filter_all();
+ size_t ancestor_null_count = 0;
+ if (fuse_nested_selection) {
+ SCOPED_RAW_TIMER(&_decode_null_map_time);
+ RETURN_IF_ERROR(_select_vector.init_nested(
+ &_rep_levels, &_def_levels, before_rep_level_sz,
+ _field_schema->repeated_parent_def_level,
_field_schema->definition_level,
+ map_data_column, &filter_map, filter_map_index,
&ancestor_null_count));
+ } else if (filter_map.has_filter()) {
RETURN_IF_ERROR(gen_filter_map(filter_map, filter_map_index,
before_rep_level_sz,
_rep_levels.size(),
_nested_filter_map_data,
&_nested_filter_map));
@@ -1075,17 +1083,20 @@ Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::_read_nested_column(
nullptr, _rep_levels.size() - before_rep_level_sz, false));
}
- _null_run_lengths.clear();
- _ancestor_null_indices.clear();
- RETURN_IF_ERROR(gen_nested_null_map(before_rep_level_sz,
_rep_levels.size(),
- _null_run_lengths,
_ancestor_null_indices));
+ if (!fuse_nested_selection) {
+ _null_run_lengths.clear();
+ _ancestor_null_indices.clear();
+ RETURN_IF_ERROR(gen_nested_null_map(before_rep_level_sz,
_rep_levels.size(),
+ _null_run_lengths,
_ancestor_null_indices));
+ ancestor_null_count = _ancestor_null_indices.size();
- {
- SCOPED_RAW_TIMER(&_decode_null_map_time);
- RETURN_IF_ERROR(_select_vector.init(
- _null_run_lengths,
- _rep_levels.size() - before_rep_level_sz -
_ancestor_null_indices.size(),
- map_data_column, &_nested_filter_map, 0,
&_ancestor_null_indices));
+ {
+ SCOPED_RAW_TIMER(&_decode_null_map_time);
+ RETURN_IF_ERROR(_select_vector.init(
+ _null_run_lengths,
+ _rep_levels.size() - before_rep_level_sz -
ancestor_null_count,
+ map_data_column, &_nested_filter_map, 0,
&_ancestor_null_indices));
+ }
}
DORIS_CHECK(_serde != nullptr);
@@ -1104,10 +1115,10 @@ Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::_read_nested_column(
map_data_column,
materialization_start_row);
}
RETURN_IF_ERROR(status);
- if (!_ancestor_null_indices.empty()) {
-
RETURN_IF_ERROR(_chunk_reader->skip_values(_ancestor_null_indices.size(),
false));
+ if (ancestor_null_count != 0) {
+ RETURN_IF_ERROR(_chunk_reader->skip_values(ancestor_null_count,
false));
}
- if (filter_map.has_filter()) {
+ if (filter_map.has_filter() && !fuse_nested_selection) {
auto new_rep_sz = before_rep_level_sz;
for (size_t idx = before_rep_level_sz; idx < _rep_levels.size();
idx++) {
if (_nested_filter_map_data[idx - before_rep_level_sz]) {
diff --git a/be/src/format_v2/parquet/reader/native/common.cpp
b/be/src/format_v2/parquet/reader/native/common.cpp
index f048d2b878e..c50488575e4 100644
--- a/be/src/format_v2/parquet/reader/native/common.cpp
+++ b/be/src/format_v2/parquet/reader/native/common.cpp
@@ -193,4 +193,88 @@ Status ColumnSelectVector::init(const
std::vector<uint16_t>& run_length_null_map
return Status::OK();
}
+void ColumnSelectVector::reset(bool has_filter) {
+ _data_map.clear();
+ _run_length_null_map = nullptr;
+ _has_filter = has_filter;
+ _num_values = 0;
+ _num_nulls = 0;
+ _num_filtered = 0;
+ _read_index = 0;
+}
+
+Status ColumnSelectVector::init_nested(std::vector<level_t>* repetition_levels,
+ std::vector<level_t>* definition_levels,
+ size_t level_start_index, level_t
repeated_parent_def_level,
+ level_t definition_level, NullMap*
null_map,
+ FilterMap* parent_filter, size_t
filter_map_index,
+ size_t* ancestor_null_count) {
+ if (repetition_levels == nullptr || definition_levels == nullptr ||
parent_filter == nullptr ||
+ ancestor_null_count == nullptr) {
+ return Status::InvalidArgument("Nested selection requires non-null
input state");
+ }
+ if (repetition_levels->size() != definition_levels->size() ||
+ level_start_index > repetition_levels->size()) {
+ return Status::InvalidArgument(
+ "Nested selection has invalid level bounds: repetition={},
definition={}, start={}",
+ repetition_levels->size(), definition_levels->size(),
level_start_index);
+ }
+ if (!parent_filter->has_filter() || parent_filter->filter_all()) {
+ return Status::InvalidArgument("Nested selection requires a partial
parent filter");
+ }
+ if (level_start_index == repetition_levels->size()) {
+ reset(true);
+ *ancestor_null_count = 0;
+ return Status::OK();
+ }
+ if (filter_map_index >= parent_filter->filter_map_size()) {
+ return Status::InvalidArgument("Nested filter row {} exceeds filter
map size {}",
+ filter_map_index,
parent_filter->filter_map_size());
+ }
+ reset(true);
+ _data_map.reserve(repetition_levels->size() - level_start_index);
+ *ancestor_null_count = 0;
+ size_t current_parent = filter_map_index;
+ size_t output_level = level_start_index;
+ for (size_t input_level = level_start_index; input_level <
repetition_levels->size();
+ ++input_level) {
+ if (input_level != level_start_index &&
(*repetition_levels)[input_level] == 0) {
+ ++current_parent;
+ if (current_parent >= parent_filter->filter_map_size()) {
+ return Status::InvalidArgument("Nested filter row {} exceeds
filter map size {}",
+ current_parent,
parent_filter->filter_map_size());
+ }
+ }
+ const bool selected = parent_filter->filter_map_data()[current_parent]
!= 0;
+ if (selected) {
+ (*repetition_levels)[output_level] =
(*repetition_levels)[input_level];
+ (*definition_levels)[output_level] =
(*definition_levels)[input_level];
+ ++output_level;
+ }
+ // An ancestor-null placeholder belongs to the surviving parent's
level shape, but it must
+ // never consume a leaf selection entry because no physical leaf value
exists for it.
+ const level_t def_level = (*definition_levels)[input_level];
+ if (def_level < repeated_parent_def_level) {
+ ++*ancestor_null_count;
+ continue;
+ }
+ const bool is_null = def_level < definition_level;
+ ++_num_values;
+ _num_nulls += is_null;
+ if (!selected) {
+ ++_num_filtered;
+ } else if (null_map != nullptr) {
+ null_map->push_back(static_cast<UInt8>(is_null));
+ }
+ DataReadType read_type = is_null ? FILTERED_NULL : FILTERED_CONTENT;
+ if (selected) {
+ read_type = is_null ? NULL_DATA : CONTENT;
+ }
+ _data_map.push_back(read_type);
+ }
+ repetition_levels->resize(output_level);
+ definition_levels->resize(output_level);
+ return Status::OK();
+}
+
} // namespace doris::format::parquet::native
diff --git a/be/src/format_v2/parquet/reader/native/common.h
b/be/src/format_v2/parquet/reader/native/common.h
index 4ad204da2ff..eb6848ee299 100644
--- a/be/src/format_v2/parquet/reader/native/common.h
+++ b/be/src/format_v2/parquet/reader/native/common.h
@@ -65,6 +65,12 @@ public:
NullMap* null_map, FilterMap* filter_map, size_t
filter_map_index,
const std::unordered_set<size_t>* skipped_indices = nullptr);
+ Status init_nested(std::vector<level_t>* repetition_levels,
+ std::vector<level_t>* definition_levels, size_t
level_start_index,
+ level_t repeated_parent_def_level, level_t
definition_level,
+ NullMap* null_map, FilterMap* parent_filter, size_t
filter_map_index,
+ size_t* ancestor_null_count);
+
size_t num_values() const { return _num_values; }
size_t num_nulls() const { return _num_nulls; }
size_t num_filtered() const { return _num_filtered; }
@@ -99,6 +105,8 @@ public:
}
private:
+ void reset(bool has_filter);
+
std::vector<DataReadType> _data_map;
const std::vector<uint16_t>* _run_length_null_map = nullptr;
bool _has_filter = false;
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 89e1f56ac3e..72ccbf44844 100644
--- a/be/test/format_v2/parquet/parquet_benchmark_scenarios_test.cpp
+++ b/be/test/format_v2/parquet/parquet_benchmark_scenarios_test.cpp
@@ -66,7 +66,7 @@ TEST(ParquetBenchmarkScenariosTest,
DecoderMatrixCoversNativeEncodingAndTypeFami
TEST(ParquetBenchmarkScenariosTest,
KernelMatrixCoversEverySimdStageAndBoundaryShape) {
const auto scenarios = kernel_scenarios();
- EXPECT_EQ(scenarios.size(), size_t {80});
+ EXPECT_EQ(scenarios.size(), size_t {92});
const std::map<Kernel, std::vector<ValueType>> expected_types {
{Kernel::BYTE_STREAM_SPLIT, {ValueType::FLOAT, ValueType::DOUBLE}},
{Kernel::DELTA_PREFIX_SUM, {ValueType::INT32, ValueType::INT64}},
@@ -76,6 +76,7 @@ TEST(ParquetBenchmarkScenariosTest,
KernelMatrixCoversEverySimdStageAndBoundaryS
{ValueType::INT32, ValueType::INT64, ValueType::FLOAT,
ValueType::DOUBLE}},
{Kernel::RAW_PREDICATE,
{ValueType::INT32, ValueType::INT64, ValueType::FLOAT,
ValueType::DOUBLE}},
+ {Kernel::NESTED_SELECTION, {ValueType::INT32}},
};
for (const auto& [kernel, value_types] : expected_types) {
for (const auto value_type : value_types) {
@@ -109,6 +110,23 @@ TEST(ParquetBenchmarkScenariosTest,
KernelMatrixCoversEverySimdStageAndBoundaryS
}
}
+TEST(ParquetBenchmarkScenariosTest,
NestedSelectionCoversSparseParentSurvivors) {
+ const auto scenarios = kernel_scenarios();
+ for (const int selectivity : {1, 10, 50}) {
+ for (const auto pattern : {Pattern::CLUSTERED, Pattern::ALTERNATING}) {
+ for (const auto implementation :
+ {NestedSelectionImplementation::LEGACY,
NestedSelectionImplementation::FUSED}) {
+ EXPECT_TRUE(std::ranges::any_of(scenarios, [&](const
KernelScenario& scenario) {
+ return scenario.kernel == Kernel::NESTED_SELECTION &&
+ scenario.selectivity_percent == selectivity &&
+ scenario.null_percent == 10 && scenario.pattern ==
pattern &&
+ scenario.nested_implementation == implementation;
+ })) << "missing nested sparse-selection implementation";
+ }
+ }
+ }
+}
+
TEST(ParquetBenchmarkScenariosTest,
ReaderMatrixCoversNullableSparseAndProjectionAxes) {
const auto scenarios = reader_scenarios();
// Keep the exact count aligned with the upstream complex-residual
scenario retained by rebase.
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 9de951ec5e4..0c70ddaf01e 100644
--- a/be/test/format_v2/parquet/parquet_reader_control_test.cpp
+++ b/be/test/format_v2/parquet/parquet_reader_control_test.cpp
@@ -31,6 +31,7 @@
#include "format_v2/parquet/parquet_scan.h"
#include "format_v2/parquet/reader/column_reader.h"
#include "format_v2/parquet/reader/global_rowid_column_reader.h"
+#include "format_v2/parquet/reader/native/common.h"
#include "format_v2/parquet/reader/row_position_column_reader.h"
#include "format_v2/parquet/selection_vector.h"
#include "storage/utils.h"
@@ -184,6 +185,80 @@ TEST(SelectionVectorTest,
IdentitySelectionDoesNotMaterializeFilter) {
EXPECT_EQ(filter, nullptr);
}
+TEST(NativeNestedSelectionTest,
BuildsSelectionAndCompactsSurvivingParentLevels) {
+ using native::ColumnSelectVector;
+ using native::FilterMap;
+ using native::level_t;
+
+ std::vector<level_t> repetition_levels {0, 1, 1, 0, 0, 1};
+ std::vector<level_t> definition_levels {3, 2, 1, 3, 0, 3};
+ std::vector<uint8_t> parent_filter_data {1, 0, 1};
+ FilterMap parent_filter;
+ ASSERT_TRUE(
+ parent_filter.init(parent_filter_data.data(),
parent_filter_data.size(), false).ok());
+
+ ColumnSelectVector selection;
+ NullMap selected_nulls;
+ size_t ancestor_null_count = 0;
+ ASSERT_TRUE(selection
+ .init_nested(&repetition_levels, &definition_levels, 0,
+ /*repeated_parent_def_level=*/2,
+ /*definition_level=*/3, &selected_nulls,
&parent_filter, 0,
+ &ancestor_null_count)
+ .ok());
+
+ EXPECT_EQ(ancestor_null_count, 2);
+ EXPECT_EQ(selection.num_values(), 4);
+ EXPECT_EQ(selection.num_nulls(), 1);
+ EXPECT_EQ(selection.num_filtered(), 1);
+ EXPECT_EQ(selected_nulls, NullMap({0, 1, 0}));
+ EXPECT_EQ(repetition_levels, (std::vector<level_t> {0, 1, 1, 0, 1}));
+ EXPECT_EQ(definition_levels, (std::vector<level_t> {3, 2, 1, 0, 3}));
+
+ ColumnSelectVector::DataReadType type;
+ EXPECT_EQ(selection.get_next_run<true>(&type), 1);
+ EXPECT_EQ(type, ColumnSelectVector::CONTENT);
+ EXPECT_EQ(selection.get_next_run<true>(&type), 1);
+ EXPECT_EQ(type, ColumnSelectVector::NULL_DATA);
+ EXPECT_EQ(selection.get_next_run<true>(&type), 1);
+ EXPECT_EQ(type, ColumnSelectVector::FILTERED_CONTENT);
+ EXPECT_EQ(selection.get_next_run<true>(&type), 1);
+ EXPECT_EQ(type, ColumnSelectVector::CONTENT);
+ EXPECT_EQ(selection.get_next_run<true>(&type), 0);
+}
+
+TEST(NativeNestedSelectionTest, PreservesPriorLevelsAcrossPageContinuation) {
+ using native::ColumnSelectVector;
+ using native::FilterMap;
+ using native::level_t;
+
+ std::vector<level_t> repetition_levels {0, 1, 1, 0, 1};
+ std::vector<level_t> definition_levels {3, 3, 2, 3, 1};
+ std::vector<uint8_t> parent_filter_data {1, 0};
+ FilterMap parent_filter;
+ ASSERT_TRUE(
+ parent_filter.init(parent_filter_data.data(),
parent_filter_data.size(), false).ok());
+
+ ColumnSelectVector selection;
+ NullMap selected_nulls;
+ size_t ancestor_null_count = 0;
+ ASSERT_TRUE(selection
+ .init_nested(&repetition_levels, &definition_levels,
+ /*level_start_index=*/2,
+ /*repeated_parent_def_level=*/2,
+ /*definition_level=*/3, &selected_nulls,
&parent_filter, 0,
+ &ancestor_null_count)
+ .ok());
+
+ EXPECT_EQ(ancestor_null_count, 1);
+ EXPECT_EQ(selection.num_values(), 2);
+ EXPECT_EQ(selection.num_nulls(), 1);
+ EXPECT_EQ(selection.num_filtered(), 1);
+ EXPECT_EQ(selected_nulls, NullMap({1}));
+ EXPECT_EQ(repetition_levels, (std::vector<level_t> {0, 1, 1}));
+ EXPECT_EQ(definition_levels, (std::vector<level_t> {3, 3, 2}));
+}
+
TEST(ParquetColumnReaderControlTest, BaseSelectUsesSkipReadRanges) {
CursorColumnReader reader;
SelectionVector selection(3);
diff --git a/be/test/format_v2/parquet/parquet_reader_test.cpp
b/be/test/format_v2/parquet/parquet_reader_test.cpp
index af0897d6ef2..bc73055ba88 100644
--- a/be/test/format_v2/parquet/parquet_reader_test.cpp
+++ b/be/test/format_v2/parquet/parquet_reader_test.cpp
@@ -21,7 +21,9 @@
#include <arrow/io/api.h>
#include <gtest/gtest.h>
#include <parquet/api/reader.h>
+#include <parquet/api/writer.h>
#include <parquet/arrow/writer.h>
+#include <parquet/column_page.h>
#include <parquet/page_index.h>
#include <array>
@@ -835,6 +837,123 @@ std::shared_ptr<arrow::Array>
build_nullable_struct_with_list_array(bool list_fi
return finish_array(&builder);
}
+constexpr size_t SPANNING_NESTED_VALUES = 128;
+
+void write_sparse_filter_nested_parquet_file(const std::string& file_path) {
+ auto file_result = arrow::io::FileOutputStream::Open(file_path);
+ ASSERT_TRUE(file_result.ok()) << file_result.status();
+ std::shared_ptr<arrow::io::FileOutputStream> out = *file_result;
+
+ const auto id = ::parquet::schema::PrimitiveNode::Make("id",
::parquet::Repetition::REQUIRED,
+
::parquet::LogicalType::None(),
+
::parquet::Type::INT32);
+ const auto map_key = ::parquet::schema::PrimitiveNode::Make(
+ "key", ::parquet::Repetition::REQUIRED,
::parquet::LogicalType::None(),
+ ::parquet::Type::INT32);
+ const auto map_value = ::parquet::schema::PrimitiveNode::Make(
+ "value", ::parquet::Repetition::OPTIONAL,
::parquet::LogicalType::String(),
+ ::parquet::Type::BYTE_ARRAY);
+ const auto key_value = ::parquet::schema::GroupNode::Make(
+ "key_value", ::parquet::Repetition::REPEATED, {map_key,
map_value});
+ const auto map = ::parquet::schema::GroupNode::Make("m",
::parquet::Repetition::OPTIONAL,
+ {key_value},
::parquet::LogicalType::Map());
+ const auto element = ::parquet::schema::PrimitiveNode::Make(
+ "element", ::parquet::Repetition::OPTIONAL,
::parquet::LogicalType::None(),
+ ::parquet::Type::INT32);
+ const auto list =
+ ::parquet::schema::GroupNode::Make("list",
::parquet::Repetition::REPEATED, {element});
+ const auto items = ::parquet::schema::GroupNode::Make("items",
::parquet::Repetition::OPTIONAL,
+ {list},
::parquet::LogicalType::List());
+ const auto marker = ::parquet::schema::PrimitiveNode::Make(
+ "marker", ::parquet::Repetition::REQUIRED,
::parquet::LogicalType::None(),
+ ::parquet::Type::INT32);
+ const auto nested_struct = ::parquet::schema::GroupNode::Make(
+ "s", ::parquet::Repetition::OPTIONAL, {items, marker});
+ const auto schema_node = ::parquet::schema::GroupNode::Make(
+ "schema", ::parquet::Repetition::REQUIRED, {id, map,
nested_struct});
+ const auto schema =
std::static_pointer_cast<::parquet::schema::GroupNode>(schema_node);
+
+ ::parquet::WriterProperties::Builder builder;
+ builder.version(::parquet::ParquetVersion::PARQUET_2_6);
+ // V2 and page-index writers preserve record boundaries, so use V1 here to
produce the
+ // continuation pages that the reader must still handle correctly.
+ builder.data_page_version(::parquet::ParquetDataPageVersion::V1);
+ builder.compression(::parquet::Compression::UNCOMPRESSED);
+ builder.disable_dictionary();
+ builder.write_batch_size(8);
+ builder.data_pagesize(64);
+ auto writer = ::parquet::ParquetFileWriter::Open(out, schema,
builder.build());
+ auto* row_group = writer->AppendRowGroup();
+
+ auto* id_writer =
static_cast<::parquet::Int32Writer*>(row_group->NextColumn());
+ const int32_t ids[] = {1, 2, 3, 4, 5, 6};
+ EXPECT_EQ(id_writer->WriteBatch(6, nullptr, nullptr, ids), 6);
+ id_writer->Close();
+
+ std::vector<int16_t> map_repetition_levels {0, 0, 0, 0};
+ std::vector<int16_t> map_key_definition_levels {2, 0, 2, 1};
+ std::vector<int16_t> map_value_definition_levels {3, 0, 2, 1};
+ std::vector<int32_t> map_keys {10, 30};
+ std::vector<::parquet::ByteArray> map_values;
+ const std::string rejected_value = "rejected";
+ const std::string selected_value = "selected-wide-value";
+ map_values.emplace_back(static_cast<uint32_t>(rejected_value.size()),
+ reinterpret_cast<const
uint8_t*>(rejected_value.data()));
+ for (size_t value = 0; value < SPANNING_NESTED_VALUES; ++value) {
+ map_repetition_levels.push_back(value == 0 ? 0 : 1);
+ map_key_definition_levels.push_back(2);
+ map_value_definition_levels.push_back(3);
+ map_keys.push_back(static_cast<int32_t>(5000 + value));
+ map_values.emplace_back(static_cast<uint32_t>(selected_value.size()),
+ reinterpret_cast<const
uint8_t*>(selected_value.data()));
+ }
+ map_repetition_levels.push_back(0);
+ map_key_definition_levels.push_back(2);
+ map_value_definition_levels.push_back(2);
+ map_keys.push_back(6000);
+
+ auto* map_key_writer =
static_cast<::parquet::Int32Writer*>(row_group->NextColumn());
+
EXPECT_EQ(map_key_writer->WriteBatch(static_cast<int64_t>(map_repetition_levels.size()),
+ map_key_definition_levels.data(),
+ map_repetition_levels.data(),
map_keys.data()),
+ static_cast<int64_t>(map_keys.size()));
+ map_key_writer->Close();
+ auto* map_value_writer =
static_cast<::parquet::ByteArrayWriter*>(row_group->NextColumn());
+
EXPECT_EQ(map_value_writer->WriteBatch(static_cast<int64_t>(map_repetition_levels.size()),
+ map_value_definition_levels.data(),
+ map_repetition_levels.data(),
map_values.data()),
+ static_cast<int64_t>(map_values.size()));
+ map_value_writer->Close();
+
+ std::vector<int16_t> element_repetition_levels {0, 0, 0, 1, 0};
+ std::vector<int16_t> element_definition_levels {4, 0, 4, 3, 1};
+ std::vector<int32_t> element_values {10, 30};
+ for (size_t value = 0; value < SPANNING_NESTED_VALUES; ++value) {
+ element_repetition_levels.push_back(value == 0 ? 0 : 1);
+ element_definition_levels.push_back(4);
+ element_values.push_back(static_cast<int32_t>(5000 + value));
+ }
+ element_repetition_levels.push_back(0);
+ element_definition_levels.push_back(4);
+ element_values.push_back(6000);
+ element_repetition_levels.push_back(1);
+ element_definition_levels.push_back(3);
+
+ auto* element_writer =
static_cast<::parquet::Int32Writer*>(row_group->NextColumn());
+
EXPECT_EQ(element_writer->WriteBatch(static_cast<int64_t>(element_repetition_levels.size()),
+ element_definition_levels.data(),
+ element_repetition_levels.data(),
element_values.data()),
+ static_cast<int64_t>(element_values.size()));
+ element_writer->Close();
+ auto* marker_writer =
static_cast<::parquet::Int32Writer*>(row_group->NextColumn());
+ const int16_t marker_definition_levels[] = {1, 0, 1, 1, 1, 1};
+ const int32_t marker_values[] = {10, 30, 40, 50, 60};
+ EXPECT_EQ(marker_writer->WriteBatch(6, marker_definition_levels, nullptr,
marker_values), 5);
+ marker_writer->Close();
+ row_group->Close();
+ writer->Close();
+}
+
void write_nullable_map_parquet_file(const std::string& file_path) {
auto array = build_nullable_int_string_map_array();
auto field = arrow::field("arr", array->type(), true);
@@ -1645,6 +1764,139 @@ TEST_F(NewParquetReaderTest,
NativeComplexColumnsMaterializeDirectlyAcrossBatchC
EXPECT_GT(profile.get_counter("NestedBatches")->value(), 0);
}
+TEST_F(NewParquetReaderTest,
SparseFilterPreservesNestedShapeAcrossPhysicalPages) {
+ write_sparse_filter_nested_parquet_file(_file_path);
+
+ auto physical_reader = ::parquet::ParquetFileReader::OpenFile(_file_path,
false);
+ auto row_group_metadata = physical_reader->metadata()->RowGroup(0);
+ auto row_group_reader = physical_reader->RowGroup(0);
+ for (const std::string path :
+ {"m.key_value.key", "m.key_value.value", "s.items.list.element"}) {
+ int column_ordinal = -1;
+ for (int column = 0; column < row_group_metadata->num_columns();
++column) {
+ if
(row_group_metadata->ColumnChunk(column)->path_in_schema()->ToDotString() ==
path) {
+ column_ordinal = column;
+ break;
+ }
+ }
+ ASSERT_GE(column_ordinal, 0) << path;
+ auto page_reader =
row_group_reader->GetColumnPageReader(column_ordinal);
+ bool saw_continuation_page = false;
+ std::vector<int16_t> first_repetition_levels;
+ while (auto page = page_reader->NextPage()) {
+ if (page->type() != ::parquet::PageType::DATA_PAGE) {
+ continue;
+ }
+ auto data_page =
std::static_pointer_cast<::parquet::DataPageV1>(page);
+ ::parquet::LevelDecoder repetition_decoder;
+ repetition_decoder.SetData(data_page->repetition_level_encoding(),
1,
+ data_page->num_values(),
data_page->data(),
+ data_page->size());
+ int16_t first_repetition_level = 0;
+ ASSERT_EQ(repetition_decoder.Decode(1, &first_repetition_level),
1);
+ first_repetition_levels.push_back(first_repetition_level);
+ saw_continuation_page |= first_repetition_level > 0;
+ }
+ EXPECT_TRUE(saw_continuation_page)
+ << path << " must contain a parent row split across data
pages; page starts: "
+ << testing::PrintToString(first_repetition_levels);
+ }
+ physical_reader.reset();
+
+ RuntimeProfile profile("sparse_filter_nested_pages");
+ auto reader = create_reader(0, -1, &profile);
+ reader->set_batch_size(2);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ ASSERT_EQ(schema.size(), 3);
+
+ auto request = std::make_shared<format::FileScanRequest>();
+ request->predicate_columns = {field_projection(0)};
+ request->non_predicate_columns = {field_projection(1),
field_projection(2)};
+ request->conjuncts.push_back(create_int32_greater_than_conjunct(0, 3));
+ use_schema_order_positions(request.get(), schema);
+ ASSERT_TRUE(reader->open(request).ok());
+
+ MutableColumns output;
+ for (const auto& field : schema) {
+ output.push_back(field.type->create_column());
+ }
+ bool eof = false;
+ while (!eof) {
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ for (size_t column = 0; column < output.size(); ++column) {
+
output[column]->insert_range_from(*block.get_by_position(column).column, 0,
rows);
+ }
+ }
+
+ const auto& ids = assert_cast<const ColumnNullable&>(*output[0]);
+ const auto& id_values = assert_cast<const
ColumnInt32&>(ids.get_nested_column());
+ EXPECT_EQ(std::vector<int32_t>(id_values.get_data().begin(),
id_values.get_data().end()),
+ std::vector<int32_t>({4, 5, 6}));
+
+ const auto& nullable_map = assert_cast<const ColumnNullable&>(*output[1]);
+ EXPECT_EQ(nullable_map.get_null_map_data(), NullMap({0, 0, 0}));
+ const auto& map = assert_cast<const
ColumnMap&>(nullable_map.get_nested_column());
+ EXPECT_EQ(map.get_offsets(),
+ ColumnArray::Offsets64({0, SPANNING_NESTED_VALUES,
SPANNING_NESTED_VALUES + 1}));
+ const auto& map_keys = assert_cast<const ColumnNullable&>(map.get_keys());
+ const auto& key_values = assert_cast<const
ColumnInt32&>(map_keys.get_nested_column());
+ ASSERT_EQ(key_values.size(), SPANNING_NESTED_VALUES + 1);
+ EXPECT_EQ(std::count(map_keys.get_null_map_data().begin(),
map_keys.get_null_map_data().end(),
+ uint8_t {1}),
+ 0);
+ for (size_t value = 0; value < SPANNING_NESTED_VALUES; ++value) {
+ EXPECT_EQ(key_values.get_element(value), 5000 + value);
+ }
+ EXPECT_EQ(key_values.get_element(SPANNING_NESTED_VALUES), 6000);
+ const auto& map_values = assert_cast<const
ColumnNullable&>(map.get_values());
+ const auto& map_strings = assert_cast<const
ColumnString&>(map_values.get_nested_column());
+ EXPECT_EQ(std::count(map_values.get_null_map_data().begin(),
+ map_values.get_null_map_data().end(), uint8_t {1}),
+ 1);
+ for (size_t value = 0; value < SPANNING_NESTED_VALUES; ++value) {
+ EXPECT_EQ(map_strings.get_data_at(value).to_string(),
"selected-wide-value");
+ }
+ EXPECT_TRUE(map_values.is_null_at(SPANNING_NESTED_VALUES));
+
+ const auto& nullable_struct = assert_cast<const
ColumnNullable&>(*output[2]);
+ EXPECT_EQ(nullable_struct.get_null_map_data(), NullMap({0, 0, 0}));
+ const auto& struct_column =
+ assert_cast<const
ColumnStruct&>(nullable_struct.get_nested_column());
+ const auto& nullable_list = assert_cast<const
ColumnNullable&>(struct_column.get_column(0));
+ EXPECT_EQ(nullable_list.get_null_map_data(), NullMap({1, 0, 0}));
+ const auto& list = assert_cast<const
ColumnArray&>(nullable_list.get_nested_column());
+ EXPECT_EQ(list.get_offsets(),
+ ColumnArray::Offsets64({0, SPANNING_NESTED_VALUES,
SPANNING_NESTED_VALUES + 2}));
+ const auto& nullable_elements = assert_cast<const
ColumnNullable&>(list.get_data());
+ const auto& element_values =
+ assert_cast<const
ColumnInt32&>(nullable_elements.get_nested_column());
+ ASSERT_EQ(element_values.size(), SPANNING_NESTED_VALUES + 2);
+ for (size_t value = 0; value < SPANNING_NESTED_VALUES; ++value) {
+ EXPECT_EQ(element_values.get_element(value), 5000 + value);
+ EXPECT_FALSE(nullable_elements.is_null_at(value));
+ }
+ EXPECT_EQ(element_values.get_element(SPANNING_NESTED_VALUES), 6000);
+ EXPECT_TRUE(nullable_elements.is_null_at(SPANNING_NESTED_VALUES + 1));
+ const auto& markers = assert_cast<const
ColumnNullable&>(struct_column.get_column(1));
+ const auto& marker_values = assert_cast<const
ColumnInt32&>(markers.get_nested_column());
+ EXPECT_EQ(std::count(markers.get_null_map_data().begin(),
markers.get_null_map_data().end(),
+ uint8_t {1}),
+ 0);
+ EXPECT_EQ(
+ std::vector<int32_t>(marker_values.get_data().begin(),
marker_values.get_data().end()),
+ std::vector<int32_t>({40, 50, 60}));
+
+ ASSERT_NE(profile.get_counter("SelectedRows"), nullptr);
+ EXPECT_EQ(profile.get_counter("SelectedRows")->value(), 3);
+ ASSERT_NE(profile.get_counter("NestedBatches"), nullptr);
+ EXPECT_GT(profile.get_counter("NestedBatches")->value(), 0);
+}
+
TEST_F(NewParquetReaderTest,
FullComplexChildUnderPartialParentReadsItsWholeSubtree) {
write_nested_complex_under_struct_parquet_file(_file_path);
for (size_t child_index = 0; child_index < 3; ++child_index) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
