github-actions[bot] commented on code in PR #66360:
URL: https://github.com/apache/doris/pull/66360#discussion_r3698725294
##########
be/src/format_v2/table_reader.cpp:
##########
@@ -765,6 +765,104 @@ Status TableReader::_build_table_filters_from_conjuncts()
{
return Status::OK();
}
+namespace {
+
+bool same_scan_projection(const LocalColumnIndex& lhs, const LocalColumnIndex&
rhs) {
+ if (lhs.index != rhs.index || lhs.project_all_children !=
rhs.project_all_children ||
+ lhs.children.size() != rhs.children.size()) {
+ return false;
+ }
+ for (size_t index = 0; index < lhs.children.size(); ++index) {
+ if (!same_scan_projection(lhs.children[index], rhs.children[index])) {
+ return false;
+ }
+ }
+ return true;
+}
+
+const LocalColumnIndex* find_scan_projection(const FileScanRequest& request,
+ LocalColumnId column_id) {
+ const auto find_by_id = [column_id](const std::vector<LocalColumnIndex>&
projections) {
+ return std::ranges::find_if(projections, [column_id](const
LocalColumnIndex& projection) {
+ return projection.column_id() == column_id;
+ });
+ };
+ auto it = find_by_id(request.predicate_columns);
+ if (it != request.predicate_columns.end()) {
+ return &*it;
+ }
+ it = find_by_id(request.non_predicate_columns);
+ return it == request.non_predicate_columns.end() ? nullptr : &*it;
+}
+
+bool same_physical_scan_layout(const FileScanRequest& lhs, const
FileScanRequest& rhs) {
+ if (lhs.local_positions != rhs.local_positions) {
+ return false;
+ }
+ for (const auto& [column_id, _] : lhs.local_positions) {
+ const auto* lhs_projection = find_scan_projection(lhs, column_id);
+ const auto* rhs_projection = find_scan_projection(rhs, column_id);
+ if (lhs_projection == nullptr || rhs_projection == nullptr ||
+ !same_scan_projection(*lhs_projection, *rhs_projection)) {
+ return false;
+ }
+ }
+ return true;
+}
+
+} // namespace
+
+Status TableReader::refresh_conjuncts(VExprContextSPtrs conjuncts) {
+ _conjuncts = std::move(conjuncts);
+ if (_data_reader.reader == nullptr) {
+ // The split is prepared but its physical reader has not opened yet.
open_reader() will use
+ // this newest snapshot directly, so no pending request is needed.
+ return Status::OK();
+ }
+ if (!_data_reader.reader->supports_scan_request_refresh()) {
+ return Status::OK();
+ }
+
+ RETURN_IF_ERROR(_build_table_filters_from_conjuncts());
+ // create_scan_request() rebuilds mapping projections in place. Build late
predicates with an
+ // isolated mapper so the active row group cannot observe an unprepared or
incompatible mapper
+ // before its physical request reaches the reader's safe activation
boundary.
+ auto refreshed_mapper =
_data_reader.reader->create_column_mapper(_mapper_options);
+ DORIS_CHECK(refreshed_mapper != nullptr);
+ RETURN_IF_ERROR(refreshed_mapper->create_mapping(_projected_columns,
_partition_values,
+
_data_reader.file_schema));
+ auto refreshed_request = std::make_shared<FileScanRequest>();
+ RETURN_IF_ERROR(refreshed_mapper->create_scan_request(
+ _table_filters, _projected_columns, refreshed_request.get(),
_runtime_state,
+ _file_scan_request == nullptr ? nullptr :
&_file_scan_request->local_positions));
+ if (_push_down_agg_type == TPushAggOp::type::COUNT &&
_push_down_count_columns.has_value() &&
+ _push_down_count_columns->empty()) {
+ for (const auto& column : refreshed_request->non_predicate_columns) {
+
refreshed_request->count_star_placeholder_columns.push_back(column.column_id());
+ }
+ }
+ RETURN_IF_ERROR(customize_file_scan_request(refreshed_request.get()));
+ if (_file_scan_request == nullptr ||
+ !same_physical_scan_layout(*refreshed_request, *_file_scan_request)) {
Review Comment:
[P1] Keep real COUNT(*) carrier values available until the refreshed request
activates. When an RF promotes the current placeholder slot to a predicate,
this compatibility check still accepts the request because
positions/projections match. The Parquet scheduler does not activate it until
the next row-group boundary, while `Scanner::_filter_output_block()` starts
evaluating the new RF immediately; the active request has no reader for this
slot and materializes defaults. For example, a current group containing ids
`[1,10]` with a late `id > 5` can test `[0,0]` and discard the matching row.
Please avoid placeholdering possible RF carriers while filters are pending (or
otherwise make the transition/fallback expose real values), and add a
mid-row-group COUNT(*) refresh result test.
##########
be/src/format_v2/parquet/parquet_scan.cpp:
##########
@@ -908,6 +961,29 @@ void ParquetScanScheduler::reset() {
reset_current_row_group();
}
+void
ParquetScanScheduler::set_scan_request(std::shared_ptr<format::FileScanRequest>
request) {
+ DORIS_CHECK(request != nullptr);
+ _active_request = std::move(request);
+ _pending_request.reset();
+ _predicate_schedule_request = nullptr;
+}
+
+void
ParquetScanScheduler::queue_scan_request(std::shared_ptr<format::FileScanRequest>
request) {
+ DORIS_CHECK(request != nullptr);
+ _pending_request = std::move(request);
+}
+
+void
ParquetScanScheduler::activate_pending_scan_request_at_row_group_boundary() {
+ if (_has_current_row_group || !_pending_predicate_selection.empty() ||
+ _pending_request == nullptr) {
+ return;
+ }
+ // Column readers and predicate schedules retain request-derived state for
one row group. Swap
+ // only after they are gone; the refreshed request may promote a lazy
column to a predicate.
+ _active_request = std::move(_pending_request);
Review Comment:
[P2] Invalidate all predicate-snapshot-derived optimization state when this
request becomes active. The unopened Row Group plans were footer-pruned with
the old request, but their deferred finalization is `EXPENSIVE_ONLY`, so a late
RF whose min/max rejects a group never gets the cheap footer-statistics pass.
The old `_predicate_survival_ratio`/position stats also survive, allowing
pre-RF dense batches to trigger full lazy-output prefetch before the new
selective RF is sampled. Please make remaining planning/adaptive state
request-versioned (or rerun current-request footer pruning and reset/preserve
stats only when predicates are proven equivalent), with a multi-group late-RF
I/O regression.
--
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]
