[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 #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 #22041.

## Implementation

`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 #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.

## QTF correctness — why `global_base` is preserved on reversal

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.

## 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`:
- `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>

Author: alchemist51

sandbox/plugins/analytics-backend-datafusion/rust/src/api.rs

@@ -333,9 +333,14 @@ pub struct ShardView {
     /// Index sort fields, in priority order. Sourced from the index's
     /// `index.sort.field` setting on the Java side. Empty when the index has
     /// no `index.sort.field` configured. Parallel to `sort_orders`.
-    /// Used to build `ListingOptions.with_file_sort_order(...)` so DataFusion
-    /// advertises `output_ordering` from the scan and enables the
-    /// `sort_prefix` optimization on TopK / SortPreservingMerge.
+    ///
+    /// Two consumers today:
+    ///   - Vanilla path: `ListingOptions.with_file_sort_order(...)` so DataFusion advertises
+    ///     `output_ordering` from the scan and the `sort_prefix` optimization fires on
+    ///     TopK / SortPreservingMerge.
+    ///   - Indexed path (`indexed_executor`): when the query's leading ORDER BY runs counter
+    ///     to catalog-snapshot order, the per-shard segment iteration is reversed so a TopK
+    ///     above us can prune via parquet page statistics.
     pub sort_fields: Vec<String>,
     /// Index sort directions per field — values: `"asc"` or `"desc"`.
     /// Parallel to `sort_fields`. Sourced from `index.sort.order`.

sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_executor.rs

@@ -176,6 +176,8 @@ pub async fn execute_indexed_query(
         table_path: shard_view.table_path.clone(),
         object_metas: shard_view.object_metas.clone(),
         writer_generations: shard_view.writer_generations.clone(),
+        sort_fields: shard_view.sort_fields.clone(),
+        sort_orders: shard_view.sort_orders.clone(),
         query_context: crate::query_tracker::QueryTrackingContext::new(context_id, runtime.runtime_env.memory_pool.clone(), crate::query_tracker::QueryType::Shard),
         table_name: table_name.clone(),
         indexed_config: None, // derive classification from tree
@@ -199,6 +201,101 @@ pub async fn execute_indexed_query(
 
 // ── Helpers ───────────────────────────────────────────────────────────
 
+/// Result of walking a logical plan looking for the leading top-of-plan ORDER BY.
+///
+/// `column` is the bare column name (no qualifier — we compare against `index.sort.field`
+/// which is also unqualified). `descending` is `true` for `ORDER BY x DESC`.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct TopSort {
+    pub column: String,
+    pub descending: bool,
+}
+
+/// Walk through the top of a logical plan to find the leading sort expression.
+///
+/// Descends through `Projection`, `Limit`, `SubqueryAlias`, `Distinct`, and `Filter` —
+/// nodes that don't reorder rows or rewrite sort key columns. On reaching
+/// `LogicalPlan::Sort`, returns the leading sort key's bare column name and direction.
+/// Returns `None` if the plan has no top-level Sort, or if the first Sort key is not
+/// a plain `Expr::Column` (e.g. `ORDER BY lower(x)` — we can't claim catalog monotonicity
+/// after a function).
+pub(crate) fn analyze_top_sort(plan: &datafusion::logical_expr::LogicalPlan) -> Option<TopSort> {
+    use datafusion::logical_expr::{Expr, LogicalPlan};
+    let mut current = plan;
+    loop {
+        match current {
+            LogicalPlan::Sort(s) => {
+                let leading = s.expr.first()?;
+                let col = match &leading.expr {
+                    Expr::Column(c) => c.name.clone(),
+                    _ => return None,
+                };
+                return Some(TopSort {
+                    column: col,
+                    descending: !leading.asc,
+                });
+            }
+            LogicalPlan::Projection(p) => current = p.input.as_ref(),
+            LogicalPlan::Limit(l) => current = l.input.as_ref(),
+            LogicalPlan::SubqueryAlias(a) => current = a.input.as_ref(),
+            LogicalPlan::Distinct(d) => match d {
+                datafusion::logical_expr::Distinct::All(input) => current = input.as_ref(),
+                datafusion::logical_expr::Distinct::On(on) => current = on.input.as_ref(),
+            },
+            LogicalPlan::Filter(f) => current = f.input.as_ref(),
+            _ => return None,
+        }
+    }
+}
+
+/// Decide whether to flip segment iteration order for the indexed scan.
+///
+/// Returns `true` iff the catalog has a sort declaration, the query has a top-level
+/// ORDER BY whose leading key matches the catalog's leading sort field by name, and
+/// the query's direction is the **opposite** of the catalog's. In that case the
+/// segments — laid down newest-last by the writer — are in reverse order from what
+/// the query wants, so iterating them tail-first feeds the largest values to a
+/// `TopK` first and parquet page stats prune the rest.
+///
+/// All comparisons are case-sensitive on the field name (matching PR #22041's
+/// `Column::from_name`). Direction comparison is case-insensitive on `"asc"`/`"desc"`.
+pub(crate) fn should_reverse_segments(
+    top_sort: Option<&TopSort>,
+    sort_fields: &[String],
+    sort_orders: &[String],
+) -> bool {
+    let Some(top) = top_sort else { return false };
+    let Some(catalog_field) = sort_fields.first() else { return false };
+    let Some(catalog_order) = sort_orders.first() else { return false };
+    if top.column != *catalog_field {
+        return false;
+    }
+    let catalog_descending = catalog_order.eq_ignore_ascii_case("desc");
+    top.descending != catalog_descending
+}
+
+/// Reverse the iteration order of `segments` in place. Per-segment `global_base`
+/// values are deliberately **left untouched** so each segment keeps its
+/// catalog-order shard-global row ID space.
+///
+/// Why not recompute global_base after reversing?
+///
+/// `global_base` is the additive offset used to compute the QTF `__row_id__`:
+/// `id = segment.global_base + position`. The fetch phase (`api::fetch_by_row_ids`)
+/// rebuilds segments via `build_segments` against `ShardView.object_metas` (always
+/// in catalog order) and reverses the mapping back via `partition_point` on
+/// `global_base`. For the round trip to hold, both phases must agree on each
+/// segment's `global_base` — and the only way to guarantee that without changing
+/// the fetch path is to keep query-phase `global_base` at its catalog-order value.
+///
+/// Reversing only the iteration order — i.e. the order in which chunks/RGs are
+/// scheduled by `compute_assignments` — is the *whole* point of this optimization
+/// (newest-segment-first feeds TopK earlier and lets page stats prune older
+/// segments). The `global_base` values are unrelated to that pruning win.
+pub(crate) fn reverse_segment_iteration_order(segments: &mut [SegmentFileInfo]) {
+    segments.reverse();
+}
+
 /// Collect all `Predicate(expr)` leaves in DFS order. Used by the
 /// dispatcher to build a per-leaf `PruningPredicate` cache keyed by
 /// `Arc::as_ptr` identity.
@@ -422,6 +519,191 @@ mod tests {
         ]);
         assert!(extract_single_collector_residual(&tree).is_none());
     }
+
+    // ── analyze_top_sort / should_reverse_segments ────────────────────
+
+    fn build_logical_plan(sql: &str) -> datafusion::logical_expr::LogicalPlan {
+        use datafusion::execution::SessionStateBuilder;
+        use datafusion::execution::context::SessionContext;
+        let state = SessionStateBuilder::new().with_default_features().build();
+        let ctx = SessionContext::new_with_state(state);
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("id", DataType::Int64, false),
+            Field::new("ts", DataType::Int64, false),
+            Field::new("v", DataType::Int32, false),
+        ]));
+        ctx.register_batch(
+            "t",
+            datafusion::arrow::record_batch::RecordBatch::new_empty(schema),
+        )
+        .expect("register_batch");
+        let rt = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap();
+        let df = rt.block_on(ctx.sql(sql)).expect("sql");
+        df.into_unoptimized_plan()
+    }
+
+    #[test]
+    fn analyze_top_sort_finds_leading_desc() {
+        let plan = build_logical_plan("SELECT id FROM t ORDER BY id DESC");
+        let ts = analyze_top_sort(&plan).expect("expected Sort");
+        assert_eq!(ts.column, "id");
+        assert!(ts.descending);
+    }
+
+    #[test]
+    fn analyze_top_sort_descends_through_limit() {
+        let plan = build_logical_plan("SELECT id FROM t ORDER BY id ASC LIMIT 10");
+        let ts = analyze_top_sort(&plan).expect("expected Sort");
+        assert_eq!(ts.column, "id");
+        assert!(!ts.descending);
+    }
+
+    #[test]
+    fn analyze_top_sort_descends_through_projection() {
+        let plan = build_logical_plan("SELECT v FROM t ORDER BY ts DESC");
+        let ts = analyze_top_sort(&plan).expect("expected Sort");
+        assert_eq!(ts.column, "ts");
+        assert!(ts.descending);
+    }
+
+    #[test]
+    fn analyze_top_sort_returns_none_when_no_sort() {
+        let plan = build_logical_plan("SELECT id FROM t");
+        assert!(analyze_top_sort(&plan).is_none());
+    }
+
+    #[test]
+    fn analyze_top_sort_returns_none_for_function_sort_key() {
+        // `ORDER BY abs(v)` — leading sort key isn't a plain column. Catalog monotonicity
+        // doesn't transfer through a function, so we conservatively decline to reverse.
+        let plan = build_logical_plan("SELECT id FROM t ORDER BY abs(v)");
+        assert!(analyze_top_sort(&plan).is_none());
+    }
+
+    #[test]
+    fn should_reverse_segments_matches_leading_field_opposite_direction() {
+        let top = TopSort { column: "id".to_string(), descending: true };
+        let fields = vec!["id".to_string()];
+        let orders = vec!["asc".to_string()];
+        assert!(should_reverse_segments(Some(&top), &fields, &orders));
+    }
+
+    #[test]
+    fn should_reverse_segments_matches_leading_field_same_direction() {
+        let top = TopSort { column: "id".to_string(), descending: false };
+        let fields = vec!["id".to_string()];
+        let orders = vec!["asc".to_string()];
+        assert!(!should_reverse_segments(Some(&top), &fields, &orders));
+    }
+
+    #[test]
+    fn should_reverse_segments_catalog_desc_query_asc() {
+        let top = TopSort { column: "id".to_string(), descending: false };
+        let fields = vec!["id".to_string()];
+        let orders = vec!["desc".to_string()];
+        assert!(should_reverse_segments(Some(&top), &fields, &orders));
+    }
+
+    #[test]
+    fn should_reverse_segments_no_query_sort() {
+        let fields = vec!["id".to_string()];
+        let orders = vec!["asc".to_string()];
+        assert!(!should_reverse_segments(None, &fields, &orders));
+    }
+
+    #[test]
+    fn should_reverse_segments_no_catalog_sort() {
+        let top = TopSort { column: "id".to_string(), descending: true };
+        assert!(!should_reverse_segments(Some(&top), &[], &[]));
+    }
+
+    #[test]
+    fn should_reverse_segments_query_sort_on_non_leading_catalog_field() {
+        // Catalog: [a ASC, b ASC]; query: ORDER BY b DESC. Segments are monotonic on `a`
+        // (the leading key), not `b`. Reversing won't help — decline.
+        let top = TopSort { column: "b".to_string(), descending: true };
+        let fields = vec!["a".to_string(), "b".to_string()];
+        let orders = vec!["asc".to_string(), "asc".to_string()];
+        assert!(!should_reverse_segments(Some(&top), &fields, &orders));
+    }
+
+    #[test]
+    fn should_reverse_segments_field_name_case_sensitive() {
+        // Match PR #22041 — `Column::from_name` is case-sensitive. If casing differs,
+        // we don't claim the catalog ordering applies. Safe default: no reversal.
+        let top = TopSort { column: "ID".to_string(), descending: true };
+        let fields = vec!["id".to_string()];
+        let orders = vec!["asc".to_string()];
+        assert!(!should_reverse_segments(Some(&top), &fields, &orders));
+    }
+
+    // ── reverse_segment_iteration_order ───────────────────────────────
+
+    fn dummy_segment(max_doc: i64, global_base: u64) -> SegmentFileInfo {
+        use datafusion::parquet::file::metadata::{FileMetaData, ParquetMetaData};
+        // Build a minimal ParquetMetaData. We never read it back in these tests.
+        let schema = std::sync::Arc::new(
+            datafusion::parquet::schema::types::SchemaDescriptor::new(
+                std::sync::Arc::new(
+                    datafusion::parquet::schema::types::Type::group_type_builder("schema")
+                        .build()
+                        .unwrap(),
+                ),
+            ),
+        );
+        let file_meta = FileMetaData::new(0, 0, None, None, schema, None);
+        let pq_meta = ParquetMetaData::new(file_meta, vec![]);
+        SegmentFileInfo {
+            writer_generation: global_base as i64 + 1, // arbitrary, just to vary
+            max_doc,
+            object_path: object_store::path::Path::from(format!("seg-{}.parquet", global_base)),
+            parquet_size: 0,
+            row_groups: vec![],
+            metadata: std::sync::Arc::new(pq_meta),
+            global_base,
+        }
+    }
+
+    #[test]
+    fn reverse_segments_preserves_global_base() {
+        // Original: A(max_doc=10, base=0), B(max_doc=20, base=10), C(max_doc=30, base=30).
+        // Reversal must keep each segment's original `global_base` intact so QTF row IDs
+        // emitted in query phase remain interpretable by the fetch phase (which always
+        // computes catalog-order bases).
+        let mut segs = vec![
+            dummy_segment(10, 0),
+            dummy_segment(20, 10),
+            dummy_segment(30, 30),
+        ];
+        reverse_segment_iteration_order(&mut segs);
+        assert_eq!(segs.len(), 3);
+        // New iteration order: C, B, A.
+        assert_eq!(segs[0].max_doc, 30);
+        assert_eq!(segs[0].global_base, 30); // C's catalog base, unchanged.
+        assert_eq!(segs[1].max_doc, 20);
+        assert_eq!(segs[1].global_base, 10); // B's catalog base, unchanged.
+        assert_eq!(segs[2].max_doc, 10);
+        assert_eq!(segs[2].global_base, 0);  // A's catalog base, unchanged.
+    }
+
+    #[test]
+    fn reverse_segments_empty_is_noop() {
+        let mut segs: Vec<SegmentFileInfo> = vec![];
+        reverse_segment_iteration_order(&mut segs);
+        assert!(segs.is_empty());
+    }
+
+    #[test]
+    fn reverse_segments_single_keeps_its_base() {
+        let mut segs = vec![dummy_segment(42, 7)];
+        reverse_segment_iteration_order(&mut segs);
+        assert_eq!(segs.len(), 1);
+        assert_eq!(segs[0].global_base, 7);
+        assert_eq!(segs[0].max_doc, 42);
+    }
 }
 
 /// Instruction-based indexed execution path. Consumes a pre-configured SessionContextHandle
@@ -508,6 +790,8 @@ async unsafe fn execute_indexed_with_context_inner(
     let table_path = handle.table_path;
     let object_metas = handle.object_metas;
     let writer_generations = handle.writer_generations;
+    let sort_fields = handle.sort_fields;
+    let sort_orders = handle.sort_orders;
     let query_context = handle.query_context;
     let io_handle = handle.io_handle;
     // Extract context_id early so it can be captured by the per-segment closures
@@ -550,6 +834,25 @@ async unsafe fn execute_indexed_with_context_inner(
         .map_err(|e| DataFusionError::Execution(format!("decode substrait: {}", e)))?;
     let logical_plan = from_substrait_plan(&ctx.state(), &plan).await?;
 
+    // Sort-aware segment iteration. Mirror of `ContextIndexSearcher.shouldUseTimeSeriesDescSortOptimization`
+    // for the indexed-parquet path. When the index has `index.sort.field` and the query's leading
+    // ORDER BY runs counter to the catalog's stored direction, reverse the segment vector so a
+    // TopK above us pulls the highest-priority segment first and parquet page stats prune the rest.
+    //
+    // QTF safe: `reverse_segment_iteration_order` deliberately does NOT recompute `global_base`
+    // — each segment retains its catalog-order base, so the row IDs query phase emits are still
+    // interpretable by `api::fetch_by_row_ids` (which builds its own segments from
+    // `ShardView.object_metas` in catalog order).
+    let mut segments = segments;
+    if should_reverse_segments(analyze_top_sort(&logical_plan).as_ref(), &sort_fields, &sort_orders) {
+        log_debug!(
+            "indexed_executor: reversing segment iteration (catalog leading sort={:?} {:?}, query opposite)",
+            sort_fields.first(),
+            sort_orders.first()
+        );
+        reverse_segment_iteration_order(&mut segments);
+    }
+
     let emit_row_ids = requests_row_ids;
     let filter_expr = extract_filter_expr(&logical_plan);
     let extraction = match filter_expr {

sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_table/tests_e2e/mod.rs

@@ -49,6 +49,7 @@ mod qtf_fetch_phase;
 mod row_id_emission;
 mod row_id_strategies;
 mod schema_drift;
+mod sort_reverse_row_id;
 mod streaming_at_scale;
 
 // ── Test fixture: parquet table with 16 rows ────────────────────────