refactor: migrate merge target scan to DeltaScanNext (delta-io#4266)

# Description
Migrate merge target scan to `DeltaScanNext`. Route the merge predicate
into file skipping instead of scan filters, and match rewritten files
against full file IDs.

# Related Issue(s)
- delta-io#4239
<!---
For example:

- closes #106
--->

# Documentation

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

---------

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

Author: ethan-tyler

crates/core/src/delta_datafusion/mod.rs

@@ -71,12 +71,14 @@ pub(crate) use data_validation::{
     DataValidationExec, constraints_to_exprs, generated_columns_to_exprs, validation_predicates,
 };
 pub(crate) use find_files::*;
+pub(crate) use table_provider::next::normalize_path_as_file_id;
 pub use table_provider::{
     DeltaScan, DeltaScanConfig, DeltaScanConfigBuilder, DeltaTableProvider, TableProviderBuilder,
     next::DeltaScanExec,
 };
 pub(crate) use table_provider::{
-    DeltaScanBuilder, next::FILE_ID_COLUMN_DEFAULT, update_datafusion_session,
+    DeltaScanBuilder, next::FILE_ID_COLUMN_DEFAULT, resolve_file_column_name,
+    update_datafusion_session,
 };
 
 pub(crate) const PATH_COLUMN: &str = "__delta_rs_path";

crates/core/src/delta_datafusion/table_provider.rs

@@ -65,6 +65,41 @@ pub(crate) mod next;
 
 const PATH_COLUMN: &str = "__delta_rs_path";
 
+pub(crate) fn resolve_file_column_name(
+    input_schema: &Schema,
+    file_column_name: Option<&str>,
+) -> DeltaResult<String> {
+    let column_names: HashSet<&str> = input_schema
+        .fields()
+        .iter()
+        .map(|field| field.name().as_str())
+        .collect();
+
+    match file_column_name {
+        Some(name) => {
+            if column_names.contains(name) {
+                return Err(DeltaTableError::Generic(format!(
+                    "Unable to add file path column since column with name {name} exists"
+                )));
+            }
+
+            Ok(name.to_owned())
+        }
+        None => {
+            let prefix = PATH_COLUMN;
+            let mut idx = 0;
+            let mut name = prefix.to_owned();
+
+            while column_names.contains(name.as_str()) {
+                idx += 1;
+                name = format!("{prefix}_{idx}");
+            }
+
+            Ok(name)
+        }
+    }
+}
+
 #[derive(Debug, Clone)]
 /// Used to specify if additional metadata columns are exposed to the user
 pub struct DeltaScanConfigBuilder {
@@ -138,35 +173,10 @@ impl DeltaScanConfigBuilder {
     /// Build a DeltaScanConfig and ensure no column name conflicts occur during downstream processing
     pub fn build(&self, snapshot: &EagerSnapshot) -> DeltaResult<DeltaScanConfig> {
         let file_column_name = if self.include_file_column {
-            let input_schema = snapshot.input_schema();
-            let mut column_names: HashSet<&String> = HashSet::new();
-            for field in input_schema.fields.iter() {
-                column_names.insert(field.name());
-            }
-
-            match &self.file_column_name {
-                Some(name) => {
-                    if column_names.contains(name) {
-                        return Err(DeltaTableError::Generic(format!(
-                            "Unable to add file path column since column with name {name} exists"
-                        )));
-                    }
-
-                    Some(name.to_owned())
-                }
-                None => {
-                    let prefix = PATH_COLUMN;
-                    let mut idx = 0;
-                    let mut name = prefix.to_owned();
-
-                    while column_names.contains(&name) {
-                        idx += 1;
-                        name = format!("{prefix}_{idx}");
-                    }
-
-                    Some(name)
-                }
-            }
+            Some(resolve_file_column_name(
+                snapshot.input_schema().as_ref(),
+                self.file_column_name.as_deref(),
+            )?)
         } else {
             None
         };
@@ -1325,6 +1335,32 @@ mod tests {
         Arc::new(SessionStateBuilder::new().with_config(config).build())
     }
 
+    #[test]
+    fn test_resolve_file_column_name_avoids_collisions() {
+        let schema = Schema::new(vec![
+            Field::new("id", ArrowDataType::Int64, false),
+            Field::new(PATH_COLUMN, ArrowDataType::Utf8, true),
+            Field::new(format!("{PATH_COLUMN}_1"), ArrowDataType::Utf8, true),
+        ]);
+
+        let resolved = resolve_file_column_name(&schema, None).unwrap();
+        assert_eq!(resolved, format!("{PATH_COLUMN}_2"));
+    }
+
+    #[test]
+    fn test_resolve_file_column_name_rejects_explicit_collision() {
+        let schema = Schema::new(vec![
+            Field::new("id", ArrowDataType::Int64, false),
+            Field::new("file_col", ArrowDataType::Utf8, true),
+        ]);
+
+        let err = resolve_file_column_name(&schema, Some("file_col")).unwrap_err();
+        assert!(
+            err.to_string()
+                .contains("Unable to add file path column since column with name file_col exists")
+        );
+    }
+
     #[tokio::test]
     async fn test_insert_into_simple() {
         let table = create_test_table().await.unwrap();