feat: optimize plan files memory consumption (#64)

* optimize file scan task deletes by arc

* fmt

* fix

* fmt

* fmt

Author: chenzl25

crates/iceberg/src/delete_file_index.rs

@@ -34,12 +34,14 @@ pub(crate) struct DeleteFileIndex {
     ready_notify: Arc<Notify>,
 }
 
+type DeleteFileContextAndTask = (Arc<DeleteFileContext>, Arc<FileScanTask>);
+
 #[derive(Debug)]
 struct PopulatedDeleteFileIndex {
     #[allow(dead_code)]
-    global_deletes: Vec<Arc<DeleteFileContext>>,
-    eq_deletes_by_partition: HashMap<Struct, Vec<Arc<DeleteFileContext>>>,
-    pos_deletes_by_partition: HashMap<Struct, Vec<Arc<DeleteFileContext>>>,
+    global_deletes: Vec<DeleteFileContextAndTask>,
+    eq_deletes_by_partition: HashMap<Struct, Vec<DeleteFileContextAndTask>>,
+    pos_deletes_by_partition: HashMap<Struct, Vec<DeleteFileContextAndTask>>,
     // TODO: do we need this?
     // pos_deletes_by_path: HashMap<String, Vec<Arc<DeleteFileContext>>>,
 
@@ -84,7 +86,7 @@ impl DeleteFileIndex {
         &self,
         data_file: &DataFile,
         seq_num: Option<i64>,
-    ) -> Result<Vec<FileScanTask>> {
+    ) -> Result<Vec<Arc<FileScanTask>>> {
         match self.index.get() {
             Some(idx) => Ok(idx.get_deletes_for_data_file(data_file, seq_num)),
             None => {
@@ -109,15 +111,14 @@ impl PopulatedDeleteFileIndex {
     /// 3. Otherwise, the delete file is added to one of two hash maps based on its content type.
 
     fn new(files: Vec<DeleteFileContext>) -> PopulatedDeleteFileIndex {
-        let mut eq_deletes_by_partition: HashMap<Struct, Vec<Arc<DeleteFileContext>>> =
-            HashMap::default();
-        let mut pos_deletes_by_partition: HashMap<Struct, Vec<Arc<DeleteFileContext>>> =
-            HashMap::default();
+        let mut eq_deletes_by_partition = HashMap::default();
+        let mut pos_deletes_by_partition = HashMap::default();
 
-        let mut global_deletes: Vec<Arc<DeleteFileContext>> = vec![];
+        let mut global_deletes: Vec<(Arc<DeleteFileContext>, Arc<FileScanTask>)> = vec![];
 
         files.into_iter().for_each(|ctx| {
             let arc_ctx = Arc::new(ctx);
+            let file_scan_task: Arc<FileScanTask> = Arc::new(arc_ctx.as_ref().into());
 
             let partition = arc_ctx.manifest_entry.data_file().partition();
 
@@ -126,7 +127,7 @@ impl PopulatedDeleteFileIndex {
                 // TODO: confirm we're good to skip here if we encounter a pos del
                 // FIXME(Dylan): allow putting position delete to global deletes.
                 // if arc_ctx.manifest_entry.content_type() != DataContentType::PositionDeletes {
-                global_deletes.push(arc_ctx);
+                global_deletes.push((arc_ctx, file_scan_task.clone()));
                 return;
                 // }
             }
@@ -139,10 +140,10 @@ impl PopulatedDeleteFileIndex {
 
             destination_map
                 .entry(partition.clone())
-                .and_modify(|entry| {
-                    entry.push(arc_ctx.clone());
+                .and_modify(|entry: &mut Vec<DeleteFileContextAndTask>| {
+                    entry.push((arc_ctx.clone(), file_scan_task.clone()));
                 })
-                .or_insert(vec![arc_ctx.clone()]);
+                .or_insert(vec![(arc_ctx.clone(), file_scan_task)]);
         });
 
         PopulatedDeleteFileIndex {
@@ -157,29 +158,29 @@ impl PopulatedDeleteFileIndex {
         &self,
         data_file: &DataFile,
         seq_num: Option<i64>,
-    ) -> Vec<FileScanTask> {
+    ) -> Vec<Arc<FileScanTask>> {
         let mut results = vec![];
 
         self.global_deletes
             .iter()
             // filter that returns true if the provided delete file's sequence number is **greater than or equal to** `seq_num`
-            .filter(|&delete| {
+            .filter(|&(delete, _)| {
                 seq_num
                     .map(|seq_num| delete.manifest_entry.sequence_number() >= Some(seq_num))
                     .unwrap_or_else(|| true)
             })
-            .for_each(|delete| results.push(delete.as_ref().into()));
+            .for_each(|(_, task)| results.push(task.clone()));
 
         if let Some(deletes) = self.eq_deletes_by_partition.get(data_file.partition()) {
             deletes
                 .iter()
                 // filter that returns true if the provided delete file's sequence number is **greater than** `seq_num`
-                .filter(|&delete| {
+                .filter(|&(delete, _)| {
                     seq_num
                         .map(|seq_num| delete.manifest_entry.sequence_number() > Some(seq_num))
                         .unwrap_or_else(|| true)
                 })
-                .for_each(|delete| results.push(delete.as_ref().into()));
+                .for_each(|(_, task)| results.push(task.clone()));
         }
 
         // TODO: the spec states that:
@@ -190,12 +191,12 @@ impl PopulatedDeleteFileIndex {
             deletes
                 .iter()
                 // filter that returns true if the provided delete file's sequence number is **greater than or equal to** `seq_num`
-                .filter(|&delete| {
+                .filter(|&(delete, _)| {
                     seq_num
                         .map(|seq_num| delete.manifest_entry.sequence_number() >= Some(seq_num))
                         .unwrap_or_else(|| true)
                 })
-                .for_each(|delete| results.push(delete.as_ref().into()));
+                .for_each(|(_, task)| results.push(task.clone()));
         }
 
         results

crates/iceberg/src/scan/task.rs

@@ -58,7 +58,7 @@ pub struct FileScanTask {
     pub predicate: Option<BoundPredicate>,
 
     /// The list of delete files that may need to be applied to this data file
-    pub deletes: Vec<FileScanTask>,
+    pub deletes: Vec<Arc<FileScanTask>>,
     /// sequence number
     pub sequence_number: i64,
     /// equality ids