yihua commented on code in PR #666:
URL: https://github.com/apache/hudi-rs/pull/666#discussion_r4404754794
##########
crates/core/src/metadata/table/mod.rs:
##########
@@ -455,6 +469,7 @@ impl Table {
let data_pending = self.timeline.all_pending_instant_times().await?;
let mut valid: HashSet<String> =
self.valid_from_completed_data_instants();
+ valid.extend(self.archived_data_instants().await?);
Review Comment:
Fixed in 9c2777c8: metadata-table reads use only the active timeline.
##########
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> {
Review Comment:
Fixed in 55112538: the metadata-table handle is cached for reads and dropped
when the handle writes.
##########
crates/core/src/write/rollback.rs:
##########
@@ -0,0 +1,438 @@
+/*
+ * 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.
+ */
+//! Eager marker-based rollback of failed writes (Java `EAGER` cleaning
policy).
+//!
+//! At write start, any pending data instant (fenced but never completed) is
+//! rolled back: its marker-listed base/log files are deleted (tv8+ rollback is
+//! deletion-only), its completed-but-orphaned MDT deltacommit and MDT log
+//! files are removed, a `rollback` instant records the operation, and the
+//! failed instant's fencing files and marker dir are cleaned up.
+
+use std::collections::HashMap;
+
+use crate::Result;
+use crate::metadata::rollback::{
+ HoodieInstantInfo, HoodieRollbackMetadata,
HoodieRollbackPartitionMetadata, HoodieRollbackPlan,
+ HoodieRollbackRequest,
+};
+use crate::storage::Storage;
+use crate::table::Table;
+use crate::write::append::{generate_instant_time, is_layout_two, timeline_dir};
+use crate::write::markers::{delete_marker_dir, parse_marker_name,
read_markers};
+
+const METADATA_BASE: &str = ".hoodie/metadata";
+
+/// A pending (fenced, never completed) instant on the data timeline.
+#[derive(Debug, Clone)]
+struct PendingInstant {
+ timestamp: String,
+ action: String,
+ /// Fencing file names present on the timeline (to delete after rollback).
+ fencing_files: Vec<String>,
+}
+
+/// Roll back all pending data instants (Java `rollbackFailedWrites`, EAGER).
+///
+/// Single-writer: any pending instant belongs to a crashed previous write.
+/// Rollback instants themselves and MDT-internal instants are never targets.
+pub(crate) async fn rollback_failed_writes(table: &Table) -> Result<()> {
+ let storage = table.file_system_view.storage.clone();
+ let timeline = timeline_dir(table);
+ let pending = find_pending_instants(storage.as_ref(), &timeline).await?;
+ for instant in pending {
+ rollback_instant(table, storage.as_ref(), &timeline, &instant).await?;
+ }
+ Ok(())
+}
+
+/// The action infix of a fencing file name: `{ts}.{action}.requested|inflight`
+/// or the bare COW forms `{ts}.inflight` / `{ts}.requested` (empty action).
+///
+/// Must not depend on which of an instant's fencing files is listed first —
+/// directory listing order is filesystem-specific (ext4 returns hash order,
+/// so `.inflight` can precede `.commit.requested`).
+fn action_from_fencing_name(name: &str, timestamp: &str) -> String {
+ let rest = name.trim_start_matches(timestamp).trim_start_matches('.');
+ match rest {
+ "inflight" | "requested" => String::new(),
+ other => other
+ .trim_end_matches(".requested")
+ .trim_end_matches(".inflight")
+ .to_string(),
+ }
+}
+
+async fn find_pending_instants(
+ storage: &Storage,
+ timeline_dir: &str,
+) -> Result<Vec<PendingInstant>> {
+ let files = match storage.list_files(Some(timeline_dir)).await {
+ Ok(files) => files,
+ Err(crate::storage::error::StorageError::ObjectStoreError(
+ object_store::Error::NotFound { .. },
+ )) => return Ok(Vec::new()),
+ Err(error) => return Err(error.into()),
+ };
+ let mut completed = std::collections::HashSet::new();
+ let mut fenced: HashMap<String, (String, Vec<String>)> = HashMap::new();
+ for file in &files {
+ let name = &file.name;
+ let Some(first) = name.chars().next() else {
+ continue;
+ };
+ if !first.is_ascii_digit() {
+ continue;
+ }
+ let timestamp: String =
name.chars().take_while(char::is_ascii_digit).collect();
+ if name.ends_with(".requested") || name.ends_with(".inflight") {
+ let action = action_from_fencing_name(name, ×tamp);
+ let entry = fenced
+ .entry(timestamp)
+ .or_insert_with(|| (action.clone(), Vec::new()));
+ // Prefer the action-bearing name (COW inflight has no action
infix).
+ if entry.0.is_empty() && !action.is_empty() {
+ entry.0 = action;
+ }
+ entry.1.push(name.clone());
+ } else {
+ completed.insert(timestamp);
+ }
+ }
+ let mut pending: Vec<PendingInstant> = fenced
+ .into_iter()
+ .filter(|(ts, _)| !completed.contains(ts))
+ // Rollback/compaction-style instants are never rolled back here.
+ .filter(|(_, (action, _))| {
+ matches!(
+ action.as_str(),
+ "commit" | "deltacommit" | "replacecommit" | ""
+ )
+ })
+ .map(|(timestamp, (action, fencing_files))| PendingInstant {
+ timestamp,
+ action: if action.is_empty() {
+ "commit".to_string()
+ } else {
+ action
+ },
+ fencing_files,
+ })
+ .collect();
+ pending.sort_by(|a, b| a.timestamp.cmp(&b.timestamp));
+ Ok(pending)
+}
+
+async fn rollback_instant(
+ table: &Table,
+ storage: &Storage,
+ timeline_dir: &str,
+ pending: &PendingInstant,
+) -> Result<()> {
+ let start_time = std::time::Instant::now();
+ let failed_ts = &pending.timestamp;
+
+ // Markers tell us exactly what the failed write intended to put. A crash
+ // before markers means no data or MDT files were written (markers are
+ // written first), so an empty list still produces a valid rollback.
+ let markers = read_markers(storage, failed_ts).await?;
+ let mut requests_by_partition: HashMap<String, HoodieRollbackRequest> =
HashMap::new();
+ let mut partition_meta: HashMap<String, HoodieRollbackPartitionMetadata> =
HashMap::new();
+ let mut total_deleted = 0i32;
+ for marker in &markers {
+ let Some((file_path, _io_type)) = parse_marker_name(marker) else {
+ continue;
+ };
+ let (partition, file_name) = match file_path.rsplit_once('/') {
+ Some((partition, name)) => (partition.to_string(),
name.to_string()),
+ None => (String::new(), file_path.clone()),
+ };
+ let is_log_file = file_name.starts_with('.') &&
file_name.contains(".log.");
+ let request = requests_by_partition
+ .entry(partition.clone())
+ .or_insert_with(|| HoodieRollbackRequest {
+ partition_path: partition.clone(),
+ file_id: None,
+ latest_base_instant: None,
+ files_to_be_deleted: Vec::new(),
+ log_blocks_to_be_deleted: None,
+ });
+ if is_log_file {
+ request
+ .log_blocks_to_be_deleted
+ .get_or_insert_with(HashMap::new)
+ .insert(file_path.clone(), 1);
+ } else {
+ request.files_to_be_deleted.push(file_path.clone());
+ }
+ let deleted = storage.delete_file(&file_path).await.is_ok();
+ let meta = partition_meta.entry(partition.clone()).or_insert_with(|| {
+ HoodieRollbackPartitionMetadata {
+ partition_path: partition.clone(),
+ ..Default::default()
+ }
+ });
+ if deleted {
+ total_deleted += 1;
+ meta.success_delete_files.push(file_path);
+ } else {
+ meta.failed_delete_files.push(file_path);
Review Comment:
Fixed in 1c2fce80 and 7877f6b0: the rollback fails when a file it must
delete survives or a metadata-table partition can't be listed.
##########
crates/core/src/write/rollback.rs:
##########
@@ -0,0 +1,438 @@
+/*
+ * 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.
+ */
+//! Eager marker-based rollback of failed writes (Java `EAGER` cleaning
policy).
+//!
+//! At write start, any pending data instant (fenced but never completed) is
+//! rolled back: its marker-listed base/log files are deleted (tv8+ rollback is
+//! deletion-only), its completed-but-orphaned MDT deltacommit and MDT log
+//! files are removed, a `rollback` instant records the operation, and the
+//! failed instant's fencing files and marker dir are cleaned up.
+
+use std::collections::HashMap;
+
+use crate::Result;
+use crate::metadata::rollback::{
+ HoodieInstantInfo, HoodieRollbackMetadata,
HoodieRollbackPartitionMetadata, HoodieRollbackPlan,
+ HoodieRollbackRequest,
+};
+use crate::storage::Storage;
+use crate::table::Table;
+use crate::write::append::{generate_instant_time, is_layout_two, timeline_dir};
+use crate::write::markers::{delete_marker_dir, parse_marker_name,
read_markers};
+
+const METADATA_BASE: &str = ".hoodie/metadata";
+
+/// A pending (fenced, never completed) instant on the data timeline.
+#[derive(Debug, Clone)]
+struct PendingInstant {
+ timestamp: String,
+ action: String,
+ /// Fencing file names present on the timeline (to delete after rollback).
+ fencing_files: Vec<String>,
+}
+
+/// Roll back all pending data instants (Java `rollbackFailedWrites`, EAGER).
+///
+/// Single-writer: any pending instant belongs to a crashed previous write.
+/// Rollback instants themselves and MDT-internal instants are never targets.
+pub(crate) async fn rollback_failed_writes(table: &Table) -> Result<()> {
+ let storage = table.file_system_view.storage.clone();
+ let timeline = timeline_dir(table);
+ let pending = find_pending_instants(storage.as_ref(), &timeline).await?;
+ for instant in pending {
+ rollback_instant(table, storage.as_ref(), &timeline, &instant).await?;
+ }
+ Ok(())
+}
+
+/// The action infix of a fencing file name: `{ts}.{action}.requested|inflight`
+/// or the bare COW forms `{ts}.inflight` / `{ts}.requested` (empty action).
+///
+/// Must not depend on which of an instant's fencing files is listed first —
+/// directory listing order is filesystem-specific (ext4 returns hash order,
+/// so `.inflight` can precede `.commit.requested`).
+fn action_from_fencing_name(name: &str, timestamp: &str) -> String {
+ let rest = name.trim_start_matches(timestamp).trim_start_matches('.');
+ match rest {
+ "inflight" | "requested" => String::new(),
+ other => other
+ .trim_end_matches(".requested")
+ .trim_end_matches(".inflight")
+ .to_string(),
+ }
+}
+
+async fn find_pending_instants(
+ storage: &Storage,
+ timeline_dir: &str,
+) -> Result<Vec<PendingInstant>> {
+ let files = match storage.list_files(Some(timeline_dir)).await {
+ Ok(files) => files,
+ Err(crate::storage::error::StorageError::ObjectStoreError(
+ object_store::Error::NotFound { .. },
+ )) => return Ok(Vec::new()),
+ Err(error) => return Err(error.into()),
+ };
+ let mut completed = std::collections::HashSet::new();
+ let mut fenced: HashMap<String, (String, Vec<String>)> = HashMap::new();
+ for file in &files {
+ let name = &file.name;
+ let Some(first) = name.chars().next() else {
+ continue;
+ };
+ if !first.is_ascii_digit() {
+ continue;
+ }
+ let timestamp: String =
name.chars().take_while(char::is_ascii_digit).collect();
+ if name.ends_with(".requested") || name.ends_with(".inflight") {
+ let action = action_from_fencing_name(name, ×tamp);
+ let entry = fenced
+ .entry(timestamp)
+ .or_insert_with(|| (action.clone(), Vec::new()));
+ // Prefer the action-bearing name (COW inflight has no action
infix).
+ if entry.0.is_empty() && !action.is_empty() {
+ entry.0 = action;
+ }
+ entry.1.push(name.clone());
+ } else {
+ completed.insert(timestamp);
+ }
+ }
+ let mut pending: Vec<PendingInstant> = fenced
+ .into_iter()
+ .filter(|(ts, _)| !completed.contains(ts))
+ // Rollback/compaction-style instants are never rolled back here.
+ .filter(|(_, (action, _))| {
+ matches!(
+ action.as_str(),
+ "commit" | "deltacommit" | "replacecommit" | ""
+ )
+ })
+ .map(|(timestamp, (action, fencing_files))| PendingInstant {
+ timestamp,
+ action: if action.is_empty() {
+ "commit".to_string()
+ } else {
+ action
+ },
+ fencing_files,
+ })
+ .collect();
+ pending.sort_by(|a, b| a.timestamp.cmp(&b.timestamp));
+ Ok(pending)
+}
+
+async fn rollback_instant(
+ table: &Table,
+ storage: &Storage,
+ timeline_dir: &str,
+ pending: &PendingInstant,
+) -> Result<()> {
+ let start_time = std::time::Instant::now();
+ let failed_ts = &pending.timestamp;
+
+ // Markers tell us exactly what the failed write intended to put. A crash
+ // before markers means no data or MDT files were written (markers are
+ // written first), so an empty list still produces a valid rollback.
+ let markers = read_markers(storage, failed_ts).await?;
+ let mut requests_by_partition: HashMap<String, HoodieRollbackRequest> =
HashMap::new();
+ let mut partition_meta: HashMap<String, HoodieRollbackPartitionMetadata> =
HashMap::new();
+ let mut total_deleted = 0i32;
+ for marker in &markers {
+ let Some((file_path, _io_type)) = parse_marker_name(marker) else {
+ continue;
+ };
+ let (partition, file_name) = match file_path.rsplit_once('/') {
+ Some((partition, name)) => (partition.to_string(),
name.to_string()),
+ None => (String::new(), file_path.clone()),
+ };
+ let is_log_file = file_name.starts_with('.') &&
file_name.contains(".log.");
+ let request = requests_by_partition
+ .entry(partition.clone())
+ .or_insert_with(|| HoodieRollbackRequest {
+ partition_path: partition.clone(),
+ file_id: None,
+ latest_base_instant: None,
+ files_to_be_deleted: Vec::new(),
+ log_blocks_to_be_deleted: None,
Review Comment:
Fixed in aad6285a: rollback plans use the version 8 marker-based shape and
pending rollbacks resume on their own instant. 4ebb5d43, 563619cd, f33d446c and
ccd4adbd harden plan path matching and resume checks.
##########
crates/core/src/write/archival.rs:
##########
@@ -0,0 +1,562 @@
+/*
+ * 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.
+ */
+//! Timeline archival into the LSM history timeline (Java `TimelineArchiverV2`
+//! + `LSMTimelineWriter`).
+//!
+//! When the active timeline holds more than `hoodie.keep.max.commits`
+//! completed commits, the oldest are archived down to
`hoodie.keep.min.commits`:
+//! written as one level-0 parquet of `HoodieLSMTimelineInstant` rows under
+//! `{timeline}/history/`, registered in a new `manifest_{N}` (JSON) pointed at
+//! by `_version_`, then removed from the active timeline (pending fencing
+//! files first, completed files after — Java's ordering). Runs inside the
+//! completion critical section (user design; Java runs it post-commit).
+//!
+//! LSM level compaction (merging L0 files into L1+) is not yet implemented —
+//! readers are unaffected since every file stays manifest-listed.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use arrow_array::{Array, ArrayRef, BinaryArray, Int32Array, RecordBatch,
StringArray};
+use arrow_schema::{DataType, Field, Schema};
+use serde::{Deserialize, Serialize};
+
+use crate::Result;
+use crate::error::CoreError;
+use crate::storage::Storage;
+
+/// Java `hoodie.keep.min.commits` default.
+pub(crate) const DEFAULT_KEEP_MIN_COMMITS: usize = 20;
+/// Java `hoodie.keep.max.commits` default.
+pub(crate) const DEFAULT_KEEP_MAX_COMMITS: usize = 30;
+/// Java `hoodie.timeline.manifest.retained.versions` default.
+const RETAINED_MANIFEST_VERSIONS: i64 = 3;
+
+/// Java `HoodieLSMTimelineManifest`:
`{"files":[{"fileName":..,"fileLen":..}]}`.
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+struct LsmManifest {
+ files: Vec<LsmFileEntry>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct LsmFileEntry {
+ file_name: String,
+ file_len: i64,
+}
+
+#[derive(Debug, Clone)]
+struct CompletedInstant {
+ requested: String,
+ completion: String,
+ action: String,
+ file_name: String,
+}
+
+/// Archive the timeline at `timeline_dir` if it exceeds `max_keep` completed
+/// commits, retaining the newest `min_keep`.
+pub(crate) async fn archive_timeline_if_needed(
+ storage: &Storage,
+ timeline_dir: &str,
+ min_keep: usize,
+ max_keep: usize,
+) -> Result<()> {
+ let listed = match storage.list_files(Some(timeline_dir)).await {
+ Ok(files) => files,
+ Err(crate::storage::error::StorageError::ObjectStoreError(
+ object_store::Error::NotFound { .. },
+ )) => return Ok(()),
+ Err(error) => return Err(error.into()),
+ };
+
+ let mut completed: Vec<CompletedInstant> = Vec::new();
+ let mut fencing_by_instant: HashMap<String, Vec<String>> = HashMap::new();
+ let mut earliest_pending: Option<String> = None;
+ for file in &listed {
+ let name = &file.name;
+ let Some(first) = name.chars().next() else {
+ continue;
+ };
+ if !first.is_ascii_digit() {
+ continue;
+ }
+ let requested: String =
name.chars().take_while(char::is_ascii_digit).collect();
+ if name.ends_with(".requested") || name.ends_with(".inflight") {
+ fencing_by_instant
+ .entry(requested.clone())
+ .or_default()
+ .push(name.clone());
+ continue;
+ }
+ // Completed layout-v2 name: `{req}_{comp}.{action}`.
+ let Some((stem, action)) = name.rsplit_once('.') else {
+ continue;
+ };
+ let Some((req, comp)) = stem.split_once('_') else {
+ continue;
+ };
+ completed.push(CompletedInstant {
+ requested: req.to_string(),
+ completion: comp.to_string(),
+ action: action.to_string(),
+ file_name: name.clone(),
+ });
+ let _ = requested;
+ }
+ // Pending = fenced without a completed file.
+ let completed_set: std::collections::HashSet<&str> =
+ completed.iter().map(|c| c.requested.as_str()).collect();
+ for ts in fencing_by_instant.keys() {
+ if !completed_set.contains(ts.as_str()) &&
earliest_pending.as_ref().is_none_or(|e| ts < e)
+ {
+ earliest_pending = Some(ts.clone());
+ }
+ }
+
+ completed.sort_by(|a, b| a.requested.cmp(&b.requested));
+ // A replacecommit's partitionToReplaceFileIds is what hides the file
+ // groups it replaced; until a cleaner physically deletes those groups,
+ // archiving it resurrects their rows. Java blocks archival at the first
+ // uncleaned replace instant; with no cleaner implemented, that means
+ // never archiving at or past the earliest active replacecommit.
+ let earliest_replace = completed
+ .iter()
+ .filter(|c| c.action == "replacecommit")
+ .map(|c| c.requested.clone())
+ .min();
+ // A savepoint pins its commit and everything after it: archiving past one
+ // would discard the very files the savepoint exists to retain (Java's
+ // TimelineArchiverV2 stops at the first savepoint unless
+ // hoodie.archive.beyond.savepoint, default false).
+ let earliest_savepoint = completed
+ .iter()
+ .filter(|c| c.action == "savepoint")
+ .map(|c| c.requested.clone())
+ .min();
+ let earliest_pending = [earliest_pending, earliest_replace,
earliest_savepoint]
+ .into_iter()
+ .flatten()
+ .min();
+ let is_commit_action =
+ |action: &str| matches!(action, "commit" | "deltacommit" |
"replacecommit");
+ let commit_count = completed
+ .iter()
+ .filter(|c| is_commit_action(&c.action))
+ .count();
+ // Misconfigured min > max would underflow below and archive the whole
+ // timeline; treat min as the effective floor.
+ let min_keep = min_keep.min(max_keep);
+ if commit_count <= max_keep {
+ return Ok(());
+ }
+
+ // Archive the oldest commits down to `min_keep`, but never at or past the
+ // earliest pending instant; sweep along non-commit actions (rollbacks) up
+ // to the newest archived commit.
+ let mut commits_to_archive = commit_count - min_keep;
+ let mut max_archived_commit: Option<String> = None;
+ let mut to_archive: Vec<CompletedInstant> = Vec::new();
Review Comment:
Fixed in 56fe8cd6, and b4bf0925 stops archival when a fencing file delete
fails.
##########
crates/core/src/write/keygen.rs:
##########
@@ -0,0 +1,351 @@
+/*
+ * 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.
+ */
+//! Derive Hoodie record keys and hive-style partition paths from write
batches.
+
+use arrow::array::{Array, StringArray};
+use arrow::record_batch::RecordBatch;
+
+use crate::Result;
+use crate::config::HudiConfigs;
+use crate::config::table::HudiTableConfig::{
+ IsHiveStylePartitioning, PartitionFields, RecordKeyFields,
+};
+use crate::error::CoreError;
+use crate::index::HoodieKey;
+
+/// Build per-row [`HoodieKey`]s from a write batch.
+///
+/// When `hoodie.table.recordkey.fields` is unset/empty, keys are
auto-generated
+/// like Java `AutoRecordGenWrapper*KeyGenerator`:
`{instant}_{partitionId}_{rowId}`.
+///
+/// `row_id_offset` continues the auto-key row counter across size-split
chunks and
+/// partitions within one commit. `partition_id` distinguishes writer
partitions
+/// (Java `HoodieRecord.generateSequenceId`).
+///
+/// Partition paths use hive-style (`field=value/...`) when
+/// `hoodie.datasource.write.hive_style_partitioning` is true; otherwise
value-only
+/// segments. Null partition values are rejected. Partition columns are not
stripped
+/// from the batch — callers must leave them in the written Parquet.
+/// Placeholder instant for pre-fence validation passes:
[`hoodie_keys_for_batch`]
+/// uses the instant only to FORMAT auto-generated keys, never to decide
errors,
+/// so deriving keys with this dummy value surfaces every deterministic input
+/// error (missing/null/wrong-type key or partition columns) before an instant
+/// is requested.
+pub(crate) const VALIDATION_INSTANT: &str = "00000000000000000";
+
+/// Run key/partition derivation for its error side only — call before fencing
+/// an instant so invalid input leaves no timeline trace.
+pub(crate) fn validate_keygen_inputs(
+ hudi_configs: &HudiConfigs,
+ batches: &[RecordBatch],
+) -> Result<()> {
+ for batch in batches {
+ hoodie_keys_for_batch(hudi_configs, batch, Some(VALIDATION_INSTANT))?;
+ }
+ Ok(())
+}
+
+pub fn hoodie_keys_for_batch(
+ hudi_configs: &HudiConfigs,
+ batch: &RecordBatch,
+ instant: Option<&str>,
+) -> Result<Vec<HoodieKey>> {
+ hoodie_keys_for_batch_with_offset(hudi_configs, batch, instant, 0, 0)
+}
+
+/// Like [`hoodie_keys_for_batch`], with an explicit auto-key row offset and
partition id.
+pub fn hoodie_keys_for_batch_with_offset(
+ hudi_configs: &HudiConfigs,
+ batch: &RecordBatch,
+ instant: Option<&str>,
+ row_id_offset: usize,
+ partition_id: u32,
+) -> Result<Vec<HoodieKey>> {
+ let record_key_fields: Option<Vec<String>> =
+ hudi_configs.try_get(RecordKeyFields)?.map(|v| v.into());
+ let auto_keys = record_key_fields
+ .as_ref()
+ .map(|f| f.is_empty() || (f.len() == 1 && f[0].is_empty()))
+ .unwrap_or(true);
+
+ let partition_fields: Vec<String> =
hudi_configs.get_or_default(PartitionFields).into();
+ let hive_style: bool =
hudi_configs.get_or_default(IsHiveStylePartitioning).into();
+
+ if auto_keys {
+ let instant = instant.ok_or_else(|| {
+ CoreError::Write(
+ "auto record-key generation requires a commit instant
timestamp".to_string(),
+ )
+ })?;
+ let mut keys = Vec::with_capacity(batch.num_rows());
+ for row in 0..batch.num_rows() {
+ // Match Java HoodieRecord.generateSequenceId(instant,
partitionId, rowId).
+ let record_key = format!("{instant}_{partition_id}_{}",
row_id_offset + row);
+ let partition_path = partition_path_for_row(batch, row,
&partition_fields, hive_style)?;
+ keys.push(HoodieKey {
+ record_key,
+ partition_path,
+ });
+ }
+ return Ok(keys);
+ }
+
+ let Some(fields) = record_key_fields else {
+ // Unreachable: auto_keys above covers None, but the no-panic rule
+ // wants an error, not an unwrap.
+ return Err(CoreError::Config(
+ crate::config::error::ConfigError::NotFound("record key
fields".to_string()),
+ ));
+ };
+ if fields.len() != 1 {
+ return Err(CoreError::Unsupported(
+ "writes currently require exactly one record key field (or none
for auto keys)"
+ .to_string(),
+ ));
+ }
+ let key_name = &fields[0];
+ let key_array = batch
+ .column_by_name(key_name)
+ .ok_or_else(|| {
+ CoreError::Schema(format!(
+ "record key field '{key_name}' is missing from write batch"
+ ))
+ })?
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .ok_or_else(|| {
+ CoreError::Unsupported(format!(
+ "record key field '{key_name}' must be Utf8 for writes"
+ ))
+ })?;
+
+ let mut keys = Vec::with_capacity(batch.num_rows());
+ for row in 0..batch.num_rows() {
+ if key_array.is_null(row) {
+ return Err(CoreError::Write(
+ "record key values must be non-null".to_string(),
+ ));
+ }
+ let record_key = key_array.value(row).to_string();
+ let partition_path = partition_path_for_row(batch, row,
&partition_fields, hive_style)?;
+ keys.push(HoodieKey {
+ record_key,
+ partition_path,
+ });
+ }
+ Ok(keys)
+}
+
+fn partition_path_for_row(
Review Comment:
Fixed in 60d96990: write keys honor the table key generator, url-encoded
partitions, and the default partition. bda5901f, 91ef9912, a14c7ced and
60b768a2 cover complex-keygen validation, guard placement, key-filter deletes,
and partition SET.
##########
crates/core/src/write/rewrite.rs:
##########
@@ -0,0 +1,3924 @@
+/*
+ * 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, HashSet};
+use std::str::FromStr;
+
+use arrow::array::{Array, ArrayRef, BooleanArray, StringArray, UInt32Array};
+use arrow::compute::concat_batches;
+use arrow::record_batch::RecordBatch;
+use arrow_select::take::take;
+use arrow_select::zip::zip;
+
+use crate::Result;
+use crate::config::table::HudiTableConfig::{OrderingFields, RecordKeyFields,
RecordMergeStrategy};
+use crate::error::CoreError;
+use crate::expr::ExprOperator;
+use crate::expr::filter::{Filter, filters_to_row_mask,
validate_fields_against_schemas};
+use crate::file_group::record_batches::RecordBatches;
+use crate::index::{HoodieIndex, HoodieKey, is_record_index_enabled};
+use crate::merge::RecordMergeStrategyValue;
+use crate::merge::record_merger::RecordMerger;
+use crate::metadata::commit::{HoodieCommitMetadata, HoodieWriteStat};
+use crate::metadata::meta_field::MetaField;
+use crate::metadata::table::encode::RecordIndexEntry;
+use crate::table::{ReadOptions, Table};
+use crate::timeline::instant::{Action, Instant};
+use crate::write::append::{
+ ensure_copy_on_write, generate_instant_time, is_layout_two,
prepare_batches_for_write,
+ timeline_dir,
+};
+use crate::write::keygen::{hoodie_keys_for_batch, relative_data_path};
+use crate::write::metadata::{
+ instant_to_epoch_millis, is_column_stats_enabled,
update_column_stats_partitions,
+ update_files_partition_entries, update_record_index,
+};
+
+/// Result of an upsert, delete, or overwrite write.
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct WriteResult {
+ /// Completed instant timestamp.
+ pub instant: String,
+ /// Number of rows written to replacement base files.
+ pub num_writes: usize,
+ /// Number of input keys that replaced existing rows.
+ pub num_updates: usize,
+ /// Number of input keys not previously present.
+ pub num_inserts: usize,
+ /// Number of rows deleted.
+ pub num_deletes: usize,
+}
+
+/// Options controlling an upsert.
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct UpsertOptions {
+ /// Data columns to overwrite for matched rows. `None` replaces complete
rows.
+ pub update_columns: Option<Vec<String>>,
+}
+
+/// How a COW rewrite commits: same file-group slice vs replacecommit
overwrite.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum RewriteKind {
+ /// New base file slice in the same file group(s); timeline action =
commit.
+ Commit,
+ /// Replace file groups; timeline action = replacecommit (overwrite only
for now).
+ Replace,
+}
+
+/// Critical section 1: request the action — mint the instant time and fence
+/// it (requested + inflight) under the table lock, then release; the write's
+/// heavy work runs unlocked. `request_commit` / `request_replacecommit` pick
+/// the timeline action from the rewrite kind.
+async fn request_rewrite_instant(
+ table: &Table,
+ operation: &str,
+ kind: RewriteKind,
+) -> Result<String> {
+ let storage = table.file_system_view.storage.clone();
+ let lock = crate::write::lock::lock_provider_for(table);
+ let _cs1 = lock.lock().await?;
+ let instant = generate_instant_time().await;
+ let action = match kind {
+ RewriteKind::Commit => Action::Commit,
+ RewriteKind::Replace => Action::ReplaceCommit,
+ };
+ let requested_bytes = if kind == RewriteKind::Replace {
+
crate::metadata::replace_commit::HoodieRequestedReplaceMetadata::for_operation(operation)
+ .to_avro_bytes()?
+ } else {
+ Vec::new()
+ };
+ crate::write::fence_timeline_instant(
+ storage.as_ref(),
+ &timeline_dir(table),
+ &instant,
+ action,
+ requested_bytes,
+ crate::write::inflight_commit_metadata_bytes(operation,
is_layout_two(table))?,
+ )
+ .await?;
+ Ok(instant)
+}
+
+/// Critical section 1 for MOR deltacommits.
+async fn request_deltacommit(table: &Table, operation: &str) -> Result<String>
{
+ let storage = table.file_system_view.storage.clone();
+ let lock = crate::write::lock::lock_provider_for(table);
+ let _cs1 = lock.lock().await?;
+ let instant = generate_instant_time().await;
+ crate::write::fence_timeline_instant(
+ storage.as_ref(),
+ &timeline_dir(table),
+ &instant,
+ Action::DeltaCommit,
+ Vec::new(),
+ crate::write::inflight_commit_metadata_bytes(operation,
is_layout_two(table))?,
+ )
+ .await?;
+ Ok(instant)
+}
+
+pub async fn upsert_batches(
+ table: &mut Table,
+ batches: &[RecordBatch],
+ options: UpsertOptions,
+) -> Result<WriteResult> {
+ table.reload_timeline_for_write().await?;
+ ensure_configured_record_key(table)?;
+ if table.is_mor() {
+ return mor_upsert_batches(table, batches, options).await;
+ }
+ ensure_rewrite_supported(table)?;
+ if batches.is_empty() {
+ return Err(CoreError::Write(
+ "upsert requires at least one RecordBatch".to_string(),
+ ));
+ }
+ validate_write_input(table, batches, "upsert").await?;
+ if let Some(columns) = &options.update_columns {
+ let schema = batches[0].schema();
+ for column in columns {
+ if schema.field_with_name(column).is_err() {
+ return Err(CoreError::Schema(format!(
+ "update column '{column}' is not in the write batch schema"
+ )));
+ }
+ if
MetaField::field_names_with_operation().contains(&column.as_str()) {
+ return Err(CoreError::Write(format!(
+ "update column '{column}' is a Hudi meta field"
+ )));
+ }
+ }
+ }
+ crate::write::keygen::validate_keygen_inputs(&table.hudi_configs,
batches)?;
+ let instant = request_rewrite_instant(table, "UPSERT",
RewriteKind::Commit).await?;
+ let incoming = prepare_batches_for_write(&table.hudi_configs, batches,
&instant, "pending")?;
+ let incoming = concat(&incoming)?;
+ let key_name = record_key_name(table, &incoming)?;
+ let incoming = deduplicate_last_by_key(&incoming, &key_name,
&table.hudi_configs)?;
+ let incoming_hoodie_keys = hoodie_keys_for_batch(&table.hudi_configs,
&incoming, None)?;
+ let locations = crate::index::for_table_checked(table)
+ .await
+ .tag_location(table, &incoming_hoodie_keys)
+ .await?;
+ // Small-file packing (Java UpsertPartitioner.assignInserts): assign insert
+ // rows to existing small file groups first; chosen groups join the read
+ // set so their rows are preserved through the rewrite.
+ let sizing = crate::write::sizing::SizingConfig::from_table(table);
+ let avg_record_size = crate::write::sizing::average_record_size(
+ table,
+ table.file_system_view.storage.as_ref(),
+ &sizing,
+ )
+ .await;
+ let slices_by_partition =
crate::write::sizing::latest_slices_by_partition(table).await?;
+ let mut insert_rows_by_partition: HashMap<String, Vec<usize>> =
HashMap::new();
+ for (index, hoodie_key) in incoming_hoodie_keys.iter().enumerate() {
+ if locations.get(hoodie_key).and_then(Option::as_ref).is_none() {
+ insert_rows_by_partition
+ .entry(hoodie_key.partition_path.clone())
+ .or_default()
+ .push(index);
+ }
+ }
+ let mut insert_target_by_row: HashMap<usize, Option<String>> =
HashMap::new();
+ for (partition, rows) in &insert_rows_by_partition {
+ let slices: Vec<&crate::file_group::file_slice::FileSlice> =
slices_by_partition
+ .get(partition)
+ .map(|v| v.iter().collect())
+ .unwrap_or_default();
+ let small = crate::write::sizing::small_file_groups(&slices, false,
&sizing);
+ let assignments = crate::write::sizing::assign_insert_buckets(
+ rows.len(),
+ &small,
+ avg_record_size,
+ &sizing,
+ );
+ let mut cursor = 0usize;
+ for (bucket, count) in assignments {
+ for row in &rows[cursor..cursor + count] {
+ match &bucket {
+ crate::write::sizing::InsertBucket::Existing { file_id }
=> {
+ insert_target_by_row.insert(*row,
Some(file_id.clone()));
+ }
+ crate::write::sizing::InsertBucket::New => {
+ insert_target_by_row.insert(*row, None);
+ }
+ }
+ }
+ cursor += count;
+ }
+ }
+ // MergeHandle model: route each input row to its target file group, then
+ // per-group workers read+merge+write. The driver only holds the input and
+ // plan metadata.
+ let mut slice_by_file_id: HashMap<String, (String,
crate::file_group::file_slice::FileSlice)> =
+ HashMap::new();
+ for (partition, slices) in &slices_by_partition {
+ for slice in slices {
+ slice_by_file_id.insert(
+ slice.file_id().to_string(),
+ (partition.clone(), slice.clone()),
+ );
+ }
+ }
+
+ let mut updates = 0usize;
+ let mut inserts = 0usize;
+ let mut rows_by_group: HashMap<String, Vec<u32>> = HashMap::new();
+ let mut new_rows_by_partition: HashMap<String, Vec<u32>> = HashMap::new();
+ let mut row_target: Vec<Option<String>> =
Vec::with_capacity(incoming.num_rows());
+ // Update-partition-path (global index): a key upserted with a different
+ // partition value MOVES — evicted from the group in its old partition and
+ // inserted where its values say. Anything else leaves rows whose stored
+ // partition columns disagree with the path they physically live in.
+ let mut evictions_by_group: HashMap<String, HashSet<String>> =
HashMap::new();
+ // Merge-before-move (Java parity): find the cross-partition updates whose
+ // incoming row loses the event-time merge; they stay in their old group.
+ let mut moved_candidates: HashMap<String, Vec<String>> = HashMap::new();
Review Comment:
Kept as is and documented in 5c966d3f: upserts move a record whose partition
changes, unlike Java's default
`hoodie.record.index.update.partition.path=false`.
--
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]
