feat(analytics): Sort multi-value fields by min

Signed-off-by: Varun Bansal <bansvaru@amazon.com>

Author: linuxpi

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

@@ -115,10 +115,6 @@ pub async fn register_listing_table(
     let mut listing_options = ListingOptions::new(Arc::new(ParquetFormat::new()))
         .with_file_extension(".parquet")
         .with_collect_stat(true);
-    if let Some(sort_exprs) = build_file_sort_order(sort_fields, sort_orders) {
-        listing_options = listing_options.with_file_sort_order(vec![sort_exprs]);
-    }
-
     let resolved_schema = listing_options
         .infer_schema(&ctx.state(), &table_path)
         .await
@@ -127,6 +123,11 @@ pub async fn register_listing_table(
             e
         })?;
     let resolved_schema = coerce_inferred_schema(resolved_schema);
+    if let Some(sort_exprs) =
+        build_file_sort_order(sort_fields, sort_orders, resolved_schema.as_ref())
+    {
+        listing_options = listing_options.with_file_sort_order(vec![sort_exprs]);
+    }
 
     let table_config = ListingTableConfig::new(table_path)
         .with_listing_options(listing_options)

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

@@ -975,11 +975,17 @@ async unsafe fn execute_indexed_with_context_inner(
     // 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,
-    ) {
+    let lead_sort_is_list = sort_fields
+        .first()
+        .and_then(|field| schema.field_with_name(field).ok())
+        .is_some_and(|field| matches!(field.data_type(), arrow::datatypes::DataType::List(_)));
+    if !lead_sort_is_list
+        && 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(),

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

@@ -208,7 +208,10 @@ fn compute_segment_sort_bounds(
     file_schema: &arrow::datatypes::SchemaRef,
     pq_meta: &ParquetMetaData,
 ) -> (Option<ScalarValue>, Option<ScalarValue>) {
-    if file_schema.index_of(lead_field).is_err() {
+    let Ok(field) = file_schema.field_with_name(lead_field) else {
+        return (None, None);
+    };
+    if matches!(field.data_type(), arrow::datatypes::DataType::List(_)) {
         return (None, None);
     }
 
@@ -526,6 +529,61 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn scalar_and_list_field_promote_to_list_schema() {
+        let dir = tempdir().unwrap();
+        let scalar_schema = Arc::new(Schema::new(vec![Field::new("tags", DataType::Utf8, true)]));
+        let list_child = Arc::new(Field::new("element", DataType::Utf8, true));
+        let list_schema = Arc::new(Schema::new(vec![Field::new(
+            "tags",
+            DataType::List(Arc::clone(&list_child)),
+            true,
+        )]));
+        let scalar_path = write_parquet(
+            dir.path(),
+            "a.parquet",
+            scalar_schema,
+            vec![Arc::new(StringArray::from(vec![Some("prod")]))],
+        );
+        let list_path = write_parquet(
+            dir.path(),
+            "b.parquet",
+            list_schema,
+            vec![Arc::new(ListArray::new(
+                list_child,
+                OffsetBuffer::new(vec![0_i32, 2].into()),
+                Arc::new(StringArray::from(vec!["prod", "error"])),
+                None,
+            ))],
+        );
+
+        let store: Arc<dyn ObjectStore> = Arc::new(LocalFileSystem::new());
+        let metas = object_metas(store.as_ref(), &[scalar_path, list_path]).await;
+        let ctx = SessionContext::new();
+        let generations: Vec<i64> = (0..metas.len() as i64).collect();
+        let (segments, schema) = build_segments(
+            &ctx.state(),
+            Arc::clone(&store),
+            &metas,
+            &generations,
+            default_metadata_cache(),
+            &["tags".to_string()],
+        )
+        .await
+        .unwrap();
+
+        assert!(
+            segments[1].sort_min.is_none() && segments[1].sort_max.is_none(),
+            "LIST child statistics must not be used as per-row list_min bounds"
+        );
+
+        assert!(matches!(
+            schema.field_with_name("tags").unwrap().data_type(),
+            DataType::List(child)
+                if matches!(child.data_type(), DataType::Utf8 | DataType::Utf8View)
+        ));
+    }
+
     /// Incompatible types (Int32 vs Int64 on the same field name) is
     /// a fail-fast: the Arrow `Schema::try_merge` rejects it, and we
     /// bubble the error up. Catches accidental type widening at the

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

@@ -150,7 +150,18 @@ fn build_projected_lex_ordering(
     let mut exprs: Vec<PhysicalSortExpr> = Vec::with_capacity(sort_fields.len());
     for (i, field) in sort_fields.iter().enumerate() {
         let phys = match physical_col(field, projected_schema) {
-            Ok(e) => e,
+            Ok(expr) => match projected_schema
+                .field_with_name(field)
+                .map(|field| field.data_type())
+            {
+                Ok(DataType::List(_)) => {
+                    match crate::udf::list_min::physical_expr(expr, projected_schema.as_ref()) {
+                        Ok(expr) => expr,
+                        Err(_) => break,
+                    }
+                }
+                _ => expr,
+            },
             Err(_) => break,
         };
         let descending = sort_orders
@@ -866,6 +877,21 @@ mod tests {
         }
     }
 
+    #[test]
+    fn list_sort_key_advertises_list_min_physical_ordering() {
+        let child = Arc::new(Field::new("element", DataType::Utf8View, true));
+        let schema = Arc::new(Schema::new(vec![Field::new(
+            "tags",
+            DataType::List(child),
+            true,
+        )]));
+        let ordering =
+            build_projected_lex_ordering(&schema, &["tags".into()], &["desc".into()]).unwrap();
+        assert!(format!("{}", ordering[0].expr).contains("list_min"));
+        assert!(ordering[0].options.descending);
+        assert!(!ordering[0].options.nulls_first);
+    }
+
     // QueryShardExec holds an ExecutionPlanMetricsSet (not Clone). We only
     // need to inspect `.predicate`, so read through a reference.
     async fn scan_predicate(

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

@@ -306,12 +306,6 @@ pub async unsafe fn create_session_context(
         .with_collect_stat(true)
         .with_target_partitions(effective_partitions);
 
-    if let Some(sort_exprs) =
-        build_file_sort_order(&shard_view.sort_fields, &shard_view.sort_orders)
-    {
-        listing_options = listing_options.with_file_sort_order(vec![sort_exprs]);
-    }
-
     // Register under the planner's logical table name (alias / index pattern / index), shipped
     // explicitly as logicalTableName on the shard-scan instruction node. See
     // resolve_register_name for why we do NOT reverse-engineer this from the plan bytes. The
@@ -368,11 +362,18 @@ pub async unsafe fn create_session_context(
     // failing with "Cannot merge statistics with different number of columns". Non-widened
     // (single-index) scans keep full stats.
     // TODO: re-enable once DataFusion's Statistics::try_merge tolerates a column-count delta.
-    let listing_options = if resolved_schema.fields().len() != inferred_field_count {
+    let mut listing_options = if resolved_schema.fields().len() != inferred_field_count {
         listing_options.with_collect_stat(false)
     } else {
         listing_options
     };
+    if let Some(sort_exprs) = build_file_sort_order(
+        &shard_view.sort_fields,
+        &shard_view.sort_orders,
+        resolved_schema.as_ref(),
+    ) {
+        listing_options = listing_options.with_file_sort_order(vec![sort_exprs]);
+    }
 
     let table_config = ListingTableConfig::new(shard_view.table_path.clone())
         .with_listing_options(listing_options)
@@ -737,6 +738,7 @@ fn try_acquire_budget(
 pub(crate) fn build_file_sort_order(
     sort_fields: &[String],
     sort_orders: &[String],
+    schema: &arrow::datatypes::Schema,
 ) -> Option<Vec<datafusion::logical_expr::SortExpr>> {
     if sort_fields.is_empty() {
         return None;
@@ -749,7 +751,12 @@ pub(crate) fn build_file_sort_order(
         .map(|(name, order)| {
             let ascending = order.eq_ignore_ascii_case("asc");
             let nulls_first = ascending;
-            Expr::Column(Column::from_name(name.clone())).sort(ascending, nulls_first)
+            let column = Expr::Column(Column::from_name(name.clone()));
+            let key = match schema.field_with_name(name).map(|field| field.data_type()) {
+                Ok(arrow::datatypes::DataType::List(_)) => crate::udf::list_min::expr(column),
+                _ => column,
+            };
+            key.sort(ascending, nulls_first)
         })
         .collect();
     Some(sort_exprs)
@@ -772,6 +779,26 @@ mod tests {
     use crate::agg_mode::Mode;
     use crate::query_tracker::QueryTrackingContext;
 
+    #[test]
+    fn file_sort_order_uses_fixed_list_min_for_both_directions() {
+        let child = Arc::new(Field::new("element", DataType::Utf8View, true));
+        let schema = Schema::new(vec![
+            Field::new("tags", DataType::List(child), true),
+            Field::new("id", DataType::Int64, true),
+        ]);
+
+        for (order, ascending) in [("asc", true), ("desc", false)] {
+            let ordering =
+                build_file_sort_order(&["tags".into()], &[order.into()], &schema).unwrap();
+            assert!(format!("{}", ordering[0].expr).contains("list_min"));
+            assert_eq!(ordering[0].asc, ascending);
+            assert_eq!(ordering[0].nulls_first, ascending);
+        }
+
+        let scalar = build_file_sort_order(&["id".into()], &["asc".into()], &schema).unwrap();
+        assert!(!format!("{}", scalar[0].expr).contains("list_min"));
+    }
+
     #[tokio::test]
     async fn test_widen_schema_noop_when_plan_empty() {
         let ctx = SessionContext::new();