feat: report the sort order of sorted Iceberg scans to DataFusion

[anuragmantri]

Which issue does this PR close?

• Part 2 of [Epic] Sort-order reporting for iceberg-datafusion scans iceberg-rust#3126. Part 1 (per-file sort order on FileScanTask) merged in feat(iceberg): thread per-file sort order into FileScanTask iceberg-rust#3128 and is included in the pinned iceberg-rust rev.

What changes are included in this PR?

IcebergDataSource can now report an output ordering to DataFusion when the data files it reads are already sorted, so DataFusion can drop the SortExec it would otherwise add above the scan for an ORDER BY or a sort-merge join.

The feature is off by default. It is enabled per session with SET iceberg.execution.preserve_data_ordering = true, after registering the new IcebergDataFusionConfig extension. The playground and the sqllogictest engine register it. The option name follows Spark's spark.sql.iceberg.planning.preserve-data-ordering, which apache/iceberg#16750 proposes, and sits in DataFusion's execution group. DataFusion lists extension options in information_schema.df_settings by their last name only, and SHOW and RESET don't support them.

When it is on, TableProvider::scan lists the data files up front. The source reports an ordering only if:

there are at most iceberg.execution.max_merge_files_per_partition files (default 64, matching Comet's sortMerge.maxFilesPerPartition in apache/datafusion-comet#5331),

• every file records the same sort_order_id, and it resolves to a sorted order, and
• every output column has the same type in the provider's declared schema as in the snapshot schema the files are read with, and is declared nullable wherever the snapshot column is optional. A catalog-backed provider caches its schema, so the two can differ, and DataFusion ignores the null order of a column it believes is non-nullable.

It reports the leading fields of that order that are identity transforms of projected top-level columns, stopping at the first float, double, or UUID field. DataFusion sorts floats in the order the Iceberg spec defines, with -0.0 before 0.0 and negative NaN first, but Spark, the usual writer of sorted files, treats the zeros as equal and every NaN as the largest value, so the files it writes may not be sorted in the spec's order. The spec doesn't define UUID ordering (apache/iceberg#14216). In every other case the source is planned exactly as before.

Two files with the same sort order can have overlapping key ranges, so open() reads each file as its own stream and k-way merges them with DataFusion's StreamingMergeBuilder, which is the machinery SortPreservingMergeExec uses. A pushed-down limit becomes the merge's fetch. All the readers are clones of one ArrowReader, so they share its delete-file cache.

The scan reports each sort field's null order as the table declares it. Writers commonly sort ascending fields with nulls first, the default of Iceberg Java's SortOrderBuilder.asc, while DataFusion's ORDER BY c puts them last. On a nullable column sorted that way, only ORDER BY c NULLS FIRST, or a session with datafusion.sql_parser.default_null_ordering = 'nulls_min', can skip its sort. Other queries still sort, above the merge. TableProvider::scan can't tell whether anything above the scan needs the order, so it merges whenever it reports one. Merging only when something above uses the order is tracked in #72.

The source keeps a single output partition, so this doesn't depend on apache/iceberg-rust#2671, and it composes with it. DataFusion's output ordering is per partition, so once apache/iceberg-rust#2671 groups files into partitions, each group can be merged the same way.

Public API changes:

New: IcebergDataFusionConfig and IcebergExecutionConfig, the session options.

• New: IcebergDataSource::sorted_tasks, the files a sorted source merges. IcebergDataSource::builder takes them as a new sorted_tasks argument, which changes the signature from refactor: replace IcebergDataSource::new_with_predicate with a builder #63 before its first release. Since feat: make the plan nodes inspectable and rebuildable #20, a source can be rebuilt from its accessors, for example to ship it to another process. A sorted source rebuilt without its files would read them unordered beneath a plan that already dropped the sort, so a rebuild must pass them, and build() fails if they don't share an order the source can report. Table providers pass None.

When the option is on but a scan falls back, the files are listed again in open(), as without the option. The table's manifest cache makes the second listing cheap.

Merging opens every file at once. The cap bounds that. Bounding it instead by how many files' key ranges overlap needs per-file column bounds on FileScanTask, and is tracked in #71.

Are these changes tested?

Yes.

• Unit tests for mapping a sort order to a DataFusion ordering:
  • direction and null order,
  • stopping at bucket, truncate, and day transforms,
  • stopping at float, double, UUID, nested, unprojected, and missing fields.
• Integration tests over tables with hand-written sorted files:
  • files with overlapping key ranges and duplicate keys merge into sorted order, with struct and list columns carried through;
  • files longer than a reader batch merge correctly across batch boundaries;
  • LIMIT;
  • no ordering when files disagree, have no sort order id, or have the unsorted order, and the plan then matches the one without the option;
  • no ordering when the declared schema is stale, after a column is made optional or its type is promoted;
  • the option is off by default, and the cap is honored;
  • ORDER BY loses its SortExec, while ORDER BY ... DESC and the option-off case keep it;
  • on a nullable sort key, ORDER BY id NULLS FIRST loses its SortExec, while ORDER BY id keeps it;
  • a sort-merge join with target_partitions = 4 has no SortExec;
  • a rebuilt source keeps its ordering when given its sorted tasks, and reports none when given None;
  • positional deletes are applied on the merge path.
  • a sorted scan of one file reports the order and returns its rows, and a sorted table with no files reports no order;
  • a rebuild fails when tasks share a sort order id but record different orders;
• sqllogictest: sort_order_reporting.slt checks the EXPLAIN plans over a pre-created sorted table: the option off and on, a descending sort, a table created through SQL, and the file cap.
• Mutation checks:
  • Replacing the merge with concatenation fails 8 of the 14 sorted-scan integration tests.
  • Truncating each file's stream to its first batch fails the batch-boundary test.
  • Dropping the check that every file has the same sort order fails the disagreement test.
  • Dropping the schema check fails both stale-schema tests.
  • Making the builder ignore its sorted_tasks argument fails the rebuilt-source test and the positional-delete test.
  • Making the single-file path return no rows fails the one-file test.
  • Comparing sort order ids instead of the orders fails the different-orders test.

AI Disclosure

I'm new to Rust. I used Claude Code (Claude Opus 5.5) to write this change and its tests, and I reviewed it manually.

[mbutrovich]

Nice, I'll take a look! Thanks @anuragmantri!

[andygrove]

Thanks @anuragmantri, this is careful work, and making it opt-in is the right call. I found one problem I think should be fixed before merging, plus some smaller points. Line numbers are at cb0414f.

1. A mismatch between the declared and snapshot schemas gives wrong results or a panic (should fix)

The reported order and the merge both trust the scan's declared schema, but the rows come from the snapshot schema. The two can differ: the catalog-backed provider caches its schema (#24), and the providers declare the current schema while iceberg-rust reads the snapshot schema (#42). With the option on, I reproduced two failures (repro below):

• Wrong NULL placement. A provider is created while id is NOT NULL. Another engine then makes id nullable and writes one sorted file containing [NULL, 0, 7]. With the option on, SELECT id FROM t ORDER BY id drops its sort and returns NULL, 0, 7. With it off, the result is the correct 0, 7, NULL. The cause is that DataFusion ignores NULLS FIRST/LAST on columns declared non-nullable (options_compatible in datafusion-physical-expr-common's sort_expr.rs). So it treats the scan's id ASC NULLS FIRST as satisfying ORDER BY id, which means ASC NULLS LAST. With two or more files, both modes already fail with "declared as non-nullable but contains null values" (#24). The single-file case is the one that becomes silently wrong.
• Panic. Another engine promotes id from int to long without writing data. A provider created afterwards declares Int64, but the files are read as Int32. With two sorted files and the option on, both SELECT * and ORDER BY id panic in DataFusion's merge cursor (sorts/stream.rs:264, expect("field values")). With it off, SELECT * returns the rows and ORDER BY returns an error.

A guard in shared_ordering (scan.rs:547) fixes both. It reports no order unless each output column has the same type as in the snapshot schema, and is nullable wherever the snapshot column is optional:

let read = iceberg::arrow::schema_to_arrow_schema(first.schema()).ok()?;
if !output.fields().iter().all(|field| {
    read.field_with_name(field.name()).is_ok_and(|read| {
        read.data_type() == field.data_type()
            && (field.is_nullable() || !read.is_nullable())
    })
}) {
    return None;
}

With it, both repro tests pass, and so do the PR's own tests (8 integration, 6 unit). Because it lives in shared_ordering, it also covers with_sorted_tasks.

Repro: two tests that fail on this branch

// Repro for apache/datafusion-iceberg#55: the scan's declared schema can differ
// from the snapshot schema it reads (#24, #42). Drop into
// crates/datafusion/tests/ and run with `cargo test --test <file name>`.

use std::collections::HashMap;
use std::str::FromStr;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};

use datafusion::arrow::array::{AsArray, Int32Array, RecordBatch, StringArray};
use datafusion::arrow::datatypes::Int32Type;
use datafusion::parquet::arrow::ArrowWriter;
use datafusion::prelude::{SessionConfig, SessionContext};
use datafusion_iceberg::{IcebergDataFusionConfig, IcebergTableProvider};
use iceberg::arrow::schema_to_arrow_schema;
use iceberg::io::LocalFsStorageFactory;
use iceberg::memory::{MEMORY_CATALOG_WAREHOUSE, MemoryCatalogBuilder};
use iceberg::spec::{
    DataContentType, DataFileBuilder, DataFileFormat, NestedField, NullOrder,
    PrimitiveType, Schema, SortDirection, SortField, SortOrder, Struct, Transform, Type,
};
use iceberg::table::Table;
use iceberg::transaction::{ApplyTransactionAction, Transaction};
use iceberg::{
    Catalog, CatalogBuilder, MetadataLocation, NamespaceIdent, TableCreation, TableIdent,
};

static FILES: AtomicUsize = AtomicUsize::new(0);

fn schema(id_type: PrimitiveType, id_required: bool) -> Schema {
    let id = if id_required {
        NestedField::required(1, "id", Type::Primitive(id_type))
    } else {
        NestedField::optional(1, "id", Type::Primitive(id_type))
    };
    Schema::builder()
        .with_fields(vec![
            id.into(),
            NestedField::required(2, "data", Type::Primitive(PrimitiveType::String))
                .into(),
        ])
        .build()
        .unwrap()
}

/// Creates table `ns.t` with a required int `id`, sorted by `id` ASC NULLS FIRST.
async fn create_table(dir: &std::path::Path) -> (Arc<dyn Catalog>, NamespaceIdent, Table) {
    let warehouse = dir.display().to_string();
    let catalog = MemoryCatalogBuilder::default()
        .with_storage_factory(Arc::new(LocalFsStorageFactory))
        .load(
            "memory",
            HashMap::from([(MEMORY_CATALOG_WAREHOUSE.to_string(), warehouse)]),
        )
        .await
        .unwrap();
    let ns = NamespaceIdent::new("ns".to_string());
    catalog.create_namespace(&ns, HashMap::new()).await.unwrap();
    let sort_order = SortOrder::builder()
        .with_sort_field(
            SortField::builder()
                .source_id(1)
                .transform(Transform::Identity)
                .direction(SortDirection::Ascending)
                .null_order(NullOrder::First)
                .build(),
        )
        .build_unbound()
        .unwrap();
    let creation = TableCreation::builder()
        .location(dir.join("t").display().to_string())
        .name("t".to_string())
        .schema(schema(PrimitiveType::Int, true))
        .sort_order(sort_order)
        .build();
    let table = catalog.create_table(&ns, creation).await.unwrap();
    (Arc::new(catalog), ns, table)
}

/// Commits one data file per entry of `files`, each already sorted by `id` and
/// recording sort order 1.
async fn append(
    catalog: &Arc<dyn Catalog>,
    table: &Table,
    files: &[&[Option<i32>]],
) -> Table {
    let arrow_schema =
        Arc::new(schema_to_arrow_schema(table.metadata().current_schema()).unwrap());
    let data_dir = format!("{}/data", table.metadata().location());
    std::fs::create_dir_all(&data_dir).unwrap();
    let mut data_files = Vec::new();
    for ids in files {
        let path = format!("{data_dir}/{}.parquet", FILES.fetch_add(1, Ordering::SeqCst));
        let batch = RecordBatch::try_new(arrow_schema.clone(), vec![
            Arc::new(Int32Array::from(ids.to_vec())),
            Arc::new(StringArray::from_iter_values(
                ids.iter().map(|id| format!("{id:?}")),
            )),
        ])
        .unwrap();
        let file = std::fs::File::create(&path).unwrap();
        let mut writer = ArrowWriter::try_new(file, arrow_schema.clone(), None).unwrap();
        writer.write(&batch).unwrap();
        writer.close().unwrap();
        let mut builder = DataFileBuilder::default();
        builder
            .content(DataContentType::Data)
            .file_path(path.clone())
            .file_format(DataFileFormat::Parquet)
            .file_size_in_bytes(std::fs::metadata(&path).unwrap().len())
            .record_count(ids.len() as u64)
            .partition_spec_id(0)
            .partition(Struct::empty())
            .sort_order_id(1);
        data_files.push(builder.build().unwrap());
    }
    let tx = Transaction::new(table);
    let tx = tx.fast_append().add_data_files(data_files).apply(tx).unwrap();
    tx.commit(catalog.as_ref()).await.unwrap()
}

/// Stands in for another engine committing `new_schema` without writing data.
/// iceberg-rust has no action that makes a column optional or promotes its
/// type, so this rewrites the table's current metadata file in place.
async fn commit_schema_externally(table: &Table, new_schema: Schema) {
    let metadata = table
        .metadata()
        .clone()
        .into_builder(None)
        .add_current_schema(new_schema)
        .unwrap()
        .build()
        .unwrap()
        .metadata;
    let location = MetadataLocation::from_str(table.metadata_location().unwrap()).unwrap();
    metadata.write_to(table.file_io(), &location).await.unwrap();
}

async fn session(preserve: bool) -> SessionContext {
    let config = SessionConfig::new()
        .with_target_partitions(1)
        .with_option_extension(IcebergDataFusionConfig::default());
    let ctx = SessionContext::new_with_config(config);
    let set = format!("SET iceberg.planning.preserve_data_ordering = {preserve}");
    ctx.sql(&set).await.unwrap().collect().await.unwrap();
    ctx
}

/// A provider created while `id` is required caches it as non-nullable. Another
/// engine then makes `id` optional and writes a sorted file containing a NULL.
/// DataFusion ignores the null order of non-nullable columns, so the scan's
/// `id ASC NULLS FIRST` satisfies `ORDER BY id` (NULLS LAST) and the sort is
/// dropped.
#[tokio::test(flavor = "multi_thread")]
async fn order_by_after_column_made_optional() {
    let dir = tempfile::tempdir().unwrap();
    let (catalog, ns, table) = create_table(dir.path()).await;
    let provider = IcebergTableProvider::try_new(catalog.clone(), ns.clone(), "t")
        .await
        .unwrap();

    commit_schema_externally(&table, schema(PrimitiveType::Int, false)).await;
    let table = catalog
        .load_table(&TableIdent::new(ns, "t".to_string()))
        .await
        .unwrap();
    append(&catalog, &table, &[&[None, Some(0), Some(7)]]).await;

    let provider = Arc::new(provider);
    for preserve in [false, true] {
        let ctx = session(preserve).await;
        ctx.register_table("t", provider.clone()).unwrap();
        let batches = ctx
            .sql("SELECT id FROM t ORDER BY id")
            .await
            .unwrap()
            .collect()
            .await
            .unwrap();
        let ids: Vec<Option<i32>> = batches
            .iter()
            .flat_map(|batch| batch.column(0).as_primitive::<Int32Type>().iter())
            .collect();
        // Fails with the option on: [None, Some(0), Some(7)].
        assert_eq!(
            ids,
            vec![Some(0), Some(7), None],
            "preserve_data_ordering = {preserve}"
        );
    }
}

/// Another engine promotes `id` from int to long without writing data. A
/// provider created afterwards declares Int64, while the scan reads the
/// snapshot schema's Int32.
#[tokio::test(flavor = "multi_thread")]
async fn select_after_type_promotion() {
    let dir = tempfile::tempdir().unwrap();
    let (catalog, ns, table) = create_table(dir.path()).await;
    let table = append(&catalog, &table, &[&[Some(1), Some(3)], &[Some(2), Some(4)]]).await;
    commit_schema_externally(&table, schema(PrimitiveType::Long, true)).await;
    let provider = IcebergTableProvider::try_new(catalog.clone(), ns, "t")
        .await
        .unwrap();

    let provider = Arc::new(provider);
    for preserve in [false, true] {
        let ctx = session(preserve).await;
        ctx.register_table("t", provider.clone()).unwrap();
        // Panics with the option on, in DataFusion's merge: "field values".
        let batches = ctx.sql("SELECT * FROM t").await.unwrap().collect().await.unwrap();
        let rows: usize = batches.iter().map(RecordBatch::num_rows).sum();
        assert_eq!(rows, 4, "preserve_data_ordering = {preserve}");
    }
}

2. A rebuilt scan can silently lose its order

A codec written against the accessors from #20 still compiles after this PR. Once a user turns the option on, that codec rebuilds the scan without its sorted tasks, under a plan whose SortExec has already been removed. Sort-merge joins, sorted aggregations and ORDER BY then get unsorted rows with no error. The rustdoc example on new_with_predicate (scan.rs:164-183) doesn't show the with_sorted_tasks step, and codec authors will copy it. The crate hasn't been released yet, so I'd make the sorted tasks a parameter of new_with_predicate rather than a separate method. Then the compiler makes every codec handle them.

3. The merge runs even when nothing needs the order

Whenever the leading sort column is projected, a scan of up to max_merge_files files becomes a single-threaded merge with every file open, whether or not anything above it uses the order. Iceberg and Spark default ascending sort fields to NULLS FIRST, but DataFusion's ORDER BY id means NULLS LAST. So on a nullable sort column, ORDER BY id keeps its SortExec on top of the merging scan and pays for both. Only ORDER BY id NULLS FIRST, or datafusion.sql_parser.default_null_ordering = 'nulls_min', benefits. The tests all use a NOT NULL id, so this case isn't covered. I'd add a test with a nullable key and mention this in the option's docs.

Smaller points

• with_data_file_concurrency_limit(1) (scan.rs:356) isn't needed for ordering, since each reader already gets exactly one file. It also makes the reader load that file's delete files one at a time.
• Neither the playground REPL (crates/playground/src/main.rs:83) nor the sqllogictest engine (crates/sqllogictest/src/engine/datafusion.rs:68) registers IcebergDataFusionConfig, so SET iceberg.planning.… fails in both. Registering it in the sqllogictest engine would also allow EXPLAIN-based .slt coverage.
• DataFusion 55 lists extension options in information_schema.df_settings without their prefix (max_merge_files, not iceberg.planning.max_merge_files), and SHOW and RESET fail for them. That's a DataFusion limitation, but worth a sentence in the docs.
• apache/iceberg#16750, which adds the Spark option this mirrors, is still an open PR.

What I checked and found sound

• The type allowlist, and stopping at transforms, unprojected columns and nested fields. Sort fields are resolved by field ID against the snapshot schema.
• One ArrowReader per file keeps the file's row order, and the clones share the delete-file cache.
• iceberg-rust's writers never set sort_order_id on data files, so this crate's own INSERTs can't produce files that falsely claim to be sorted.
• The planning-time listing is capped with take(max_merge_files + 1), and the fallback re-listing hits the same table's manifest cache.

[anuragmantri]

Thanks @andygrove for a thorough review, and the repros. I've addressed your comments.

For 3. Merging when nothing needs the order., I agree. test_sorted_scan_on_nullable_key shows ORDER BY id NULLS FIRST dropping its sort while ORDER BY id keeps it above the merge, and the preserve_data_ordering docs now describe this, including datafusion.sql_parser.default_null_ordering = 'nulls_min'. TableProvider::scan can't tell whether anything above the scan needs the order, so I'd rather look at DataFusion's sort pushdown as a follow-up than grow this PR if that works for you.

This is ready for another round of review.

[mbutrovich]

First pass, thanks @anuragmantri!

[mbutrovich]

I'd rather look at DataFusion's sort pushdown as a follow-up than grow this PR if that works for you.

Could you file an issue for the follow-up and link it from the PR description? The issue should cover more than sort pushdown. DataFusion 55's PushdownSort rule offers a SortExec's order to the plan below it through ExecutionPlan::try_pushdown_sort, and drops the sort when the input answers Exact. That would cover ORDER BY, where the plan is SortExec over CooperativeExec over the scan. It wouldn't cover the sort-merge join that test_sorted_scan_removes_sort_from_sort_merge_join checks. With no ordering reported, each join input plans as SortExec(preserve_partitioning) over RepartitionExec: partitioning=Hash([id@0], 4), and RepartitionExec::try_pushdown_sort returns Unsupported when the repartition doesn't keep its input's order. A design that merges only on pushdown would lose the join case, so the issue should record that.

[anuragmantri]

Filed #72. It covers more than sort pushdown and records the sort-merge join constraint you described, with the plans measured with and without a WHERE on a join side.

[mbutrovich]

Bounding it instead by how many files' key ranges overlap would need per-file column bounds on FileScanTask, which is follow-up work.

Could you file an issue for this one too? It needs an iceberg-rust change first, so the issue can track both sides.

[anuragmantri]

Filed #71. The iceberg-rust change comes first, and I will link it from there.

[mbutrovich]

As I read the spec's sort order section, -NaN < -Infinity < -value < -0 < 0 < value < Infinity < NaN is the order writers should produce, and Arrow sorts floats in that order. So the files this guards against come from writers like Spark that don't follow the spec here. Should the comment say so? As written, it reads as though DataFusion's order is the one that differs. Someone deciding later whether to lift the guard would need to know that it depends on the writers.

Suggested change

	/// Whether DataFusion orders values of `field_type` as the engines that write
	/// sorted Iceberg files do. It orders -0.0 before 0.0 and negative NaN before
	/// every number, where Spark treats the zeros as equal and NaN as the largest
	/// value, and the Iceberg spec leaves the order of UUIDs undefined
	/// (apache/iceberg#14216).
	/// Whether DataFusion orders values of `field_type` as the engines that write
	/// sorted Iceberg files do. For floating point, DataFusion uses the order the
	/// spec defines (<https://iceberg.apache.org/spec/#sort-orders>), with -0.0
	/// before 0.0 and negative NaN first. Spark treats the zeros as equal and
	/// every NaN as the largest value, so files it writes may not be sorted in the
	/// spec's order. The spec leaves the order of UUIDs undefined
	/// (apache/iceberg#14216).

[anuragmantri]

Applied your suggestion in ab16901, thanks.

[mbutrovich]

Each file in the sorted-scan tests holds at most four rows, so every per-file stream yields a single batch, and the merge never advances a stream past its first batch. What do you think about making one of these files longer than a reader batch, or writing it with several row groups, so that a test covers per-file streams that yield more than one batch? The order within each stream comes from ArrowReader, not from DataFusion's merge, and no test exercises it across batch boundaries.

[anuragmantri]

Added test_sorted_scan_merges_files_longer_than_a_batch in ab16901. It merges two files of 3,000 interleaved ids each, and first checks that each file's stream yields more than one batch. Truncating each per-file stream to its first batch fails it.

[anuragmantri]

Thanks @mbutrovich for the review, and @andygrove again.

I rebased onto main after the read path moved to IcebergDataSource (the DataSource refactor and #63).

One open question : with #63's builder, the sorted tasks are now with_sorted_tasks(...), validated in build(). That means the compiler can no longer force a codec that uses the builder to pass them, which was the point of andygrove's suggestion. Is the setter plus its docs enough, or should they become an argument to IcebergDataSource::builder?

[anuragmantri]

Filed #72. It covers more than sort pushdown and records the sort-merge join constraint you described, with the plans measured with and without a WHERE on a join side.

[anuragmantri]

Filed #71. The iceberg-rust change comes first, and I will link it from there.

[anuragmantri]

Applied your suggestion in ab16901, thanks.

[anuragmantri]

Added test_sorted_scan_merges_files_longer_than_a_batch in ab16901. It merges two files of 3,000 interleaved ids each, and first checks that each file's stream yields more than one batch. Truncating each per-file stream to its first batch fails it.

[andygrove]

Thanks @anuragmantri, this addresses everything from my last review. I re-reviewed at ab16901:

• The schema guard landed as reads_as_declared, and both stale-schema tests fail without it.
• The REPL and the sqllogictest engine register the extension, and the df_settings limitation is documented.
• The nullable-key test and the option's docs cover the NULLS FIRST case, and #71 and #72 track the follow-ups.
• with_data_file_concurrency_limit(1) is gone, and reader_builder()'s defaults match TableScan::to_arrow's.

The unit, integration, and sqllogictest suites pass locally. I also tried a table identity-partitioned on category and sorted by (category, id). The scan reports category@1 ASC, id@0 ASC, the sort is dropped, and the rows come back in order.

On your question about the builder: I'd make the sorted tasks an argument of IcebergDataSource::builder. A codec written against #63's builder still compiles after this PR and silently drops the order. I tried it with SELECT id FROM t ORDER BY id over three sorted files: rebuilding the scan as #63's docs describe, without with_sorted_tasks, returns [1, 4, 7, 2, 5, 8, 3, 6, 9] with no error. The builder hasn't been released yet, so changing its signature is free now and a breaking change after 0.12.0. The cost is a None in the two providers. I'd keep the check in build() as well.

Rebase: this conflicts with main in integration_datafusion_test.rs after #64. #64 also removed expect-test from the dev-dependencies, so the expect! in test_sorted_scan_merges_files_with_overlapping_ranges needs to become an insta snapshot.

With those two changes, this looks good to merge from my side.

[anuragmantri]

Thanks @andygrove, and for trying the partitioned (category, id) table. Both points are done.

• The sorted tasks are now the third argument of IcebergDataSource::builder, and with_sorted_tasks is gone (0ffdbbc). The providers pass None, and build() still rejects tasks that don't share a sort order the source can report. A test covers a rebuilt source: given its tasks it keeps the order, and given None it reports none.
• Rebased onto main after #64, which is a force-push. The expect! snapshot is now an insta snapshot. The earlier commits keep their content, and ab16901 is now c6542e4.

@NoahKusaba, this changes the signature of #63's builder. Please let me know if that is a problem.

[andygrove]

Thanks @anuragmantri, both changes look good. I re-reviewed at 0ffdbbc:

• The range-diff against ab16901 only shows the expect! becoming an insta snapshot, so the rest is what I reviewed last time.
• builder() now takes the sorted tasks, and both providers and every test pass them explicitly. A codec written against #63's two-argument builder no longer compiles, which was the point. The new assertions in test_rebuilt_sorted_scan_keeps_order cover the None case.

One leftover, suggested inline: AGENTS.md still says a rebuilt source gets its tasks "through with_sorted_tasks", which no longer exists.

Approving, with that sentence fixed before merge.

[mbutrovich]

Taking a look this afternoon.

[mbutrovich]

Thanks @anuragmantri! A few more questions inline. The ones I'd most like to settle before merge are the option names, since renaming them after a release breaks users' SET statements.

[mbutrovich]

Should the group follow DataFusion's layout? DataFusion puts its options under fixed groups (ConfigOptions: catalog, execution, optimizer, sql_parser, explain, format), and none of them is planning. That name comes from Spark's spark.sql.iceberg.planning.*.

The DataFusion option closest to this one is split_file_groups_by_statistics, which also arranges files so that a sort can be removed. It lives under execution, next to the Parquet reader options in execution.parquet. This crate may later expose iceberg-rust's reader settings as session options (ArrowReaderBuilder: data_file_concurrency_limit, row_group_filtering_enabled, and so on). If so, iceberg.execution.* with iceberg-rust's names could hold those too. What do you think about iceberg.execution.preserve_data_ordering?

[anuragmantri]

Agreed, the group is now execution in 6a0c8f1, so the options are iceberg.execution.preserve_data_ordering and iceberg.execution.max_merge_files_per_partition. The struct is renamed IcebergExecutionConfig. Following DataFusion's layout makes more sense here than following Spark's, and it leaves room for iceberg-rust's reader settings.

[mbutrovich]

Could this be named for what it caps once the scan has more than one partition? The PR description says that once apache/iceberg-rust#2671 groups files into partitions, each group can be merged the same way, and from then on this cap applies per partition. apache/datafusion-comet#5331 names the same limit, with the same default of 64, sortMerge.maxFilesPerPartition. Spark's in apache/iceberg#16750 is preserve-data-ordering.max-files-per-task. max_files_per_partition would match Comet and DataFusion's own use of "partition", and it would stay accurate after apache/iceberg-rust#2671 without a rename.

[anuragmantri]

Agreed on naming it per partition, so it stays accurate after apache/iceberg-rust#2671. I kept "merge" in it, though: max_merge_files_per_partition. Under iceberg.execution, a bare max_files_per_partition reads like a general limit on how files are grouped into partitions, which is the kind of option apache/iceberg-rust#2671 might add itself. Happy to drop "merge" if you prefer Comet's exact name.

[anuragmantri]

Thanks @mbutrovich, these are all good catches, and settling the names before a release makes sense. Everything is in 6a0c8f1, on top of the earlier commits.

Thanks @andygrove for the approval. The AGENTS.md sentence is fixed in 6370be9.

[anuragmantri]

Agreed, the group is now execution in 6a0c8f1, so the options are iceberg.execution.preserve_data_ordering and iceberg.execution.max_merge_files_per_partition. The struct is renamed IcebergExecutionConfig. Following DataFusion's layout makes more sense here than following Spark's, and it leaves room for iceberg-rust's reader settings.

[anuragmantri]

Agreed on naming it per partition, so it stays accurate after apache/iceberg-rust#2671. I kept "merge" in it, though: max_merge_files_per_partition. Under iceberg.execution, a bare max_files_per_partition reads like a general limit on how files are grouped into partitions, which is the kind of option apache/iceberg-rust#2671 might add itself. Happy to drop "merge" if you prefer Comet's exact name.

[mbutrovich]

Thanks @anuragmantri, 6a0c8f1 addresses everything from my last round.

Could you update the PR description for the renames in 6a0c8f1? It still says SET iceberg.planning.preserve_data_ordering, iceberg.planning.max_merge_files, and IcebergPlanningConfig. It also still says "Iceberg sorts ascending fields with nulls first by default", which you mentioned you'd fix along with the rustdoc.

[anuragmantri]

Could you update the PR description for the renames in 6a0c8f1?

Thanks @mbutrovich - I updated the description, I will also change the descriptions of follow-up PRs, that refer the old option name.