github-actions[bot] commented on code in PR #66261:
URL: https://github.com/apache/doris/pull/66261#discussion_r3687395250
##########
be/src/format_v2/table_reader.cpp:
##########
@@ -736,32 +741,263 @@ Status TableReader::init(TableReadOptions&& options) {
}
_system_properties = create_system_properties(_scan_params);
_mapper_options.mode = TableColumnMappingMode::BY_NAME;
- _conjuncts = std::move(options.conjuncts);
+ return _replace_conjuncts(options.conjuncts);
+}
+
+Status TableReader::_clone_conjunct(const VExprContextSPtr& source,
VExprContextSPtr* cloned) {
+ DORIS_CHECK(source != nullptr);
+ DORIS_CHECK(source->root() != nullptr);
+ DORIS_CHECK(cloned != nullptr);
+ VExprSPtr root;
+ RETURN_IF_ERROR(clone_table_expr_tree(source->root(), &root));
+ *cloned = VExprContext::create_shared(std::move(root));
+ return Status::OK();
+}
+
+Status TableReader::_prepare_conjunct(const VExprContextSPtr& source,
VExprContextSPtr* prepared) {
+ DORIS_CHECK(prepared != nullptr);
+ VExprContextSPtr conjunct;
+ RETURN_IF_ERROR(_clone_conjunct(source, &conjunct));
+ RETURN_IF_ERROR(conjunct->prepare(_runtime_state, RowDescriptor {}));
+ RETURN_IF_ERROR(conjunct->open(_runtime_state));
+ *prepared = std::move(conjunct);
return Status::OK();
}
+Status TableReader::_replace_conjuncts(const VExprContextSPtrs& conjuncts) {
+ VExprContextSPtrs prepared;
+ prepared.reserve(conjuncts.size());
+ for (size_t conjunct_index = 0; conjunct_index < conjuncts.size();
++conjunct_index) {
+ VExprContextSPtr conjunct;
+ if (conjunct_index < _table_reader_owned_conjunct_count) {
+ RETURN_IF_ERROR(_prepare_conjunct(conjuncts[conjunct_index],
&conjunct));
+ } else {
+ // Scanner-owned suffixes are cloned only for pruning analysis;
preparing them here
+ // would duplicate expression state that must remain exclusively
in Scanner.
+ RETURN_IF_ERROR(_clone_conjunct(conjuncts[conjunct_index],
&conjunct));
+ }
+ prepared.push_back(std::move(conjunct));
+ }
+ _conjuncts = std::move(prepared);
+ return Status::OK();
+}
+
+Status TableReader::append_conjuncts_with_ownership(const VExprContextSPtrs&
conjuncts,
+ size_t
table_reader_owned_conjunct_count) {
+ DORIS_CHECK(!_appended_table_reader_owned_conjunct_count.has_value());
+ _appended_table_reader_owned_conjunct_count =
table_reader_owned_conjunct_count;
+ auto status = append_conjuncts(conjuncts);
+ _appended_table_reader_owned_conjunct_count.reset();
+ return status;
+}
+
+Status TableReader::append_conjuncts(const VExprContextSPtrs& conjuncts) {
+ const size_t owned_count =
+
_appended_table_reader_owned_conjunct_count.value_or(conjuncts.size());
+ DORIS_CHECK_LE(owned_count, conjuncts.size());
+ // Once Scanner owns a suffix, later predicates cannot be inserted into
the TableReader-owned
+ // prefix without reordering them ahead of that stateful/error-preserving
barrier.
+ DORIS_CHECK(owned_count == 0 || _table_reader_owned_conjunct_count ==
_conjuncts.size());
+ for (size_t conjunct_index = 0; conjunct_index < conjuncts.size();
++conjunct_index) {
+ const auto& source = conjuncts[conjunct_index];
+ VExprContextSPtr conjunct;
+ if (conjunct_index < owned_count) {
+ RETURN_IF_ERROR(_prepare_conjunct(source, &conjunct));
+ } else {
+ // Preserve Scanner as the sole owner of runtime state for
appended residuals.
+ RETURN_IF_ERROR(_clone_conjunct(source, &conjunct));
+ }
+ _conjuncts.push_back(conjunct);
+ if (conjunct_index < owned_count) {
+ ++_table_reader_owned_conjunct_count;
+ }
+ if (_current_task != nullptr && conjunct_index < owned_count) {
+ // Keep late predicates residual for this split even after a
refreshed request is
+ // queued: the physical reader may not activate it until the next
row-group boundary.
+ _remaining_conjuncts.push_back(std::move(conjunct));
+ }
+ }
+ return _queue_refreshed_scan_request();
+}
+
Status TableReader::_build_table_filters_from_conjuncts() {
_table_filters.clear();
- _constant_pruning_safe_filter_count = 0;
- bool in_safe_prefix = true;
- for (const auto& conjunct : _conjuncts) {
+ for (size_t conjunct_index = 0; conjunct_index < _conjuncts.size();
++conjunct_index) {
+ const auto& conjunct = _conjuncts[conjunct_index];
DORIS_CHECK(conjunct != nullptr);
DORIS_CHECK(conjunct->root() != nullptr);
- // `_table_filters` omits expressions without slot references, but
such an expression still
- // occupies a position in the row-level conjunct order. Record how
many localized filters
- // precede the first unsafe original conjunct so constant pruning
cannot jump over a
- // slotless non-deterministic/error-preserving barrier. Unsafe
predicates remain solely on
- // Scanner's original row-level path because localizing a clone would
execute their state
- // twice with independent state.
- if (in_safe_prefix && !_is_safe_to_pre_execute(conjunct)) {
- in_safe_prefix = false;
- }
+ const bool can_localize = is_safe_to_pre_execute(conjunct);
Review Comment:
[P1] Keep unsafe conjuncts as ordering barriers
This marks each later conjunct localizable independently, so an ordered pair
such as an earlier `assert_true`/failing cast/stateful predicate followed by a
safe predicate can execute in the opposite order: FileReader or split/constant
pruning evaluates the later predicate first. If it rejects the batch or split,
`_filter_remaining_conjuncts()` never executes the earlier unsafe predicate,
turning an error or required stateful evaluation into silent empty output.
`is_safe_to_execute_on_selected_rows()` only proves the moved predicate itself
tolerates selected rows; it does not prove that crossing an earlier unsafe
predicate preserves semantics. Please retain the safe-prefix barrier for
localization and pruning.
##########
be/src/exec/scan/file_scanner_v2.cpp:
##########
@@ -483,18 +485,22 @@ Status FileScannerV2::_get_block_impl(RuntimeState*
state, Block* block, bool* e
}
Status FileScannerV2::_filter_output_block(Block* block) {
- return
_contextualize_output_filter_status(Scanner::_filter_output_block(block),
- _get_current_format_type());
+ // TableReader is the single owner of predicates that cannot be localized,
while FileReader is
+ // the exact owner of localized predicates. Re-evaluating either set here
would duplicate
+ // mutable expression state and force predicate-only columns to be
materialized for Scanner.
+ DORIS_CHECK(_conjuncts.empty());
+ (void)block;
+ return Status::OK();
}
-Status FileScannerV2::_contextualize_output_filter_status(Status status,
-
TFileFormatType::type format_type) {
- if (!status.ok() && format_type == TFileFormatType::FORMAT_ORC) {
- // Error-preserving expressions cannot be reordered into the ORC
reader and therefore run
- // at the scanner boundary; keep their error context identical to ORC
callback failures.
- status.prepend("Orc row reader nextBatch failed. reason = ");
- }
- return status;
+size_t FileScannerV2::_last_block_rows_read(const Block& block) const {
Review Comment:
[P2] Preserve filtered-reader progress before budgeting
A late RF is now moved below Scanner for an active split.
`JniTableReader::get_block()` loops when it rejects a Java batch, and once
Parquet activates a queued refreshed request at a row-group boundary,
`ParquetScanScheduler::read_next_batch()` likewise loops and widens
zero-survivor predicate batches. Neither path returns or publishes the rejected
batch's pre-filter shape, so this fallback sees only a survivor or EOF and
Scanner's row/byte budget can be bypassed while the remaining split is drained.
Please make each predicate-owning reader publish and yield bounded pre-filter
progress, including zero-survivor batches.
--
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]
