github-actions[bot] commented on code in PR #66356:
URL: https://github.com/apache/doris/pull/66356#discussion_r3698349055
##########
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 {
+ const auto& stats = _table_reader->last_materialized_block_stats();
Review Comment:
[P2] Account for JNI batches before falling back to survivor rows.
`JniTableReader::get_block()` applies the now-owned (including late) conjuncts
and loops again whenever a Java batch is fully rejected, but it never records
`MaterializedBlockStats`. Since Scanner filtering is now a no-op, a late
selective RF can drain the rest of a large JNI split in one scheduler turn,
while these hooks see only zero/survivor rows and cannot enforce the row/byte
budget or learn the pre-filter width. Please record the pre-filter JNI
rows/bytes and yield after one rejected materialized batch, as the base
TableReader path now does.
##########
be/src/format_v2/table_reader.cpp:
##########
@@ -1085,11 +1324,10 @@ Status
TableReader::_evaluate_partition_prune_conjuncts(const VExprContextSPtrs&
for (const auto& conjunct : conjuncts) {
DORIS_CHECK(conjunct != nullptr);
DORIS_CHECK(conjunct->root() != nullptr);
- // Keep only the safe prefix of the original conjunct order. If an
unsafe conjunct is
- // skipped, a later predicate could prune the split before the unsafe
one reaches its
- // normal row-level evaluation point.
- if (!_is_safe_to_pre_execute(conjunct)) {
- break;
+ // Unsafe or non-deterministic predicates remain at TableReader. Safe
predicates are
+ // independent pruning candidates even when an earlier predicate
cannot be pre-executed.
+ if (!is_safe_to_pre_execute(conjunct)) {
+ continue;
Review Comment:
[P1] Preserve the unsafe conjunct as an ordering barrier. With a partition
value `p = BIGINT_MIN` and ordered predicates `mod(p, -1) > 0 AND p > 0`, the
first predicate is unsafe because it must raise `INVALID_ARGUMENT`, while the
second is safely false. Continuing here lets the second predicate prune the
split before the modulo executes, suppressing that required error; V1 and the
pre-change V2 path stop at the first unsafe conjunct for exactly this reason.
Please restore the safe-prefix barrier for partition/constant pruning and later
localization, and cover this error-preserving order.
##########
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));
Review Comment:
[P1] Open rebuilt predicate contexts with per-scanner function state.
`_clone_conjunct()` creates a fresh `VExprContext`, so this `open()` uses
`FRAGMENT_LOCAL`; however residual functions such as LIKE/REGEXP initialize
their `LikeState` only for `THREAD_LOCAL` and then dereference it during
execution (Java/Python UDFs have the same contract). Because LIKE is unsafe to
localize, `WHERE s LIKE 'x%'` now reaches TableReader's residual filter after
Scanner relinquishes the predicate and fails with missing/null function state.
Please preserve the original fragment state and clone/open the rewritten
context with thread-local state, and add a residual LIKE regression.
##########
be/src/format_v2/parquet/parquet_scan.cpp:
##########
@@ -2731,9 +2743,10 @@ Status ParquetScanScheduler::read_next_batch(
};
while (true) {
if (!_has_current_row_group) {
+ activate_pending_scan_request_at_row_group_boundary();
Review Comment:
[P2] Bound all-reject work after activating the refreshed request. If a late
RF rejects the remaining row groups, `read_next_batch()` keeps growing and
retrying zero-survivor probes until a survivor or EOF; Parquet publishes its
raw-row delta only after that call returns, so Scanner cannot charge its row
budget or reach cancellation checkpoints during the drain. This PR newly moves
active-split late RFs into that loop. Please return a bounded empty progress
unit (with consumed raw rows/bytes) and add an all-reject late-RF regression.
##########
be/src/format_v2/column_mapper.cpp:
##########
@@ -2197,27 +2235,44 @@ Status TableColumnMapper::create_scan_request(
// 2. Build referenced predicate columns
// Hidden filter mappings must be built before localizing filters, so that
they can be localized together with visible mappings and referenced by
localized filter expressions.
RETURN_IF_ERROR(_build_hidden_filter_mappings(table_filters));
- RETURN_IF_ERROR(localize_filters(table_filters, file_request,
runtime_state));
- for (const auto& mapping : _hidden_mappings) {
- if (!mapping.file_local_id.has_value()) {
- continue;
- }
- const auto local_id = LocalColumnId(*mapping.file_local_id);
- const bool is_visible_output =
+ FilterLocalizationResult local_localization_result;
+ auto* exact_localization_result =
+ localization_result == nullptr ? &local_localization_result :
localization_result;
+ RETURN_IF_ERROR(localize_filters(table_filters, file_request,
runtime_state,
+ exact_localization_result));
+ for (const auto& predicate_column : file_request->predicate_columns) {
+ const auto local_id = predicate_column.column_id();
+ const bool value_required_after_filter =
std::ranges::any_of(_mappings, [local_id](const ColumnMapping&
visible_mapping) {
return visible_mapping.file_local_id.has_value() &&
- LocalColumnId(*visible_mapping.file_local_id) ==
local_id;
+ LocalColumnId(*visible_mapping.file_local_id) ==
local_id &&
+ visible_mapping.value_required_after_filter;
});
- if (is_visible_output) {
+ if (value_required_after_filter) {
continue;
}
- // File-local filtering is an optimization; Scanner still evaluates
the original
- // table-level conjunct after TableReader returns. Only truly hidden
mappings are absent
- // from that scanner-visible block and may safely discard their
payload here.
- if (std::ranges::any_of(file_request->predicate_columns,
- [local_id](const LocalColumnIndex& projection)
{
- return projection.column_id() == local_id;
- }) &&
+ bool referenced_by_filter = false;
+ bool referenced_only_by_localized_filters = true;
+ for (size_t filter_index = 0; filter_index < table_filters.size();
++filter_index) {
+ const bool references_local_column = std::ranges::any_of(
+ table_filters[filter_index].global_indices,
[&](GlobalIndex global_index) {
+ const auto* mapping =
_find_filter_mapping(global_index);
+ return mapping != nullptr &&
mapping->file_local_id.has_value() &&
+ LocalColumnId(*mapping->file_local_id) ==
local_id;
+ });
+ if (!references_local_column) {
+ continue;
+ }
+ referenced_by_filter = true;
+ referenced_only_by_localized_filters &=
+ exact_localization_result->localized_filters[filter_index];
+ }
+ // A scan tuple can contain a filter slot that its upstream projection
does not consume.
+ // For example, `SELECT SUM(measure) FROM t WHERE filter_key BETWEEN 1
AND 20` needs
+ // `filter_key` while evaluating the predicate, but not afterwards.
Discard its payload
+ // only when every predicate referencing the physical column was
localized exactly for this
+ // split; schema evolution or a TableReader residual must retain the
value.
+ if (referenced_by_filter && referenced_only_by_localized_filters &&
Review Comment:
[P1] Keep predicate-only values needed by later consumers. This mapper
decision only proves that query filters localized into FileReader; it does not
prove Parquet can discard the value while staging them. On dictionary encoding,
the direct path checks only same-column dictionary residuals, emits a default
placeholder, and marks the position materialized even when
`residual_predicate_positions` says a multi-column residual or Iceberg
equality-delete conjunct still needs the real value. Those later predicates
then evaluate defaults and can return wrong rows. Please gate dictionary
predicate-only mode on the residual/delete position set (as the fixed/runtime
direct path does) and cover both overlaps.
--
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]
