yihua commented on code in PR #931: URL: https://github.com/apache/hudi-rs/pull/931#discussion_r4427166623
########## crates/core/src/write/metadata_compaction.rs: ########## @@ -0,0 +1,1355 @@ +/* + * 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. + */ +//! Inline compaction of the metadata table (Java +//! `HoodieBackedTableMetadataWriter.compactIfNecessary`). +//! +//! Runs inside a data write's completion critical section once the metadata +//! table holds `hoodie.metadata.compact.max.delta.commits` delta commits since +//! its latest compaction. Every file slice with log files gets a new base +//! HFile. A base record no log block touches is copied as its key and value +//! bytes, never decoded; a touched record is merged with Java's +//! `HoodieMetadataPayload` semantics, which for `record_index` is the newer +//! record's bytes as they are. + +use std::collections::{BTreeMap, HashMap, HashSet}; +use std::sync::Arc; + +use apache_avro::Schema as AvroSchema; + +use crate::Result; +use crate::error::CoreError; +use crate::file_group::log_file::scanner::{HFileRecordGroup, LogFileScanner}; +use crate::hfile::{HFileBuilder, HFileReader, HFileRecord}; +use crate::metadata::commit::{HoodieCommitMetadata, HoodieWriteStat}; +use crate::metadata::compaction::{HoodieCompactionOperation, HoodieCompactionPlan}; +use crate::metadata::payload_merger::MetadataPayloadMerger; +use crate::metadata::table::encode::{ + ColumnStatsMetadata, FilesMetadataEntry, StatsIndexVersion, decode_column_stats_entry, + encode_column_stats, encode_files_record, encode_partition_stats, hoodie_metadata_schema_json, + metadata_schema, +}; +use crate::metadata::table::records::{ + FilesPartitionRecord, MetadataPartitionType, MetadataRecordType, + decode_files_partition_record_with_schema, +}; +use crate::storage::Storage; +use crate::table::Table; +use crate::timeline::instant::Action; +use crate::timeline::selector::InstantRange; + +const METADATA_BASE: &str = ".hoodie/metadata"; +const MDT_TIMELINE: &str = ".hoodie/metadata/.hoodie/timeline"; +const MAX_DELTA_COMMITS: &str = "hoodie.metadata.compact.max.delta.commits"; +const DEFAULT_MAX_DELTA_COMMITS: usize = 10; +/// Java's default `hoodie.hfile.block.size`. +const HFILE_BLOCK_SIZE: usize = 1024 * 1024; +/// Slices compacted at once. Each holds its whole base file and its output in +/// memory, so this keeps the peak near two record index shards however many +/// cores the writer has. +const MAX_CONCURRENT_SLICES: usize = 2; +/// Java `HoodieAvroHFileWriter` records a base file's key range in its file info. +const MIN_RECORD_KEY: &str = "minRecordKey"; +const MAX_RECORD_KEY: &str = "maxRecordKey"; + +/// How a metadata partition combines successive records of one key. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum MergeKind { + /// `combineFileSystemMetadata`. + Files, + /// `mergeColumnStatsRecords`; `partition_stats` encodes with its own type. + Stats { partition_stats: bool }, + /// The newer record replaces the older one. + NewerWins, +} + +impl MergeKind { + /// The partitions this writer knows the payload semantics of; others are + /// left for the engine that created them. + fn of(partition: &str) -> Option<Self> { + if partition == MetadataPartitionType::Files.partition_name() { + Some(Self::Files) + } else if partition == MetadataPartitionType::ColumnStats.partition_name() { + Some(Self::Stats { + partition_stats: false, + }) + } else if partition == MetadataPartitionType::PartitionStats.partition_name() { + Some(Self::Stats { + partition_stats: true, + }) + } else if partition == MetadataPartitionType::RecordIndex.partition_name() { + Some(Self::NewerWins) + } else { + None + } + } +} + +/// One metadata file slice to compact. +struct SliceToCompact { + partition: String, + file_id: String, + kind: MergeKind, + /// Relative to the data table base, like the log paths. + base_path: Option<String>, + log_paths: Vec<String>, +} + +impl SliceToCompact { + fn base_instant(&self) -> Option<String> { + let name = self.base_path.as_deref().map(file_name)?; + name.rsplit_once('_') + .map(|(_, rest)| rest.trim_end_matches(".hfile").to_string()) + } +} + +fn file_name(path: &str) -> &str { + path.rsplit('/').next().unwrap_or(path) +} + +/// Compact the metadata table when its delta commits since the latest +/// compaction reach the threshold. Called with the table lock held, after +/// the data instant completed; returns the compaction instant, if one ran. +/// +/// Skips while any data instant is pending: its metadata log blocks would be +/// named below the compaction instant and so attach to the superseded slice +/// (Java holds compaction behind pending data instants the same way). Review Comment: Fixed in 4cb1b6c5: the comment and the design doc now describe Java scheduling the compaction below the earliest pending data instant, and porting that rule is left as a follow-up. -- 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]
