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

AGENTS.md

@@ -73,6 +73,8 @@ Tests are self-contained (in-memory catalogs, temp dirs, checked-in metadata JSO
 
 `IcebergTableScan` converts the projection to column names and the filters to a single Iceberg `Predicate` (`physical_plan/expr_to_predicate.rs`). It applies `limit` in-stream, has one output partition, and delegates the actual read to iceberg's `TableScan::to_arrow()`.
 
+With the session option `iceberg.execution.preserve_data_ordering` (`config.rs`; off by default), `TableProvider::scan` lists the data files up front. If there are at most `iceberg.execution.max_merge_files_per_partition` of them and all record the same resolvable sort order, `IcebergDataSource` reports that order's leading fields as its output ordering and k-way merges one reader per file in `open()`. Only identity transforms of projected top-level columns are reported, stopping at the first float, double, or UUID field, because Arrow orders -0.0 and NaN differently from Spark-written files. Otherwise the source is built exactly as without the option. A source rebuilt with `IcebergDataSource::builder` must be given what `sorted_tasks()` returns through `with_sorted_tasks`, or it will read the files unordered under a plan that relies on their order. The providers build a sorted source the same way: `plan_sorted_tasks` lists the files, and they pass its result to `with_sorted_tasks`. The order is also not reported unless the provider's declared schema matches the snapshot schema the files are read with, since DataFusion ignores the null order of a column declared non-nullable.
+
 Both providers report every filter as `Inexact`, so DataFusion re-applies the original filters after the scan. The pushed-down predicate therefore only has to be *implied by* the original filter. It may match extra rows but must never drop a row that matches. Unconvertible parts are dropped: an `AND` keeps whichever side converted, while an `OR` needs both sides. Preserve this soundness rule when extending `expr_to_predicate.rs`; for example, it is why date casts and some NaN arithmetic are deliberately not pushed down.
 
 ### Write path (`IcebergTableProvider::insert_into`)
@@ -98,11 +100,12 @@ Library code returns `datafusion::error::Result`. Convert iceberg errors with `t
 ## sqllogictest harness (`crates/sqllogictest`)
 
 - It's a custom `harness = false` runner using libtest-mimic. Each TOML file in `testdata/schedules/` is one test: it declares engines and ordered `[[steps]]` that point at `.slt` files under `testdata/slts/`. An `.slt` file that no schedule references never runs.
-- Each schedule gets a fresh `SessionContext` (`target_partitions = 4`, information_schema enabled) and a fresh in-memory Iceberg catalog. The catalog is registered as `default` with namespace `default`, so SQL addresses tables as `default.default.<table>`. `datafusion` is the only engine type. The `catalog` setting in a schedule is parsed but ignored.
+- Each schedule gets a fresh `SessionContext` (`target_partitions = 4`, information_schema enabled, `IcebergDataFusionConfig` registered) and a fresh in-memory Iceberg catalog. The catalog is registered as `default` with namespace `default`, so SQL addresses tables as `default.default.<table>`. `datafusion` is the only engine type. The `catalog` setting in a schedule is parsed but ignored.
 - `engine/datafusion.rs` pre-creates the tables SQL can't express yet:
   - `test_partitioned_table`: identity-partitioned on `category`
   - `test_binary_table`
   - `test_encrypted_round_trip`: format V3, encrypted with in-memory KMS key `test-master-key`
+  - `test_sorted_table`: sorted by `id`, with two data files whose ids overlap, each recording that sort order
 
   All other tables are created by `CREATE TABLE` inside the `.slt` files.
 - Several `.slt` files assert `EXPLAIN` output, such as `IcebergTableScan projection:[...] predicate:[...] limit:[...]`. Changes to the scan's `DisplayAs`, to pushdown, or to the DataFusion version require updating those expectations by hand; the harness has no auto-complete mode.

crates/datafusion/src/config.rs

@@ -0,0 +1,92 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Session options for Iceberg tables, set with `SET iceberg.<group>.<option>`.
+
+use datafusion::common::config::ConfigExtension;
+use datafusion::common::extensions_options;
+
+extensions_options! {
+    /// Session options for reading Iceberg tables through DataFusion.
+    ///
+    /// `SET iceberg.<group>.<option> = <value>` only works once this extension
+    /// is registered on the session, with
+    /// [`SessionConfig::with_option_extension`](datafusion::prelude::SessionConfig::with_option_extension).
+    /// Without it, every option keeps its default. DataFusion lists these
+    /// options in `information_schema.df_settings` by their last name alone,
+    /// such as `max_merge_files_per_partition`, and `SHOW` and `RESET` do not
+    /// support them.
+    ///
+    /// ```
+    /// # #[tokio::main(flavor = "current_thread")]
+    /// # async fn main() -> datafusion::error::Result<()> {
+    /// use datafusion::prelude::{SessionConfig, SessionContext};
+    /// use iceberg_datafusion::IcebergDataFusionConfig;
+    ///
+    /// let config =
+    ///     SessionConfig::new().with_option_extension(IcebergDataFusionConfig::default());
+    /// let ctx = SessionContext::new_with_config(config);
+    /// ctx.sql("SET iceberg.execution.preserve_data_ordering = true")
+    ///     .await?
+    ///     .collect()
+    ///     .await?;
+    ///
+    /// let state = ctx.state();
+    /// let options = state.config().options();
+    /// let iceberg = options.extensions.get::<IcebergDataFusionConfig>().unwrap();
+    /// assert!(iceberg.execution.preserve_data_ordering);
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub struct IcebergDataFusionConfig {
+        /// Options for executing table scans.
+        pub execution: IcebergExecutionConfig, default = IcebergExecutionConfig::default()
+    }
+}
+
+extensions_options! {
+    /// Options for executing Iceberg table scans, under `iceberg.execution`.
+    pub struct IcebergExecutionConfig {
+        /// When true, a scan lists its data files while it is planned and, if
+        /// every file records the same sort order, reports that order to
+        /// DataFusion so that it can skip sorting the scan's output. The scan
+        /// then merges the files in that order, with every file open at once.
+        ///
+        /// Only the leading sort fields that are identity transforms of
+        /// projected top-level columns are reported, stopping at the first
+        /// floating point or UUID field, whose orders in Iceberg writers and in
+        /// DataFusion can differ.
+        ///
+        /// 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 sort the merged output again.
+        pub preserve_data_ordering: bool, default = false
+        /// The most data files a scan merges in one output partition to
+        /// preserve their sort order. A scan with more files in a partition
+        /// reports no order, and DataFusion sorts its output instead, which
+        /// can spill to disk where the merge cannot.
+        pub max_merge_files_per_partition: usize, default = 64
+    }
+}
+
+impl ConfigExtension for IcebergDataFusionConfig {
+    const PREFIX: &'static str = "iceberg";
+}

crates/datafusion/src/lib.rs

@@ -18,6 +18,9 @@
 mod catalog;
 pub use catalog::*;
 
+mod config;
+pub use config::{IcebergDataFusionConfig, IcebergExecutionConfig};
+
 mod error;
 pub use error::*;