You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by gi...@apache.org on 2026/09/14 17:14:56 UTC
(iceberg-rust) branch main updated: feat(iceberg): thread per-file sort order into FileScanTask (#3128)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
The following commit(s) were added to refs/heads/main by this push:
new fabb70707 feat(iceberg): thread per-file sort order into FileScanTask (#3128)
fabb70707 is described below
commit fabb707075303d85f7e1e4b1b45834c3ec7cf68c
Author: Anurag Mantripragada <an...@gmail.com>
AuthorDate: Mon Sep 14 17:04:03 2026 +0000
feat(iceberg): thread per-file sort order into FileScanTask (#3128)
Resolves each data file's sort_order_id against the table's known sort
orders during scan planning and carries the result on FileScanTask as
sort_order (Some only when genuinely sorted; the reserved unsorted
order, id 0, resolves to None) alongside the raw sort_order_id (kept
unresolved so a non-zero id that fails to resolve is still
recoverable as "physically sorted").
PlanContext precomputes a narrow sort-orders map once per scan,
matching the existing unified_partition_type pattern, rather than
carrying the full TableMetadataRef into every ManifestFileContext.
---
crates/iceberg/public-api.txt | 4 +-
crates/iceberg/src/scan/context.rs | 28 ++++-
crates/iceberg/src/scan/mod.rs | 205 ++++++++++++++++++++++++++++++++++++-
crates/iceberg/src/scan/task.rs | 48 ++++++++-
4 files changed, 276 insertions(+), 9 deletions(-)
diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt
index ca5580496..40987ef1d 100644
--- a/crates/iceberg/public-api.txt
+++ b/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
diff --git a/crates/iceberg/src/scan/context.rs b/crates/iceberg/src/scan/context.rs
index a598975b9..4635325fb 100644
--- a/crates/iceberg/src/scan/context.rs
+++ b/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(),
}
}
}
diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs
index db7080647..9bdf42beb 100644
--- a/crates/iceberg/src/scan/mod.rs
+++ b/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()
diff --git a/crates/iceberg/src/scan/task.rs b/crates/iceberg/src/scan/task.rs
index f1a799fcc..5fd0f81bb 100644
--- a/crates/iceberg/src/scan/task.rs
+++ b/crates/iceberg/src/scan/task.rs
@@ -24,7 +24,7 @@ use typed_builder::TypedBuilder;
use crate::expr::BoundPredicate;
use crate::spec::{
DataContentType, DataFileFormat, ManifestEntryRef, NameMapping, PartitionSpec, Schema,
- SchemaRef, Struct, StructType,
+ SchemaRef, SortOrderRef, Struct, StructType,
};
use crate::{Error, ErrorKind, Result};
@@ -116,6 +116,26 @@ pub struct FileScanTask {
#[builder(default)]
unified_partition_type: Option<Arc<StructType>>,
+ /// The raw `sort_order_id` recorded on the data file (spec field 140), carried through
+ /// unresolved. See [`DataFile::sort_order_id()`](crate::spec::DataFile::sort_order_id).
+ /// `None` if the file has no sort order id.
+ ///
+ /// This preserves the distinction that [`sort_order`](Self::sort_order) collapses: a
+ /// non-zero id that does not resolve against the table's sort orders still marks the file
+ /// as physically sorted, which a consumer such as the DataFusion sorted-scan optimizer
+ /// needs even when the order definition itself is unavailable.
+ #[builder(default)]
+ sort_order_id: Option<i32>,
+
+ /// The sort order that this file's rows are sorted by, resolved from the data file's
+ /// [`sort_order_id`](Self::sort_order_id) against the table's known sort orders. `Some`
+ /// only when the id resolves to an order that has sort fields. `None` if the file has no
+ /// sort order id, the id does not resolve, or it resolves to an order with no sort fields
+ /// (the reserved unsorted order, id 0 per the spec). Use [`sort_order_id`](Self::sort_order_id)
+ /// to tell those cases apart.
+ #[builder(default)]
+ sort_order: Option<SortOrderRef>,
+
/// Whether this scan task should treat column names as case-sensitive when binding predicates.
case_sensitive: bool,
@@ -217,6 +237,16 @@ impl FileScanTask {
self.unified_partition_type.as_ref()
}
+ /// Returns the raw sort order id recorded on the data file, unresolved.
+ pub fn sort_order_id(&self) -> Option<i32> {
+ self.sort_order_id
+ }
+
+ /// Returns the sort order this file's rows are sorted by, if resolved.
+ pub fn sort_order(&self) -> Option<&SortOrderRef> {
+ self.sort_order.as_ref()
+ }
+
/// Returns whether names are treated as case-sensitive.
pub fn case_sensitive(&self) -> bool {
self.case_sensitive
@@ -372,8 +402,8 @@ mod _serde {
use super::{FileScanTask, FileScanTaskDeleteFile};
use crate::expr::BoundPredicate;
use crate::spec::{
- DataFileFormat, Literal, NameMapping, PartitionSpec, RawLiteral, SchemaRef, StructType,
- Type,
+ DataFileFormat, Literal, NameMapping, PartitionSpec, RawLiteral, SchemaRef, SortOrderRef,
+ StructType, Type,
};
use crate::{Error, ErrorKind, Result};
@@ -399,6 +429,10 @@ mod _serde {
name_mapping: Option<Arc<NameMapping>>,
#[serde(default)]
unified_partition_type: Option<Arc<StructType>>,
+ #[serde(default)]
+ sort_order_id: Option<i32>,
+ #[serde(default)]
+ sort_order: Option<SortOrderRef>,
case_sensitive: bool,
#[serde(default)]
key_metadata: Option<Box<[u8]>>,
@@ -430,6 +464,10 @@ mod _serde {
name_mapping: Option<&'a Arc<NameMapping>>,
#[serde(skip_serializing_if = "Option::is_none")]
unified_partition_type: Option<&'a Arc<StructType>>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ sort_order_id: Option<i32>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ sort_order: Option<&'a SortOrderRef>,
case_sensitive: bool,
#[serde(skip_serializing_if = "Option::is_none")]
key_metadata: Option<&'a [u8]>,
@@ -477,6 +515,8 @@ mod _serde {
partition_spec: value.partition_spec.as_ref(),
name_mapping: value.name_mapping.as_ref(),
unified_partition_type: value.unified_partition_type.as_ref(),
+ sort_order_id: value.sort_order_id,
+ sort_order: value.sort_order.as_ref(),
case_sensitive: value.case_sensitive,
key_metadata: value.key_metadata.as_deref(),
})
@@ -528,6 +568,8 @@ mod _serde {
.with_partition_spec(value.partition_spec)
.with_name_mapping(value.name_mapping)
.with_unified_partition_type(value.unified_partition_type)
+ .with_sort_order_id(value.sort_order_id)
+ .with_sort_order(value.sort_order)
.with_case_sensitive(value.case_sensitive)
.with_key_metadata(value.key_metadata)
.build()