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:

// 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:

// 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 Strings.

Migration guide:

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

// 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:

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:

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

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:

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:

// 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:

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:

  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:

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

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

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

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:

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.

// 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:

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

// 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:

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 Boxed:

// 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:

// 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():

// 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 *:

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

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

See PR #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, 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 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:

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

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:

SET datafusion.spark.map_key_dedup_policy = 'LAST_WIN';

See PR #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:

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

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

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

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:

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 and PR #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:

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:

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:

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

// 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 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:

// 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:

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.

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:

// 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:

// 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:

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:

// 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 for details.

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:

// 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 for details.