github-actions[bot] commented on code in PR #65965:
URL: https://github.com/apache/doris/pull/65965#discussion_r3639933927
##########
be/src/format_v2/parquet/parquet_scan.cpp:
##########
@@ -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));
Review Comment:
[P1] Preserve decode failures for rows skipped by later stages
This safety check covers the `VExpr` tree, but the later column is also
subject to physical-to-logical conversion while it is materialized. After an
earlier AND child rejects a row, `materialize_predicate_positions()` calls the
native reader with that row filtered out; `decode_selected_values()` handles
`FILTERED_CONTENT` with `skip_values()`, and the sparse path runs the SerDe
only for selected ranges. For example, with `enable_strict_cast=true`, an
out-of-range INT96 value in a row rejected by stage 1 used to make eager
materialization return `DataQualityError`, while the staged path skips that
payload and can return rows successfully. This is below the `format::Cast`
safeguard from the existing thread. Please either limit staged materialization
to file-column conversions that are total for every physical value, or validate
filtered payloads equivalently to eager decoding, and cover a strict-cast
partial-survival invalid-INT96 case.
##########
be/benchmark/parquet/benchmark_parquet_reader.hpp:
##########
@@ -284,10 +339,17 @@ inline Block make_block(const
std::vector<format::ColumnDefinition>& schema) {
}
struct ReaderSession {
+ ~ReaderSession() {
Review Comment:
[P2] Destroy the expression session while timing is paused
`run_reader()` resumes timing immediately after `reader->close()`, but the
local `session` is destroyed only at the end of the loop body. For this new
case, that destructor calls `context->close()` in the measured region. The
reader still retains the request that owns the same prepared context, and
`VExprContext::~VExprContext()` closes it again because explicit `close()` does
not clear `_opened`; request/reader destruction is timed as well. That makes
the complex-residual number include duplicate expression teardown even though
the benchmark contract excludes setup/close work. Please reset/destroy the
session while timing is paused and use a single close ownership path.
--
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]
