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

### Filter predicate evaluation order may differ from query text

The logical optimizer now reorders filters so that cheap predicates (most binary
comparisons, `IS NULL`, `Between`, `InList`, etc.) evaluate before expensive
ones (`LIKE`, regex, scalar function calls, subqueries). For example,
`WHERE col LIKE '%foo%' AND col2 = 5` may evaluate `col2 = 5` before
`col LIKE '%foo%'`.

**Evaluation order has never been guaranteed to match the order written in the
query.** The SQL standard explicitly allows implementations to evaluate operands
in any order; major engines (PostgreSQL, SQL Server, Oracle, MySQL) document the
same. Queries should not rely on left-to-right evaluation or short-circuit
semantics for `AND` or `OR`. Previous versions of DataFusion already reordered
predicates (e.g., as part of expression simplification or predicate pushdown);
the new reordering pass just increases the scenarios where the optimizer will
change predicate evaluation order.

**Fallible-predicate patterns are particularly affected.** For example:

```sql
WHERE s ~ '^[0-9]+$' AND CAST(s AS INT) > 0
```

The intent is likely to filter non-numeric strings before the `CAST` runs,
but this depends on evaluation-order behavior the SQL standard does not
guaranteed. The new reorder makes this kind of pattern more likely to fail
at runtime if the optimizer moves the `CAST` ahead of the regex. To force
conditional evaluation, rewrite using `CASE`, which has standardized
short-circuit semantics:

```sql
WHERE CASE WHEN s ~ '^[0-9]+$' THEN CAST(s AS INT) > 0 ELSE false END
```

Volatile expressions (`random()`, `now()`, etc.) are exempt — their position
in the conjunct list is preserved so the number of times they evaluate per
query does not change.

### Map casts with nested Structs use schema-evolution rules

Casts of a `Map` whose key or value recursively contains a `Struct` now adapt
nested Struct fields by case-sensitive name. Nullable fields added to the target
Struct are filled with `NULL`, and source-only fields are omitted. Missing
non-nullable target fields, nullable-to-non-nullable changes, and incompatible
nested types are rejected.

Map keys define map identity. During this nested-Struct adaptation, key Struct
fields cannot be removed and primitive key type changes must preserve every
source value (for example, `Int32` to `Int64` is allowed, while `Int64` to
`Int32` is rejected). Sorted Maps also require an unchanged key type and the
same sorted flag.

**Migration guide:** Schemas that relied on positional Struct-field matching,
key-field removal, or narrowing Map-key casts must be updated to retain
case-sensitive field names and key identity. Ordinary Maps without semantic
Struct children continue to use Arrow's normal Map cast behavior.

### `TableProvider::scan` takes `Option<&[usize]>` for the projection

`TableProvider::scan` and the related APIs listed below previously took the
projection as `Option<&Vec<usize>>`:

```rust,ignore
// Before
async fn scan(
    &self,
    state: &dyn Session,
    projection: Option<&Vec<usize>>,
    filters: &[Expr],
    limit: Option<usize>,
) -> Result<Arc<dyn ExecutionPlan>>;

// After
async fn scan(
    &self,
    state: &dyn Session,
    projection: Option<&[usize]>,
    filters: &[Expr],
    limit: Option<usize>,
) -> Result<Arc<dyn ExecutionPlan>>;
```

Implementations only need the signature updated. Two patterns inside a `scan`
body need adjusting, because `Option<&[usize]>` is not an `Option<&T>` over a
sized type:

```rust,ignore
// Before                        // After
projection.cloned()             projection.map(|p| p.to_vec())
```

Callers holding an `Option<Vec<usize>>` should pass `.as_deref()` rather than
`.as_ref()`.

The same change applies to `StreamingTableExec::try_new`,
`FilterExec::with_projection` and `batch_filter`, which took the projection in
the same shape.

`datafusion_common::project_schema` now accepts unsized projections
(`projection: Option<&T> where T: AsRef<[usize]> + ?Sized`), so it takes both
`Option<&Vec<usize>>` and `Option<&[usize]>`; no caller changes are required.

Besides being the more idiomatic signature, this lets
`TableProvider::scan_with_args`'s default body hand the projection straight to
`scan` instead of allocating a `Vec` that then has to be captured by the
returned future -- which is what made that method expensive to compile
(`datafusion-session` compiles ~3.3x faster as a result).

### `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

### New `Dialect::Spark` variant

The `Dialect` enum in `datafusion_common::config` now includes a `Spark` variant.
If you match exhaustively on `Dialect`, add a `Dialect::Spark` arm.

### `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.

### `ExecutionPlan::with_new_children` and `ExecutionPlan::with_new_children_and_same_properties` deprecated

`with_new_children` and `with_new_children_and_same_properties` have been
deprecated. These methods are used to replace the child plans of an
`ExecutionPlan` while leaving the plan otherwise identical.

`with_new_children_if_necessary` has also been deprecated in favor of
`replace_children_if_necessary` for consistency in naming.

As noted [here](https://github.com/apache/datafusion/pull/23332#discussion_r3554897693),
while the addition of `with_new_children_and_same_properties` has the benefit
of skipping potentially expensive computation in the case that replacement children
have the same properties as the original children, it widens the API surface area
of `ExecutionPlan` in a way that could be confusing for users.

Thus, to rectify this, we unify these methods by introducing `replace_children`.
`replace_children` solves this problem by taking `ReplaceChildrenOptions`,
which includes a `ChildrenPropertiesMode`. The mode has two variants,
`Keep` and `Recompute`, which tell `replace_children` whether plan
properties can be reused or need to be recomputed.

This method is called from `replace_children_if_necessary`, which is the
standard entry point that should be used for replacing the children of a node.

**Migration guide:**

To migrate from `with_new_children` and `with_new_children_and_same_properties`
to `replace_children`, it is recommended to implement `replace_children` with
a `match` statement matching on the `ChildrenPropertiesMode`. In the case that
the properties match the children, `ChildrenPropertiesMode::Keep`,
follow the body of `with_new_children_and_same_properties`. In the case that
the properties do not match the children, `ChildrenPropertiesMode::Recompute`,
follow the body of `with_new_children`.

For example, take a look at the implementation for `FilterExec`:

```
    fn replace_children(
        self: Arc<Self>,
        mut children: Vec<Arc<dyn ExecutionPlan>>,
        options: ReplaceChildrenOptions,
    ) -> Result<Arc<dyn ExecutionPlan>> {
        validate_child_count!(self, children);
        match options.children_properties {
            ChildrenPropertiesMode::Keep => Ok(Arc::new(Self {
                input: children.swap_remove(0),
                metrics: ExecutionPlanMetricsSet::new(),
                ..Self::clone(&*self)
            })),
            ChildrenPropertiesMode::Recompute => {
                let new_input = children.swap_remove(0);
                FilterExecBuilder::from(&*self)
                    .with_input(new_input)
                    .build()
                    .map(|e| Arc::new(e) as _)
            }
        }
    }
```

In the case that the options indicate the properties are the same, we can simply
swap the children without having to recompute the properties. In the other case,
we create a new node from scratch.

To ensure that this works correctly, it is recommended that users also look
through their codebase and ensure that they use `replace_children_if_necessary`
for these changes — `replace_children_if_necessary` should be preferred over
manual use of `replace_children`, since `replace_children_if_necessary` will
call `replace_children` with the correct options filled in.

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

### `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.

### `RecursiveQuery` gained a `schema` field

`datafusion_expr::logical_plan::RecursiveQuery` exposed the static (anchor) term's
schema to parent plans, which ignored columns that only the recursive term makes
nullable. The node now stores the schema reconciled from both terms explicitly, so
it can no longer be built with an exhaustive struct literal.

**Who is affected:**

- Code constructing `RecursiveQuery` with a struct literal. This change also
  shipped in `54.1.0`, so the same migration applies within the `54` series.

**Migration guide:**

Build the node with `RecursiveQuery::try_new`, which computes the schema and
returns an error if the two terms do not have the same number of columns:

```rust,ignore
// Before
let query = RecursiveQuery {
    name,
    static_term,
    recursive_term,
    is_distinct,
};

// After
let query = RecursiveQuery::try_new(name, static_term, recursive_term, is_distinct)?;
```

`RecursiveQueryExec::try_new` takes the reconciled schema as its second argument
for the same reason, rather than deriving it from its children.

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

### `ExecutionPlan::apply_expressions` is now a required method

`apply_expressions` has been added as a **required** method on the `ExecutionPlan`, `FileSource`, and `DataSource` traits. Any custom implementation of
these traits must now implement `apply_expressions`. See docs on `ExecutionPlan::apply_expressions` for migration 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.

### `ExprType` gained a `SqlSimilarToPattern` variant

`SIMILAR TO` with a non-literal (column- or expression-based) pattern is now
represented by a new `SqlSimilarToPattern` physical expression, so it can be
serialized. The generated `ExprType` enum on `PhysicalExprNode` gained a
matching `SqlSimilarToPattern` variant (tag 28).

**Who is affected:**

- Users matching exhaustively on `ExprType` (for example in a custom
  physical-plan encoder/decoder).

**Migration guide:**

Add a `SqlSimilarToPattern` arm, or fall back to a wildcard arm:

```rust,ignore
match expr_type {
    // ...
    ExprType::SqlSimilarToPattern(node) => { /* ... */ }
    _ => { /* ... */ }
}
```

Plans encoded before this variant existed are unaffected: they never produced
this variant, so decoding is unchanged. Plans containing a dynamic
`SIMILAR TO` pattern encoded after this change fail to decode on older
readers that don't recognize the new tag.

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

### `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.

### `SortProperties::and_or` replaced by `and` and `or`

`datafusion_expr_common::sort_properties::SortProperties::and_or` has been
removed. Call `and` for `AND` and `or` for `OR`:

```rust,ignore
// Before (54.0.0)
let props = lhs.and_or(&rhs);

// After (55.0.0)
let props = lhs.and(&rhs); // for AND
let props = lhs.or(&rhs);  // for OR
```

The two operators do not preserve the same sort orders. `AND` keeps an ordering
only when nulls sort after `true` (`ASC NULLS LAST` or `DESC NULLS FIRST`), and
`OR` only when nulls sort before `false` (`ASC NULLS FIRST` or
`DESC NULLS LAST`). `and_or` did not say which operator the caller meant, so it
could not be right for both.

See [PR #24276](https://github.com/apache/datafusion/pull/24276) 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

### `datafusion-proto`: parquet options conversions are fallible

`protobuf::ParquetOptions` and `protobuf::TableParquetOptions` validate
`writer_version` when converting into their `datafusion-common` counterparts, so
those conversions are `TryFrom` rather than `From`.

Every other `From` / `TryFrom` conversion between DataFusion types and
`datafusion_proto::protobuf` messages is unchanged. Several impls moved to the
crate that owns their DataFusion type, but trait impls are global, so
`X::try_from(&proto)` and `proto.try_into()` still resolve with no import
changes.

**Migration guide:**

```rust,ignore
// Before
let opts = ParquetOptions::from(&proto_opts);
let table_opts = TableParquetOptions::from(&proto_table_opts);

// After
let opts = ParquetOptions::try_from(&proto_opts)?;
let table_opts = TableParquetOptions::try_from(&proto_table_opts)?;
```

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

### RANGE window frames with offsets now reject ORDER BY types without arithmetic

A `RANGE` frame offset such as `1 PRECEDING` is measured in `ORDER BY` values
rather than in rows, so the frame bound is computed as `current_value - 1`. That
requires arithmetic on the `ORDER BY` type. For types that have no such
arithmetic -- `Utf8`, `Binary`, `Boolean`, `List` and `Null` -- such frames are
now rejected when the query is planned:

```sql
SELECT x, COUNT(*) OVER (ORDER BY x RANGE BETWEEN 1 PRECEDING AND CURRENT ROW)
FROM (VALUES ('a'), ('b'), ('c')) t(x)
ORDER BY x;
-- Error during planning: RANGE with offset PRECEDING/FOLLOWING is not supported for ORDER BY type Utf8
```

What this replaces depends on the `ORDER BY` type. Over `Utf8`, `Boolean` and
`List` keys such queries planned successfully, and in DataFusion 54 the failing
frame-bound arithmetic was silently treated as an overflow
([#22140](https://github.com/apache/datafusion/pull/22140)): the bound collapsed
to the partition edge, and the query returned results computed over that wrongly
widened frame instead of an offset window. DataFusion 53 and earlier failed
during execution with an arithmetic error. Rows whose order key value was `NULL`
never reached the arithmetic, so queries where every order key value was `NULL`
returned correct results throughout; those also now fail at planning. Over a
`Null`-typed key the offset itself failed to coerce, erroring at planning time
with `Cast error: Casting from Utf8 to Null not supported`, and binary keys were
rejected for every `RANGE` frame with the internal error quoted below -- for
those two, only the error message changes.

Such an `ORDER BY` type is still usable with the `UNBOUNDED PRECEDING`,
`CURRENT ROW` and `UNBOUNDED FOLLOWING` bounds, which are located by comparing
`ORDER BY` values rather than by computing them. Binary order keys additionally
gained support for those bounds in this release, having previously failed with
`Internal error: Cannot run range queries on datatype: Binary`.

**Who is affected:**

- Queries using `offset PRECEDING` or `offset FOLLOWING` in a `RANGE` frame over
  a non-numeric, non-temporal `ORDER BY` key. To count rows rather than compare
  values, use a `ROWS` frame; to include everything up to the partition edge,
  use `UNBOUNDED PRECEDING` or `UNBOUNDED FOLLOWING`.

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