Skip to content

feat: declare sort orders and surface them to DataFusion for attested files - #391

Merged
JanKaul merged 2 commits into
JanKaul:mainfrom
cedricziel:feat/declared-sort-orders
Aug 20, 2026
Merged

feat: declare sort orders and surface them to DataFusion for attested files#391
JanKaul merged 2 commits into
JanKaul:mainfrom
cedricziel:feat/declared-sort-orders

Conversation

@cedricziel

Copy link
Copy Markdown
Contributor

Problem

A table's declared sort order never reached the data it describes, and readers had no way to find out that a file was sorted:

  • CreateTable stored the declared write order under DEFAULT_SORT_ORDER_ID (0, the spec's reserved "unsorted" id), so a table created with a sort order reported the unsorted order as its default.
  • Nothing wrote Parquet sorting_columns or set a DataFile's sort_order_id, so a sorted file and an unsorted file were indistinguishable in the manifest.
  • datafusion_iceberg therefore reported ordering: None for every file, and DataFusion had to re-sort data that was already sorted on disk, with no TopK or sort pushdown available.

Approach

Declare the order, attest it per file, and only claim it for files that carry the attestation.

Declaration (iceberg-rust)

  • CreateTable keeps the declared write order under its own order_id and makes it the default. A fielded order using the reserved id 0 is rejected.
  • TableMetadata::default_sort_order() and TableTransaction::replace_sort_order() declare an order on an existing table (AddSortOrder + SetDefaultSortOrder, guarded by AssertDefaultSortOrderId).

Attestation (iceberg-rust)

  • New write_sorted_parquet_partitioned(): the caller promises the batch stream is sorted by the table's default sort order; every written file records it as per-row-group sorting_columns plus an iceberg.sort-order-id footer entry, and parquet_to_datafile lifts that footer entry into the manifest entry's sort_order_id. write_parquet_partitioned() is unchanged and keeps writing unattested files, so a producer that cannot guarantee the order stays honest by default.
  • sorting_columns are all-or-nothing: they are only emitted when every sort field is an identity transform on a top-level primitive column, because a partial list would claim a different (prefix) order than the one the file honors.

Consumption (datafusion_iceberg)

  • The scan derives a DataFusion ordering from the table's default sort order (leading run of identity-transformed fields only) and splits data files by whether their manifest entry attests that order id. Attested files go into a scan carrying output_ordering; unattested files go into a scan that claims nothing. A table holding any unattested file therefore keeps its explicit sort — the failure mode of a false claim is silently mis-ordered results, so the classification is conservative by construction.
  • When split_file_groups_by_statistics is enabled, the attested groups are regrouped into non-overlapping, statistics-ordered groups the way DataFusion's listing table does. Falling back to the partition-shaped groups is always safe, since DataFusion re-validates the declared ordering against each group's file statistics at plan time.

Tests

  • iceberg-rust/tests/sort_order_test.rs (6 tests): a table created with a sort order reports it as default; replace_sort_order declares one on an existing table and rejects the reserved id; a sorted write attests the order on every file (footer plus sort_order_id); unsorted writes and tables without a declared order attest nothing.
  • datafusion_iceberg/tests/sort_order_scan.rs (3 tests): a fully attested table declares output_ordering and elides the sort; one unattested file keeps the sort and the results stay exact; attested-but-overlapping files also keep the sort and stay exact, with and without split_file_groups_by_statistics.

All green locally, along with the make test-* suites.

A table's default sort order was stored under the reserved unsorted id and
never reached the files: nothing wrote Parquet sorting_columns or the
manifest entry's sort_order_id, so readers could not tell sorted files
from unsorted ones.

- CreateTable keeps the declared write order under its own order id and
  makes it the default; a fielded order with id 0 is rejected.
- TableMetadata::default_sort_order() and
  TableTransaction::replace_sort_order() declare an order on an existing
  table (AddSortOrder + SetDefaultSortOrder, guarded by
  AssertDefaultSortOrderId).
- write_sorted_parquet_partitioned() writes rows the caller has sorted by
  the default order and attests it per file: row-group sorting_columns
  for identity fields, an iceberg.sort-order-id footer entry, and the
  DataFile sort_order_id lifted from that footer by parquet_to_datafile.
  write_parquet_partitioned() keeps writing unattested files.
The provider reported ordering: None for every file, so a scan of files
written under a declared sort order still had to be re-sorted and could not
feed TopK pushdown.

The scan now derives a DataFusion ordering from the table's default sort
order and splits data files by whether their manifest entry attests that
order id. Attested files go into a scan that declares the ordering (and,
when split_file_groups_by_statistics is on, gets regrouped into
non-overlapping statistics-ordered groups); unattested files go into a scan
that claims nothing. A table holding any unattested file therefore keeps its
explicit sort, and only the leading run of identity-transformed sort fields
is ever claimed.
@cedricziel

Copy link
Copy Markdown
Contributor Author

Self-review notes, recorded so the follow-up is specific rather than remembered. Queued for the next free build slot on our side; happy to fold any of it in if you would rather see it before merge.

1. The two SortOrder validity checks are implemented twice.
iceberg-rust/src/catalog/create.rs (in CreateTableBuilder::create) and iceberg-rust/src/table/transaction/operation.rs (in the ReplaceSortOrder arm) each independently do the same two things: reject order_id == DEFAULT_SORT_ORDER_ID when the order has fields, and check that every sort field's source_id exists in the current schema. They differ only in error wording — create.rs names the table, operation.rs does not — and in which error variant the first check returns.

The natural fix is one SortOrder::validate_against_schema(&self, schema: &Schema) -> Result<(), Error> living next to SortOrder and DEFAULT_SORT_ORDER_ID in iceberg-rust-spec/src/spec/sort.rs, called from both sites. Roughly 25 lines of near-duplicate logic collapse, and a future fix to one copy can no longer miss the other. Behaviour-preserving.

2. Those checks scan where a lookup exists.
Both write schema.fields().iter().any(|f| f.id == source_id), a linear scan, while StructType::get(id) already does the same lookup through its lookup map — and this PR uses get correctly in arrow/write.rs and in datafusion_iceberg. The scan matches the shape of the pre-existing partition-field validation just above it in create.rs, so it is at least locally consistent, but it makes the PR inconsistent with itself. Worth folding into the same change as #1, since it is the same code.

3. Optional readability nit.
declared_output_ordering in datafusion_iceberg/src/table/mod.rs chains .take_while(identity transform) then .map(…) .map_while(|expr| expr) — two short-circuit conditions expressing one rule. A single map_while that returns None for both a non-identity transform and an unresolvable column would say "stop at the first field we cannot claim" once, matching the doc comment above it. No behaviour change; only worth doing if that file is being touched anyway.

None of these change behaviour, and CI is green as it stands.

@JanKaul

JanKaul commented Aug 20, 2026

Copy link
Copy Markdown
Owner

As always, great work. I also like the approach. Thanks!

@JanKaul
JanKaul merged commit ce2d3a1 into JanKaul:main Aug 20, 2026
2 checks passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants