This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hudi-rs.git
The following commit(s) were added to refs/heads/main by this push:
new 10d45378 feat(core): return the commit completion time from writes
(#877)
10d45378 is described below
commit 10d453786efaeaa3751fc88edb0a43ad34f17486
Author: Y Ethan Guo <[email protected]>
AuthorDate: Thu Oct 29 19:59:30 2026 -0700
feat(core): return the commit completion time from writes (#877)
On timeline layout v2 an incremental read's bounds are compared against
completion times, but a write returned only the instant's requested time, so
a caller could not read strictly after its own write. WriteResult and
AppendResult now carry completion_time, the value written into the completed
instant; it is None on layout v1 and when no commit is written.
Closes #876
---
crates/core/src/write/append.rs | 10 +++++---
crates/core/src/write/mod.rs | 21 ++++++++++++---
crates/core/src/write/rewrite/cow.rs | 3 ++-
crates/core/src/write/rewrite/mod.rs | 6 ++++-
crates/core/src/write/rewrite/mor.rs | 6 +++--
crates/core/tests/table_write_tests.rs | 47 ++++++++++++++++++++++++++++++++++
6 files changed, 82 insertions(+), 11 deletions(-)
diff --git a/crates/core/src/write/append.rs b/crates/core/src/write/append.rs
index 813ef058..0fa25c79 100644
--- a/crates/core/src/write/append.rs
+++ b/crates/core/src/write/append.rs
@@ -53,8 +53,10 @@ static LAST_EPOCH_MILLIS: AtomicI64 = AtomicI64::new(0);
/// What an append wrote.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct AppendResult {
- /// Timestamp of the completed instant.
+ /// Timestamp of the completed instant (the requested time).
pub instant: String,
+ /// Completion time of the instant on timeline layout v2; `None` on v1.
+ pub completion_time: Option<String>,
/// Path of the completed instant, relative to the table base.
pub commit_relative_path: String,
/// Path of the first base file written, relative to the table base.
@@ -327,7 +329,7 @@ pub async fn append_batches(table: &mut Table, batches:
&[RecordBatch]) -> Resul
let commit_bytes =
crate::write::commit_metadata_bytes(table, "INSERT", write_stats,
None, schema.as_ref())?;
- let commit_relative_path = crate::write::complete_write(
+ let completed = crate::write::complete_write(
table,
&request_instant,
action,
@@ -348,7 +350,8 @@ pub async fn append_batches(table: &mut Table, batches:
&[RecordBatch]) -> Resul
Ok(AppendResult {
instant: request_instant,
- commit_relative_path,
+ completion_time: completed.completion_time,
+ commit_relative_path: completed.commit_relative_path,
base_file_path: primary_base_path,
num_rows,
})
@@ -502,6 +505,7 @@ async fn append_version_6(table: &mut Table, batches:
&[RecordBatch]) -> Result<
table.clear_read_caches();
Ok(AppendResult {
instant,
+ completion_time: None,
commit_relative_path,
base_file_path,
num_rows,
diff --git a/crates/core/src/write/mod.rs b/crates/core/src/write/mod.rs
index e30b8e83..76719655 100644
--- a/crates/core/src/write/mod.rs
+++ b/crates/core/src/write/mod.rs
@@ -484,7 +484,8 @@ pub(crate) fn commit_metadata_bytes(
/// effort: a transient failure must not turn a committed write into an error
/// the caller would retry.
///
-/// Returns the completed instant's path relative to the table base.
+/// Returns the completed instant's path relative to the table base and its
+/// completion time (timeline layout v2 only).
pub(crate) async fn complete_write(
table: &mut crate::table::Table,
instant: &str,
@@ -493,7 +494,7 @@ pub(crate) async fn complete_write(
schema: &arrow_schema::Schema,
updates: MetadataTableUpdates,
written_paths: &[String],
-) -> Result<String> {
+) -> Result<CompletedWrite> {
let storage = table.file_system_view.storage.clone();
let mut mdt_stats = HashMap::new();
if table.is_metadata_table_enabled() {
@@ -555,7 +556,7 @@ pub(crate) async fn complete_write(
} else {
None
};
- let completed = Instant::new_completed(instant.to_string(), action,
completion)?;
+ let completed = Instant::new_completed(instant.to_string(), action,
completion.clone())?;
let commit_relative_path =
completed.relative_path_with_base(&append::timeline_dir(table))?;
storage
.put_file_if_absent(&commit_relative_path, commit_bytes)
@@ -565,7 +566,19 @@ pub(crate) async fn complete_write(
let _ = table.timeline.reload_completed_commits().await;
table.clear_read_caches();
- Ok(commit_relative_path)
+ Ok(CompletedWrite {
+ commit_relative_path,
+ completion_time: completion,
+ })
+}
+
+/// What [`complete_write`] stored on the timeline.
+pub(crate) struct CompletedWrite {
+ /// Path of the completed instant, relative to the table base.
+ pub commit_relative_path: String,
+ /// Completion time of the instant on timeline layout v2; `None` on v1,
+ /// which records none.
+ pub completion_time: Option<String>,
}
#[cfg(test)]
diff --git a/crates/core/src/write/rewrite/cow.rs
b/crates/core/src/write/rewrite/cow.rs
index e139b119..a0b45e46 100644
--- a/crates/core/src/write/rewrite/cow.rs
+++ b/crates/core/src/write/rewrite/cow.rs
@@ -372,7 +372,7 @@ pub(super) async fn finalize_rewrite_commit(
},
schema.as_ref(),
)?;
- crate::write::complete_write(
+ let completed = crate::write::complete_write(
table,
instant,
action,
@@ -389,6 +389,7 @@ pub(super) async fn finalize_rewrite_commit(
.await?;
Ok(WriteResult {
instant: instant.to_string(),
+ completion_time: completed.completion_time,
num_writes,
num_updates: updates,
num_inserts: inserts,
diff --git a/crates/core/src/write/rewrite/mod.rs
b/crates/core/src/write/rewrite/mod.rs
index ecc0d80e..577c1ce7 100644
--- a/crates/core/src/write/rewrite/mod.rs
+++ b/crates/core/src/write/rewrite/mod.rs
@@ -63,8 +63,12 @@ use validate::*;
/// Result of an upsert, delete, or overwrite write.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct WriteResult {
- /// Completed instant timestamp.
+ /// Completed instant timestamp (the requested time).
pub instant: String,
+ /// Completion time of the instant on timeline layout v2, which is what an
+ /// incremental read's start bound is compared against on that layout.
+ /// `None` on layout v1 and when no commit was written.
+ pub completion_time: Option<String>,
/// Number of rows written to replacement base files.
pub num_writes: usize,
/// Number of input keys that replaced existing rows.
diff --git a/crates/core/src/write/rewrite/mor.rs
b/crates/core/src/write/rewrite/mor.rs
index 921e8fc2..30303736 100644
--- a/crates/core/src/write/rewrite/mor.rs
+++ b/crates/core/src/write/rewrite/mor.rs
@@ -626,7 +626,7 @@ pub(super) async fn mor_upsert_batches(
None,
incoming.schema().as_ref(),
)?;
- crate::write::complete_write(
+ let completed = crate::write::complete_write(
table,
&instant,
Action::DeltaCommit,
@@ -643,6 +643,7 @@ pub(super) async fn mor_upsert_batches(
.await?;
Ok(WriteResult {
instant,
+ completion_time: completed.completion_time,
num_writes: incoming.num_rows(),
num_updates: updates,
num_inserts: inserts,
@@ -805,7 +806,7 @@ pub(super) async fn mor_delete_keys(
None,
&schema,
)?;
- crate::write::complete_write(
+ let completed = crate::write::complete_write(
table,
&instant,
Action::DeltaCommit,
@@ -822,6 +823,7 @@ pub(super) async fn mor_delete_keys(
.await?;
Ok(WriteResult {
instant,
+ completion_time: completed.completion_time,
num_writes: 0,
num_updates: 0,
num_inserts: 0,
diff --git a/crates/core/tests/table_write_tests.rs
b/crates/core/tests/table_write_tests.rs
index a844eb09..533dbd82 100644
--- a/crates/core/tests/table_write_tests.rs
+++ b/crates/core/tests/table_write_tests.rs
@@ -304,6 +304,53 @@ async fn
test_append_writes_column_stats_and_partition_stats_mdt() {
);
}
+#[tokio::test]
+async fn test_write_result_completion_time_matches_timeline() {
+ for table_type in [TableTypeValue::CopyOnWrite,
TableTypeValue::MergeOnRead] {
+ let dir = tempdir().unwrap();
+ let mut table = Table::create(dir.path().to_str().unwrap())
+ .with_table_name("trips")
+ .with_table_type(table_type.clone())
+ .with_record_key_fields(["id"])
+ .create()
+ .await
+ .unwrap();
+ let append = table.append([sample_batch()]).await.unwrap();
+ let upsert = table
+ .upsert([batch(vec![("a", 10), ("d", 4)])])
+ .await
+ .unwrap();
+ let delete = table.delete("id = 'a'").await.unwrap();
+ assert_eq!(delete.num_deletes, 1, "{table_type:?}");
+
+ let timeline = table.get_timeline();
+ let instants = match table_type {
+ TableTypeValue::CopyOnWrite =>
timeline.get_completed_commits(false).await,
+ TableTypeValue::MergeOnRead =>
timeline.get_completed_deltacommits(false).await,
+ }
+ .unwrap();
+ for (verb, instant, completion_time) in [
+ ("append", &append.instant, &append.completion_time),
+ ("upsert", &upsert.instant, &upsert.completion_time),
+ ("delete", &delete.instant, &delete.completion_time),
+ ] {
+ let completed = instants
+ .iter()
+ .find(|i| &i.timestamp == instant)
+ .unwrap_or_else(|| panic!("{table_type:?}: {verb} instant not
on the timeline"));
+ assert!(completion_time.is_some(), "{table_type:?} {verb}");
+ assert_eq!(
+ completion_time, &completed.completion_timestamp,
+ "{table_type:?} {verb}"
+ );
+ }
+
+ let noop = table.delete("id = 'missing'").await.unwrap();
+ assert_eq!(noop.num_deletes, 0, "{table_type:?}");
+ assert_eq!(noop.completion_time, None, "{table_type:?}");
+ }
+}
+
#[tokio::test]
async fn test_upsert_delete_and_overwrite() {
let dir = tempdir().unwrap();
