github-actions[bot] commented on code in PR #66412:
URL: https://github.com/apache/doris/pull/66412#discussion_r3710207447
##########
be/src/format_v2/parquet/parquet_scan.cpp:
##########
@@ -2421,6 +3207,230 @@ Status
ParquetScanScheduler::read_filter_columns(int64_t batch_rows,
return status;
};
+ auto execute_raw_dnf = [&](const detail::PredicateConjunctStage& stage,
+ bool* applied) -> Status {
+ DORIS_CHECK(applied != nullptr);
+ *applied = false;
+ if (stage.raw_dnf_columns.empty() || *selected_rows == 0) {
+ return Status::OK();
+ }
+ auto readers_it =
_current_raw_dnf_readers.find(stage.expression.get());
+ if (readers_it == _current_raw_dnf_readers.end()) {
+ return Status::OK();
+ }
+ size_t dictionary_columns = 0;
+ size_t raw_value_columns = 0;
+ size_t fixed_width_columns = 0;
+ const uint16_t selected_rows_before = *selected_rows;
+ const size_t branch_count =
stage.raw_dnf_columns.front().branch_conjuncts.size();
+ _raw_disjunction_filter_scratch.resize(selected_rows_before);
+ // The scratch buffer is retained across batches; every DNF
intersection must start from
+ // the full branch mask rather than a previous batch's survivors.
+ std::ranges::fill(_raw_disjunction_filter_scratch,
+ static_cast<uint8_t>((1U << branch_count) - 1));
+ for (auto& column : readers_it->second) {
+ bool used_filter = false;
+ if (column.dictionary_filter.has_value()) {
+ uint16_t survivors = 0;
+ RETURN_IF_ERROR(column.reader->select_with_dictionary_filter(
+ *selection, selected_rows_before, batch_rows,
*column.dictionary_filter,
+ nullptr, &_raw_disjunction_branch_filter_scratch,
&survivors, &used_filter,
+ true));
+ if (used_filter) {
+ ++dictionary_columns;
+ }
+ } else {
+ DirectPredicateExecutionKind execution_kind =
DirectPredicateExecutionKind::NONE;
+ const VExprSPtrs conjuncts {column.expression};
+ RETURN_IF_ERROR(column.reader->select_with_fixed_width_filter(
+ *selection, selected_rows_before, batch_rows,
conjuncts,
+ cast_set<int>(column.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_columns;
+ }
+ if (used_filter &&
+ (execution_kind == DirectPredicateExecutionKind::RAW_FIXED
||
+ execution_kind ==
DirectPredicateExecutionKind::CONVERTED_FIXED)) {
+ ++fixed_width_columns;
+ }
+ }
+ if (!used_filter) {
Review Comment:
[P1] Fall back for mixed encodings in the DNF executor
This is a second, distinct mixed-encoding failure from the simple-OR branch
below. Rectangular-DNF preflight classifies every non-fully-dictionary chunk
from only the expression/type and NULL-density gates, so a valid
dictionary-to-PLAIN chunk at 50%-<90% NULL installs these auxiliary readers.
`ScalarColumnReader::read_fixed_width_filter()` then checks the complete
advertised encoding set, sees `RLE_DICTIONARY` beside PLAIN, and safely returns
`used_filter=false` before consuming—but this path converts that refusal into
`InternalError`. The ordinary residual reader supports this transition, and all
DNF readers are auxiliary, so please include the complete per-Column-Chunk
encoding capability in DNF preflight and disable the whole mask stage while
ordinary cursors are untouched. Fixing only the `branch.reader == nullptr`
logic in the existing simple-OR thread will not cover this executor.
##########
be/src/format_v2/parquet/parquet_profile.cpp:
##########
@@ -211,6 +211,12 @@ void ParquetProfile::init(RuntimeProfile* profile) {
profile, "DictionaryPredicateDirectRows", TUnit::UNIT,
parquet_profile, 1);
dictionary_predicate_projected_rows = ADD_CHILD_COUNTER_WITH_LEVEL(
profile, "DictionaryPredicateProjectedRows", TUnit::UNIT,
parquet_profile, 1);
+ multi_column_or_raw_filter_batches = ADD_CHILD_COUNTER_WITH_LEVEL(
+ profile, "MultiColumnOrRawFilterBatches", TUnit::UNIT,
parquet_profile, 1);
+ multi_column_or_raw_filter_fallback_batches = ADD_CHILD_COUNTER_WITH_LEVEL(
Review Comment:
[P2] Count policy-gated raw-filter fallbacks
This counter only advances when a reader was already created and then
returns `used_filter=false`. The normal path for the new policy is different:
file/Row-Group NULL-density, projection, encoding, and dictionary-survival
gates insert the stage into `_disabled_raw_disjunctions` and run the residual
expression without incrementing this or any attempted/disabled-by-reason
counter. A profile with raw batches = 0 and fallback batches = 0 therefore
cannot distinguish an ineligible request from a recognized raw OR/DNF that the
policy rejected, including the cases exercised by the new
`KeepsLegacy*OutsideProfitableRange` tests. Please publish the decision where
those gates fire (ideally attempted/enabled and disabled-by-reason counters),
or count residual batches for disabled recognized stages and cover that meaning
in the tests.
##########
be/src/format_v2/parquet/parquet_scan.cpp:
##########
@@ -1834,6 +2293,297 @@ Status
ParquetScanScheduler::prepare_current_dictionary_filters(
_current_predicate_columns.emplace(local_id, std::move(column_reader));
update_counter_if_not_null(_scan_profile.dict_filter_columns, 1);
}
+
+ std::unordered_map<size_t, size_t> remaining_position_occurrences;
+ for (const auto& remaining_stage : schedule.remaining_stages) {
+ for (const size_t position : remaining_stage.required_positions) {
+ ++remaining_position_occurrences[position];
+ }
+ }
+ std::unordered_set<size_t> delete_positions;
+ for (const auto& conjunct : request.delete_conjuncts) {
+ std::set<int> positions;
+ conjunct->root()->collect_slot_column_ids(positions);
+ for (const int position : positions) {
+ if (position >= 0) {
+ delete_positions.insert(cast_set<size_t>(position));
+ }
+ }
+ }
+
+ for (const auto& stage : schedule.remaining_stages) {
+ if (stage.raw_disjunction_branches.empty() ||
+ _disabled_raw_disjunctions.contains(stage.expression.get())) {
+ continue;
+ }
+ std::vector<RawDisjunctionBranchReader> branch_readers;
+ branch_readers.reserve(stage.raw_disjunction_branches.size());
+ bool usable = true;
+ for (const auto& branch : stage.raw_disjunction_branches) {
+ const auto predicate_index_it =
+
_predicate_indices_by_position_scratch.find(branch.position);
+ if (predicate_index_it ==
_predicate_indices_by_position_scratch.end()) {
+ usable = false;
+ break;
+ }
+ const auto& column =
request.predicate_columns[predicate_index_it->second];
+ const auto local_id = column.column_id();
+ if (!local_id.is_valid() ||
+ local_id.value() >= static_cast<int32_t>(file_schema.size())) {
+ usable = false;
+ break;
+ }
+ const auto& column_schema = file_schema[local_id.value()];
+ DORIS_CHECK(column_schema != nullptr);
+ if (column_schema->leaf_column_id < 0 ||
+ column_schema->leaf_column_id >=
+ static_cast<int>(row_group_metadata.columns.size())) {
+ usable = false;
+ break;
+ }
+ const auto& column_chunk =
row_group_metadata.columns[column_schema->leaf_column_id];
+ if (!column_chunk.__isset.meta_data) {
+ usable = false;
+ break;
+ }
+ const int expression_column_id = cast_set<int>(branch.position);
+ const bool raw_eligible =
branch.expression->can_execute_on_raw_fixed_values(
+ column_schema->type,
expression_column_id) ||
+
branch.expression->can_execute_on_raw_binary_values(
+ column_schema->type,
expression_column_id) ||
+
branch.expression->can_execute_on_null_map(
+ column_schema->type,
expression_column_id);
+ if (!raw_eligible) {
+ usable = false;
+ break;
+ }
+
+ bool dictionary_eligible = false;
+ if (!branch.expression->raw_predicate_result_for_null()) {
+ dictionary_eligible = supports_row_level_dictionary_filter(
+ *column_schema,
column_chunk.meta_data) &&
+
(branch.expression->can_execute_on_raw_fixed_values(
+ column_schema->type,
expression_column_id) ||
+
branch.expression->can_execute_on_raw_binary_values(
+ column_schema->type,
expression_column_id));
+ }
+ const bool projected = !request.is_predicate_only(local_id);
+ const int minimum_null_denominator = dictionary_eligible &&
!projected ? 10 : 2;
+ if ((!dictionary_eligible && projected) ||
+ !has_null_fraction_at_least(column_chunk.meta_data, 1,
minimum_null_denominator)) {
+ // Branch readers pay off only after avoiding enough nullable
materialization. The
+ // paired matrix establishes conservative 10% dictionary and
50% PLAIN thresholds;
+ // projected PLAIN branches retain the residual path because
they decode twice.
+ usable = false;
+ break;
+ }
+ const bool use_predicate_reader =
+ request.is_predicate_only(local_id) &&
+ remaining_position_occurrences[branch.position] == 1 &&
+ !delete_positions.contains(branch.position) &&
+
!schedule.single_column_conjuncts.contains(branch.position);
+ if (dictionary_eligible) {
+
update_counter_if_not_null(_scan_profile.dict_filter_candidate_columns, 1);
+ }
+
+ std::unique_ptr<ParquetColumnReader> column_reader;
+ if (!use_predicate_reader || dictionary_eligible) {
+ RETURN_IF_ERROR(NativeColumnReader::create(
+ *column_schema, &column, file_context.native_file,
Review Comment:
[P1] Build raw execution readers on the Row-Group I/O wrapper
This object is used later to scan data pages, but it permanently captures
`file_context.native_file` while `prepare_current_dictionary_filters()` still
runs before `set_native_random_access_ranges()`; the DNF construction below has
the same ordering. For a projected-only OR request
(`non_predicate_positions.empty()`), the scheduler can then enable MergeRange
and build the ordinary survivor readers from `native_data_file()`, while the
actual raw branch scans continue through these base-file auxiliaries. DNF
similarly scans through its auxiliaries and then skips duplicate ordinary
readers. Because active MergeRange also suppresses the early FileCache
predicate prefetch, the new hot path gets neither the shared ranges nor that
warm-up and restores one independent remote stream per mask/branch before
rereading or skipping the ordinary readers. Please make duplicate raw-reader
layouts part of the Row-Group I/O decision: preferably eliminate the duplicate
cursor/consume the ordinary rea
der once, or otherwise disable MergeRange and retain explicit prefetch. Two
independent same-leaf readers should not simply be rebound to the wrapper's
forward-consumed range cache.
##########
be/src/format_v2/parquet/parquet_scan.cpp:
##########
@@ -2421,6 +3207,230 @@ Status
ParquetScanScheduler::read_filter_columns(int64_t batch_rows,
return status;
};
+ auto execute_raw_dnf = [&](const detail::PredicateConjunctStage& stage,
+ bool* applied) -> Status {
+ DORIS_CHECK(applied != nullptr);
+ *applied = false;
+ if (stage.raw_dnf_columns.empty() || *selected_rows == 0) {
+ return Status::OK();
+ }
+ auto readers_it =
_current_raw_dnf_readers.find(stage.expression.get());
+ if (readers_it == _current_raw_dnf_readers.end()) {
+ return Status::OK();
+ }
+ size_t dictionary_columns = 0;
+ size_t raw_value_columns = 0;
+ size_t fixed_width_columns = 0;
+ const uint16_t selected_rows_before = *selected_rows;
+ const size_t branch_count =
stage.raw_dnf_columns.front().branch_conjuncts.size();
+ _raw_disjunction_filter_scratch.resize(selected_rows_before);
+ // The scratch buffer is retained across batches; every DNF
intersection must start from
+ // the full branch mask rather than a previous batch's survivors.
+ std::ranges::fill(_raw_disjunction_filter_scratch,
+ static_cast<uint8_t>((1U << branch_count) - 1));
+ for (auto& column : readers_it->second) {
+ bool used_filter = false;
+ if (column.dictionary_filter.has_value()) {
+ uint16_t survivors = 0;
+ RETURN_IF_ERROR(column.reader->select_with_dictionary_filter(
+ *selection, selected_rows_before, batch_rows,
*column.dictionary_filter,
+ nullptr, &_raw_disjunction_branch_filter_scratch,
&survivors, &used_filter,
+ true));
+ if (used_filter) {
+ ++dictionary_columns;
+ }
+ } else {
+ DirectPredicateExecutionKind execution_kind =
DirectPredicateExecutionKind::NONE;
+ const VExprSPtrs conjuncts {column.expression};
+ RETURN_IF_ERROR(column.reader->select_with_fixed_width_filter(
+ *selection, selected_rows_before, batch_rows,
conjuncts,
+ cast_set<int>(column.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_columns;
+ }
+ if (used_filter &&
+ (execution_kind == DirectPredicateExecutionKind::RAW_FIXED
||
+ execution_kind ==
DirectPredicateExecutionKind::CONVERTED_FIXED)) {
+ ++fixed_width_columns;
+ }
+ }
+ if (!used_filter) {
+ return Status::InternalError(
+ "Validated multi-column DNF column {} could not
execute directly",
+ column.position);
+ }
+ DORIS_CHECK_EQ(_raw_disjunction_branch_filter_scratch.size(),
selected_rows_before);
+ for (size_t row = 0; row < selected_rows_before; ++row) {
+ _raw_disjunction_filter_scratch[row] &=
_raw_disjunction_branch_filter_scratch[row];
+ }
+ }
+ const uint16_t new_selected_rows =
count_selected_rows(_raw_disjunction_filter_scratch);
+ if (conjunct_filtered_rows != nullptr) {
+ *conjunct_filtered_rows +=
static_cast<int64_t>(selected_rows_before) -
+ static_cast<int64_t>(new_selected_rows);
+ }
+ if (new_selected_rows != selected_rows_before) {
+ predicate_columns_need_alignment = true;
+ *selected_rows = new_selected_rows == 0 ? 0
+ :
apply_compact_filter_to_selection(
+
_raw_disjunction_filter_scratch,
+ selection,
selected_rows_before);
+ }
+ for (const auto& column : readers_it->second) {
+ if (!materialized_positions.contains(column.position)) {
+ const auto predicate_reader_it =
_current_predicate_columns.find(column.local_id);
+ DORIS_CHECK(predicate_reader_it !=
_current_predicate_columns.end());
+ RETURN_IF_ERROR(predicate_reader_it->second->skip(batch_rows));
+ materialized_positions.insert(column.position);
+
read_column_positions.push_back(cast_set<uint32_t>(column.position));
+ }
+ auto placeholder =
file_block->get_by_position(column.position).column->clone_empty();
+ placeholder->insert_many_defaults(*selected_rows);
+ file_block->replace_by_position(column.position,
std::move(placeholder));
+ remember_column_selection(cast_set<uint32_t>(column.position));
+ }
+
update_counter_if_not_null(_scan_profile.multi_column_or_raw_filter_batches, 1);
+
update_counter_if_not_null(_scan_profile.multi_column_or_raw_filter_branches,
+ cast_set<int64_t>(branch_count));
+
update_counter_if_not_null(_scan_profile.dictionary_predicate_direct_batches,
+ cast_set<int64_t>(dictionary_columns));
+
update_counter_if_not_null(_scan_profile.dictionary_predicate_direct_rows,
Review Comment:
[P2] Keep DNF direct-filter accounting in selected-row units
Each DNF reader receives the current `selection` and returns one mask byte
per `selected_rows_before`, so rows rejected by earlier predicate rounds are
skipped rather than evaluated by these mask expressions. These updates instead
charge `batch_rows` for every DNF column: an 8K batch narrowed to one row makes
two mask columns report 16K direct rows instead of two, while the existing
single-column and sibling raw-OR paths use `selected_rows_before`. This block
also never adds `selected_rows_before - new_selected_rows` to
`RowsFilteredByDictFilter` when every DNF column used a dictionary bitmap,
although those same two paths publish that benefit. Please keep the direct rows
in selected-input units, add the all-dictionary rejected-row attribution, and
cover both with a DNF stage after a selective conjunct.
--
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]
