fix(datafusion): resolve DML predicates against execution scan schema (delta-io#4127)

# Description
After the DataFusion 52 upgrade, DELETE and UPDATE operations fail with:

`Invalid comparison operation: Utf8 == Utf8View`

Error: `Arrow error: Invalid comparison operation: Utf8 == Utf8View`

This occurs when using string equality predicates like
table.delete().with_predicate("name = 'foo'").

## Root Cause:
With DataFusion 52, Parquet string/binary columns may be represented as
view types (`Utf8View`/`BinaryView`) when
`datafusion.execution.parquet.schema_force_view_types` is enabled
(default `true`).

DELETE/UPDATE were resolving predicates and SET expressions against
`snapshot.arrow_schema()` (base `Utf8`/`Binary`), but executing against
the scan/provider schema (which can include view types). This mismatch
caused filter evaluation errors like `Utf8 == Utf8View` (and similarly
`Binary == BinaryView`).

## Fix:
Resolve predicates against DeltaScanConfig::table_schema() instead of
snapshot.arrow_schema(). This ensures predicate literals are coerced to
match the actual execution schema (view types when enabled, base types
when disabled).

## Changes:
- delete.rs: Use DeltaScanConfig::table_schema() for predicate
resolution
- update.rs: Use DeltaScanConfig::table_schema() for both SET
expressions and WHERE predicates
- expr.rs: Add ScalarValue::Dictionary support to fmt_expr_to_sql
(partition columns are dictionary-encoded in execution schema)

##  Tests:
- test_delete_string_equality_utf8view_regression_4125
- test_update_string_equality_non_partition
- test_delete_partition_string_predicate_dictionary_formatting
- test_delete_binary_equality_non_partition
- test_delete_custom_session_schema_force_view_types_disabled
- Dictionary scalar formatting unit tests in expr.rs

# Related Issue(s)
- Fixes delta-io#4125

<!---
For example:

- closes #106
--->

# Documentation

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

---------

Signed-off-by: Ethan Urbanski <ethan@urbanskitech.com>
Co-authored-by: R. Tyler Croy <rtyler@brokenco.de>

Author: ethan-tyler

crates/core/src/delta_datafusion/expr.rs

@@ -495,6 +495,13 @@ struct ScalarValueFormat<'a> {
 impl fmt::Display for ScalarValueFormat<'_> {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         match self.scalar {
+            ScalarValue::Dictionary(_key_type, inner) => write!(
+                f,
+                "{}",
+                ScalarValueFormat {
+                    scalar: inner.as_ref()
+                }
+            )?,
             ScalarValue::Boolean(e) => format_option!(f, e)?,
             ScalarValue::Float32(e) => format_option!(f, e)?,
             ScalarValue::Float64(e) => format_option!(f, e)?,
@@ -761,6 +768,24 @@ mod test {
                 expected: "_binary = decode('aa00ff', 'hex')".to_string(),
                 override_expected_expr: Some(col("_binary").eq(decode(lit("aa00ff"), lit("hex")))),
             },
+            ParseTest {
+                expr: col("id").eq(lit(ScalarValue::Dictionary(
+                    Box::new(ArrowDataType::UInt16),
+                    Box::new(ScalarValue::Utf8(Some("A".into()))),
+                ))),
+                expected: "id = 'A'".to_string(),
+                // Parsing canonicalizes away dictionary wrappers.
+                override_expected_expr: Some(col("id").eq(lit(ScalarValue::Utf8(Some("A".into()))))),
+            },
+            ParseTest {
+                expr: col("value").eq(lit(ScalarValue::Dictionary(
+                    Box::new(ArrowDataType::UInt16),
+                    Box::new(ScalarValue::Int32(Some(3))),
+                ))),
+                expected: "value = 3".to_string(),
+                // Parsing canonicalizes away dictionary wrappers.
+                override_expected_expr: Some(col("value").eq(lit(ScalarValue::Int64(Some(3))))),
+            },
             simple!(
                 col("value").between(lit(20_i64), lit(30_i64)),
                 "value BETWEEN 20 AND 30".to_string()

crates/core/src/operations/delete.rs

@@ -53,6 +53,7 @@ use uuid::Uuid;
 use super::Operation;
 use super::cdc::should_write_cdc;
 use crate::DeltaTable;
+use crate::delta_datafusion::DeltaScanConfig;
 use crate::delta_datafusion::expr::fmt_expr_to_sql;
 use crate::delta_datafusion::logical::{
     LogicalPlanBuilderExt as _, LogicalPlanExt, MetricObserver,
@@ -200,7 +201,13 @@ impl std::future::IntoFuture for DeleteBuilder {
 
             let predicate = this
                 .predicate
-                .map(|p| p.resolve(session.as_ref(), snapshot.arrow_schema().to_dfschema_ref()?))
+                .map(|p| {
+                    let scan_config = DeltaScanConfig::new_from_session(session.as_ref());
+                    let predicate_schema = scan_config
+                        .table_schema(snapshot.table_configuration())?
+                        .to_dfschema_ref()?;
+                    p.resolve(session.as_ref(), predicate_schema)
+                })
                 .transpose()?;
 
             let operation = DeltaOperation::Delete {
@@ -440,6 +447,7 @@ mod tests {
     use arrow::datatypes::{Field, Schema};
     use arrow::record_batch::RecordBatch;
     use arrow_array::ArrayRef;
+    use arrow_array::BinaryArray;
     use arrow_array::StringArray;
     use arrow_array::StructArray;
     use arrow_buffer::NullBuffer;
@@ -449,6 +457,7 @@ mod tests {
     use datafusion::physical_plan::ExecutionPlan;
     use datafusion::physical_plan::collect;
     use datafusion::prelude::*;
+    use datafusion::scalar::ScalarValue;
     use delta_kernel::engine::arrow_conversion::TryIntoKernel;
     use delta_kernel::schema::PrimitiveType;
     use delta_kernel::schema::StructType;
@@ -913,6 +922,201 @@ mod tests {
         assert_batches_sorted_eq!(&expected, &actual);
     }
 
+    #[tokio::test]
+    async fn test_delete_partition_string_predicate_dictionary_formatting() -> DeltaResult<()> {
+        // Regression test: resolving predicates against the execution scan schema can
+        // dictionary-encode partition columns, so fmt_expr_to_sql must support
+        // ScalarValue::Dictionary scalars.
+        let schema = get_arrow_schema(&None);
+        let table = setup_table(Some(vec!["id"])).await;
+
+        let batch = RecordBatch::try_new(
+            Arc::clone(&schema),
+            vec![
+                Arc::new(arrow::array::StringArray::from(vec!["A", "B", "A", "C"])),
+                Arc::new(arrow::array::Int32Array::from(vec![1, 2, 3, 4])),
+                Arc::new(arrow::array::StringArray::from(vec![
+                    "2021-02-02",
+                    "2021-02-02",
+                    "2021-02-02",
+                    "2021-02-02",
+                ])),
+            ],
+        )?;
+
+        let table = write_batch(table, batch).await;
+
+        let (table, _metrics) = table.delete().with_predicate("id = 'A'").await?;
+
+        let last_commit = table.last_commit().await.unwrap();
+        let parameters = last_commit.operation_parameters.clone().unwrap();
+        assert_eq!(parameters["predicate"], json!("id = 'A'"));
+
+        let expected = [
+            "+----+-------+------------+",
+            "| id | value | modified   |",
+            "+----+-------+------------+",
+            "| B  | 2     | 2021-02-02 |",
+            "| C  | 4     | 2021-02-02 |",
+            "+----+-------+------------+",
+        ];
+        let actual = get_data(&table).await;
+        assert_batches_sorted_eq!(&expected, &actual);
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_delete_binary_equality_non_partition() -> DeltaResult<()> {
+        // Regression test: DF52 view types can cause predicate evaluation to compare
+        // Binary against BinaryView for equality predicates.
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("data", DataType::Binary, true),
+            Field::new("int32", DataType::Int32, true),
+        ]));
+
+        let batch = RecordBatch::try_new(
+            schema,
+            vec![
+                Arc::new(BinaryArray::from_opt_vec(vec![
+                    Some(b"aaa".as_slice()),
+                    Some(b"bbb".as_slice()),
+                    Some(b"ccc".as_slice()),
+                ])) as ArrayRef,
+                Arc::new(Int32Array::from(vec![0, 1, 2])) as ArrayRef,
+            ],
+        )?;
+
+        let table = DeltaTable::new_in_memory().write(vec![batch]).await?;
+
+        let (table, _metrics) = table
+            .delete()
+            .with_predicate(col("data").eq(lit(ScalarValue::Binary(Some(b"bbb".to_vec())))))
+            .await?;
+
+        let last_commit = table.last_commit().await.unwrap();
+        let parameters = last_commit.operation_parameters.clone().unwrap();
+        assert_eq!(
+            parameters["predicate"],
+            json!("data = decode('626262', 'hex')")
+        );
+
+        let mut values = Vec::new();
+        for batch in get_data(&table).await {
+            let int32 = batch
+                .column(1)
+                .as_any()
+                .downcast_ref::<Int32Array>()
+                .unwrap();
+            for idx in 0..batch.num_rows() {
+                values.push(int32.value(idx));
+            }
+        }
+        values.sort();
+        assert_eq!(values, vec![0, 2]);
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_delete_string_equality_utf8view_regression_4125() -> DeltaResult<()> {
+        // Direct regression test for GitHub issue #4125:
+        // https://github.com/delta-io/delta-rs/issues/4125
+        //
+        // Non-partition string column with default session (view types enabled in DF52+).
+        // Was failing with: Invalid comparison operation: Utf8 == Utf8View
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("name", DataType::Utf8, false),
+            Field::new("value", DataType::Int32, false),
+        ]));
+
+        let batch = RecordBatch::try_new(
+            Arc::clone(&schema),
+            vec![
+                Arc::new(StringArray::from(vec!["alice", "bob", "charlie"])),
+                Arc::new(Int32Array::from(vec![1, 2, 3])),
+            ],
+        )?;
+
+        let table = DeltaTable::new_in_memory().write(vec![batch]).await?;
+
+        // This was the failing operation in issue #4125
+        let (table, _metrics) = table.delete().with_predicate("name = 'bob'").await?;
+
+        let last_commit = table.last_commit().await.unwrap();
+        let parameters = last_commit.operation_parameters.clone().unwrap();
+        assert_eq!(parameters["predicate"], json!("name = 'bob'"));
+
+        let expected = [
+            "+---------+-------+",
+            "| name    | value |",
+            "+---------+-------+",
+            "| alice   | 1     |",
+            "| charlie | 3     |",
+            "+---------+-------+",
+        ];
+        let actual = get_data(&table).await;
+        assert_batches_sorted_eq!(&expected, &actual);
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_delete_custom_session_schema_force_view_types_disabled() -> DeltaResult<()> {
+        // Integration guardrail: user-supplied DataFusion sessions may disable view types.
+        // Predicate resolution must respect the provided session's config.
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("name", DataType::Utf8, false),
+            Field::new("value", DataType::Int32, false),
+        ]));
+
+        let batch = RecordBatch::try_new(
+            Arc::clone(&schema),
+            vec![
+                Arc::new(StringArray::from(vec!["alice", "bob", "charlie"])),
+                Arc::new(Int32Array::from(vec![1, 2, 3])),
+            ],
+        )?;
+
+        let table = DeltaTable::new_in_memory().write(vec![batch]).await?;
+
+        let config: datafusion::prelude::SessionConfig =
+            crate::delta_datafusion::DeltaSessionConfig::default().into();
+        let config = config.set_bool(
+            "datafusion.execution.parquet.schema_force_view_types",
+            false,
+        );
+        let runtime_env = datafusion::execution::runtime_env::RuntimeEnvBuilder::new()
+            .build_arc()
+            .unwrap();
+        let state = datafusion::execution::SessionStateBuilder::new()
+            .with_default_features()
+            .with_config(config)
+            .with_runtime_env(runtime_env)
+            .with_query_planner(crate::delta_datafusion::planner::DeltaPlanner::new())
+            .build();
+        let session = Arc::new(state);
+
+        let (table, _metrics) = table
+            .delete()
+            .with_session_state(session)
+            .with_predicate("name = 'bob'")
+            .await?;
+
+        let last_commit = table.last_commit().await.unwrap();
+        let parameters = last_commit.operation_parameters.clone().unwrap();
+        assert_eq!(parameters["predicate"], json!("name = 'bob'"));
+
+        let expected = [
+            "+---------+-------+",
+            "| name    | value |",
+            "+---------+-------+",
+            "| alice   | 1     |",
+            "| charlie | 3     |",
+            "+---------+-------+",
+        ];
+        let actual = get_data(&table).await;
+        assert_batches_sorted_eq!(&expected, &actual);
+        Ok(())
+    }
+
     #[tokio::test]
     async fn test_failure_nondeterministic_query() {
         // Deletion requires a deterministic predicate

crates/core/src/operations/update.rs

@@ -46,7 +46,9 @@ use super::{
     CustomExecuteHandler, Operation,
     write::execution::{write_execution_plan, write_execution_plan_cdc},
 };
-use crate::delta_datafusion::{Expression, scan_files_where_matches, update_datafusion_session};
+use crate::delta_datafusion::{
+    DeltaScanConfig, Expression, scan_files_where_matches, update_datafusion_session,
+};
 use crate::kernel::resolve_snapshot;
 use crate::logstore::LogStoreRef;
 use crate::operations::cdc::*;
@@ -272,7 +274,10 @@ async fn execute(
     let exec_start = Instant::now();
     let mut metrics = UpdateMetrics::default();
 
-    let schema = snapshot.arrow_schema().to_dfschema_ref()?;
+    let scan_config = DeltaScanConfig::new_from_session(session);
+    let schema = scan_config
+        .table_schema(snapshot.table_configuration())?
+        .to_dfschema_ref()?;
     let updates: HashMap<_, _> = updates
         .into_iter()
         .map(|(key, expr)| expr.resolve(session, schema.clone()).map(|e| (key.name, e)))
@@ -442,7 +447,13 @@ impl std::future::IntoFuture for UpdateBuilder {
 
             let predicate = this
                 .predicate
-                .map(|p| p.resolve(session.as_ref(), snapshot.arrow_schema().to_dfschema_ref()?))
+                .map(|p| {
+                    let scan_config = DeltaScanConfig::new_from_session(session.as_ref());
+                    let predicate_schema = scan_config
+                        .table_schema(snapshot.table_configuration())?
+                        .to_dfschema_ref()?;
+                    p.resolve(session.as_ref(), predicate_schema)
+                })
                 .transpose()?;
 
             let predicate = predicate.unwrap_or(lit(true));

crates/core/src/operations/update/tests.rs

@@ -2,8 +2,7 @@ use super::*;
 
 use crate::kernel::{Action, PrimitiveType, StructField, StructType};
 use crate::kernel::{DataType as DeltaDataType, ProtocolInner};
-use crate::writer::test_utils::datafusion::get_data;
-use crate::writer::test_utils::datafusion::write_batch;
+use crate::writer::test_utils::datafusion::{get_data, write_batch};
 use crate::writer::test_utils::{
     get_arrow_schema, get_delta_schema, get_record_batch, setup_table_with_configuration,
 };
@@ -124,6 +123,48 @@ async fn test_update_predicate_left_in_data() -> DeltaResult<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn test_update_string_equality_non_partition() -> DeltaResult<()> {
+    // Regression test: DF52 view types can cause predicate evaluation to compare
+    // Utf8 against Utf8View for string equality predicates.
+    let schema = Arc::new(Schema::new(vec![
+        Field::new("utf8", DataType::Utf8, true),
+        Field::new("int32", DataType::Int32, true),
+    ]));
+
+    let batch = RecordBatch::try_new(
+        schema,
+        vec![
+            Arc::new(StringArray::from(vec![Some("0"), Some("1"), Some("2")])),
+            Arc::new(Int32Array::from(vec![0, 1, 2])),
+        ],
+    )?;
+
+    let table = DeltaTable::new_in_memory().write(vec![batch]).await?;
+
+    let (table, _metrics) = table
+        .update()
+        .with_predicate("utf8 = '1'")
+        .with_update(
+            "utf8",
+            "CASE WHEN utf8 = '1' THEN 'hello world' ELSE utf8 END",
+        )
+        .await?;
+
+    let expected = [
+        "+-------------+-------+",
+        "| utf8        | int32 |",
+        "+-------------+-------+",
+        "| 0           | 0     |",
+        "| 2           | 2     |",
+        "| hello world | 1     |",
+        "+-------------+-------+",
+    ];
+    let actual = get_data(&table).await;
+    assert_batches_sorted_eq!(&expected, &actual);
+    Ok(())
+}
+
 #[tokio::test]
 async fn test_update_no_predicate() {
     let schema = get_arrow_schema(&None);