Aitozi commented on code in PR #400:
URL: https://github.com/apache/paimon-rust/pull/400#discussion_r3445502404
##########
crates/paimon/src/table/table_commit.rs:
##########
@@ -541,6 +554,262 @@ impl TableCommit {
self.snapshot_commit.commit(&snapshot, &statistics).await
}
+ async fn compact_manifest_files_if_needed(
+ &self,
+ file_io: &FileIO,
+ manifest_dir: &str,
+ manifest_files: Vec<ManifestFileMeta>,
+ ) -> Result<Vec<ManifestFileMeta>> {
+ if manifest_files.len() <= 1 {
+ return Ok(manifest_files);
+ }
+
+ if let Some(compacted) = self
+ .full_compact_manifest_files(file_io, manifest_dir,
&manifest_files)
+ .await?
+ {
+ return Ok(compacted);
+ }
+
+ self.minor_compact_manifest_files(file_io, manifest_dir,
manifest_files)
+ .await
+ }
+
+ fn should_full_compact_manifests(&self, manifest_files:
&[ManifestFileMeta]) -> bool {
+ let delta_size: i64 = manifest_files
+ .iter()
+ .filter(|file| {
+ file.num_deleted_files() > 0 || file.file_size() <
self.manifest_target_file_size
+ })
+ .map(ManifestFileMeta::file_size)
+ .sum();
+ delta_size >= self.manifest_full_compaction_threshold_size
+ }
+
+ async fn full_compact_manifest_files(
+ &self,
+ file_io: &FileIO,
+ manifest_dir: &str,
+ manifest_files: &[ManifestFileMeta],
+ ) -> Result<Option<Vec<ManifestFileMeta>>> {
+ if !self.should_full_compact_manifests(manifest_files) {
+ return Ok(None);
+ }
+
+ let delete_identifiers = self
+ .read_deleted_manifest_identifiers(file_io, manifest_dir,
manifest_files)
+ .await?;
+ let delete_partitions: HashSet<Vec<u8>> = delete_identifiers
+ .iter()
+ .map(|identifier| identifier.partition.clone())
+ .collect();
+
+ let mut result = Vec::new();
+ let mut candidates = Vec::new();
+ for manifest_file in manifest_files {
+ let must_change = self.manifest_must_change(manifest_file);
+ let affected_by_deletes =
+ self.manifest_may_contain_deleted_partitions(manifest_file,
&delete_partitions);
+ if must_change || affected_by_deletes {
+ candidates.push(manifest_file.clone());
+ } else {
+ result.push(manifest_file.clone());
+ }
+ }
+
+ if candidates.len() <= 1 {
+ return Ok(None);
+ }
+
+ result.extend(
+ self.merge_manifest_candidates(file_io, manifest_dir, &candidates)
+ .await?,
+ );
+ Ok(Some(result))
+ }
+
+ async fn read_deleted_manifest_identifiers(
+ &self,
+ file_io: &FileIO,
+ manifest_dir: &str,
+ manifest_files: &[ManifestFileMeta],
+ ) -> Result<HashSet<crate::spec::Identifier>> {
+ let mut identifiers = HashSet::new();
+ for manifest_file in manifest_files {
+ if manifest_file.num_deleted_files() == 0 {
+ continue;
+ }
+ let path = format!("{manifest_dir}/{}", manifest_file.file_name());
+ for entry in Manifest::read(file_io, &path).await? {
+ if *entry.kind() == FileKind::Delete {
+ identifiers.insert(entry.into_identifier());
+ }
+ }
+ }
+ Ok(identifiers)
+ }
+
+ fn manifest_must_change(&self, manifest_file: &ManifestFileMeta) -> bool {
+ manifest_file.num_deleted_files() > 0
+ || manifest_file.file_size() < self.manifest_target_file_size
+ }
+
+ fn manifest_may_contain_deleted_partitions(
+ &self,
+ manifest_file: &ManifestFileMeta,
+ delete_partitions: &HashSet<Vec<u8>>,
+ ) -> bool {
+ if delete_partitions.is_empty() {
+ return false;
+ }
+
+ let partition_fields = self.table.schema().partition_fields();
+ if partition_fields.is_empty() {
+ return true;
+ }
+
+ delete_partitions.iter().any(|partition| {
+ self.partition_may_match_manifest_stats(
+ partition,
+ manifest_file.partition_stats(),
+ &partition_fields,
+ )
+ })
+ }
+
+ fn partition_may_match_manifest_stats(
+ &self,
+ partition: &[u8],
+ stats: &BinaryTableStats,
+ partition_fields: &[crate::spec::DataField],
+ ) -> bool {
+ let Ok(partition_row) = BinaryRow::from_serialized_bytes(partition)
else {
+ return true;
+ };
+ let Ok(min_row) = BinaryRow::from_serialized_bytes(stats.min_values())
else {
+ return true;
+ };
+ let Ok(max_row) = BinaryRow::from_serialized_bytes(stats.max_values())
else {
+ return true;
+ };
+ if partition_row.arity() < partition_fields.len() as i32
+ || min_row.arity() < partition_fields.len() as i32
+ || max_row.arity() < partition_fields.len() as i32
+ {
+ return true;
+ }
+
+ for (idx, field) in partition_fields.iter().enumerate() {
+ let data_type = field.data_type();
+ let Ok(partition_datum) = extract_datum(&partition_row, idx,
data_type) else {
+ return true;
+ };
+ let Ok(min_datum) = extract_datum(&min_row, idx, data_type) else {
+ return true;
+ };
+ let Ok(max_datum) = extract_datum(&max_row, idx, data_type) else {
+ return true;
+ };
+
+ match partition_datum {
+ Some(datum) => {
+ let (Some(min), Some(max)) = (min_datum, max_datum) else {
+ return true;
+ };
+ if datum < min || datum > max {
+ return false;
+ }
+ }
+ None => {
+ if matches!(stats.null_counts().get(idx), Some(Some(0))) {
+ return false;
+ }
+ }
+ }
+ }
+
+ true
+ }
+
+ async fn minor_compact_manifest_files(
+ &self,
+ file_io: &FileIO,
+ manifest_dir: &str,
+ manifest_files: Vec<ManifestFileMeta>,
+ ) -> Result<Vec<ManifestFileMeta>> {
+ let mut result = Vec::new();
+ let mut candidates = Vec::new();
+ let mut total_size = 0;
+
+ for manifest_file in manifest_files {
+ total_size += manifest_file.file_size();
+ candidates.push(manifest_file);
+ if total_size >= self.manifest_target_file_size {
+ let merged = self
+ .merge_manifest_candidates(file_io, manifest_dir,
&candidates)
+ .await?;
+ result.extend(merged);
+ candidates.clear();
+ total_size = 0;
+ }
+ }
+
+ if candidates.len() >= self.manifest_merge_min_count {
+ let merged = self
+ .merge_manifest_candidates(file_io, manifest_dir, &candidates)
+ .await?;
+ result.extend(merged);
+ } else {
+ result.extend(candidates);
+ }
+
+ Ok(result)
+ }
+
+ async fn merge_manifest_candidates(
+ &self,
+ file_io: &FileIO,
+ manifest_dir: &str,
+ candidates: &[ManifestFileMeta],
+ ) -> Result<Vec<ManifestFileMeta>> {
+ if candidates.len() == 1 {
+ return Ok(vec![candidates[0].clone()]);
+ }
+
+ let entries = self
+ .read_manifest_entries(file_io, manifest_dir, candidates)
+ .await?;
+ let active_entries = merge_active_entries(entries);
Review Comment:
Thanks for catching this. I fixed the merge semantics to align with Java.
The minor compaction path now uses Java `FileEntry.mergeEntries`-style
behavior, so ADD/DELETE pairs in the same candidate set cancel, while unmatched
DELETE entries are preserved.
For full compaction, I aligned with Java `ManifestFileMerger`: it reads ADD
entries and filters them by collected DELETE identifiers, so delete-only
manifests are removed instead of writing DELETE entries back.
I also extracted the logic into a separate `ManifestFileMerger` module and
moved the compaction coverage there.
--
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]
