github-actions[bot] commented on code in PR #66261:
URL: https://github.com/apache/doris/pull/66261#discussion_r3684658688
##########
be/src/format_v2/table_reader.cpp:
##########
@@ -736,35 +741,272 @@ 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));
Review Comment:
[P1] Preserve late-filter values until ownership transfers
This appends the late RF to TableReader residuals immediately, but the
active Parquet request may already have marked its slot predicate-only and
replaced its values with defaults. The new residual then filters those
placeholders instead of the real values; after the next row-group boundary, the
refreshed request can still be predicate-only while this residual remains, so
valid rows continue to be dropped. Please keep every possible late-RF slot
materialized, or defer this residual until the reader acknowledges activation
and retire it atomically when file-level ownership begins.
##########
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 values needed by later file-local stages
`referenced_only_by_localized_filters` only says each table filter moved
into the file request; it does not mean this column has no later consumer. With
dictionary-encoded Parquet, `id > 1` can run as the single-column dictionary
stage and, because `id` is predicate-only here, replace surviving IDs with
defaults. A later localized `id + other > 10` stage then evaluates those
defaults. Iceberg has the same problem because `customize_file_scan_request()`
appends equality-delete conjuncts only after this flag is computed, allowing
deleted rows to survive when their key is defaulted. Please derive
predicate-only status from all request consumers/stages, including delete
conjuncts, or defer payload destruction until the last such consumer has run.
--
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]
