QuakeWang commented on code in PR #400:
URL: https://github.com/apache/paimon-rust/pull/400#discussion_r3445080466
##########
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:
This merge helper drops DELETE entries whose matching ADD is not in the
compacted candidate set. That is unsafe for manifest compaction: a DELETE may
refer to an ADD kept in an older/base manifest that is not being rewritten.
Dropping it can make the deleted data file visible again in later scans.
Java `FileEntry.mergeEntries` keeps unmatched DELETE entries for this
reason. We should use equivalent compaction merge semantics here: ADD/DELETE in
the same candidate set can cancel, but unmatched DELETE entries must be
preserved.
--
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]
