yihua commented on code in PR #666: URL: https://github.com/apache/hudi-rs/pull/666#discussion_r4404751515
########## crates/core/src/write/append.rs: ########## @@ -0,0 +1,1227 @@ +/* + * 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 std::collections::HashMap; +use std::io::Cursor; +use std::str::FromStr; +use std::sync::Arc; +use std::sync::atomic::{AtomicI64, Ordering}; + +use arrow::array::{ArrayRef, StringArray, UInt32Array}; +use arrow::record_batch::RecordBatch; +use arrow_select::take::take; +use chrono::Utc; +use parquet::arrow::ArrowWriter; + +use crate::Result; +use crate::config::table::HudiTableConfig::{ + PopulatesMetaFields, RecordMergeStrategy, TableType, TableVersion, TimelineLayoutVersion, + TimelinePath, +}; +use crate::config::table::TableTypeValue; +use crate::error::CoreError; +use crate::index::is_record_index_enabled; +use crate::merge::RecordMergeStrategyValue; +use crate::metadata::HUDI_METADATA_DIR; +use crate::metadata::commit::{HoodieCommitMetadata, HoodieWriteStat}; +use crate::metadata::meta_field::MetaField; +use crate::metadata::table::encode::RecordIndexEntry; +use crate::table::Table; +use crate::timeline::instant::{Action, Instant}; +use crate::write::keygen::{ + hoodie_keys_for_batch, hoodie_keys_for_batch_with_offset, relative_data_path, +}; +use crate::write::metadata::{ + is_column_stats_enabled, update_column_stats_partitions, update_files_partition, + update_record_index, +}; + +/// Result of an append write. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct AppendResult { + pub instant: String, + pub commit_relative_path: String, + pub base_file_path: String, + pub num_rows: usize, +} + +static LAST_EPOCH_MILLIS: AtomicI64 = AtomicI64::new(0); + +/// Append one or more record batches as a single INSERT commit. +pub async fn append_batches(table: &mut Table, batches: &[RecordBatch]) -> Result<AppendResult> { + append_batches_inner(table, batches, false).await +} + +/// Append one or more record batches, requiring append-only merge mode. +pub async fn append_batches_only( + table: &mut Table, + batches: &[RecordBatch], +) -> Result<AppendResult> { + append_batches_inner(table, batches, true).await +} + +async fn append_batches_inner( + table: &mut Table, + batches: &[RecordBatch], + require_append_only: bool, +) -> Result<AppendResult> { + table.reload_timeline_for_write().await?; + if !table.is_mor() { + if require_append_only { + ensure_append_only_strategy(table)?; + } else { + ensure_insert_allowed(table)?; + } + } + ensure_supported_table_type(table)?; + + if batches.is_empty() { + return Err(CoreError::Write( + "append requires at least one RecordBatch".to_string(), + )); + } + let schema = batches[0].schema(); + for batch in batches.iter().skip(1) { + if batch.schema().as_ref() != schema.as_ref() { + return Err(CoreError::Write( + "All RecordBatches in append must share the same schema".to_string(), + )); + } + } + ensure_append_schema_matches_table(table, schema.as_ref()).await?; + // Must be expressible in commit metadata (Avro JSON) before fencing. + arrow_schema_to_avro_json(&strip_meta_fields_from_schema(schema.as_ref()))?; + let num_rows: usize = batches.iter().map(|b| b.num_rows()).sum(); + if num_rows == 0 { + return Err(CoreError::Write( + "append requires at least one row".to_string(), + )); + } + + crate::write::keygen::validate_keygen_inputs(&table.hudi_configs, batches)?; + + let layout_two = is_layout_two(table); + let action = if table.is_mor() { + Action::DeltaCommit + } else { + Action::Commit + }; + let timeline_dir = timeline_dir(table); + let storage = table.file_system_view.storage.clone(); + let lock = crate::write::lock::lock_provider_for(table); + // Critical section 1: mint the instant and fence it, then work unlocked. + let cs1 = lock.lock().await?; + let request_instant = generate_instant_time().await; + crate::write::fence_timeline_instant( + storage.as_ref(), + &timeline_dir, + &request_instant, + action.clone(), + Vec::new(), + crate::write::inflight_commit_metadata_bytes("INSERT", layout_two)?, + ) + .await?; + drop(cs1); + + // Group rows across batches by partition path and write one base file per partition. + let mut partition_rows: HashMap<String, Vec<(usize, usize)>> = HashMap::new(); + for (batch_idx, batch) in batches.iter().enumerate() { + let keys = hoodie_keys_for_batch(&table.hudi_configs, batch, Some(&request_instant))?; + for (row_idx, key) in keys.iter().enumerate() { + partition_rows + .entry(key.partition_path.clone()) + .or_default() + .push((batch_idx, row_idx)); + } + } + + let mut write_stats: HashMap<String, Vec<HoodieWriteStat>> = HashMap::new(); + let mut files_mdt = Vec::new(); + let mut rli_entries = Vec::new(); + let mut stats_updates = Vec::new(); + let mut written_paths: Vec<String> = Vec::new(); + let mut primary_base_path = String::new(); + let max_records = max_records_per_file(table); + // Monotonic auto-key row id across partitions and size-split chunks in this commit. + let mut auto_key_row_offset = 0usize; + let mut auto_key_partition_id = 0u32; + + // Stable partition iteration so partition_id assignment is deterministic. + let mut partition_rows_sorted: Vec<_> = partition_rows.into_iter().collect(); + partition_rows_sorted.sort_by(|a, b| a.0.cmp(&b.0)); + + // Plan every output file name up front and write markers before any data + // file is put — rollback of a crashed write is driven by these markers. + let mut file_plan: HashMap<String, std::collections::VecDeque<(String, String)>> = + HashMap::new(); + let mut planned_markers = Vec::new(); + for (partition_path, row_refs) in &partition_rows_sorted { + let chunks = row_refs.len().div_ceil(max_records).max(1); + let entries = file_plan.entry(partition_path.clone()).or_default(); + for _ in 0..chunks { + let file_id = crate::write::new_file_id(); + let file_name = format!("{file_id}_0-0-0_{request_instant}.parquet"); + planned_markers.push(crate::write::markers::Marker::create( + partition_path, + &file_name, + )); + entries.push_back((file_id, file_name)); + } + } + crate::write::markers::write_markers(storage.as_ref(), &request_instant, &planned_markers) + .await?; + + // Phase 1 (sequential): slice + prepare every chunk, keeping table borrows + // out of the parallel phase. + struct ChunkPlan { + partition_path: String, + file_id: String, + file_name: String, + base_file_path: String, + row_count: i64, + chunk_keys: Vec<crate::index::HoodieKey>, + prepared: Vec<RecordBatch>, + } + let mut chunk_plans: Vec<ChunkPlan> = Vec::new(); + for (partition_path, row_refs) in partition_rows_sorted { + // BTreeMap: deterministic row order in the written file. + let batch_indices: std::collections::BTreeMap<usize, Vec<u32>> = { + let mut map: std::collections::BTreeMap<usize, Vec<u32>> = + std::collections::BTreeMap::new(); + for (batch_idx, row_idx) in &row_refs { + map.entry(*batch_idx).or_default().push(*row_idx as u32); + } + map + }; + let mut partition_batches = Vec::new(); + for (batch_idx, rows) in batch_indices { + partition_batches.push(take_rows(&batches[batch_idx], &rows)?); + } + let partition_batch = match partition_batches.pop() { + Some(only) if partition_batches.is_empty() => only, + Some(last) => { + partition_batches.push(last); + arrow::compute::concat_batches(&schema, &partition_batches)? + } + None => continue, + }; + crate::write::ensure_partition_metadata( + storage.as_ref(), + &partition_path, + &request_instant, + ) + .await?; + + // Split oversized partitions into multiple base files (Java max.file.size). + let total_rows = partition_batch.num_rows(); + let mut offset = 0usize; + while offset < total_rows { + let end = (offset + max_records).min(total_rows); + let indices: Vec<u32> = (offset as u32..end as u32).collect(); + let chunk = take_rows(&partition_batch, &indices)?; + let chunk_row_offset = auto_key_row_offset; + let chunk_keys = hoodie_keys_for_batch_with_offset( + &table.hudi_configs, + &chunk, + Some(&request_instant), + chunk_row_offset, + auto_key_partition_id, + )?; + auto_key_row_offset += chunk.num_rows(); + let (file_id, file_name) = file_plan + .get_mut(&partition_path) + .and_then(std::collections::VecDeque::pop_front) + .ok_or_else(|| { + CoreError::Write(format!( + "file plan exhausted for partition '{partition_path}'" + )) + })?; + let base_file_path = relative_data_path(&partition_path, &file_name); + if primary_base_path.is_empty() { + primary_base_path = base_file_path.clone(); + } + let prepared = prepare_batches_for_write_with_offset( + &table.hudi_configs, + &[chunk], + &request_instant, + &file_name, + chunk_row_offset, + auto_key_partition_id, + )?; + chunk_plans.push(ChunkPlan { + partition_path: partition_path.clone(), + file_id, + file_name, + base_file_path, + row_count: indices.len() as i64, + chunk_keys, + prepared, + }); + offset = end; + } + auto_key_partition_id = auto_key_partition_id.saturating_add(1); + } + + // Phase 2 (parallel): encode + put each file on the write task pool. + let props = parquet_writer_props(table); + let collect_ranges = is_column_stats_enabled(table); + let tasks = chunk_plans + .iter() + .map(|plan| { + crate::write::parquet_file_task( + storage.clone(), + props.clone(), + plan.prepared.clone(), + plan.base_file_path.clone(), + collect_ranges, + ) + }) + .collect(); + let results = + crate::write::run_write_tasks(tasks, crate::write::write_task_parallelism(table)).await; + + // Phase 3 (sequential): assemble outputs in plan order; clean up on error. + let mut first_error = None; + for (plan, result) in chunk_plans.iter().zip(results) { + match result { + Ok(output) => { + written_paths.push(plan.base_file_path.clone()); + if let Some(ranges) = output.ranges { + stats_updates.push(crate::write::metadata::StatsFileUpdate { + partition_path: plan.partition_path.clone(), + file_name: plan.file_name.clone(), + is_deleted: false, + ranges, + }); + } + write_stats + .entry(plan.partition_path.clone()) + .or_default() + .push(HoodieWriteStat { + file_id: Some(plan.file_id.clone()), + path: Some(plan.base_file_path.clone()), + // Basename only — FileGroup builder joins partition_path + name. + base_file: Some(plan.file_name.clone()), + prev_commit: Some("null".to_string()), + num_writes: Some(plan.row_count), + num_inserts: Some(plan.row_count), + total_write_bytes: Some(output.size), + file_size_in_bytes: Some(output.size), + partition_path: Some(plan.partition_path.clone()), + ..Default::default() + }); + files_mdt.push(( + plan.partition_path.clone(), + plan.file_name.clone(), + output.size, + )); + for key in &plan.chunk_keys { + rli_entries.push(RecordIndexEntry { + record_key: key.record_key.clone(), + partition_path: plan.partition_path.clone(), + file_id: plan.file_id.clone(), + instant_time_millis: crate::write::metadata::instant_to_epoch_millis_in( + &request_instant, + &crate::write::metadata::timeline_timezone(&table.hudi_configs), + ), + is_deleted: false, + }); + } + } + Err(error) => { + if first_error.is_none() { + first_error = Some(error); + } + } + } + } + if let Some(error) = first_error { + for path in &written_paths { + let _ = storage.delete_file(path).await; + } + return Err(error); + } + + let commit_metadata = HoodieCommitMetadata { + version: Some(1), + operation_type: Some("INSERT".to_string()), + partition_to_write_stats: Some(write_stats), + compacted: Some(false), + // Java HoodieCommitMetadata.extraMetadata is non-null; null → NPE in Spark. + extra_metadata: Some(HashMap::from([( + "schema".to_string(), + arrow_schema_to_avro_json(&schema)?, + )])), + }; + let commit_bytes = if layout_two { + commit_metadata.to_avro_bytes()? + } else { + commit_metadata.to_json_bytes()? + }; + + // Crash-consistency order: data files → MDT files → MDT deltacommit → + // completed data commit. Finalizing the data timeline last means readers + // fence MDT contents by completed data instants. MDT log-file writes are + // "work" (unlocked); only completion runs in critical section 2. + let mut mdt_stats: HashMap<String, Vec<HoodieWriteStat>> = HashMap::new(); + if table.is_metadata_table_enabled() { + let mdt_result = async { + mdt_stats.insert( + "files".to_string(), + update_files_partition(storage.as_ref(), &request_instant, &files_mdt).await?, + ); + if is_record_index_enabled(table) { + let stats = + update_record_index(storage.as_ref(), &request_instant, &rli_entries).await?; + if !stats.is_empty() { + mdt_stats.insert("record_index".to_string(), stats); + } + } + if is_column_stats_enabled(table) && !stats_updates.is_empty() { + let stats = update_column_stats_partitions( + table, + &request_instant, + &stats_updates, + schema.as_ref(), + &[], + ) + .await?; + mdt_stats.extend(stats); + } + Ok::<(), crate::error::CoreError>(()) + } + .await; + if let Err(error) = mdt_result { + for path in &written_paths { + let _ = storage.delete_file(path).await; + } + return Err(error); + } + } + + // Critical section 2: complete the action (MDT deltacommit + data commit + // with a completion time minted under the lock), then bookkeeping. + let cs2 = lock.lock().await?; + if table.is_metadata_table_enabled() + && let Err(error) = crate::write::metadata::write_metadata_commit( + storage.as_ref(), + &request_instant, + mdt_stats, + ) + .await + { + for path in &written_paths { + let _ = storage.delete_file(path).await; + } + return Err(error); + } + let completion = if layout_two { + Some(generate_instant_time().await) + } else { + None + }; + let instant = Instant::new_completed(request_instant.clone(), action, completion)?; + let commit_relative_path = instant.relative_path_with_base(&timeline_dir)?; + if let Err(error) = storage Review Comment: Fixed in 2992f536: data files are kept when the completed-commit PUT fails, since the PUT may have landed. ########## crates/core/src/metadata/table/column_stats.rs: ########## @@ -0,0 +1,638 @@ +/* + * 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. + */ +//! Extract column-range metadata from Parquet footers for MDT writes. + +use std::collections::HashMap; + +use arrow_schema::{DataType, Schema, TimeUnit}; +use bytes::Bytes; +use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder; +use parquet::file::statistics::Statistics as ParquetStatistics; + +use crate::Result; +use crate::error::CoreError; +use crate::metadata::meta_field::MetaField; +use crate::metadata::table::encode::{ColumnStatValue, ColumnStatsMetadata}; + +/// Per-column range extracted from a written Parquet base file. +#[derive(Debug, Clone)] +pub struct ColumnRangeStats { + pub column_name: String, + pub min_value: Option<ColumnStatValue>, + pub max_value: Option<ColumnStatValue>, + /// False when the parquet footer gave truncated/absent statistics, so the + /// bounds are UNKNOWN rather than absent. Readers treat a published record + /// with null min/max as "all nulls" and prune the file, so a record with + /// an unknown range must not be published at all (the file then falls into + /// the reader's not-indexed rescue set). + pub range_known: bool, + pub value_count: i64, + pub null_count: i64, + pub total_size: i64, + pub total_uncompressed_size: i64, +} + +/// Read file-level column ranges from in-memory Parquet bytes (Java `ParquetUtils`). +/// +/// Covers exactly [`columns_to_index`]: the three always-indexed meta fields plus +/// the first 32 supported data fields. Nested (multi-part path) columns and types +/// Spark does not index (bytes/fixed/decimal) are omitted rather than mis-encoded. +pub fn column_ranges_from_parquet_bytes(bytes: &[u8]) -> Result<Vec<ColumnRangeStats>> { + let builder = ParquetRecordBatchReaderBuilder::try_new(Bytes::copy_from_slice(bytes)) + .map_err(|e| CoreError::Write(format!("Failed to open Parquet for column stats: {e}")))?; + let schema = builder.schema().clone(); + let metadata = builder.metadata().clone(); + Ok(column_ranges_from_parquet_metadata( + metadata.as_ref(), + schema.as_ref(), + )) +} + +fn column_ranges_from_parquet_metadata( + metadata: &parquet::file::metadata::ParquetMetaData, + schema: &Schema, +) -> Vec<ColumnRangeStats> { + let type_by_name: HashMap<&str, &DataType> = schema + .fields() + .iter() + .map(|f| (f.name().as_str(), f.data_type())) + .collect(); + let indexed_columns: std::collections::HashSet<String> = + columns_to_index(schema).into_iter().collect(); + + #[derive(Default)] + struct Acc { + min: Option<ColumnStatValue>, + max: Option<ColumnStatValue>, + value_count: i64, + null_count: i64, + total_size: i64, + total_uncompressed_size: i64, + data_type: Option<DataType>, + /// A chunk carried no statistics at all, so the accumulated range does + /// not cover the whole column and must not be published as a bound. + /// Truncated bounds do NOT set this: parquet truncates min downward + /// and max upward, so they stay valid (just not tight) bounds, and + /// Java publishes them as-is (ParquetUtils.readColumnStatsFromMetadata). + range_incomplete: bool, + } + + let mut by_col: HashMap<String, Acc> = HashMap::new(); + for row_group in metadata.row_groups() { + for col_chunk in row_group.columns() { + let path = col_chunk.column_descr().path(); + if path.parts().len() != 1 { + continue; + } + let Some(col_name) = path.parts().first().map(|s| s.as_str()) else { + continue; + }; + if !indexed_columns.contains(col_name) { + continue; + } + let Some(&data_type) = type_by_name.get(col_name) else { + continue; + }; + + let acc = by_col.entry(col_name.to_string()).or_default(); + acc.data_type = Some(data_type.clone()); + acc.value_count += col_chunk.num_values(); + acc.total_size += col_chunk.compressed_size(); + acc.total_uncompressed_size += col_chunk.uncompressed_size(); + + if let Some(stats) = col_chunk.statistics() { + // Null count only when actually recorded; never fabricate + // "all null" for chunks with stats disabled. + acc.null_count += stats.null_count_opt().unwrap_or(0) as i64; + match parquet_min_max_to_stat_values(stats, data_type) { + // Truncated bounds are still bounds; publish them (the + // record is never marked tight). + Some((min, max)) => { + acc.min = merge_min(acc.min.take(), Some(min)); + acc.max = merge_max(acc.max.take(), Some(max)); + } + // Parquet emits stubbed stats for an all-null chunk: no + // bounds exist, and none are needed — it cannot widen the + // column's range (Java equates nulls to the value count). + None if stats + .null_count_opt() + .is_some_and(|nulls| nulls as i64 == col_chunk.num_values()) => {} + None => acc.range_incomplete = true, + } + } else { + // Stats disabled for the chunk: range and null count unknown. + acc.range_incomplete = true; + } + } + } + + let mut out: Vec<ColumnRangeStats> = by_col + .into_iter() + .map(|(column_name, acc)| ColumnRangeStats { + column_name, + range_known: !acc.range_incomplete, + min_value: (!acc.range_incomplete).then_some(acc.min).flatten(), + max_value: (!acc.range_incomplete).then_some(acc.max).flatten(), + value_count: acc.value_count, + null_count: acc.null_count, + total_size: acc.total_size, + total_uncompressed_size: acc.total_uncompressed_size, + }) + .collect(); + out.sort_by(|a, b| a.column_name.cmp(&b.column_name)); + out +} + +/// Build MDT column_stats payloads for a written (or deleted) file. +pub fn column_stats_for_file( + file_name: &str, + ranges: &[ColumnRangeStats], + is_deleted: bool, +) -> Vec<ColumnStatsMetadata> { + ranges + .iter() + .filter(|r| is_deleted || r.range_known || r.null_count == r.value_count) + .map(|r| ColumnStatsMetadata { + file_name: file_name.to_string(), + column_name: r.column_name.clone(), + min_value: if is_deleted { + None + } else { + r.min_value.clone() + }, + max_value: if is_deleted { + None + } else { + r.max_value.clone() + }, + value_count: if is_deleted { 0 } else { r.value_count }, + null_count: if is_deleted { 0 } else { r.null_count }, + total_size: if is_deleted { 0 } else { r.total_size }, + total_uncompressed_size: if is_deleted { + 0 + } else { + r.total_uncompressed_size + }, + is_deleted, + is_tight_bound: false, + decoded_value_type_ordinal: None, + }) + .collect() +} + +/// Tombstone column_stats for a deleted file when footer ranges are unavailable. +pub fn column_stats_tombstones( + file_name: &str, + column_names: &[String], +) -> Vec<ColumnStatsMetadata> { + column_names + .iter() + .map(|column_name| ColumnStatsMetadata { + file_name: file_name.to_string(), + column_name: column_name.clone(), + min_value: None, + max_value: None, + value_count: 0, + null_count: 0, + total_size: 0, + total_uncompressed_size: 0, + is_deleted: true, + is_tight_bound: false, + decoded_value_type_ordinal: None, + }) + .collect() +} + +/// Aggregate file-level ranges into partition-level stats (same column name). +pub fn aggregate_partition_stats( + partition_path: &str, + ranges_by_file: &[Vec<ColumnRangeStats>], + is_tight_bound: bool, +) -> Vec<ColumnStatsMetadata> { + #[derive(Default)] + struct Acc { + min: Option<ColumnStatValue>, + max: Option<ColumnStatValue>, + value_count: i64, + null_count: i64, + total_size: i64, + total_uncompressed_size: i64, + range_incomplete: bool, + } + let mut by_col: HashMap<String, Acc> = HashMap::new(); + for ranges in ranges_by_file { + for r in ranges { + let acc = by_col.entry(r.column_name.clone()).or_default(); + // One file with unknown bounds makes the partition aggregate a + // non-tight (advisory) bound: publishing it tight would let + // pruning skip a partition whose unknown file may match. + if !r.range_known && r.null_count != r.value_count { + acc.range_incomplete = true; + } + acc.min = merge_min(acc.min.take(), r.min_value.clone()); + acc.max = merge_max(acc.max.take(), r.max_value.clone()); + acc.value_count += r.value_count; + acc.null_count += r.null_count; + acc.total_size += r.total_size; + acc.total_uncompressed_size += r.total_uncompressed_size; + } + } + // Java partition_stats stores the partition path in fileName. + let file_name = if partition_path.is_empty() { + ".".to_string() + } else { + partition_path.to_string() + }; + let mut out: Vec<ColumnStatsMetadata> = by_col + .into_iter() + // A partition record with null bounds reads as "all nulls" and Spark's + // partition pruning has no not-indexed rescue set, so it would prune + // the whole partition. Publish nothing instead — unless the column + // really is all nulls, where null bounds are the correct encoding. + .filter(|(_, acc)| { + acc.min.is_some() || acc.max.is_some() || acc.null_count == acc.value_count + }) + .map(|(column_name, acc)| ColumnStatsMetadata { + file_name: file_name.clone(), + column_name, + min_value: acc.min, + max_value: acc.max, + value_count: acc.value_count, + null_count: acc.null_count, + total_size: acc.total_size, + total_uncompressed_size: acc.total_uncompressed_size, + is_deleted: false, + is_tight_bound: is_tight_bound && !acc.range_incomplete, + decoded_value_type_ordinal: None, + }) + .collect(); + out.sort_by(|a, b| a.column_name.cmp(&b.column_name)); + out +} + +/// Java `HoodieMetadataConfig.COLUMN_STATS_INDEX_MAX_COLUMNS` default: only the +/// first 32 supported data fields (schema order) are indexed. +pub const MAX_COLUMNS_TO_INDEX: usize = 32; + +/// Columns eligible for MDT indexing from a table/data schema. +/// +/// Java `HoodieTableMetadataUtil.getColumnsToIndex`: the three always-indexed +/// meta fields (when present, i.e. `populateMetaFields`) plus the first +/// [`MAX_COLUMNS_TO_INDEX`] supported non-meta fields in schema order. +pub fn columns_to_index(schema: &Schema) -> Vec<String> { + // Java META_COLS_TO_ALWAYS_INDEX_SCHEMA_MAP is a TreeMap: alphabetical order. + let mut cols: Vec<String> = [ + MetaField::CommitTime, + MetaField::PartitionPath, + MetaField::RecordKey, + ] + .iter() + .map(|m| m.as_ref().to_string()) + .filter(|name| schema.fields().iter().any(|f| f.name() == name)) + .collect(); + cols.extend( + schema + .fields() + .iter() + .filter(|f| !MetaField::is_meta_field(f.name()) && is_indexable_type(f.data_type())) + .take(MAX_COLUMNS_TO_INDEX) + .map(|f| f.name().clone()), + ); + cols +} + +/// Java `isColumnTypeSupported` for the SPARK record type: nested types, enums, +/// raw bytes/fixed — and hence decimals, stored as bytes/fixed — are not indexed. +fn is_indexable_type(data_type: &DataType) -> bool { + matches!( + data_type, + DataType::Boolean + | DataType::Int8 + | DataType::Int16 + | DataType::Int32 + | DataType::Int64 + | DataType::UInt8 + | DataType::UInt16 + | DataType::UInt32 + | DataType::UInt64 + | DataType::Float32 + | DataType::Float64 + | DataType::Utf8 + | DataType::LargeUtf8 + | DataType::Date32 + | DataType::Time64(TimeUnit::Microsecond) + | DataType::Timestamp(_, _) + ) +} + +fn parquet_min_max_to_stat_values( + stats: &ParquetStatistics, + data_type: &DataType, +) -> Option<(ColumnStatValue, ColumnStatValue)> { + match (stats, data_type) { + (ParquetStatistics::Boolean(s), DataType::Boolean) => Some(( + ColumnStatValue::Boolean(*s.min_opt()?), + ColumnStatValue::Boolean(*s.max_opt()?), + )), + (ParquetStatistics::Int32(s), DataType::Date32) => Some(( + ColumnStatValue::Date(*s.min_opt()?), + ColumnStatValue::Date(*s.max_opt()?), + )), + (ParquetStatistics::Int32(s), _) => Some(( + ColumnStatValue::Int(*s.min_opt()?), + ColumnStatValue::Int(*s.max_opt()?), + )), + (ParquetStatistics::Int64(s), DataType::Timestamp(unit, tz)) => { + let min = to_micros(*s.min_opt()?, *unit)?; + let max = to_micros(*s.max_opt()?, *unit)?; + // tz-less arrow timestamps are Java LOCAL_TIMESTAMP_* semantics. + if tz.is_some() { + Some(( + ColumnStatValue::TimestampMicros(min), + ColumnStatValue::TimestampMicros(max), + )) + } else { + Some(( + ColumnStatValue::LocalTimestampMicros(min), + ColumnStatValue::LocalTimestampMicros(max), + )) + } + } + (ParquetStatistics::Int64(s), DataType::Time64(TimeUnit::Microsecond)) => Some(( + ColumnStatValue::TimeMicros(*s.min_opt()?), + ColumnStatValue::TimeMicros(*s.max_opt()?), + )), + (ParquetStatistics::Int64(s), _) => Some(( + ColumnStatValue::Long(*s.min_opt()?), + ColumnStatValue::Long(*s.max_opt()?), + )), + (ParquetStatistics::Float(s), _) => Some(( + ColumnStatValue::Float(*s.min_opt()?), + ColumnStatValue::Float(*s.max_opt()?), + )), + (ParquetStatistics::Double(s), _) => Some(( + ColumnStatValue::Double(*s.min_opt()?), + ColumnStatValue::Double(*s.max_opt()?), + )), + (ParquetStatistics::ByteArray(s), DataType::Utf8 | DataType::LargeUtf8) => { + let min = String::from_utf8_lossy(s.min_opt()?.data()).into_owned(); + let max = String::from_utf8_lossy(s.max_opt()?.data()).into_owned(); + Some((ColumnStatValue::String(min), ColumnStatValue::String(max))) + } + (ParquetStatistics::ByteArray(s), _) => Some(( + ColumnStatValue::Bytes(s.min_opt()?.data().to_vec()), + ColumnStatValue::Bytes(s.max_opt()?.data().to_vec()), + )), + (ParquetStatistics::FixedLenByteArray(s), _) => Some(( + ColumnStatValue::Bytes(s.min_opt()?.data().to_vec()), + ColumnStatValue::Bytes(s.max_opt()?.data().to_vec()), + )), + _ => None, + } +} + +fn to_micros(value: i64, unit: TimeUnit) -> Option<i64> { + match unit { + TimeUnit::Second => value.checked_mul(1_000_000), + TimeUnit::Millisecond => value.checked_mul(1_000), + TimeUnit::Microsecond => Some(value), + TimeUnit::Nanosecond => Some(value / 1_000), + } +} + +fn merge_min(a: Option<ColumnStatValue>, b: Option<ColumnStatValue>) -> Option<ColumnStatValue> { Review Comment: Fixed in 7b08eb7f: partition_stats ranges merge in the column's own type. ########## crates/core/src/write/append.rs: ########## @@ -0,0 +1,1227 @@ +/* + * 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 std::collections::HashMap; +use std::io::Cursor; +use std::str::FromStr; +use std::sync::Arc; +use std::sync::atomic::{AtomicI64, Ordering}; + +use arrow::array::{ArrayRef, StringArray, UInt32Array}; +use arrow::record_batch::RecordBatch; +use arrow_select::take::take; +use chrono::Utc; +use parquet::arrow::ArrowWriter; + +use crate::Result; +use crate::config::table::HudiTableConfig::{ + PopulatesMetaFields, RecordMergeStrategy, TableType, TableVersion, TimelineLayoutVersion, + TimelinePath, +}; +use crate::config::table::TableTypeValue; +use crate::error::CoreError; +use crate::index::is_record_index_enabled; +use crate::merge::RecordMergeStrategyValue; +use crate::metadata::HUDI_METADATA_DIR; +use crate::metadata::commit::{HoodieCommitMetadata, HoodieWriteStat}; +use crate::metadata::meta_field::MetaField; +use crate::metadata::table::encode::RecordIndexEntry; +use crate::table::Table; +use crate::timeline::instant::{Action, Instant}; +use crate::write::keygen::{ + hoodie_keys_for_batch, hoodie_keys_for_batch_with_offset, relative_data_path, +}; +use crate::write::metadata::{ + is_column_stats_enabled, update_column_stats_partitions, update_files_partition, + update_record_index, +}; + +/// Result of an append write. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct AppendResult { + pub instant: String, + pub commit_relative_path: String, + pub base_file_path: String, + pub num_rows: usize, +} + +static LAST_EPOCH_MILLIS: AtomicI64 = AtomicI64::new(0); + +/// Append one or more record batches as a single INSERT commit. +pub async fn append_batches(table: &mut Table, batches: &[RecordBatch]) -> Result<AppendResult> { + append_batches_inner(table, batches, false).await +} + +/// Append one or more record batches, requiring append-only merge mode. +pub async fn append_batches_only( + table: &mut Table, + batches: &[RecordBatch], +) -> Result<AppendResult> { + append_batches_inner(table, batches, true).await +} + +async fn append_batches_inner( + table: &mut Table, + batches: &[RecordBatch], + require_append_only: bool, +) -> Result<AppendResult> { + table.reload_timeline_for_write().await?; + if !table.is_mor() { + if require_append_only { + ensure_append_only_strategy(table)?; + } else { + ensure_insert_allowed(table)?; + } + } + ensure_supported_table_type(table)?; + + if batches.is_empty() { + return Err(CoreError::Write( + "append requires at least one RecordBatch".to_string(), + )); + } + let schema = batches[0].schema(); + for batch in batches.iter().skip(1) { + if batch.schema().as_ref() != schema.as_ref() { + return Err(CoreError::Write( + "All RecordBatches in append must share the same schema".to_string(), + )); + } + } + ensure_append_schema_matches_table(table, schema.as_ref()).await?; + // Must be expressible in commit metadata (Avro JSON) before fencing. + arrow_schema_to_avro_json(&strip_meta_fields_from_schema(schema.as_ref()))?; + let num_rows: usize = batches.iter().map(|b| b.num_rows()).sum(); + if num_rows == 0 { + return Err(CoreError::Write( + "append requires at least one row".to_string(), + )); + } + + crate::write::keygen::validate_keygen_inputs(&table.hudi_configs, batches)?; + + let layout_two = is_layout_two(table); + let action = if table.is_mor() { + Action::DeltaCommit + } else { + Action::Commit + }; + let timeline_dir = timeline_dir(table); + let storage = table.file_system_view.storage.clone(); + let lock = crate::write::lock::lock_provider_for(table); + // Critical section 1: mint the instant and fence it, then work unlocked. + let cs1 = lock.lock().await?; + let request_instant = generate_instant_time().await; + crate::write::fence_timeline_instant( + storage.as_ref(), + &timeline_dir, + &request_instant, + action.clone(), + Vec::new(), + crate::write::inflight_commit_metadata_bytes("INSERT", layout_two)?, + ) + .await?; + drop(cs1); + + // Group rows across batches by partition path and write one base file per partition. + let mut partition_rows: HashMap<String, Vec<(usize, usize)>> = HashMap::new(); + for (batch_idx, batch) in batches.iter().enumerate() { + let keys = hoodie_keys_for_batch(&table.hudi_configs, batch, Some(&request_instant))?; + for (row_idx, key) in keys.iter().enumerate() { + partition_rows + .entry(key.partition_path.clone()) + .or_default() + .push((batch_idx, row_idx)); + } + } + + let mut write_stats: HashMap<String, Vec<HoodieWriteStat>> = HashMap::new(); + let mut files_mdt = Vec::new(); + let mut rli_entries = Vec::new(); + let mut stats_updates = Vec::new(); + let mut written_paths: Vec<String> = Vec::new(); + let mut primary_base_path = String::new(); + let max_records = max_records_per_file(table); + // Monotonic auto-key row id across partitions and size-split chunks in this commit. + let mut auto_key_row_offset = 0usize; + let mut auto_key_partition_id = 0u32; + + // Stable partition iteration so partition_id assignment is deterministic. + let mut partition_rows_sorted: Vec<_> = partition_rows.into_iter().collect(); + partition_rows_sorted.sort_by(|a, b| a.0.cmp(&b.0)); + + // Plan every output file name up front and write markers before any data + // file is put — rollback of a crashed write is driven by these markers. + let mut file_plan: HashMap<String, std::collections::VecDeque<(String, String)>> = + HashMap::new(); + let mut planned_markers = Vec::new(); + for (partition_path, row_refs) in &partition_rows_sorted { + let chunks = row_refs.len().div_ceil(max_records).max(1); + let entries = file_plan.entry(partition_path.clone()).or_default(); + for _ in 0..chunks { + let file_id = crate::write::new_file_id(); + let file_name = format!("{file_id}_0-0-0_{request_instant}.parquet"); + planned_markers.push(crate::write::markers::Marker::create( + partition_path, + &file_name, + )); + entries.push_back((file_id, file_name)); + } + } + crate::write::markers::write_markers(storage.as_ref(), &request_instant, &planned_markers) + .await?; + + // Phase 1 (sequential): slice + prepare every chunk, keeping table borrows + // out of the parallel phase. + struct ChunkPlan { + partition_path: String, + file_id: String, + file_name: String, + base_file_path: String, + row_count: i64, + chunk_keys: Vec<crate::index::HoodieKey>, + prepared: Vec<RecordBatch>, + } + let mut chunk_plans: Vec<ChunkPlan> = Vec::new(); + for (partition_path, row_refs) in partition_rows_sorted { + // BTreeMap: deterministic row order in the written file. + let batch_indices: std::collections::BTreeMap<usize, Vec<u32>> = { + let mut map: std::collections::BTreeMap<usize, Vec<u32>> = + std::collections::BTreeMap::new(); + for (batch_idx, row_idx) in &row_refs { + map.entry(*batch_idx).or_default().push(*row_idx as u32); + } + map + }; + let mut partition_batches = Vec::new(); + for (batch_idx, rows) in batch_indices { + partition_batches.push(take_rows(&batches[batch_idx], &rows)?); + } + let partition_batch = match partition_batches.pop() { + Some(only) if partition_batches.is_empty() => only, + Some(last) => { + partition_batches.push(last); + arrow::compute::concat_batches(&schema, &partition_batches)? + } + None => continue, + }; + crate::write::ensure_partition_metadata( + storage.as_ref(), + &partition_path, + &request_instant, + ) + .await?; + + // Split oversized partitions into multiple base files (Java max.file.size). + let total_rows = partition_batch.num_rows(); + let mut offset = 0usize; + while offset < total_rows { + let end = (offset + max_records).min(total_rows); + let indices: Vec<u32> = (offset as u32..end as u32).collect(); + let chunk = take_rows(&partition_batch, &indices)?; + let chunk_row_offset = auto_key_row_offset; + let chunk_keys = hoodie_keys_for_batch_with_offset( + &table.hudi_configs, + &chunk, + Some(&request_instant), + chunk_row_offset, + auto_key_partition_id, + )?; + auto_key_row_offset += chunk.num_rows(); + let (file_id, file_name) = file_plan + .get_mut(&partition_path) + .and_then(std::collections::VecDeque::pop_front) + .ok_or_else(|| { + CoreError::Write(format!( + "file plan exhausted for partition '{partition_path}'" + )) + })?; + let base_file_path = relative_data_path(&partition_path, &file_name); + if primary_base_path.is_empty() { + primary_base_path = base_file_path.clone(); + } + let prepared = prepare_batches_for_write_with_offset( + &table.hudi_configs, + &[chunk], + &request_instant, + &file_name, + chunk_row_offset, + auto_key_partition_id, + )?; + chunk_plans.push(ChunkPlan { + partition_path: partition_path.clone(), + file_id, + file_name, + base_file_path, + row_count: indices.len() as i64, + chunk_keys, + prepared, + }); + offset = end; + } + auto_key_partition_id = auto_key_partition_id.saturating_add(1); + } + + // Phase 2 (parallel): encode + put each file on the write task pool. + let props = parquet_writer_props(table); + let collect_ranges = is_column_stats_enabled(table); + let tasks = chunk_plans + .iter() + .map(|plan| { + crate::write::parquet_file_task( + storage.clone(), + props.clone(), + plan.prepared.clone(), + plan.base_file_path.clone(), + collect_ranges, + ) + }) + .collect(); + let results = + crate::write::run_write_tasks(tasks, crate::write::write_task_parallelism(table)).await; + + // Phase 3 (sequential): assemble outputs in plan order; clean up on error. + let mut first_error = None; + for (plan, result) in chunk_plans.iter().zip(results) { + match result { + Ok(output) => { + written_paths.push(plan.base_file_path.clone()); + if let Some(ranges) = output.ranges { + stats_updates.push(crate::write::metadata::StatsFileUpdate { + partition_path: plan.partition_path.clone(), + file_name: plan.file_name.clone(), + is_deleted: false, + ranges, + }); + } + write_stats + .entry(plan.partition_path.clone()) + .or_default() + .push(HoodieWriteStat { + file_id: Some(plan.file_id.clone()), + path: Some(plan.base_file_path.clone()), + // Basename only — FileGroup builder joins partition_path + name. + base_file: Some(plan.file_name.clone()), + prev_commit: Some("null".to_string()), + num_writes: Some(plan.row_count), + num_inserts: Some(plan.row_count), + total_write_bytes: Some(output.size), + file_size_in_bytes: Some(output.size), + partition_path: Some(plan.partition_path.clone()), + ..Default::default() + }); + files_mdt.push(( + plan.partition_path.clone(), + plan.file_name.clone(), + output.size, + )); + for key in &plan.chunk_keys { + rli_entries.push(RecordIndexEntry { + record_key: key.record_key.clone(), + partition_path: plan.partition_path.clone(), + file_id: plan.file_id.clone(), + instant_time_millis: crate::write::metadata::instant_to_epoch_millis_in( + &request_instant, + &crate::write::metadata::timeline_timezone(&table.hudi_configs), + ), + is_deleted: false, + }); + } + } + Err(error) => { + if first_error.is_none() { + first_error = Some(error); + } + } + } + } + if let Some(error) = first_error { + for path in &written_paths { + let _ = storage.delete_file(path).await; + } + return Err(error); + } + + let commit_metadata = HoodieCommitMetadata { + version: Some(1), + operation_type: Some("INSERT".to_string()), + partition_to_write_stats: Some(write_stats), + compacted: Some(false), + // Java HoodieCommitMetadata.extraMetadata is non-null; null → NPE in Spark. + extra_metadata: Some(HashMap::from([( + "schema".to_string(), + arrow_schema_to_avro_json(&schema)?, + )])), + }; + let commit_bytes = if layout_two { + commit_metadata.to_avro_bytes()? + } else { + commit_metadata.to_json_bytes()? + }; + + // Crash-consistency order: data files → MDT files → MDT deltacommit → + // completed data commit. Finalizing the data timeline last means readers + // fence MDT contents by completed data instants. MDT log-file writes are + // "work" (unlocked); only completion runs in critical section 2. + let mut mdt_stats: HashMap<String, Vec<HoodieWriteStat>> = HashMap::new(); + if table.is_metadata_table_enabled() { + let mdt_result = async { + mdt_stats.insert( + "files".to_string(), + update_files_partition(storage.as_ref(), &request_instant, &files_mdt).await?, + ); + if is_record_index_enabled(table) { + let stats = + update_record_index(storage.as_ref(), &request_instant, &rli_entries).await?; + if !stats.is_empty() { + mdt_stats.insert("record_index".to_string(), stats); + } + } + if is_column_stats_enabled(table) && !stats_updates.is_empty() { + let stats = update_column_stats_partitions( + table, + &request_instant, + &stats_updates, + schema.as_ref(), + &[], + ) + .await?; + mdt_stats.extend(stats); + } + Ok::<(), crate::error::CoreError>(()) + } + .await; + if let Err(error) = mdt_result { + for path in &written_paths { + let _ = storage.delete_file(path).await; + } + return Err(error); + } + } + + // Critical section 2: complete the action (MDT deltacommit + data commit + // with a completion time minted under the lock), then bookkeeping. + let cs2 = lock.lock().await?; + if table.is_metadata_table_enabled() + && let Err(error) = crate::write::metadata::write_metadata_commit( + storage.as_ref(), + &request_instant, + mdt_stats, + ) + .await + { + for path in &written_paths { + let _ = storage.delete_file(path).await; + } + return Err(error); + } + let completion = if layout_two { + Some(generate_instant_time().await) + } else { + None + }; + let instant = Instant::new_completed(request_instant.clone(), action, completion)?; + let commit_relative_path = instant.relative_path_with_base(&timeline_dir)?; + if let Err(error) = storage + .put_file_if_absent(&commit_relative_path, commit_bytes) + .await + { + for path in &written_paths { + let _ = storage.delete_file(path).await; + } + return Err(error.into()); + } + // The commit is durable from here: post-commit maintenance (markers, + // archival) and handle refresh are best-effort — a transient failure must + // not turn a committed write into an Err the caller would retry. + let _ = + crate::write::post_complete_bookkeeping(table, storage.as_ref(), &request_instant).await; + drop(cs2); + + let _ = table.timeline.reload_completed_commits().await; + table.file_system_view.clear_cache(); + + Ok(AppendResult { + instant: request_instant, + commit_relative_path, + base_file_path: primary_base_path, + num_rows, + }) +} + +fn take_rows(batch: &RecordBatch, indices: &[u32]) -> Result<RecordBatch> { + let indices = UInt32Array::from(indices.to_vec()); + let columns = batch + .columns() + .iter() + .map(|column| take(column.as_ref(), &indices, None)) + .collect::<std::result::Result<Vec<_>, _>>()?; + RecordBatch::try_new(batch.schema(), columns).map_err(Into::into) +} + +fn ensure_insert_allowed(table: &Table) -> Result<()> { + let strategy: String = table + .hudi_configs + .get_or_default(RecordMergeStrategy) + .into(); + // Insert/append is valid for both append-only and upsertable (commit/event-time) tables. + match RecordMergeStrategyValue::from_str(&strategy)? { + RecordMergeStrategyValue::AppendOnly | RecordMergeStrategyValue::OverwriteWithLatest => { + Ok(()) + } + } +} + +fn ensure_append_only_strategy(table: &Table) -> Result<()> { + let strategy: String = table + .hudi_configs + .get_or_default(RecordMergeStrategy) + .into(); + if RecordMergeStrategyValue::from_str(&strategy)? != RecordMergeStrategyValue::AppendOnly { + return Err(CoreError::Unsupported( + "append_only requires hoodie.record.merge.mode=append_only".to_string(), + )); + } + Ok(()) +} + +pub(crate) fn ensure_copy_on_write(table: &Table) -> Result<()> { + let table_type: String = table.hudi_configs.get(TableType)?.into(); + if TableTypeValue::from_str(&table_type)? != TableTypeValue::CopyOnWrite { + return Err(CoreError::Unsupported(format!( + "append is currently only supported for COPY_ON_WRITE tables. Found '{table_type}'" + ))); + } + Ok(()) +} + +fn ensure_supported_table_type(table: &Table) -> Result<()> { + let table_type: String = table.hudi_configs.get(TableType)?.into(); + match TableTypeValue::from_str(&table_type)? { + TableTypeValue::CopyOnWrite | TableTypeValue::MergeOnRead => Ok(()), + } +} + +pub(crate) fn is_layout_two(table: &Table) -> bool { + let table_version: isize = match table.hudi_configs.get(TableVersion) { + Ok(v) => v.into(), + Err(_) => return false, + }; + let layout_version: isize = match table.hudi_configs.get(TimelineLayoutVersion) { + Ok(v) => v.into(), + Err(_) => return false, + }; + table_version >= 8 && layout_version == 2 +} + +pub(crate) fn timeline_dir(table: &Table) -> String { + if is_layout_two(table) { + let timeline_path: String = table.hudi_configs.get_or_default(TimelinePath).into(); + format!("{HUDI_METADATA_DIR}/{timeline_path}") + } else { + HUDI_METADATA_DIR.to_string() + } +} + +/// Java `HoodieTimeGeneratorConfig` infers a 1 ms max expected clock skew when +/// the (default) `InProcessLockProvider` is used — the single-writer case. +const MAX_EXPECTED_CLOCK_SKEW_MS: u64 = 1; + +/// Skew-adjusting monotonic instant time generator (Java +/// `SkewAdjustingTimeGenerator` + `HoodieInstantTimeGenerator.createNewInstantTime`). +/// +/// Under a process-wide lock: capture the clock, wait out the max expected +/// clock skew, and return the captured time — retrying (never bumping +1) until +/// it exceeds every previously minted instant. Used for both requested and +/// completion timestamps, so completion times are strictly greater than the +/// requested times minted before them. Format `yyyyMMddHHmmssSSS` (UTC). +/// Whether instants are formatted in local time (Java's +/// `HoodieTimelineTimeZone.LOCAL`, the Hudi default) or UTC. Process-wide, +/// mirroring `HoodieInstantTimeGenerator.setCommitTimeZone`; set from the +/// table's `hoodie.table.timeline.timezone` when a table is created or +/// loaded for writes. Spark writers mint LOCAL-time instants by default, so +/// matching the table's declared zone keeps mixed-writer timelines ordered. +static COMMIT_TIME_IS_LOCAL: std::sync::atomic::AtomicBool = + std::sync::atomic::AtomicBool::new(true); + +pub(crate) fn set_commit_timezone(timezone: &str) { + COMMIT_TIME_IS_LOCAL.store( + !timezone.eq_ignore_ascii_case("utc"), + std::sync::atomic::Ordering::Release, + ); +} + +pub(crate) async fn generate_instant_time() -> String { + static TIME_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); + let _guard = TIME_LOCK.lock().await; + loop { + let now = Utc::now(); + tokio::time::sleep(std::time::Duration::from_millis(MAX_EXPECTED_CLOCK_SKEW_MS)).await; + let candidate_millis = now.timestamp_millis(); + // Store is race-free: all mints serialize through TIME_LOCK. + if candidate_millis > LAST_EPOCH_MILLIS.load(Ordering::Acquire) { + LAST_EPOCH_MILLIS.store(candidate_millis, Ordering::Release); + let format = "%Y%m%d%H%M%S%3f"; + return if COMMIT_TIME_IS_LOCAL.load(std::sync::atomic::Ordering::Acquire) { + now.with_timezone(&chrono::Local).format(format).to_string() + } else { + now.format(format).to_string() + }; + } + } +} + +/// Default max base-file size (Java `hoodie.parquet.max.file.size` = 120 MiB). +pub(crate) const DEFAULT_MAX_FILE_SIZE_BYTES: i64 = 120 * 1024 * 1024; +/// Default record size estimate used for insert splits. +pub(crate) const DEFAULT_RECORD_SIZE_ESTIMATE: i64 = 1024; + +pub(crate) fn max_records_per_file(table: &Table) -> usize { + let max_file_size = table + .hudi_configs + .as_options() + .get("hoodie.parquet.max.file.size") + .and_then(|v| v.parse::<i64>().ok()) + .unwrap_or(DEFAULT_MAX_FILE_SIZE_BYTES) + .max(1); + let record_size = table + .hudi_configs + .as_options() + .get("hoodie.copyonwrite.record.size.estimate") + .and_then(|v| v.parse::<i64>().ok()) + .unwrap_or(DEFAULT_RECORD_SIZE_ESTIMATE) + .max(1); + ((max_file_size / record_size) as usize).max(1) +} + +/// Parquet `WriterProperties` resolved from `hoodie.parquet.*` configs. +pub(crate) fn parquet_writer_props(table: &Table) -> parquet::file::properties::WriterProperties { + use parquet::basic::{Compression, GzipLevel, ZstdLevel}; + use parquet::file::properties::WriterProperties; + + let options = table.hudi_configs.as_options(); + let page_size = options + .get("hoodie.parquet.page.size") + .and_then(|v| v.parse::<usize>().ok()) + .unwrap_or(1024 * 1024) + .max(1); + let block_size = options + .get("hoodie.parquet.block.size") + .and_then(|v| v.parse::<i64>().ok()) + .unwrap_or(DEFAULT_MAX_FILE_SIZE_BYTES) + .max(1); + let record_size = options + .get("hoodie.copyonwrite.record.size.estimate") + .and_then(|v| v.parse::<i64>().ok()) + .unwrap_or(DEFAULT_RECORD_SIZE_ESTIMATE) + .max(1); + // parquet-rs row-group limit is row-count; approximate Java's byte block size. + let max_row_group_rows = ((block_size / record_size) as usize).max(1); + let dictionary_enabled = options + .get("hoodie.parquet.dictionary.enabled") + .map(|v| v.eq_ignore_ascii_case("true")) + .unwrap_or(true); + let codec = options + .get("hoodie.parquet.compression.codec") + .map(|s| s.as_str()) + .unwrap_or("zstd"); + let compression = match codec.to_ascii_lowercase().as_str() { + "uncompressed" | "none" => Compression::UNCOMPRESSED, + "snappy" => Compression::SNAPPY, + "gzip" | "gz" => Compression::GZIP(GzipLevel::try_new(6).unwrap_or_default()), + "lz4" => Compression::LZ4, + "brotli" => Compression::BROTLI(Default::default()), + // Default zstd — intentional vs Java's gzip; honor hoodie.parquet.compression.codec. + _ => Compression::ZSTD(ZstdLevel::try_new(3).unwrap_or_default()), + }; + + WriterProperties::builder() + .set_compression(compression) + .set_dictionary_enabled(dictionary_enabled) + .set_data_page_size_limit(page_size) + .set_dictionary_page_size_limit(page_size) + .set_max_row_group_row_count(Some(max_row_group_rows)) + .build() +} + +/// Encode batches to parquet bytes with pre-resolved properties (safe to run +/// on a blocking thread with owned inputs). +pub(crate) fn write_parquet_bytes_with_props( + props: parquet::file::properties::WriterProperties, + batches: &[RecordBatch], +) -> Result<Vec<u8>> { + let cursor = Cursor::new(Vec::new()); + let mut writer = ArrowWriter::try_new(cursor, batches[0].schema(), Some(props))?; + for batch in batches { + writer.write(batch)?; + } + let cursor = writer.into_inner()?; + Ok(cursor.into_inner()) +} + +/// Arrow→Avro schema JSON for `HoodieCommitMetadata.extraMetadata.schema`. +/// +/// Unmapped Arrow types error out — silent `"string"` fallback corrupts Spark's +/// TableSchemaResolver (Parquet types diverge from commit schema). +pub(crate) fn arrow_schema_to_avro_json(schema: &arrow_schema::Schema) -> Result<String> { + let fields: Vec<String> = schema + .fields() + .iter() + .enumerate() + .map(|(index, f)| { + ensure_legal_avro_name(f.name())?; + // Field INDEX, not name: a name-derived path is not injective + // (`a.b` and `a_b` collide), and Avro refuses a redefined record. + let avro_type = arrow_type_to_avro_type_json_named( + f.data_type(), + &format!("hoodie_record_f{index}"), + )?; + let ty = if f.is_nullable() { + format!("[\"null\",{avro_type}]") + } else { + avro_type + }; + Ok(format!( + "{{\"name\":\"{}\",\"type\":{}}}", + f.name().replace('\"', "\\\""), + ty + )) + }) + .collect::<Result<Vec<_>>>()?; + Ok(format!( + "{{\"type\":\"record\",\"name\":\"hoodie_record\",\"fields\":[{}]}}", + fields.join(",") + )) +} + +/// Avro field names must match [A-Za-z_][A-Za-z0-9_]*. A name that does not is +/// rejected rather than renamed: renaming would diverge from the parquet column +/// and silently mislabel the data, while an unparseable commit schema makes the +/// table unreadable by every engine (including this one). +fn ensure_legal_avro_name(name: &str) -> Result<()> { + let mut chars = name.chars(); + let legal = chars + .next() + .is_some_and(|c| c.is_ascii_alphabetic() || c == '_') + && chars.all(|c| c.is_ascii_alphanumeric() || c == '_'); + if legal { + return Ok(()); + } + Err(CoreError::Unsupported(format!( + "column name '{name}' is not a legal Avro name ([A-Za-z_][A-Za-z0-9_]*), \ + so it cannot be written into Hudi commit metadata" + ))) +} + +/// `path` makes every nested Avro record name unique: Avro forbids redefining +/// a named type, so a fixed name breaks any schema with two structs — Java's +/// `Schema.Parser` rejects the whole commit schema. +fn arrow_type_to_avro_type_json_named(dt: &arrow_schema::DataType, path: &str) -> Result<String> { + use arrow_schema::DataType; + match dt { + DataType::Boolean => Ok("\"boolean\"".to_string()), + DataType::Int8 | DataType::Int16 | DataType::Int32 => Ok("\"int\"".to_string()), + DataType::Int64 => Ok("\"long\"".to_string()), + DataType::Float32 => Ok("\"float\"".to_string()), + DataType::Float64 => Ok("\"double\"".to_string()), + DataType::Utf8 | DataType::LargeUtf8 => { + Ok("{\"type\":\"string\",\"avro.java.string\":\"String\"}".to_string()) + } + DataType::Binary | DataType::LargeBinary => Ok("\"bytes\"".to_string()), + DataType::Date32 => Ok("{\"type\":\"int\",\"logicalType\":\"date\"}".to_string()), + DataType::Timestamp(unit, tz) => { + // A timezone-less Arrow timestamp is Spark's TimestampNTZType, + // which maps to Avro local-timestamp-*; mapping it to the + // UTC-adjusted logical type shifts values on read. + let local = tz.is_none(); + let logical = match unit { + arrow_schema::TimeUnit::Millisecond if local => "local-timestamp-millis", + arrow_schema::TimeUnit::Microsecond if local => "local-timestamp-micros", + arrow_schema::TimeUnit::Millisecond => "timestamp-millis", + arrow_schema::TimeUnit::Microsecond => "timestamp-micros", + other => { + return Err(CoreError::Unsupported(format!( + "commit schema does not support Arrow timestamp unit {other:?}" + ))); + } + }; + Ok(format!( + "{{\"type\":\"long\",\"logicalType\":\"{logical}\"}}" + )) + } + DataType::Decimal128(precision, scale) | DataType::Decimal256(precision, scale) => { + Ok(format!( + "{{\"type\":\"bytes\",\"logicalType\":\"decimal\",\"precision\":{precision},\"scale\":{scale}}}" + )) + } + DataType::List(field) | DataType::LargeList(field) => { + let item = + arrow_type_to_avro_type_json_named(field.data_type(), &format!("{path}_item"))?; + let item = if field.is_nullable() { + format!("[\"null\",{item}]") + } else { + item + }; + Ok(format!("{{\"type\":\"array\",\"items\":{item}}}")) + } + DataType::Struct(fields) => { + let nested: Vec<String> = fields + .iter() + .enumerate() + .map(|(index, f)| { + ensure_legal_avro_name(f.name())?; + let avro_type = arrow_type_to_avro_type_json_named( + f.data_type(), + &format!("{path}f{index}"), + )?; + let ty = if f.is_nullable() { + format!("[\"null\",{avro_type}]") + } else { + avro_type + }; + Ok(format!( + "{{\"name\":\"{}\",\"type\":{}}}", + f.name().replace('\"', "\\\""), + ty + )) + }) + .collect::<Result<Vec<_>>>()?; + Ok(format!( + "{{\"type\":\"record\",\"name\":\"{path}_record\",\"fields\":[{}]}}", + nested.join(",") + )) + } + other => Err(CoreError::Unsupported(format!( + "cannot map Arrow type {other:?} to Avro for commit metadata schema" + ))), + } +} + +async fn ensure_append_schema_matches_table( + table: &Table, + batch_schema: &arrow_schema::Schema, +) -> Result<()> { + if table.timeline.completed_commits.is_empty() { + return Ok(()); + } + let table_schema = table.get_schema().await?; + let batch_data = strip_meta_fields_from_schema(batch_schema); + if schemas_equal_by_fields(&table_schema, &batch_data) { + return Ok(()); + } + Err(CoreError::Write(format!( + "append batch schema does not match table schema (table fields={:?}, batch fields={:?})", + table_schema + .fields() + .iter() + .map(|f| (f.name().clone(), f.data_type().clone())) + .collect::<Vec<_>>(), + batch_data + .fields() + .iter() + .map(|f| (f.name().clone(), f.data_type().clone())) + .collect::<Vec<_>>(), + ))) +} + +pub(crate) fn strip_meta_fields_from_schema(schema: &arrow_schema::Schema) -> arrow_schema::Schema { + let fields: Vec<_> = schema + .fields() + .iter() + .filter(|f| !f.name().starts_with("_hoodie_")) + .cloned() + .collect(); + arrow_schema::Schema::new(fields) +} + +/// Prepend the Hudi meta fields to a data schema when absent — written files +/// always carry them, so index bookkeeping must consider them. +pub(crate) fn schema_with_meta_fields(schema: &arrow_schema::Schema) -> arrow_schema::Schema { + // Match the meta fields by exact name: a data column may legitimately + // begin with `_hoodie_` (Hudi's own `_hoodie_is_deleted` marker does), and + // treating it as proof the meta fields are present would silently skip + // them. + let meta: std::collections::HashSet<&str> = + crate::metadata::meta_field::MetaField::field_names() + .into_iter() + .collect(); + if schema + .fields() + .iter() + .any(|f| meta.contains(f.name().as_str())) + { + return schema.clone(); + } + let mut fields: Vec<arrow_schema::FieldRef> = + crate::metadata::meta_field::MetaField::field_names() + .into_iter() + .map(|name| { + std::sync::Arc::new(arrow_schema::Field::new( + name, + arrow_schema::DataType::Utf8, + true, + )) + }) + .collect(); + fields.extend(schema.fields().iter().cloned()); + arrow_schema::Schema::new(fields) +} + +fn schemas_equal_by_fields(a: &arrow_schema::Schema, b: &arrow_schema::Schema) -> bool { + if a.fields().len() != b.fields().len() { + return false; + } + a.fields() + .iter() + .zip(b.fields().iter()) + .all(|(fa, fb)| fa.name() == fb.name() && fa.data_type() == fb.data_type()) +} + +pub(crate) fn prepare_batches_for_write( + hudi_configs: &crate::config::HudiConfigs, + batches: &[RecordBatch], + instant: &str, + file_name: &str, +) -> Result<Vec<RecordBatch>> { + prepare_batches_for_write_with_offset(hudi_configs, batches, instant, file_name, 0, 0) +} + +pub(crate) fn prepare_batches_for_write_with_offset( + hudi_configs: &crate::config::HudiConfigs, + batches: &[RecordBatch], + instant: &str, + file_name: &str, + row_id_offset: usize, + partition_id: u32, +) -> Result<Vec<RecordBatch>> { + let populates_meta_fields: bool = hudi_configs.get_or_default(PopulatesMetaFields).into(); + if !populates_meta_fields { + return Ok(batches.to_vec()); + } + + batches Review Comment: Fixed in c01ec3c3: auto-generated keys are numbered across all batches of a write. ########## crates/core/src/write/append.rs: ########## @@ -0,0 +1,1227 @@ +/* + * 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 std::collections::HashMap; +use std::io::Cursor; +use std::str::FromStr; +use std::sync::Arc; +use std::sync::atomic::{AtomicI64, Ordering}; + +use arrow::array::{ArrayRef, StringArray, UInt32Array}; +use arrow::record_batch::RecordBatch; +use arrow_select::take::take; +use chrono::Utc; +use parquet::arrow::ArrowWriter; + +use crate::Result; +use crate::config::table::HudiTableConfig::{ + PopulatesMetaFields, RecordMergeStrategy, TableType, TableVersion, TimelineLayoutVersion, + TimelinePath, +}; +use crate::config::table::TableTypeValue; +use crate::error::CoreError; +use crate::index::is_record_index_enabled; +use crate::merge::RecordMergeStrategyValue; +use crate::metadata::HUDI_METADATA_DIR; +use crate::metadata::commit::{HoodieCommitMetadata, HoodieWriteStat}; +use crate::metadata::meta_field::MetaField; +use crate::metadata::table::encode::RecordIndexEntry; +use crate::table::Table; +use crate::timeline::instant::{Action, Instant}; +use crate::write::keygen::{ + hoodie_keys_for_batch, hoodie_keys_for_batch_with_offset, relative_data_path, +}; +use crate::write::metadata::{ + is_column_stats_enabled, update_column_stats_partitions, update_files_partition, + update_record_index, +}; + +/// Result of an append write. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct AppendResult { + pub instant: String, + pub commit_relative_path: String, + pub base_file_path: String, + pub num_rows: usize, +} + +static LAST_EPOCH_MILLIS: AtomicI64 = AtomicI64::new(0); + +/// Append one or more record batches as a single INSERT commit. +pub async fn append_batches(table: &mut Table, batches: &[RecordBatch]) -> Result<AppendResult> { + append_batches_inner(table, batches, false).await +} + +/// Append one or more record batches, requiring append-only merge mode. +pub async fn append_batches_only( + table: &mut Table, + batches: &[RecordBatch], +) -> Result<AppendResult> { + append_batches_inner(table, batches, true).await +} + +async fn append_batches_inner( + table: &mut Table, + batches: &[RecordBatch], + require_append_only: bool, +) -> Result<AppendResult> { + table.reload_timeline_for_write().await?; + if !table.is_mor() { + if require_append_only { + ensure_append_only_strategy(table)?; + } else { + ensure_insert_allowed(table)?; + } + } + ensure_supported_table_type(table)?; + + if batches.is_empty() { + return Err(CoreError::Write( + "append requires at least one RecordBatch".to_string(), + )); + } + let schema = batches[0].schema(); + for batch in batches.iter().skip(1) { + if batch.schema().as_ref() != schema.as_ref() { + return Err(CoreError::Write( + "All RecordBatches in append must share the same schema".to_string(), + )); + } + } + ensure_append_schema_matches_table(table, schema.as_ref()).await?; + // Must be expressible in commit metadata (Avro JSON) before fencing. + arrow_schema_to_avro_json(&strip_meta_fields_from_schema(schema.as_ref()))?; + let num_rows: usize = batches.iter().map(|b| b.num_rows()).sum(); + if num_rows == 0 { + return Err(CoreError::Write( + "append requires at least one row".to_string(), + )); + } + + crate::write::keygen::validate_keygen_inputs(&table.hudi_configs, batches)?; + + let layout_two = is_layout_two(table); + let action = if table.is_mor() { + Action::DeltaCommit + } else { + Action::Commit + }; + let timeline_dir = timeline_dir(table); + let storage = table.file_system_view.storage.clone(); + let lock = crate::write::lock::lock_provider_for(table); + // Critical section 1: mint the instant and fence it, then work unlocked. + let cs1 = lock.lock().await?; + let request_instant = generate_instant_time().await; + crate::write::fence_timeline_instant( + storage.as_ref(), + &timeline_dir, + &request_instant, + action.clone(), + Vec::new(), + crate::write::inflight_commit_metadata_bytes("INSERT", layout_two)?, + ) + .await?; + drop(cs1); + + // Group rows across batches by partition path and write one base file per partition. + let mut partition_rows: HashMap<String, Vec<(usize, usize)>> = HashMap::new(); + for (batch_idx, batch) in batches.iter().enumerate() { + let keys = hoodie_keys_for_batch(&table.hudi_configs, batch, Some(&request_instant))?; + for (row_idx, key) in keys.iter().enumerate() { + partition_rows + .entry(key.partition_path.clone()) + .or_default() + .push((batch_idx, row_idx)); + } + } + + let mut write_stats: HashMap<String, Vec<HoodieWriteStat>> = HashMap::new(); + let mut files_mdt = Vec::new(); + let mut rli_entries = Vec::new(); + let mut stats_updates = Vec::new(); + let mut written_paths: Vec<String> = Vec::new(); + let mut primary_base_path = String::new(); + let max_records = max_records_per_file(table); + // Monotonic auto-key row id across partitions and size-split chunks in this commit. + let mut auto_key_row_offset = 0usize; + let mut auto_key_partition_id = 0u32; + + // Stable partition iteration so partition_id assignment is deterministic. + let mut partition_rows_sorted: Vec<_> = partition_rows.into_iter().collect(); + partition_rows_sorted.sort_by(|a, b| a.0.cmp(&b.0)); + + // Plan every output file name up front and write markers before any data + // file is put — rollback of a crashed write is driven by these markers. + let mut file_plan: HashMap<String, std::collections::VecDeque<(String, String)>> = + HashMap::new(); + let mut planned_markers = Vec::new(); + for (partition_path, row_refs) in &partition_rows_sorted { + let chunks = row_refs.len().div_ceil(max_records).max(1); + let entries = file_plan.entry(partition_path.clone()).or_default(); + for _ in 0..chunks { + let file_id = crate::write::new_file_id(); + let file_name = format!("{file_id}_0-0-0_{request_instant}.parquet"); + planned_markers.push(crate::write::markers::Marker::create( + partition_path, + &file_name, + )); + entries.push_back((file_id, file_name)); + } + } + crate::write::markers::write_markers(storage.as_ref(), &request_instant, &planned_markers) + .await?; + + // Phase 1 (sequential): slice + prepare every chunk, keeping table borrows + // out of the parallel phase. + struct ChunkPlan { + partition_path: String, + file_id: String, + file_name: String, + base_file_path: String, + row_count: i64, + chunk_keys: Vec<crate::index::HoodieKey>, + prepared: Vec<RecordBatch>, + } + let mut chunk_plans: Vec<ChunkPlan> = Vec::new(); + for (partition_path, row_refs) in partition_rows_sorted { + // BTreeMap: deterministic row order in the written file. + let batch_indices: std::collections::BTreeMap<usize, Vec<u32>> = { + let mut map: std::collections::BTreeMap<usize, Vec<u32>> = + std::collections::BTreeMap::new(); + for (batch_idx, row_idx) in &row_refs { + map.entry(*batch_idx).or_default().push(*row_idx as u32); + } + map + }; + let mut partition_batches = Vec::new(); + for (batch_idx, rows) in batch_indices { + partition_batches.push(take_rows(&batches[batch_idx], &rows)?); + } + let partition_batch = match partition_batches.pop() { + Some(only) if partition_batches.is_empty() => only, + Some(last) => { + partition_batches.push(last); + arrow::compute::concat_batches(&schema, &partition_batches)? + } + None => continue, + }; + crate::write::ensure_partition_metadata( + storage.as_ref(), + &partition_path, + &request_instant, + ) + .await?; + + // Split oversized partitions into multiple base files (Java max.file.size). + let total_rows = partition_batch.num_rows(); + let mut offset = 0usize; + while offset < total_rows { + let end = (offset + max_records).min(total_rows); + let indices: Vec<u32> = (offset as u32..end as u32).collect(); + let chunk = take_rows(&partition_batch, &indices)?; + let chunk_row_offset = auto_key_row_offset; + let chunk_keys = hoodie_keys_for_batch_with_offset( + &table.hudi_configs, + &chunk, + Some(&request_instant), + chunk_row_offset, + auto_key_partition_id, + )?; + auto_key_row_offset += chunk.num_rows(); + let (file_id, file_name) = file_plan + .get_mut(&partition_path) + .and_then(std::collections::VecDeque::pop_front) + .ok_or_else(|| { + CoreError::Write(format!( + "file plan exhausted for partition '{partition_path}'" + )) + })?; + let base_file_path = relative_data_path(&partition_path, &file_name); + if primary_base_path.is_empty() { + primary_base_path = base_file_path.clone(); + } + let prepared = prepare_batches_for_write_with_offset( + &table.hudi_configs, + &[chunk], + &request_instant, + &file_name, + chunk_row_offset, + auto_key_partition_id, + )?; + chunk_plans.push(ChunkPlan { + partition_path: partition_path.clone(), + file_id, + file_name, + base_file_path, + row_count: indices.len() as i64, + chunk_keys, + prepared, + }); + offset = end; + } + auto_key_partition_id = auto_key_partition_id.saturating_add(1); + } + + // Phase 2 (parallel): encode + put each file on the write task pool. + let props = parquet_writer_props(table); + let collect_ranges = is_column_stats_enabled(table); + let tasks = chunk_plans + .iter() + .map(|plan| { + crate::write::parquet_file_task( + storage.clone(), + props.clone(), + plan.prepared.clone(), + plan.base_file_path.clone(), + collect_ranges, + ) + }) + .collect(); + let results = + crate::write::run_write_tasks(tasks, crate::write::write_task_parallelism(table)).await; + + // Phase 3 (sequential): assemble outputs in plan order; clean up on error. + let mut first_error = None; + for (plan, result) in chunk_plans.iter().zip(results) { + match result { + Ok(output) => { + written_paths.push(plan.base_file_path.clone()); + if let Some(ranges) = output.ranges { + stats_updates.push(crate::write::metadata::StatsFileUpdate { + partition_path: plan.partition_path.clone(), + file_name: plan.file_name.clone(), + is_deleted: false, + ranges, + }); + } + write_stats + .entry(plan.partition_path.clone()) + .or_default() + .push(HoodieWriteStat { + file_id: Some(plan.file_id.clone()), + path: Some(plan.base_file_path.clone()), + // Basename only — FileGroup builder joins partition_path + name. + base_file: Some(plan.file_name.clone()), + prev_commit: Some("null".to_string()), + num_writes: Some(plan.row_count), + num_inserts: Some(plan.row_count), + total_write_bytes: Some(output.size), + file_size_in_bytes: Some(output.size), + partition_path: Some(plan.partition_path.clone()), + ..Default::default() + }); + files_mdt.push(( + plan.partition_path.clone(), + plan.file_name.clone(), + output.size, + )); + for key in &plan.chunk_keys { + rli_entries.push(RecordIndexEntry { + record_key: key.record_key.clone(), + partition_path: plan.partition_path.clone(), + file_id: plan.file_id.clone(), + instant_time_millis: crate::write::metadata::instant_to_epoch_millis_in( + &request_instant, + &crate::write::metadata::timeline_timezone(&table.hudi_configs), + ), + is_deleted: false, + }); + } + } + Err(error) => { + if first_error.is_none() { + first_error = Some(error); + } + } + } + } + if let Some(error) = first_error { + for path in &written_paths { + let _ = storage.delete_file(path).await; + } + return Err(error); + } + + let commit_metadata = HoodieCommitMetadata { + version: Some(1), + operation_type: Some("INSERT".to_string()), + partition_to_write_stats: Some(write_stats), + compacted: Some(false), + // Java HoodieCommitMetadata.extraMetadata is non-null; null → NPE in Spark. + extra_metadata: Some(HashMap::from([( + "schema".to_string(), + arrow_schema_to_avro_json(&schema)?, + )])), + }; + let commit_bytes = if layout_two { + commit_metadata.to_avro_bytes()? + } else { + commit_metadata.to_json_bytes()? + }; + + // Crash-consistency order: data files → MDT files → MDT deltacommit → + // completed data commit. Finalizing the data timeline last means readers + // fence MDT contents by completed data instants. MDT log-file writes are + // "work" (unlocked); only completion runs in critical section 2. + let mut mdt_stats: HashMap<String, Vec<HoodieWriteStat>> = HashMap::new(); + if table.is_metadata_table_enabled() { + let mdt_result = async { + mdt_stats.insert( + "files".to_string(), + update_files_partition(storage.as_ref(), &request_instant, &files_mdt).await?, + ); + if is_record_index_enabled(table) { + let stats = + update_record_index(storage.as_ref(), &request_instant, &rli_entries).await?; + if !stats.is_empty() { + mdt_stats.insert("record_index".to_string(), stats); + } + } + if is_column_stats_enabled(table) && !stats_updates.is_empty() { + let stats = update_column_stats_partitions( + table, + &request_instant, + &stats_updates, + schema.as_ref(), + &[], + ) + .await?; + mdt_stats.extend(stats); + } + Ok::<(), crate::error::CoreError>(()) + } + .await; + if let Err(error) = mdt_result { + for path in &written_paths { + let _ = storage.delete_file(path).await; + } + return Err(error); + } + } + + // Critical section 2: complete the action (MDT deltacommit + data commit + // with a completion time minted under the lock), then bookkeeping. + let cs2 = lock.lock().await?; + if table.is_metadata_table_enabled() + && let Err(error) = crate::write::metadata::write_metadata_commit( + storage.as_ref(), + &request_instant, + mdt_stats, + ) + .await + { + for path in &written_paths { + let _ = storage.delete_file(path).await; + } + return Err(error); + } + let completion = if layout_two { + Some(generate_instant_time().await) + } else { + None + }; + let instant = Instant::new_completed(request_instant.clone(), action, completion)?; + let commit_relative_path = instant.relative_path_with_base(&timeline_dir)?; + if let Err(error) = storage + .put_file_if_absent(&commit_relative_path, commit_bytes) + .await + { + for path in &written_paths { + let _ = storage.delete_file(path).await; + } + return Err(error.into()); + } + // The commit is durable from here: post-commit maintenance (markers, + // archival) and handle refresh are best-effort — a transient failure must + // not turn a committed write into an Err the caller would retry. + let _ = + crate::write::post_complete_bookkeeping(table, storage.as_ref(), &request_instant).await; + drop(cs2); + + let _ = table.timeline.reload_completed_commits().await; + table.file_system_view.clear_cache(); + + Ok(AppendResult { + instant: request_instant, + commit_relative_path, + base_file_path: primary_base_path, + num_rows, + }) +} + +fn take_rows(batch: &RecordBatch, indices: &[u32]) -> Result<RecordBatch> { + let indices = UInt32Array::from(indices.to_vec()); + let columns = batch + .columns() + .iter() + .map(|column| take(column.as_ref(), &indices, None)) + .collect::<std::result::Result<Vec<_>, _>>()?; + RecordBatch::try_new(batch.schema(), columns).map_err(Into::into) +} + +fn ensure_insert_allowed(table: &Table) -> Result<()> { + let strategy: String = table + .hudi_configs + .get_or_default(RecordMergeStrategy) + .into(); + // Insert/append is valid for both append-only and upsertable (commit/event-time) tables. + match RecordMergeStrategyValue::from_str(&strategy)? { + RecordMergeStrategyValue::AppendOnly | RecordMergeStrategyValue::OverwriteWithLatest => { + Ok(()) + } + } +} + +fn ensure_append_only_strategy(table: &Table) -> Result<()> { + let strategy: String = table + .hudi_configs + .get_or_default(RecordMergeStrategy) + .into(); + if RecordMergeStrategyValue::from_str(&strategy)? != RecordMergeStrategyValue::AppendOnly { + return Err(CoreError::Unsupported( + "append_only requires hoodie.record.merge.mode=append_only".to_string(), + )); + } + Ok(()) +} + +pub(crate) fn ensure_copy_on_write(table: &Table) -> Result<()> { + let table_type: String = table.hudi_configs.get(TableType)?.into(); + if TableTypeValue::from_str(&table_type)? != TableTypeValue::CopyOnWrite { + return Err(CoreError::Unsupported(format!( + "append is currently only supported for COPY_ON_WRITE tables. Found '{table_type}'" + ))); + } + Ok(()) +} + +fn ensure_supported_table_type(table: &Table) -> Result<()> { + let table_type: String = table.hudi_configs.get(TableType)?.into(); + match TableTypeValue::from_str(&table_type)? { + TableTypeValue::CopyOnWrite | TableTypeValue::MergeOnRead => Ok(()), + } +} + +pub(crate) fn is_layout_two(table: &Table) -> bool { + let table_version: isize = match table.hudi_configs.get(TableVersion) { + Ok(v) => v.into(), + Err(_) => return false, + }; + let layout_version: isize = match table.hudi_configs.get(TimelineLayoutVersion) { + Ok(v) => v.into(), + Err(_) => return false, + }; + table_version >= 8 && layout_version == 2 +} + +pub(crate) fn timeline_dir(table: &Table) -> String { + if is_layout_two(table) { + let timeline_path: String = table.hudi_configs.get_or_default(TimelinePath).into(); + format!("{HUDI_METADATA_DIR}/{timeline_path}") + } else { + HUDI_METADATA_DIR.to_string() + } +} + +/// Java `HoodieTimeGeneratorConfig` infers a 1 ms max expected clock skew when +/// the (default) `InProcessLockProvider` is used — the single-writer case. +const MAX_EXPECTED_CLOCK_SKEW_MS: u64 = 1; + +/// Skew-adjusting monotonic instant time generator (Java +/// `SkewAdjustingTimeGenerator` + `HoodieInstantTimeGenerator.createNewInstantTime`). +/// +/// Under a process-wide lock: capture the clock, wait out the max expected +/// clock skew, and return the captured time — retrying (never bumping +1) until +/// it exceeds every previously minted instant. Used for both requested and +/// completion timestamps, so completion times are strictly greater than the +/// requested times minted before them. Format `yyyyMMddHHmmssSSS` (UTC). +/// Whether instants are formatted in local time (Java's +/// `HoodieTimelineTimeZone.LOCAL`, the Hudi default) or UTC. Process-wide, +/// mirroring `HoodieInstantTimeGenerator.setCommitTimeZone`; set from the +/// table's `hoodie.table.timeline.timezone` when a table is created or +/// loaded for writes. Spark writers mint LOCAL-time instants by default, so +/// matching the table's declared zone keeps mixed-writer timelines ordered. +static COMMIT_TIME_IS_LOCAL: std::sync::atomic::AtomicBool = + std::sync::atomic::AtomicBool::new(true); + +pub(crate) fn set_commit_timezone(timezone: &str) { + COMMIT_TIME_IS_LOCAL.store( + !timezone.eq_ignore_ascii_case("utc"), + std::sync::atomic::Ordering::Release, + ); +} + +pub(crate) async fn generate_instant_time() -> String { + static TIME_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); + let _guard = TIME_LOCK.lock().await; + loop { + let now = Utc::now(); + tokio::time::sleep(std::time::Duration::from_millis(MAX_EXPECTED_CLOCK_SKEW_MS)).await; + let candidate_millis = now.timestamp_millis(); + // Store is race-free: all mints serialize through TIME_LOCK. + if candidate_millis > LAST_EPOCH_MILLIS.load(Ordering::Acquire) { + LAST_EPOCH_MILLIS.store(candidate_millis, Ordering::Release); + let format = "%Y%m%d%H%M%S%3f"; + return if COMMIT_TIME_IS_LOCAL.load(std::sync::atomic::Ordering::Acquire) { + now.with_timezone(&chrono::Local).format(format).to_string() + } else { + now.format(format).to_string() + }; + } + } +} + +/// Default max base-file size (Java `hoodie.parquet.max.file.size` = 120 MiB). +pub(crate) const DEFAULT_MAX_FILE_SIZE_BYTES: i64 = 120 * 1024 * 1024; +/// Default record size estimate used for insert splits. +pub(crate) const DEFAULT_RECORD_SIZE_ESTIMATE: i64 = 1024; + +pub(crate) fn max_records_per_file(table: &Table) -> usize { + let max_file_size = table + .hudi_configs + .as_options() + .get("hoodie.parquet.max.file.size") + .and_then(|v| v.parse::<i64>().ok()) + .unwrap_or(DEFAULT_MAX_FILE_SIZE_BYTES) + .max(1); + let record_size = table + .hudi_configs + .as_options() + .get("hoodie.copyonwrite.record.size.estimate") + .and_then(|v| v.parse::<i64>().ok()) + .unwrap_or(DEFAULT_RECORD_SIZE_ESTIMATE) + .max(1); + ((max_file_size / record_size) as usize).max(1) +} + +/// Parquet `WriterProperties` resolved from `hoodie.parquet.*` configs. +pub(crate) fn parquet_writer_props(table: &Table) -> parquet::file::properties::WriterProperties { + use parquet::basic::{Compression, GzipLevel, ZstdLevel}; + use parquet::file::properties::WriterProperties; + + let options = table.hudi_configs.as_options(); + let page_size = options + .get("hoodie.parquet.page.size") + .and_then(|v| v.parse::<usize>().ok()) + .unwrap_or(1024 * 1024) + .max(1); + let block_size = options + .get("hoodie.parquet.block.size") + .and_then(|v| v.parse::<i64>().ok()) + .unwrap_or(DEFAULT_MAX_FILE_SIZE_BYTES) + .max(1); + let record_size = options + .get("hoodie.copyonwrite.record.size.estimate") + .and_then(|v| v.parse::<i64>().ok()) + .unwrap_or(DEFAULT_RECORD_SIZE_ESTIMATE) + .max(1); + // parquet-rs row-group limit is row-count; approximate Java's byte block size. + let max_row_group_rows = ((block_size / record_size) as usize).max(1); + let dictionary_enabled = options + .get("hoodie.parquet.dictionary.enabled") + .map(|v| v.eq_ignore_ascii_case("true")) + .unwrap_or(true); + let codec = options + .get("hoodie.parquet.compression.codec") + .map(|s| s.as_str()) + .unwrap_or("zstd"); + let compression = match codec.to_ascii_lowercase().as_str() { + "uncompressed" | "none" => Compression::UNCOMPRESSED, + "snappy" => Compression::SNAPPY, + "gzip" | "gz" => Compression::GZIP(GzipLevel::try_new(6).unwrap_or_default()), + "lz4" => Compression::LZ4, + "brotli" => Compression::BROTLI(Default::default()), + // Default zstd — intentional vs Java's gzip; honor hoodie.parquet.compression.codec. + _ => Compression::ZSTD(ZstdLevel::try_new(3).unwrap_or_default()), + }; + + WriterProperties::builder() + .set_compression(compression) + .set_dictionary_enabled(dictionary_enabled) + .set_data_page_size_limit(page_size) + .set_dictionary_page_size_limit(page_size) + .set_max_row_group_row_count(Some(max_row_group_rows)) + .build() +} + +/// Encode batches to parquet bytes with pre-resolved properties (safe to run +/// on a blocking thread with owned inputs). +pub(crate) fn write_parquet_bytes_with_props( + props: parquet::file::properties::WriterProperties, + batches: &[RecordBatch], +) -> Result<Vec<u8>> { + let cursor = Cursor::new(Vec::new()); + let mut writer = ArrowWriter::try_new(cursor, batches[0].schema(), Some(props))?; + for batch in batches { + writer.write(batch)?; + } + let cursor = writer.into_inner()?; + Ok(cursor.into_inner()) +} + +/// Arrow→Avro schema JSON for `HoodieCommitMetadata.extraMetadata.schema`. +/// +/// Unmapped Arrow types error out — silent `"string"` fallback corrupts Spark's +/// TableSchemaResolver (Parquet types diverge from commit schema). +pub(crate) fn arrow_schema_to_avro_json(schema: &arrow_schema::Schema) -> Result<String> { + let fields: Vec<String> = schema + .fields() + .iter() + .enumerate() + .map(|(index, f)| { + ensure_legal_avro_name(f.name())?; + // Field INDEX, not name: a name-derived path is not injective + // (`a.b` and `a_b` collide), and Avro refuses a redefined record. + let avro_type = arrow_type_to_avro_type_json_named( + f.data_type(), + &format!("hoodie_record_f{index}"), + )?; + let ty = if f.is_nullable() { + format!("[\"null\",{avro_type}]") + } else { + avro_type + }; + Ok(format!( + "{{\"name\":\"{}\",\"type\":{}}}", + f.name().replace('\"', "\\\""), + ty + )) + }) + .collect::<Result<Vec<_>>>()?; + Ok(format!( + "{{\"type\":\"record\",\"name\":\"hoodie_record\",\"fields\":[{}]}}", + fields.join(",") + )) +} + +/// Avro field names must match [A-Za-z_][A-Za-z0-9_]*. A name that does not is +/// rejected rather than renamed: renaming would diverge from the parquet column +/// and silently mislabel the data, while an unparseable commit schema makes the +/// table unreadable by every engine (including this one). +fn ensure_legal_avro_name(name: &str) -> Result<()> { + let mut chars = name.chars(); + let legal = chars + .next() + .is_some_and(|c| c.is_ascii_alphabetic() || c == '_') + && chars.all(|c| c.is_ascii_alphanumeric() || c == '_'); + if legal { + return Ok(()); + } + Err(CoreError::Unsupported(format!( + "column name '{name}' is not a legal Avro name ([A-Za-z_][A-Za-z0-9_]*), \ + so it cannot be written into Hudi commit metadata" + ))) +} + +/// `path` makes every nested Avro record name unique: Avro forbids redefining +/// a named type, so a fixed name breaks any schema with two structs — Java's +/// `Schema.Parser` rejects the whole commit schema. +fn arrow_type_to_avro_type_json_named(dt: &arrow_schema::DataType, path: &str) -> Result<String> { + use arrow_schema::DataType; + match dt { + DataType::Boolean => Ok("\"boolean\"".to_string()), + DataType::Int8 | DataType::Int16 | DataType::Int32 => Ok("\"int\"".to_string()), + DataType::Int64 => Ok("\"long\"".to_string()), + DataType::Float32 => Ok("\"float\"".to_string()), + DataType::Float64 => Ok("\"double\"".to_string()), + DataType::Utf8 | DataType::LargeUtf8 => { + Ok("{\"type\":\"string\",\"avro.java.string\":\"String\"}".to_string()) + } + DataType::Binary | DataType::LargeBinary => Ok("\"bytes\"".to_string()), + DataType::Date32 => Ok("{\"type\":\"int\",\"logicalType\":\"date\"}".to_string()), + DataType::Timestamp(unit, tz) => { + // A timezone-less Arrow timestamp is Spark's TimestampNTZType, + // which maps to Avro local-timestamp-*; mapping it to the + // UTC-adjusted logical type shifts values on read. + let local = tz.is_none(); + let logical = match unit { + arrow_schema::TimeUnit::Millisecond if local => "local-timestamp-millis", + arrow_schema::TimeUnit::Microsecond if local => "local-timestamp-micros", + arrow_schema::TimeUnit::Millisecond => "timestamp-millis", + arrow_schema::TimeUnit::Microsecond => "timestamp-micros", + other => { + return Err(CoreError::Unsupported(format!( + "commit schema does not support Arrow timestamp unit {other:?}" + ))); + } + }; + Ok(format!( + "{{\"type\":\"long\",\"logicalType\":\"{logical}\"}}" + )) + } + DataType::Decimal128(precision, scale) | DataType::Decimal256(precision, scale) => { + Ok(format!( + "{{\"type\":\"bytes\",\"logicalType\":\"decimal\",\"precision\":{precision},\"scale\":{scale}}}" + )) + } + DataType::List(field) | DataType::LargeList(field) => { + let item = + arrow_type_to_avro_type_json_named(field.data_type(), &format!("{path}_item"))?; + let item = if field.is_nullable() { + format!("[\"null\",{item}]") + } else { + item + }; + Ok(format!("{{\"type\":\"array\",\"items\":{item}}}")) + } + DataType::Struct(fields) => { + let nested: Vec<String> = fields + .iter() + .enumerate() + .map(|(index, f)| { + ensure_legal_avro_name(f.name())?; + let avro_type = arrow_type_to_avro_type_json_named( + f.data_type(), + &format!("{path}f{index}"), + )?; + let ty = if f.is_nullable() { + format!("[\"null\",{avro_type}]") + } else { + avro_type + }; + Ok(format!( + "{{\"name\":\"{}\",\"type\":{}}}", + f.name().replace('\"', "\\\""), + ty + )) + }) + .collect::<Result<Vec<_>>>()?; + Ok(format!( + "{{\"type\":\"record\",\"name\":\"{path}_record\",\"fields\":[{}]}}", + nested.join(",") + )) + } + other => Err(CoreError::Unsupported(format!( + "cannot map Arrow type {other:?} to Avro for commit metadata schema" + ))), + } +} + +async fn ensure_append_schema_matches_table( + table: &Table, + batch_schema: &arrow_schema::Schema, +) -> Result<()> { + if table.timeline.completed_commits.is_empty() { + return Ok(()); + } + let table_schema = table.get_schema().await?; + let batch_data = strip_meta_fields_from_schema(batch_schema); + if schemas_equal_by_fields(&table_schema, &batch_data) { + return Ok(()); + } + Err(CoreError::Write(format!( + "append batch schema does not match table schema (table fields={:?}, batch fields={:?})", + table_schema + .fields() + .iter() + .map(|f| (f.name().clone(), f.data_type().clone())) + .collect::<Vec<_>>(), + batch_data + .fields() + .iter() + .map(|f| (f.name().clone(), f.data_type().clone())) + .collect::<Vec<_>>(), + ))) +} + +pub(crate) fn strip_meta_fields_from_schema(schema: &arrow_schema::Schema) -> arrow_schema::Schema { + let fields: Vec<_> = schema + .fields() + .iter() + .filter(|f| !f.name().starts_with("_hoodie_")) Review Comment: Fixed in 30342034 and d6c1e48c: `_hoodie_is_deleted` is a data column and marks deletes on upsert and on copy-on-write update. ########## crates/core/src/table/mod.rs: ########## @@ -201,17 +200,27 @@ fn instant_time_minus_one(instant_time: &str) -> String { } impl Table { - /// Get or initialize the cached metadata table instance. + /// Open a fresh metadata-table handle from storage (no instance cache). /// - /// Returns `Ok(&Table)` if metadata table is successfully created or was already cached. - /// The instance is created once and reused across all subsequent calls. - pub(crate) async fn get_or_init_metadata_table(&self) -> Result<&Table> { - self.cached_metadata_table - .get_or_try_init(|| async { - log::debug!("Initializing cached metadata table instance"); - self.new_metadata_table().await - }) - .await + /// Writers and readers reload the data timeline first, then open MDT relative + /// to that view. Commit locking / conflict checks are deferred to a later + /// concurrency-control project. + pub(crate) async fn get_or_init_metadata_table(&self) -> Result<Table> { + self.new_metadata_table().await + } + + /// Reload the active timeline and drop listing caches before a write plans. + pub(crate) async fn reload_timeline_for_write(&mut self) -> Result<()> { + crate::write::ensure_writable_table_version(self)?; + // Instants must be minted in the table's declared timeline timezone + // (Spark writers default to LOCAL); set before any instant is minted. + crate::write::set_commit_timezone(&self.timezone()); + // Eager rollback of crashed writes (Java rollbackFailedWrites) before + // any new instant is minted. + crate::write::rollback::rollback_failed_writes(self).await?; Review Comment: Kept as a single-writer limitation and documented in b6cfdb86 (`write/lock.rs` and the writer design doc). -- 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]
