github-actions[bot] commented on code in PR #69646:
URL: https://github.com/apache/doris/pull/69646#discussion_r4379658688
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/TopnFilterPushDownVisitor.java:
##########
@@ -92,6 +92,11 @@ public Boolean visit(Plan plan, PushDownContext ctx) {
@Override
public Boolean visitPhysicalProject(
PhysicalProject<? extends Plan> project, PushDownContext ctx) {
+ // TopN can skip or reorder input before this project. Crossing
assert_true(k < 3), for
+ // example, could suppress errors in pruned row groups instead of
preserving evaluation.
+ if
(project.getProjects().stream().anyMatch(Expression::containsVolatileOrNoneMovableExpression))
{
Review Comment:
[P2] Fence throwing date arithmetic before pushing the TopN filter
This marker check still allows `weeks_add(d, 1)`, which implements
`DateAddSubMonotonic` rather than either checked marker, although its DATEV2
implementation throws on overflow. A reduced triggering plan is:
```text
TopN(k ASC, v ASC LIMIT 1)
Project(k, weeks_add(d, 1) AS v)
FileScan(k, d)
```
The secondary order key keeps `v` below TopN. In one Parquet split/scanner,
let physical RG0 contain `(k=100, d=9999-12-31)` and RG1 contain `(k=0,
d=2020-01-01)` for several full output batches. Previously RG0 reached the
scanner projection and raised overflow before any frontier existed. The new
ordering reads RG1 first; when its output publishes bound `k=0` before the next
group opens, the footer recheck skips RG0 and the query can succeed.
`allow_row_reordering` only inspects scan conjuncts, so it cannot see the
projection fused onto ScanNode by the translator. Extend the project fence to
throwing date-arithmetic semantics and add this secondary-key/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] Include join predicates in the reordering safety fence
These request flags only describe scan conjuncts; they cannot establish
safety for expressions retained on a join above the scan. For example, this
reduced plan is still allowed by
`TopnFilterPushDownVisitor.visitPhysicalHashJoin`:
```text
TopN(l.k ASC LIMIT 1)
InnerHashJoin(l.id = r.id;
other: assert_true(l.k < 3, 'bad'))
FileScan(l.id, l.k)
small matching build r
```
`PushDownJoinOtherCondition` deliberately keeps this NoneMovable predicate
on the join, and the translator preserves it as an other-join conjunct. With
one broadcast-build/local-TopN fragment, one Parquet split/scanner, all IDs
matching, physical RG0 containing `k=100` and RG1 containing several full
batches of `k=0`, the old physical order raises `bad` before a frontier exists.
Reordering visits RG1 first; after its joined rows publish bound `0`, the new
footer check can reject RG0 before the assertion executes. The scan's reorder
flag remains true, and the new project-only visitor guard is never applicable.
Fence TopN pushdown at evaluation-sensitive join predicates and cover this
reversed-group ON-assertion case. Expanding the project-expression check in the
other finding does not fix this separate join boundary.
--
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]
