This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new e493c61d44a [improvement](be) Prioritize Parquet V2 row groups for
TopN scans (#69646)
e493c61d44a is described below
commit e493c61d44af3de66134c96ef25796492e61c79b
Author: Gabriel <[email protected]>
AuthorDate: Wed Oct 28 10:55:36 2026 +0800
[improvement](be) Prioritize Parquet V2 row groups for TopN scans (#69646)
### What problem does this PR solve?
Problem Summary:
Parquet V2 currently visits row groups in physical file order. For a
small `ORDER BY ... LIMIT`, the best values can be near the end of a
file. Earlier groups are then decoded before the TopN sink publishes a
useful bound. Even after a bound arrives, the deferred row-group path
does not recheck the footer unless a new scan request is published; TopN
updates its bound in place. `VTopNPred` also lacks the ZoneMap interface
needed to reject whole row groups.
For example, consider four groups containing consecutive integer keys
`[300, 399]`, `[200, 299]`, `[100, 199]`, `[0, 99]` and `ORDER BY k ASC
LIMIT 3`. Reading the last group first produces a bound of `2`; the
other three groups can be rejected using their footer minimum before
dictionary/index probes, prefetch, or data decoding. The descending case
uses the footer maximum.
### Optimization approach
- Add conservative ZoneMap evaluation to `VTopNPred`. Snapshot the
current runtime bound at each evaluation, retain equality for secondary
sort keys, honor NULL ordering, and fall back when types, ranges, or
floating-point NaN information are unsuitable.
- Prioritize selected row groups by the leading TopN slot's sanitized
footer minimum/maximum. Resolve the file-local slot through the
request's column mapping. Keep execution indices separate from the
physical plans so row IDs, delete coordinates, and Condition Cache
boundaries remain valid.
- Recheck the latest TopN bound at every row-group boundary before
expensive metadata probes and I/O. Files without statistics retain the
existing path. Reordering respects row-ID fetch order and the request's
existing metadata-pruning boundary; missing or unusable statistics never
exclude a group by themselves.
### Result and evaluation contract
The optimization must preserve the final TopN result, including
secondary-key ties and NULL placement. It may avoid evaluating
expressions or materializing columns for rows that a valid TopN bound
proves cannot contribute to that result. Preserving potential exceptions
from such excluded rows is not a requirement: a DATE overflow in a
discarded row group, for example, does not require reading that group.
Existing TopN/runtime-filter pushdown eligibility and file-to-table
localization remain unchanged.
No executed error is caught, ignored, or converted to success. A
conversion that is actually needed still follows the existing
strict/non-strict rules and propagates its failure. This distinction
also applies to unprojected equality-delete keys and set-operation
inputs reduced through existing runtime filters. It does not permit
skipping candidate rows, bypassing deletes for retained rows,
re-evaluating a volatile leading key as a different value, or trusting
incompatible statistics.
Earlier follow-ups added FE expression whitelists and table-mapping
restrictions to preserve errors in rows outside the result. Those
restrictions are removed; they unnecessarily reduced existing
optimization eligibility. The production change is confined to the TopN
ZoneMap capability and Parquet scheduling. Existing metadata-pruning
boundaries and ordered row-ID fetches remain respected. Upgrade
compatibility for legacy FE requests using Parquet V1 is outside the
scope of this PR; the follow-up dedicated to that case has been
reverted.
### Eligibility and limits
- FE must already generate a TopN runtime filter for the plan and
leading-key type. Its existing selectivity check remains
`max(estimated_scan_rows, 1) * topn_filter_ratio > LIMIT + OFFSET`
(default ratio: `0.5`); existing window, CTE, nested-TopN and join-side
rules remain unchanged.
- This row-group scheduling optimization applies to Parquet V2. After
localization, the leading-key filter must reference a direct,
non-repeated primitive file column with a compatible type. Arbitrary
computed keys such as `abs(k)` are not mapped to footer bounds.
Projected or secondary-key expressions do not need a new non-throwing
proof.
- Reordering requires at least two selected row groups, at least one
usable leading-key range, no ordered row-ID fetch, and all localized
conjuncts inside the existing metadata-pruning-safe boundary. Ordering
is within the current file/split, without coordinating a global order
across files or scanners. Current-bound footer rechecks can still apply
without reordering when their own eligibility holds.
- ASC/DESC, NULL ordering, OFFSET and multiple sort keys retain their
existing semantics. Leading-key equality is retained for secondary keys.
Missing/untrusted statistics, incompatible types, or an unavailable
bound retain candidates; they cannot by themselves reject a group.
- Benefits depend on reaching promising groups before the others are
read and on having separable ranges. Overlapping ranges, many
leading-key ties, one group per split, or late feedback can
substantially reduce the gain. No new production setting is introduced.
### Measured optimization effect
The included opt-in `ParquetScanTest.TopNRowGroupOrderMicrobench`
exercises the actual Parquet V2 reader plus a bounded TopK heap and
per-block updates of the real runtime predicate. Its workload
corresponds to:
```sql
SELECT k FROM parquet_source ORDER BY k ASC LIMIT 10;
SELECT k FROM parquet_source ORDER BY k DESC LIMIT 10;
```
Fixture: one required INT32 column, 128 row groups x 4,096 rows (524,288
rows), disjoint key ranges, and the best group physically last for each
direction. Files use uncompressed PLAIN DataPageV2 with dictionary and
PageIndex disabled, isolating footer pruning. Every iteration verifies
all ten final values.
| Per reader query | Before | After |
| --- | ---: | ---: |
| Row groups decoded | 128 | 1 |
| RawRowsRead | 524,288 | 4,096 (-99.22%) |
| RowGroupsFilteredByMinMax | 0 | 127 |
| Actual local read bytes | 2,149,888 | 65,602 (-96.95%) |
| ASC median latency | 10.698 ms | 0.669 ms (~16.0x) |
| DESC median latency | 8.590 ms | 0.660 ms (~13.0x) |
Method: Clang 20 RELEASE builds, warm local filesystem cache, the same
pinned CPU, nine paired samples alternating baseline/candidate order,
and 32 reader queries per sample after warmup. These paired timings were
collected before the eligibility cleanup; the direct one-column workload
exercises the same scheduling path. Fresh reader correctness tests still
verify best-first reads and late-bound group rejection. The baseline
restores `vtopn_pred.{h,cpp}` and `parquet_scan.{h,cpp}` from the
pre-optimization base `8d2bd1e14b6d95d923d43e00857da326639cc466` while
retaining the identical benchmark harness and request plumbing; the
benchmark calls ParquetReader directly. All paired checksums match. Work
counters are identical across all samples. ASC candidate samples range
from 0.647 to 1.097 ms and DESC from 0.642 to 0.745 ms; the shared host
has timing outliers. These are reader microbenchmark results; complete
SQL gains depend on file splits, parallelism, feedback latency, and
storage.
### Validation and reproduction
- Fresh verification of the revised implementation: 497 BE tests passed
(one opt-in microbenchmark skipped in the ordinary run), and all 31 FE
tests passed across the four suites below. The three aggregation query
shapes also pass the final translated `TOPN OPT:` EXPLAIN assertion. The
output-evolution optimization test and FE filter-generation test first
failed on the preceding implementation, then passed after removal of the
unnecessary fences.
- Reader integration covers ASC/DESC, top-level/nested
DATE-to-TIMESTAMP_NS output evolution, strict cast on/off, strict
identity DATE materialization, and an unprojected Iceberg
equality-delete key. Excluded groups are skipped. The native and
hidden-delete cases include ordinary-scan and reversed-TopN controls
that still raise the original conversion error when the invalid group is
required; existing output-evolution tests also retain their overflow
checks. The tests verify all surviving keys/converted values and row
counts with real per-block TopN feedback.
- FE Checkstyle (zero violations), clang-format 16, BE build hygiene,
and changed-line clang-tidy for every affected C++ file passed.
- Existing reader tests cover exact TopK results, physical row IDs,
localized slot mapping, missing statistics, metadata-pruning boundaries,
ordered row-ID fetches, NULL placement, and equality boundaries.
- FE coverage checks that potentially throwing secondary expressions do
not disable existing TopN runtime-filter generation. The existing
`nereids_tpch_p0/tpch/topn-filter.groovy` also exposed an actual
regression: the blanket visitor stop removed `TOPN OPT` for `GROUP BY k
ORDER BY k LIMIT 1`. Restore the original visitor traversal and cover
grouped keys, aggregation with HAVING and a join, and grouping sets with
`push_topn_to_agg=false`. These cases run full FE planning and check
`TOPN OPT:` in the translated EXPLAIN fragments. The same unnecessary
expression whitelist also disabled `TOPN OPT` for a nested STRUCT order
key with an `IS NOT NULL` predicate on a sibling field, as exposed by
`nereids_rules_p0/column_pruning/topn_filter_nested_column_pruning.groovy`.
Additional translated-plan checks cover TIMESTAMPTZ, INT and DATE fields
in ASC/DESC, including the complete and predicate-only nested access
paths. Both existing regression suites and their assertions are
unchanged. FE production logic matches the pre-PR base; only an
explanatory traversal comment remains in the FE production diff.
- The earlier SQL suite asserting that all excluded-row errors must be
preserved is removed. Complete SQL performance is not claimed;
performance evidence comes from the reader microbenchmark below.
Correctness suites:
```bash
./run-be-ut.sh -j8 --run \
--filter='*Parquet*:*ColumnMapper*:*TableReader*:*Iceberg*:*EvolutionCast*:*RuntimePredicate*'
./run-fe-ut.sh --run \
org.apache.doris.nereids.postprocess.TopNRuntimeFilterTest,org.apache.doris.nereids.rules.rewrite.DateArithmeticFilterPushDownTest,org.apache.doris.nereids.postprocess.PushDownFilterThroughProjectTest,org.apache.doris.nereids.util.ExpressionUtilsTest
```
Opt-in benchmark (use a RELEASE UT build):
```bash
DORIS_TOPN_ROW_GROUP_BENCHMARK=1 BUILD_TYPE_UT=RELEASE \
./run-be-ut.sh -j8 --run
--filter=ParquetScanTest.TopNRowGroupOrderMicrobench
```
For paired comparison, build and save baseline/candidate UT executables
with the same harness, then use the existing runner:
```bash
DORIS_TOPN_ROW_GROUP_BENCHMARK=1 \
DORIS_BE_UT_EXECUTABLE=build-support/run-query-engine-paired-benchmarks.py \
QUERY_ENGINE_BENCH_BASELINE=./baseline/doris_be_test \
QUERY_ENGINE_BENCH_CANDIDATE=./candidate/doris_be_test \
QUERY_ENGINE_BENCH_CPU=0 \
QUERY_ENGINE_BENCH_OUTPUT=./paired-results \
BUILD_TYPE_UT=RELEASE ./run-be-ut.sh -j8 --run \
--filter=ParquetScanTest.TopNRowGroupOrderMicrobench
```
Select an available idle CPU for `QUERY_ENGINE_BENCH_CPU`. The runner
emits paired timings/checksums; `PARQUET_TOPN_WORK` lines report
per-query raw rows, groups filtered by min/max, and actual local read
bytes.
### Release note
Improve Parquet V2 TopN scans by reading promising row groups first and
applying the latest runtime bound before opening subsequent groups.
### Check List (For Author)
- Test
- [x] Unit Test
- [x] Manual test (reader microbenchmark and commands above)
- Behavior changed:
- [x] Yes. Eligible Parquet V2 TopN scans may read row groups in a
different order and skip more groups. SQL ordering remains defined by
the TopN operator. Expressions and conversions in excluded groups may
not execute, so an exception from an excluded row may no longer be
observed.
- Does this need documentation?
- [x] No. Internal scan optimization with no new user-facing option.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
be/src/exprs/vtopn_pred.cpp | 36 +++
be/src/exprs/vtopn_pred.h | 6 +
be/src/format_v2/parquet/parquet_scan.cpp | 138 ++++++++-
be/src/format_v2/parquet/parquet_scan.h | 8 +
be/test/format_v2/parquet/parquet_scan_test.cpp | 312 ++++++++++++++++++++-
be/test/format_v2/table/iceberg_reader_test.cpp | 269 ++++++++++++++++++
.../processor/post/TopnFilterPushDownVisitor.java | 2 +
.../nereids/postprocess/TopNRuntimeFilterTest.java | 101 +++++++
8 files changed, 867 insertions(+), 5 deletions(-)
diff --git a/be/src/exprs/vtopn_pred.cpp b/be/src/exprs/vtopn_pred.cpp
index ab7ac572b85..0ef432cde95 100644
--- a/be/src/exprs/vtopn_pred.cpp
+++ b/be/src/exprs/vtopn_pred.cpp
@@ -97,6 +97,42 @@ bool topn_string_matches(int comparison, RawComparisonOp op)
{
} // namespace
+bool VTopNPred::can_evaluate_zonemap_filter() const {
+ // Capability must remain available before the first bound; readers cache
it during open.
+ return _predicate != nullptr && _children.size() == 1 &&
_children[0]->is_slot_ref();
+}
+
+ZoneMapFilterResult VTopNPred::evaluate_zonemap_filter(const
ZoneMapEvalContext& ctx) const {
+ if (!can_evaluate_zonemap_filter()) {
+ return unsupported_zonemap_filter(ctx);
+ }
+ const Field bound = _predicate->get_value();
+ if (bound.is_null()) {
+ return ZoneMapFilterResult::kMayMatch;
+ }
+ const auto* slot = assert_cast<const VSlotRef*>(_children[0].get());
+ const auto zone = ctx.zone_map(slot->column_id());
+ const auto type = ctx.data_type(slot->column_id());
+ if (zone == nullptr || !expr_zonemap::data_types_compatible(type,
slot->data_type())) {
+ return unsupported_zonemap_filter(ctx);
+ }
+ if (_predicate->nulls_first() && zone->has_null) {
+ return ZoneMapFilterResult::kMayMatch;
+ }
+ if (!zone->has_not_null) {
+ return ZoneMapFilterResult::kNoMatch;
+ }
+ if (ctx.floating_nan_count_unknown(slot->column_id()) || bound.is_nan() ||
+ !expr_zonemap::range_stats_usable_for_zonemap(*zone, type) ||
+ !expr_zonemap::field_types_compatible(bound.get_type(),
+
remove_nullable(type)->get_primitive_type())) {
+ return unsupported_zonemap_filter(ctx);
+ }
+ // Keep equality: with ORDER BY a, b a row tied on a can still improve b.
+ const bool excluded = _predicate->is_asc() ? bound < zone->min_value :
zone->max_value < bound;
+ return excluded ? ZoneMapFilterResult::kNoMatch :
ZoneMapFilterResult::kMayMatch;
+}
+
bool VTopNPred::can_execute_on_raw_fixed_values(const DataTypePtr& data_type,
int column_id) const {
if (_predicate == nullptr || data_type == nullptr || _children.size() !=
1) {
return false;
diff --git a/be/src/exprs/vtopn_pred.h b/be/src/exprs/vtopn_pred.h
index af680955d05..612e411427f 100644
--- a/be/src/exprs/vtopn_pred.h
+++ b/be/src/exprs/vtopn_pred.h
@@ -63,6 +63,12 @@ public:
}
int source_node_id() const { return _source_node_id; }
+ bool is_asc() const { return _predicate->is_asc(); }
+ bool nulls_first() const { return _predicate->nulls_first(); }
+
+ bool can_evaluate_zonemap_filter() const override;
+ ZoneMapFilterResult evaluate_zonemap_filter(const ZoneMapEvalContext& ctx)
const override;
+
Status clone_node(VExprSPtr* cloned_expr) const override {
DORIS_CHECK(cloned_expr != nullptr);
*cloned_expr = VTopNPred::create_shared(clone_texpr_node(),
_source_node_id, nullptr);
diff --git a/be/src/format_v2/parquet/parquet_scan.cpp
b/be/src/format_v2/parquet/parquet_scan.cpp
index e24c3bc5812..4ecea4ab08e 100644
--- a/be/src/format_v2/parquet/parquet_scan.cpp
+++ b/be/src/format_v2/parquet/parquet_scan.cpp
@@ -20,6 +20,7 @@
#include <iterator>
#include <limits>
#include <memory>
+#include <numeric>
#include <optional>
#include <ranges>
#include <set>
@@ -43,6 +44,7 @@
#include "exprs/vcompound_pred.h"
#include "exprs/vectorized_fn_call.h"
#include "exprs/vexpr_context.h"
+#include "exprs/vtopn_pred.h"
#include "format_v2/parquet/parquet_column_schema.h"
#include "format_v2/parquet/parquet_file_context.h"
#include "format_v2/parquet/parquet_statistics.h"
@@ -1230,6 +1232,9 @@ void
ParquetScanScheduler::set_condition_cache_context(std::shared_ptr<Condition
void ParquetScanScheduler::reset() {
_next_row_group_plan_idx = 0;
+ _row_group_order.clear();
+ _row_group_order_prepared = false;
+ _topn_footer_recheck = false;
_raw_rows_read = 0;
_predicate_schedule_request = nullptr;
_predicate_schedule = {};
@@ -1409,6 +1414,125 @@ 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;
+ });
+ // Scheduling need not evaluate conversions in groups excluded by TopN.
Reuse the request's
+ // existing metadata-pruning boundary; row-ID fetches still require
increasing file positions.
+ if (!_topn_footer_recheck || request.row_ids.has_value() ||
+ safe_count != request.conjuncts.size() ||
_scan_plan->row_groups.size() < 2) {
+ return;
+ }
+ struct Priority {
+ std::optional<Field> best;
+ bool may_have_null = false;
+ };
+ std::vector<Priority> priorities;
+ priorities.reserve(_scan_plan->row_groups.size());
+ const bool trusted_nulls =
+ native::parquet_reader_compat(metadata.__isset.created_by ?
metadata.created_by : "")
+ .null_count_trusted;
+ bool usable = false;
+ for (const auto& plan : _scan_plan->row_groups) {
+ 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];
+ Priority priority;
+ if (chunk.__isset.meta_data && chunk.meta_data.__isset.statistics) {
+ const auto safe_statistics =
detail::sanitize_native_footer_statistics(
+ schema.type_descriptor, chunk.meta_data.statistics,
+ detail::has_supported_type_defined_order(metadata,
schema.leaf_column_id));
+ const auto stats =
ParquetStatisticsUtils::TransformColumnStatistics(
+ schema, &safe_statistics, chunk.meta_data.num_values,
_timezone, trusted_nulls);
+ if (stats.has_min_max) {
+ priority.best = topn->is_asc() ? stats.min_value :
stats.max_value;
+ usable = true;
+ }
+ priority.may_have_null = stats.has_null;
+ }
+ priorities.push_back(std::move(priority));
+ }
+ if (!usable) {
+ return;
+ }
+ _row_group_order.resize(priorities.size());
+ std::iota(_row_group_order.begin(), _row_group_order.end(), 0);
+ std::stable_sort(_row_group_order.begin(), _row_group_order.end(),
[&](size_t lhs, size_t rhs) {
+ const auto& left = priorities[lhs];
+ const auto& right = priorities[rhs];
+ if (topn->nulls_first() && left.may_have_null != right.may_have_null) {
+ return left.may_have_null;
+ }
+ if (left.best.has_value() != right.best.has_value()) {
+ return left.best.has_value();
+ }
+ if (!left.best.has_value()) {
+ return false;
+ }
+ return topn->is_asc() ? *left.best < *right.best : *right.best <
*left.best;
+ });
+}
+
Status ParquetScanScheduler::open_next_row_group(
ParquetFileContext& file_context,
const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
@@ -1416,6 +1540,9 @@ Status ParquetScanScheduler::open_next_row_group(
*has_row_group = false;
RowGroupReadPlan* selected_plan = nullptr;
DORIS_CHECK(_scan_plan != nullptr);
+ if (!_row_group_order_prepared) {
+ prepare_topn_row_group_order(file_context, file_schema, request);
+ }
if (_requested_leaf_ids_need_refresh) {
// A runtime-filter refresh changes the immutable projection baseline
exactly once; later
// row groups only apply their full-Variant fallback delta to this
rebuilt set.
@@ -1423,7 +1550,11 @@ Status ParquetScanScheduler::open_next_row_group(
_requested_leaf_ids_need_refresh = false;
}
while (_next_row_group_plan_idx < _scan_plan->row_groups.size()) {
- RowGroupReadPlan& candidate_plan =
_scan_plan->row_groups[_next_row_group_plan_idx++];
+ const size_t plan_idx = _row_group_order.empty()
+ ? _next_row_group_plan_idx
+ :
_row_group_order[_next_row_group_plan_idx];
+ ++_next_row_group_plan_idx;
+ RowGroupReadPlan& candidate_plan = _scan_plan->row_groups[plan_idx];
// Probe only the row group about to execute. This keeps
LIMIT/cancellation latency
// independent of the number of later remote row groups while
preserving eager footer
// statistics pruning during open.
@@ -1440,6 +1571,11 @@ Status ParquetScanScheduler::open_next_row_group(
const auto& row_group = file_context.native_metadata->to_thrift()
.row_groups[candidate_plan.row_group_id];
prepare_row_group_physical_projection(row_group, file_schema,
request, &candidate_plan);
+ }
+ // TopN changes its bound in place, without publishing a new
FileScanRequest. For example,
+ // reading [0, 99] first can tighten ASC LIMIT 3 to 2 and exclude a
later group [100, 199].
+ // Recheck before dictionary/index probes or prefetch, using the
current monotonic bound.
+ if (_remaining_plans_need_replanning || _topn_footer_recheck) {
const auto row_group_leaf_ids = row_group_leaf_column_ids(
file_schema, request, candidate_plan,
_scan_plan->requested_leaf_column_ids);
const auto& requested_leaf_ids = row_group_leaf_ids.has_value()
diff --git a/be/src/format_v2/parquet/parquet_scan.h
b/be/src/format_v2/parquet/parquet_scan.h
index b8c2d80c057..a4220304286 100644
--- a/be/src/format_v2/parquet/parquet_scan.h
+++ b/be/src/format_v2/parquet/parquet_scan.h
@@ -239,6 +239,10 @@ private:
static constexpr size_t PROFILE_FLUSH_BATCH_INTERVAL = 16;
void reset_current_row_group();
+ void prepare_topn_row_group_order(
+ const ParquetFileContext& file_context,
+ const std::vector<std::unique_ptr<ParquetColumnSchema>>&
file_schema,
+ const format::FileScanRequest& request);
void activate_pending_scan_request_at_row_group_boundary();
void flush_current_reader_profiles();
bool finish_current_reader_batch_profiles();
@@ -287,6 +291,10 @@ private:
std::shared_ptr<RowGroupScanPlan> _scan_plan; // shared with the reader's
aggregate path
size_t _next_row_group_plan_idx = 0; // index of the next row
group to process
+ // Execution order is separate from physical plans: Condition Cache uses
their file-row order.
+ std::vector<size_t> _row_group_order;
+ bool _row_group_order_prepared = false;
+ bool _topn_footer_recheck = false;
bool _has_current_row_group = false;
// Readers retain pointers into this immutable row-group map, so it must
outlive both maps below.
diff --git a/be/test/format_v2/parquet/parquet_scan_test.cpp
b/be/test/format_v2/parquet/parquet_scan_test.cpp
index c8ac9506387..af4a6fe0633 100644
--- a/be/test/format_v2/parquet/parquet_scan_test.cpp
+++ b/be/test/format_v2/parquet/parquet_scan_test.cpp
@@ -24,6 +24,7 @@
#include <parquet/api/writer.h>
#include <parquet/arrow/writer.h>
#include <parquet/encoding.h>
+#include <unistd.h>
#include <atomic>
#include <bit>
@@ -36,6 +37,7 @@
#include <memory>
#include <numeric>
#include <optional>
+#include <queue>
#include <set>
#include <string>
#include <thread>
@@ -43,6 +45,7 @@
#include <vector>
#include "common/config.h"
+#include "common/metrics/doris_metrics.h"
#include "core/assert_cast.h"
#include "core/block/block.h"
#include "core/column/column_array.h"
@@ -85,6 +88,7 @@
#include "storage/index/zone_map/zonemap_eval_context.h"
#include "storage/index/zone_map/zonemap_filter_result.h"
#include "storage/utils.h"
+#include "testutil/benchmark_control.h"
#include "testutil/mock/mock_query_context.h"
#include "util/coding.h"
#include "util/defer_op.h"
@@ -961,7 +965,8 @@ struct PreparedTopNConjunct {
PreparedTopNConjunct create_topn_conjunct(RuntimeState* state, int column_id,
const DataTypePtr& data_type, const
Field& bound,
- bool publish_bound = true) {
+ bool publish_bound = true, bool
is_asc = true,
+ bool nulls_first = false) {
DORIS_CHECK(state != nullptr);
DORIS_CHECK(data_type != nullptr);
TExpr target_expr;
@@ -977,8 +982,8 @@ PreparedTopNConjunct create_topn_conjunct(RuntimeState*
state, int column_id,
TTopnFilterDesc desc;
desc.__set_source_node_id(10);
- desc.__set_is_asc(true);
- desc.__set_null_first(false);
+ desc.__set_is_asc(is_asc);
+ desc.__set_null_first(nulls_first);
desc.__set_target_node_id_to_target_expr({{20, std::move(target_expr)}});
auto query_context = MockQueryContext::create();
state->_query_ctx = query_context.get();
@@ -1927,7 +1932,9 @@ void use_schema_order_positions(format::FileScanRequest*
request,
class ParquetScanTest : public testing::Test {
protected:
void SetUp() override {
- _test_dir = std::filesystem::temp_directory_path() /
"doris_format_v2_parquet_scan_test";
+ // Paired baseline/candidate benchmarks must not remove the other
process's fixtures.
+ _test_dir = std::filesystem::temp_directory_path() /
+ ("doris_format_v2_parquet_scan_test_" +
std::to_string(getpid()));
std::filesystem::remove_all(_test_dir);
std::filesystem::create_directories(_test_dir);
_file_path = (_test_dir / "scan.parquet").string();
@@ -4790,5 +4797,302 @@ TEST_F(ParquetScanTest,
OpenDefersPageIndexProbeToCurrentRowGroup) {
EXPECT_EQ(counter_value(profile, "PageIndexReadCalls"), 1);
}
+TEST_F(ParquetScanTest, TopNReadsBestRowGroupBeforePublishingBound) {
+ constexpr int GROUP_ROWS = 128;
+ constexpr int GROUPS = 4;
+ constexpr int K = 3;
+ for (const bool ascending : {true, false}) {
+ _file_path = (_test_dir / (ascending ? "asc.parquet" :
"desc.parquet")).string();
+ std::vector<int32_t> values;
+ for (int group = 0; group < GROUPS; ++group) {
+ const int base = (ascending ? GROUPS - group - 1 : group) *
GROUP_ROWS;
+ for (int row = 0; row < GROUP_ROWS; ++row) {
+ values.push_back(base + row);
+ }
+ }
+ auto table = arrow::Table::Make(arrow::schema({arrow::field("k",
arrow::int32(), false)}),
+ {build_int32_array(values)});
+ // PLAIN, no dictionary or PageIndex: only Footer pruning can avoid
reading later groups.
+ write_table(_file_path, table, GROUP_ROWS, false, false, true);
+ RuntimeProfile profile("topn_row_group_order");
+ const format::GlobalRowIdContext rowid_context {
+ .version = 1, .backend_id = 2, .file_id = 3};
+ auto reader = create_reader(0, -1, &profile, rowid_context);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+ reader->set_batch_size(GROUP_ROWS);
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto prepared = create_topn_conjunct(&state, 0, schema[0].type, Field
{}, false, ascending);
+ auto request = std::make_shared<format::FileScanRequest>();
+ format::FileScanRequestBuilder builder(request.get());
+
ASSERT_TRUE(builder.add_predicate_column(format::LocalColumnId(0)).ok());
+ ASSERT_TRUE(builder.add_non_predicate_column(
+
format::LocalColumnId(format::GLOBAL_ROWID_COLUMN_ID))
+ .ok());
+ request->conjuncts.push_back(prepared.conjunct);
+ ASSERT_TRUE(reader->open(request).ok());
+ std::vector<int32_t> top;
+ bool eof = false;
+ while (!eof) {
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ const auto& column =
int32_data_column(*block.get_by_position(0).column);
+ for (size_t row = 0; row < rows; ++row) {
+ const int32_t value = column.get_element(row);
+ top.push_back(value);
+ const auto location =
+
decode_rowid(string_data_column(*block.get_by_position(1).column), row);
+ const int expected_position =
+ ascending ? (GROUPS - 1 - value / GROUP_ROWS) *
GROUP_ROWS +
+ value % GROUP_ROWS
+ : value;
+ EXPECT_EQ(location.row_id, expected_position);
+ EXPECT_EQ(location.file_id, rowid_context.file_id);
+ }
+ std::ranges::sort(
+ top, [ascending](int32_t a, int32_t b) { return ascending
? a < b : a > b; });
+ if (top.size() >= K) {
+ top.resize(K);
+ // Model the SortSink feedback after each block without
replacing the scan request.
+ ASSERT_TRUE(prepared.query_context->get_runtime_predicate(10)
+
.update(Field::create_field<TYPE_INT>(top.back()))
+ .ok());
+ }
+ }
+ const std::vector<int32_t> expected =
+ ascending ? std::vector<int32_t> {0, 1, 2} :
std::vector<int32_t> {511, 510, 509};
+ EXPECT_EQ(top, expected);
+ EXPECT_EQ(counter_value(profile, "RawRowsRead"), GROUP_ROWS);
+ EXPECT_EQ(counter_value(profile, "RowGroupsFilteredByMinMax"), GROUPS
- 1);
+ prepared.conjunct->close();
+ }
+}
+
+TEST_F(ParquetScanTest,
TopNOrderUsesLocalizedSlotInsteadOfPhysicalColumnOrdinal) {
+ auto table = arrow::Table::Make(
+ arrow::schema({arrow::field("padding", arrow::int32(), false),
+ arrow::field("k", arrow::int32(), false)}),
+ {build_int32_array({-1, -1, -1, -1, -1, -1}),
build_int32_array({8, 9, 4, 5, 0, 1})});
+ write_table(_file_path, table, 2, false, false, true);
+ RuntimeProfile profile("localized_topn");
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ auto reader = create_reader(0, -1, &profile);
+ ASSERT_TRUE(reader->init(&state).ok());
+ reader->set_batch_size(2);
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ // The physical key is column 1, but the mapper binds it to slot 0 in the
file-local block.
+ auto prepared = create_topn_conjunct(&state, 0, schema[1].type, Field {},
false);
+ auto request = std::make_shared<format::FileScanRequest>();
+ format::FileScanRequestBuilder builder(request.get());
+ ASSERT_TRUE(builder.add_predicate_column(format::LocalColumnId(1)).ok());
+ request->conjuncts.push_back(prepared.conjunct);
+ ASSERT_TRUE(reader->open(request).ok());
+ Block block = build_file_block({schema[1]});
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ ASSERT_EQ(rows, 2);
+ EXPECT_EQ(int32_data_column(*block.get_by_position(0).column).get_data(),
+ (ColumnInt32::Container {0, 1}));
+ ASSERT_TRUE(prepared.query_context->get_runtime_predicate(10)
+ .update(Field::create_field<TYPE_INT>(1))
+ .ok());
+ block.clear_column_data();
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ EXPECT_TRUE(eof);
+ EXPECT_EQ(rows, 0);
+ EXPECT_EQ(counter_value(profile, "RawRowsRead"), 2);
+ EXPECT_EQ(counter_value(profile, "RowGroupsFilteredByMinMax"), 2);
+ prepared.conjunct->close();
+}
+
+TEST_F(ParquetScanTest, TopNMissingStatisticsAndPruningFenceKeepPhysicalOrder)
{
+ for (const int scenario : {0, 1, 2}) {
+ auto table = arrow::Table::Make(arrow::schema({arrow::field("k",
arrow::int32(), false)}),
+ {build_int32_array({8, 9, 4, 5, 0,
1})});
+ write_table(_file_path, table, 2, false, false, scenario != 0);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ auto reader = create_reader();
+ ASSERT_TRUE(reader->init(&state).ok());
+ reader->set_batch_size(2);
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto prepared = create_topn_conjunct(&state, 0, schema[0].type, Field
{}, false);
+ auto request = std::make_shared<format::FileScanRequest>();
+ format::FileScanRequestBuilder builder(request.get());
+
ASSERT_TRUE(builder.add_predicate_column(format::LocalColumnId(0)).ok());
+ request->conjuncts.push_back(prepared.conjunct);
+ if (scenario == 1) {
+ // A preceding evaluation-sensitive conjunct prevents metadata
pruning and reordering.
+ request->metadata_pruning_safe_conjunct_count = 0;
+ } else if (scenario == 2) {
+ request->row_ids = std::vector<int64_t> {0, 3, 4};
+ }
+ ASSERT_TRUE(reader->open(request).ok());
+ std::vector<int32_t> actual;
+ bool eof = false;
+ while (!eof) {
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ const auto& column =
int32_data_column(*block.get_by_position(0).column);
+ for (size_t row = 0; row < rows; ++row) {
+ actual.push_back(column.get_element(row));
+ }
+ }
+ EXPECT_EQ(actual, scenario == 2 ? (std::vector<int32_t> {8, 5, 0})
+ : (std::vector<int32_t> {8, 9, 4, 5,
0, 1}));
+ prepared.conjunct->close();
+ }
+}
+
+TEST_F(ParquetScanTest, TopNZoneMapObservesLateBoundsAndNullOrdering) {
+ for (const bool ascending : {true, false}) {
+ for (const bool nulls_first : {true, false}) {
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ auto type = make_nullable(std::make_shared<DataTypeInt32>());
+ auto prepared =
+ create_topn_conjunct(&state, 0, type, Field {}, false,
ascending, nulls_first);
+ auto zone = std::make_shared<segment_v2::ZoneMap>();
+ zone->has_not_null = true;
+ zone->min_value = Field::create_field<TYPE_INT>(10);
+ zone->max_value = Field::create_field<TYPE_INT>(20);
+ ZoneMapEvalContext ctx;
+ ctx.slots[0].data_type = type;
+ ctx.slots[0].zone_map = zone;
+ const auto& expression = prepared.conjunct->root();
+ EXPECT_TRUE(expression->can_evaluate_zonemap_filter());
+ EXPECT_NE(expression->evaluate_zonemap_filter(ctx),
ZoneMapFilterResult::kNoMatch);
+ ASSERT_TRUE(prepared.query_context->get_runtime_predicate(10)
+
.update(Field::create_field<TYPE_INT>(ascending ? 9 : 21))
+ .ok());
+ EXPECT_EQ(expression->evaluate_zonemap_filter(ctx),
ZoneMapFilterResult::kNoMatch);
+ // Equal leading keys may still win on a later ORDER BY key and
must remain candidates.
+ if (ascending) {
+ zone->min_value = Field::create_field<TYPE_INT>(9);
+ } else {
+ zone->max_value = Field::create_field<TYPE_INT>(21);
+ }
+ EXPECT_EQ(expression->evaluate_zonemap_filter(ctx),
ZoneMapFilterResult::kMayMatch);
+ zone->min_value = Field::create_field<TYPE_INT>(10);
+ zone->max_value = Field::create_field<TYPE_INT>(20);
+ zone->has_null = true;
+ EXPECT_EQ(expression->evaluate_zonemap_filter(ctx),
+ nulls_first ? ZoneMapFilterResult::kMayMatch :
ZoneMapFilterResult::kNoMatch);
+ zone->has_not_null = false;
+ EXPECT_EQ(expression->evaluate_zonemap_filter(ctx),
+ nulls_first ? ZoneMapFilterResult::kMayMatch :
ZoneMapFilterResult::kNoMatch);
+ ctx.slots.clear();
+ EXPECT_EQ(expression->evaluate_zonemap_filter(ctx),
ZoneMapFilterResult::kUnsupported);
+ prepared.conjunct->close();
+ }
+ }
+}
+
+TEST_F(ParquetScanTest, TopNRowGroupOrderMicrobench) {
+ if (std::getenv("DORIS_TOPN_ROW_GROUP_BENCHMARK") == nullptr) {
+ GTEST_SKIP() << "Set DORIS_TOPN_ROW_GROUP_BENCHMARK=1 to run the
reader microbenchmark";
+ }
+ constexpr int GROUPS = 128;
+ constexpr int GROUP_ROWS = 4096;
+ constexpr int K = 10;
+ constexpr int SAMPLES = 9;
+ constexpr int ITERATIONS = 32;
+ for (const bool ascending : {true, false}) {
+ _file_path = (_test_dir / (ascending ? "asc.parquet" :
"desc.parquet")).string();
+ std::vector<int32_t> values;
+ values.reserve(GROUPS * GROUP_ROWS);
+ for (int group = 0; group < GROUPS; ++group) {
+ const int base = (ascending ? GROUPS - group - 1 : group) *
GROUP_ROWS;
+ for (int row = 0; row < GROUP_ROWS; ++row) {
+ values.push_back(base + row);
+ }
+ }
+ auto table = arrow::Table::Make(arrow::schema({arrow::field("k",
arrow::int32(), false)}),
+ {build_int32_array(values)});
+ // Fixture generation and the warmup are outside the reported
reader-plus-TopK samples.
+ write_table(_file_path, table, GROUP_ROWS, false, false, true);
+ for (int sample = -1; sample < SAMPLES; ++sample) {
+ const std::string label = ascending ? "parquet_topn_asc" :
"parquet_topn_desc";
+ if (sample >= 0) {
+ benchmark::wait_for_turn(label, sample);
+ }
+ uint64_t total_elapsed = 0;
+ int64_t total_checksum = 0;
+ int64_t total_rows = 0;
+ int64_t total_filtered = 0;
+ int64_t total_bytes = 0;
+ for (int iteration = 0; iteration < ITERATIONS; ++iteration) {
+ const auto bytes_before =
DorisMetrics::instance()->local_bytes_read_total->value();
+ const auto start = std::chrono::steady_clock::now();
+ RuntimeProfile profile("topn_microbench");
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ auto reader = create_reader(0, -1, &profile);
+ ASSERT_TRUE(reader->init(&state).ok());
+ reader->set_batch_size(GROUP_ROWS);
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto prepared =
+ create_topn_conjunct(&state, 0, schema[0].type, Field
{}, false, ascending);
+ auto request = std::make_shared<format::FileScanRequest>();
+ format::FileScanRequestBuilder builder(request.get());
+
ASSERT_TRUE(builder.add_predicate_column(format::LocalColumnId(0)).ok());
+ request->conjuncts.push_back(prepared.conjunct);
+ ASSERT_TRUE(reader->open(request).ok());
+ auto compare = [ascending](int32_t a, int32_t b) {
+ return ascending ? a < b : a > b;
+ };
+ std::priority_queue<int32_t, std::vector<int32_t>,
decltype(compare)> top(compare);
+ bool eof = false;
+ while (!eof) {
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ const auto& column =
int32_data_column(*block.get_by_position(0).column);
+ for (size_t row = 0; row < rows; ++row) {
+ top.push(column.get_element(row));
+ if (top.size() > K) {
+ top.pop();
+ }
+ }
+ if (top.size() == K) {
+
ASSERT_TRUE(prepared.query_context->get_runtime_predicate(10)
+
.update(Field::create_field<TYPE_INT>(top.top()))
+ .ok());
+ }
+ }
+ ASSERT_TRUE(reader->close().ok());
+ const auto elapsed =
std::chrono::duration_cast<std::chrono::nanoseconds>(
+ std::chrono::steady_clock::now()
- start)
+ .count();
+ ASSERT_EQ(top.size(), K);
+ int64_t checksum = 0;
+ for (int rank = K - 1; rank >= 0; --rank) {
+ EXPECT_EQ(top.top(), ascending ? rank : GROUPS *
GROUP_ROWS - 1 - rank);
+ checksum += top.top();
+ top.pop();
+ }
+ total_elapsed += elapsed;
+ total_checksum += checksum;
+ total_rows += counter_value(profile, "RawRowsRead");
+ total_filtered += counter_value(profile,
"RowGroupsFilteredByMinMax");
+ total_bytes +=
+
DorisMetrics::instance()->local_bytes_read_total->value() - bytes_before;
+ prepared.conjunct->close();
+ }
+ if (sample >= 0) {
+ benchmark::report_sample(label, sample, ITERATIONS,
total_elapsed, total_checksum,
+ true);
+ std::cout << "PARQUET_TOPN_WORK," << label << ',' << sample <<
','
+ << total_rows / ITERATIONS << ',' << total_filtered
/ ITERATIONS << ','
+ << total_bytes / ITERATIONS << std::endl;
+ }
+ }
+ }
+}
+
} // namespace
} // namespace doris
diff --git a/be/test/format_v2/table/iceberg_reader_test.cpp
b/be/test/format_v2/table/iceberg_reader_test.cpp
index d905b8cb726..2651501c62e 100644
--- a/be/test/format_v2/table/iceberg_reader_test.cpp
+++ b/be/test/format_v2/table/iceberg_reader_test.cpp
@@ -72,6 +72,7 @@
#include "exprs/vexpr.h"
#include "exprs/vliteral.h"
#include "exprs/vslot_ref.h"
+#include "exprs/vtopn_pred.h"
#include "format/format_common.h"
#include "format/orc/orc_memory_stream_test.h"
#include "format/table/deletion_vector_reader.h"
@@ -85,7 +86,9 @@
#include "runtime/runtime_profile.h"
#include "runtime/runtime_state.h"
#include "storage/segment/condition_cache.h"
+#include "testutil/mock/mock_query_context.h"
#include "util/debug_points.h"
+#include "util/defer_op.h"
#include "util/hash_util.hpp"
#include "util/url_coding.h"
@@ -1726,6 +1729,272 @@ void apply_final_conjuncts(Block* block, const
VExprContextSPtrs& conjuncts) {
ASSERT_TRUE(status.ok()) << status;
}
+struct EvolutionTopNFilter {
+ std::shared_ptr<MockQueryContext> query_context;
+ VExprContextSPtr conjunct;
+};
+
+EvolutionTopNFilter make_evolution_topn_filter(RuntimeState* state, bool
ascending) {
+ const auto type = make_nullable(std::make_shared<DataTypeInt32>());
+ TExpr target;
+ TExprNode slot;
+ slot.__set_node_type(TExprNodeType::SLOT_REF);
+ slot.__set_type(type->to_thrift());
+ slot.__set_num_children(0);
+ TSlotRef slot_ref;
+ slot_ref.__set_slot_id(0);
+ slot_ref.__set_tuple_id(0);
+ slot.__set_slot_ref(slot_ref);
+ target.nodes.push_back(slot);
+ TTopnFilterDesc desc;
+ desc.__set_source_node_id(10);
+ desc.__set_is_asc(ascending);
+ desc.__set_null_first(false);
+ desc.__set_target_node_id_to_target_expr({{20, target}});
+ auto query = MockQueryContext::create();
+ state->_query_ctx = query.get();
+ query->init_runtime_predicates({desc});
+ auto& predicate = query->get_runtime_predicate(10);
+ predicate.set_detected_source();
+ DORIS_CHECK(predicate.init_target(20, {}, -1).ok());
+ TExprNode topn_node;
+ topn_node.__set_type(std::make_shared<DataTypeUInt8>()->to_thrift());
+ topn_node.__set_is_nullable(true);
+ auto topn = VTopNPred::create_shared(topn_node, 10, nullptr);
+ topn->add_child(VSlotRef::create_shared(0, 0, -1, type, "k"));
+ return {.query_context = std::move(query), .conjunct =
prepared_conjunct(state, topn)};
+}
+
+void write_topn_evolution_file(const std::string& path, bool ascending, bool
nested,
+ int32_t bad_date = -354285) {
+ constexpr int GROUP_ROWS = 4096;
+ arrow::Int32Builder keys;
+ arrow::Date32Builder dates;
+ for (int row = 0; row < GROUP_ROWS * 4; ++row) {
+ const bool bad = row < GROUP_ROWS;
+ const int key = (bad == ascending) ? 100 : 0;
+ DORIS_CHECK(keys.Append(key).ok());
+ DORIS_CHECK(dates.Append(bad ? bad_date : 19724).ok());
+ }
+ auto key_field = arrow::field("k", arrow::int32(), true,
+
arrow::key_value_metadata({"PARQUET:field_id"}, {"1"}));
+ auto date_field = arrow::field("ts", arrow::date32(), true,
+
arrow::key_value_metadata({"PARQUET:field_id"}, {"2"}));
+ std::shared_ptr<arrow::Array> date_values = dates.Finish().ValueOrDie();
+ if (nested) {
+ date_values = arrow::StructArray::Make({date_values},
{date_field}).ValueOrDie();
+ date_field = arrow::field("outer", date_values->type(), true,
+
arrow::key_value_metadata({"PARQUET:field_id"}, {"3"}));
+ }
+ auto table = arrow::Table::Make(arrow::schema({key_field, date_field}),
+ {keys.Finish().ValueOrDie(), date_values});
+ auto out = arrow::io::FileOutputStream::Open(path).ValueOrDie();
+ ::parquet::WriterProperties::Builder properties;
+ properties.disable_dictionary();
+ properties.data_page_version(::parquet::ParquetDataPageVersion::V2);
+ DORIS_CHECK(::parquet::arrow::WriteTable(*table,
arrow::default_memory_pool(), out, GROUP_ROWS,
+ properties.build())
+ .ok());
+ DORIS_CHECK(out->Close().ok());
+}
+
+std::vector<ColumnDefinition> topn_evolution_columns(bool promote, bool
nested) {
+ DataTypePtr date_type = promote ?
DataTypePtr(std::make_shared<DataTypeTimeStampNs>())
+ :
DataTypePtr(std::make_shared<DataTypeDateV2>());
+ auto date = make_table_column(2, "ts", date_type);
+ if (nested) {
+ auto outer = make_table_column(
+ 3, "outer",
+ std::make_shared<DataTypeStruct>(DataTypes {date.type},
Strings {"ts"}));
+ outer.children = {date};
+ date = outer;
+ }
+ return {make_table_column(1, "k", std::make_shared<DataTypeInt32>()),
date};
+}
+
+TEST(IcebergV2ReaderTest, TopNPrunesRowsBeforeUnneededEvolutionConversion) {
+ const auto dir = std::filesystem::temp_directory_path() /
"topn_evolution_projection";
+ std::filesystem::create_directories(dir);
+ Defer cleanup([&] { std::filesystem::remove_all(dir); });
+ for (const bool ascending : {false, true}) {
+ for (const bool nested : {false, true}) {
+ const auto path = (dir / (std::string(ascending ? "asc" : "desc") +
+ (nested ? "_nested.parquet" :
".parquet")))
+ .string();
+ // RG0 cannot beat any later group on k. Its DATE -> TIMESTAMP_NS
conversion is not
+ // needed for TopN, while the same conversion must still fail in
an unfiltered scan.
+ write_topn_evolution_file(path, ascending, nested);
+ for (const bool promote : {false, true}) {
+ for (const bool strict : {false, true}) {
+ SCOPED_TRACE(testing::Message()
+ << "ascending=" << ascending << " nested=" <<
nested
+ << " promote=" << promote << " strict=" <<
strict);
+ TQueryOptions options;
+ options.__set_enable_strict_cast(strict);
+ options.__set_batch_size(4096);
+ RuntimeState state {options, TQueryGlobals()};
+ auto filter = make_evolution_topn_filter(&state,
ascending);
+ const auto columns = topn_evolution_columns(promote,
nested);
+ auto scan_params = make_local_parquet_scan_params();
+ RuntimeProfile profile("topn_evolution");
+ iceberg::IcebergTableReader reader;
+ ASSERT_TRUE(reader.init({.projected_columns = columns,
+ .conjuncts = {filter.conjunct},
+ .format = FileFormat::PARQUET,
+ .scan_params = &scan_params,
+ .io_ctx = nullptr,
+ .runtime_state = &state,
+ .scanner_profile = &profile})
+ .ok());
+
ASSERT_TRUE(reader.prepare_split(build_split_options(path)).ok());
+ bool eos = false;
+ size_t rows = 0;
+ Status status;
+ while (!eos) {
+ Block block = build_table_block(columns);
+ status = reader.get_block(&block, &eos);
+ if (!status.ok()) {
+ break;
+ }
+ rows += block.rows();
+ if (block.rows() > 0) {
+ const auto& values = assert_cast<const
ColumnNullable&>(
+
*block.get_by_position(0).column)
+ .get_nested_column();
+ EXPECT_EQ(assert_cast<const
ColumnInt32&>(values).get_element(0),
+ ascending ? 0 : 100);
+ const IColumn* dates =
block.get_by_position(1).column.get();
+ if (nested) {
+ dates = &assert_cast<const ColumnStruct&>(
+
expect_not_null_table_column(block, 1))
+ .get_column(0);
+ }
+ const auto type =
+ promote ?
DataTypePtr(std::make_shared<DataTypeTimeStampNs>())
+ :
DataTypePtr(std::make_shared<DataTypeDateV2>());
+ const auto& materialized =
+
expect_not_null_nullable_nested_column(*dates);
+ for (size_t row = 0; row < block.rows(); ++row) {
+ EXPECT_EQ(type->to_string(materialized, row),
+ promote ? "2024-01-02
00:00:00.000000000" : "2024-01-02");
+ }
+
ASSERT_TRUE(filter.query_context->get_runtime_predicate(10)
+
.update(Field::create_field<TYPE_INT>(
+ ascending ? 0 : 100))
+ .ok());
+ }
+ }
+ EXPECT_TRUE(status.ok()) << status;
+ EXPECT_EQ(rows, 3 * 4096);
+ ASSERT_TRUE(reader.close().ok());
+ filter.conjunct->close();
+ }
+ }
+ }
+ }
+}
+
+TEST(IcebergV2ReaderTest,
TopNPruningSkipsNativeAndHiddenDeleteConversionsOnlyForExcludedRows) {
+ const auto dir = std::filesystem::temp_directory_path() /
"topn_conversion_pruning";
+ std::filesystem::create_directories(dir);
+ Defer cleanup([&] { std::filesystem::remove_all(dir); });
+ for (const bool hidden_delete : {false, true}) {
+ for (const bool ascending : {false, true}) {
+ const auto suffix = std::string(hidden_delete ? "delete_" :
"native_") +
+ (ascending ? "asc" : "desc");
+ const auto path = (dir / (suffix + ".parquet")).string();
+ // DATE's INT32 domain includes year 10000, which Doris DATE
cannot materialize.
+ // Year 1000 is valid as DATE but overflows the hidden equality
key's TIMESTAMP_NS.
+ write_topn_evolution_file(path, ascending, false, hidden_delete ?
-354285 : 2932897);
+ const auto delete_path = (dir / (suffix +
"_delete.parquet")).string();
+ if (hidden_delete) {
+ const auto type = arrow::timestamp(arrow::TimeUnit::NANO);
+ arrow::TimestampBuilder builder(type,
arrow::default_memory_pool());
+ ASSERT_TRUE(builder.Append(1'704'067'200'000'000'000L).ok());
+ const auto field = arrow::field(
+ "ts", type, true,
arrow::key_value_metadata({"PARQUET:field_id"}, {"2"}));
+ const auto table =
+ arrow::Table::Make(arrow::schema({field}),
{builder.Finish().ValueOrDie()});
+ auto out =
arrow::io::FileOutputStream::Open(delete_path).ValueOrDie();
+ ASSERT_TRUE(
+ ::parquet::arrow::WriteTable(*table,
arrow::default_memory_pool(), out, 1)
+ .ok());
+ ASSERT_TRUE(out->Close().ok());
+ }
+ for (const int scenario : {0, 1, 2}) {
+ SCOPED_TRACE(testing::Message()
+ << "hidden_delete=" << hidden_delete << "
ascending=" << ascending
+ << " scenario=" << scenario);
+ TQueryOptions options;
+ options.__set_enable_strict_cast(true);
+ options.__set_batch_size(4096);
+ RuntimeState state {options, TQueryGlobals()};
+ // No filter and a reversed TopN both need RG0; only scenario
1 excludes it.
+ auto filter =
+ make_evolution_topn_filter(&state, scenario == 2 ?
!ascending : ascending);
+ auto columns = topn_evolution_columns(false, false);
+ if (hidden_delete) {
+ columns.resize(1);
+ }
+ auto scan_params = make_local_parquet_scan_params();
+
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
+ RuntimeProfile profile("topn_conversion");
+ ShardedKVCache cache(1);
+ iceberg::IcebergTableReader reader;
+ ASSERT_TRUE(reader.init({.projected_columns = columns,
+ .conjuncts = scenario == 0
+ ?
VExprContextSPtrs {}
+ :
VExprContextSPtrs {filter.conjunct},
+ .format = FileFormat::PARQUET,
+ .scan_params = &scan_params,
+ .io_ctx = nullptr,
+ .runtime_state = &state,
+ .scanner_profile = &profile})
+ .ok());
+ auto split = build_split_options(path);
+ split.cache = &cache;
+ if (hidden_delete) {
+
split.current_range.__set_table_format_params(make_iceberg_table_format_desc(
+ path,
{make_iceberg_equality_delete_file(delete_path, {2})}, 3));
+ }
+ ASSERT_TRUE(reader.prepare_split(split).ok());
+ bool eos = false;
+ size_t rows = 0;
+ Status status;
+ while (!eos) {
+ Block block = build_table_block(columns);
+ status = reader.get_block(&block, &eos);
+ if (!status.ok()) {
+ break;
+ }
+ rows += block.rows();
+ if (block.rows() > 0) {
+ expect_int32_column_values(
+ *block.get_by_position(0).column,
+ std::vector<int32_t>(block.rows(), ascending ?
0 : 100));
+ ASSERT_TRUE(
+ filter.query_context->get_runtime_predicate(10)
+
.update(Field::create_field<TYPE_INT>(ascending ? 0 : 100))
+ .ok());
+ }
+ }
+ EXPECT_EQ(status.ok(), scenario == 1) << status;
+ if (scenario == 1) {
+ EXPECT_EQ(rows, 3 * 4096);
+ } else {
+ EXPECT_NE(status.to_string().find(
+ hidden_delete ? "TIMESTAMP_NS overflow"
+ : "Parquet DATE value is
out of range"),
+ std::string::npos)
+ << status;
+ }
+ ASSERT_TRUE(reader.close().ok());
+ filter.conjunct->close();
+ }
+ }
+ }
+}
+
TEST(IcebergV2ReaderTest, DatePromotionToNanosecondsChecksOverflow) {
const auto test_dir = std::filesystem::temp_directory_path() /
"iceberg_date_promotion";
std::filesystem::create_directories(test_dir);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/TopnFilterPushDownVisitor.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/TopnFilterPushDownVisitor.java
index bd72b8aaaef..e90306455c6 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/TopnFilterPushDownVisitor.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/TopnFilterPushDownVisitor.java
@@ -82,6 +82,8 @@ public class TopnFilterPushDownVisitor extends
PlanVisitor<Boolean, PushDownCont
@Override
public Boolean visit(Plan plan, PushDownContext ctx) {
+ // A bound on a grouping key can exclude whole groups. Stopping at
unlisted operators
+ // here would disable existing scan filters below aggregates and their
join inputs.
boolean pushed = false;
for (Plan child : plan.children()) {
pushed |= child.accept(this, ctx);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/TopNRuntimeFilterTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/TopNRuntimeFilterTest.java
index 07fbaf84219..63654f74851 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/TopNRuntimeFilterTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/TopNRuntimeFilterTest.java
@@ -37,6 +37,7 @@ import
org.apache.doris.nereids.trees.plans.physical.TopnFilter;
import org.apache.doris.nereids.util.MemoPatternMatchSupported;
import org.apache.doris.nereids.util.PlanChecker;
import org.apache.doris.planner.ScanNode;
+import org.apache.doris.thrift.TExplainLevel;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -44,13 +45,20 @@ import org.mockito.Mockito;
import java.util.List;
import java.util.Map;
+import java.util.stream.Collectors;
public class TopNRuntimeFilterTest extends SSBTestBase implements
MemoPatternMatchSupported {
@Override
public void runBeforeAll() throws Exception {
super.runBeforeAll();
+ createTable("CREATE TABLE topn_safety (id INT, k INT, d DATE, ts
BIGINT) DUPLICATE KEY(id) "
+ + "DISTRIBUTED BY HASH(id) BUCKETS 1
PROPERTIES('replication_num'='1')");
createTable("CREATE TABLE uuid_topn (id INT, u UUID) DUPLICATE KEY(id)
"
+ "DISTRIBUTED BY HASH(id) BUCKETS 1
PROPERTIES('replication_num'='1')");
+ createTable("CREATE TABLE topn_nested_fields (id INT, "
+ + "ts STRUCT<a: TIMESTAMPTZ(0), b: TIMESTAMPTZ(0)>, "
+ + "si STRUCT<a: INT, b: INT>, sd STRUCT<a: DATE, b: DATE>)
DUPLICATE KEY(id) "
+ + "DISTRIBUTED BY HASH(id) BUCKETS 1
PROPERTIES('replication_num'='1')");
}
@Test
@@ -68,6 +76,99 @@ public class TopNRuntimeFilterTest extends SSBTestBase
implements MemoPatternMat
Assertions.assertTrue(checker.getCascadesContext().getTopnFilterContext().isTopnFilterSource(localTopN));
}
+ @Test
+ public void testTopNFilterDoesNotRequireTotalSecondaryExpressions() {
+ double previousRatio =
connectContext.getSessionVariable().topnFilterRatio;
+ connectContext.getSessionVariable().topnFilterRatio = 1000000;
+ try {
+ for (String expression : List.of("weeks_add(d, 1)",
"from_microsecond(ts)",
+ "cast(ts AS TINYINT)", "assert_true(k < 3, 'out of
range')")) {
+ // A row excluded by the leading key need not evaluate its
secondary expression.
+ PlanChecker checker = PlanChecker.from(connectContext)
+ .analyze("SELECT k, " + expression + " AS v FROM
topn_safety "
+ + "ORDER BY k, v LIMIT 3")
+ .rewrite().implement();
+ PhysicalPlan plan = new
PlanPostProcessors(checker.getCascadesContext())
+ .process(checker.getPhysicalPlan());
+
Assertions.assertFalse(checker.getCascadesContext().getTopnFilterContext()
+ .getTopnFilters().isEmpty(), plan.treeString());
+ }
+ } finally {
+ connectContext.getSessionVariable().topnFilterRatio =
previousRatio;
+ }
+ }
+
+ @Test
+ public void testTopNFilterCrossesGroupedKeyAggregation() {
+ double previousRatio =
connectContext.getSessionVariable().topnFilterRatio;
+ boolean previousPushTopn =
connectContext.getSessionVariable().pushTopnToAgg;
+ boolean previousJoinReorder =
connectContext.getSessionVariable().isDisableJoinReorder();
+ String previousDisabledRules = String.join(",",
+
connectContext.getSessionVariable().getDisableNereidsRuleNames());
+ connectContext.getSessionVariable().topnFilterRatio = 1000000;
+ connectContext.getSessionVariable().pushTopnToAgg = false;
+ connectContext.getSessionVariable().setDisableJoinReorder(true);
+ // Empty unit-test tables must retain scans to exercise the final
translated plan.
+
connectContext.getSessionVariable().setDisableNereidsRules(previousDisabledRules
+ ",PRUNE_EMPTY_PARTITION");
+ try {
+ for (String sql : List.of(
+ "SELECT k, count(1) FROM topn_safety GROUP BY k ORDER BY k
LIMIT 1",
+ "SELECT t.k, t.total FROM (SELECT k, count(1) AS total
FROM topn_safety "
+ + "GROUP BY k HAVING total > 10) t JOIN customer c
ON t.k = c.c_custkey "
+ + "ORDER BY t.k LIMIT 1",
+ "SELECT k, sum(ts) FROM topn_safety GROUP BY GROUPING
SETS((k)) ORDER BY k LIMIT 2")) {
+ // Exercise the runtime filter across aggregation,
independently of TopN-to-agg pushdown.
+ PlanChecker.from(connectContext).checkPlannerResult(sql,
planner -> {
+ String explain = planner.getFragments().stream()
+ .map(fragment ->
fragment.getExplainString(TExplainLevel.NORMAL))
+ .collect(Collectors.joining("\n"));
+ Assertions.assertTrue(explain.contains("TOPN OPT:"),
explain);
+ });
+ }
+ } finally {
+ connectContext.getSessionVariable().topnFilterRatio =
previousRatio;
+ connectContext.getSessionVariable().pushTopnToAgg =
previousPushTopn;
+
connectContext.getSessionVariable().setDisableJoinReorder(previousJoinReorder);
+
connectContext.getSessionVariable().setDisableNereidsRules(previousDisabledRules);
+ }
+ }
+
+ @Test
+ public void testTopNFilterSupportsNestedOrderKeyWithSiblingPredicate() {
+ double previousRatio =
connectContext.getSessionVariable().topnFilterRatio;
+ boolean previousPrune =
connectContext.getSessionVariable().enablePruneNestedColumns;
+ String previousDisabledRules = String.join(",",
+
connectContext.getSessionVariable().getDisableNereidsRuleNames());
+ connectContext.getSessionVariable().topnFilterRatio = 1000000;
+ connectContext.getSessionVariable().enablePruneNestedColumns = true;
+ // Empty unit-test tables must retain scans to exercise nested
access-path translation.
+
connectContext.getSessionVariable().setDisableNereidsRules(previousDisabledRules
+ ",PRUNE_EMPTY_PARTITION");
+ try {
+ for (String column : List.of("ts", "si", "sd")) {
+ for (String direction : List.of("ASC", "DESC")) {
+ String sql = "SELECT struct_element(" + column + ", 'a')
FROM topn_nested_fields "
+ + "WHERE struct_element(" + column + ", 'b') IS
NOT NULL "
+ + "ORDER BY 1 " + direction + " NULLS LAST LIMIT
1";
+ PlanChecker.from(connectContext).checkPlannerResult(sql,
planner -> {
+ String explain = planner.getFragments().stream()
+ .map(fragment ->
fragment.getExplainString(TExplainLevel.NORMAL))
+ .collect(Collectors.joining("\n"));
+ // TopN must retain its order field when the predicate
reads only a sibling's nulls.
+ Assertions.assertTrue(explain.contains("TOPN OPT:"),
explain);
+ Assertions.assertTrue(explain.contains("all access
paths: ["
+ + column + ".a, " + column + ".b.NULL]"),
explain);
+ Assertions.assertTrue(explain.contains("predicate
access paths: ["
+ + column + ".b.NULL]"), explain);
+ });
+ }
+ }
+ } finally {
+ connectContext.getSessionVariable().topnFilterRatio =
previousRatio;
+ connectContext.getSessionVariable().enablePruneNestedColumns =
previousPrune;
+
connectContext.getSessionVariable().setDisableNereidsRules(previousDisabledRules);
+ }
+ }
+
@Test
public void testUuidTopNFilterPreservesOrderingAndNullPlacement() {
double previousRatio =
connectContext.getSessionVariable().topnFilterRatio;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
