fix: fix plan files with deletes (#61)

* fix plan files with deletes

* fmt

Author: chenzl25

crates/iceberg/src/scan/context.rs

@@ -269,7 +269,8 @@ impl PlanContext {
         };
 
         // TODO: Ideally we could ditch this intermediate Vec as we return an iterator.
-        let mut filtered_mfcs = vec![];
+        let mut filtered_deletes_mfcs = vec![];
+        let mut filtered_data_mfcs = vec![];
 
         for manifest_file in &manifest_files {
             let (delete_file_idx, tx) = if manifest_file.content == ManifestContentType::Deletes {
@@ -313,10 +314,22 @@ impl PlanContext {
                 filter_fn.clone(),
             );
 
-            filtered_mfcs.push(Ok(mfc));
+            match manifest_file.content {
+                ManifestContentType::Deletes => {
+                    filtered_deletes_mfcs.push(Ok(mfc));
+                }
+                ManifestContentType::Data => {
+                    filtered_data_mfcs.push(Ok(mfc));
+                }
+            }
         }
 
-        Ok(Box::new(filtered_mfcs.into_iter()))
+        // Push deletes manifest first then data manifest files.
+        Ok(Box::new(
+            filtered_deletes_mfcs
+                .into_iter()
+                .chain(filtered_data_mfcs.into_iter()),
+        ))
     }
 
     fn create_manifest_file_context(

crates/iceberg/src/scan/mod.rs

@@ -456,8 +456,7 @@ impl TableScan {
                         .send(Err(error))
                         .await;
                 }
-            })
-            .await;
+            });
         }
 
         let mut channel_for_data_manifest_entry_error = file_scan_task_tx.clone();