laskoviymishka commented on code in PR #3128:
URL: https://github.com/apache/iceberg-rust/pull/3128#discussion_r3972092775
##########
crates/iceberg/src/scan/context.rs:
##########
@@ -86,11 +88,19 @@ impl ManifestFileContext {
case_sensitive,
partition_spec,
unified_partition_type,
+ table_metadata,
} = self;
let manifest = object_cache.get_manifest(&manifest_file).await?;
for manifest_entry in manifest.entries() {
+ let sort_order = manifest_entry
+ .data_file()
+ .sort_order_id()
+ .and_then(|id| table_metadata.sort_order_by_id(id as i64))
Review Comment:
This resolves and then discards the raw `sort_order_id`, which collapses
three different situations into a single `None`: no id at all, the reserved
unsorted order, and an id that doesn't resolve against the table's sort orders.
For reading that's spec-compliant, but for the Part 2 sorted-scan work it's
the difference between "this file was never sorted" and "this file is
physically sorted but its order definition was dropped" — both arrive as `None`
and the optimizer can't tell them apart.
I'd carry the raw `sort_order_id: Option<i32>` on `FileScanTask` alongside
the resolved `sort_order` so a caller can still distinguish those cases. At
minimum, a `tracing::warn!` when a non-zero id fails to resolve would keep this
from silently degrading with no signal. wdyt?
##########
crates/iceberg/src/scan/context.rs:
##########
@@ -86,11 +88,19 @@ impl ManifestFileContext {
case_sensitive,
partition_spec,
unified_partition_type,
+ table_metadata,
} = self;
let manifest = object_cache.get_manifest(&manifest_file).await?;
for manifest_entry in manifest.entries() {
+ let sort_order = manifest_entry
+ .data_file()
+ .sort_order_id()
+ .and_then(|id| table_metadata.sort_order_by_id(id as i64))
+ .filter(|order| !order.is_unsorted())
Review Comment:
Small thing: `!order.is_unsorted()` keys off `fields.is_empty()` rather than
the order id, so a malformed or forward-compat order with a non-zero id and
empty fields would also be treated as unsorted. In a valid table that's
equivalent to `order_id == 0` and it matches Java's `isUnsorted()`, so it's
fine as-is.
The field doc over in `task.rs` says it resolves "to the reserved unsorted
order (id 0, per the spec)", which is a bit more specific than what the check
actually does — worth either softening the wording or comparing against
`SortOrder::UNSORTED_ORDER_ID` if we want the doc to be literally true.
##########
crates/iceberg/src/scan/mod.rs:
##########
@@ -1640,6 +1640,76 @@ pub mod tests {
manifest_list_write.close().await.unwrap();
}
+ /// Writes a manifest with four live "Added" data-file entries
(partitioned on `x`
+ /// = 100, 200, 300, 400), 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(vec![data_file_manifest].into_iter())
Review Comment:
`std::iter::once(data_file_manifest)` avoids the throwaway heap `Vec` here —
the sibling helpers use `vec![...]` because they genuinely have multiple
manifests, but this one only ever has the one.
##########
crates/iceberg/src/scan/task.rs:
##########
@@ -150,6 +150,22 @@ pub struct FileScanTask {
#[builder(default)]
pub unified_partition_type: Option<Arc<StructType>>,
+ /// The sort order that this file's rows are sorted by, resolved from the
data file's
+ /// `sort_order_id` against the table's known sort orders. `Some` only if
the id resolves
+ /// to a sort order with at least one field. `None` if the file has no
sort order id, the
+ /// id doesn't resolve against the table's sort orders, or it resolves to
the reserved
+ /// unsorted order (id 0, per the spec).
+ ///
+ /// Note: this reflects only the file's own recorded sort order id, not
necessarily the
+ /// table's current default sort order — see
[`crate::spec::DataFile::sort_order_id`].
+ /// Serde: not yet implemented.
+ #[serde(default)]
+ #[serde(skip_serializing_if = "Option::is_none")]
+ #[serde(serialize_with = "serialize_not_implemented")]
Review Comment:
This is the one I'd really want to settle before merge. Unlike `partition` /
`partition_spec` / `unified_partition_type`, `SortOrder` already derives
`Serialize`/`Deserialize` and gets round-tripped in table metadata JSON today —
so there's no technical reason for the `not_implemented` guard here.
The effect is that any `FileScanTask` whose `sort_order` is `Some(_)` errors
on serialize. `skip_serializing_if` covers the `None` case, but the moment Part
2 populates this field — which is the whole point of the PR — every task with a
resolved order silently fails serde on a public field.
I'd drop the `serialize_with`/`deserialize_with` guards and use plain
`#[serde(default, skip_serializing_if = "Option::is_none")]`, which just works
for a round-trippable `SortOrder`. If there's a wire-format reason to omit it,
I'd love a comment spelling that out specifically rather than the
`not_implemented` stub. wdyt?
##########
crates/iceberg/testdata/example_table_metadata_v2.json:
##########
@@ -41,6 +41,10 @@
"last-partition-id": 1000,
"default-sort-order-id": 3,
"sort-orders": [
+ {
+ "order-id": 0,
Review Comment:
Adding order-id 0 here works, but this fixture is shared by four test
modules (transaction/append, io/object_cache, the scan tests), and the change
flips `sort_order_by_id(0)` from `None` to `Some(unsorted)` for all of them — a
silent behavioral change for one test's benefit.
It also makes the file-4 assertion a bit of an illusion: without this entry,
`sort_order_by_id(0)` returns `None`, `and_then` short-circuits, and
`sort_order.is_none()` passes either way — the `!is_unsorted()` filter never
actually fires. And the more common Java-written case (id 0 with no explicit
order-0 entry) stays untested.
I'd either drop this change and add a dedicated fixture that omits order-id
0 (proving the real short-circuit path), or mutate the metadata clone inline in
the helper the way `new_unpartitioned` does, so the filter branch gets
exercised without touching a shared fixture. wdyt?
##########
crates/iceberg/src/scan/context.rs:
##########
@@ -50,6 +50,7 @@ pub(crate) struct ManifestFileContext {
case_sensitive: bool,
partition_spec: Option<PartitionSpecRef>,
unified_partition_type: Option<Arc<StructType>>,
+ table_metadata: TableMetadataRef,
Review Comment:
Carrying the whole `TableMetadataRef` into every `ManifestFileContext`
works, but it hands each context a much wider capability than it needs — the
existing `unified_partition_type` pattern precomputes the derived value in
`PlanContext` and carries only that.
Sort-order resolution is genuinely per-entry, so I get why full metadata is
tempting. But we could build an `Arc<HashMap<i64, SortOrderRef>>` once in
`create_manifest_file_context` and carry that instead, which keeps the context
narrow and matches the surrounding pattern. wdyt?
##########
crates/iceberg/src/scan/task.rs:
##########
@@ -150,6 +150,22 @@ pub struct FileScanTask {
#[builder(default)]
pub unified_partition_type: Option<Arc<StructType>>,
+ /// The sort order that this file's rows are sorted by, resolved from the
data file's
+ /// `sort_order_id` against the table's known sort orders. `Some` only if
the id resolves
+ /// to a sort order with at least one field. `None` if the file has no
sort order id, the
+ /// id doesn't resolve against the table's sort orders, or it resolves to
the reserved
+ /// unsorted order (id 0, per the spec).
+ ///
+ /// Note: this reflects only the file's own recorded sort order id, not
necessarily the
+ /// table's current default sort order — see
[`crate::spec::DataFile::sort_order_id`].
Review Comment:
The intra-doc link `crate::spec::DataFile::sort_order_id` points at a method
but omits the `()`, so rustdoc may resolve it as a field or emit a
`broken_intra_doc_links` warning — writing it as `sort_order_id()` fixes it.
##########
crates/iceberg/src/scan/mod.rs:
##########
@@ -1968,6 +2038,73 @@ pub mod tests {
}
}
+ #[tokio::test]
+ async fn test_plan_files_carries_sort_order_into_file_scan_task() {
+ let mut fixture = TableTestFixture::new();
+
+ let expected_sort_order = fixture
+ .table
+ .metadata()
+ .sort_order_by_id(3)
+ .unwrap()
+ .clone();
+
+ 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");
Review Comment:
The per-file assertions are thorough, but nothing asserts the aggregate — a
regression that resolved every entry to id 3, or dropped resolution entirely,
would still pass three of the four checks.
A single `assert_eq!(tasks.iter().filter(|t|
t.sort_order.is_some()).count(), 1)` up top would catch that class of systemic
regression cheaply.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
