github-actions[bot] commented on code in PR #69646:
URL: https://github.com/apache/doris/pull/69646#discussion_r4384892963
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/TopnFilterPushDownVisitor.java:
##########
@@ -108,9 +136,46 @@ public Boolean visitPhysicalProject(
return project.child().accept(this, ctxProjectProbeExpr);
}
+ private static boolean hasEvaluationSensitiveExpressions(Plan plan) {
+ return plan.getExpressions().stream().anyMatch(expression ->
!safeOnReducedInput(expression));
+ }
+
+ private static boolean safeOnReducedInput(Expression expression) {
+ // Determinism and monotonicity do not imply total evaluation:
from_second(-1), for example,
+ // throws. Keep a closed set of proven-total expressions instead of
enumerating throwing
+ // functions. Unknown functions, casts and UDFs must preserve their
original evaluation domain.
+ if (expression instanceof Slot || expression instanceof Literal) {
+ return true;
+ }
+ boolean safeNode = expression instanceof Alias || expression
instanceof Nullable
+ || expression instanceof And || expression instanceof Or ||
expression instanceof Not
+ || expression instanceof IsNull;
+ if (expression instanceof EqualTo || expression instanceof
NullSafeEqual
+ || expression instanceof LessThan || expression instanceof
LessThanEqual
+ || expression instanceof GreaterThan || expression instanceof
GreaterThanEqual
+ || expression instanceof InPredicate) {
+ safeNode = expression.children().stream().allMatch(child -> {
+ DataType type = child.getDataType();
+ return type.isNumericType() || type.isBooleanType() ||
type.isStringLikeType()
+ || type.isDateLikeType() || type.isTimeType() ||
type.isIPType()
+ || type.isUuidType() || type.isNullType();
+ });
+ } else if (expression instanceof Substring) {
+ safeNode = expression.children().subList(1,
expression.arity()).stream()
+ .allMatch(Literal.class::isInstance);
+ } else if (expression instanceof DateAddSubMonotonic) {
+ // Zero intervals are identities on already-typed dates; nonzero
arithmetic can overflow.
+ safeNode =
!ExpressionUtils.containsUnsafeDateArithmetic(expression);
+ }
+ return safeNode &&
expression.children().stream().allMatch(TopnFilterPushDownVisitor::safeOnReducedInput);
+ }
+
@Override
public Boolean visitPhysicalSetOperation(
PhysicalSetOperation setOperation, PushDownContext ctx) {
+ if (hasEvaluationSensitiveExpressions(setOperation)) {
Review Comment:
[P2] Check sensitive siblings before reducing a set build input
The set operator's `getExpressions()` contains output slots, so this check
does not cover expressions in its later inputs. Those inputs can still be
reduced indirectly:
`RuntimeFilterGenerator.generateRuntimeFilterForSetOperation` builds a filter
from child 0 into later children, and runtime-filter pushdown crosses a project
without checking its unrelated outputs. A reduced triggering tree is:
```text
TopN(k ASC, v ASC LIMIT 1)
UnionAll
Safe(k, v)
Intersect
ParquetScan A(k INT, v DATETIMEV2(6))
Project(k, from_microsecond(ts) AS v)
Scan B(k INT, ts BIGINT)
```
On one BE, preserve this tree by disabling `PUSH_DOWN_TOP_N_THROUGH_UNION`,
`REORDER_INTERSECT`, and `INFER_SET_OPERATOR_DISTINCT`. Use a high
`topn_filter_ratio`, `enable_runtime_filter_prune=false`,
`runtime_filter_type=1`, and `runtime_filter_wait_infinitely=true`; keep A
below the set-filter build-row threshold. The set filter targets the first
nonuniform output `k`. A's `v` is a matching nullable Parquet
TIMESTAMP(MICROS), so its mappings are trivial. Put `k=100` in physical RG0 and
full `k=0` groups later; B includes `(100,-1)`. Choose the reachable
interleaving where A returns its first block before Safe supplies the outer
bound `k=0`, then A finishes before B opens.
Physical-first A already retained 100, so the published set filter includes
it and B evaluates `from_microsecond(-1)`, raising an error. Best-first A
initially returns 0; after Safe tightens the bound, RG0 is skipped, the set
filter excludes 100, and B's error disappears. UNION streams its independent
branch while the set build is unfinished; query-owned TopN predicates are
shared across fragments on the same BE.
The existing epoch-project fix checks only the traversed path; this is a
separate sibling consumer through a generated filter. Fence sensitive later
inputs before permitting this reduction, and add the set-RF/sibling-error
regression.
##########
be/src/format_v2/parquet/parquet_scan.cpp:
##########
@@ -1409,21 +1414,147 @@ std::vector<format::LocalColumnIndex>
ParquetScanScheduler::adaptive_predicate_p
return result;
}
+namespace {
+
+struct TopNOrderingColumn {
+ const VTopNPred* predicate;
+ const ParquetColumnSchema* schema;
+};
+
+std::optional<TopNOrderingColumn> find_topn_ordering_column(
+ const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
+ const format::FileScanRequest& request) {
+ const size_t safe_count =
+ std::min(request.metadata_pruning_safe_conjunct_count,
request.conjuncts.size());
+ const VTopNPred* topn = nullptr;
+ for (size_t idx = 0; idx < safe_count; ++idx) {
+ const auto& root = request.conjuncts[idx]->root();
+ if (root->is_topn_filter() && root->can_evaluate_zonemap_filter()) {
+ topn = assert_cast<const VTopNPred*>(root.get());
+ break;
+ }
+ }
+ if (topn == nullptr) {
+ return std::nullopt;
+ }
+ const auto* slot = assert_cast<const VSlotRef*>(topn->get_child(0).get());
+ const auto position = std::ranges::find_if(request.local_positions,
[&](const auto& entry) {
+ return entry.second.value() == slot->column_id();
+ });
+ if (position == request.local_positions.end()) {
+ return std::nullopt;
+ }
+ const auto column = std::ranges::find_if(file_schema, [&](const auto&
schema) {
+ return schema->local_id == position->first.value();
+ });
+ if (column == file_schema.end() || (*column)->kind !=
ParquetColumnSchemaKind::PRIMITIVE ||
+ (*column)->max_repetition_level != 0 ||
+ !expr_zonemap::data_types_compatible((*column)->type,
slot->data_type())) {
+ return std::nullopt;
+ }
+ return TopNOrderingColumn {.predicate = topn, .schema = column->get()};
+}
+
+} // namespace
+
+void ParquetScanScheduler::prepare_topn_row_group_order(
+ const ParquetFileContext& file_context,
+ const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
+ const format::FileScanRequest& request) {
+ _row_group_order_prepared = true;
+ const auto binding = find_topn_ordering_column(file_schema, request);
+ if (!binding.has_value()) {
+ return;
+ }
+ const auto* topn = binding->predicate;
+ const auto& schema = *binding->schema;
+ const size_t safe_count =
+ std::min(request.metadata_pruning_safe_conjunct_count,
request.conjuncts.size());
+ const auto& metadata = file_context.native_metadata->to_thrift();
+ _topn_footer_recheck = std::ranges::any_of(_scan_plan->row_groups,
[&](const auto& plan) {
+ const auto& group = metadata.row_groups[plan.row_group_id];
+ DORIS_CHECK(schema.leaf_column_id >= 0 && schema.leaf_column_id <
group.columns.size());
+ const auto& chunk = group.columns[schema.leaf_column_id];
+ return chunk.__isset.meta_data && chunk.meta_data.__isset.statistics;
+ });
+ // Files without statistics cannot benefit from repeated footer checks.
Row-ID fetches promise
+ // increasing positions, and an unsafe table expression must continue
observing file order.
+ if (!_topn_footer_recheck || request.row_ids.has_value() ||
!request.allow_row_reordering ||
Review Comment:
[P2] Fence throwing native materialization for identity mappings
A trivial table mapping does not establish that physical Parquet
materialization is total. This reduced plan remains eligible with
`enable_strict_cast=true` and matching nullable file/table types:
```text
TopN(k ASC, d ASC LIMIT 1)
FileScan(k INT, d DATE)
```
In one split/scanner using PLAIN pages, let physical RG0 contain `k=100` and
DATE day `2932897` (`10000-01-01`), then place several full batches of `k=0`
with ordinary dates in later groups. Parquet DATE uses a signed INT32 epoch-day
domain ([format
specification](https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#date));
this value exceeds Doris's date range without requiring a malformed file or
schema evolution.
Logical DATE infers `TYPE_DATEV2`, so identical file/table types keep
`mapping.is_trivial=true`. Nevertheless, `DateV2ParquetConsumer` calls
`decode_parquet_date`, which returns `DataQualityError` for this day number;
the strict mode passed by `NativeColumnReader` prevents NULL substitution.
Physical order therefore fails on RG0 before a bound exists. The new ordering
reads valid low-key groups first, publishes `k=0`, and prunes RG0 before native
conversion, suppressing the error.
This is distinct from the existing table `EvolutionCast` report: no table
cast or nontrivial mapping exists here. Require a file-reader proof that native
logical materialization is safe to reorder, or conservatively fence the
affected conversion paths, and add a strict identity-DATE late-bound regression.
##########
be/src/format_v2/table_reader.cpp:
##########
@@ -1614,6 +1615,15 @@ bool TableReader::_is_safe_to_pre_execute(const
VExprContextSPtr& conjunct) {
return predicate->is_safe_to_execute_on_selected_rows();
}
+bool TableReader::_is_safe_to_reorder_rows(const TableColumnMapper& mapper)
const {
+ // FE/scan conjuncts cannot see implicit materialization: an unrelated
DATE -> TIMESTAMP_NS
+ // output can overflow. Only direct mappings are proven safe, including
their nested children;
+ // a complex remap can run child casts even when its outer projection is a
SlotRef.
+ return std::ranges::all_of(_conjuncts, _is_safe_to_pre_execute) &&
+ std::ranges::all_of(mapper.mappings(),
Review Comment:
[P2] Fence hidden equality-delete key casts before reordering
`mapper.mappings()` covers projected output mappings, while Iceberg
customization can append a throwing cast to `FileScanRequest::delete_conjuncts`
for an unprojected equality-delete key. A reduced triggering plan is:
```text
TopN(k ASC LIMIT 1)
Iceberg FileScan(output: k INT; hidden delete key: ts TIMESTAMP_NS)
retained Parquet: k INT, ts DATE
```
Use an Iceberg v3 DATE-to-TIMESTAMP_NS promotion with stable field IDs and
an applicable newer equality-delete file containing a representable,
nonmatching `ts`. `_append_equality_delete_predicates` inserts `EvolutionCast`
for the hidden data key; `EvolutionCast::prepare` forces strict conversion for
this promotion. Neither that expression nor its hidden mapping is in the two
collections checked here. Reconciliation sees the unchanged projected `k`, so
it retains the TopN conjunct and permits reordering.
In one split/scanner, put `(k=100, ts=1000-01-01)` in physical RG0, followed
by several full low-key groups `(k=0, ts=2024-01-02)`. Physical order evaluates
the hidden delete-key cast and raises overflow before any bound. Best-first
reads the valid groups, publishes `k=0`, and rejects RG0 at the new footer
check before delete evaluation, allowing the query to succeed.
This is separate from the existing projected-output conversion thread: its
repaired visible-mapping proof cannot inspect this customization-created delete
expression. Include finalized delete-key evaluation in the initial and refresh
safety proof and cover the output-only-`k` case.
--
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]
