yihua commented on code in PR #666: URL: https://github.com/apache/hudi-rs/pull/666#discussion_r4404757908
########## crates/core/src/index/record.rs: ########## @@ -0,0 +1,194 @@ +/* + * 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. + */ +//! Record-level index backed by the metadata table `record_index` partition. + +use std::collections::HashMap; +use std::sync::Arc; + +use crate::Result; +use crate::error::CoreError; +use crate::file_group::log_file::scanner::{LogFileScanner, ScanResult}; +use crate::hfile::{HFileReader, HFileRecord}; +use crate::index::{HoodieIndex, HoodieKey, RecordLocation}; +use crate::metadata::table::encode::decode_record_index_entry; +use crate::metadata::table::records::MetadataPartitionType; +use crate::storage::Storage; +use crate::table::Table; +use crate::timeline::selector::InstantRange; +use crate::write::metadata::epoch_millis_to_instant; + +/// Metadata-table record-level index (global: key = record key). +#[derive(Clone, Debug, Default)] +pub struct RecordIndex; + +impl HoodieIndex for RecordIndex { + async fn tag_location( + &self, + table: &Table, + keys: &[HoodieKey], + ) -> Result<HashMap<HoodieKey, Option<RecordLocation>>> { + let mut locations = keys + .iter() + .cloned() + .map(|key| (key, None)) + .collect::<HashMap<_, _>>(); + if keys.is_empty() { + return Ok(locations); + } + + let storage = table.file_system_view.storage.clone(); + let merged = load_record_index_map(table, storage).await?; + for key in keys { + if let Some(entry) = merged.get(&key.record_key) { + locations.insert( + key.clone(), + Some(RecordLocation { + file_id: entry.file_id.clone(), + instant_time: epoch_millis_to_instant(entry.instant_time_millis), + partition_path: entry.partition_path.clone(), + }), + ); + } + } + Ok(locations) + } +} + +struct LoadedLocation { + file_id: String, + partition_path: String, + instant_time_millis: i64, +} + +async fn load_record_index_map( + table: &Table, + storage: Arc<Storage>, +) -> Result<HashMap<String, LoadedLocation>> { + let partition = MetadataPartitionType::RecordIndex.partition_name(); + // Slice-resolved: only each file group's LIVE slice (multi-base aware + // after a Spark MDT compaction), logs in version order. + let mut base_paths = Vec::new(); + let mut log_paths = Vec::new(); + for (_, base, logs) in + crate::write::metadata::mdt_partition_latest_slices(storage.as_ref(), partition).await? + { + base_paths.push(base); + log_paths.extend(logs); + } + + let mut merged: HashMap<String, Option<LoadedLocation>> = HashMap::new(); + for base in base_paths { + let mut reader = HFileReader::open(storage.as_ref(), &base) + .await + .map_err(|e| CoreError::HFile(format!("failed to open record_index base: {e:?}")))?; + let records = reader + .collect_records() + .map_err(|e| CoreError::HFile(format!("failed to read record_index base: {e:?}")))?; + apply_hfile_records(&mut merged, &records)?; + } + if !log_paths.is_empty() { + // Only trust log blocks whose instant completed on the data timeline + // (Java getValidInstantTimestamps) — orphan MDT commits are skipped. + let valid = table.valid_metadata_instants().await?; + let scanner = LogFileScanner::new(table.hudi_configs.clone(), storage); + let range = InstantRange::exact_match(valid, "UTC"); + match scanner.scan(log_paths, &range).await? { + ScanResult::HFileRecords(records) => apply_hfile_records(&mut merged, &records)?, + ScanResult::Empty => {} + ScanResult::RecordBatches(_) => { + return Err(CoreError::MetadataTable( + "record_index logs must contain HFile data blocks".to_string(), Review Comment: Fixed in 20e317bb: record_index logs holding only delete blocks read as tombstones. ########## crates/core/src/write/metadata.rs: ########## @@ -0,0 +1,1320 @@ +/* + * 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. + */ +//! Metadata table bootstrap and files / record_index / stats partition updates. + +use std::collections::{BTreeMap, HashMap}; + +use crate::Result; +use crate::file_group::log_file::scanner::{LogFileScanner, ScanResult}; +use crate::file_group::log_file::writer::LogFileWriter; +use crate::file_group::log_file::{BlockMetadataKey, BlockType}; +use crate::hfile::{HFileReader, HFileWriter}; +use crate::metadata::commit::{HoodieCommitMetadata, HoodieWriteStat}; +use crate::metadata::table::column_stats::{ + ColumnRangeStats, aggregate_partition_stats, column_stats_for_file, column_stats_tombstones, + columns_to_index, +}; +use crate::metadata::table::encode::{ + ColumnStatsMetadata, FilesMetadataEntry, RecordIndexEntry, StatsIndexVersion, + decode_column_stats_entry, encode_all_partitions, encode_column_stats, encode_files_record, + encode_partition_stats, encode_record_index_entry, files_metadata_avro_schema_json, + hoodie_metadata_schema_json, record_index_metadata_avro_schema_json, +}; +use crate::metadata::table::hash::{column_stats_index_key, partition_stats_index_key}; +use crate::metadata::table::records::{ + FilesPartitionRecord, MetadataPartitionType, MetadataRecordType, +}; +use crate::storage::Storage; +use crate::table::{ReadOptions, Table}; +use crate::timeline::instant::{Action, Instant}; +use crate::timeline::selector::InstantRange; +use arrow_schema::Schema; + +const METADATA_BASE: &str = ".hoodie/metadata"; +const FILES_FILE_ID: &str = "files-0000-0"; +/// Java `hoodie.metadata.record.index.min.filegroup.count` default. +pub(crate) const DEFAULT_RLI_NUM_FILE_GROUPS: usize = 10; +/// Java `hoodie.metadata.index.column.stats.file.group.count` default. +pub(crate) const DEFAULT_COLUMN_STATS_NUM_FILE_GROUPS: usize = 2; +/// Java `hoodie.metadata.index.partition.stats.file.group.count` default. +pub(crate) const DEFAULT_PARTITION_STATS_NUM_FILE_GROUPS: usize = 1; + +fn column_stats_file_id(shard: usize) -> String { + format!("col-stats-{shard:04}-0") +} + +fn partition_stats_file_id(shard: usize) -> String { + format!("partition-stats-{shard:04}-0") +} + +/// A parquet (or deleted) data file whose column ranges should update MDT stats. +#[derive(Debug, Clone)] +pub struct StatsFileUpdate { + pub partition_path: String, + pub file_name: String, + pub is_deleted: bool, + /// Footer-derived ranges for adds; ignored when `is_deleted`. + pub ranges: Vec<ColumnRangeStats>, +} + +fn record_index_file_id(shard: usize) -> String { + format!("record-index-{shard:04}-0") +} + +/// Java `HoodieTableMetadataUtil.mapRecordKeyToFileGroupIndex` (31-hash, signed i32). +pub(crate) fn map_record_key_to_file_group_index( + record_key: &str, + num_file_groups: usize, +) -> usize { + if num_file_groups <= 1 { + return 0; + } + let mut h: i32 = 0; + // Java's String.hashCode iterates UTF-16 code units (charAt), not Unicode + // scalars; supplementary-plane keys must hash to the same shard as Java. + for unit in record_key.encode_utf16() { + h = h.wrapping_mul(31).wrapping_add(unit as i32); + } + (h.unsigned_abs() as usize) % num_file_groups +} + +/// Java `HoodieBackedTableMetadataWriter.generateUniqueInstantTime` on a fresh table: +/// `SOLO_COMMIT_TIMESTAMP` plus `offset` millis, zero-padded to 17 digits. +fn bootstrap_instant(offset: usize) -> String { + format!("{offset:017}") +} + +/// Create the metadata table properties and its initial partitions. +/// +/// Each MDT partition is initialized in its own deltacommit with a unique +/// instant (`00000000000000000`, `...001`, ...) in Java's enum order: +/// files, column_stats, record_index, partition_stats. +pub async fn bootstrap_metadata_table( + data_table_storage: &Storage, + table_name: &str, + table_version: isize, + record_index_enabled: bool, + column_stats_enabled: bool, + partition_stats_enabled: bool, +) -> Result<()> { + let mdt_name = format!("{table_name}_metadata"); + let checksum = crc32(format!(".{mdt_name}").as_bytes()); + // The MDT's table version follows the data table's (Java initializes the + // MDT meta client with the data write config's version). + let properties = format!( + "# Generated by hudi-rs\n\ + hoodie.table.name={mdt_name}\n\ + hoodie.table.type=MERGE_ON_READ\n\ + hoodie.table.base.file.format=HFILE\n\ + hoodie.table.recordkey.fields=key\n\ + hoodie.populate.meta.fields=false\n\ + hoodie.table.version={table_version}\n\ + hoodie.table.initial.version={table_version}\n\ + hoodie.timeline.layout.version=2\n\ + hoodie.timeline.path=timeline\n\ + hoodie.timeline.history.path=history\n\ + hoodie.archivelog.folder=history\n\ + hoodie.table.keygenerator.type=HOODIE_TABLE_METADATA\n\ + hoodie.compaction.payload.class=org.apache.hudi.metadata.HoodieMetadataPayload\n\ + hoodie.record.merge.mode=CUSTOM\n\ + hoodie.record.merge.strategy.id=00000000-0000-0000-0000-000000000000\n\ + hoodie.datasource.write.drop.partition.columns=false\n\ + hoodie.table.checksum={checksum}\n" + ); + data_table_storage + .put_file( + &format!("{METADATA_BASE}/.hoodie/hoodie.properties"), + properties.into_bytes(), + ) + .await?; + data_table_storage + .put_file( + &format!("{METADATA_BASE}/.hoodie/timeline/.keep"), + b"".as_slice(), + ) + .await?; + + // Java initializeFilesPartition on an empty table writes only the + // __all_partitions__ record with an empty partition list — no "." entry. + let files_instant = bootstrap_instant(0); + let all_partitions = encode_all_partitions(Vec::new())?; + let files_base_name = format!("{FILES_FILE_ID}_0-0-0_{files_instant}.hfile"); + let hfile = HFileWriter::write( + &[( + FilesPartitionRecord::ALL_PARTITIONS_KEY.to_string(), + all_partitions, + )], + BTreeMap::from([( + "schema".to_string(), + files_metadata_avro_schema_json()?.as_bytes().to_vec(), + )]), + ) + .map_err(|error| crate::error::CoreError::HFile(error.to_string()))?; + let files_base_size = hfile.len() as i64; + crate::write::ensure_partition_metadata( + data_table_storage, + &format!("{METADATA_BASE}/files"), + &files_instant, + ) + .await?; + data_table_storage + .put_file(&format!("{METADATA_BASE}/files/{files_base_name}"), hfile) + .await?; + write_metadata_commit( + data_table_storage, + &files_instant, + HashMap::from([( + "files".to_string(), + vec![HoodieWriteStat { + file_id: Some(FILES_FILE_ID.to_string()), + path: Some(format!("files/{files_base_name}")), + base_file: Some(files_base_name), + prev_commit: Some("null".to_string()), + num_writes: Some(1), + total_write_bytes: Some(files_base_size), + file_size_in_bytes: Some(files_base_size), + partition_path: Some("files".to_string()), + ..Default::default() + }], + )]), + ) + .await?; + + // Remaining partitions in Java MetadataPartitionType enum order, each with + // the next unique bootstrap instant and its own deltacommit. + let mut next_offset = 1usize; + if column_stats_enabled { + bootstrap_empty_hfile_partition( + data_table_storage, + &bootstrap_instant(next_offset), + MetadataPartitionType::ColumnStats.partition_name(), + DEFAULT_COLUMN_STATS_NUM_FILE_GROUPS, + &column_stats_file_id, + hoodie_metadata_schema_json()?, + ) + .await?; + next_offset += 1; + } + if record_index_enabled { + bootstrap_empty_hfile_partition( + data_table_storage, + &bootstrap_instant(next_offset), + MetadataPartitionType::RecordIndex.partition_name(), + DEFAULT_RLI_NUM_FILE_GROUPS, + &record_index_file_id, + record_index_metadata_avro_schema_json()?, + ) + .await?; + next_offset += 1; + } + if partition_stats_enabled { + bootstrap_empty_hfile_partition( + data_table_storage, + &bootstrap_instant(next_offset), + MetadataPartitionType::PartitionStats.partition_name(), + DEFAULT_PARTITION_STATS_NUM_FILE_GROUPS, + &partition_stats_file_id, + hoodie_metadata_schema_json()?, + ) + .await?; + } + Ok(()) +} + +async fn bootstrap_empty_hfile_partition( + storage: &Storage, + instant: &str, + partition_name: &str, + num_file_groups: usize, + file_id_fn: &dyn Fn(usize) -> String, + schema_json: &str, +) -> Result<()> { + crate::write::ensure_partition_metadata( + storage, + &format!("{METADATA_BASE}/{partition_name}"), + instant, + ) + .await?; + let mut stats = Vec::with_capacity(num_file_groups); + for shard in 0..num_file_groups { + let file_id = file_id_fn(shard); + let base_name = format!("{file_id}_0-0-0_{instant}.hfile"); + let empty = HFileWriter::write( + &[], + BTreeMap::from([("schema".to_string(), schema_json.as_bytes().to_vec())]), + ) + .map_err(|error| crate::error::CoreError::HFile(error.to_string()))?; + let size = empty.len() as i64; + storage + .put_file( + &format!("{METADATA_BASE}/{partition_name}/{base_name}"), + empty, + ) + .await?; + stats.push(HoodieWriteStat { + file_id: Some(file_id), + path: Some(format!("{partition_name}/{base_name}")), + base_file: Some(base_name), + prev_commit: Some("null".to_string()), + num_writes: Some(0), + total_write_bytes: Some(size), + file_size_in_bytes: Some(size), + partition_path: Some(partition_name.to_string()), + ..Default::default() + }); + } + write_metadata_commit( + storage, + instant, + HashMap::from([(partition_name.to_string(), stats)]), + ) + .await +} + +/// Append a files-partition HFile data block to the metadata table log file. +/// +/// Returns the `files` partition write stats for the caller's single MDT +/// deltacommit (see [`write_metadata_commit`]). +pub async fn update_files_partition( + data_storage: &Storage, + instant: &str, + files: &[(String, String, i64)], +) -> Result<Vec<HoodieWriteStat>> { + let entries = files + .iter() + .map(|(partition, file_name, size)| (partition.clone(), file_name.clone(), *size, false)) + .collect::<Vec<_>>(); + update_files_partition_entries(data_storage, instant, &entries).await +} + +/// Apply file additions and deletions to the metadata table files partition. +/// +/// Returns the `files` partition write stats for the caller's single MDT +/// deltacommit (see [`write_metadata_commit`]). +pub async fn update_files_partition_entries( + data_storage: &Storage, + instant: &str, + files: &[(String, String, i64, bool)], +) -> Result<Vec<HoodieWriteStat>> { + let mut by_partition: BTreeMap<String, Vec<FilesMetadataEntry>> = BTreeMap::new(); + for (partition, file_name, size, is_deleted) in files { + let partition = if partition.is_empty() { + FilesPartitionRecord::NON_PARTITIONED_NAME.to_string() + } else { + partition.clone() + }; + by_partition + .entry(partition) + .or_default() + .push(FilesMetadataEntry { + name: file_name.clone(), + size: *size, + is_deleted: *is_deleted, + }); + } + let partitions = by_partition.keys().cloned().collect::<Vec<_>>(); + let all_partition_entries: Vec<FilesMetadataEntry> = partitions + .iter() + .map(|name| FilesMetadataEntry { + name: name.clone(), + size: 0, + is_deleted: false, + }) + .collect(); + let mut entries = vec![( + FilesPartitionRecord::ALL_PARTITIONS_KEY.to_string(), + encode_files_record( + FilesPartitionRecord::ALL_PARTITIONS_KEY, + MetadataRecordType::AllPartitions, + all_partition_entries, + )?, + )]; + for (partition, files) in by_partition { + entries.push(( + partition.clone(), + encode_files_record(&partition, MetadataRecordType::Files, files)?, + )); + } + let schema_json = files_metadata_avro_schema_json()?.to_string(); + let hfile = HFileWriter::write( + &entries, + BTreeMap::from([("schema".to_string(), schema_json.as_bytes().to_vec())]), + ) + .map_err(|error| crate::error::CoreError::HFile(error.to_string()))?; + // Java HFile data blocks embed SCHEMA (ordinal 2) alongside INSTANT_TIME. + let block = LogFileWriter::write_log_block( + BlockType::HfileData, + HashMap::from([ + (BlockMetadataKey::InstantTime, instant.to_string()), + (BlockMetadataKey::Schema, schema_json), + ]), + &hfile, + ); + let log_name = format!(".{FILES_FILE_ID}_{instant}.log.1_0-0-0"); + let log_rel = format!("files/{log_name}"); + let log_size = block.len() as i64; + data_storage + .put_file(&format!("{METADATA_BASE}/{log_rel}"), block) + .await?; + Ok(vec![HoodieWriteStat { + file_id: Some(FILES_FILE_ID.to_string()), + path: Some(log_rel), + log_files: Some(vec![log_name]), + prev_commit: Some("null".to_string()), + num_writes: Some(entries.len() as i64), + total_write_bytes: Some(log_size), + file_size_in_bytes: Some(log_size), + partition_path: Some("files".to_string()), + ..Default::default() + }]) +} + +/// Append record-index put/delete entries. +/// +/// Puts are HFile data blocks. Deletes are v3 delete log blocks — never type-5 +/// records with null `recordIndexMetadata`, which NPEs Java's RLI reader. +/// +/// Returns the `record_index` partition write stats for the caller's single +/// MDT deltacommit (see [`write_metadata_commit`]). +pub async fn update_record_index( + data_storage: &Storage, + instant: &str, + entries: &[RecordIndexEntry], +) -> Result<Vec<HoodieWriteStat>> { + if entries.is_empty() { + return Ok(Vec::new()); + } + let num_groups = detect_rli_num_file_groups(data_storage).await?; + let schema_json = record_index_metadata_avro_schema_json()?.to_string(); + let mut puts_by_shard: HashMap<usize, Vec<(String, Vec<u8>)>> = HashMap::new(); + let mut deletes_by_shard: HashMap<usize, Vec<(String, String)>> = HashMap::new(); + for entry in entries { + let shard = map_record_key_to_file_group_index(&entry.record_key, num_groups); + if entry.is_deleted { + // DeleteRecord.partitionPath is the MDT partition (Java HoodieKey for RLI deletes). + deletes_by_shard.entry(shard).or_default().push(( + entry.record_key.clone(), + MetadataPartitionType::RecordIndex + .partition_name() + .to_string(), + )); + } else { + let bytes = encode_record_index_entry(entry)?; + puts_by_shard + .entry(shard) + .or_default() + .push((entry.record_key.clone(), bytes)); + } + } + + let shards: std::collections::BTreeSet<usize> = puts_by_shard + .keys() + .chain(deletes_by_shard.keys()) + .copied() + .collect(); + let mut stats = Vec::new(); + for shard in shards { + let file_id = record_index_file_id(shard); + let puts = puts_by_shard.remove(&shard).unwrap_or_default(); + let deletes = deletes_by_shard.remove(&shard).unwrap_or_default(); + let mut log_bytes = Vec::new(); + let mut num_writes = 0i64; + if !puts.is_empty() { + num_writes += puts.len() as i64; + let hfile = HFileWriter::write( + &puts, + BTreeMap::from([("schema".to_string(), schema_json.as_bytes().to_vec())]), + ) + .map_err(|error| crate::error::CoreError::HFile(error.to_string()))?; + log_bytes.extend(LogFileWriter::write_log_block( + BlockType::HfileData, + HashMap::from([ + (BlockMetadataKey::InstantTime, instant.to_string()), + (BlockMetadataKey::Schema, schema_json.clone()), + ]), + &hfile, + )); + } + if !deletes.is_empty() { + num_writes += deletes.len() as i64; + log_bytes.extend(crate::write::build_delete_log_block(instant, &deletes, 0)?); + } + if log_bytes.is_empty() { + continue; + } + let log_name = format!(".{file_id}_{instant}.log.1_0-0-0"); + let log_rel = format!("record_index/{log_name}"); + let log_size = log_bytes.len() as i64; + data_storage + .put_file(&format!("{METADATA_BASE}/{log_rel}"), log_bytes) + .await?; + stats.push(HoodieWriteStat { + file_id: Some(file_id), + path: Some(log_rel), + log_files: Some(vec![log_name]), + prev_commit: Some("null".to_string()), + num_writes: Some(num_writes), + total_write_bytes: Some(log_size), + file_size_in_bytes: Some(log_size), + partition_path: Some("record_index".to_string()), + ..Default::default() + }); + } + Ok(stats) +} + +async fn detect_rli_num_file_groups(storage: &Storage) -> Result<usize> { + detect_num_file_groups( + storage, + MetadataPartitionType::RecordIndex.partition_name(), + "record-index-", + DEFAULT_RLI_NUM_FILE_GROUPS, + ) + .await +} + +async fn detect_num_file_groups( Review Comment: Fixed in 8c36b334: writes are refused when the metadata table has partitions, inflight partitions, or a partitioned record index the writer doesn't maintain (documented in 7d91e592). -- 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]
