alessandro-nori commented on code in PR #3064:
URL: https://github.com/apache/iceberg-rust/pull/3064#discussion_r3851178040
##########
crates/iceberg/src/transaction/expire_snapshots.rs:
##########
@@ -209,15 +227,112 @@ impl ExpireSnapshotsAction {
}
}
+ // The snapshots that survive this expiry. Note this is not
`retained_ids`: a snapshot named
+ // explicitly is expired even when the age/branch policy would have
retained it.
+ let surviving_ids: Vec<i64> = metadata
+ .snapshots()
+ .map(|snapshot| snapshot.snapshot_id())
+ .filter(|id| !expiring_ids.contains(id))
+ .collect();
+
let mut ids_to_remove: Vec<i64> = expiring_ids.into_iter().collect();
ids_to_remove.sort_unstable();
removed_ref_names.sort();
Ok(ExpirePlan {
ids_to_remove,
refs_to_remove: removed_ref_names,
+ surviving_ids,
})
}
+ /// Computes the `RemoveSchemas`/`RemovePartitionSpecs` updates for specs
and schemas that no
+ /// surviving snapshot references anymore.
+ ///
+ /// The current schema and the default spec are always treated as
reachable, so they are never
+ /// removed even when no snapshot uses them yet.
+ async fn clean_expired_metadata_updates(
+ &self,
+ table: &Table,
+ surviving_ids: &[i64],
+ ) -> Result<Vec<TableUpdate>> {
+ let metadata = table.metadata();
+ let mut updates = vec![];
+
+ // Reading a manifest list per surviving snapshot is the expensive
part, so skip it when
+ // there is at most one spec: the default spec is reachable by
definition, leaving nothing
+ // to remove.
+ if metadata.partition_specs_iter().len() > 1 {
+ let mut reachable_specs: HashSet<i32> =
+ HashSet::from([metadata.default_partition_spec_id()]);
+ let total_specs = metadata.partition_specs_iter().len();
+
+ let readers: Vec<_> = surviving_ids
+ .iter()
+ .filter_map(|id| metadata.snapshot_by_id(*id))
+ .map(|snapshot| table.manifest_list_reader(snapshot))
+ .collect();
+ let mut manifest_lists = stream::iter(readers)
+ .map(|reader| async move {
+ let manifest_list = reader.load().await?;
+ Ok::<HashSet<i32>, Error>(
+ manifest_list
+ .entries()
+ .iter()
+ .map(|entry| entry.partition_spec_id)
+ .collect(),
+ )
+ })
+ .buffer_unordered(available_parallelism().get());
+ while let Some(spec_ids) = manifest_lists.try_next().await? {
+ reachable_specs.extend(spec_ids);
+ // Every spec is in use, so no further reads can change the
outcome.
+ if reachable_specs.len() >= total_specs {
+ break;
+ }
+ }
+
+ let mut spec_ids: Vec<i32> = metadata
+ .partition_specs_iter()
+ .map(|spec| spec.spec_id())
+ .filter(|spec_id| !reachable_specs.contains(spec_id))
+ .collect();
+ if !spec_ids.is_empty() {
+ spec_ids.sort_unstable();
+ updates.push(TableUpdate::RemovePartitionSpecs { spec_ids });
+ }
+ }
+
+ // Unlike specs, a snapshot's schema is recorded in metadata, so this
needs no I/O. A
+ // snapshot that predates `schema-id` leaves its schema unknown;
treating it as referencing
Review Comment:
schema-id is an optional field on a snapshot and snapshots written by older
writers don't have it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
