github-actions[bot] commented on code in PR #66412:
URL: https://github.com/apache/doris/pull/66412#discussion_r3709424578
##########
be/src/format_v2/parquet/parquet_scan.cpp:
##########
@@ -2421,6 +2759,121 @@ Status
ParquetScanScheduler::read_filter_columns(int64_t batch_rows,
return status;
};
+ auto execute_raw_disjunction = [&](const detail::PredicateConjunctStage&
stage,
+ bool* applied) -> Status {
+ DORIS_CHECK(applied != nullptr);
+ *applied = false;
+ if (stage.raw_disjunction_branches.empty() || *selected_rows == 0) {
+ return Status::OK();
+ }
+ auto readers_it =
_current_raw_disjunction_readers.find(stage.expression.get());
+ if (readers_it == _current_raw_disjunction_readers.end()) {
+ return Status::OK();
+ }
+
+ const uint16_t selected_rows_before = *selected_rows;
+ // PODArray resize preserves retained bytes, but every OR group must
start from FALSE or a
+ // survivor bit from the previous batch could admit a row that matches
no current branch.
+ _raw_disjunction_filter_scratch.resize(selected_rows_before);
+ std::ranges::fill(_raw_disjunction_filter_scratch, 0);
+ size_t dictionary_branches = 0;
+ size_t raw_value_branches = 0;
+ size_t fixed_width_branches = 0;
+ for (auto& branch : readers_it->second) {
+ ParquetColumnReader* branch_reader = branch.reader.get();
+ if (branch_reader == nullptr) {
+ const auto predicate_reader_it =
_current_predicate_columns.find(branch.local_id);
+ DORIS_CHECK(predicate_reader_it !=
_current_predicate_columns.end());
+ branch_reader = predicate_reader_it->second.get();
+ }
+ bool used_filter = false;
+ if (branch.dictionary_filter.has_value()) {
+ uint16_t branch_survivors = 0;
+ RETURN_IF_ERROR(branch_reader->select_with_dictionary_filter(
+ *selection, selected_rows_before, batch_rows,
*branch.dictionary_filter,
+ nullptr, &_raw_disjunction_branch_filter_scratch,
&branch_survivors,
+ &used_filter));
+ if (used_filter) {
+ ++dictionary_branches;
+ }
+ } else {
+ DirectPredicateExecutionKind execution_kind =
DirectPredicateExecutionKind::NONE;
+ const VExprSPtrs branch_conjuncts {branch.expression};
+ RETURN_IF_ERROR(branch_reader->select_with_fixed_width_filter(
+ *selection, selected_rows_before, batch_rows,
branch_conjuncts,
+ cast_set<int>(branch.position), nullptr,
+ &_raw_disjunction_branch_filter_scratch, &used_filter,
&execution_kind));
+ if (used_filter &&
+ (execution_kind == DirectPredicateExecutionKind::RAW_FIXED
||
+ execution_kind ==
DirectPredicateExecutionKind::RAW_BINARY ||
+ execution_kind ==
DirectPredicateExecutionKind::CONVERTED_FIXED)) {
+ ++raw_value_branches;
+ }
+ if (used_filter &&
+ (execution_kind == DirectPredicateExecutionKind::RAW_FIXED
||
+ execution_kind ==
DirectPredicateExecutionKind::CONVERTED_FIXED)) {
+ ++fixed_width_branches;
+ }
+ }
+ if (!used_filter) {
+ if (!branch.reader) {
+ return Status::InternalError(
+ "Validated multi-column OR branch {} could not
execute directly",
+ branch.position);
+ }
+ // No ordinary predicate reader has advanced yet. Dropping the
independent raw
Review Comment:
[P1] Do not fall back after an ordinary branch has advanced
The assertion in this comment is not guaranteed: branches run in expression
order, and an earlier hidden single-use branch may already have consumed its
ordinary predicate reader. If a later auxiliary branch then returns
`used_filter=false` (for example, a >=50%-NULL mixed dictionary/PLAIN chunk),
this erases the auxiliaries and resumes residual evaluation.
`materialize_predicate_positions()` then reads the earlier ordinary reader
again for the same batch, so that branch supplies the next span's values and
can silently select the wrong source rows. The stage needs a non-consuming
all-branch capability check, or all branches must execute on independent
readers until success can be committed.
##########
be/benchmark/parquet/benchmark_parquet_reader.hpp:
##########
@@ -234,7 +253,7 @@ inline std::filesystem::path ensure_fixture(const
ReaderScenario& scenario) {
properties.version(::parquet::ParquetVersion::PARQUET_2_6);
properties.data_page_version(::parquet::ParquetDataPageVersion::V2);
properties.compression(::parquet::Compression::UNCOMPRESSED);
- properties.disable_statistics();
+ // Keep footer statistics enabled to match the common production fixture
shape.
Review Comment:
[P1] Keep statistics out of the decode benchmark fixtures
This changes the entire existing reader matrix, not just the new OR cases
that need footer `null_count`. For example, fixture values are `row % 100` and
`sel_0` is `c < 0`; with statistics enabled, ZoneMap planning proves every Row
Group has no match during open, so those timed cases perform no row-level
decode, predicate compaction, or lazy materialization. That invalidates their
role as decode/filter benchmarks (and the benchmark guide still says statistics
are disabled and pruning cases must be separate). Please keep the existing
fixtures statistics-free and introduce a distinct statistics-enabled
fixture/mode for the OR policy or explicit pruning benchmarks.
##########
be/src/format_v2/parquet/parquet_scan.cpp:
##########
@@ -2421,6 +2759,121 @@ Status
ParquetScanScheduler::read_filter_columns(int64_t batch_rows,
return status;
};
+ auto execute_raw_disjunction = [&](const detail::PredicateConjunctStage&
stage,
+ bool* applied) -> Status {
+ DORIS_CHECK(applied != nullptr);
+ *applied = false;
+ if (stage.raw_disjunction_branches.empty() || *selected_rows == 0) {
+ return Status::OK();
+ }
+ auto readers_it =
_current_raw_disjunction_readers.find(stage.expression.get());
+ if (readers_it == _current_raw_disjunction_readers.end()) {
+ return Status::OK();
+ }
+
+ const uint16_t selected_rows_before = *selected_rows;
+ // PODArray resize preserves retained bytes, but every OR group must
start from FALSE or a
+ // survivor bit from the previous batch could admit a row that matches
no current branch.
+ _raw_disjunction_filter_scratch.resize(selected_rows_before);
+ std::ranges::fill(_raw_disjunction_filter_scratch, 0);
+ size_t dictionary_branches = 0;
+ size_t raw_value_branches = 0;
+ size_t fixed_width_branches = 0;
+ for (auto& branch : readers_it->second) {
+ ParquetColumnReader* branch_reader = branch.reader.get();
+ if (branch_reader == nullptr) {
+ const auto predicate_reader_it =
_current_predicate_columns.find(branch.local_id);
+ DORIS_CHECK(predicate_reader_it !=
_current_predicate_columns.end());
+ branch_reader = predicate_reader_it->second.get();
+ }
+ bool used_filter = false;
+ if (branch.dictionary_filter.has_value()) {
+ uint16_t branch_survivors = 0;
+ RETURN_IF_ERROR(branch_reader->select_with_dictionary_filter(
+ *selection, selected_rows_before, batch_rows,
*branch.dictionary_filter,
+ nullptr, &_raw_disjunction_branch_filter_scratch,
&branch_survivors,
+ &used_filter));
+ if (used_filter) {
+ ++dictionary_branches;
+ }
+ } else {
+ DirectPredicateExecutionKind execution_kind =
DirectPredicateExecutionKind::NONE;
+ const VExprSPtrs branch_conjuncts {branch.expression};
+ RETURN_IF_ERROR(branch_reader->select_with_fixed_width_filter(
+ *selection, selected_rows_before, batch_rows,
branch_conjuncts,
+ cast_set<int>(branch.position), nullptr,
+ &_raw_disjunction_branch_filter_scratch, &used_filter,
&execution_kind));
+ if (used_filter &&
+ (execution_kind == DirectPredicateExecutionKind::RAW_FIXED
||
+ execution_kind ==
DirectPredicateExecutionKind::RAW_BINARY ||
+ execution_kind ==
DirectPredicateExecutionKind::CONVERTED_FIXED)) {
+ ++raw_value_branches;
+ }
+ if (used_filter &&
+ (execution_kind == DirectPredicateExecutionKind::RAW_FIXED
||
+ execution_kind ==
DirectPredicateExecutionKind::CONVERTED_FIXED)) {
+ ++fixed_width_branches;
+ }
+ }
+ if (!used_filter) {
+ if (!branch.reader) {
+ return Status::InternalError(
Review Comment:
[P1] Fall back before reusing an unsupported ordinary reader
A valid mixed dictionary/PLAIN chunk can reach this branch.
`is_fully_dictionary_encoded_chunk()` classifies it as non-dictionary, a hidden
column with at least 50% NULLs passes the raw gate, but
`ScalarColumnReader::read_fixed_width_filter()` rejects the advertised
`RLE_DICTIONARY` encoding and returns `used_filter=false` before consuming.
Because a hidden single-use branch stores `reader == nullptr`, enabling this
optimization turns a supported residual scan into this `InternalError`. Please
include the native reader's complete per-Row-Group encoding capability in the
all-branch preflight (or otherwise keep all ordinary cursors untouched) so an
unsupported chunk retains the residual path.
##########
be/benchmark/parquet/benchmark_parquet_reader.hpp:
##########
@@ -493,6 +560,29 @@ inline std::unique_ptr<ReaderSession> open_reader(const
std::filesystem::path& p
throw_if_error(context->open(&session->runtime_state));
session->request->conjuncts.push_back(context);
session->opened_conjuncts.push_back(std::move(context));
+ } else if (scenario.operation == ReaderOperation::MULTI_COLUMN_OR_SCAN) {
+ DORIS_CHECK(scenario.value_type == ValueType::DECIMAL64);
+ DORIS_CHECK(scenario.schema_width >= 4);
+ std::array<int, 3> predicate_positions {};
+ for (int column = 0; column < 3; ++column) {
+ const auto predicate_id = format::LocalColumnId(column);
+ throw_if_error(request_builder.add_predicate_column(predicate_id));
+ if (scenario.projection == Projection::PREDICATE_ONLY || column !=
0) {
+
session->request->predicate_only_columns.push_back(predicate_id);
+ }
+ predicate_positions[column] =
+
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_multi_column_or_predicate(scenario,
predicate_positions,
+ 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->request->enable_multi_column_or_raw_filter =
Review Comment:
[P1] Make the raw benchmark execute and verify the raw path
This flag only requests the production policy; it does not force the
implementation named by the scenario. Under the checked-in gates, 65 of the 100
`impl_raw_disjunction` cases (including every projected PLAIN case) fall back,
so those registrations measure the residual path on both sides. The PR body
acknowledges the 35/65 split, but the regressions from the earlier unrestricted
implementation are what justify the hard-coded 10%/50% cutoffs, and this
benchmark has neither a force-raw mode nor a profile/path assertion to
reproduce them. Please add a benchmark-only capability-checked force mode and
verify the selected path before timing, while retaining the current
policy-fallback cases as separately named controls.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
