github-actions[bot] commented on code in PR #65369:
URL: https://github.com/apache/doris/pull/65369#discussion_r3555935435
##########
be/src/format_v2/parquet/parquet_scan.cpp:
##########
@@ -597,42 +786,517 @@ 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);
+ DORIS_CHECK(conjunct->root() != nullptr);
+ if (!conjunct->root()->is_safe_to_execute_on_selected_rows()) {
Review Comment:
This safety gate only inspects `request.conjuncts`, but delete predicates
can also be error-preserving. Iceberg equality deletes populate
`request.delete_conjuncts` separately, and when a data key type differs from
the delete-key type `_append_equality_delete_predicates()` wraps the slot in
`format::Cast`; that cast propagates `_function->execute()` errors. With a
normal predicate such as `a > 0` enabling the round-by-round path, rows
rejected by `a` can be compacted away before
`execute_scheduled_delete_conjuncts()` runs, so a bad equality-delete cast on
one of those rows is suppressed. The old batch path kept the full `batch_rows`
block for delete conjunct evaluation after regular filters updated only the
selection vector. Please include delete conjunct roots in the selected-row
safety decision, or force the full-batch path whenever a delete predicate is
not proven safe on compacted rows.
##########
be/src/format_v2/parquet/reader/scalar_column_reader.cpp:
##########
@@ -215,6 +251,174 @@ Status ScalarColumnReader::skip(int64_t rows) {
return Status::OK();
}
+Status ScalarColumnReader::select_with_dictionary_filter(const
SelectionVector& sel,
+ uint16_t
selected_rows, int64_t batch_rows,
+ const
IColumn::Filter& dictionary_filter,
+ MutableColumnPtr&
column,
+ IColumn::Filter*
row_filter,
+ bool* used_filter) {
+ DORIS_CHECK(column.get() != nullptr);
+ DORIS_CHECK(row_filter != nullptr);
+ DORIS_CHECK(used_filter != nullptr);
+ RETURN_IF_ERROR(sel.verify(selected_rows, batch_rows));
+ *used_filter = false;
+ row_filter->clear();
+ row_filter->reserve(selected_rows);
+
+ const auto ranges = selection_to_ranges(sel, selected_rows);
+ int64_t cursor = 0;
+ for (const auto& range : ranges) {
+ if (range.start < cursor || range.start + range.length > batch_rows) {
+ return Status::InvalidArgument(
+ "Invalid parquet dictionary selection range [{}, {}) for
column {}",
+ range.start, range.start + range.length, _name);
+ }
+ RETURN_IF_ERROR(skip(range.start - cursor));
+
+ int64_t range_rows_read = 0;
+ RETURN_IF_ERROR(read_range_with_dictionary_filter(range.length,
dictionary_filter, column,
+ row_filter,
&range_rows_read,
+ used_filter));
+ if (!*used_filter) {
+ return Status::OK();
+ }
+ if (range_rows_read != range.length) {
+ return Status::Corruption(
+ "Parquet dictionary selected read returned {} rows,
expected {} rows for "
+ "column {}",
+ range_rows_read, range.length, _name);
+ }
+ cursor = range.start + range.length;
+ }
+ RETURN_IF_ERROR(skip(batch_rows - cursor));
+ if (_profile.reader_select_rows != nullptr) {
+ COUNTER_UPDATE(_profile.reader_select_rows, selected_rows);
+ }
+ return Status::OK();
+}
+
+Status ScalarColumnReader::read_range_with_dictionary_filter(
+ int64_t rows, const IColumn::Filter& dictionary_filter,
MutableColumnPtr& column,
+ IColumn::Filter* row_filter, int64_t* rows_read, bool* used_filter) {
+ DORIS_CHECK(row_filter != nullptr);
+ DORIS_CHECK(rows_read != nullptr);
+ DORIS_CHECK(used_filter != nullptr);
+ DORIS_CHECK(_record_reader != nullptr);
+ if (!_record_reader->read_dictionary()) {
+ *used_filter = false;
+ return Status::OK();
+ }
+
+ ParquetLeafBatch leaf_batch;
+ RETURN_IF_ERROR(leaf_reader().read_batch(rows, &leaf_batch, rows_read));
+ int64_t matched_rows = 0;
+
RETURN_IF_ERROR(append_dictionary_filtered_values(leaf_batch.binary_chunks(),
dictionary_filter,
+ column, row_filter,
&matched_rows,
+ used_filter));
+ if (!*used_filter) {
+ return Status::Corruption(
+ "Parquet dictionary reader did not return dictionary batches
for column {}", _name);
+ }
+ if (row_filter->size() < static_cast<size_t>(*rows_read)) {
+ return Status::Corruption(
+ "Parquet dictionary filter produced too few row decisions for
column {}: "
+ "filter={}, rows={}",
+ _name, row_filter->size(), *rows_read);
+ }
+ advance_rows_read(*rows_read);
+ update_reader_read_rows(*rows_read);
+ return Status::OK();
+}
+
+Status ScalarColumnReader::append_dictionary_filtered_values(
+ const std::vector<std::shared_ptr<::arrow::Array>>& chunks,
+ const IColumn::Filter& dictionary_filter, MutableColumnPtr& column,
+ IColumn::Filter* row_filter, int64_t* matched_rows, bool* used_filter)
const {
+ DORIS_CHECK(row_filter != nullptr);
+ DORIS_CHECK(matched_rows != nullptr);
+ DORIS_CHECK(used_filter != nullptr);
+ *matched_rows = 0;
+ *used_filter = false;
+
+ std::vector<StringRef> selected_values;
+ for (const auto& chunk : chunks) {
+ DORIS_CHECK(chunk != nullptr);
+ const auto* dict_array = dynamic_cast<const
::arrow::DictionaryArray*>(chunk.get());
+ if (dict_array == nullptr) {
+ // The caller has already consumed rows from a
DictionaryRecordReader. Falling back to a
+ // normal selected read would desynchronize the Parquet stream, so
absence of a
+ // DictionaryArray is reported as corruption by
read_range_with_dictionary_filter().
+ return Status::OK();
+ }
+ *used_filter = true;
+ const auto& dictionary = dict_array->dictionary();
+ if (dictionary == nullptr) {
+ return Status::Corruption("Parquet dictionary array has null
dictionary for column {}",
+ _name);
+ }
+
+ // Dictionary predicates are evaluated once against the dictionary
page and produce a
+ // dictionary-entry bitmap. DATA_PAGE rows then only need an
integer-index lookup. NULL rows
+ // do not have a dictionary entry and cannot satisfy the supported
equality/IN predicates.
+ for (int64_t row = 0; row < dict_array->length(); ++row) {
+ bool keep = false;
+ if (!dict_array->IsNull(row)) {
+ const int64_t dictionary_index =
dict_array->GetValueIndex(row);
+ if (dictionary_index >= 0 &&
+ dictionary_index <
static_cast<int64_t>(dictionary_filter.size())) {
+ keep =
dictionary_filter[static_cast<size_t>(dictionary_index)] != 0;
+ }
+ if (keep) {
+ RETURN_IF_ERROR(append_arrow_binary_dictionary_value(
+ _name, *dictionary, dictionary_index,
&selected_values));
+ ++*matched_rows;
+ }
+ }
+ row_filter->push_back(keep ? 1 : 0);
+ }
+ }
+
+ if (!*used_filter) {
+ return Status::OK();
+ }
+ return append_decoded_binary_values(selected_values, column);
+}
+
+Status ScalarColumnReader::append_decoded_binary_values(const
std::vector<StringRef>& values,
+ MutableColumnPtr&
column) const {
+ DecodedColumnView view;
+ view.value_kind = decoded_value_kind(_type_descriptor);
+ view.row_count = static_cast<int64_t>(values.size());
+ view.logical_integer_bit_width = _type_descriptor.integer_bit_width;
+ view.logical_integer_is_signed = !_type_descriptor.is_unsigned_integer;
+ view.fixed_length = _type_descriptor.fixed_length;
+ view.binary_values = &values;
+
+ SCOPED_TIMER(_profile.materialization_time);
+ if (!_type->is_nullable()) {
+ if (auto* nullable_column =
check_and_get_column<ColumnNullable>(*column);
+ nullable_column != nullptr) {
+ auto& nested_column = nullable_column->get_nested_column();
+ auto& null_map = nullable_column->get_null_map_data();
+ const auto old_nested_size = nested_column.size();
+ const auto old_null_map_size = null_map.size();
+ auto st =
_type->get_serde()->read_column_from_decoded_values(nested_column, view);
+ if (!st.ok()) {
+ nested_column.resize(old_nested_size);
+ return st;
+ }
+ null_map.resize(old_null_map_size + nested_column.size() -
old_nested_size);
+ memset(null_map.data() + old_null_map_size, 0, null_map.size() -
old_null_map_size);
Review Comment:
The new `append_decoded_binary_values()` helper uses `memset` here, but this
translation unit does not include `<cstring>` directly. It may compile today
through an Arrow/Parquet transitive include, but the declaration is not
guaranteed by the headers this file owns. Please add `#include <cstring>` near
the other standard headers so this file is not include-order dependent.
--
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]
