github-actions[bot] commented on code in PR #66035:
URL: https://github.com/apache/doris/pull/66035#discussion_r3649611584
##########
be/src/exec/scan/file_scanner_v2.cpp:
##########
@@ -450,18 +461,33 @@ 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());
-}
-
-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.
+ if (_scanner_residual_conjuncts.empty() || block->rows() == 0) {
+ return Status::OK();
+ }
+ SCOPED_TIMER(_scanner_residual_filter_timer);
+ const size_t rows_before_filter = block->rows();
+ auto status = VExprContext::filter_block(_scanner_residual_conjuncts,
block, block->columns());
+ if (!status.ok() && _params != nullptr &&
+ _get_current_format_type() == TFileFormatType::FORMAT_ORC) {
status.prepend("Orc row reader nextBatch failed. reason = ");
}
- return status;
+ RETURN_IF_ERROR(status);
+ const int64_t filtered_rows = cast_set<int64_t>(rows_before_filter -
block->rows());
+ _counter.num_rows_unselected += filtered_rows;
+ if (_scanner_residual_rows_filtered_counter != nullptr) {
+ COUNTER_UPDATE(_scanner_residual_rows_filtered_counter, filtered_rows);
+ }
+ return Status::OK();
+}
+
+size_t FileScannerV2::_last_block_rows_read(const Block& block) const {
Review Comment:
[P1] Preserve the Scanner progress bound for JNI batches
This fallback is useful only when the reader returns once per materialized
batch. `JniTableReader::get_block()` neither records `MaterializedBlockStats`
nor returns when a Java batch is fully filtered; it keeps reading until a
survivor or EOF. Therefore direct JNI and Hudi/Paimon JNI splits can drain an
entire reject-all split in one Scanner task, while adaptive sizing learns from
survivors instead of the pre-filter Java block. Please reset/record JNI
materialization stats before predicate filtering and return an empty progress
batch after one rejected Java batch, with a JNI coverage case.
##########
be/src/format_v2/table_reader.cpp:
##########
@@ -734,35 +739,170 @@ 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));
Review Comment:
[P1] Keep analysis-only late conjuncts out of JNI filtering
When a Scanner-owned suffix already exists, `_sync_table_reader_conjuncts()`
passes `owned_count=0`, so this branch deliberately creates an
unprepared/unopened analysis clone. The JNI path still prepares and filters the
complete `_conjuncts` vector (`JniTableReader` plus the JDBC/Hudi/MaxCompute
finalizers), so the next JNI batch executes this clone;
`RuntimeFilterExpr::execute_filter()` DCHECKs `_open_finished`, and Scanner
would evaluate its separately owned RF again. Please make every JNI finalizer
evaluate only the TableReader-owned prefix (without re-opening analysis-only
suffixes) and add an active-split late-RF JNI test.
--
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]
