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 61720ba4927 [branch-4.1][pick] Backport Paimon and Parquet scan
improvements (#66359)
61720ba4927 is described below
commit 61720ba49278b6a7b954c250e581f26f6629edd6
Author: Gabriel <[email protected]>
AuthorDate: Mon Aug 3 13:02:10 2026 +0800
[branch-4.1][pick] Backport Paimon and Parquet scan improvements (#66359)
### What problem does this PR solve?
Original PRs (master merge order):
- [#66244 — Reduce Paimon JNI read object
allocations](https://github.com/apache/doris/pull/66244)
- [#66263 — Fuse Parquet nested sparse selection
planning](https://github.com/apache/doris/pull/66263)
- [#66357 — Keep File Scanner V2 predicate pruning
enabled](https://github.com/apache/doris/pull/66357)
Backport the three changes to `branch-4.1` in their master integration
order. This branch contains the complete final three-commit chain merged
by #66357.
The backport reduces Paimon JNI allocation overhead, fuses Parquet V2
nested sparse-selection planning, and keeps safe FileScannerV2 predicate
pruning independent of legacy session gates. It also adapts the
FileScannerV2 test fixture to the `branch-4.1` row-descriptor
nullability invariant.
### Release note
None
### Check List (For Author)
- Test
- [x] Unit Test
- [ ] Regression test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason
- Behavior changed:
- [ ] No.
- [x] Yes. Safe FileScannerV2 partition and expression ZoneMap pruning
is no longer disabled by legacy session gates.
- Does this need documentation?
- [x] No.
- [ ] Yes.
Verification:
- FE Checkstyle: 28 modules passed with zero violations.
- Paimon unit tests: 42 tests passed with zero failures or errors.
- Focused BE ASAN unit tests: 6 tests passed.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
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 +-
be/src/exec/scan/file_scanner_v2.cpp | 5 +-
be/src/format_v2/parquet/parquet_statistics.cpp | 10 +-
.../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 +
be/test/exec/scan/file_scanner_v2_test.cpp | 62 +++++
.../parquet/parquet_benchmark_scenarios_test.cpp | 20 +-
.../parquet/parquet_reader_control_test.cpp | 75 ++++++
be/test/format_v2/parquet/parquet_reader_test.cpp | 252 +++++++++++++++++++
.../format_v2/parquet/parquet_statistics_test.cpp | 74 ++++++
.../org/apache/doris/paimon/PaimonColumnValue.java | 131 ++++++++--
.../apache/doris/paimon/PaimonColumnValueTest.java | 236 +++++++++++++++++-
.../apache/doris/paimon/PaimonJniScannerTest.java | 10 +
.../java/org/apache/doris/qe/SessionVariable.java | 16 +-
gensrc/thrift/PaloInternalService.thrift | 6 +-
19 files changed, 1306 insertions(+), 69 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/exec/scan/file_scanner_v2.cpp
b/be/src/exec/scan/file_scanner_v2.cpp
index c568832d67c..6a7bf74571e 100644
--- a/be/src/exec/scan/file_scanner_v2.cpp
+++ b/be/src/exec/scan/file_scanner_v2.cpp
@@ -642,7 +642,10 @@ Status FileScannerV2::_prepare_table_reader_split(const
TFileRangeDesc& range,
VExprContextSPtrs conjuncts;
RETURN_IF_ERROR(_build_table_conjuncts(&conjuncts));
VExprContextSPtrs partition_prune_conjuncts;
- if (_state->query_options().enable_runtime_filter_partition_prune) {
+ if (!partition_values.empty()) {
+ // A split without partition constants cannot be pruned here, so avoid
cloning every
+ // conjunct solely for a consumer that must return immediately.
FileScannerV2 otherwise
+ // keeps safe partition pruning enabled independently of the legacy
session gate.
RETURN_IF_ERROR(_build_table_conjuncts(&partition_prune_conjuncts));
}
RETURN_IF_ERROR(_table_reader->prepare_split({
diff --git a/be/src/format_v2/parquet/parquet_statistics.cpp
b/be/src/format_v2/parquet/parquet_statistics.cpp
index 7d1afb39240..7f35b087910 100644
--- a/be/src/format_v2/parquet/parquet_statistics.cpp
+++ b/be/src/format_v2/parquet/parquet_statistics.cpp
@@ -452,11 +452,11 @@ std::optional<format::LocalColumnId>
file_column_id_by_block_position(
return std::nullopt;
}
-bool has_expr_zonemap_filter(const format::FileScanRequest& request,
- const RuntimeState* runtime_state) {
- if (!expr_zonemap::is_expr_zonemap_filter_enabled(runtime_state)) {
- return false;
- }
+bool has_expr_zonemap_filter(const format::FileScanRequest& request, const
RuntimeState*) {
+ // FileScannerV2 metadata pruning is a fixed part of its scan pipeline and
must not inherit
+ // the legacy scanner's expression ZoneMap session gate.
+ // TODO: Fence metadata pruning at the first unsafe/error-preserving
conjunct so a later
+ // ZoneMap predicate cannot bypass its row-level evaluation.
for (const auto& conjunct : request.conjuncts) {
if (conjunct != nullptr && conjunct->root() != nullptr &&
conjunct->root()->can_evaluate_zonemap_filter()) {
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/exec/scan/file_scanner_v2_test.cpp
b/be/test/exec/scan/file_scanner_v2_test.cpp
index 2950fa4911d..c73ed981491 100644
--- a/be/test/exec/scan/file_scanner_v2_test.cpp
+++ b/be/test/exec/scan/file_scanner_v2_test.cpp
@@ -141,6 +141,18 @@ private:
std::shared_ptr<RetryableCloseState> _state;
};
+class CapturingSplitTableReader final : public format::TableReader {
+public:
+ Status prepare_split(const format::SplitReadOptions& options) override {
+ conjunct_count = options.conjuncts.has_value() ?
options.conjuncts->size() : 0;
+ partition_prune_conjunct_count =
options.partition_prune_conjuncts.size();
+ return Status::OK();
+ }
+
+ size_t conjunct_count = 0;
+ size_t partition_prune_conjunct_count = 0;
+};
+
VExprSPtr slot_ref(int slot_id, int column_id, DataTypePtr type, const
std::string& name) {
return VSlotRef::create_shared(slot_id, column_id, -1, std::move(type),
name);
}
@@ -445,6 +457,56 @@ TEST(FileScannerV2Test,
FailedTableReaderCloseCanBeRetriedThroughScanner) {
EXPECT_EQ(close_state->close_calls, 2);
}
+TEST(FileScannerV2Test,
PartitionPruningRemainsEnabledWhenSessionSwitchIsFalse) {
+ TQueryOptions query_options;
+ query_options.__set_enable_runtime_filter_partition_prune(false);
+ RuntimeState state {query_options, TQueryGlobals()};
+ ObjectPool pool;
+ TDescriptorTable thrift_descriptors;
+ TTupleDescriptor tuple_descriptor;
+ tuple_descriptor.id = 0;
+ tuple_descriptor.byteSize = 0;
+ tuple_descriptor.numNullBytes = 0;
+ thrift_descriptors.tupleDescriptors.push_back(tuple_descriptor);
+ DescriptorTbl* descriptors = nullptr;
+ ASSERT_TRUE(DescriptorTbl::create(&pool, thrift_descriptors,
&descriptors).ok());
+ TPlanNode plan_node;
+ plan_node.node_id = 0;
+ plan_node.node_type = TPlanNodeType::FILE_SCAN_NODE;
+ plan_node.num_children = 0;
+ plan_node.limit = -1;
+ plan_node.row_tuples.push_back(0);
+ // RowDescriptor requires one nullability entry for every row tuple on
branch-4.1.
+ plan_node.nullable_tuples.push_back(false);
+ plan_node.file_scan_node.tuple_id = 0;
+ plan_node.__isset.file_scan_node = true;
+ FileScanOperatorX parent(&pool, plan_node, 0, *descriptors, 1);
+ FileScanLocalState local_state(&state, &parent);
+ RuntimeProfile profile("file_scanner_v2_partition_prune");
+ auto table_reader = std::make_unique<CapturingSplitTableReader>();
+ auto* captured = table_reader.get();
+ FileScannerV2 scanner(&state, &profile, std::move(table_reader));
+ scanner._local_state = &local_state;
+
+ TFileScanRangeParams params;
+ params.__set_format_type(TFileFormatType::FORMAT_PARQUET);
+ scanner._params = ¶ms;
+ scanner._slot_id_to_global_index.emplace(7, format::GlobalIndex(0));
+ scanner._conjuncts = {VExprContext::create_shared(
+ slot_ref(7, 7, std::make_shared<DataTypeInt32>(),
"partition_col"))};
+
+ const auto range = range_with_format("hive",
TFileFormatType::FORMAT_PARQUET);
+ ASSERT_TRUE(scanner._prepare_table_reader_split(range, {}).ok());
+ EXPECT_EQ(captured->conjunct_count, 1);
+ EXPECT_EQ(captured->partition_prune_conjunct_count, 0);
+
+ ASSERT_TRUE(scanner._prepare_table_reader_split(
+ range, {{"partition_col",
Field::create_field<TYPE_INT>(1)}})
+ .ok());
+ EXPECT_EQ(captured->conjunct_count, 1);
+ EXPECT_EQ(captured->partition_prune_conjunct_count, 1);
+}
+
// Scenario: Once FileScannerV2 is selected, an unsupported range must fail
instead of falling back
// to FileScanner.
TEST(FileScannerV2Test, ValidateScanRangeRejectsUnsupportedRange) {
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) {
diff --git a/be/test/format_v2/parquet/parquet_statistics_test.cpp
b/be/test/format_v2/parquet/parquet_statistics_test.cpp
index e69f1311e4a..740892367bb 100644
--- a/be/test/format_v2/parquet/parquet_statistics_test.cpp
+++ b/be/test/format_v2/parquet/parquet_statistics_test.cpp
@@ -47,6 +47,7 @@
#include "format_v2/parquet/parquet_file_context.h"
#include "format_v2/parquet/reader/native/block_split_bloom_filter.h"
#include "io/fs/file_reader.h"
+#include "runtime/runtime_state.h"
#include "util/thrift_util.h"
namespace doris {
namespace {
@@ -669,6 +670,79 @@ TEST(NativeParquetStatisticsTest,
TypeDefinedBoundsRequireSupportedColumnOrder)
EXPECT_TRUE(selected_ranges.empty());
}
+TEST(NativeParquetStatisticsTest, ZonemapPruningIgnoresDisabledSessionSwitch) {
+ auto encode_int32 = [](int32_t value) {
+ std::string bytes(sizeof(value), '\0');
+ memcpy(bytes.data(), &value, sizeof(value));
+ return bytes;
+ };
+
+ auto column_schema =
std::make_unique<format::parquet::ParquetColumnSchema>();
+ column_schema->kind = format::parquet::ParquetColumnSchemaKind::PRIMITIVE;
+ column_schema->local_id = 0;
+ column_schema->leaf_column_id = 0;
+ column_schema->type = std::make_shared<DataTypeInt32>();
+ column_schema->type_descriptor.doris_type = column_schema->type;
+ column_schema->type_descriptor.physical_type = tparquet::Type::INT32;
+ std::vector<std::unique_ptr<format::parquet::ParquetColumnSchema>> schema;
+ schema.push_back(std::move(column_schema));
+
+ tparquet::Statistics statistics;
+ statistics.__set_min_value(encode_int32(1));
+ statistics.__set_max_value(encode_int32(2));
+ statistics.__set_null_count(0);
+ tparquet::ColumnMetaData column_metadata;
+ column_metadata.__set_type(tparquet::Type::INT32);
+ column_metadata.__set_num_values(1);
+ column_metadata.__set_statistics(statistics);
+ tparquet::ColumnChunk chunk;
+ chunk.__set_meta_data(column_metadata);
+ tparquet::RowGroup row_group;
+ row_group.__set_columns({chunk});
+ row_group.__set_num_rows(1);
+ tparquet::ColumnOrder order;
+ order.__set_TYPE_ORDER(tparquet::TypeDefinedOrder());
+ tparquet::FileMetaData metadata;
+ metadata.__set_column_orders({order});
+ metadata.__set_row_groups({row_group});
+
+ format::FileScanRequest request;
+ request.local_positions.emplace(format::LocalColumnId(0),
format::LocalIndex(0));
+ request.predicate_columns =
{format::LocalColumnIndex::top_level(format::LocalColumnId(0))};
+ request.conjuncts = {
+
VExprContext::create_shared(std::make_shared<MetadataInt32GreaterThanExpr>(100))};
+
+ TQueryOptions query_options;
+ query_options.__set_enable_expr_zonemap_filter(false);
+ RuntimeState state {query_options, TQueryGlobals()};
+ std::vector<int> selected_row_groups;
+ ASSERT_TRUE(format::parquet::select_row_groups_by_metadata(metadata,
schema, request, nullptr,
+
&selected_row_groups, false, nullptr,
+ nullptr, &state)
+ .ok());
+ EXPECT_TRUE(selected_row_groups.empty());
+
+ format::parquet::NativeParquetPageIndex page_index;
+ page_index.column_index.__set_min_values({encode_int32(1)});
+ page_index.column_index.__set_max_values({encode_int32(2)});
+ page_index.column_index.__set_null_pages({false});
+ page_index.column_index.__set_null_counts({0});
+ tparquet::PageLocation location;
+ location.__set_offset(0);
+ location.__set_compressed_page_size(10);
+ location.__set_first_row_index(0);
+ page_index.offset_index.__set_page_locations({location});
+ std::unordered_map<int, format::parquet::NativeParquetPageIndex>
page_indexes;
+ page_indexes.emplace(0, std::move(page_index));
+ std::vector<format::parquet::RowRange> selected_ranges;
+ std::map<int, format::parquet::ParquetPageSkipPlan> skip_plans;
+ ASSERT_TRUE(format::parquet::select_row_group_ranges_by_native_page_index(
+ metadata, page_indexes, schema, request, 1,
&selected_ranges, &skip_plans,
+ nullptr, nullptr, &state)
+ .ok());
+ EXPECT_TRUE(selected_ranges.empty());
+}
+
TEST(NativeParquetStatisticsTest,
ContradictoryAllNullPageCountsDisablePruning) {
auto column_schema =
std::make_unique<format::parquet::ParquetColumnSchema>();
column_schema->kind = format::parquet::ParquetColumnSchemaKind::PRIMITIVE;
diff --git
a/fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonColumnValue.java
b/fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonColumnValue.java
index f60bb996703..339feb97c9b 100644
---
a/fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonColumnValue.java
+++
b/fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonColumnValue.java
@@ -38,20 +38,47 @@ import java.math.BigInteger;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
public class PaimonColumnValue implements ColumnValue {
private static final Logger LOG =
LoggerFactory.getLogger(PaimonColumnValue.class);
+ private static final Map<String, String> DORIS_TIME_ZONE_ALIASES;
+
+ static {
+ Map<String, String> aliases = new HashMap<>(ZoneId.SHORT_IDS);
+ // The scanner cannot depend on FE's TimeUtils, so keep its accepted
aliases and CST
+ // interpretation identical at this JNI boundary.
+ aliases.put("CST", "Asia/Shanghai");
+ aliases.put("PRC", "Asia/Shanghai");
+ aliases.put("UTC", "UTC");
+ aliases.put("GMT", "UTC");
+ DORIS_TIME_ZONE_ALIASES = Collections.unmodifiableMap(aliases);
+ }
+
private int idx;
private DataGetters record;
private ColumnType dorisType;
private DataType dataType;
- private String timeZone;
+ private ZoneId timeZone;
+ // Keep these caches lazy so scalar columns do not pay for complex-type
reuse bookkeeping.
+ private List<PaimonColumnValue> arrayValues;
+ private List<PaimonColumnValue> mapKeys;
+ private List<PaimonColumnValue> mapValues;
+ private List<PaimonColumnValue> structValues;
public PaimonColumnValue() {
}
public PaimonColumnValue(DataGetters record, int idx, ColumnType
columnType, DataType dataType, String timeZone) {
+ this(record, idx, columnType, dataType, resolveTimeZone(timeZone));
+ }
+
+ private PaimonColumnValue(
+ DataGetters record, int idx, ColumnType columnType, DataType
dataType, ZoneId timeZone) {
this.idx = idx;
this.record = record;
this.dorisType = columnType;
@@ -70,7 +97,7 @@ public class PaimonColumnValue implements ColumnValue {
}
public void setTimeZone(String timeZone) {
- this.timeZone = timeZone;
+ this.timeZone = resolveTimeZone(timeZone);
}
@Override
@@ -142,8 +169,8 @@ public class PaimonColumnValue implements ColumnValue {
public LocalDateTime getDateTime() {
Timestamp ts = record.getTimestamp(idx, dorisType.getPrecision());
if (dataType instanceof LocalZonedTimestampType) {
- return ts.toLocalDateTime().atZone(ZoneId.of("UTC"))
-
.withZoneSameInstant(ZoneId.of(timeZone)).toLocalDateTime();
+ // Paimon stores TIMESTAMP_LTZ as an epoch instant, so convert it
directly in the cached session zone.
+ return LocalDateTime.ofInstant(ts.toInstant(), timeZone);
} else {
return ts.toLocalDateTime();
}
@@ -152,15 +179,18 @@ public class PaimonColumnValue implements ColumnValue {
@Override
public LocalDateTime getTimeStampTz() {
Timestamp ts = record.getTimestamp(idx, dorisType.getPrecision());
- LocalDateTime v = ts.toInstant()
- .atZone(ZoneId.of("UTC"))
- .toLocalDateTime();
- return v;
+ // Timestamp's local representation is identical to converting its
epoch instant in UTC.
+ return ts.toLocalDateTime();
}
@Override
public boolean isNull() {
- return record.isNullAt(idx);
+ boolean isNull = record.isNullAt(idx);
+ if (isNull) {
+ // A null complex value has no live descendants; release wrappers
retained by its prior row.
+ clearChildCaches();
+ }
+ return isNull;
}
@Override
@@ -171,28 +201,41 @@ public class PaimonColumnValue implements ColumnValue {
@Override
public void unpackArray(List<ColumnValue> values) {
InternalArray recordArray = record.getArray(idx);
+ if (arrayValues == null) {
+ arrayValues = new ArrayList<>();
+ }
+ ColumnType elementDorisType = dorisType.getChildTypes().get(0);
+ DataType elementPaimonType = ((ArrayType) dataType).getElementType();
for (int i = 0; i < recordArray.size(); i++) {
- PaimonColumnValue arrayColumnValue = new
PaimonColumnValue((DataGetters) recordArray, i,
- dorisType.getChildTypes().get(0), ((ArrayType)
dataType).getElementType(), timeZone);
- values.add(arrayColumnValue);
+ values.add(reuseColumnValue(arrayValues, i, (DataGetters)
recordArray, i,
+ elementDorisType, elementPaimonType));
}
+ trimCache(arrayValues, recordArray.size());
}
@Override
public void unpackMap(List<ColumnValue> keys, List<ColumnValue> values) {
InternalMap map = record.getMap(idx);
+ if (mapKeys == null) {
+ mapKeys = new ArrayList<>();
+ mapValues = new ArrayList<>();
+ }
InternalArray key = map.keyArray();
+ ColumnType keyDorisType = dorisType.getChildTypes().get(0);
+ DataType keyPaimonType = ((MapType) dataType).getKeyType();
for (int i = 0; i < key.size(); i++) {
- PaimonColumnValue keyColumnValue = new
PaimonColumnValue((DataGetters) key, i,
- dorisType.getChildTypes().get(0), ((MapType)
dataType).getKeyType(), timeZone);
- keys.add(keyColumnValue);
+ keys.add(reuseColumnValue(mapKeys, i, (DataGetters) key, i,
+ keyDorisType, keyPaimonType));
}
+ trimCache(mapKeys, key.size());
InternalArray value = map.valueArray();
+ ColumnType valueDorisType = dorisType.getChildTypes().get(1);
+ DataType valuePaimonType = ((MapType) dataType).getValueType();
for (int i = 0; i < value.size(); i++) {
- PaimonColumnValue valueColumnValue = new
PaimonColumnValue((DataGetters) value, i,
- dorisType.getChildTypes().get(1), ((MapType)
dataType).getValueType(), timeZone);
- values.add(valueColumnValue);
+ values.add(reuseColumnValue(mapValues, i, (DataGetters) value, i,
+ valueDorisType, valuePaimonType));
}
+ trimCache(mapValues, value.size());
}
@Override
@@ -200,9 +243,57 @@ public class PaimonColumnValue implements ColumnValue {
RowType rowType = (RowType) dataType;
// Projection entries are original child indexes, so the binary row
must keep the full RowType arity.
InternalRow row = record.getRow(idx, rowType.getFieldCount());
+ if (structValues == null) {
+ structValues = new ArrayList<>();
+ }
for (int i : structFieldIndex) {
- values.add(new PaimonColumnValue(row, i,
dorisType.getChildTypes().get(i),
- rowType.getFields().get(i).type(), timeZone));
+ values.add(reuseColumnValue(structValues, i, row, i,
dorisType.getChildTypes().get(i),
+ rowType.getFields().get(i).type()));
+ }
+ }
+
+ private PaimonColumnValue reuseColumnValue(
+ List<PaimonColumnValue> cache, int cacheIndex, DataGetters
childRecord, int childIndex,
+ ColumnType childDorisType, DataType childPaimonType) {
+ while (cache.size() <= cacheIndex) {
+ cache.add(null);
+ }
+ PaimonColumnValue value = cache.get(cacheIndex);
+ if (value == null) {
+ value = new PaimonColumnValue(
+ childRecord, childIndex, childDorisType, childPaimonType,
timeZone);
+ cache.set(cacheIndex, value);
+ } else {
+ // VectorColumn consumes unpacked values synchronously, before
this parent advances to another value.
+ value.reset(childRecord, childIndex, childDorisType,
childPaimonType, timeZone);
}
+ return value;
+ }
+
+ private void reset(
+ DataGetters record, int idx, ColumnType dorisType, DataType
dataType, ZoneId timeZone) {
+ this.record = record;
+ this.idx = idx;
+ this.dorisType = dorisType;
+ this.dataType = dataType;
+ this.timeZone = timeZone;
+ }
+
+ private static ZoneId resolveTimeZone(String timeZone) {
+ return ZoneId.of(timeZone, DORIS_TIME_ZONE_ALIASES);
+ }
+
+ private static void trimCache(List<PaimonColumnValue> cache, int liveSize)
{
+ if (cache.size() > liveSize) {
+ // Retain only wrappers addressable by the current container, not
its historical maximum.
+ cache.subList(liveSize, cache.size()).clear();
+ }
+ }
+
+ private void clearChildCaches() {
+ arrayValues = null;
+ mapKeys = null;
+ mapValues = null;
+ structValues = null;
}
}
diff --git
a/fe/be-java-extensions/paimon-connector/src/test/java/org/apache/doris/paimon/PaimonColumnValueTest.java
b/fe/be-java-extensions/paimon-connector/src/test/java/org/apache/doris/paimon/PaimonColumnValueTest.java
index 639a2e87f30..20062ff2f6e 100644
---
a/fe/be-java-extensions/paimon-connector/src/test/java/org/apache/doris/paimon/PaimonColumnValueTest.java
+++
b/fe/be-java-extensions/paimon-connector/src/test/java/org/apache/doris/paimon/PaimonColumnValueTest.java
@@ -21,21 +21,33 @@ import org.apache.doris.common.jni.vec.ColumnType;
import org.apache.doris.common.jni.vec.ColumnValue;
import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericArray;
+import org.apache.paimon.data.GenericMap;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.Timestamp;
import org.apache.paimon.data.serializer.InternalRowSerializer;
+import org.apache.paimon.types.ArrayType;
import org.apache.paimon.types.BigIntType;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.LocalZonedTimestampType;
+import org.apache.paimon.types.MapType;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.TimestampType;
import org.apache.paimon.types.VarCharType;
import org.junit.Assert;
import org.junit.Test;
+import java.lang.reflect.Field;
+import java.time.Instant;
+import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
public class PaimonColumnValueTest {
private final RowType paimonStructType = RowType.of(
@@ -80,10 +92,228 @@ public class PaimonColumnValueTest {
Assert.assertEquals(100L, values.get(2).getLong());
}
+ @Test
+ public void testReuseArrayElementsAcrossRows() {
+ ArrayType paimonArrayType = new ArrayType(new IntType());
+ ColumnType dorisArrayType = ColumnType.parseType("a", "array<int>");
+ PaimonColumnValue arrayValue = new PaimonColumnValue(
+ GenericRow.of(new GenericArray(new int[] {1, 2})), 0,
+ dorisArrayType, paimonArrayType, "UTC");
+
+ List<ColumnValue> firstValues = new ArrayList<>();
+ arrayValue.unpackArray(firstValues);
+ arrayValue.setOffsetRow(GenericRow.of(new GenericArray(new int[] {3,
4, 5})));
+ List<ColumnValue> secondValues = new ArrayList<>();
+ arrayValue.unpackArray(secondValues);
+
+ Assert.assertSame(firstValues.get(0), secondValues.get(0));
+ Assert.assertSame(firstValues.get(1), secondValues.get(1));
+ Assert.assertEquals(Arrays.asList(3, 4, 5), getInts(secondValues));
+
+ arrayValue.setOffsetRow(GenericRow.of(new GenericArray(new int[]
{6})));
+ List<ColumnValue> thirdValues = new ArrayList<>();
+ arrayValue.unpackArray(thirdValues);
+ Assert.assertSame(firstValues.get(0), thirdValues.get(0));
+ Assert.assertEquals(Collections.singletonList(6),
getInts(thirdValues));
+ }
+
+ @Test
+ public void testReuseMapEntriesAcrossRows() {
+ MapType paimonMapType = new MapType(new IntType(), new BigIntType());
+ ColumnType dorisMapType = ColumnType.parseType("m", "map<int,bigint>");
+ PaimonColumnValue mapValue = new PaimonColumnValue(
+ GenericRow.of(new GenericMap(linkedMap(1, 10L, 2, 20L))), 0,
+ dorisMapType, paimonMapType, "UTC");
+
+ List<ColumnValue> firstKeys = new ArrayList<>();
+ List<ColumnValue> firstValues = new ArrayList<>();
+ mapValue.unpackMap(firstKeys, firstValues);
+ mapValue.setOffsetRow(GenericRow.of(new GenericMap(linkedMap(3, 30L,
4, 40L))));
+ List<ColumnValue> secondKeys = new ArrayList<>();
+ List<ColumnValue> secondValues = new ArrayList<>();
+ mapValue.unpackMap(secondKeys, secondValues);
+
+ Assert.assertSame(firstKeys.get(0), secondKeys.get(0));
+ Assert.assertSame(firstKeys.get(1), secondKeys.get(1));
+ Assert.assertSame(firstValues.get(0), secondValues.get(0));
+ Assert.assertSame(firstValues.get(1), secondValues.get(1));
+ Assert.assertEquals(Arrays.asList(3, 4), getInts(secondKeys));
+ Assert.assertEquals(Arrays.asList(30L, 40L), getLongs(secondValues));
+ }
+
+ @Test
+ public void testReuseProjectedStructFieldsAcrossRows() {
+ PaimonColumnValue structValue = createStructValue();
+ List<Integer> projectedFields = Arrays.asList(0, 2);
+ List<ColumnValue> firstValues = new ArrayList<>();
+ structValue.unpackStruct(projectedFields, firstValues);
+
+ structValue.setOffsetRow(createStructRow(20, "y", 200L));
+ List<ColumnValue> secondValues = new ArrayList<>();
+ structValue.unpackStruct(projectedFields, secondValues);
+
+ Assert.assertSame(firstValues.get(0), secondValues.get(0));
+ Assert.assertSame(firstValues.get(1), secondValues.get(1));
+ Assert.assertEquals(20, secondValues.get(0).getInt());
+ Assert.assertEquals(200L, secondValues.get(1).getLong());
+ }
+
+ @Test
+ public void testReuseNestedArrayElementsAcrossRows() {
+ ArrayType paimonNestedArrayType = new ArrayType(new ArrayType(new
IntType()));
+ ColumnType dorisNestedArrayType = ColumnType.parseType("a",
"array<array<int>>");
+ PaimonColumnValue arrayValue = new PaimonColumnValue(
+ GenericRow.of(new GenericArray(new Object[] {new
GenericArray(new int[] {1, 2})})), 0,
+ dorisNestedArrayType, paimonNestedArrayType, "UTC");
+
+ List<ColumnValue> firstOuterValues = new ArrayList<>();
+ arrayValue.unpackArray(firstOuterValues);
+ List<ColumnValue> firstInnerValues = new ArrayList<>();
+ firstOuterValues.get(0).unpackArray(firstInnerValues);
+
+ arrayValue.setOffsetRow(
+ GenericRow.of(new GenericArray(new Object[] {new
GenericArray(new int[] {3, 4})})));
+ List<ColumnValue> secondOuterValues = new ArrayList<>();
+ arrayValue.unpackArray(secondOuterValues);
+ List<ColumnValue> secondInnerValues = new ArrayList<>();
+ secondOuterValues.get(0).unpackArray(secondInnerValues);
+
+ Assert.assertSame(firstOuterValues.get(0), secondOuterValues.get(0));
+ Assert.assertSame(firstInnerValues.get(0), secondInnerValues.get(0));
+ Assert.assertSame(firstInnerValues.get(1), secondInnerValues.get(1));
+ Assert.assertEquals(Arrays.asList(3, 4), getInts(secondInnerValues));
+ }
+
+ @Test
+ public void testNestedArrayCacheRetainsOnlyCurrentShape() throws Exception
{
+ int outerSize = 4;
+ int innerSize = 32;
+ ArrayType paimonNestedArrayType = new ArrayType(new ArrayType(new
IntType()));
+ ColumnType dorisNestedArrayType = ColumnType.parseType("a",
"array<array<int>>");
+ PaimonColumnValue arrayValue = new PaimonColumnValue(
+ nestedArrayRow(outerSize, innerSize, 0, -1), 0,
+ dorisNestedArrayType, paimonNestedArrayType, "UTC");
+
+ consumeNestedArray(arrayValue);
+ Assert.assertEquals(outerSize + innerSize,
retainedCachedValues(arrayValue));
+
+ // Moving the large child to position 1 makes position 0 a smaller,
non-null array.
+ arrayValue.setOffsetRow(nestedArrayRow(outerSize, innerSize, 1, -1));
+ consumeNestedArray(arrayValue);
+ Assert.assertEquals(outerSize + innerSize,
retainedCachedValues(arrayValue));
+
+ // Moving it again makes position 1 null, which must release that
position's descendants.
+ arrayValue.setOffsetRow(nestedArrayRow(outerSize, innerSize, 2, 1));
+ consumeNestedArray(arrayValue);
+ Assert.assertEquals(outerSize + innerSize,
retainedCachedValues(arrayValue));
+ }
+
+ @Test
+ public void testTimestampConversionsPreserveBoundaryValues() {
+ Timestamp timestamp = Timestamp.fromEpochMillis(-1, 999_999);
+ ColumnType dorisTimestampType = ColumnType.parseType("t",
"datetimev2(9)");
+
+ PaimonColumnValue timestampValue = new PaimonColumnValue(
+ GenericRow.of(timestamp), 0, dorisTimestampType, new
TimestampType(9), "UTC");
+ Assert.assertEquals(
+ LocalDateTime.of(1969, 12, 31, 23, 59, 59, 999_999_999),
+ timestampValue.getDateTime());
+ Assert.assertEquals(
+ LocalDateTime.of(1969, 12, 31, 23, 59, 59, 999_999_999),
+ timestampValue.getTimeStampTz());
+
+ PaimonColumnValue localZonedValue = new PaimonColumnValue(
+
GenericRow.of(Timestamp.fromInstant(Instant.parse("2024-03-10T10:30:00.123456789Z"))),
0,
+ dorisTimestampType, new LocalZonedTimestampType(9),
"America/Los_Angeles");
+ Assert.assertEquals(
+ LocalDateTime.of(2024, 3, 10, 3, 30, 0, 123_456_789),
+ localZonedValue.getDateTime());
+
+ localZonedValue.setTimeZone("Asia/Shanghai");
+ Assert.assertEquals(
+ LocalDateTime.of(2024, 3, 10, 18, 30, 0, 123_456_789),
+ localZonedValue.getDateTime());
+
+ localZonedValue.setTimeZone("CST");
+ Assert.assertEquals(
+ LocalDateTime.of(2024, 3, 10, 18, 30, 0, 123_456_789),
+ localZonedValue.getDateTime());
+ }
+
+ private InternalRow nestedArrayRow(int outerSize, int innerSize, int
populatedIndex, int nullIndex) {
+ Object[] outerValues = new Object[outerSize];
+ for (int i = 0; i < outerSize; i++) {
+ if (i == nullIndex) {
+ outerValues[i] = null;
+ } else if (i == populatedIndex) {
+ outerValues[i] = new GenericArray(new int[innerSize]);
+ } else {
+ outerValues[i] = new GenericArray(new int[0]);
+ }
+ }
+ return GenericRow.of(new GenericArray(outerValues));
+ }
+
+ private void consumeNestedArray(PaimonColumnValue arrayValue) {
+ List<ColumnValue> outerValues = new ArrayList<>();
+ arrayValue.unpackArray(outerValues);
+ for (ColumnValue outerValue : outerValues) {
+ if (!outerValue.isNull()) {
+ outerValue.unpackArray(new ArrayList<>());
+ }
+ }
+ }
+
+ private int retainedCachedValues(PaimonColumnValue value) throws Exception
{
+ int retained = 0;
+ for (String fieldName : Arrays.asList("arrayValues", "mapKeys",
"mapValues", "structValues")) {
+ Field field = PaimonColumnValue.class.getDeclaredField(fieldName);
+ field.setAccessible(true);
+ List<?> children = (List<?>) field.get(value);
+ if (children == null) {
+ continue;
+ }
+ for (Object child : children) {
+ if (child != null) {
+ retained++;
+ retained += retainedCachedValues((PaimonColumnValue)
child);
+ }
+ }
+ }
+ return retained;
+ }
+
private PaimonColumnValue createStructValue() {
- GenericRow nestedRow = GenericRow.of(10, BinaryString.fromString("x"),
100L);
+ return new PaimonColumnValue(createStructRow(10, "x", 100L), 0,
+ dorisStructType, paimonStructType, "UTC");
+ }
+
+ private InternalRow createStructRow(int intValue, String stringValue, long
longValue) {
+ GenericRow nestedRow = GenericRow.of(intValue,
BinaryString.fromString(stringValue), longValue);
RowType outerType = RowType.of(new DataType[] {paimonStructType}, new
String[] {"s"});
- InternalRow outerRow = new
InternalRowSerializer(outerType).toBinaryRow(GenericRow.of(nestedRow));
- return new PaimonColumnValue(outerRow, 0, dorisStructType,
paimonStructType, "UTC");
+ return new
InternalRowSerializer(outerType).toBinaryRow(GenericRow.of(nestedRow));
+ }
+
+ private List<Integer> getInts(List<ColumnValue> values) {
+ List<Integer> result = new ArrayList<>();
+ for (ColumnValue value : values) {
+ result.add(value.getInt());
+ }
+ return result;
+ }
+
+ private List<Long> getLongs(List<ColumnValue> values) {
+ List<Long> result = new ArrayList<>();
+ for (ColumnValue value : values) {
+ result.add(value.getLong());
+ }
+ return result;
+ }
+
+ private Map<Integer, Long> linkedMap(int key1, long value1, int key2, long
value2) {
+ Map<Integer, Long> result = new LinkedHashMap<>();
+ result.put(key1, value1);
+ result.put(key2, value2);
+ return result;
}
}
diff --git
a/fe/be-java-extensions/paimon-connector/src/test/java/org/apache/doris/paimon/PaimonJniScannerTest.java
b/fe/be-java-extensions/paimon-connector/src/test/java/org/apache/doris/paimon/PaimonJniScannerTest.java
index 563d2fbf659..9918618c07a 100644
---
a/fe/be-java-extensions/paimon-connector/src/test/java/org/apache/doris/paimon/PaimonJniScannerTest.java
+++
b/fe/be-java-extensions/paimon-connector/src/test/java/org/apache/doris/paimon/PaimonJniScannerTest.java
@@ -493,6 +493,16 @@ public class PaimonJniScannerTest {
Assert.assertEquals(2, sampleCalls.get());
}
+ @Test
+ public void
testConstructorAcceptsDorisShortTimeZoneForNonTemporalProjection() {
+ Map<String, String> params = createBaseParams();
+ params.put("required_fields", "id");
+ params.put("columns_types", "int");
+ params.put("time_zone", "EST");
+
+ new PaimonJniScanner(128, params);
+ }
+
@Test
public void testIOManagerOptionHelpers() throws Exception {
Map<String, String> params = createBaseParams();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
index 74caa6d0fad..1c59b4fa1ea 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
@@ -2129,7 +2129,15 @@ public class SessionVariable implements Serializable,
Writable {
@VariableMgr.VarAttr(name = ENABLE_RUNTIME_FILTER_PRUNE, needForward =
true, fuzzy = true)
public boolean enableRuntimeFilterPrune = true;
- @VariableMgr.VarAttr(name = ENABLE_RUNTIME_FILTER_PARTITION_PRUNE,
needForward = true, fuzzy = true)
+ @VariableMgr.VarAttr(
+ name = ENABLE_RUNTIME_FILTER_PARTITION_PRUNE,
+ description = {"控制支持该变量的 scanner 是否启用运行时过滤器分区裁剪。"
+ + "File Scanner V2 始终启用安全的分区裁剪。默认为 true。",
+ "Controls runtime-filter partition pruning in scanners
that honor this variable. "
+ + "File Scanner V2 always enables safe partition
pruning. "
+ + "The default value is true."},
+ needForward = true,
+ fuzzy = true)
public boolean enableRuntimeFilterPartitionPrune = true;
/**
@@ -2613,8 +2621,10 @@ public class SessionVariable implements Serializable,
Writable {
@VariableMgr.VarAttr(
name = ENABLE_EXPR_ZONEMAP_FILTER,
fuzzy = true,
- description = {"控制 scanner 是否启用表达式 ZoneMap 过滤。默认为 false。",
- "Controls whether to enable expression ZoneMap filtering
in scanners. "
+ description = {"控制支持该变量的 scanner 是否启用表达式 ZoneMap 过滤。"
+ + "File Scanner V2 始终启用安全的表达式 ZoneMap 过滤。默认为 false。",
+ "Controls expression ZoneMap filtering in scanners that
honor this variable. "
+ + "File Scanner V2 always enables safe expression
ZoneMap filtering. "
+ "The default value is false."},
needForward = true)
public boolean enableExprZonemapFilter = false;
diff --git a/gensrc/thrift/PaloInternalService.thrift
b/gensrc/thrift/PaloInternalService.thrift
index 05deef4efa5..64a8331dd93 100644
--- a/gensrc/thrift/PaloInternalService.thrift
+++ b/gensrc/thrift/PaloInternalService.thrift
@@ -374,6 +374,8 @@ struct TQueryOptions {
148: optional i32 min_scanners_concurrency = 1;
149: optional i32 min_scan_scheduler_concurrency = 0; //deprecated
+ // Controls runtime-filter partition pruning for readers that honor this
option.
+ // FileScannerV2 always enables safe partition pruning.
150: optional bool enable_runtime_filter_partition_prune = true;
// The minimum memory that an operator required to run.
@@ -496,7 +498,9 @@ struct TQueryOptions {
219: optional i64 ann_index_candidate_rows_threshold = 0
// Candidate row ratio threshold against segment rows. Existing default is
0.3.
220: optional double ann_index_candidate_rows_percent_threshold = 0.3
- // To control whether BE scan readers may apply expression-based ZoneMap
pruning.
+
+ // Controls expression-based ZoneMap pruning for readers that honor this
option.
+ // FileScannerV2 always enables safe expression ZoneMap pruning.
224: optional bool enable_expr_zonemap_filter = true
225: optional i64 runtime_filter_tree_publish_max_send_bytes = 268435456
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
