Reverse indexed-parquet segment iteration for opposite-direction sorts#22056
Conversation
PR Code Analyzer ❗AI-powered 'Code-Diff-Analyzer' found issues on commit c38fabb.
The table above displays the top 10 most important findings. Pull Requests Author(s): Please update your Pull Request according to the report above. Repository Maintainer(s): You can Thanks. |
8d5b4ce to
c38fabb
Compare
c38fabb to
1a53c01
Compare
PR Reviewer Guide 🔍(Review updated until commit 291a49f)Here are some key observations to aid the review process:
|
PR Code Suggestions ✨Latest suggestions up to 291a49f Explore these optional code suggestions:
Previous suggestionsSuggestions up to commit 35801ae
Suggestions up to commit 2fa982e
Suggestions up to commit ece6b9e
Suggestions up to commit 3536f9d
Suggestions up to commit b714624
|
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #22056 +/- ##
=========================================
Coverage 73.35% 73.36%
- Complexity 75854 75880 +26
=========================================
Files 6070 6070
Lines 344609 344609
Branches 49576 49576
=========================================
+ Hits 252799 252812 +13
+ Misses 71684 71674 -10
+ Partials 20126 20123 -3 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
1a53c01 to
0295440
Compare
|
Persistent review updated to latest commit 0295440 |
0295440 to
3dc538e
Compare
|
Persistent review updated to latest commit 3dc538e |
3dc538e to
60d3bdc
Compare
|
Persistent review updated to latest commit 60d3bdc |
60d3bdc to
decfdd4
Compare
|
Persistent review updated to latest commit decfdd4 |
|
❌ Gradle check result for decfdd4: FAILURE Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change? |
decfdd4 to
ca6250c
Compare
|
Persistent review updated to latest commit ca6250c |
ca6250c to
846878c
Compare
|
Persistent review updated to latest commit 846878c |
|
❌ Gradle check result for 846878c: FAILURE Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change? |
846878c to
32dd4f9
Compare
|
❌ Gradle check result for 67bfb10: FAILURE Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change? |
|
Persistent review updated to latest commit ccfe95e |
|
Persistent review updated to latest commit b714624 |
b714624 to
3536f9d
Compare
|
Persistent review updated to latest commit 3536f9d |
|
Persistent review updated to latest commit ece6b9e |
|
❌ Gradle check result for ece6b9e: null Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change? |
ece6b9e to
2fa982e
Compare
|
Persistent review updated to latest commit 2fa982e |
…ction sorts Mirror of `ContextIndexSearcher.shouldUseTimeSeriesDescSortOptimization` for the indexed-parquet path. When `index.sort.field` is configured and the query's leading ORDER BY runs counter to the catalog's stored direction, reverse the per-shard segment iteration so a TopK above the scan pulls the highest-priority segment first and parquet page stats prune the rest. Reuses PR opensearch-project#22041's `ShardView.sort_fields` / `sort_orders` plumbing — no new Java/FFM surface. Only the indexed scan path (`IndexedTableProvider` / `QueryShardExec`) consumes them; the vanilla `ListingTable` path remains covered by opensearch-project#22041. `indexed_executor.rs`: - `analyze_top_sort(&LogicalPlan)` walks Projection/Limit/SubqueryAlias/ Distinct/Filter to find the leading top-of-plan ORDER BY. Returns None if the leading sort key isn't a plain `Expr::Column` (function on the sort key breaks catalog monotonicity). - `should_reverse_segments` triggers iff the catalog has a sort, the query's leading sort column matches `sort_fields[0]` (case-sensitive, matching PR opensearch-project#22041), and the directions differ. - `reverse_segment_iteration_order` flips `Vec<SegmentFileInfo>` in place. Per-segment `global_base` is deliberately NOT recomputed — see below. - Wired into `execute_indexed_with_context_inner` after `from_substrait_plan` decodes the LogicalPlan. `SessionContextHandle` carries `sort_fields` / `sort_orders` cloned from `ShardView` so both `create_session_context` and `create_session_context_indexed` populate the indexed-path consumer. The `__row_id__` emitted by the indexed scan is `segment.global_base + position`. QTF's fetch phase (`api::fetch_by_row_ids`) goes through a different code path (`ShardTableProvider`, not `IndexedTableProvider`) and rebuilds segments via `build_segments(ShardView.object_metas, ShardView.writer_generations)` — always in catalog order. Fetch then partition_points the cached row IDs against catalog-order `global_base`s to find the owning segment. For the round trip to hold, query-phase IDs must match catalog-order bases. `reverse_segment_iteration_order` therefore flips iteration order ONLY — each segment retains its catalog-order `global_base`. Reversal becomes purely a scheduling decision; the row-id space is unchanged. Rust unit tests in `indexed_executor::tests`: - `analyze_top_sort` (top-level desc, descends through Limit/Projection, None for no-sort, None for function sort key). - `should_reverse_segments` (matching field opposite direction, matching same direction, catalog desc query asc, no query sort, no catalog sort, non-leading catalog field, case-sensitive name). - `reverse_segments_preserves_global_base`, `reverse_segments_empty_is_noop`, `reverse_segments_single_keeps_its_base`. Rust e2e tests in `indexed_table/tests_e2e/sort_reverse_row_id.rs`: - `three_segments_row_ids_invariant_under_reversal`, `four_segments_row_ids_invariant_under_reversal` — pin the load-bearing invariant: same shard-global `__row_id__` set whether iteration is natural or reversed. Java IT in `sandbox/qa/analytics-engine-coordinator/.../SortReverseQtfIT.java` (parquet+lucene composite, 2 shards, 3 flushes → 2-4 segments per shard): - `testReversalTriggers_orderByTsDescLimit_correctResults` — captures the reversal log line via `MockLogAppender` on `RustLoggerBridge` AND asserts result correctness (top-k by ts DESC). - `testNoReversal_orderByTsAscLimit_correctResults` — `UnseenEventExpectation` proves reversal does NOT fire; results still correct. - `testReversal_keywordFilterDelegationPath_correctResults` — keyword filter shipped to Lucene perf peer + reversal; verifies QTF row-id round-trip composes with delegation. Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
2fa982e to
35801ae
Compare
|
Persistent review updated to latest commit 35801ae |
|
❌ Gradle check result for 35801ae: null Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change? |
35801ae to
7ed685e
Compare
Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
7ed685e to
291a49f
Compare
|
Persistent review updated to latest commit 291a49f |
opensearch-project#22056) * [Sandbox] Reverse indexed-parquet segment iteration for opposite-direction sorts Mirror of `ContextIndexSearcher.shouldUseTimeSeriesDescSortOptimization` for the indexed-parquet path. When `index.sort.field` is configured and the query's leading ORDER BY runs counter to the catalog's stored direction, reverse the per-shard segment iteration so a TopK above the scan pulls the highest-priority segment first and parquet page stats prune the rest. Reuses PR opensearch-project#22041's `ShardView.sort_fields` / `sort_orders` plumbing — no new Java/FFM surface. Only the indexed scan path (`IndexedTableProvider` / `QueryShardExec`) consumes them; the vanilla `ListingTable` path remains covered by opensearch-project#22041. `indexed_executor.rs`: - `analyze_top_sort(&LogicalPlan)` walks Projection/Limit/SubqueryAlias/ Distinct/Filter to find the leading top-of-plan ORDER BY. Returns None if the leading sort key isn't a plain `Expr::Column` (function on the sort key breaks catalog monotonicity). - `should_reverse_segments` triggers iff the catalog has a sort, the query's leading sort column matches `sort_fields[0]` (case-sensitive, matching PR opensearch-project#22041), and the directions differ. - `reverse_segment_iteration_order` flips `Vec<SegmentFileInfo>` in place. Per-segment `global_base` is deliberately NOT recomputed — see below. - Wired into `execute_indexed_with_context_inner` after `from_substrait_plan` decodes the LogicalPlan. `SessionContextHandle` carries `sort_fields` / `sort_orders` cloned from `ShardView` so both `create_session_context` and `create_session_context_indexed` populate the indexed-path consumer. The `__row_id__` emitted by the indexed scan is `segment.global_base + position`. QTF's fetch phase (`api::fetch_by_row_ids`) goes through a different code path (`ShardTableProvider`, not `IndexedTableProvider`) and rebuilds segments via `build_segments(ShardView.object_metas, ShardView.writer_generations)` — always in catalog order. Fetch then partition_points the cached row IDs against catalog-order `global_base`s to find the owning segment. For the round trip to hold, query-phase IDs must match catalog-order bases. `reverse_segment_iteration_order` therefore flips iteration order ONLY — each segment retains its catalog-order `global_base`. Reversal becomes purely a scheduling decision; the row-id space is unchanged. Rust unit tests in `indexed_executor::tests`: - `analyze_top_sort` (top-level desc, descends through Limit/Projection, None for no-sort, None for function sort key). - `should_reverse_segments` (matching field opposite direction, matching same direction, catalog desc query asc, no query sort, no catalog sort, non-leading catalog field, case-sensitive name). - `reverse_segments_preserves_global_base`, `reverse_segments_empty_is_noop`, `reverse_segments_single_keeps_its_base`. Rust e2e tests in `indexed_table/tests_e2e/sort_reverse_row_id.rs`: - `three_segments_row_ids_invariant_under_reversal`, `four_segments_row_ids_invariant_under_reversal` — pin the load-bearing invariant: same shard-global `__row_id__` set whether iteration is natural or reversed. Java IT in `sandbox/qa/analytics-engine-coordinator/.../SortReverseQtfIT.java` (parquet+lucene composite, 2 shards, 3 flushes → 2-4 segments per shard): - `testReversalTriggers_orderByTsDescLimit_correctResults` — captures the reversal log line via `MockLogAppender` on `RustLoggerBridge` AND asserts result correctness (top-k by ts DESC). - `testNoReversal_orderByTsAscLimit_correctResults` — `UnseenEventExpectation` proves reversal does NOT fire; results still correct. - `testReversal_keywordFilterDelegationPath_correctResults` — keyword filter shipped to Lucene perf peer + reversal; verifies QTF row-id round-trip composes with delegation. Signed-off-by: Arpit Bandejiya <abandeji@amazon.com> * trigger CI rerun (gradle-check Jenkins null-result flake) Signed-off-by: Arpit Bandejiya <abandeji@amazon.com> --------- Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
Description
Mirror of Lucene's
ContextIndexSearcher.shouldUseTimeSeriesDescSortOptimizationfor the indexed-parquet scan path (IndexedTableProvider/QueryShardExec). Whenindex.sort.fieldis configured and the query's leadingORDER BYruns counter to the catalog's stored direction, reverse the per-shard segment iteration so aTopKabove the scan pulls the highest-priority segment first and parquet page-stats pruning kicks in for the rest.Reuses PR #22041's plumbing. That PR threads
index.sort.field/index.sort.orderfromIndexSettingsthroughDatafusionReaderManager→DatafusionReader→ReaderHandle→df_create_readertoShardView.sort_fields/ShardView.sort_orders. This PR adds a second consumer of those same fields, on the indexed path, with no new Java or FFM surface area.Where the optimization fires
indexed_executor.rs::execute_indexed_with_context_inner, afterfrom_substrait_plandecodes theLogicalPlan:Three module-local helpers:
analyze_top_sort(&LogicalPlan) -> Option<TopSort>— walksProjection/Limit/SubqueryAlias/Distinct/Filterto find the firstLogicalPlan::Sort. Returns the leading sort key's bare column name and direction. ReturnsNoneif no top-level Sort, or if the leading sort key is anything but a plainExpr::Column(function on the sort key breaks catalog monotonicity).should_reverse_segments(top_sort, sort_fields, sort_orders) -> bool— true iff the catalog has a sort, the query's leading sort column matchessort_fields[0]exactly (case-sensitive, matching [Sandbox] Propagate index.sort.field to DataFusion ListingTable #22041'sColumn::from_name), and the directions differ. Multi-key sorts use onlysort_fields[0]— segments aren't monotonic on a non-leading key.reverse_segment_iteration_order(&mut [SegmentFileInfo])— flips the slice in place. Per-segmentglobal_baseis deliberately not recomputed; see below.QTF correctness — why
global_basemust be preservedThe
__row_id__emitted by the indexed scan issegment.global_base + position. QTF's fetch phase (api::fetch_by_row_ids) goes through a different code path (ShardTableProvider, notIndexedTableProvider) and rebuilds segments viabuild_segments(ShardView.object_metas, ShardView.writer_generations)— always in catalog order. Fetch thenpartition_points the cached row IDs against catalog-orderglobal_bases to find the owning segment.For the round trip to hold, query-phase IDs must match catalog-order bases.
reverse_segment_iteration_ordertherefore flips iteration order only — each segment retains its catalog-orderglobal_base. Reversal becomes purely a scheduling decision; the row-id space is unchanged. The optimization composes safely with QTF.What this PR does not change
ListingTable/DataSourceExec) — covered by [Sandbox] Propagate index.sort.field to DataFusion ListingTable #22041 viaListingOptions::with_file_sort_order(...)(different mechanism: advertiseoutput_orderingso the planner can dropSortExec).QueryShardExec.properties.equivalence_properties— no output-ordering claim. The TopK / SortPreservingMergeExec above the scan still re-sorts. Reversal is purely a pruning aid for sort-with-limit shapes; it doesn't try to make the scan's output-ordering claim load-bearing. That's a separate, more invasive change.Tests
Rust unit tests in
indexed_executor::tests:analyze_top_sort_*— top-level desc, descends through Limit/Projection, None for no-sort, None for function sort key.should_reverse_segments_*— matching field opposite direction, matching same direction, catalog desc query asc, no query sort, no catalog sort, non-leading catalog field, case-sensitive name.reverse_segments_preserves_global_base,reverse_segments_empty_is_noop,reverse_segments_single_keeps_its_base.Rust e2e tests in
indexed_table/tests_e2e/sort_reverse_row_id.rs— pin the load-bearing invariant directly: build 3- and 4-segment fixtures withemit_row_ids: true, run the query in natural order vsreverse_segment_iteration_order, assert the set of emitted__row_id__values is identical (sorted set equality). Ifglobal_basecumulation drifted under reversal, fetch would return wrong rows; this catches it locally.Java IT in
sandbox/qa/analytics-engine-coordinator/.../SortReverseQtfIT.java(parquet primary + Lucene secondary, 2 shards, 3 flushes → 2-4 parquet segments per shard,index.sort.field=ts asc):testReversalTriggers_orderByTsDescLimit_correctResults— captures the reversal log line viaMockLogAppenderonRustLoggerBridge(proves the optimization fired) and asserts top-5 rows byts DESCare correct via the QTF query→fetch round-trip.testNoReversal_orderByTsAscLimit_correctResults—UnseenEventExpectationconfirms reversal does NOT fire when query direction matches catalog; results still correct.testReversal_keywordFilterDelegationPath_correctResults—WHERE category = 'C0' ORDER BY ts DESC LIMIT 4exercises the keyword-filter delegation path (predicate shipped to Lucene perf peer) + reversal; asserts QTF row-id round-trip composes with delegation.Verification
cargo test --lib: 1045 / 1045 pass (1043 pre-existing + 22 new unit tests + 2 new e2e tests, after dedupe).:server,:sandbox:plugins:composite-engine,:sandbox:plugins:analytics-backend-datafusion,:sandbox:plugins:analytics-backend-lucene,:sandbox:qa:analytics-engine-coordinatorallcompileJava/compileTestJava/compileInternalClusterTestJavaclean.Benchmark results — full clickbench corpus
Measured on EC2
r8g.2xlarge(ARM64, 8 vCPU, 64 GB) with the full 100M-row OSB clickbench corpus (parquet primary + Lucene secondary, OSB sort[CounterID desc, EventDate desc, UserID desc, EventTime desc, WatchID desc]).Same EC2 box, same data, same script, same 13 queries (10 reversal-eligible + 3 controls), 3 passes per query. Build A = upstream main without this PR; Build B = upstream main + this PR. min_ms is reported (warm cache, ignores pass-1 cold-cache outliers; mean is misleading because Q01 cold-cache is ~2300 ms before page cache warms).
All queries:
where <field> = <literal> | sort + CounterID | head 1000(catalog leading isCounterID desc→+triggers reversal). Selectivity is the number of rows matching the filter against the 100M corpus.BrowserLanguage = 'S0'(~92M)HitColor = '5'(~80M)BrowserCountry = 'h1'(~28M)MobilePhoneModel = 'iPad'(~5M)UTMSource = 'feedburner'(~572K)MobilePhoneModel = 'iPhone'(~229K)sort - CounterID)sort + EventDate)(Q05/Q06/Q09 returned 0 rows on this corpus slice — literal absent from the data — and are omitted; their flat timings are uninformative.)
Reading the result:
head 1000against tens of millions of matching rows actually has work to do — TopK has to scan many segments before its threshold tightens. With reversal, the highest-CounterIDsegment is hit first, the threshold tightens fast, and parquet page stats prune most of the remaining segments.Variance across warm passes is ~3–5 ms, well below the 60–140 ms deltas on Q01–Q04.
Related Issues / PRs
Builds on #22041 (vanilla-path index.sort.field propagation).
Check List
By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.