JingsongLi commented on code in PR #400: URL: https://github.com/apache/paimon-rust/pull/400#discussion_r3457653572
########## crates/paimon/src/table/manifest_file_merger.rs: ########## @@ -0,0 +1,919 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use crate::io::FileIO; +use crate::spec::stats::BinaryTableStats; +use crate::spec::{ + extract_datum, BinaryRow, BinaryRowBuilder, DataField, DataType, Datum, FileKind, Identifier, + Manifest, ManifestEntry, ManifestFileMeta, +}; +use crate::Result; +use std::collections::hash_map::Entry; +use std::collections::{HashMap, HashSet}; + +/// Manifest file merger with Java `ManifestFileMerger`-style full and minor compaction. +pub(crate) struct ManifestFileMerger<'a> { + file_io: &'a FileIO, + manifest_dir: &'a str, + partition_fields: &'a [DataField], + schema_id: i64, + target_file_size: i64, + full_compaction_threshold_size: i64, + merge_min_count: usize, +} + +impl<'a> ManifestFileMerger<'a> { + pub(crate) fn new( + file_io: &'a FileIO, + manifest_dir: &'a str, + partition_fields: &'a [DataField], + schema_id: i64, + target_file_size: i64, + full_compaction_threshold_size: i64, + merge_min_count: usize, + ) -> Self { + Self { + file_io, + manifest_dir, + partition_fields, + schema_id, + target_file_size, + full_compaction_threshold_size, + merge_min_count, + } + } + + pub(crate) async fn merge( + &self, + manifest_files: Vec<ManifestFileMeta>, + ) -> Result<Vec<ManifestFileMeta>> { + if manifest_files.len() <= 1 { + return Ok(manifest_files); + } + + if let Some(compacted) = self.try_full_compaction(&manifest_files).await? { + return Ok(compacted); + } + + self.try_minor_compaction(manifest_files).await + } + + fn should_full_compact(&self, manifest_files: &[ManifestFileMeta]) -> bool { + let delta_size: i64 = manifest_files + .iter() + .filter(|file| self.must_change(file)) + .map(ManifestFileMeta::file_size) + .sum(); + delta_size >= self.full_compaction_threshold_size + } + + async fn try_full_compaction( + &self, + manifest_files: &[ManifestFileMeta], + ) -> Result<Option<Vec<ManifestFileMeta>>> { + if !self.should_full_compact(manifest_files) { + return Ok(None); + } + + let delete_identifiers = self.read_deleted_identifiers(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 to_be_merged = Vec::new(); + for manifest_file in manifest_files { + if self.must_change(manifest_file) + || self.may_contain_deleted_partitions(manifest_file, &delete_partitions) + { + to_be_merged.push(manifest_file.clone()); + } else { + result.push(manifest_file.clone()); + } + } + + if to_be_merged.len() <= 1 { + return Ok(None); + } + + let mut new_files = Vec::new(); + for manifest_file in to_be_merged { + let read_result = self + .read_for_full_compaction(&manifest_file, &delete_identifiers) + .await?; + if read_result.require_change { + new_files.extend(self.write_compacted_entries(&read_result.entries).await?); Review Comment: Full compaction should feed all changed entries into one rolling writer instead of rewriting each input manifest independently. With the current loop, if a table has many small manifests and `manifest.full-compaction-threshold-size` is reached without deletes, every manifest is `must_change`, but each one is written back as its own new manifest and `merge()` returns before the minor compaction path can merge them. The result is N small manifests with new names, so full compaction does a lot of I/O while leaving the manifest count and small-file problem unchanged. Java keeps a single `RollingFileWriter` for the whole `toBeMerged` set and rolls only when `manifest.target-file-size` is reached; this path should do the same, or fall back to minor compaction when no entries were actually removed. -- 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]
