Add NameMapping. Still need to populate from table metadata. Want to test with DataFusion Comet first.

Author: mbutrovich

crates/iceberg/src/arrow/delete_filter.rs

@@ -341,6 +341,7 @@ pub(crate) mod tests {
                 deletes: vec![pos_del_1, pos_del_2.clone()],
                 partition: None,
                 partition_spec: None,
+            name_mapping: None,
             },
             FileScanTask {
                 start: 0,
@@ -354,6 +355,7 @@ pub(crate) mod tests {
                 deletes: vec![pos_del_3],
                 partition: None,
                 partition_spec: None,
+            name_mapping: None,
             },
         ];
 

crates/iceberg/src/arrow/reader.rs

@@ -243,6 +243,7 @@ impl ArrowReader {
                     task.project_field_ids(),
                     task.partition_spec.clone(),
                     task.partition.clone(),
+                    task.name_mapping.clone(),
                 )
             } else {
                 // Fallback to build without partition data
@@ -1961,6 +1962,7 @@ message schema {
                 deletes: vec![],
                 partition: None,
                 partition_spec: None,
+            name_mapping: None,
             })]
             .into_iter(),
         )) as FileScanTaskStream;
@@ -2281,6 +2283,7 @@ message schema {
             deletes: vec![],
             partition: None,
             partition_spec: None,
+            name_mapping: None,
         };
 
         // Task 2: read the second and third row groups
@@ -2296,6 +2299,7 @@ message schema {
             deletes: vec![],
             partition: None,
             partition_spec: None,
+            name_mapping: None,
         };
 
         let tasks1 = Box::pin(futures::stream::iter(vec![Ok(task1)])) as FileScanTaskStream;
@@ -2427,6 +2431,7 @@ message schema {
                 deletes: vec![],
                 partition: None,
                 partition_spec: None,
+            name_mapping: None,
             })]
             .into_iter(),
         )) as FileScanTaskStream;
@@ -2597,6 +2602,7 @@ message schema {
             }],
             partition: None,
             partition_spec: None,
+            name_mapping: None,
         };
 
         let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream;
@@ -2814,6 +2820,7 @@ message schema {
             }],
             partition: None,
             partition_spec: None,
+            name_mapping: None,
         };
 
         let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream;
@@ -3024,6 +3031,7 @@ message schema {
             }],
             partition: None,
             partition_spec: None,
+            name_mapping: None,
         };
 
         let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream;
@@ -3126,6 +3134,7 @@ message schema {
                 deletes: vec![],
                 partition: None,
                 partition_spec: None,
+            name_mapping: None,
             })]
             .into_iter(),
         )) as FileScanTaskStream;
@@ -3222,6 +3231,7 @@ message schema {
                 deletes: vec![],
                 partition: None,
                 partition_spec: None,
+            name_mapping: None,
             })]
             .into_iter(),
         )) as FileScanTaskStream;
@@ -3307,6 +3317,7 @@ message schema {
                 deletes: vec![],
                 partition: None,
                 partition_spec: None,
+            name_mapping: None,
             })]
             .into_iter(),
         )) as FileScanTaskStream;
@@ -3406,6 +3417,7 @@ message schema {
                 deletes: vec![],
                 partition: None,
                 partition_spec: None,
+            name_mapping: None,
             })]
             .into_iter(),
         )) as FileScanTaskStream;
@@ -3534,6 +3546,7 @@ message schema {
                 deletes: vec![],
                 partition: None,
                 partition_spec: None,
+            name_mapping: None,
             })]
             .into_iter(),
         )) as FileScanTaskStream;
@@ -3629,6 +3642,7 @@ message schema {
                 deletes: vec![],
                 partition: None,
                 partition_spec: None,
+            name_mapping: None,
             })]
             .into_iter(),
         )) as FileScanTaskStream;
@@ -3737,6 +3751,7 @@ message schema {
                 deletes: vec![],
                 partition: None,
                 partition_spec: None,
+            name_mapping: None,
             })]
             .into_iter(),
         )) as FileScanTaskStream;
@@ -3875,6 +3890,7 @@ message schema {
                 deletes: vec![],
                 partition: Some(partition_data),
                 partition_spec: Some(partition_spec),
+                name_mapping: None,
             })]
             .into_iter(),
         )) as FileScanTaskStream;