feat(iceberg): thread per-file sort order into FileScanTask

crates/iceberg/public-api.txt

@@ -1292,6 +1292,8 @@ pub fn iceberg::scan::FileScanTask::project_field_ids(&self) -> &[i32]
 pub fn iceberg::scan::FileScanTask::record_count(&self) -> core::option::Option<u64>
 pub fn iceberg::scan::FileScanTask::schema(&self) -> &iceberg::spec::Schema
 pub fn iceberg::scan::FileScanTask::schema_ref(&self) -> iceberg::spec::SchemaRef
+pub fn iceberg::scan::FileScanTask::sort_order(&self) -> core::option::Option<&iceberg::spec::SortOrderRef>
+pub fn iceberg::scan::FileScanTask::sort_order_id(&self) -> core::option::Option<i32>
 pub fn iceberg::scan::FileScanTask::start(&self) -> u64
 pub fn iceberg::scan::FileScanTask::unified_partition_type(&self) -> core::option::Option<&alloc::sync::Arc<iceberg::spec::StructType>>
 impl core::clone::Clone for iceberg::scan::FileScanTask
@@ -1304,7 +1306,7 @@ impl core::fmt::Debug for iceberg::scan::FileScanTask
 pub fn iceberg::scan::FileScanTask::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
 impl core::marker::StructuralPartialEq for iceberg::scan::FileScanTask
 impl iceberg::scan::FileScanTask
-pub fn iceberg::scan::FileScanTask::builder() -> FileScanTaskBuilder<((), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), ())>
+pub fn iceberg::scan::FileScanTask::builder() -> FileScanTaskBuilder<((), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), ())>
 impl serde_core::ser::Serialize for iceberg::scan::FileScanTask
 pub fn iceberg::scan::FileScanTask::serialize<S>(&self, serializer: S) -> core::result::Result<<S as serde_core::ser::Serializer>::Ok, <S as serde_core::ser::Serializer>::Error> where S: serde_core::ser::Serializer
 impl<'de> serde_core::de::Deserialize<'de> for iceberg::scan::FileScanTask

crates/iceberg/src/scan/context.rs

@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use std::collections::HashMap;
 use std::sync::Arc;
 
 use futures::channel::mpsc::Sender;
@@ -29,7 +30,7 @@ use crate::scan::{
 };
 use crate::spec::{
     ManifestContentType, ManifestEntryRef, ManifestFile, ManifestList, NameMapping,
-    PartitionSpecRef, SchemaRef, SnapshotRef, StructType, TableMetadataRef,
+    PartitionSpecRef, SchemaRef, SnapshotRef, SortOrderRef, StructType, TableMetadataRef,
 };
 use crate::{Error, ErrorKind, Result};
 
@@ -50,6 +51,7 @@ pub(crate) struct ManifestFileContext {
     case_sensitive: bool,
     partition_spec: Option<PartitionSpecRef>,
     unified_partition_type: Option<Arc<StructType>>,
+    sort_orders: Arc<HashMap<i64, SortOrderRef>>,
 }
 
 /// Wraps a [`ManifestEntryRef`] alongside the objects that are needed
@@ -67,6 +69,8 @@ pub(crate) struct ManifestEntryContext {
     pub case_sensitive: bool,
     pub partition_spec: Option<PartitionSpecRef>,
     pub unified_partition_type: Option<Arc<StructType>>,
+    pub sort_order_id: Option<i32>,
+    pub sort_order: Option<SortOrderRef>,
 }
 
 impl ManifestFileContext {
@@ -86,11 +90,23 @@ impl ManifestFileContext {
             case_sensitive,
             partition_spec,
             unified_partition_type,
+            sort_orders,
         } = self;
 
         let manifest = object_cache.get_manifest(&manifest_file).await?;
 
         for manifest_entry in manifest.entries() {
+            // Carry the raw id through unresolved, then resolve it to an order for the
+            // reader. The resolved order is `None` for the unsorted order (empty fields) so
+            // that `Some` always means genuinely sorted, matching Java's `isSorted()` gate;
+            // the raw id above keeps the "physically sorted but definition dropped" case
+            // recoverable.
+            let sort_order_id = manifest_entry.data_file().sort_order_id();
+            let sort_order = sort_order_id
+                .and_then(|id| sort_orders.get(&(id as i64)))
+                .filter(|order| !order.is_unsorted())
+                .cloned();
+
             let manifest_entry_context = ManifestEntryContext {
                 // TODO: refactor to avoid the expensive ManifestEntry clone
                 manifest_entry: manifest_entry.clone(),
@@ -104,6 +120,8 @@ impl ManifestFileContext {
                 case_sensitive,
                 partition_spec: partition_spec.clone(),
                 unified_partition_type: unified_partition_type.clone(),
+                sort_order_id,
+                sort_order,
             };
 
             sender
@@ -150,6 +168,8 @@ impl ManifestEntryContext {
             .with_unified_partition_type(self.unified_partition_type.clone())
             .with_case_sensitive(self.case_sensitive)
             .with_key_metadata(self.manifest_entry.data_file.key_metadata().map(Box::from))
+            .with_sort_order_id(self.sort_order_id)
+            .with_sort_order(self.sort_order)
             .build()
     }
 }
@@ -174,6 +194,11 @@ pub(crate) struct PlanContext {
     pub expression_evaluator_cache: Arc<ExpressionEvaluatorCache>,
 
     pub unified_partition_type: Option<Arc<StructType>>,
+
+    /// The table's sort orders keyed by id, precomputed once so each
+    /// [`ManifestFileContext`] carries only this narrow map rather than the full table
+    /// metadata. Mirrors how `unified_partition_type` carries a precomputed value.
+    pub sort_orders: Arc<HashMap<i64, SortOrderRef>>,
 }
 
 impl PlanContext {
@@ -304,6 +329,7 @@ impl PlanContext {
                 .partition_spec_by_id(manifest_file.partition_spec_id)
                 .cloned(),
             unified_partition_type: self.unified_partition_type.clone(),
+            sort_orders: self.sort_orders.clone(),
         }
     }
 }

crates/iceberg/src/scan/mod.rs

@@ -23,6 +23,7 @@ mod context;
 use context::*;
 mod task;
 
+use std::collections::HashMap;
 use std::sync::Arc;
 
 use arrow_array::RecordBatch;
@@ -41,7 +42,7 @@ use crate::metadata_columns::{
     RESERVED_FIELD_ID_PARTITION, get_metadata_field_id, is_metadata_column_name,
 };
 use crate::runtime::Runtime;
-use crate::spec::{DataContentType, Schema, SchemaRef, SnapshotRef, StructType};
+use crate::spec::{DataContentType, Schema, SchemaRef, SnapshotRef, SortOrderRef, StructType};
 use crate::table::Table;
 use crate::util::available_parallelism;
 use crate::{Error, ErrorKind, Result};
@@ -311,6 +312,16 @@ impl<'a> TableScanBuilder<'a> {
             .map(Arc::new);
         let unified_partition_type = projected_partition_type(self.table, &schema, &field_ids)?;
 
+        // Precompute the table's sort orders once, keyed by id, so each manifest-file
+        // context carries only this narrow map instead of the full table metadata.
+        let sort_orders = Arc::new(
+            self.table
+                .metadata()
+                .sort_orders_iter()
+                .map(|order| (order.order_id, order.clone()))
+                .collect::<HashMap<i64, SortOrderRef>>(),
+        );
+
         let plan_context = PlanContext {
             snapshot,
             table_metadata: self.table.metadata_ref(),
@@ -325,6 +336,7 @@ impl<'a> TableScanBuilder<'a> {
             manifest_evaluator_cache: Arc::new(ManifestEvaluatorCache::new()),
             expression_evaluator_cache: Arc::new(ExpressionEvaluatorCache::new()),
             unified_partition_type,
+            sort_orders,
         };
 
         Ok(TableScan {
@@ -664,9 +676,10 @@ pub mod tests {
     use crate::spec::{
         DataContentType, DataFileBuilder, DataFileFormat, Datum, FormatVersion, Literal,
         MAIN_BRANCH, ManifestEntry, ManifestListWriter, ManifestStatus, ManifestWriterBuilder,
-        MappedField, NameMapping, NestedField, Operation, PartitionSpec, PrimitiveType, Schema,
-        Snapshot, Struct, StructType, Summary, TableMetadata, TableMetadataBuilder,
-        TableProperties, Transform, Type, UnboundPartitionSpec,
+        MappedField, NameMapping, NestedField, NullOrder, Operation, PartitionSpec, PrimitiveType,
+        Schema, Snapshot, SortDirection, SortField, SortOrder, Struct, StructType, Summary,
+        TableMetadata, TableMetadataBuilder, TableProperties, Transform, Type,
+        UnboundPartitionSpec,
     };
     use crate::table::Table;
     use crate::test_utils::test_runtime;
@@ -1638,6 +1651,76 @@ pub mod tests {
             manifest_list_write.close().await.unwrap();
         }
 
+        /// Writes a manifest with three live "Added" data-file entries (partitioned on `x`
+        /// = 100, 200, 300), each with the given `sort_order_id` set on its `DataFile`
+        /// (`None` leaves the field unset). Used to test how `sort_order_id` resolution
+        /// against the table's sort orders flows into each entry's `FileScanTask`.
+        pub async fn setup_manifest_files_with_sort_order_ids(
+            &mut self,
+            sort_order_ids: [Option<i32>; 4],
+        ) {
+            let current_snapshot = self.table.metadata().current_snapshot().unwrap();
+            let current_schema = current_snapshot.schema(self.table.metadata()).unwrap();
+            let current_partition_spec = self.table.metadata().default_partition_spec();
+            let parquet_file_size = self.write_parquet_data_files();
+
+            let mut writer = ManifestWriterBuilder::new(
+                self.next_manifest_file(),
+                Some(current_snapshot.snapshot_id()),
+                current_schema.clone(),
+                current_partition_spec.as_ref().clone(),
+            )
+            .build_v2_data();
+
+            for (i, sort_order_id) in sort_order_ids.into_iter().enumerate() {
+                let mut data_file_builder = DataFileBuilder::default();
+                data_file_builder
+                    .partition_spec_id(0)
+                    .content(DataContentType::Data)
+                    .file_path(format!("{}/{}.parquet", &self.table_location, i + 1))
+                    .file_format(DataFileFormat::Parquet)
+                    .file_size_in_bytes(parquet_file_size)
+                    .record_count(1)
+                    .partition(Struct::from_iter([Some(Literal::long(
+                        100 * (i as i64 + 1),
+                    ))]));
+                if let Some(id) = sort_order_id {
+                    data_file_builder.sort_order_id(id);
+                }
+                let data_file = data_file_builder.build().unwrap();
+
+                writer
+                    .add_entry(
+                        ManifestEntry::builder()
+                            .status(ManifestStatus::Added)
+                            .data_file(data_file)
+                            .build(),
+                    )
+                    .unwrap();
+            }
+
+            let data_file_manifest = writer.write_manifest_file().await.unwrap();
+
+            let manifest_list_writer = self
+                .table
+                .file_io()
+                .new_output(current_snapshot.manifest_list())
+                .unwrap()
+                .writer()
+                .await
+                .unwrap();
+            let mut manifest_list_write = ManifestListWriter::v2(
+                manifest_list_writer,
+                current_snapshot.snapshot_id(),
+                current_snapshot.parent_snapshot_id(),
+                current_snapshot.sequence_number(),
+            );
+            manifest_list_write
+                .add_manifests(std::iter::once(data_file_manifest))
+                .unwrap();
+            manifest_list_write.close().await.unwrap();
+        }
+
         /// Writes `mrg.parquet` with three 100-row row groups. Columns `x` (field
         /// id `1`) and `y` (field id `2`) both run 1000..1300, so row position `p`
         /// carries `x = y = 1000 + p`. Returns `(path, file_size_in_bytes)`.
@@ -1969,6 +2052,104 @@ pub mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn test_plan_files_carries_sort_order_into_file_scan_task() {
+        let mut fixture = TableTestFixture::new();
+
+        // Inject the reserved unsorted order (id 0) inline rather than editing the shared
+        // testdata fixture, so the id-0 file below exercises the `!is_unsorted()` filter
+        // branch instead of the `and_then` short-circuit an absent entry would take.
+        let mut metadata = fixture.table.metadata().clone();
+        metadata
+            .sort_orders
+            .insert(0, Arc::new(SortOrder::unsorted_order()));
+        fixture.table = fixture.table.with_metadata(Arc::new(metadata));
+
+        let expected_sort_order = fixture
+            .table
+            .metadata()
+            .sort_order_by_id(3)
+            .unwrap()
+            .clone();
+
+        // sort_order_ids: resolvable (3), absent, unresolvable (99), reserved unsorted (0).
+        fixture
+            .setup_manifest_files_with_sort_order_ids([Some(3), None, Some(99), Some(0)])
+            .await;
+
+        let tasks: Vec<_> = fixture
+            .table
+            .scan()
+            .build()
+            .unwrap()
+            .plan_files()
+            .await
+            .unwrap()
+            .try_collect()
+            .await
+            .unwrap();
+
+        assert_eq!(tasks.len(), 4, "expected all four FileScanTasks");
+
+        // Aggregates catch a systemic regression (every entry resolving to id 3, or
+        // resolution dropping entirely) that the per-file checks below would each still pass.
+        assert_eq!(
+            tasks.iter().filter(|t| t.sort_order().is_some()).count(),
+            1,
+            "exactly one file resolves to a sort order"
+        );
+        assert_eq!(
+            tasks.iter().filter(|t| t.sort_order_id().is_some()).count(),
+            3,
+            "three files carry a raw sort_order_id"
+        );
+
+        let resolved = tasks
+            .iter()
+            .find(|t| t.data_file_path().ends_with("1.parquet"))
+            .unwrap();
+        assert_eq!(resolved.sort_order_id(), Some(3));
+        assert_eq!(
+            resolved.sort_order(),
+            Some(&expected_sort_order),
+            "sort_order_id 3 should resolve to the table's sort order at id 3"
+        );
+
+        let missing = tasks
+            .iter()
+            .find(|t| t.data_file_path().ends_with("2.parquet"))
+            .unwrap();
+        assert_eq!(missing.sort_order_id(), None);
+        assert!(
+            missing.sort_order().is_none(),
+            "a file with no sort_order_id carries no sort_order"
+        );
+
+        let unresolvable = tasks
+            .iter()
+            .find(|t| t.data_file_path().ends_with("3.parquet"))
+            .unwrap();
+        assert_eq!(
+            unresolvable.sort_order_id(),
+            Some(99),
+            "the raw id is preserved even when it does not resolve"
+        );
+        assert!(
+            unresolvable.sort_order().is_none(),
+            "an unresolvable sort_order_id resolves to no sort_order"
+        );
+
+        let unsorted = tasks
+            .iter()
+            .find(|t| t.data_file_path().ends_with("4.parquet"))
+            .unwrap();
+        assert_eq!(unsorted.sort_order_id(), Some(0));
+        assert!(
+            unsorted.sort_order().is_none(),
+            "the reserved unsorted order (id 0) resolves to no sort_order"
+        );
+    }
+
     #[tokio::test]
     async fn test_plan_files_on_table_without_any_snapshots() {
         let table = TableTestFixture::new_empty().table;
@@ -2746,6 +2927,20 @@ pub mod tests {
                 .unwrap(),
         );
         let unified_partition_type = Arc::new(partition_spec.partition_type(&schema).unwrap());
+        let sort_order = Arc::new(
+            SortOrder::builder()
+                .with_order_id(1)
+                .with_sort_field(
+                    SortField::builder()
+                        .source_id(1)
+                        .transform(Transform::Identity)
+                        .direction(SortDirection::Ascending)
+                        .null_order(NullOrder::First)
+                        .build(),
+                )
+                .build(&schema)
+                .unwrap(),
+        );
         let task = FileScanTask::builder()
             .with_data_file_path("data_file_path".to_string())
             .with_file_size_in_bytes(123)
@@ -2777,6 +2972,8 @@ pub mod tests {
                 vec![],
             )]))))
             .with_unified_partition_type(Some(unified_partition_type))
+            .with_sort_order_id(Some(1))
+            .with_sort_order(Some(sort_order))
             .with_case_sensitive(true)
             .with_key_metadata(Some(vec![1, 2, 3].into_boxed_slice()))
             .build()