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 72989876caa [improvement](parquet) Lazily materialize complex residual
columns (#65965)
72989876caa is described below
commit 72989876caa12a025d74c1d80b5469a25df4b421
Author: Gabriel <[email protected]>
AuthorDate: Fri Jul 24 11:24:37 2026 +0800
[improvement](parquet) Lazily materialize complex residual columns (#65965)
### What problem does this PR solve?
Issue Number: N/A
Related PR: #65921
Problem Summary:
The V2 Parquet scanner already evaluates single-column predicates round
by round, but a multi-column residual (including the children of a
compound `AND`) still caused every predicate column to be materialized
before expression evaluation. Expression short circuiting therefore
happened after Parquet decode/IO and could not avoid later-only columns.
This PR follows expression-triggered lazy materialization:
- preserve conjunct order and split safe compound `AND` residuals into
ordered expression stages;
- record each stage's slot dependencies and materialize only the next
reachable columns;
- read later columns with the surviving selection, or skip them entirely
when an earlier stage rejects the batch;
- prefetch only the first reachable predicate stage;
- retain the original eager path for stateful/error-sensitive
expressions that are unsafe on selected rows;
- add unit coverage and a Release microbenchmark scenario based on the
Parquet benchmark framework from #65921.
Correctness/counter validation (6 input rows):
- an earlier two-column `AND` child filters all rows:
`ReaderReadRows=12`, `ReaderSkipRows=6` (the third column is never
decoded);
- when 3 rows survive the first residual: the later column reads only
those 3 rows (`ReaderReadRows=15`, `ReaderSelectRows=3`,
`ReaderSkipRows=3`).
Release microbenchmark:
```text
CPU: Intel Xeon Platinum 8457C, pinned to CPU 8
Build: RELEASE
Filter:
^ParquetReader/complex_residual_scan/plain/null_10/alternating/sel_10/
Protocol: 3 warmups, then A-B-B-A; 5 repetitions per group; min_time=1s
Pair 1 median CPU: before 1,377,041 ns; after 1,318,811 ns (-4.23%)
Pair 2 median CPU: before 2,140,937 ns; after 2,124,862 ns (-0.75%)
Paired geometric normalization: -2.50% CPU time
Selected rows: 1,460 in every run
```
The host had high concurrent load and CPU scaling enabled, so the paired
groups and run conditions are reported explicitly rather than relying on
wall time. Both A/B pairs improve in the same direction.
### Release note
None
### Check List (For Author)
- Test
- [ ] Regression test
- [x] Unit Test
- [x] Manual test (Release microbenchmark above)
- [ ] No need to test or manual test.
- Behavior changed:
- [x] No. The SQL result is unchanged; only predicate-column
materialization timing changes.
- [ ] Yes.
- Does this need documentation?
- [x] No.
- [ ] Yes.
### 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 | 10 +-
be/benchmark/parquet/README.md | 15 +-
be/benchmark/parquet/benchmark_parquet_reader.hpp | 94 +++++++-
be/benchmark/parquet/parquet_benchmark_scenarios.h | 22 +-
be/src/format_v2/expr/cast.h | 3 +
be/src/format_v2/parquet/parquet_profile.cpp | 3 +
be/src/format_v2/parquet/parquet_profile.h | 2 +
be/src/format_v2/parquet/parquet_scan.cpp | 251 +++++++++++++++++----
be/src/format_v2/parquet/parquet_scan.h | 11 +-
.../parquet/parquet_benchmark_scenarios_test.cpp | 24 +-
be/test/format_v2/parquet/parquet_scan_test.cpp | 241 +++++++++++++++++++-
11 files changed, 613 insertions(+), 63 deletions(-)
diff --git a/be/benchmark/parquet/AGENTS.md b/be/benchmark/parquet/AGENTS.md
index a8667fb43cd..a558f69937c 100644
--- a/be/benchmark/parquet/AGENTS.md
+++ b/be/benchmark/parquet/AGENTS.md
@@ -42,7 +42,7 @@ be/output/lib/benchmark_test --benchmark_list_tests \
| grep -c '^ParquetDecoder/' # currently 152
be/output/lib/benchmark_test --benchmark_list_tests \
- | grep -c '^ParquetReader/' # currently 137
+ | grep -c '^ParquetReader/' # currently 152
```
When running the binary directly from `be/build_RELEASE/bin`, make sure the
JVM and third-party
@@ -111,9 +111,10 @@ selection with clustered and alternating selection ranges,
for 152 registered ca
| DELTA_BYTE_ARRAY | BYTE_ARRAY |
`ParquetReader` deliberately uses a single-variable matrix rather than a
Cartesian product. After
-deduplication it contains 137 cases covering:
+deduplication it contains 152 cases covering:
-- operations: open-to-first-block, full scan, predicate scan, limit 1, and
limit 1000;
+- operations: open-to-first-block, full scan, predicate scan, complex residual
scan, limit 1, and
+ limit 1000;
- file encodings: PLAIN, dictionary, BYTE_STREAM_SPLIT, and
DELTA_BINARY_PACKED;
- null ratios: 0%, 1%, 10%, 50%, and 90%;
- null shapes: clustered and alternating;
@@ -167,6 +168,9 @@ Fixture contents and writer settings are:
- every non-null value is `row % 100`;
- the predicate is `value < selectivity_percent`, so the threshold maps
directly to the intended
non-null selectivity;
+- the complex residual scan evaluates a production expression tree whose first
child is
+ `c0 < selectivity_percent` and whose always-true second child is `c2 = c3`,
exposing whether
+ later-only columns are decoded eagerly;
- alternating nulls use a 101-row period and `(row * 37) % 101`, avoiding
direct correlation with
the 100-value predicate period;
- clustered nulls use contiguous null prefixes inside each 1,024-row cluster;
diff --git a/be/benchmark/parquet/README.md b/be/benchmark/parquet/README.md
index 40c79bc28be..f024f7d6a24 100644
--- a/be/benchmark/parquet/README.md
+++ b/be/benchmark/parquet/README.md
@@ -36,13 +36,14 @@ be/output/lib/benchmark_test \
## Local reader cases
-`ParquetReader` measures local open-to-first-block, full scan, predicate scan,
and LIMIT-shaped
-reads. The matrix covers:
+`ParquetReader` measures local open-to-first-block, full scan, predicate scan,
complex residual
+scan, and LIMIT-shaped reads. The matrix covers:
- PLAIN, dictionary, byte-stream-split, and DELTA binary-packed files;
- NULL ratios of 0%, 1%, 10%, 50%, and 90%, with clustered and alternating
placement;
- predicate selectivities of 0%, 1%, 10%, 50%, 90%, and 100%;
- predicate-only and predicate-plus-lazy-projected reads;
+- ordered complex residuals whose later columns are reachable only after an
earlier residual;
- schemas with 4, 32, 128, and 512 columns, with the predicate first or last.
Fixtures are created lazily under the system temporary directory in
@@ -58,6 +59,16 @@ be/output/lib/benchmark_test \
--benchmark_out_format=json
```
+The complex-residual case uses a production compound `AND` tree. Its first
child,
+`c0 < selectivity_percent`, preserves the requested selectivity; its second
child, `c2 = c3`,
+references two new columns and accepts every row that reaches it:
+
+```shell
+be/output/lib/benchmark_test \
+
--benchmark_filter='^ParquetReader/complex_residual_scan/plain/null_10/alternating/sel_10/'
\
+ --benchmark_min_time=1s
+```
+
Every result reports throughput plus `raw_rows`, `selected_rows`,
`fixture_bytes`, `ns/raw_row`,
and (when at least one row survives) `ns/selected_row`. Keep CPU frequency,
build type, compiler,
machine placement, and benchmark filters fixed when comparing two commits.
diff --git a/be/benchmark/parquet/benchmark_parquet_reader.hpp
b/be/benchmark/parquet/benchmark_parquet_reader.hpp
index ae3e14c0f66..63094738040 100644
--- a/be/benchmark/parquet/benchmark_parquet_reader.hpp
+++ b/be/benchmark/parquet/benchmark_parquet_reader.hpp
@@ -24,6 +24,7 @@
#include <parquet/arrow/writer.h>
#include <algorithm>
+#include <array>
#include <cstdint>
#include <filesystem>
#include <memory>
@@ -39,13 +40,18 @@
#include "core/column/column_vector.h"
#include "core/data_type/data_type_nullable.h"
#include "core/data_type/data_type_number.h"
+#include "exprs/vcompound_pred.h"
+#include "exprs/vectorized_fn_call.h"
#include "exprs/vexpr.h"
#include "exprs/vexpr_context.h"
+#include "exprs/vliteral.h"
+#include "exprs/vslot_ref.h"
#include "format_v2/file_reader.h"
#include "format_v2/parquet/parquet_reader.h"
#include "gen_cpp/Types_types.h"
#include "io/io_common.h"
#include "parquet_benchmark_scenarios.h"
+#include "runtime/descriptors.h"
#include "runtime/runtime_state.h"
#include "storage/index/zone_map/zonemap_eval_context.h"
#include "storage/index/zone_map/zonemap_filter_result.h"
@@ -275,6 +281,55 @@ inline VExprContextSPtr make_predicate(int
column_position, int selectivity_perc
return context;
}
+inline VExprSPtr make_int32_comparison(const std::string& function_name,
TExprOpcode::type opcode,
+ VExprSPtr left, VExprSPtr right) {
+ const auto bool_type = make_nullable(std::make_shared<DataTypeUInt8>());
+ TFunctionName name;
+ name.__set_function_name(function_name);
+ TFunction function;
+ function.__set_name(name);
+ function.__set_binary_type(TFunctionBinaryType::BUILTIN);
+ function.__set_arg_types({left->data_type()->to_thrift(),
right->data_type()->to_thrift()});
+ function.__set_ret_type(bool_type->to_thrift());
+ function.__set_has_var_args(false);
+ TExprNode node;
+ node.__set_node_type(TExprNodeType::BINARY_PRED);
+ node.__set_opcode(opcode);
+ node.__set_type(bool_type->to_thrift());
+ node.__set_fn(function);
+ node.__set_num_children(2);
+ node.__set_is_nullable(true);
+ auto comparison = VectorizedFnCall::create_shared(node);
+ comparison->add_child(std::move(left));
+ comparison->add_child(std::move(right));
+ return comparison;
+}
+
+inline VExprContextSPtr make_complex_residual_predicate(int
selectivity_percent, int first_position,
+ int
later_left_position,
+ int
later_right_position,
+ const DataTypePtr&
int_type) {
+ const auto bool_type = make_nullable(std::make_shared<DataTypeUInt8>());
+ TExprNode node;
+ node.__set_node_type(TExprNodeType::COMPOUND_PRED);
+ node.__set_opcode(TExprOpcode::COMPOUND_AND);
+ node.__set_type(bool_type->to_thrift());
+ node.__set_num_children(2);
+ node.__set_is_nullable(true);
+ auto compound = VCompoundPred::create_shared(node);
+ compound->add_child(make_int32_comparison(
+ "lt", TExprOpcode::LT,
+ VSlotRef::create_shared(first_position, first_position, -1,
int_type, "c0"),
+ VLiteral::create_shared(remove_nullable(int_type),
+
Field::create_field<TYPE_INT>(selectivity_percent))));
+ compound->add_child(make_int32_comparison(
+ "eq", TExprOpcode::EQ,
+ VSlotRef::create_shared(later_left_position, later_left_position,
-1, int_type, "c2"),
+ VSlotRef::create_shared(later_right_position,
later_right_position, -1, int_type,
+ "c3")));
+ return VExprContext::create_shared(std::move(compound));
+}
+
inline Block make_block(const std::vector<format::ColumnDefinition>& schema) {
Block block;
for (const auto& column : schema) {
@@ -284,10 +339,17 @@ inline Block make_block(const
std::vector<format::ColumnDefinition>& schema) {
}
struct ReaderSession {
+ ~ReaderSession() {
+ for (const auto& context : opened_conjuncts) {
+ context->close();
+ }
+ }
+
RuntimeState runtime_state {TQueryOptions(), TQueryGlobals()};
std::unique_ptr<format::parquet::ParquetReader> reader;
std::vector<format::ColumnDefinition> schema;
std::shared_ptr<format::FileScanRequest> request;
+ VExprContextSPtrs opened_conjuncts;
};
inline std::unique_ptr<ReaderSession> open_reader(const std::filesystem::path&
path,
@@ -324,6 +386,27 @@ inline std::unique_ptr<ReaderSession> open_reader(const
std::filesystem::path& p
const auto predicate_position =
session->request->local_positions.at(predicate_id).value();
session->request->conjuncts.push_back(
make_predicate(static_cast<int>(predicate_position),
scenario.selectivity_percent));
+ } else if (scenario.operation == ReaderOperation::COMPLEX_RESIDUAL_SCAN) {
+ DORIS_CHECK(scenario.schema_width >= 5);
+ std::array<int, 3> predicate_columns {0, 2, 3};
+ std::array<int, 3> predicate_positions {};
+ for (size_t index = 0; index < predicate_columns.size(); ++index) {
+ const int column = predicate_columns[index];
+ const auto predicate_id = format::LocalColumnId(column);
+ throw_if_error(request_builder.add_predicate_column(predicate_id));
+ session->request->predicate_only_columns.push_back(predicate_id);
+ predicate_positions[index] =
+
static_cast<int>(session->request->local_positions.at(predicate_id).value());
+ }
+ throw_if_error(request_builder.add_non_predicate_column(
+ format::LocalColumnId(scenario.schema_width - 1)));
+ auto context = make_complex_residual_predicate(
+ scenario.selectivity_percent, predicate_positions[0],
predicate_positions[1],
+ predicate_positions[2], session->schema[0].type);
+ throw_if_error(context->prepare(&session->runtime_state,
RowDescriptor()));
+ throw_if_error(context->open(&session->runtime_state));
+ session->request->conjuncts.push_back(context);
+ session->opened_conjuncts.push_back(std::move(context));
} else {
throw_if_error(request_builder.add_non_predicate_column(format::LocalColumnId(0)));
if (scenario.schema_width > 1) {
@@ -383,6 +466,9 @@ inline int projected_columns(const ReaderScenario&
scenario) {
scenario.projection == Projection::PREDICATE_ONLY) {
return 1;
}
+ if (scenario.operation == ReaderOperation::COMPLEX_RESIDUAL_SCAN) {
+ return 4;
+ }
return std::min(2, scenario.schema_width);
}
@@ -428,13 +514,7 @@ inline void run_reader(benchmark::State& state,
ReaderScenario scenario) {
inline bool register_reader_benchmarks() {
for (const auto& scenario : reader_scenarios()) {
- std::string name =
- "ParquetReader/" + to_string(scenario.operation) + "/" +
- to_string(scenario.encoding) + "/null_" +
std::to_string(scenario.null_percent) +
- "/" + to_string(scenario.null_pattern) + "/sel_" +
- std::to_string(scenario.selectivity_percent) + "/" +
- to_string(scenario.projection) + "/width_" +
std::to_string(scenario.schema_width) +
- "/predicate_" + std::to_string(scenario.predicate_position);
+ std::string name = "ParquetReader/" + reader_scenario_name(scenario);
benchmark::RegisterBenchmark(name.c_str(), [=](benchmark::State&
state) {
run_reader(state, scenario);
})->Unit(benchmark::kNanosecond);
diff --git a/be/benchmark/parquet/parquet_benchmark_scenarios.h
b/be/benchmark/parquet/parquet_benchmark_scenarios.h
index a01c955c6a2..1db6b3c8fd2 100644
--- a/be/benchmark/parquet/parquet_benchmark_scenarios.h
+++ b/be/benchmark/parquet/parquet_benchmark_scenarios.h
@@ -37,7 +37,14 @@ enum class Encoding {
enum class ValueType { INT32, INT64, FLOAT, DOUBLE, BYTE_ARRAY,
FIXED_LEN_BYTE_ARRAY };
enum class Pattern { CLUSTERED, ALTERNATING };
enum class Projection { PREDICATE_ONLY, PREDICATE_PROJECTED };
-enum class ReaderOperation { OPEN_TO_FIRST_BLOCK, FULL_SCAN, PREDICATE_SCAN,
LIMIT_1, LIMIT_1000 };
+enum class ReaderOperation {
+ OPEN_TO_FIRST_BLOCK,
+ FULL_SCAN,
+ PREDICATE_SCAN,
+ COMPLEX_RESIDUAL_SCAN,
+ LIMIT_1,
+ LIMIT_1000
+};
struct DecoderScenario {
Encoding encoding;
@@ -113,7 +120,8 @@ inline std::vector<ReaderScenario> reader_scenarios() {
.predicate_position = 0};
for (const auto operation :
{ReaderOperation::OPEN_TO_FIRST_BLOCK, ReaderOperation::FULL_SCAN,
- ReaderOperation::PREDICATE_SCAN, ReaderOperation::LIMIT_1,
ReaderOperation::LIMIT_1000}) {
+ ReaderOperation::PREDICATE_SCAN,
ReaderOperation::COMPLEX_RESIDUAL_SCAN,
+ ReaderOperation::LIMIT_1, ReaderOperation::LIMIT_1000}) {
auto scenario = baseline;
scenario.operation = operation;
add(scenario);
@@ -252,6 +260,8 @@ inline std::string to_string(ReaderOperation value) {
return "full_scan";
case ReaderOperation::PREDICATE_SCAN:
return "predicate_scan";
+ case ReaderOperation::COMPLEX_RESIDUAL_SCAN:
+ return "complex_residual_scan";
case ReaderOperation::LIMIT_1:
return "limit_1";
case ReaderOperation::LIMIT_1000:
@@ -260,4 +270,12 @@ inline std::string to_string(ReaderOperation value) {
return "unknown";
}
+inline std::string reader_scenario_name(const ReaderScenario& scenario) {
+ return to_string(scenario.operation) + "/" + to_string(scenario.encoding)
+ "/null_" +
+ std::to_string(scenario.null_percent) + "/" +
to_string(scenario.null_pattern) +
+ "/sel_" + std::to_string(scenario.selectivity_percent) + "/" +
+ to_string(scenario.projection) + "/width_" +
std::to_string(scenario.schema_width) +
+ "/predicate_" + std::to_string(scenario.predicate_position);
+}
+
} // namespace doris::parquet_benchmark
diff --git a/be/src/format_v2/expr/cast.h b/be/src/format_v2/expr/cast.h
index 1dc06bcf07f..22604455e50 100644
--- a/be/src/format_v2/expr/cast.h
+++ b/be/src/format_v2/expr/cast.h
@@ -53,6 +53,9 @@ public:
std::string debug_string() const override;
uint64_t get_digest(uint64_t seed) const override { return 0; }
const std::string& expr_name() const override { return _expr_name; }
+ // Ordinary CAST can fail for data-dependent input. Localization must
retain the same
+ // full-batch error behavior as VCastExpr instead of hiding errors in
previously rejected rows.
+ bool is_safe_to_execute_on_selected_rows() const override { return false; }
Status clone_node(VExprSPtr* cloned_expr) const override {
DORIS_CHECK(cloned_expr != nullptr);
*cloned_expr = Cast::create_shared(_data_type);
diff --git a/be/src/format_v2/parquet/parquet_profile.cpp
b/be/src/format_v2/parquet/parquet_profile.cpp
index b56a57d419e..9332dac0182 100644
--- a/be/src/format_v2/parquet/parquet_profile.cpp
+++ b/be/src/format_v2/parquet/parquet_profile.cpp
@@ -175,6 +175,8 @@ void ParquetProfile::init(RuntimeProfile* profile) {
TUnit::BYTES,
parquet_profile, 1);
predicate_compaction_count = ADD_CHILD_COUNTER_WITH_LEVEL(profile,
"PredicateCompactionCount",
TUnit::UNIT,
parquet_profile, 1);
+ predicate_alignment_columns = ADD_CHILD_COUNTER_WITH_LEVEL(profile,
"PredicateAlignmentColumns",
+ TUnit::UNIT,
parquet_profile, 1);
fixed_width_predicate_direct_batches = ADD_CHILD_COUNTER_WITH_LEVEL(
profile, "FixedWidthPredicateDirectBatches", TUnit::UNIT,
parquet_profile, 1);
fixed_width_predicate_direct_rows = ADD_CHILD_COUNTER_WITH_LEVEL(
@@ -316,6 +318,7 @@ ParquetScanProfile ParquetProfile::scan_profile() const {
.predicate_compaction_time = predicate_compaction_time,
.predicate_compaction_bytes = predicate_compaction_bytes,
.predicate_compaction_count = predicate_compaction_count,
+ .predicate_alignment_columns = predicate_alignment_columns,
.fixed_width_predicate_direct_batches =
fixed_width_predicate_direct_batches,
.fixed_width_predicate_direct_rows =
fixed_width_predicate_direct_rows,
.dict_filter_rewrite_time = dict_filter_rewrite_time,
diff --git a/be/src/format_v2/parquet/parquet_profile.h
b/be/src/format_v2/parquet/parquet_profile.h
index 170f14b56c5..438d1a9a4b2 100644
--- a/be/src/format_v2/parquet/parquet_profile.h
+++ b/be/src/format_v2/parquet/parquet_profile.h
@@ -88,6 +88,7 @@ struct ParquetScanProfile {
RuntimeProfile::Counter* predicate_compaction_time = nullptr;
RuntimeProfile::Counter* predicate_compaction_bytes = nullptr;
RuntimeProfile::Counter* predicate_compaction_count = nullptr;
+ RuntimeProfile::Counter* predicate_alignment_columns = nullptr;
RuntimeProfile::Counter* fixed_width_predicate_direct_batches = nullptr;
RuntimeProfile::Counter* fixed_width_predicate_direct_rows = nullptr;
RuntimeProfile::Counter* dict_filter_rewrite_time = nullptr; // dictionary
rewrite time (ns)
@@ -203,6 +204,7 @@ struct ParquetProfile {
RuntimeProfile::Counter* predicate_compaction_time = nullptr;
RuntimeProfile::Counter* predicate_compaction_bytes = nullptr;
RuntimeProfile::Counter* predicate_compaction_count = nullptr;
+ RuntimeProfile::Counter* predicate_alignment_columns = nullptr;
RuntimeProfile::Counter* fixed_width_predicate_direct_batches = nullptr;
RuntimeProfile::Counter* fixed_width_predicate_direct_rows = nullptr;
RuntimeProfile::Counter* dict_filter_rewrite_time = nullptr;
diff --git a/be/src/format_v2/parquet/parquet_scan.cpp
b/be/src/format_v2/parquet/parquet_scan.cpp
index b92a3c27054..3730f303254 100644
--- a/be/src/format_v2/parquet/parquet_scan.cpp
+++ b/be/src/format_v2/parquet/parquet_scan.cpp
@@ -22,6 +22,7 @@
#include <optional>
#include <ranges>
#include <set>
+#include <span>
#include <unordered_set>
#include <utility>
@@ -569,8 +570,8 @@ Status finalize_parquet_row_group_plans(
namespace {
-using DictionaryResidualConjunct = std::pair<VExprContextSPtr, VExprSPtr>;
-using DictionaryResidualConjuncts = std::vector<DictionaryResidualConjunct>;
+using OwnedExpressionConjunct = std::pair<VExprContextSPtr, VExprSPtr>;
+using OwnedExpressionConjuncts = std::vector<OwnedExpressionConjunct>;
void update_counter_if_not_null(RuntimeProfile::Counter* counter, int64_t
value) {
if (counter != nullptr) {
@@ -615,10 +616,9 @@ Status execute_compact_filter_conjuncts(const
VExprContextSPtrs& conjuncts, size
return Status::OK();
}
-Status execute_compact_dictionary_residual_conjuncts(const
DictionaryResidualConjuncts& conjuncts,
- size_t rows, Block*
file_block,
- IColumn::Filter*
compact_filter,
- bool* can_filter_all) {
+Status execute_compact_owned_conjuncts(std::span<const
OwnedExpressionConjunct> conjuncts,
+ size_t rows, Block* file_block,
+ IColumn::Filter* compact_filter, bool*
can_filter_all) {
DORIS_CHECK(compact_filter != nullptr);
DORIS_CHECK(can_filter_all != nullptr);
compact_filter->resize_fill(rows, 1);
@@ -890,6 +890,7 @@ void ParquetScanScheduler::reset() {
_predicate_schedule = {};
_predicate_positions_scratch.clear();
_predicate_indices_by_position_scratch.clear();
+ _materialized_predicate_positions_scratch.clear();
_ordered_predicate_positions_scratch.clear();
_predicate_batch_sequence = 0;
reset_current_row_group();
@@ -964,8 +965,10 @@ const detail::PredicateConjunctSchedule&
ParquetScanScheduler::predicate_conjunc
_predicate_schedule_request = &request;
_predicate_positions_scratch.clear();
_predicate_indices_by_position_scratch.clear();
+ _materialized_predicate_positions_scratch.clear();
_predicate_positions_scratch.reserve(request.predicate_columns.size());
_predicate_indices_by_position_scratch.reserve(request.predicate_columns.size());
+
_materialized_predicate_positions_scratch.reserve(request.predicate_columns.size());
for (size_t idx = 0; idx < request.predicate_columns.size(); ++idx) {
const auto position_it =
request.local_positions.find(request.predicate_columns[idx].column_id());
@@ -978,18 +981,42 @@ const detail::PredicateConjunctSchedule&
ParquetScanScheduler::predicate_conjunc
}
std::vector<format::LocalColumnIndex>
ParquetScanScheduler::adaptive_predicate_prefetch_columns(
- const format::FileScanRequest& request) const {
+ const format::FileScanRequest& request) {
std::vector<size_t> positions;
std::unordered_map<size_t, const format::LocalColumnIndex*>
columns_by_position;
- positions.reserve(request.predicate_columns.size());
columns_by_position.reserve(request.predicate_columns.size());
for (const auto& column : request.predicate_columns) {
const auto position_it =
request.local_positions.find(column.column_id());
DORIS_CHECK(position_it != request.local_positions.end());
const size_t position = position_it->second.value();
- positions.push_back(position);
columns_by_position.emplace(position, &column);
}
+ const auto& schedule = predicate_conjunct_schedule(request);
+ if (!schedule.supports_lazy_materialization) {
+ positions.reserve(request.predicate_columns.size());
+ for (const auto& column : request.predicate_columns) {
+
positions.push_back(request.local_positions.at(column.column_id()).value());
+ }
+ } else if (!schedule.single_column_conjuncts.empty()) {
+ positions.reserve(schedule.single_column_conjuncts.size());
+ for (const auto& column : request.predicate_columns) {
+ const size_t position =
request.local_positions.at(column.column_id()).value();
+ if (schedule.single_column_conjuncts.contains(position)) {
+ // Cold adaptive statistics intentionally preserve request
order; iterating the
+ // hash map here would make the first decoded predicate depend
on bucket layout.
+ positions.push_back(position);
+ }
+ }
+ } else if (!schedule.remaining_stages.empty()) {
+ // Match execution's first reachable stage. Warming columns owned only
by later residuals
+ // would turn lazy decode into eager remote IO before an earlier
conjunct can reject rows.
+ positions = schedule.remaining_stages.front().required_positions;
+ } else {
+ positions.reserve(request.predicate_columns.size());
+ for (const auto& column : request.predicate_columns) {
+
positions.push_back(request.local_positions.at(column.column_id()).value());
+ }
+ }
auto ordered = detail::order_adaptive_predicates(positions,
_predicate_runtime_stats);
ordered = detail::adaptive_prefetch_prefix(ordered,
_predicate_runtime_stats, 0.25);
std::vector<format::LocalColumnIndex> result;
@@ -1214,6 +1241,37 @@ Status
ParquetScanScheduler::flush_pending_non_predicate_skip_rows() {
namespace {
+bool append_residual_stages(const VExprContextSPtr& owner_context, const
VExprSPtr& expression,
+ const std::unordered_set<size_t>&
predicate_block_positions,
+ std::vector<detail::PredicateConjunctStage>*
stages) {
+ DORIS_CHECK(owner_context != nullptr);
+ DORIS_CHECK(expression != nullptr);
+ DORIS_CHECK(stages != nullptr);
+ const auto* compound_predicate = dynamic_cast<const
VCompoundPred*>(expression.get());
+ if (compound_predicate != nullptr && compound_predicate->op() ==
TExprOpcode::COMPOUND_AND) {
+ for (const auto& child : expression->children()) {
+ if (!append_residual_stages(owner_context, child,
predicate_block_positions, stages)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ std::set<int> referenced_positions;
+ expression->collect_slot_column_ids(referenced_positions);
+ auto& stage = stages->emplace_back();
+ stage.owner_context = owner_context;
+ stage.expression = expression;
+ for (const int position : referenced_positions) {
+ if (position < 0 ||
!predicate_block_positions.contains(cast_set<size_t>(position))) {
+ stages->pop_back();
+ return false;
+ }
+ stage.required_positions.push_back(cast_set<size_t>(position));
+ }
+ return true;
+}
+
detail::PredicateConjunctSchedule build_predicate_conjunct_schedule(
const format::FileScanRequest& request) {
std::unordered_set<size_t> predicate_block_positions;
@@ -1235,18 +1293,31 @@ detail::PredicateConjunctSchedule
build_predicate_conjunct_schedule(
// optimization, so any unsafe conjunct disables the per-column
schedule for the batch.
schedule.remaining_conjuncts = request.conjuncts;
schedule.single_column_conjuncts.clear();
+ schedule.remaining_stages.clear();
+ schedule.supports_lazy_materialization = false;
return schedule;
}
std::set<int> referenced_positions;
conjunct->root()->collect_slot_column_ids(referenced_positions);
if (referenced_positions.size() != 1) {
schedule.remaining_conjuncts.push_back(conjunct);
+ if (!append_residual_stages(conjunct, conjunct->root(),
predicate_block_positions,
+ &schedule.remaining_stages)) {
+ schedule.supports_lazy_materialization = false;
+ schedule.remaining_conjuncts = request.conjuncts;
+ schedule.single_column_conjuncts.clear();
+ schedule.remaining_stages.clear();
+ return schedule;
+ }
continue;
}
const auto block_position =
static_cast<size_t>(*referenced_positions.begin());
if (!predicate_block_positions.contains(block_position)) {
- schedule.remaining_conjuncts.push_back(conjunct);
- continue;
+ schedule.supports_lazy_materialization = false;
+ schedule.remaining_conjuncts = request.conjuncts;
+ schedule.single_column_conjuncts.clear();
+ schedule.remaining_stages.clear();
+ return schedule;
}
schedule.single_column_conjuncts[block_position].push_back(conjunct);
}
@@ -1280,7 +1351,7 @@ bool can_evaluate_dictionary_exactly(const VExprSPtr&
expr) {
}
void collect_dictionary_residual_exprs(const VExprContextSPtr& owner_context,
const VExprSPtr& expr,
- DictionaryResidualConjuncts*
residual_conjuncts) {
+ OwnedExpressionConjuncts*
residual_conjuncts) {
DORIS_CHECK(owner_context != nullptr);
DORIS_CHECK(expr != nullptr);
DORIS_CHECK(residual_conjuncts != nullptr);
@@ -1304,9 +1375,8 @@ void collect_dictionary_residual_exprs(const
VExprContextSPtr& owner_context, co
residual_conjuncts->emplace_back(owner_context, expr);
}
-DictionaryResidualConjuncts build_dictionary_residual_conjuncts(
- const VExprContextSPtrs& conjuncts) {
- DictionaryResidualConjuncts residual_conjuncts;
+OwnedExpressionConjuncts build_dictionary_residual_conjuncts(const
VExprContextSPtrs& conjuncts) {
+ OwnedExpressionConjuncts residual_conjuncts;
for (const auto& conjunct : conjuncts) {
DORIS_CHECK(conjunct != nullptr);
collect_dictionary_residual_exprs(conjunct, conjunct->root(),
&residual_conjuncts);
@@ -1426,7 +1496,7 @@ Status
ParquetScanScheduler::prepare_current_dictionary_filters(
// VCompoundPred intentionally evaluates only dictionary-capable
children, so residual
// predicates still run later on surviving rows.
IColumn::Filter dictionary_filter;
- DictionaryResidualConjuncts residual_conjuncts;
+ OwnedExpressionConjuncts residual_conjuncts;
{
SCOPED_TIMER(_scan_profile.dict_filter_build_time);
dictionary_filter = build_dictionary_entry_filter(
@@ -1471,14 +1541,18 @@ Status
ParquetScanScheduler::read_filter_columns(int64_t batch_rows,
remember_residual_positions(schedule.remaining_conjuncts);
remember_residual_positions(request.delete_conjuncts);
const size_t predicate_batch_sequence = _predicate_batch_sequence++;
- const bool can_read_predicate_columns_round_by_round =
- !schedule.single_column_conjuncts.empty();
+ const bool can_read_predicate_columns_round_by_round =
schedule.supports_lazy_materialization;
auto& read_column_positions = _read_column_positions_scratch;
read_column_positions.clear();
read_column_positions.reserve(request.predicate_columns.size());
+ auto& materialized_positions = _materialized_predicate_positions_scratch;
+ materialized_positions.clear();
for (auto& rows : _predicate_column_selection_scratch |
std::views::values) {
rows.clear();
}
+ // A generation becomes dirty only when filtering changes SelectionVector.
Columns read after
+ // an all-pass stage already share its coordinates, so rewalking every
prior mapping is wasted.
+ bool predicate_columns_need_alignment = false;
auto remember_column_selection = [&](uint32_t position) {
auto& rows = _predicate_column_selection_scratch[position];
@@ -1493,6 +1567,8 @@ Status ParquetScanScheduler::read_filter_columns(int64_t
batch_rows,
auto compact_predicate_columns = [&](bool discard_predicate_only_payload)
-> Status {
bool compacted = false;
int64_t compacted_bytes = 0;
+ update_counter_if_not_null(_scan_profile.predicate_alignment_columns,
+
cast_set<int64_t>(read_column_positions.size()));
for (const uint32_t position : read_column_positions) {
auto& source_rows = _predicate_column_selection_scratch[position];
const auto& old_column =
file_block->get_by_position(position).column;
@@ -1705,6 +1781,7 @@ Status ParquetScanScheduler::read_filter_columns(int64_t
batch_rows,
static_cast<int64_t>(new_selected_rows);
}
if (new_selected_rows != selected_rows_before) {
+ predicate_columns_need_alignment = true;
*selected_rows = can_filter_all
? 0
:
apply_compact_filter_to_selection(compact_filter, selection,
@@ -1713,16 +1790,16 @@ Status
ParquetScanScheduler::read_filter_columns(int64_t batch_rows,
return Status::OK();
};
- auto execute_scheduled_dictionary_residual_conjuncts =
- [&](const DictionaryResidualConjuncts& conjuncts) -> Status {
+ auto execute_scheduled_owned_conjuncts =
+ [&](std::span<const OwnedExpressionConjunct> conjuncts) -> Status {
if (conjuncts.empty() || *selected_rows == 0) {
return Status::OK();
}
const uint16_t selected_rows_before = *selected_rows;
IColumn::Filter compact_filter;
bool can_filter_all = false;
- RETURN_IF_ERROR(execute_compact_dictionary_residual_conjuncts(
- conjuncts, selected_rows_before, file_block, &compact_filter,
&can_filter_all));
+ RETURN_IF_ERROR(execute_compact_owned_conjuncts(conjuncts,
selected_rows_before, file_block,
+ &compact_filter,
&can_filter_all));
if (can_filter_all) {
compact_filter.resize_fill(selected_rows_before, 0);
}
@@ -1732,6 +1809,7 @@ Status ParquetScanScheduler::read_filter_columns(int64_t
batch_rows,
static_cast<int64_t>(new_selected_rows);
}
if (new_selected_rows != selected_rows_before) {
+ predicate_columns_need_alignment = true;
*selected_rows = can_filter_all
? 0
:
apply_compact_filter_to_selection(compact_filter, selection,
@@ -1749,13 +1827,13 @@ Status
ParquetScanScheduler::read_filter_columns(int64_t batch_rows,
return execute_scheduled_conjuncts(conjuncts);
};
- auto execute_scheduled_dictionary_residual_conjuncts_with_profile =
- [&](const DictionaryResidualConjuncts& conjuncts) -> Status {
+ auto execute_scheduled_owned_conjuncts_with_profile =
+ [&](std::span<const OwnedExpressionConjunct> conjuncts) -> Status {
if (_scan_profile.predicate_filter_time == nullptr) {
- return execute_scheduled_dictionary_residual_conjuncts(conjuncts);
+ return execute_scheduled_owned_conjuncts(conjuncts);
}
SCOPED_TIMER(_scan_profile.predicate_filter_time);
- return execute_scheduled_dictionary_residual_conjuncts(conjuncts);
+ return execute_scheduled_owned_conjuncts(conjuncts);
};
auto execute_scheduled_delete_conjuncts = [&]() -> Status {
@@ -1772,6 +1850,7 @@ Status ParquetScanScheduler::read_filter_columns(int64_t
batch_rows,
compact_filter.resize_fill(selected_rows_before, 0);
}
if (can_filter_all || count_selected_rows(compact_filter) !=
selected_rows_before) {
+ predicate_columns_need_alignment = true;
*selected_rows = can_filter_all
? 0
:
apply_compact_filter_to_selection(compact_filter, selection,
@@ -1789,6 +1868,7 @@ Status ParquetScanScheduler::read_filter_columns(int64_t
batch_rows,
RETURN_IF_ERROR(read_predicate_column(column_reader.get(),
position_it->second.value(),
fid, nullptr,
&used_dictionary_filter,
&used_fixed_width_filter));
+ materialized_positions.insert(position_it->second.value());
}
return Status::OK();
};
@@ -1808,8 +1888,18 @@ Status ParquetScanScheduler::read_filter_columns(int64_t
batch_rows,
// Single-column conjuncts can be evaluated immediately after their
column is read. Once
// selection shrinks, later predicate columns use
ParquetColumnReader::select() so the
// reader skips rows already rejected by earlier predicates instead of
materializing them.
+ _ordered_predicate_positions_scratch.clear();
+
_ordered_predicate_positions_scratch.reserve(schedule.single_column_conjuncts.size());
+ for (const auto& column : request.predicate_columns) {
+ const size_t position =
request.local_positions.at(column.column_id()).value();
+ if (schedule.single_column_conjuncts.contains(position)) {
+ // The request order is the stable cold-start policy until
measured costs can
+ // reorder predicates; unordered-map iteration can defeat an
early selective filter.
+ _ordered_predicate_positions_scratch.push_back(position);
+ }
+ }
_ordered_predicate_positions_scratch =
detail::order_adaptive_predicates(
- _predicate_positions_scratch, _predicate_runtime_stats);
+ _ordered_predicate_positions_scratch,
_predicate_runtime_stats);
const auto& ordered_positions = _ordered_predicate_positions_scratch;
for (size_t order_idx = 0; order_idx < ordered_positions.size();
++order_idx) {
const size_t position = ordered_positions[order_idx];
@@ -1835,16 +1925,20 @@ Status
ParquetScanScheduler::read_filter_columns(int64_t batch_rows,
RETURN_IF_ERROR(read_predicate_column(reader_it->second.get(),
block_position, fid,
column_conjuncts,
&used_dictionary_filter,
&used_fixed_width_filter));
+ materialized_positions.insert(block_position);
if (*selected_rows != 0 && conjunct_it !=
schedule.single_column_conjuncts.end()) {
if (used_dictionary_filter) {
const auto residual_it =
_current_dictionary_residual_conjuncts.find(fid);
DORIS_CHECK(residual_it !=
_current_dictionary_residual_conjuncts.end());
-
RETURN_IF_ERROR(execute_scheduled_dictionary_residual_conjuncts_with_profile(
- residual_it->second));
+ RETURN_IF_ERROR(
+
execute_scheduled_owned_conjuncts_with_profile(residual_it->second));
} else if (!used_fixed_width_filter) {
RETURN_IF_ERROR(execute_scheduled_conjuncts_with_profile(conjunct_it->second));
}
}
+ if (*selected_rows != rows_before) {
+ predicate_columns_need_alignment = true;
+ }
if (sample) {
const double cost_per_row =
static_cast<double>(MonotonicNanos() - start_ns) /
std::max<uint16_t>(rows_before, 1);
@@ -1866,39 +1960,103 @@ Status
ParquetScanScheduler::read_filter_columns(int64_t batch_rows,
if (*selected_rows != 0) {
continue;
}
- for (size_t remaining_order_idx = order_idx + 1;
- remaining_order_idx < ordered_positions.size();
++remaining_order_idx) {
- const size_t remaining_idx =
_predicate_indices_by_position_scratch.at(
- ordered_positions[remaining_order_idx]);
- const auto remaining_fid =
request.predicate_columns[remaining_idx].column_id();
- auto remaining_reader_it =
_current_predicate_columns.find(remaining_fid);
- DORIS_CHECK(remaining_reader_it !=
_current_predicate_columns.end());
- RETURN_IF_ERROR(remaining_reader_it->second->skip(batch_rows));
- }
return Status::OK();
}
return Status::OK();
};
+ auto materialize_predicate_positions = [&](const std::vector<size_t>&
positions) -> Status {
+ for (const size_t position : positions) {
+ if (materialized_positions.contains(position)) {
+ continue;
+ }
+ const auto index_it =
_predicate_indices_by_position_scratch.find(position);
+ DORIS_CHECK(index_it !=
_predicate_indices_by_position_scratch.end());
+ const auto fid =
request.predicate_columns[index_it->second].column_id();
+ const auto reader_it = _current_predicate_columns.find(fid);
+ DORIS_CHECK(reader_it != _current_predicate_columns.end());
+ bool used_dictionary_filter = false;
+ bool used_fixed_width_filter = false;
+ RETURN_IF_ERROR(read_predicate_column(reader_it->second.get(),
position, fid, nullptr,
+ &used_dictionary_filter,
+ &used_fixed_width_filter));
+ materialized_positions.insert(position);
+ }
+ return Status::OK();
+ };
+
+ auto skip_unmaterialized_predicate_columns = [&]() -> Status {
+ for (const auto& col : request.predicate_columns) {
+ const auto position_it =
request.local_positions.find(col.column_id());
+ DORIS_CHECK(position_it != request.local_positions.end());
+ if (materialized_positions.contains(position_it->second.value())) {
+ continue;
+ }
+ const auto reader_it =
_current_predicate_columns.find(col.column_id());
+ DORIS_CHECK(reader_it != _current_predicate_columns.end());
+ RETURN_IF_ERROR(reader_it->second->skip(batch_rows));
+ }
+ // Every skipped column has an empty payload in the block. Suppress
the caller's
+ // batch-coordinate filter because there is no materialized
batch-sized column left.
+ *predicate_columns_filtered = true;
+ return Status::OK();
+ };
+
auto compact_predicate_columns_with_profile =
[&](bool discard_predicate_only_payload) -> Status {
+ if (!discard_predicate_only_payload &&
!predicate_columns_need_alignment) {
+ return Status::OK();
+ }
const int64_t start_ns = MonotonicNanos();
auto status =
compact_predicate_columns(discard_predicate_only_payload);
update_counter_if_not_null(_scan_profile.predicate_compaction_time,
MonotonicNanos() - start_ns);
+ if (status.ok()) {
+ predicate_columns_need_alignment = false;
+ }
return status;
};
RETURN_IF_ERROR(read_round_by_round());
- // Single-column expressions only touch the just-read column, so earlier
columns can retain
- // their own row mappings. Compact only when a later expression needs a
shared coordinate
- // space; otherwise the final boundary can discard hidden predicate
payloads without scanning
- // them again.
- if (!schedule.remaining_conjuncts.empty()) {
+ if (*selected_rows == 0) {
+ RETURN_IF_ERROR(skip_unmaterialized_predicate_columns());
+ return compact_predicate_columns_with_profile(true);
+ }
+
+ // Complex residuals keep their original conjunct order. Materialize only
the columns needed
+ // by the next reachable expression, then compact previously read columns
into the same row
+ // space before evaluating it. This is the scanner-side equivalent of
expression-triggered
+ // lazy columns: a conjunct that rejects the batch prevents later-only
columns from decoding.
+ for (const auto& stage : schedule.remaining_stages) {
+
RETURN_IF_ERROR(materialize_predicate_positions(stage.required_positions));
RETURN_IF_ERROR(compact_predicate_columns_with_profile(false));
+ const OwnedExpressionConjunct stage_conjunct {stage.owner_context,
stage.expression};
+ RETURN_IF_ERROR(execute_scheduled_owned_conjuncts_with_profile(
+ std::span<const OwnedExpressionConjunct>(&stage_conjunct, 1)));
+ if (*selected_rows == 0) {
+ RETURN_IF_ERROR(skip_unmaterialized_predicate_columns());
+ return compact_predicate_columns_with_profile(true);
+ }
}
-
RETURN_IF_ERROR(execute_scheduled_conjuncts_with_profile(schedule.remaining_conjuncts));
+
if (!request.delete_conjuncts.empty()) {
+ std::set<int> delete_positions;
+ for (const auto& conjunct : request.delete_conjuncts) {
+ DORIS_CHECK(conjunct != nullptr && conjunct->root() != nullptr);
+ conjunct->root()->collect_slot_column_ids(delete_positions);
+ }
+ std::vector<size_t> required_delete_positions;
+ required_delete_positions.reserve(delete_positions.size());
+ for (const int position : delete_positions) {
+ DORIS_CHECK(position >= 0);
+ required_delete_positions.push_back(cast_set<size_t>(position));
+ }
+ if (required_delete_positions.empty() &&
!_predicate_positions_scratch.empty()) {
+ // An all-literal equality-delete predicate has no slot
dependency, but its hidden
+ // row-count carrier must still be materialized so the result
matches selected_rows.
+
required_delete_positions.push_back(_predicate_positions_scratch.front());
+ }
+
RETURN_IF_ERROR(materialize_predicate_positions(required_delete_positions));
RETURN_IF_ERROR(compact_predicate_columns_with_profile(false));
}
if (_scan_profile.predicate_filter_time == nullptr) {
@@ -1907,6 +2065,11 @@ Status ParquetScanScheduler::read_filter_columns(int64_t
batch_rows,
SCOPED_TIMER(_scan_profile.predicate_filter_time);
RETURN_IF_ERROR(execute_scheduled_delete_conjuncts());
}
+ if (*selected_rows == 0) {
+ RETURN_IF_ERROR(skip_unmaterialized_predicate_columns());
+ return compact_predicate_columns_with_profile(true);
+ }
+
RETURN_IF_ERROR(materialize_predicate_positions(_predicate_positions_scratch));
return compact_predicate_columns_with_profile(true);
}
diff --git a/be/src/format_v2/parquet/parquet_scan.h
b/be/src/format_v2/parquet/parquet_scan.h
index ac883cbbcbe..f0e202d9942 100644
--- a/be/src/format_v2/parquet/parquet_scan.h
+++ b/be/src/format_v2/parquet/parquet_scan.h
@@ -58,9 +58,17 @@ struct ParquetScanRange;
class NativeParquetMetadata;
namespace detail {
+struct PredicateConjunctStage {
+ VExprContextSPtr owner_context;
+ VExprSPtr expression;
+ std::vector<size_t> required_positions;
+};
+
struct PredicateConjunctSchedule {
std::map<size_t, VExprContextSPtrs> single_column_conjuncts;
VExprContextSPtrs remaining_conjuncts;
+ std::vector<PredicateConjunctStage> remaining_stages;
+ bool supports_lazy_materialization = true;
};
struct AdaptivePredicateStats {
@@ -208,7 +216,7 @@ private:
const detail::PredicateConjunctSchedule& predicate_conjunct_schedule(
const format::FileScanRequest& request);
std::vector<format::LocalColumnIndex> adaptive_predicate_prefetch_columns(
- const format::FileScanRequest& request) const;
+ const format::FileScanRequest& request);
Status open_next_row_group(ParquetFileContext& file_context,
const
std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
@@ -305,6 +313,7 @@ private:
detail::PredicateConjunctSchedule _predicate_schedule;
std::vector<size_t> _predicate_positions_scratch;
std::unordered_map<size_t, size_t> _predicate_indices_by_position_scratch;
+ std::unordered_set<size_t> _materialized_predicate_positions_scratch;
std::vector<size_t> _ordered_predicate_positions_scratch;
std::unordered_map<uint32_t, std::vector<SelectionVector::Index>>
_predicate_column_selection_scratch;
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 490a0bde204..45c1fe9dd44 100644
--- a/be/test/format_v2/parquet/parquet_benchmark_scenarios_test.cpp
+++ b/be/test/format_v2/parquet/parquet_benchmark_scenarios_test.cpp
@@ -22,6 +22,7 @@
#include <algorithm>
#include <array>
#include <set>
+#include <string>
#include <tuple>
namespace doris::parquet_benchmark {
@@ -86,7 +87,8 @@ TEST(ParquetBenchmarkScenariosTest,
ReaderMatrixCoversOperationsEncodingsAndSche
const auto scenarios = reader_scenarios();
for (const auto operation :
{ReaderOperation::OPEN_TO_FIRST_BLOCK, ReaderOperation::FULL_SCAN,
- ReaderOperation::PREDICATE_SCAN, ReaderOperation::LIMIT_1,
ReaderOperation::LIMIT_1000}) {
+ ReaderOperation::PREDICATE_SCAN,
ReaderOperation::COMPLEX_RESIDUAL_SCAN,
+ ReaderOperation::LIMIT_1, ReaderOperation::LIMIT_1000}) {
EXPECT_TRUE(std::ranges::any_of(scenarios, [&](const ReaderScenario&
scenario) {
return scenario.operation == operation;
}));
@@ -109,6 +111,26 @@ TEST(ParquetBenchmarkScenariosTest,
ReaderMatrixCoversOperationsEncodingsAndSche
}
}
+TEST(ParquetBenchmarkScenariosTest,
ReaderMatrixHasExactUniqueRegistrationNames) {
+ const auto scenarios = reader_scenarios();
+ EXPECT_EQ(scenarios.size(), 152);
+
+ std::set<std::string> names;
+ for (const auto& scenario : scenarios) {
+ names.insert(reader_scenario_name(scenario));
+ }
+ EXPECT_EQ(names.size(), scenarios.size());
+}
+
+TEST(ParquetBenchmarkScenariosTest,
ReaderMatrixCoversComplexResidualLazyMaterialization) {
+ const auto scenarios = reader_scenarios();
+ EXPECT_TRUE(std::ranges::any_of(scenarios, [](const ReaderScenario&
scenario) {
+ return scenario.operation == ReaderOperation::COMPLEX_RESIDUAL_SCAN &&
+ scenario.encoding == Encoding::PLAIN &&
scenario.selectivity_percent == 10 &&
+ scenario.schema_width == 32;
+ }));
+}
+
TEST(ParquetBenchmarkScenariosTest, ReaderMatrixCoversFixedWidthRawFilterAxes)
{
const auto scenarios = reader_scenarios();
for (const auto encoding : {Encoding::BYTE_STREAM_SPLIT,
Encoding::DELTA_BINARY_PACKED}) {
diff --git a/be/test/format_v2/parquet/parquet_scan_test.cpp
b/be/test/format_v2/parquet/parquet_scan_test.cpp
index 9bdb0c7562f..a4e5abb91a4 100644
--- a/be/test/format_v2/parquet/parquet_scan_test.cpp
+++ b/be/test/format_v2/parquet/parquet_scan_test.cpp
@@ -46,11 +46,13 @@
#include "core/data_type/data_type_number.h"
#include "core/data_type/data_type_string.h"
#include "core/field.h"
+#include "exprs/vcompound_pred.h"
#include "exprs/vectorized_fn_call.h"
#include "exprs/vexpr.h"
#include "exprs/vexpr_context.h"
#include "exprs/vliteral.h"
#include "exprs/vslot_ref.h"
+#include "format_v2/expr/cast.h"
#include "format_v2/expr/delete_predicate.h"
#include "format_v2/file_reader.h"
#include "format_v2/parquet/parquet_column_schema.h"
@@ -232,6 +234,8 @@ public:
const std::string& expr_name() const override { return _expr_name; }
+ bool is_constant() const override { return false; }
+
Status execute_column_impl(VExprContext*, const Block* block, const
Selector* selector,
size_t count, ColumnPtr& result_column) const
override {
DORIS_CHECK(block != nullptr);
@@ -470,6 +474,59 @@ VExprContextSPtr create_int32_pair_sum_conjunct(int
left_column_id, int right_co
std::make_shared<Int32PairSumExpr>(left_column_id,
right_column_id, upper_bound));
}
+VExprContextSPtr create_and_conjunct(VExprSPtr left, VExprSPtr right) {
+ const auto result_type = left->data_type()->is_nullable() ||
right->data_type()->is_nullable()
+ ?
make_nullable(std::make_shared<DataTypeUInt8>())
+ : std::make_shared<DataTypeUInt8>();
+ TExprNode node;
+ node.__set_node_type(TExprNodeType::COMPOUND_PRED);
+ node.__set_opcode(TExprOpcode::COMPOUND_AND);
+ node.__set_type(result_type->to_thrift());
+ node.__set_num_children(2);
+ node.__set_is_nullable(result_type->is_nullable());
+ auto compound = VCompoundPred::create_shared(node);
+ compound->add_child(std::move(left));
+ compound->add_child(std::move(right));
+ return VExprContext::create_shared(std::move(compound));
+}
+
+VExprSPtr create_binary_predicate(const std::string& function_name,
TExprOpcode::type opcode,
+ VExprSPtr left, VExprSPtr right) {
+ const auto result_type = left->data_type()->is_nullable() ||
right->data_type()->is_nullable()
+ ?
make_nullable(std::make_shared<DataTypeUInt8>())
+ : std::make_shared<DataTypeUInt8>();
+ TFunctionName name;
+ name.__set_function_name(function_name);
+ TFunction function;
+ function.__set_name(name);
+ function.__set_binary_type(TFunctionBinaryType::BUILTIN);
+ function.__set_arg_types({left->data_type()->to_thrift(),
right->data_type()->to_thrift()});
+ function.__set_ret_type(result_type->to_thrift());
+ function.__set_has_var_args(false);
+ TExprNode node;
+ node.__set_node_type(TExprNodeType::BINARY_PRED);
+ node.__set_opcode(opcode);
+ node.__set_type(result_type->to_thrift());
+ node.__set_fn(function);
+ node.__set_num_children(2);
+ node.__set_is_nullable(result_type->is_nullable());
+ auto predicate = VectorizedFnCall::create_shared(node);
+ predicate->add_child(std::move(left));
+ predicate->add_child(std::move(right));
+ return predicate;
+}
+
+VExprSPtr create_int32_slot_comparison(const std::string& function_name,
TExprOpcode::type opcode,
+ int left_column_id, int right_column_id,
+ const DataTypePtr& type) {
+ return create_binary_predicate(
+ function_name, opcode,
+ VSlotRef::create_shared(left_column_id, left_column_id, -1, type,
+ "c" + std::to_string(left_column_id)),
+ VSlotRef::create_shared(right_column_id, right_column_id, -1, type,
+ "c" + std::to_string(right_column_id)));
+}
+
VExprContextSPtr create_int32_direct_greater_conjunct(int column_id, int32_t
lower_bound) {
return VExprContext::create_shared(
std::make_shared<Int32DirectGreaterExpr>(column_id, lower_bound));
@@ -635,6 +692,41 @@ void write_int_pair_parquet_file(const std::string&
file_path, int64_t row_group
write_table(file_path, table, row_group_size, false, false,
enable_statistics, encoding);
}
+void write_int_triple_parquet_file(const std::string& file_path) {
+ auto schema = arrow::schema({
+ arrow::field("left", arrow::int32(), false),
+ arrow::field("middle", arrow::int32(), false),
+ arrow::field("right", arrow::int32(), false),
+ });
+ auto table = arrow::Table::Make(schema, {build_int32_array({1, 2, 3, 4, 5,
6}),
+ build_int32_array({10, 20, 30, 0,
0, 0}),
+ build_int32_array({100, 200, 300,
400, 500, 600})});
+ write_table(file_path, table, 6, false, false, false);
+}
+
+void write_int_columns_parquet_file(const std::string& file_path, int
column_count) {
+ std::vector<std::shared_ptr<arrow::Field>> fields;
+ std::vector<std::shared_ptr<arrow::Array>> columns;
+ fields.reserve(column_count);
+ columns.reserve(column_count);
+ for (int column = 0; column < column_count; ++column) {
+ fields.push_back(arrow::field("c" + std::to_string(column),
arrow::int32(), false));
+ columns.push_back(build_int32_array({1, 2, 3, 4, 5, 6}));
+ }
+ write_table(file_path,
arrow::Table::Make(arrow::schema(std::move(fields)), std::move(columns)),
+ 6, false, false, false);
+}
+
+void write_int_pair_and_string_parquet_file(const std::string& file_path) {
+ auto schema = arrow::schema({arrow::field("left", arrow::int32(), false),
+ arrow::field("right", arrow::int32(), false),
+ arrow::field("text", arrow::utf8(), false)});
+ auto table =
+ arrow::Table::Make(schema, {build_int32_array({1, 2, 3}),
build_int32_array({0, 3, 4}),
+ build_string_array({"bad", "2",
"3"})});
+ write_table(file_path, table, 3, false, false, false);
+}
+
void write_uint32_pair_parquet_file(const std::string& file_path) {
auto schema = arrow::schema({arrow::field("id", arrow::uint32(), false),
arrow::field("score", arrow::int32(),
false)});
@@ -1349,8 +1441,7 @@ TEST_F(ParquetScanTest,
PredicateOnlyGlobalRowIdKeepsSignedFileLocalId) {
Block block = build_file_block(schema);
size_t rows = 0;
bool eof = false;
- const auto status = reader->get_block(&block, &rows, &eof);
- ASSERT_TRUE(status.ok()) << status;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
EXPECT_EQ(rows, 6);
EXPECT_EQ(int32_data_column(*block.get_by_position(0).column).get_data(),
(ColumnInt32::Container {1, 2, 3, 4, 5, 6}));
@@ -1372,7 +1463,8 @@ TEST_F(ParquetScanTest,
EmptyScanPlanReturnsEofWithoutReadingColumns) {
Block block = build_file_block(schema);
size_t rows = 0;
bool eof = false;
- ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ const auto status = reader->get_block(&block, &rows, &eof);
+ ASSERT_TRUE(status.ok()) << status;
EXPECT_EQ(rows, 0);
EXPECT_TRUE(eof);
}
@@ -1485,6 +1577,149 @@ TEST_F(ParquetScanTest,
PredicateColumnsSkipUnreadColumnsWhenFirstPredicateFilte
EXPECT_EQ(counter_value(profile, "ReaderSkipRows"), 6);
}
+TEST_F(ParquetScanTest,
ComplexResidualSkipsColumnsAfterEarlierAndChildFiltersAll) {
+ write_int_triple_parquet_file(_file_path);
+ RuntimeProfile profile("profile");
+ auto reader = create_reader(0, -1, &profile);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ format::FileScanRequestBuilder request_builder(request.get());
+
ASSERT_TRUE(request_builder.add_predicate_column(format::LocalColumnId(0)).ok());
+
ASSERT_TRUE(request_builder.add_predicate_column(format::LocalColumnId(1)).ok());
+
ASSERT_TRUE(request_builder.add_predicate_column(format::LocalColumnId(2)).ok());
+ auto context = create_and_conjunct(
+ create_int32_slot_comparison("eq", TExprOpcode::EQ, 0, 1,
schema[0].type),
+ create_int32_slot_comparison("lt", TExprOpcode::LT, 1, 2,
schema[1].type));
+ ASSERT_TRUE(context->prepare(&state, RowDescriptor()).ok());
+ ASSERT_TRUE(context->open(&state).ok());
+ request->conjuncts.push_back(context);
+ ASSERT_TRUE(reader->open(request).ok());
+
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ EXPECT_EQ(rows, 0);
+ EXPECT_EQ(counter_value(profile, "ReaderReadRows"), 12);
+ EXPECT_EQ(counter_value(profile, "ReaderSelectRows"), 0);
+ EXPECT_EQ(counter_value(profile, "ReaderSkipRows"), 6);
+ context->close();
+}
+
+TEST_F(ParquetScanTest, ComplexResidualSelectsLaterColumnsForSurvivingRows) {
+ write_int_triple_parquet_file(_file_path);
+ RuntimeProfile profile("profile");
+ auto reader = create_reader(0, -1, &profile);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ format::FileScanRequestBuilder request_builder(request.get());
+
ASSERT_TRUE(request_builder.add_predicate_column(format::LocalColumnId(0)).ok());
+
ASSERT_TRUE(request_builder.add_predicate_column(format::LocalColumnId(1)).ok());
+
ASSERT_TRUE(request_builder.add_predicate_column(format::LocalColumnId(2)).ok());
+ auto context = create_and_conjunct(
+ create_int32_slot_comparison("gt", TExprOpcode::GT, 1, 0,
schema[1].type),
+ create_int32_slot_comparison("lt", TExprOpcode::LT, 1, 2,
schema[1].type));
+ ASSERT_TRUE(context->prepare(&state, RowDescriptor()).ok());
+ ASSERT_TRUE(context->open(&state).ok());
+ request->conjuncts.push_back(context);
+ ASSERT_TRUE(reader->open(request).ok());
+
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ ASSERT_EQ(rows, 3);
+ EXPECT_EQ(int32_data_column(*block.get_by_position(0).column).get_data(),
+ (ColumnInt32::Container {1, 2, 3}));
+ EXPECT_EQ(int32_data_column(*block.get_by_position(2).column).get_data(),
+ (ColumnInt32::Container {100, 200, 300}));
+ EXPECT_EQ(counter_value(profile, "ReaderReadRows"), 15);
+ EXPECT_EQ(counter_value(profile, "ReaderSelectRows"), 3);
+ EXPECT_EQ(counter_value(profile, "ReaderSkipRows"), 3);
+ context->close();
+}
+
+TEST_F(ParquetScanTest, AllPassResidualChainAlignsEachColumnOnce) {
+ write_int_columns_parquet_file(_file_path, 6);
+ RuntimeProfile profile("profile");
+ auto reader = create_reader(0, -1, &profile);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ format::FileScanRequestBuilder request_builder(request.get());
+ for (int column = 0; column < 6; ++column) {
+
ASSERT_TRUE(request_builder.add_predicate_column(format::LocalColumnId(column)).ok());
+ }
+ for (int column = 0; column < 6; column += 2) {
+ auto context =
VExprContext::create_shared(create_int32_slot_comparison(
+ "eq", TExprOpcode::EQ, column, column + 1,
schema[column].type));
+ ASSERT_TRUE(context->prepare(&state, RowDescriptor()).ok());
+ ASSERT_TRUE(context->open(&state).ok());
+ request->conjuncts.push_back(std::move(context));
+ }
+ ASSERT_TRUE(reader->open(request).ok());
+
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ bool eof = false;
+ const auto status = reader->get_block(&block, &rows, &eof);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(rows, 6);
+ auto* alignment_columns = profile.get_counter("PredicateAlignmentColumns");
+ ASSERT_NE(alignment_columns, nullptr);
+ EXPECT_EQ(alignment_columns->value(), 6);
+}
+
+TEST_F(ParquetScanTest, LocalizedStrictCastPreservesRejectedRowError) {
+ write_int_pair_and_string_parquet_file(_file_path);
+ RuntimeProfile profile("profile");
+ auto reader = create_reader(0, -1, &profile);
+ TQueryOptions query_options;
+ query_options.__set_enable_strict_cast(true);
+ RuntimeState state {query_options, TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ format::FileScanRequestBuilder request_builder(request.get());
+ for (int column = 0; column < 3; ++column) {
+
ASSERT_TRUE(request_builder.add_predicate_column(format::LocalColumnId(column)).ok());
+ }
+
+ auto cast =
format::Cast::create_shared(make_nullable(std::make_shared<DataTypeInt32>()));
+ cast->add_child(VSlotRef::create_shared(2, 2, -1, schema[2].type, "text"));
+ auto first = create_int32_slot_comparison("lt", TExprOpcode::LT, 0, 1,
schema[0].type);
+ auto second =
+ create_binary_predicate("eq", TExprOpcode::EQ, std::move(cast),
+
VLiteral::create_shared(std::make_shared<DataTypeInt32>(),
+
Field::create_field<TYPE_INT>(2)));
+ auto context = create_and_conjunct(std::move(first), std::move(second));
+ EXPECT_FALSE(context->root()->is_safe_to_execute_on_selected_rows());
+ ASSERT_TRUE(context->prepare(&state, RowDescriptor()).ok());
+ ASSERT_TRUE(context->open(&state).ok());
+ request->conjuncts.push_back(context);
+ ASSERT_TRUE(reader->open(request).ok());
+
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ bool eof = false;
+ const auto status = reader->get_block(&block, &rows, &eof);
+ EXPECT_FALSE(status.ok());
+ context->close();
+}
+
TEST_F(ParquetScanTest, PredicateOnlyColumnDropsPayloadAfterFiltering) {
write_int_pair_parquet_file(_file_path, 6, false);
RuntimeProfile profile("profile");
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
