<!---
  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.
-->

# Upgrade Guides

## DataFusion 55.0.0

**Note:** DataFusion `55.0.0` has not been released yet. The information provided
in this section pertains to features and changes that have already been merged
to the main branch and are awaiting release in this version.

### `DataFrame::fill_null` now borrows its arguments

`DataFrame::fill_null` previously took its arguments by value:

```rust,ignore
// Before
pub fn fill_null(
    &self,
    value: ScalarValue,
    columns: Vec<String>,
) -> Result<DataFrame>
```

It now borrows them, matching the signature of the newly added
`DataFrame::fill_nan`:

```rust,ignore
// After
pub fn fill_null(
    &self,
    value: &ScalarValue,
    columns: &[&str],
) -> Result<DataFrame>
```

This lets callers pass a borrowed `ScalarValue` and slice literals (or
`&str` column names) without first allocating owned `String`s.

**Migration guide:**

Borrow the value and pass a slice of `&str` instead of an owned `Vec<String>`:

```rust,ignore
// Before
let df = df.fill_null(ScalarValue::from(0), vec!["a".to_owned(), "c".to_owned()])?;
let df = df.fill_null(ScalarValue::from(0), vec![])?;

// After
let df = df.fill_null(&ScalarValue::from(0), &["a", "c"])?;
let df = df.fill_null(&ScalarValue::from(0), &[])?;
```

### `FileScanConfig::partitioned_by_file_group` removed

`FileScanConfig::partitioned_by_file_group` and
`FileScanConfigBuilder::with_partitioned_by_file_group(...)` have been removed.
Use `FileScanConfig::output_partitioning` and
`FileScanConfigBuilder::with_output_partitioning(...)` instead.
The corresponding
`datafusion_proto::protobuf::FileScanExecConf::partitioned_by_file_group`
field has also been removed.

**Who is affected:**

- Users who accessed `FileScanConfig::partitioned_by_file_group` directly.
- Users who called
  `FileScanConfigBuilder::with_partitioned_by_file_group(true)`.
- Users who constructed or accessed
  `datafusion_proto::protobuf::FileScanExecConf::partitioned_by_file_group`.

**Migration guide:**

If your file groups are organized by table partition column values, declare hash
output partitioning over those partition columns:

```rust,ignore
use datafusion_datasource::file_scan_config::{
    FileScanConfigBuilder, output_partitioning_from_partition_fields,
};

let output_partitioning = output_partitioning_from_partition_fields(
    source.table_schema().table_schema(),
    source.table_schema().table_partition_cols(),
    file_groups.len(),
);

let config = FileScanConfigBuilder::new(object_store_url, source)
    .with_file_groups(file_groups)
    .with_output_partitioning(output_partitioning)
    .build();
```

`output_partitioning_from_partition_fields` returns
`Some(Partitioning::Hash(...))` when partition columns are present and `None`
otherwise. If you construct the partitioning manually, pass
`Some(Partitioning::Hash(partition_exprs, partition_count))` to
`with_output_partitioning(...)`.

When constructing `FileScanExecConf`, omit `partitioned_by_file_group` and set
`output_partitioning` instead.

### User `SpillFile` traits instead of [`RefCountedTempFile`]

Spill file APIs now use the `datafusion_execution::SpillFile` trait instead of
the concrete [`RefCountedTempFile`] type. [`DiskManager::create_tmp_file`] now
returns `Arc<dyn SpillFile>`.
This change was introduced in [PR #21882], which adds pluggable spill file
backends via `SpillFile` and `TempFileFactory`.

If your code matched on [`DiskManagerMode`], add a `DiskManagerMode::Custom(_)`
arm.

If your code wrote directly to a [`RefCountedTempFile`] or called
[`RefCountedTempFile::update_disk_usage`], open a spill writer instead:

```diff
- temp_file.inner().as_file().write_all(bytes)?;
- temp_file.update_disk_usage()?;
+ temp_file.open_writer()?.write_all(bytes)?;
```

Use `temp_file.size()` instead of [`RefCountedTempFile::current_disk_usage`].

[`diskmanager::create_tmp_file`]: https://docs.rs/datafusion-execution/latest/datafusion_execution/disk_manager/struct.DiskManager.html#method.create_tmp_file
[`diskmanagermode`]: https://docs.rs/datafusion-execution/latest/datafusion_execution/disk_manager/enum.DiskManagerMode.html
[`pr #21882`]: https://github.com/apache/datafusion/pull/21882
[`refcountedtempfile`]: https://docs.rs/datafusion-execution/latest/datafusion_execution/disk_manager/struct.RefCountedTempFile.html
[`refcountedtempfile::current_disk_usage`]: https://docs.rs/datafusion-execution/latest/datafusion_execution/disk_manager/struct.RefCountedTempFile.html#method.current_disk_usage
[`refcountedtempfile::update_disk_usage`]: https://docs.rs/datafusion-execution/latest/datafusion_execution/disk_manager/struct.RefCountedTempFile.html#method.update_disk_usage

### `Dialect::AVAILABLE` replaced by `Dialect::available()`

`datafusion_common::config::Dialect::AVAILABLE` has been removed. Use
`Dialect::available()` instead.

### `spill_record_batch_by_size` removed

`datafusion_physical_plan::spill::spill_record_batch_by_size` has been removed.
This function was deprecated in DataFusion `46.0.0`.

Use `datafusion_physical_plan::spill::SpillManager::spill_record_batch_by_size`
instead.

### `CreateExternalTable` supports multiple locations

`CREATE EXTERNAL TABLE` now accepts multiple paths in a single `LOCATION`
clause, which are read together as one table:

```sql
CREATE EXTERNAL TABLE hits
STORED AS PARQUET
LOCATION ('file_1.parquet', 'file_2.parquet');
```

To support this, the `location` field of both
`datafusion_expr::CreateExternalTable` and
`datafusion_sql::parser::CreateExternalTable` changed from a `String` to a
`Vec<String>` named `locations`:

```rust
// Before (54.0.0)
let location: String = create_external_table.location;

// After (55.0.0)
let locations: Vec<String> = create_external_table.locations;
```

The `CreateExternalTable::builder(name, location, file_type, schema)`
constructor is unchanged and still takes a single location; use the new
`CreateExternalTableBuilder::with_locations(Vec<String>)` to set more than one.
All listed locations must resolve to the same schema and reside on the same
object store. A plain string literal remains a single location, so paths that
contain commas continue to work, for example `LOCATION 'path/with,comma.csv'`.

### Decimal scalar formatting uses human-readable values

Decimal scalar literals in `EXPLAIN` output, expression display strings, and
auto-generated column names now format the decimal value using its scale while
still showing the precision and scale. For example, a `Decimal128` literal with
stored value `1`, precision `1`, and scale `1` is now rendered as
`Decimal128(0.1,1,1)` instead of `Decimal128(Some(1),1,1)`. When formatting a
`ScalarValue` directly, it now appears as `0.1` instead of `Some(1),1,1`.

`NULL` decimal literals were previously shown as `Decimal128(None,10,2)`; they
will now appear as `Decimal128(NULL,10,2)`.

Query result values already used human-readable decimal formatting and are
unchanged.

### `Coercion` supports dictionary encoding preservation

`datafusion_expr_common::signature::Coercion` now supports optional dictionary
encoding preservation. Typed coercions materialize dictionary inputs by
default, including both `TypeSignatureClass::Native(...)` and broader classes
such as `Integer`, `Numeric`, and `Binary`. When preservation is enabled,
DataFusion instead coerces dictionary inputs to
`Dictionary(original_key_type, coerced_value_type)` instead of materializing them
to the coerced value type.

User-defined functions can opt in by setting dictionary encoding preservation on
the relevant coercion:

```rust
Coercion::new_exact(TypeSignatureClass::Native(logical_string()))
    .with_encoding_preservation(EncodingPreservation::dictionary())
```

This changes the coerced argument type passed to the function. If a function
derives its return type from that coerced argument type, code that checks exact
result types may need to update its expectations or add an explicit cast to
materialize the result.

This changes the previous behavior of typed non-native classes such as
`Integer` and `Binary`, which retained the physical dictionary type by default.
UDFs relying on that behavior must now explicitly enable dictionary
preservation. `TypeSignatureClass::Any` is unaffected.

### `GroupsAccumulator::merge_batch` no longer takes `opt_filter`

The `opt_filter` argument has been removed from
`datafusion_expr_common::groups_accumulator::GroupsAccumulator::merge_batch`:

```diff
  fn merge_batch(
      &mut self,
      values: &[ArrayRef],
      group_indices: &[usize],
-     opt_filter: Option<&BooleanArray>,
      total_num_groups: usize,
  ) -> Result<()>;
```

Aggregate `FILTER` clauses only apply to raw input rows during the partial
(update) phase, so by the time intermediate states are merged there is nothing
left to filter per row. In practice `opt_filter` was always `None` here, so
removing it makes the API self-explanatory and impossible to misuse.

**Who is affected:**

- Anyone with a custom `GroupsAccumulator` implementation.
- Anyone calling `merge_batch` directly.

**Migration guide:**

Drop the `opt_filter` argument from your `merge_batch` signature and from any
call sites:

```diff
  fn merge_batch(
      &mut self,
      values: &[ArrayRef],
      group_indices: &[usize],
-     opt_filter: Option<&BooleanArray>,
      total_num_groups: usize,
  ) -> Result<()> {
      // ...
  }
```

```diff
- acc.merge_batch(values, group_indices, None, total_num_groups)?;
+ acc.merge_batch(values, group_indices, total_num_groups)?;
```

If your implementation previously inspected `opt_filter` (for example asserting
it was `None`), that code can simply be deleted.

See [issue #22775](https://github.com/apache/datafusion/issues/22775) for details.

### `GroupsAccumulator::convert_to_state` is now required

`datafusion_expr_common::groups_accumulator::GroupsAccumulator::convert_to_state`
no longer provides a default implementation, and the
`GroupsAccumulator::supports_convert_to_state` capability method has been
removed. All `GroupsAccumulator` implementations must now support converting
input batches directly to intermediate aggregate state.

**Who is affected:**

- Users with custom `GroupsAccumulator` implementations.
- FFI providers and consumers that use `FFI_GroupsAccumulator`.

**Migration guide:**

Custom `GroupsAccumulator` implementations must now provide their own
`convert_to_state` implementation.

Delete `supports_convert_to_state` implementations because `convert_to_state`
is now required:

```diff
- fn supports_convert_to_state(&self) -> bool {
-     true
- }
```

The `supports_convert_to_state` field has also been removed from
`datafusion_ffi::udaf::groups_accumulator::FFI_GroupsAccumulator`, changing its
ABI layout. Rebuild both FFI providers and consumers against DataFusion 55, and
do not exchange this struct with libraries built against older major versions.

See [issue #23081](https://github.com/apache/datafusion/issues/23081) for details.

### `is_dynamic_physical_expr` is deprecated

`datafusion_physical_expr_common::physical_expr::is_dynamic_physical_expr` is
deprecated. It was a thin wrapper over `snapshot_generation(expr) != 0` used to
ask "does this predicate contain a dynamic filter?".

Prefer asking the question directly against the concrete type. For a one-off
check, downcast to `DynamicFilterPhysicalExpr`:

```rust
use datafusion_physical_expr::expressions::DynamicFilterPhysicalExpr;
use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion};

let mut is_dynamic = false;
predicate.apply(|e| {
    if e.downcast_ref::<DynamicFilterPhysicalExpr>().is_some() {
        is_dynamic = true;
        Ok(TreeNodeRecursion::Stop)
    } else {
        Ok(TreeNodeRecursion::Continue)
    }
})?;
```

If you also need to know whether the dynamic filters can still change (and to be
notified when they do), use the new `DynamicFilterTracking` /
`DynamicFilterTracker` API in `datafusion_physical_expr`:

```rust
use datafusion_physical_expr::DynamicFilterTracking;

let tracking = DynamicFilterTracking::classify(&predicate);
if tracking.contains_dynamic_filter() {
    // worth re-evaluating the predicate at runtime
}
```

### `PruningPredicate::try_new` is deprecated

`datafusion_pruning::PruningPredicate::try_new` is deprecated. Use
`PruningPredicateBuilder` instead. The deprecated constructor remains available
in DataFusion 55 and preserves its existing behavior.

```rust
// Before
let predicate = PruningPredicate::try_new(expr, schema)?;

// After
let predicate = PruningPredicateBuilder::new()
    .with_file_schema(schema)
    .try_build(expr)?;
```

### `FilePruner::try_new` no longer builds a pruner for static predicates without statistics

`datafusion_pruning::FilePruner::try_new` now returns `None` when the predicate
is purely static _and_ the file carries no usable column statistics, because
such a pruner can never prune anything beyond what planning already did.
Previously it returned `Some` whenever a statistics struct was present (the
"is this worth pruning?" decision lived in the Parquet opener). Files with column
statistics, and predicates that carry a dynamic filter, are unaffected.

### `QueryPlanner` adds `Any` as a supertrait

To enable downcasting of `dyn QueryPlanner` to concrete query planner types (via
`is::<T>()` / `downcast_ref::<T>()`), the `QueryPlanner` trait now has `Any`
as a supertrait:

```diff
- pub trait QueryPlanner: Debug
+ pub trait QueryPlanner: Any + Debug
```

### `ExecutionPlan::partition_statistics` deprecated in favor of `statistics_from_inputs`

`ExecutionPlan::partition_statistics` is deprecated. Statistics computation is
now split into two parts:

- `StatisticsContext` owns the bottom-up plan-tree traversal and a per-walk
  cache of memoized child statistics. Call `StatisticsContext::compute` to
  obtain statistics for a plan.
- `ExecutionPlan::statistics_from_inputs` computes a node's statistics from its
  children's already-resolved statistics, which the context passes in. The node
  does not traverse the tree itself.

Existing implementations of `partition_statistics` continue to work unchanged.
The default `statistics_from_inputs` delegates to the deprecated method, so no
migration is required until the deprecated method is removed.

> **Warning:** The delegation is **one-way**: the default `statistics_from_inputs`
> calls `partition_statistics`, but the default `partition_statistics` does
> **not** call `statistics_from_inputs` — it returns `Statistics::new_unknown`.
> Nodes that override only `statistics_from_inputs` will silently return
> `Statistics::new_unknown` to any caller still using the deprecated
> `partition_statistics`.

**Who is affected:**

- Users who implement custom `ExecutionPlan` nodes (recommended to migrate)
- Users who call `partition_statistics` directly (recommended to switch to `StatisticsContext::compute`)

**Migration guide:**

For **implementations**, override `statistics_from_inputs` instead of
`partition_statistics`, plus `child_stats_requests` to declare which children to
resolve. Child statistics then arrive pre-computed in `input_stats` (one entry per
child, in `children()` order), so the node only expresses its local propagation
logic. Leaf nodes, and nodes that derive their statistics without reading children,
need neither override (the default `child_stats_requests` skips every child).

```rust,ignore
// Before:
fn partition_statistics(&self, partition: Option<usize>) -> Result<Arc<Statistics>> {
    let child_stats = self.input.partition_statistics(partition)?;
    // ... transform child_stats ...
}

// After: declare the child to resolve, then compute from its statistics.
fn child_stats_requests(&self, partition: Option<usize>) -> Vec<ChildStats> {
    vec![ChildStats::At(partition)]
}

fn statistics_from_inputs(
    &self,
    input_stats: &[Arc<Statistics>],
    args: &StatisticsArgs,
) -> Result<Arc<Statistics>> {
    let child_stats = Arc::clone(&input_stats[0]);
    // ... transform child_stats ...
}
```

> **Important:** the default `child_stats_requests` skips every child, so a node that
> reads `input_stats` must override it to declare the children it uses, or those slots
> are filled with `Statistics::new_unknown` placeholders. Request a child with
> `ChildStats::At(partition)` (`None` = overall) and omit one with `ChildStats::Skip`.
> For example, a partition-merging operator requests `ChildStats::At(None)`, and a
> broadcast join requests its build side at `None`.

For **callers**, walk a plan through `StatisticsContext::compute`. The cache is
created with the context:

```rust,ignore
use datafusion_physical_plan::{StatisticsArgs, StatisticsContext};

// Before:
let stats = plan.partition_statistics(None)?;

// After:
let stats = StatisticsContext::new().compute(plan.as_ref(), &StatisticsArgs::new())?;
```

### `DdlStatement::CreateExternalTable` and `CreateFunction` are now boxed

The two largest variants of `datafusion_expr::DdlStatement` are now
`Box`ed:

```rust,ignore
// Before
pub enum DdlStatement {
    CreateExternalTable(CreateExternalTable),
    // ...
    CreateFunction(CreateFunction),
    // ...
}

// After
pub enum DdlStatement {
    CreateExternalTable(Box<CreateExternalTable>),
    // ...
    CreateFunction(Box<CreateFunction>),
    // ...
}
```

`CreateExternalTable` is 312 bytes and `CreateFunction` is 288 bytes, so
without boxing they forced the entire `LogicalPlan` enum to 320 bytes
even on SELECT-only query paths that never instantiate them. Boxing
shrinks `LogicalPlan` from 320 → 176 bytes (−45%), making every
`mem::take` / `mem::swap` / `Arc<LogicalPlan>` store on the planning
hot path move a smaller payload.

**Who is affected:**

- Users who construct `DdlStatement::CreateExternalTable(...)` or
  `DdlStatement::CreateFunction(...)` from an owned struct.
- Users who pattern-match these variants and destructure the inner
  struct in the same pattern (e.g.
  `DdlStatement::CreateExternalTable(CreateExternalTable { name, .. })`).
- Code that consumes the inner struct out of these variants (e.g. to
  pass `CreateExternalTable` by value to another function).

**Migration guide:**

When constructing the variants, wrap the inner struct in `Box::new`:

```rust,ignore
// Before
let stmt = DdlStatement::CreateFunction(CreateFunction { name, args, .. });

// After
let stmt = DdlStatement::CreateFunction(Box::new(CreateFunction {
    name,
    args,
    ..
}));
```

When pattern-matching, bind the boxed value and either access fields
through it (Rust auto-derefs the `Box`) or destructure via `.as_ref()`:

```rust,ignore
// Before
match ddl {
    DdlStatement::CreateExternalTable(CreateExternalTable {
        name, location, ..
    }) => { /* use name, location */ }
}

// After — access fields through the box
match ddl {
    DdlStatement::CreateExternalTable(ce) => {
        let name = &ce.name;
        let location = &ce.location;
        /* ... */
    }
}

// After — destructure the dereferenced struct
match ddl {
    DdlStatement::CreateExternalTable(ce) => {
        let CreateExternalTable { name, location, .. } = ce.as_ref();
        /* ... */
    }
}
```

When you need an owned `CreateExternalTable` / `CreateFunction` out of
the variant, dereference the box with `*`:

```rust,ignore
// Before
match plan {
    LogicalPlan::Ddl(DdlStatement::CreateExternalTable(cmd)) => Ok(cmd),
    _ => { /* ... */ }
}

// After
match plan {
    LogicalPlan::Ddl(DdlStatement::CreateExternalTable(cmd)) => Ok(*cmd),
    _ => { /* ... */ }
}
```

See [PR #22733](https://github.com/apache/datafusion/pull/22733) for
details, including the per-variant size breakdown and benchmark
results.

### `ListingOptions::target_partitions` and `collect_stat` removed

The `target_partitions` and `collect_stat` fields on
`datafusion_catalog_listing::ListingOptions`, their builder methods
(`with_target_partitions`, `with_collect_stat`), and the
`with_session_config_options` helper have been removed.

`ListingTable` now reads both values directly from the active `SessionConfig`
at scan time instead of from a copy snapshotted onto the table at construction
time.

**Who is affected:**

- Code that set `target_partitions` / `collect_stat` per table via
  `ListingOptions`, or read those public fields.
- Code that relied on a `ListingTable` freezing these values at construction
  time independently of the session config. The table now always reflects the
  current `SessionConfig`.

**Migration guide:**

Configure these on the `SessionConfig` instead:

```rust,ignore
// Before
let options = ListingOptions::new(format)
    .with_target_partitions(8)
    .with_collect_stat(true);

// After
let config = SessionConfig::new()
    .with_target_partitions(8)
    .with_collect_statistics(true);
```

See [PR #22969](https://github.com/apache/datafusion/pull/22969) for details.

### Spark map functions now reject duplicate keys by default

The Spark-compatibility map-construction functions (`map_from_arrays`,
`map_from_entries`, `str_to_map`) now raise `[DUPLICATED_MAP_KEY]` at runtime
when constructing a map that contains duplicate keys. This matches the default
of Spark's [`spark.sql.mapKeyDedupPolicy`](https://github.com/apache/spark/blob/v4.0.0/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala#L4502-L4511).

A new config option, `datafusion.spark.map_key_dedup_policy`, controls the
behavior:

- `EXCEPTION` (default): raise on any duplicate key.
- `LAST_WIN`: keep the last occurrence of each duplicate key. The key stays at
  its first-seen position with the value from its last occurrence (matching
  Spark's `ArrayBasedMapBuilder`).

**Who is affected:**

- Queries calling `map_from_arrays` or `str_to_map` on data that contains
  duplicate keys. Previously these functions either tolerated duplicates
  silently or raised a non-configurable error.

**Migration guide:**

To restore lenient duplicate-key handling, set the policy to `LAST_WIN`:

```sql
SET datafusion.spark.map_key_dedup_policy = 'LAST_WIN';
```

See [PR #21720](https://github.com/apache/datafusion/pull/21720) for details.

### Unify LRU memory-limiting caches into one generic cache

The caches `DefaultFileMetadataCache`, `DefaultListFilesCache` and `DefaultFileStatisticsCache`
are merged into one generic implementation `DefaultCache`. The corresponding traits are now
type aliases:

```diff
- pub trait FileStatisticsCache: CacheAccessor<TableScopedPath, CachedFileMetadata>
- pub trait ListFilesCache: CacheAccessor<TableScopedPath, CachedFileList>
- pub trait FileMetadataCache: CacheAccessor<Path, CachedFileMetadataEntry>
+ pub type FileStatisticsCache = dyn Cache<TableScopedPath, CachedFileMetadata>;
+ pub type ListFilesCache = dyn Cache<TableScopedPath, CachedFileList>;
+ pub type FileMetadataCache = dyn Cache<Path, CachedFileMetadataEntry>;
```

**Who is affected:**

- Users who introduced their own implementation of `FileMetadataCache`, `ListFilesCache` or `FileStatisticsCache`.

**Migration guide:**

Implement the newly introduced types for your custom cache implementation.

See [PR #22613](https://github.com/apache/datafusion/pull/22613) for details.

### `CachedFileMetadata` now validates file schema

The file-statistics cache remains keyed by `TableScopedPath`, but
`CachedFileMetadata` now stores a `SchemaFingerprint` of the `file_schema` used
to compute the cached statistics. Cache hits are valid only when both the file
metadata and schema fingerprint match.

**Who is affected:**

- Users constructing `CachedFileMetadata` values directly.

**Migration guide:**

- Pass `Arc::new(SchemaFingerprint::from_schema(file_schema))` to
  `CachedFileMetadata::new`.
- Pass the current schema fingerprint to `CachedFileMetadata::is_valid_for`.

See [PR #23201](https://github.com/apache/datafusion/pull/23201) for details.

### `EmptyExecNode` and `PlaceholderRowExecNode` gained a `partitions` field

The generated protobuf structs `EmptyExecNode` and `PlaceholderRowExecNode`
encoded only a schema, so the partition count set by `EmptyExec::with_partitions`
was silently dropped when a physical plan was serialized and deserialized: a plan
that reported `n` partitions before encoding reported `1` after. Both messages now
carry a `partitions` field that round-trips the count.

**Who is affected:**

- Users constructing `EmptyExecNode` or `PlaceholderRowExecNode` with an
  exhaustive struct literal.

**Migration guide:**

Set the new field, or fill it from `Default`:

```rust,ignore
// Before
EmptyExecNode { schema: Some(schema) }

// After
EmptyExecNode { schema: Some(schema), partitions: 4 }
// or
EmptyExecNode { schema: Some(schema), ..Default::default() }
```

The wire format stays compatible in both directions. Plans encoded before this
field existed decode as a single partition, the previous default, and plans
encoded after it add a field that older readers ignore.

See [PR #23643](https://github.com/apache/datafusion/pull/23643) for details.

### `time ± interval` now returns a `time` instead of an `interval`

Adding or subtracting an `interval` to/from a `time` value now returns a `time`
that wraps within the 24-hour clock, matching PostgreSQL and DuckDB. Previously
DataFusion returned an `interval`.

```sql
-- 55.0.0 onwards: returns a time
SELECT time '23:30:00' + interval '2 hours';
-- 01:30:00
```

Only the sub-day portion of the interval affects the result; whole days and
months are ignored, as in PostgreSQL. The result keeps the input time's unit
(mirroring `timestamp + interval`), and any interval precision finer than that
unit is truncated -- so `time(s) + interval '1 nanosecond'` is a no-op.

See [PR #23279](https://github.com/apache/datafusion/pull/23279) for details.

### Physical-planning state moved to an explicit `PhysicalPlanningContext`

The `subquery_indexes` and `subquery_results` public fields on
`datafusion_expr::execution_props::ExecutionProps` have been removed. They were
added in `54.0.0` as the channel through which the physical planner passed
uncorrelated scalar-subquery state to functions that create physical
`Arc<dyn PhysicalExpr>` values from logical `Expr` values.

The `lambda_variable_qualifier` public field and the
`with_qualified_lambda_variables` method on `ExecutionProps` have been removed
for the same reason: they carried the qualifiers of the lambda variables in
scope while `create_physical_expr` descended into a lambda body.

That state is now carried by a dedicated
`datafusion_expr::physical_planning_context::PhysicalPlanningContext` passed explicitly
through functions and planner traits. Unlike `ExecutionProps`, which applies
throughout the planning of an entire query, this context is scoped to the
logical plan subtree currently being converted. This removes the need for the
physical planner to clone and mutate a `SessionState`, is a prerequisite for
letting the planner take `&dyn Session`, and lets `ExtensionPlanner`
implementations create physical
expressions containing scalar subqueries against the same subquery state as the
rest of the plan.

The following functions take a new trailing
`planning_ctx: &PhysicalPlanningContext` parameter:

- `datafusion_physical_expr::create_physical_expr` / `create_physical_exprs`
- `datafusion_physical_expr::create_physical_sort_expr` /
  `create_physical_sort_exprs` / `create_physical_partitioning`
- `datafusion::physical_planner::create_window_expr` /
  `create_window_expr_with_name`
- `datafusion_physical_expr::aggregate::LoweredAggregateBuilder::new`

The planner traits changed accordingly:

- `PhysicalPlanner::create_physical_expr` takes
  `planning_ctx: &PhysicalPlanningContext`
- `ExtensionPlanner::plan_extension` and `plan_table_scan` receive
  `planning_ctx: &PhysicalPlanningContext` and should forward it to
  `PhysicalPlanner::create_physical_expr` when creating physical expressions

Convenience methods such as `SessionContext::create_physical_expr` and
`SessionState::create_physical_expr` are unchanged.

**Who is affected:**

- Code calling the functions above: pass
  `&PhysicalPlanningContext::default()` unless you are creating physical
  expressions as part of a physical plan that contains uncorrelated scalar
  subqueries.
- Custom `PhysicalPlanner` or `ExtensionPlanner` implementations: add the new
  parameter and forward it.
- Code that read or wrote `execution_props.subquery_indexes` /
  `execution_props.subquery_results`: build a `PhysicalPlanningContext` instead.
- Code that read `execution_props.lambda_variable_qualifier` or called
  `ExecutionProps::with_qualified_lambda_variables`: remove that usage. Callers
  that only plan a `HigherOrderFunction` are not affected --
  `create_physical_expr` populates the lambda qualifiers itself as it descends
  into lambda bodies. Code that needs to read or extend the lambda
  scope should use the equivalents on `PhysicalPlanningContext`:
  `PhysicalPlanningContext::lambda_variable_qualifier` and
  `PhysicalPlanningContext::with_qualified_lambda_variables`.

**Migration guide:**

When creating a physical expression outside of physical planning, pass an empty
context:

```rust,ignore
use datafusion_expr::physical_planning_context::PhysicalPlanningContext;
use datafusion_physical_expr::create_physical_expr;

// Before
let phys = create_physical_expr(&expr, &schema, &props)?;

// After
let phys = create_physical_expr(
    &expr,
    &schema,
    &props,
    &PhysicalPlanningContext::default(),
)?;
```

For `ExtensionPlanner` implementations, accept and forward the context:

```rust,ignore
async fn plan_extension(
    &self,
    planner: &dyn PhysicalPlanner,
    node: &dyn UserDefinedLogicalNode,
    logical_inputs: &[&LogicalPlan],
    physical_inputs: &[Arc<dyn ExecutionPlan>],
    session: &dyn Session,
    planning_ctx: &PhysicalPlanningContext, // new parameter
) -> Result<Option<Arc<dyn ExecutionPlan>>> {
    for expr in node.expressions() {
        // Forward the context so scalar subqueries in this node's
        // expressions resolve against the plan's subquery state
        planner.create_physical_expr(&expr, node.schema(), session, planning_ctx)?;
    }
    // ...
}
```

See [PR #23649](https://github.com/apache/datafusion/pull/23649) and
[PR #23989](https://github.com/apache/datafusion/pull/23989) for details.

### Catalog, planner, and optimizer contracts moved to `datafusion-session`

The catalog, planner, and physical optimizer contract traits now live in the
`datafusion-session` crate. This makes them available through `Session` without
downcasting to `SessionState`, including across the FFI boundary.

The moved catalog traits are `CatalogProviderList`, `CatalogProvider`,
`SchemaProvider`, `TableProvider`, `TableProviderFactory`, and
`TableFunctionImpl`. The related `TableFunction` struct also moved. The
`datafusion-catalog` crate re-exports these items from their new location, so
paths such as `datafusion::catalog::TableProvider` and
`datafusion_catalog::CatalogProvider` continue to work unchanged.

The moved planning and optimization traits are `QueryPlanner`,
`PhysicalPlanner`, `ExtensionPlanner`, `PhysicalOptimizerRule`, and
`PhysicalOptimizerContext`. Their previous paths also continue to work through
re-exports:

- `datafusion::execution::context::QueryPlanner`
- `datafusion::physical_planner::{PhysicalPlanner, ExtensionPlanner}`
- `datafusion_physical_optimizer::{PhysicalOptimizerRule, PhysicalOptimizerContext}`

The session argument for methods on `QueryPlanner`, `PhysicalPlanner`, and
`ExtensionPlanner` changed from `&SessionState` to `&dyn Session`. Custom planner
implementations should update their signatures. Planner code should use methods
on `Session` instead of downcasting it to `SessionState`.

The `Session` trait now requires a `catalog_list` method that returns the
catalogs registered with the session:

```rust
fn catalog_list(&self) -> Arc<dyn CatalogProviderList>;
```

Custom `Session` implementations must add this method. Implementations that do
not expose a catalog can return the new `EmptyCatalogProviderList`:

```rust
use std::sync::Arc;
use datafusion_session::{CatalogProviderList, EmptyCatalogProviderList};

fn catalog_list(&self) -> Arc<dyn CatalogProviderList> {
    Arc::new(EmptyCatalogProviderList)
}
```

`Session` gains a `query_planner` method alongside `optimize`,
`physical_optimizers`, and `statistics_registry`. All four have default
implementations, so existing `Session` implementations that do not perform
physical planning require no changes: `query_planner` defaults to the new
`UnsupportedQueryPlanner`, `optimize` returns the plan unchanged,
`physical_optimizers` returns no rules, and `statistics_registry` returns
`None`.

A custom session that drives planning through `DefaultQueryPlanner` or
`DefaultPhysicalPlanner` must override these methods to expose its planning and
optimization behavior; the defaults will otherwise produce unoptimized plans or
fail to plan at all. The simplest approach is to delegate to a `SessionState`:

```rust
use std::sync::Arc;
use datafusion_session::{PhysicalOptimizerRule, QueryPlanner};

fn query_planner(&self) -> Arc<dyn QueryPlanner + Send + Sync> {
    self.inner.query_planner()
}

fn optimize(&self, plan: &LogicalPlan) -> Result<LogicalPlan> {
    self.inner.optimize(plan)
}

fn physical_optimizers(&self) -> &[Arc<dyn PhysicalOptimizerRule + Send + Sync>] {
    self.inner.physical_optimizers()
}
```

`ForeignSession::create_physical_plan` runs the complete planning pipeline in the
library that owns the session. `ForeignSession::query_planner`, `optimize`, and
`physical_optimizers` forward to the owning session across the FFI boundary. A
foreign query planner can also be installed on a session through the new
`datafusion_ffi::query_planner::FFI_QueryPlanner`; see that module's
documentation for how plans and extension codecs cross the boundary.

See [PR #23703](https://github.com/apache/datafusion/pull/23703) for details on
the catalog changes.

### `FFI_LogicalExtensionCodec::task_ctx_provider` is now private

The `task_ctx_provider` field on
`datafusion_ffi::proto::logical_extension_codec::FFI_LogicalExtensionCodec` was
`pub` and is now crate-private, matching `FFI_PhysicalExtensionCodec`.

**Who is affected:**

- Code that read or cloned `FFI_LogicalExtensionCodec::task_ctx_provider`
  directly. Pass the task context provider to `FFI_LogicalExtensionCodec::new`
  instead, and keep your own copy if you need it elsewhere.

### Unused `async` removed from several public functions

Public functions that were declared `async` but never awaited anything are now
synchronous:

- `CsvFormat::read_to_delimited_chunks_from_stream` (in
  `datafusion_datasource_csv`, re-exported as
  `datafusion::datasource::file_format::csv::CsvFormat`)
- `datafusion_substrait::serializer::deserialize_bytes`, which now also borrows
  its input as `&[u8]` instead of taking an owned `Vec<u8>`
- `datafusion::test_util::parquet::TestParquetFile::create_scan`

**Migration guide:**

Remove `.await` from call sites; the compiler flags each one, since `.await`
on a non-future value does not compile:

```rust,ignore
// Before
let stream = csv_format
    .read_to_delimited_chunks_from_stream(input)
    .await;
let plan = deserialize_bytes(proto_bytes).await?;

// After
let stream = csv_format.read_to_delimited_chunks_from_stream(input);
let plan = deserialize_bytes(&proto_bytes)?;
```

### `MovingMin` and `MovingMax` changed to `pub(crate)`

`MovingMin` and `MovingMax` in `datafusion_functions_aggregate::min_max` have been changed from `pub` to `pub(crate)` visibility as they are internal helper data structures for DataFusion's sliding window aggregators.

**Who is affected:**

- Code that directly imported `MovingMin` or `MovingMax` from `datafusion_functions_aggregate`. Standard SQL window functions (`MIN(...) OVER (...)` / `MAX(...) OVER (...)`) are unaffected.

See [PR #23827](https://github.com/apache/datafusion/pull/23827) for details.

### `WindowExpr::evaluate_stateful` now takes a `WindowEvalContext`

`WindowExpr::evaluate_stateful` (and the provided
`AggregateWindowExpr::aggregate_evaluate_stateful` method) take a new
`WindowEvalContext` argument carrying stream-level information that is shared
by all partitions:

```rust,ignore
// Before
fn evaluate_stateful(
    &self,
    partition_batches: &PartitionBatches,
    window_agg_state: &mut PartitionWindowAggStates,
) -> Result<()>

// After
fn evaluate_stateful(
    &self,
    partition_batches: &PartitionBatches,
    window_agg_state: &mut PartitionWindowAggStates,
    eval_ctx: &WindowEvalContext<'_>,
) -> Result<()>
```

`WindowEvalContext` currently carries the most recent input row, which
previously lived in each partition's `PartitionBatchState` (see the next
section). The struct is `#[non_exhaustive]` so that fields can be added
without further signature changes: construct it with
`WindowEvalContext::default()` and set fields through its builder methods.

**Who is affected:**

- Implementations of the `WindowExpr` trait that override `evaluate_stateful`
  must add the new parameter.
- Callers of `evaluate_stateful` or `aggregate_evaluate_stateful` must pass a
  context.

**Migration guide:**

```rust,ignore
use datafusion_physical_expr::window::WindowEvalContext;

// Before
window_expr.evaluate_stateful(&partition_batches, &mut window_agg_state)?;

// After
let eval_ctx = WindowEvalContext::default()
    .with_most_recent_row(most_recent_row.as_ref());
window_expr.evaluate_stateful(
    &partition_batches,
    &mut window_agg_state,
    &eval_ctx,
)?;
```

Pass `WindowEvalContext::default()` when no most-recent-row watermark is
available (for example, when the input is sorted by the partition keys and
partition ends are detected directly).

### `PartitionBatchState::most_recent_row` removed

The `most_recent_row` field and the `set_most_recent_row` method have been
removed from `datafusion_expr::window_state::PartitionBatchState`. The most
recent input row is a property of the whole input stream rather than
per-partition state: every partition observed the same value. It is now
tracked once by the operator driving the evaluation and passed to window
expressions through the new `WindowEvalContext` argument of
`WindowExpr::evaluate_stateful` described above.

**Who is affected:**

- Code that read `PartitionBatchState::most_recent_row` or called
  `set_most_recent_row`, such as custom streaming window operators.

**Migration guide:**

Track the most recent input row once per stream (for example, a one-row
slice of the last non-empty input batch) and pass it to window expressions
via `WindowEvalContext::with_most_recent_row` instead of copying it into
each partition's state.

### `MSRV` updated to 1.94.0

The Minimum Supported Rust Version (MSRV) has been updated to [`1.94.0`].

[`1.94.0`]: https://releases.rs/docs/1.94.0/

### `CachedParquetFileReader` removed; `ParquetFileReader` fields are now private

`CachedParquetFileReader` duplicated `ParquetFileReader` and has been removed;
`ParquetFileReader`'s fields are also now private, with
`file_metrics()` and `partitioned_file()` accessors added for the two that
were previously public.

**Who is affected:**

- Code that names the `CachedParquetFileReader` type.
- Code that constructs a `ParquetFileReader` directly via a struct literal, or
  reads/writes its fields.

**Migration guide:**

`ParquetFileReader::new` is no longer public; build a reader through
`ParquetFileReaderFactory::create_reader` (via `DefaultParquetFileReaderFactory`
or `CachedParquetFileReaderFactory`) instead of constructing one directly:

```rust,ignore
// Before
let inner = ParquetObjectReader::new(Arc::clone(&store), location).with_file_size(size);
let reader = CachedParquetFileReader::new(
    file_metrics,
    store,
    inner,
    partitioned_file,
    metadata_cache,
    metadata_size_hint,
);

// After
let reader = CachedParquetFileReaderFactory::new(store, metadata_cache)
    .create_reader(partition_index, partitioned_file, metadata_size_hint, &metrics)?;
```

Replace field access with the new accessor methods:

```rust,ignore
// Before
let bytes_scanned = reader.file_metrics.bytes_scanned.value();
let location = &reader.partitioned_file.object_meta.location;

// After
let bytes_scanned = reader.file_metrics().bytes_scanned.value();
let location = &reader.partitioned_file().object_meta.location;
```

### `array_distance` scalar function now rejects multidimensional arrays

`array_distance` only supports one-dimensional arrays. Previously, when given
multidimensional arrays, it computed the distance using only the first
subarray and ignored the remaining subarrays. For example:

```sql
SELECT array_distance(
    [[1, 2], [100, 100]],
    [[1, 4], [0, 0]]
);
```

Previously, this query returned `2.0`, the distance between `[1, 2]` and
`[1, 4]`. It now returns a planning error stating that `array_distance` only
supports one-dimensional arrays.

### `ParquetObjectReader` / `ParquetObjectWriter` deprecated upstream

The [`parquet` crate] deprecated [`ParquetObjectReader`]
and [`ParquetObjectWriter`] in favor of implementing
[`AsyncFileReader`] directly (see the example on the [`AsyncFileReader`] trait and
[`parquet/examples/object_store.rs`] in `arrow-rs`) or passing an
[`BufWriter`] straight to [`AsyncArrowWriter`].

**Who is affected:**

- Custom [`ParquetFileReaderFactory`] implementations that construct a
  [`ParquetObjectReader`] directly and now see a deprecation warning after
  upgrading the `parquet` dependency.

**Migration guide:**

If your [`AsyncFileReader`] implementation exists mainly to read from an
[`ObjectStore`] and track metrics, consider using DataFusion's
[`ParquetFileReader`] instead of wrapping a [`ParquetObjectReader`]:

```rust,ignore
// Before
let inner = ParquetObjectReader::new(store, location).with_file_size(size);
Ok(Box::new(MyReader { inner, file_metrics, partitioned_file }))

// After
Ok(Box::new(ParquetFileReader {
    file_metrics,
    store,
    metadata_size_hint,
    partitioned_file,
}))
```

If you need custom behavior (I/O coalescing, byte caching, a dedicated I/O
runtime), implement `AsyncFileReader` directly against your `ObjectStore`,
following the pattern in [`parquet/examples/object_store.rs`]

See [PR #24030](https://github.com/apache/datafusion/pull/24030) for details.

[`parquet` crate]: https://crates.io/crates/parquet
[`parquetobjectreader`]: https://docs.rs/parquet/59.1.0/parquet/arrow/async_reader/struct.ParquetObjectReader.html
[`parquetobjectwriter`]: https://docs.rs/parquet/59.1.0/parquet/arrow/async_writer/struct.ParquetObjectWriter.html
[`parquetfilereader`]: https://docs.rs/datafusion/latest/datafusion/datasource/physical_plan/parquet/struct.ParquetFileReader.html
[`parquetfilereaderfactory`]: https://docs.rs/datafusion/latest/datafusion/datasource/physical_plan/parquet/trait.ParquetFileReaderFactory.html
[`asyncfilereader`]: https://docs.rs/parquet/59.1.0/parquet/arrow/async_reader/trait.AsyncFileReader.html
[`objectstore`]: https://docs.rs/object_store/latest/object_store/trait.ObjectStore.html
[`bufwriter`]: https://docs.rs/tokio/latest/tokio/io/struct.BufWriter.html
[`asyncarrowwriter`]: https://docs.rs/parquet/59.1.0/parquet/arrow/async_writer/struct.AsyncArrowWriter.html
[`parquet/examples/object_store.rs`]: https://github.com/apache/arrow-rs/blob/main/parquet/examples/object_store.rs