Resolve commit entries and merge index entries based on the plan type.
(
&self,
plan: &CommitEntriesPlan,
latest_snapshot: &Option<Snapshot>,
)
| 523 | |
| 524 | /// Resolve commit entries and merge index entries based on the plan type. |
| 525 | async fn resolve_commit( |
| 526 | &self, |
| 527 | plan: &CommitEntriesPlan, |
| 528 | latest_snapshot: &Option<Snapshot>, |
| 529 | ) -> Result<ResolvedCommit> { |
| 530 | let file_io = self.snapshot_manager.file_io(); |
| 531 | let manifest_dir = self.snapshot_manager.manifest_dir(); |
| 532 | |
| 533 | match plan { |
| 534 | CommitEntriesPlan::Direct { |
| 535 | entries, |
| 536 | new_index_entries, |
| 537 | } => { |
| 538 | if self.row_tracking_enabled { |
| 539 | self.validate_row_id_alignment(entries, latest_snapshot) |
| 540 | .await?; |
| 541 | } |
| 542 | self.validate_deleted_files(entries, latest_snapshot) |
| 543 | .await?; |
| 544 | // Auto-promote to OVERWRITE when CoW rewrites produce Delete entries. |
| 545 | // This ensures the snapshot correctly reflects file replacements. |
| 546 | let has_delete = entries.iter().any(|e| *e.kind() == FileKind::Delete); |
| 547 | let kind = if has_delete { |
| 548 | CommitKind::OVERWRITE |
| 549 | } else { |
| 550 | CommitKind::APPEND |
| 551 | }; |
| 552 | |
| 553 | let index_manifest_name = if new_index_entries.is_empty() { |
| 554 | latest_snapshot |
| 555 | .as_ref() |
| 556 | .and_then(|s| s.index_manifest().map(|s| s.to_string())) |
| 557 | } else { |
| 558 | let mut all = |
| 559 | Self::read_prev_index_entries(file_io, &manifest_dir, latest_snapshot) |
| 560 | .await?; |
| 561 | let new_keys: HashSet<(Vec<u8>, i32)> = new_index_entries |
| 562 | .iter() |
| 563 | .filter(|e| e.index_file.index_type == "HASH") |
| 564 | .map(|e| (e.partition.clone(), e.bucket)) |
| 565 | .collect(); |
| 566 | all.retain(|e| { |
| 567 | if e.index_file.index_type == "HASH" { |
| 568 | !new_keys.contains(&(e.partition.clone(), e.bucket)) |
| 569 | } else { |
| 570 | true |
| 571 | } |
| 572 | }); |
| 573 | all.extend_from_slice(new_index_entries); |
| 574 | Self::write_index_manifest(file_io, &manifest_dir, &all).await? |
| 575 | }; |
| 576 | |
| 577 | Ok(ResolvedCommit { |
| 578 | entries: entries.clone(), |
| 579 | kind, |
| 580 | index_manifest_name, |
| 581 | }) |
| 582 | } |
no test coverage detected