feat: session-first DataFusion integration + session resolution policies (delta-io#4145)

# Description

Make delta-rs' DataFusion integration consistently honor caller provided
sessions and introduce a session first API for registering Delta object
stores.

**Changes:** 

Session resolution: 
- Add `resolve_session_state(...)` & `SessionFallbackPolicy`
(`InternalDefaults` / `DeriveFromTrait` / `RequireSessionState`)
- Builders expose `with_session_fallback_policy(...)` to control
strictness
- Migrate operations to use resolver
(optimize/merge/update/write/delete)

Session first registration:
- Add `DeltaSessionExt` trait:
  - `ensure_object_store_registered(...)`
  - `ensure_log_store_registered(...)`
- Deprecate `DeltaTable::update_datafusion_session(...)` (shim kept for
compatibility)

Predicate parsing:
- Non-`SessionState` sessions can preserve UDFs when configured via
`DeriveFromTrait`

**Compatibility:**
- Default is backward compatible:
`SessionFallbackPolicy::InternalDefaults` warns but doesn't break
- Strict mode available:
`with_session_fallback_policy(RequireSessionState)` errors instead of
falling back
- `DeltaTable::update_datafusion_session` remains but is deprecated

**Tests:**
- Regression tests for fallback policy wiring across builders
(`RequireSessionState` path)
- Existing `deltalake-core` DataFusion test suite passes with
`--features datafusion`

# Related Issue(s)
**Addresses:** 
- delta-io#4081
- delta-io#4139
<!---
For example:


- closes #106
--->

# Documentation

<!---
Share links to useful documentation
--->

# Follow ups:

- Flip default to `RequireSessionState` (breaking change)
- Remove deprecated `DeltaTable::update_datafusion_session` after
deprecation window

---------

Signed-off-by: Ethan Urbanski <ethan@urbanskitech.com>

Author: ethan-tyler

crates/core/src/delta_datafusion/expr.rs

@@ -194,30 +194,34 @@ pub(crate) struct DeltaContextProvider<'a> {
 }
 
 impl<'a> DeltaContextProvider<'a> {
-    fn new(session: &'a dyn Session) -> Self {
-        // default planners are [CoreFunctionPlanner, NestedFunctionPlanner, FieldAccessPlanner,
-        // UserDefinedFunctionPlanner]
+    fn try_new(session: &'a dyn Session) -> DeltaResult<Self> {
         let planners: Vec<Arc<dyn ExprPlanner>> = vec![
             Arc::new(CoreFunctionPlanner::default()),
             Arc::new(CustomNestedFunctionPlanner::default()),
             Arc::new(FieldAccessPlanner),
             Arc::new(datafusion::functions::unicode::planner::UnicodeFunctionPlanner),
             Arc::new(datafusion::functions::datetime::planner::DatetimeFunctionPlanner),
         ];
-        // Disable the above for testing
-        //let planners = state.expr_planners();
-        let new_state = session
-            .as_any()
-            .downcast_ref::<SessionState>()
-            .map(|state| SessionStateBuilder::new_from_existing(state.clone()))
-            .unwrap_or_default()
+        let (base_state, _) = crate::delta_datafusion::resolve_session_state(
+            Some(session),
+            crate::delta_datafusion::SessionFallbackPolicy::DeriveFromTrait,
+            || SessionStateBuilder::new().with_default_features().build(),
+            crate::delta_datafusion::SessionResolveContext {
+                operation: "parse_predicate_expression",
+                table_uri: None,
+                cdc: false,
+            },
+        )?;
+
+        let new_state = SessionStateBuilder::new_from_existing(base_state)
             .with_expr_planners(planners.clone())
             .build();
-        DeltaContextProvider {
+
+        Ok(DeltaContextProvider {
             planners,
             state: new_state,
             _phantom: PhantomData,
-        }
+        })
     }
 }
 
@@ -283,7 +287,7 @@ pub fn parse_predicate_expression(
             source: Box::new(err),
         })?;
 
-    let context_provider = DeltaContextProvider::new(session);
+    let context_provider = DeltaContextProvider::try_new(session)?;
     let sql_to_rel =
         SqlToRel::new_with_options(&context_provider, DeltaParserOptions::default().into());
 
@@ -585,7 +589,9 @@ impl fmt::Display for ScalarValueFormat<'_> {
 #[cfg(test)]
 mod test {
     use arrow_schema::DataType as ArrowDataType;
-    use datafusion::common::{Column, ScalarValue, ToDFSchema};
+    use datafusion::common::{Column, DFSchema, ScalarValue, ToDFSchema};
+    use datafusion::execution::SessionStateBuilder;
+    use datafusion::execution::runtime_env::RuntimeEnvBuilder;
     use datafusion::functions::core::arrow_cast;
     use datafusion::functions::core::expr_ext::FieldAccessor;
     use datafusion::functions::encoding::expr_fn::decode;
@@ -594,13 +600,18 @@ mod test {
     use datafusion::functions_nested::expr_fn::cardinality;
     use datafusion::logical_expr::expr::ScalarFunction;
     use datafusion::logical_expr::{BinaryExpr, Cast, Expr, ExprSchemable, col, lit};
-    use datafusion::prelude::SessionContext;
+    use datafusion::prelude::{SessionConfig, SessionContext};
 
     use crate::DeltaTable;
+    use crate::delta_datafusion::planner::DeltaPlanner;
     use crate::delta_datafusion::{DataFusionMixins, DeltaSessionContext};
     use crate::kernel::{ArrayType, DataType, PrimitiveType, StructField, StructType};
+    use crate::test_utils::datafusion::{WrapperSession, make_test_scalar_udf};
 
     use super::fmt_expr_to_sql;
+    use super::parse_predicate_expression;
+
+    const TEST_UDF_NAME: &str = "delta_rs_parse_expr_test_udf";
 
     struct ParseTest {
         expr: Expr,
@@ -722,6 +733,34 @@ mod test {
         table
     }
 
+    fn make_incompatible_session_with_udf() -> WrapperSession {
+        let runtime_env = RuntimeEnvBuilder::new().build_arc().unwrap();
+        let config = SessionConfig::new();
+        let udf = make_test_scalar_udf(TEST_UDF_NAME);
+
+        let state = SessionStateBuilder::new()
+            .with_default_features()
+            .with_config(config)
+            .with_runtime_env(runtime_env)
+            .with_query_planner(DeltaPlanner::new())
+            .with_scalar_functions(vec![udf])
+            .build();
+
+        WrapperSession::new(state)
+    }
+
+    #[test]
+    fn parse_predicate_expression_preserves_udfs_for_non_session_state() {
+        let wrapper = make_incompatible_session_with_udf();
+        let schema = DFSchema::empty();
+
+        let expr = parse_predicate_expression(&schema, format!("{TEST_UDF_NAME}(1) = 1"), &wrapper);
+        assert!(
+            expr.is_ok(),
+            "Expected UDF to be available during parsing but got: {expr:?}"
+        );
+    }
+
     #[tokio::test]
     async fn test_expr_sql() {
         let table = setup_table().await;

crates/core/src/delta_datafusion/mod.rs

@@ -42,7 +42,6 @@ use datafusion::datasource::TableProvider;
 use datafusion::datasource::physical_plan::wrap_partition_type_in_dict;
 use datafusion::execution::TaskContext;
 use datafusion::execution::context::SessionContext;
-use datafusion::execution::runtime_env::RuntimeEnv;
 use datafusion::logical_expr::logical_plan::CreateExternalTable;
 use datafusion::logical_expr::utils::conjunction;
 use datafusion::logical_expr::{Expr, Extension, LogicalPlan};
@@ -52,17 +51,16 @@ use datafusion_proto::logical_plan::LogicalExtensionCodec;
 use datafusion_proto::physical_plan::PhysicalExtensionCodec;
 use delta_kernel::engine::arrow_conversion::TryIntoArrow as _;
 use either::Either;
-use url::Url;
 
 use crate::delta_datafusion::expr::parse_predicate_expression;
 use crate::delta_datafusion::table_provider::DeltaScanWire;
 use crate::ensure_table_uri;
 use crate::errors::{DeltaResult, DeltaTableError};
 use crate::kernel::{Add, EagerSnapshot, LogDataHandler, Snapshot};
-use crate::logstore::{LogStore, LogStoreRef};
 use crate::table::state::DeltaTableState;
 use crate::{open_table, open_table_with_storage_options};
 
+pub(crate) use self::session::DeltaSessionExt;
 pub use self::session::{
     DeltaParserOptions, DeltaRuntimeEnvBuilder, DeltaSessionConfig, DeltaSessionContext,
     create_session,
@@ -93,6 +91,8 @@ pub mod logical;
 pub mod physical;
 pub mod planner;
 mod session;
+pub use session::SessionFallbackPolicy;
+pub(crate) use session::{SessionResolveContext, resolve_session_state};
 mod table_provider;
 pub(crate) mod utils;
 
@@ -307,14 +307,6 @@ impl DeltaTableState {
     }
 }
 
-// each delta table must register a specific object store, since paths are internally
-// handled relative to the table root.
-pub(crate) fn register_store(store: LogStoreRef, env: &RuntimeEnv) {
-    let object_store_url = store.object_store_url();
-    let url: &Url = object_store_url.as_ref();
-    env.register_object_store(url, store.object_store(None));
-}
-
 pub(crate) fn get_null_of_arrow_type(t: &ArrowDataType) -> DeltaResult<ScalarValue> {
     match t {
         ArrowDataType::Null => Ok(ScalarValue::Null),
@@ -641,6 +633,7 @@ mod tests {
     use std::fmt::{self, Debug, Display, Formatter};
     use std::ops::Range;
     use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel};
+    use url::Url;
 
     use super::*;