Validate that files marked for deletion actually exist in the current snapshot. For CoW UPDATE/DELETE, the commit contains `FileKind::Delete` entries for files being replaced. If a concurrent commit has already removed or rewritten those files, the delete entries become stale and the commit must be rejected.
(
&self,
commit_entries: &[ManifestEntry],
latest_snapshot: &Option<Snapshot>,
)
| 796 | /// files being replaced. If a concurrent commit has already removed or rewritten |
| 797 | /// those files, the delete entries become stale and the commit must be rejected. |
| 798 | async fn validate_deleted_files( |
| 799 | &self, |
| 800 | commit_entries: &[ManifestEntry], |
| 801 | latest_snapshot: &Option<Snapshot>, |
| 802 | ) -> Result<()> { |
| 803 | let delete_entries: Vec<_> = commit_entries |
| 804 | .iter() |
| 805 | .filter(|e| *e.kind() == FileKind::Delete) |
| 806 | .collect(); |
| 807 | |
| 808 | if delete_entries.is_empty() { |
| 809 | return Ok(()); |
| 810 | } |
| 811 | |
| 812 | let snap = match latest_snapshot { |
| 813 | Some(s) => s, |
| 814 | None => { |
| 815 | let entry = &delete_entries[0]; |
| 816 | return Err(crate::Error::DataInvalid { |
| 817 | message: format!( |
| 818 | "Delete conflict: file '{}' is marked for deletion but no snapshot exists.", |
| 819 | entry.file().file_name, |
| 820 | ), |
| 821 | source: None, |
| 822 | }); |
| 823 | } |
| 824 | }; |
| 825 | |
| 826 | let partition_filter = self.build_entries_partition_filter(&delete_entries)?; |
| 827 | let scan = TableScan::new(&self.table, partition_filter, vec![], None, None, None) |
| 828 | .with_scan_all_files(); |
| 829 | let existing_entries = scan.plan_manifest_entries(snap).await?; |
| 830 | |
| 831 | let existing_files: HashSet<(&[u8], i32, &str)> = existing_entries |
| 832 | .iter() |
| 833 | .map(|e| (e.partition(), e.bucket(), e.file().file_name.as_str())) |
| 834 | .collect(); |
| 835 | |
| 836 | for entry in &delete_entries { |
| 837 | let key = ( |
| 838 | entry.partition(), |
| 839 | entry.bucket(), |
| 840 | entry.file().file_name.as_str(), |
| 841 | ); |
| 842 | if !existing_files.contains(&key) { |
| 843 | return Err(crate::Error::DataInvalid { |
| 844 | message: format!( |
| 845 | "Delete conflict: file '{}' in partition/bucket ({}) \ |
| 846 | does not exist in the current snapshot. \ |
| 847 | It may have been removed by a concurrent operation.", |
| 848 | entry.file().file_name, |
| 849 | entry.bucket(), |
| 850 | ), |
| 851 | source: None, |
| 852 | }); |
| 853 | } |
| 854 | } |
| 855 |
no test coverage detected