github-actions[bot] commented on code in PR #65369:
URL: https://github.com/apache/doris/pull/65369#discussion_r3548814999
##########
be/src/format_v2/parquet/parquet_scan.cpp:
##########
@@ -597,42 +818,433 @@ Status
ParquetScanScheduler::skip_current_row_group_rows(int64_t rows) {
return Status::OK();
}
+namespace {
+
+struct PredicateConjunctSchedule {
+ std::map<size_t, VExprContextSPtrs> single_column_conjuncts;
+ VExprContextSPtrs remaining_conjuncts;
+};
+
+PredicateConjunctSchedule build_predicate_conjunct_schedule(
+ const format::FileScanRequest& request) {
+ std::unordered_set<size_t> predicate_block_positions;
+ predicate_block_positions.reserve(request.predicate_columns.size());
+ 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());
+ predicate_block_positions.insert(position_it->second.value());
+ }
+
+ PredicateConjunctSchedule schedule;
+ for (const auto& conjunct : request.conjuncts) {
+ DORIS_CHECK(conjunct != nullptr);
+ std::set<int> referenced_positions;
+ conjunct->root()->collect_slot_column_ids(referenced_positions);
+ if (referenced_positions.size() != 1) {
+ schedule.remaining_conjuncts.push_back(conjunct);
+ 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.single_column_conjuncts[block_position].push_back(conjunct);
+ }
+ return schedule;
+}
+
+bool can_evaluate_all_with_dictionary(const VExprContextSPtrs& conjuncts) {
+ if (conjuncts.empty()) {
+ return false;
+ }
+ return std::ranges::all_of(conjuncts, [](const auto& conjunct) {
+ return conjunct != nullptr && conjunct->root() != nullptr &&
+ conjunct->root()->can_evaluate_dictionary_filter();
+ });
+}
+
+size_t file_block_column_count(const format::FileScanRequest& request) {
+ size_t column_count = 0;
+ for (const auto& [_, block_position] : request.local_positions) {
+ column_count = std::max(column_count,
static_cast<size_t>(block_position.value()) + 1);
+ }
+ return column_count;
+}
+
+Status build_dictionary_value_column(const ParquetColumnSchema& column_schema,
+ const OwnedDictionaryWords& dict_words,
+ ColumnWithTypeAndName* column) {
+ DORIS_CHECK(column != nullptr);
+ auto string_column = ColumnString::create();
+ for (const auto& value : dict_words.values) {
+ string_column->insert_data(value.data(), value.size());
+ }
+
+ MutableColumnPtr result_column = std::move(string_column);
+ if (column_schema.type->is_nullable()) {
+ result_column = ColumnNullable::create(std::move(result_column),
+
ColumnUInt8::create(dict_words.values.size(), 0));
+ }
+ *column =
+ ColumnWithTypeAndName(std::move(result_column),
column_schema.type, column_schema.name);
+ return Status::OK();
+}
+
+Status build_dictionary_filter_block(
+ const format::FileScanRequest& request,
+ const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
+ format::LocalColumnId dictionary_column_id, const
OwnedDictionaryWords& dict_words,
+ Block* dict_block) {
+ DORIS_CHECK(dict_block != nullptr);
+ dict_block->clear();
+ // VExpr conjuncts address columns by their file-block position, not by
Parquet local column id.
+ // Keep the temporary dictionary block in the same layout as normal scan
blocks so existing
+ // expression evaluation can be reused without a dictionary-specific
expression path.
+ std::vector<ColumnWithTypeAndName>
columns(file_block_column_count(request));
+ for (const auto& [file_column_id, block_position] :
request.local_positions) {
+ DORIS_CHECK(file_column_id.is_valid() &&
+ file_column_id.value() <
static_cast<int>(file_schema.size()));
+ const auto& column_schema = file_schema[file_column_id.value()];
+ DORIS_CHECK(column_schema != nullptr);
+ ColumnWithTypeAndName column;
+ if (file_column_id == dictionary_column_id) {
+ RETURN_IF_ERROR(build_dictionary_value_column(*column_schema,
dict_words, &column));
+ } else {
+ column =
ColumnWithTypeAndName(column_schema->type->create_column(),
+ column_schema->type,
column_schema->name);
+ }
+ columns[static_cast<size_t>(block_position.value())] =
std::move(column);
+ }
+ for (auto& column : columns) {
+ DORIS_CHECK(column.column.get() != nullptr);
+ dict_block->insert(std::move(column));
+ }
+ return Status::OK();
+}
+
+uint16_t count_selected_rows(const IColumn::Filter& filter) {
+ uint16_t selected_rows = 0;
+ for (const auto value : filter) {
+ selected_rows += value != 0;
+ }
+ return selected_rows;
+}
+
+Status filter_read_predicate_columns(Block* file_block, const
std::vector<uint32_t>& positions,
+ const IColumn::Filter& compact_filter) {
+ if (positions.empty()) {
+ return Status::OK();
+ }
+ RETURN_IF_CATCH_EXCEPTION(Block::filter_block_internal(file_block,
positions, compact_filter));
+ return Status::OK();
+}
+
+} // namespace
+
+Status ParquetScanScheduler::prepare_current_dictionary_filters(
+ ParquetFileContext& file_context,
+ const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
+ const format::FileScanRequest& request, int row_group_idx,
+ const ::parquet::RowGroupMetaData& row_group_metadata) {
+ _current_dictionary_filters.clear();
+ if (request.conjuncts.empty()) {
+ return Status::OK();
+ }
+ const auto schedule = build_predicate_conjunct_schedule(request);
+ if (schedule.single_column_conjuncts.empty()) {
+ return Status::OK();
+ }
+
+ SCOPED_TIMER(_scan_profile.dict_filter_rewrite_time);
+ for (const auto& col : request.predicate_columns) {
+ const auto local_id = col.local_id();
+ if (local_id < 0 || local_id >=
static_cast<int32_t>(file_schema.size())) {
+ continue;
+ }
+ const auto position_it = request.local_positions.find(col.column_id());
+ DORIS_CHECK(position_it != request.local_positions.end());
+ const auto block_position =
static_cast<size_t>(position_it->second.value());
+ const auto conjunct_it =
schedule.single_column_conjuncts.find(block_position);
+ if (conjunct_it == schedule.single_column_conjuncts.end() ||
+ !can_evaluate_all_with_dictionary(conjunct_it->second)) {
+ continue;
+ }
+
+ // This optimization is deliberately limited to single-column
predicates that all support
+ // dictionary evaluation. Mixed predicates still run after predicate
columns are read, using
+ // the normal row-level expression path.
+ const auto& column_schema = file_schema[local_id];
+ DORIS_CHECK(column_schema != nullptr);
+ if (column_schema->leaf_column_id < 0 ||
+ column_schema->leaf_column_id >= row_group_metadata.num_columns())
{
+ continue;
+ }
+ auto column_chunk =
row_group_metadata.ColumnChunk(column_schema->leaf_column_id);
+ if (column_chunk == nullptr ||
+ !supports_row_level_dictionary_filter(*column_schema,
*column_chunk)) {
+ continue;
+ }
+
+ OwnedDictionaryWords dict_words;
+ if (!read_byte_array_dictionary_words(file_context.file_reader.get(),
row_group_idx,
+ column_schema->leaf_column_id,
*column_schema,
+ &dict_words)) {
+ continue;
+ }
+
+ Block dict_block;
+ RETURN_IF_ERROR(build_dictionary_filter_block(request, file_schema,
col.column_id(),
+ dict_words,
&dict_block));
+ IColumn::Filter dictionary_filter;
+ bool can_filter_all = false;
+ // Evaluate the original conjuncts against dictionary entries once.
The resulting bitmap is
+ // indexed by Parquet dictionary id, so the data-page reader only
performs an integer lookup
+ // for each selected row instead of materializing every
predicate-column value first.
+ RETURN_IF_ERROR(execute_compact_filter_conjuncts(conjunct_it->second,
+
dict_words.values.size(), &dict_block,
+ &dictionary_filter,
&can_filter_all));
+ if (can_filter_all) {
+ dictionary_filter.resize_fill(dict_words.values.size(), 0);
+ }
+
+ // The bitmap is keyed by Parquet dictionary id. Later data-page reads
evaluate the
+ // predicate with an integer lookup and only materialize STRING values
for surviving rows.
+ _current_dictionary_filters.emplace(local_id,
std::move(dictionary_filter));
+ }
+ return Status::OK();
+}
+
Status ParquetScanScheduler::read_filter_columns(int64_t batch_rows,
const
format::FileScanRequest& request,
Block* file_block,
SelectionVector* selection,
uint16_t* selected_rows,
- int64_t*
conjunct_filtered_rows) {
+ int64_t*
conjunct_filtered_rows,
+ bool*
predicate_columns_filtered) {
+ DORIS_CHECK(predicate_columns_filtered != nullptr);
+ *predicate_columns_filtered = false;
if (!request.conjuncts.empty() || !request.delete_conjuncts.empty()) {
selection->resize(static_cast<size_t>(batch_rows));
}
- for (const auto& [fid, column_reader] : _current_predicate_columns) {
- auto position_it =
request.local_positions.find(format::LocalColumnId(fid));
- DORIS_CHECK(position_it != request.local_positions.end());
- const auto block_position = position_it->second.value();
+ const auto schedule = build_predicate_conjunct_schedule(request);
+ const bool can_read_predicate_columns_round_by_round =
+ !schedule.single_column_conjuncts.empty();
+ std::vector<uint32_t> read_column_positions;
+ read_column_positions.reserve(request.predicate_columns.size());
+
+ auto read_predicate_column = [&](ParquetColumnReader* column_reader,
size_t block_position,
+ ColumnId local_id, bool*
used_dictionary_filter) -> Status {
+ DORIS_CHECK(used_dictionary_filter != nullptr);
+ *used_dictionary_filter = false;
DCHECK(remove_nullable(column_reader->type())
->equals(*remove_nullable(file_block->get_by_position(block_position).type)))
<< column_reader->type()->get_name() << " "
<<
file_block->get_by_position(block_position).type->get_name() << " "
<< column_reader->name() << " " <<
file_block->get_by_position(block_position).name;
auto column =
file_block->get_by_position(block_position).column->assert_mutable();
- int64_t column_rows = 0;
- {
- SCOPED_TIMER(_scan_profile.column_read_time);
- RETURN_IF_ERROR(column_reader->read(batch_rows, column,
&column_rows));
+ SCOPED_TIMER(_scan_profile.column_read_time);
+ const auto dictionary_filter_it =
_current_dictionary_filters.find(local_id);
+ if (dictionary_filter_it != _current_dictionary_filters.end()) {
+ const uint16_t selected_rows_before = *selected_rows;
+ IColumn::Filter compact_filter;
+ bool used_filter = false;
+ RETURN_IF_ERROR(column_reader->select_with_dictionary_filter(
+ *selection, *selected_rows, batch_rows,
dictionary_filter_it->second, column,
+ &compact_filter, &used_filter));
+ if (used_filter) {
+ DORIS_CHECK(compact_filter.size() == selected_rows_before);
+ const uint16_t new_selected_rows =
count_selected_rows(compact_filter);
+ 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) {
+ // The dictionary reader has already appended only
surviving values for the
+ // current column. Apply the compact row filter only to
columns read before this
+ // one, then update the shared selection for later
predicate/lazy columns.
+ RETURN_IF_ERROR(filter_read_predicate_columns(file_block,
read_column_positions,
+
compact_filter));
+ *selected_rows =
apply_compact_filter_to_selection(compact_filter, selection,
+
selected_rows_before);
+ *predicate_columns_filtered = true;
+ }
+ file_block->replace_by_position(block_position,
std::move(column));
+
read_column_positions.push_back(cast_set<uint32_t>(block_position));
+ *used_dictionary_filter = true;
+ return Status::OK();
+ }
}
- if (column_rows != batch_rows) {
- return Status::Corruption("Parquet filter column {} returned {}
rows, expected {} rows",
- column_reader->name(), column_rows,
batch_rows);
+
+ if (*selected_rows == batch_rows) {
+ int64_t column_rows = 0;
+ RETURN_IF_ERROR(column_reader->read(batch_rows, column,
&column_rows));
+ if (column_rows != batch_rows) {
+ return Status::Corruption(
+ "Parquet filter column {} returned {} rows, expected
{} rows",
+ column_reader->name(), column_rows, batch_rows);
+ }
+ } else {
+ [[maybe_unused]] auto old_size = column->size();
+ RETURN_IF_ERROR(column_reader->select(*selection, *selected_rows,
batch_rows, column));
+ if (column->size() != old_size + *selected_rows) {
+ return Status::Corruption(
+ "Parquet selected filter column {} returned {} rows,
expected {} rows",
+ column_reader->name(), column->size(), old_size +
*selected_rows);
+ }
+ *predicate_columns_filtered = true;
}
file_block->replace_by_position(block_position, std::move(column));
- }
- if (_scan_profile.predicate_filter_time == nullptr) {
+ read_column_positions.push_back(cast_set<uint32_t>(block_position));
+ return Status::OK();
+ };
+
+ auto execute_scheduled_conjuncts = [&](const VExprContextSPtrs& 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_filter_conjuncts(
+ conjuncts, selected_rows_before, file_block, &compact_filter,
&can_filter_all));
+ if (can_filter_all) {
+ compact_filter.resize_fill(selected_rows_before, 0);
+ }
+ const uint16_t new_selected_rows = can_filter_all ? 0 :
count_selected_rows(compact_filter);
+ 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) {
+ // All columns read so far are already compacted to the current
selection. Apply the
+ // compact filter to those columns and the selection vector
together, so later predicate
+ // columns can read only rows that survived previous predicate
rounds.
+ RETURN_IF_ERROR(filter_read_predicate_columns(file_block,
read_column_positions,
+ compact_filter));
+ *selected_rows = can_filter_all
+ ? 0
+ :
apply_compact_filter_to_selection(compact_filter, selection,
+
selected_rows_before);
+ *predicate_columns_filtered = true;
+ }
+ return Status::OK();
+ };
+
+ auto execute_scheduled_conjuncts_with_profile =
+ [&](const VExprContextSPtrs& conjuncts) -> Status {
+ if (_scan_profile.predicate_filter_time == nullptr) {
+ return execute_scheduled_conjuncts(conjuncts);
+ }
+ SCOPED_TIMER(_scan_profile.predicate_filter_time);
+ return execute_scheduled_conjuncts(conjuncts);
+ };
+
+ auto execute_scheduled_delete_conjuncts = [&]() -> Status {
+ if (request.delete_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_delete_conjuncts(request.delete_conjuncts,
+ selected_rows_before,
file_block,
+ &compact_filter,
&can_filter_all));
+ if (can_filter_all) {
+ compact_filter.resize_fill(selected_rows_before, 0);
+ }
+ if (can_filter_all || count_selected_rows(compact_filter) !=
selected_rows_before) {
+ RETURN_IF_ERROR(filter_read_predicate_columns(file_block,
read_column_positions,
+ compact_filter));
+ *selected_rows = can_filter_all
+ ? 0
+ :
apply_compact_filter_to_selection(compact_filter, selection,
+
selected_rows_before);
+ *predicate_columns_filtered = true;
+ }
+ return Status::OK();
+ };
+
+ auto read_all_predicate_columns = [&]() -> Status {
+ for (const auto& [fid, column_reader] : _current_predicate_columns) {
+ auto position_it =
request.local_positions.find(format::LocalColumnId(fid));
+ DORIS_CHECK(position_it != request.local_positions.end());
+ bool used_dictionary_filter = false;
+ RETURN_IF_ERROR(read_predicate_column(column_reader.get(),
position_it->second.value(),
+ fid,
&used_dictionary_filter));
+ }
+ return Status::OK();
+ };
+
+ if (!can_read_predicate_columns_round_by_round) {
+ RETURN_IF_ERROR(read_all_predicate_columns());
+ if (_scan_profile.predicate_filter_time == nullptr) {
+ return execute_batch_filters(request, batch_rows, file_block,
selection, selected_rows,
+ conjunct_filtered_rows);
+ }
+ SCOPED_TIMER(_scan_profile.predicate_filter_time);
return execute_batch_filters(request, batch_rows, file_block,
selection, selected_rows,
conjunct_filtered_rows);
}
+
+ auto read_round_by_round = [&]() -> Status {
+ // 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.
+ for (size_t idx = 0; idx < request.predicate_columns.size(); ++idx) {
+ const auto& col = request.predicate_columns[idx];
+ const auto fid = col.local_id();
+ auto reader_it = _current_predicate_columns.find(fid);
+ DORIS_CHECK(reader_it != _current_predicate_columns.end());
+ auto position_it = request.local_positions.find(col.column_id());
+ DORIS_CHECK(position_it != request.local_positions.end());
+ const auto block_position = position_it->second.value();
+ bool used_dictionary_filter = false;
+ RETURN_IF_ERROR(read_predicate_column(reader_it->second.get(),
block_position, fid,
+ &used_dictionary_filter));
+ if (*selected_rows == 0) {
+ for (size_t remaining_idx = idx + 1;
+ remaining_idx < request.predicate_columns.size();
++remaining_idx) {
+ const auto remaining_fid =
request.predicate_columns[remaining_idx].local_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();
+ }
+ if (used_dictionary_filter) {
+ // The dictionary bitmap has already applied the single-column
conjunct for this
Review Comment:
`can_evaluate_dictionary_filter()` is only a safe pruning signal for `AND`:
`VCompoundPred` returns true when any child is dictionary-filterable, not when
the whole predicate can be replaced by a dictionary-entry bitmap. After this
reader returns `used_dictionary_filter`, the scheduler skips the original
scheduled conjunct entirely, so a predicate such as `value = 'az' AND random()
< 0.5` would evaluate `random()` once for the dictionary entry and then keep or
drop every row with that value together. The dictionary bitmap should be used
only as a prefilter unless the code can prove the whole conjunct is
dictionary-equivalent; residual or compound predicates still need to run on the
surviving rows.
--
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]
