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 334c7e02 refactor(core): return one WriteResult from every table write
(#945)
334c7e02 is described below
commit 334c7e0215b5e08d9843804ddf6a89ec482e90af
Author: Y Ethan Guo <[email protected]>
AuthorDate: Fri Oct 30 13:04:29 2026 -0700
refactor(core): return one WriteResult from every table write (#945)
* refactor(core): return one WriteResult from every table write
Table::append returns WriteResult like the other write verbs, and
AppendResult is removed. WriteResult.instant becomes
requested_time: Option<String>, None when the write made no commit,
next to completion_time. The Python binding maps it to
HudiWriteResult.instant_time directly.
* Read requested_time in the compaction tests and keep the Python
compaction instant mapping
* Give CompactionResult requested_time, move WriteResult to the write
module, and pin the upsert benchmark's ruff rules
* Describe the empty compaction result by its requested time
---
README.md | 4 +-
benchmark/upsert/ruff.toml | 21 ++++++
crates/core/src/table/mod.rs | 8 +--
crates/core/src/write/append.rs | 45 ++++--------
crates/core/src/write/mod.rs | 28 ++++++--
crates/core/src/write/rewrite/compaction.rs | 53 ++++++++------
crates/core/src/write/rewrite/cow.rs | 2 +-
crates/core/src/write/rewrite/mod.rs | 22 +-----
crates/core/src/write/rewrite/mor.rs | 4 +-
crates/core/tests/spark_parity_tests.rs | 2 +-
crates/core/tests/table_bulk_insert_tests.rs | 70 ++++++++++---------
crates/core/tests/table_write_archival_tests.rs | 3 +-
.../core/tests/table_write_composite_keys_tests.rs | 6 +-
crates/core/tests/table_write_lifecycle_tests.rs | 18 ++---
crates/core/tests/table_write_rollback_tests.rs | 11 +--
crates/core/tests/table_write_tests.rs | 81 +++++++++++++++-------
python/src/internal.rs | 56 ++-------------
17 files changed, 223 insertions(+), 211 deletions(-)
diff --git a/README.md b/README.md
index 72c80d1a..b4cd1fa6 100644
--- a/README.md
+++ b/README.md
@@ -464,7 +464,9 @@ table.upsert(batch, update_columns=["fare"])
let result = table.upsert([batch]).await?;
println!(
"instant={} updates={} inserts={}",
- result.instant, result.num_updates, result.num_inserts
+ result.requested_time.as_deref().unwrap_or("none"),
+ result.num_updates,
+ result.num_inserts
);
```
diff --git a/benchmark/upsert/ruff.toml b/benchmark/upsert/ruff.toml
new file mode 100644
index 00000000..1739702d
--- /dev/null
+++ b/benchmark/upsert/ruff.toml
@@ -0,0 +1,21 @@
+# 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.
+
+# Pin the rule set rather than follow ruff's built-in defaults, which change
+# across ruff releases.
+target-version = "py310"
+lint.select = ["E4", "E7", "E9", "F"]
diff --git a/crates/core/src/table/mod.rs b/crates/core/src/table/mod.rs
index 4bbda1cf..cd61dcfd 100644
--- a/crates/core/src/table/mod.rs
+++ b/crates/core/src/table/mod.rs
@@ -98,7 +98,6 @@ pub(crate) mod validation;
pub use crate::config::read_options::{QueryType, ReadOptions};
pub use crate::index::{HoodieKey, SimpleIndex};
pub use crate::write::TableCreateBuilder;
-pub use crate::write::append::AppendResult;
pub use crate::write::{CompactionResult, UpsertOptions, WriteResult};
use crate::Result;
@@ -461,8 +460,9 @@ impl Table {
/// under one compaction instant that completes as a `commit`.
///
/// Rows keep their `_hoodie_commit_time`, so incremental reads see no new
- /// rows from a compaction. Returns a result with an empty instant when no
- /// file group had log files, in which case nothing is written.
+ /// rows from a compaction. Returns a result whose `requested_time` is
+ /// `None` when no file group had log files, in which case nothing is
+ /// written.
///
/// Compaction also runs on its own after a delta commit when
/// `hoodie.compact.inline` is `true` and the completed delta commits since
@@ -1590,7 +1590,7 @@ impl Table {
pub async fn append(
&mut self,
batches: impl IntoIterator<Item = RecordBatch>,
- ) -> Result<AppendResult> {
+ ) -> Result<WriteResult> {
let batches: Vec<RecordBatch> = batches.into_iter().collect();
crate::write::append_batches(self, &batches).await
}
diff --git a/crates/core/src/write/append.rs b/crates/core/src/write/append.rs
index 2aff8fd3..823a321b 100644
--- a/crates/core/src/write/append.rs
+++ b/crates/core/src/write/append.rs
@@ -45,6 +45,7 @@ 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::WriteResult;
use crate::write::keygen::{
hoodie_keys_for_batch, hoodie_keys_for_batch_with_offset,
relative_data_path,
};
@@ -52,23 +53,8 @@ use crate::write::metadata::is_column_stats_enabled;
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 (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.
- pub base_file_path: String,
- /// Rows written.
- pub num_rows: usize,
-}
-
/// Append one or more record batches as a single INSERT commit.
-pub async fn append_batches(table: &mut Table, batches: &[RecordBatch]) ->
Result<AppendResult> {
+pub async fn append_batches(table: &mut Table, batches: &[RecordBatch]) ->
Result<WriteResult> {
let batches = &crate::write::normalize_for_write(batches)?;
if table_version(table) < 8 {
return append_version_6(table, batches).await;
@@ -131,7 +117,6 @@ pub async fn append_batches(table: &mut Table, batches:
&[RecordBatch]) -> Resul
.then(|| crate::write::metadata::instant_millis(&table.hudi_configs,
&request_instant));
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;
@@ -227,9 +212,6 @@ pub async fn append_batches(table: &mut Table, batches:
&[RecordBatch]) -> Resul
))
})?;
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],
@@ -357,12 +339,12 @@ pub async fn append_batches(table: &mut Table, batches:
&[RecordBatch]) -> Resul
crate::write::maybe_compact_inline(table).await;
}
- Ok(AppendResult {
- instant: request_instant,
+ Ok(WriteResult {
+ requested_time: Some(request_instant),
completion_time: completed.completion_time,
- commit_relative_path: completed.commit_relative_path,
- base_file_path: primary_base_path,
- num_rows,
+ num_writes: num_rows,
+ num_inserts: num_rows,
+ ..Default::default()
})
}
@@ -457,7 +439,7 @@ fn ensure_rows_present(batches: &[RecordBatch]) ->
Result<usize> {
/// failed write, and writes no markers and no archival, so it does not share
/// `complete_write` with the version 8 path, and a crashed write leaves
/// nothing on the timeline for a later writer to roll back.
-async fn append_version_6(table: &mut Table, batches: &[RecordBatch]) ->
Result<AppendResult> {
+async fn append_version_6(table: &mut Table, batches: &[RecordBatch]) ->
Result<WriteResult> {
ensure_batches_share_schema(batches)?;
let batches = &match_table_schema(table, batches).await?;
let schema = batches[0].schema();
@@ -525,12 +507,11 @@ async fn append_version_6(table: &mut Table, batches:
&[RecordBatch]) -> Result<
// The commit is durable; a failed refresh must not turn it into an error.
let _ = table.timeline.reload_completed_commits().await;
table.clear_read_caches();
- Ok(AppendResult {
- instant,
- completion_time: None,
- commit_relative_path,
- base_file_path,
- num_rows,
+ Ok(WriteResult {
+ requested_time: Some(instant),
+ num_writes: num_rows,
+ num_inserts: num_rows,
+ ..Default::default()
})
}
diff --git a/crates/core/src/write/mod.rs b/crates/core/src/write/mod.rs
index 2ed9bfbe..c81505d7 100644
--- a/crates/core/src/write/mod.rs
+++ b/crates/core/src/write/mod.rs
@@ -39,7 +39,7 @@ pub(crate) use keygen::{
};
pub use lock::{InProcessLockProvider, LockLease, LockProvider};
pub(crate) use rewrite::ensure_writable_table_version;
-pub use rewrite::{CompactionResult, UpsertOptions, WriteResult};
+pub use rewrite::{CompactionResult, UpsertOptions};
pub(crate) use rewrite::{
await_async_compaction, compact_table, delete_filter, delete_keys,
dynamic_partition_overwrite_batches, maybe_compact_inline,
overwrite_batches, update_filter,
@@ -761,15 +761,35 @@ pub(crate) async fn complete_write(
let _ = table.timeline.reload_completed_commits().await;
table.clear_read_caches();
Ok(CompletedWrite {
- commit_relative_path,
completion_time: completion,
})
}
+/// Result of a write on [`Table`](crate::table::Table), shared by every write
verb.
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct WriteResult {
+ /// Requested time of the completed commit's instant, or `None` when the
+ /// write matched nothing and made no commit.
+ pub requested_time: Option<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>,
+ /// Rows in the data files this write produced: every row of each base
+ /// file it wrote, or each row appended to a merge-on-read log (deletes
+ /// append none).
+ pub num_writes: usize,
+ /// Number of input keys that replaced existing rows.
+ pub num_updates: usize,
+ /// Number of input keys not previously present. An append looks up no
+ /// keys and counts every row.
+ pub num_inserts: usize,
+ /// Number of rows deleted.
+ pub num_deletes: usize,
+}
+
/// 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>,
diff --git a/crates/core/src/write/rewrite/compaction.rs
b/crates/core/src/write/rewrite/compaction.rs
index ce09d30d..803a4db6 100644
--- a/crates/core/src/write/rewrite/compaction.rs
+++ b/crates/core/src/write/rewrite/compaction.rs
@@ -36,9 +36,9 @@ use crate::write::markers::Marker;
/// Result of a compaction.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct CompactionResult {
- /// Completed instant timestamp (the requested time); empty when no file
- /// group carried log files and nothing was written.
- pub instant: String,
+ /// Requested time of the compaction commit's instant, or `None` when no
+ /// file group carried log files and nothing was written.
+ pub requested_time: Option<String>,
/// Completion time of the instant on timeline layout v2; `None` on
/// layout v1 and when nothing was written.
pub completion_time: Option<String>,
@@ -185,13 +185,15 @@ pub(crate) async fn maybe_compact_inline(table: &mut
Table) {
}
if inline {
match compact_table(table).await {
- Ok(result) if result.instant.is_empty() => {}
- Ok(result) => log::info!(
- "inline compaction {} folded {} log files of {} file groups",
- result.instant,
- result.num_log_files,
- result.num_file_groups
+ Ok(CompactionResult {
+ requested_time: Some(instant),
+ num_file_groups,
+ num_log_files,
+ ..
+ }) => log::info!(
+ "inline compaction {instant} folded {num_log_files} log files
of {num_file_groups} file groups"
),
+ Ok(_) => {}
Err(error) => log::warn!("inline compaction failed: {error}"),
}
return;
@@ -499,7 +501,7 @@ async fn execute_compaction(
)
.await?;
Ok(CompactionResult {
- instant: instant.to_string(),
+ requested_time: Some(instant.to_string()),
completion_time: completed.completion_time,
num_file_groups: groups.len(),
num_log_files,
@@ -668,7 +670,10 @@ mod tests {
assert_eq!(with_logs.len(), 2, "a's group in sf and c's group in ny");
let result = table.compact().await.unwrap();
- assert!(!result.instant.is_empty());
+ let instant = result
+ .requested_time
+ .as_deref()
+ .expect("compaction commits");
assert!(result.completion_time.is_some());
assert_eq!(result.num_file_groups, 2);
assert_eq!(result.num_log_files, 2);
@@ -680,7 +685,11 @@ mod tests {
let after = snapshot(&table).await;
assert_eq!(data_only(&after), data_only(&before));
let row_a = after.iter().find(|r| r.0 == "a").unwrap();
- assert_eq!(row_a.4, upsert.instant, "compaction preserves commit
time");
+ assert_eq!(
+ Some(row_a.4.clone()),
+ upsert.requested_time,
+ "compaction preserves commit time"
+ );
assert_eq!(
row_a.5,
format!(
@@ -690,7 +699,7 @@ mod tests {
.find(|s| s.partition_path == "city=sf")
.unwrap()
.file_id(),
- result.instant
+ instant
),
"compaction stamps the new base file name"
);
@@ -699,7 +708,7 @@ mod tests {
assert_eq!(
slices
.iter()
- .filter(|s| s.creation_instant_time() == result.instant)
+ .filter(|s| s.creation_instant_time() == instant)
.count(),
2
);
@@ -707,8 +716,7 @@ mod tests {
// Timeline: plan in the requested file, empty inflight, compacted
commit.
let timeline = dir.path().join(".hoodie/timeline");
let plan = CompactionPlan::from_avro_bytes(
- &std::fs::read(timeline.join(format!("{}.compaction.requested",
result.instant)))
- .unwrap(),
+ &std::fs::read(timeline.join(format!("{}.compaction.requested",
instant))).unwrap(),
)
.unwrap();
let operations = plan.operations.clone().unwrap();
@@ -735,13 +743,13 @@ mod tests {
assert_eq!(plan.preserve_hoodie_metadata, Some(true));
assert!(plan.is_scheduled_by_hudi_rs());
assert!(
- std::fs::read(timeline.join(format!("{}.compaction.inflight",
result.instant)))
+ std::fs::read(timeline.join(format!("{}.compaction.inflight",
instant)))
.unwrap()
.is_empty()
);
let commit_name = format!(
"{}_{}.commit",
- result.instant,
+ instant,
result.completion_time.as_deref().unwrap()
);
assert_eq!(completed_commits(dir.path()), vec![commit_name.clone()]);
@@ -779,14 +787,17 @@ mod tests {
.into_iter()
.find(|s| s.partition_path == "city=sf")
.unwrap();
- assert_eq!(sf.creation_instant_time(), result.instant);
+ assert_eq!(
+ Some(sf.creation_instant_time()),
+ result.requested_time.as_deref()
+ );
assert_eq!(sf.log_files.len(), 1);
let row_b = snapshot(&table)
.await
.into_iter()
.find(|r| r.0 == "b")
.unwrap();
- assert_eq!((row_b.3, row_b.4), (21, later.instant));
+ assert_eq!((row_b.3, row_b.4), (21, later.requested_time.unwrap()));
}
#[tokio::test]
@@ -1065,7 +1076,7 @@ mod tests {
assert_eq!(slices[0].log_files.len(), 1, "a pending log is not
planned");
finish_delta_commit_unlocked(&table, before).await;
let during = table.upsert([batch(&[("b", "sf", 2,
21)])]).await.unwrap();
- assert!(during.instant > instant);
+ assert!(during.requested_time.unwrap() > instant);
execute_compaction(&mut table, &instant, slices)
.await
.unwrap();
diff --git a/crates/core/src/write/rewrite/cow.rs
b/crates/core/src/write/rewrite/cow.rs
index ab688126..6fbf1e22 100644
--- a/crates/core/src/write/rewrite/cow.rs
+++ b/crates/core/src/write/rewrite/cow.rs
@@ -317,7 +317,7 @@ pub(super) async fn finalize_rewrite_commit(
)
.await?;
Ok(WriteResult {
- instant: instant.to_string(),
+ requested_time: Some(instant.to_string()),
completion_time: completed.completion_time,
num_writes,
num_updates: updates,
diff --git a/crates/core/src/write/rewrite/mod.rs
b/crates/core/src/write/rewrite/mod.rs
index 087f6496..c692d229 100644
--- a/crates/core/src/write/rewrite/mod.rs
+++ b/crates/core/src/write/rewrite/mod.rs
@@ -45,6 +45,7 @@ use crate::metadata::meta_field::MetaField;
use crate::metadata::table::encode::RecordIndexEntry;
use crate::table::{ReadOptions, Table};
use crate::timeline::instant::Action;
+use crate::write::WriteResult;
use crate::write::append::{
ensure_copy_on_write, generate_instant_time, is_layout_two,
prepare_batches_for_write,
timeline_dir,
@@ -68,27 +69,6 @@ use rows::*;
pub(crate) use validate::ensure_writable_table_version;
use validate::*;
-/// Result of an upsert, delete, or overwrite write.
-#[derive(Clone, Debug, Default, PartialEq, Eq)]
-pub struct WriteResult {
- /// 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>,
- /// Rows in the data files this write produced: every row of each
- /// rewritten copy-on-write base file, or each row appended to a
- /// merge-on-read log (deletes append none).
- 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 {
diff --git a/crates/core/src/write/rewrite/mor.rs
b/crates/core/src/write/rewrite/mor.rs
index cacfc897..8e03e75a 100644
--- a/crates/core/src/write/rewrite/mor.rs
+++ b/crates/core/src/write/rewrite/mor.rs
@@ -594,7 +594,7 @@ pub(super) async fn mor_upsert_batches(
.await?;
crate::write::maybe_compact_inline(table).await;
Ok(WriteResult {
- instant,
+ requested_time: Some(instant),
completion_time: completed.completion_time,
num_writes: incoming.num_rows(),
num_updates: updates,
@@ -771,7 +771,7 @@ pub(super) async fn mor_delete_keys(
.await?;
crate::write::maybe_compact_inline(table).await;
Ok(WriteResult {
- instant,
+ requested_time: Some(instant),
completion_time: completed.completion_time,
num_writes: 0,
num_updates: 0,
diff --git a/crates/core/tests/spark_parity_tests.rs
b/crates/core/tests/spark_parity_tests.rs
index e9b600b3..77bb2049 100644
--- a/crates/core/tests/spark_parity_tests.rs
+++ b/crates/core/tests/spark_parity_tests.rs
@@ -1369,7 +1369,7 @@ async fn parity_k_rs_compaction_spark_interop() {
);
assert_eq!(rs_read(path).await, expected, "rs read after rs compaction");
assert_eq!(
- spark_read_as_of(&env, path, &delete.instant),
+ spark_read_as_of(&env, path,
delete.requested_time.as_deref().unwrap()),
expected,
"spark time travel to the delta commit before the compaction"
);
diff --git a/crates/core/tests/table_bulk_insert_tests.rs
b/crates/core/tests/table_bulk_insert_tests.rs
index 17f449ae..e7c7cc7a 100644
--- a/crates/core/tests/table_bulk_insert_tests.rs
+++ b/crates/core/tests/table_bulk_insert_tests.rs
@@ -107,16 +107,13 @@ async fn append_only_writes_records_and_commit() ->
Result<()> {
// Instants other writers mint before this write must order before it.
let earlier_instant =
chrono::Local::now().format("%Y%m%d%H%M%S%3f").to_string();
let result = table.append([batch]).await?;
- assert_eq!(result.num_rows, 3);
- assert_eq!(
- result.commit_relative_path,
- format!(".hoodie/{}.commit", result.instant)
- );
+ assert_eq!((result.num_writes, result.num_inserts), (3, 3));
+ assert_eq!(result.completion_time, None);
+ let instant = result.requested_time.as_deref().expect("append commits");
assert!(
- chrono::NaiveDateTime::parse_from_str(&result.instant,
"%Y%m%d%H%M%S%3f").is_ok()
- && result.instant >= earlier_instant,
- "instant {} must be a yyyyMMddHHmmssSSS time at or after
{earlier_instant}",
- result.instant
+ chrono::NaiveDateTime::parse_from_str(instant,
"%Y%m%d%H%M%S%3f").is_ok()
+ && instant >= earlier_instant.as_str(),
+ "instant {instant} must be a yyyyMMddHHmmssSSS time at or after
{earlier_instant}"
);
let reopened = Table::new(temp_dir.path().to_str().expect("valid temp
path")).await?;
@@ -140,7 +137,7 @@ async fn append_only_writes_records_and_commit() ->
Result<()> {
assert_eq!(names.value(1), "bob");
assert_eq!(names.value(2), "carol");
- let commit_file = temp_dir.path().join(&result.commit_relative_path);
+ let commit_file =
temp_dir.path().join(format!(".hoodie/{instant}.commit"));
assert!(commit_file.exists(), "commit file should exist");
Ok(())
@@ -185,10 +182,14 @@ async fn append_only_writes_layout_v2_commit() ->
Result<()> {
)?;
let result = table.append([batch]).await?;
- assert!(result.commit_relative_path.starts_with(".hoodie/timeline/"));
- assert!(result.commit_relative_path.ends_with(".commit"));
-
- let commit_file = temp_dir.path().join(&result.commit_relative_path);
+ let commit_file = temp_dir.path().join(format!(
+ ".hoodie/timeline/{}_{}.commit",
+ result.requested_time.as_deref().expect("append commits"),
+ result
+ .completion_time
+ .as_deref()
+ .expect("layout v2 records completion")
+ ));
assert!(commit_file.exists(), "layout v2 commit file should exist");
let reopened = Table::new(temp_dir.path().to_str().expect("valid temp
path")).await?;
@@ -236,7 +237,7 @@ async fn append_only_writes_multiple_batches_as_one_file()
-> Result<()> {
)?;
let result = table.append([first, second]).await?;
- assert_eq!(result.num_rows, 3);
+ assert_eq!((result.num_writes, result.num_inserts), (3, 3));
let reopened = Table::new(temp_dir.path().to_str().expect("valid temp
path")).await?;
let records = reopened.read(&ReadOptions::new()).await?;
@@ -404,7 +405,7 @@ async fn
append_only_accepts_cleared_metadata_partitions_property() -> Result<()
let result = table
.append([id_name_batch(vec![1], vec!["alice"])])
.await?;
- assert_eq!(result.num_rows, 1);
+ assert_eq!((result.num_writes, result.num_inserts), (1, 1));
Ok(())
}
@@ -419,8 +420,9 @@ async fn append_only_writes_version_6_commit_metadata() ->
Result<()> {
.append([id_name_batch(vec![1, 2, 3], vec!["alice", "bob", "carol"])])
.await?;
+ let instant = result.requested_time.as_deref().expect("append commits");
let commit: serde_json::Value = serde_json::from_slice(&fs::read(
- temp_dir.path().join(&result.commit_relative_path),
+ temp_dir.path().join(format!(".hoodie/{instant}.commit")),
)?)
.expect("v1 commit metadata is JSON");
assert_eq!(commit["operationType"], "BULK_INSERT");
@@ -430,15 +432,12 @@ async fn append_only_writes_version_6_commit_metadata()
-> Result<()> {
assert_eq!(stats.len(), 1);
let stat = &stats[0];
let file_id = stat["fileId"].as_str().expect("fileId");
- assert_eq!(
- result.base_file_path,
- format!("{file_id}_0-0-0_{}.parquet", result.instant)
- );
- assert_eq!(stat["path"], result.base_file_path.as_str());
+ let base_file = format!("{file_id}_0-0-0_{instant}.parquet");
+ assert_eq!(stat["path"], base_file.as_str());
assert_eq!(stat["partitionPath"], "");
assert_eq!(stat["numWrites"], 3);
assert_eq!(stat["numInserts"], 3);
- let file_size =
fs::metadata(temp_dir.path().join(&result.base_file_path))?.len();
+ let file_size = fs::metadata(temp_dir.path().join(&base_file))?.len();
assert_eq!(stat["fileSizeInBytes"], file_size);
let schema: serde_json::Value = serde_json::from_str(
@@ -573,13 +572,16 @@ async fn
append_only_version_6_commits_support_time_travel_and_incremental_reads
.await?;
let as_of = |instant: &str|
ReadOptions::new().with_as_of_timestamp(instant);
- assert_eq!(ids_read_with(uri, as_of(&first.instant)).await?, vec![1, 2]);
assert_eq!(
- ids_read_with(uri, as_of(&second.instant)).await?,
+ ids_read_with(uri,
as_of(first.requested_time.as_deref().unwrap())).await?,
+ vec![1, 2]
+ );
+ assert_eq!(
+ ids_read_with(uri,
as_of(second.requested_time.as_deref().unwrap())).await?,
vec![1, 2, 3]
);
assert_eq!(
- ids_read_with(uri, as_of(&third.instant)).await?,
+ ids_read_with(uri,
as_of(third.requested_time.as_deref().unwrap())).await?,
vec![1, 2, 3, 4, 5]
);
@@ -591,17 +593,18 @@ async fn
append_only_version_6_commits_support_time_travel_and_incremental_reads
assert_eq!(
ids_read_with(
uri,
- incremental(&first.instant).with_end_timestamp(&second.instant)
+ incremental(first.requested_time.as_deref().unwrap())
+ .with_end_timestamp(second.requested_time.as_deref().unwrap())
)
.await?,
vec![3]
);
assert_eq!(
- ids_read_with(uri, incremental(&first.instant)).await?,
+ ids_read_with(uri,
incremental(first.requested_time.as_deref().unwrap())).await?,
vec![3, 4, 5]
);
assert_eq!(
- ids_read_with(uri, incremental(&third.instant)).await?,
+ ids_read_with(uri,
incremental(third.requested_time.as_deref().unwrap())).await?,
Vec::<i32>::new()
);
@@ -735,9 +738,12 @@ async fn
append_only_without_table_version_takes_the_version_6_path() -> Result<
let result = table
.append([id_name_batch(vec![1, 2], vec!["alice", "bob"])])
.await?;
- assert_eq!(
- result.commit_relative_path,
- format!(".hoodie/{}.commit", result.instant)
+ let instant = result.requested_time.as_deref().expect("append commits");
+ assert!(
+ temp_dir
+ .path()
+ .join(format!(".hoodie/{instant}.commit"))
+ .is_file()
);
let rows: usize = table
.read(&ReadOptions::new())
diff --git a/crates/core/tests/table_write_archival_tests.rs
b/crates/core/tests/table_write_archival_tests.rs
index 04ede12e..9335b810 100644
--- a/crates/core/tests/table_write_archival_tests.rs
+++ b/crates/core/tests/table_write_archival_tests.rs
@@ -456,7 +456,8 @@ async fn plant_pending_then_archive(
.append([batch(vec![("k0", "sf", 0)])])
.await
.unwrap()
- .instant;
+ .requested_time
+ .unwrap();
let pending_ts = next_instant(&first);
let timeline = dir.join(".hoodie/timeline");
diff --git a/crates/core/tests/table_write_composite_keys_tests.rs
b/crates/core/tests/table_write_composite_keys_tests.rs
index 73687f5f..bb977698 100644
--- a/crates/core/tests/table_write_composite_keys_tests.rs
+++ b/crates/core/tests/table_write_composite_keys_tests.rs
@@ -243,7 +243,7 @@ async fn test_simple_index_tags_virtual_composite_keys() {
.create()
.await
.unwrap();
- let append = table
+ table
.append([order_batch(&[(1, "a", 10), (1, "b", 20)])])
.await
.unwrap();
@@ -254,7 +254,9 @@ async fn test_simple_index_tags_virtual_composite_keys() {
.await
.unwrap();
let location = locations[&stored].as_ref().expect("stored key is tagged");
- assert!(append.base_file_path.contains(&location.file_id));
+ let slices = table.get_file_slices(&ReadOptions::new()).await.unwrap();
+ assert_eq!(slices.len(), 1);
+ assert_eq!(slices[0].file_id(), location.file_id);
assert!(locations[&absent].is_none());
}
diff --git a/crates/core/tests/table_write_lifecycle_tests.rs
b/crates/core/tests/table_write_lifecycle_tests.rs
index 936a599e..63969f58 100644
--- a/crates/core/tests/table_write_lifecycle_tests.rs
+++ b/crates/core/tests/table_write_lifecycle_tests.rs
@@ -209,7 +209,7 @@ async fn
test_cow_lifecycle_append_upsert_delete_overwrite_with_mdt() {
.is_file()
);
- let append = table
+ table
.append([batch(vec![("a", 1), ("b", 2), ("c", 3)])])
.await
.unwrap();
@@ -223,11 +223,9 @@ async fn
test_cow_lifecycle_append_upsert_delete_overwrite_with_mdt() {
replacecommits: 0,
}
);
- assert!(
- mdt_active_files(&table)
- .await
- .contains(&append.base_file_path)
- );
+ let slices = table.get_file_slices(&ReadOptions::new()).await.unwrap();
+ let base_file = slices[0].base_file_relative_path().unwrap().unwrap();
+ assert!(mdt_active_files(&table).await.contains(&base_file));
assert_snapshot_async(&table, vec![("a", 1), ("b", 2), ("c", 3)]).await;
// 2) Upsert over time: update + insert; state stays consistent.
@@ -299,11 +297,10 @@ async fn
test_mor_lifecycle_asserts_base_and_log_file_counts() {
.unwrap();
// Append → base parquet + deltacommit, no logs yet.
- let append = table
+ table
.append([ordered_batch(vec![("a", 1, 1), ("b", 2, 1)])])
.await
.unwrap();
- assert!(append.commit_relative_path.ends_with(".deltacommit"));
assert_eq!(
data_layout(root),
FileLayout {
@@ -341,7 +338,10 @@ async fn
test_mor_lifecycle_asserts_base_and_log_file_counts() {
.unwrap()
.to_str()
.unwrap()
- .contains(&format!("_{}.log.", update.instant)),
+ .contains(&format!(
+ "_{}.log.",
+ update.requested_time.as_deref().unwrap()
+ )),
"log file should include upsert instant: {:?}",
logs[0]
);
diff --git a/crates/core/tests/table_write_rollback_tests.rs
b/crates/core/tests/table_write_rollback_tests.rs
index 8c5433c2..a6ad38f2 100644
--- a/crates/core/tests/table_write_rollback_tests.rs
+++ b/crates/core/tests/table_write_rollback_tests.rs
@@ -117,7 +117,10 @@ async fn test_marker_dir_removed_after_successful_commit()
{
let mut table = create_partitioned(dir.path()).await;
let result = table.append([batch(vec![("a", "sf", 1)])]).await.unwrap();
- let marker_dir = dir.path().join(".hoodie/.temp").join(&result.instant);
+ let marker_dir = dir
+ .path()
+ .join(".hoodie/.temp")
+ .join(result.requested_time.as_deref().unwrap());
let leftovers = std::fs::read_dir(&marker_dir)
.map(|entries| entries.count())
.unwrap_or(0);
@@ -789,13 +792,13 @@ async fn
test_pending_rollback_of_completed_instant_is_refused() {
let rollback_ts = "30000101000000017";
let plan = HoodieRollbackPlan {
instant_to_rollback: Some(HoodieInstantInfo {
- commit_time: committed.instant.clone(),
+ commit_time: committed.requested_time.clone().unwrap(),
action: "commit".to_string(),
}),
rollback_requests: Some(vec![HoodieRollbackRequest {
partition_path: "city=sf".to_string(),
file_id: Some(data_file.split('_').next().unwrap().to_string()),
- latest_base_instant: Some(committed.instant.clone()),
+ latest_base_instant:
Some(committed.requested_time.clone().unwrap()),
files_to_be_deleted: vec![absolute(&table, &relative)],
log_blocks_to_be_deleted: Some(Default::default()),
}]),
@@ -815,7 +818,7 @@ async fn
test_pending_rollback_of_completed_instant_is_refused() {
let err = result.unwrap_err();
assert!(
matches!(&err, CoreError::Unsupported(msg)
- if msg.contains(rollback_ts) && msg.contains(&committed.instant)),
+ if msg.contains(rollback_ts) &&
msg.contains(committed.requested_time.as_deref().unwrap())),
"{err}"
);
assert_eq!(pending_rollbacks(&timeline), vec![rollback_ts.to_string()]);
diff --git a/crates/core/tests/table_write_tests.rs
b/crates/core/tests/table_write_tests.rs
index 41a00d3e..78169026 100644
--- a/crates/core/tests/table_write_tests.rs
+++ b/crates/core/tests/table_write_tests.rs
@@ -137,6 +137,13 @@ fn rows_by_id(batches: &[RecordBatch]) -> Vec<(String,
i64)> {
rows
}
+/// The base file of the table's only file slice, relative to the table base.
+async fn only_base_file(table: &Table) -> String {
+ let slices = table.get_file_slices(&ReadOptions::new()).await.unwrap();
+ assert_eq!(slices.len(), 1);
+ slices[0].base_file_relative_path().unwrap().unwrap()
+}
+
#[tokio::test]
async fn test_create_and_append_then_read() {
let dir = tempdir().unwrap();
@@ -150,28 +157,31 @@ async fn test_create_and_append_then_read() {
.unwrap();
let result = table.append([sample_batch()]).await.unwrap();
- assert_eq!(result.num_rows, 3);
+ assert_eq!((result.num_writes, result.num_inserts), (3, 3));
+ let instant = result.requested_time.as_deref().unwrap();
+ let base_file = only_base_file(&table).await;
assert!(
- dir.path().join(&result.base_file_path).is_file(),
- "base file should exist at {}",
- result.base_file_path
+ dir.path().join(&base_file).is_file(),
+ "base file should exist at {base_file}"
);
+ let timeline = dir.path().join(".hoodie/timeline");
+ let commit_file = timeline.join(format!(
+ "{instant}_{}.commit",
+ result.completion_time.as_deref().unwrap()
+ ));
assert!(
- dir.path().join(&result.commit_relative_path).is_file(),
+ commit_file.is_file(),
"commit file should exist at {}",
- result.commit_relative_path
+ commit_file.display()
);
- let timeline = dir.path().join(".hoodie/timeline");
assert!(
timeline
- .join(format!("{}.commit.requested", result.instant))
+ .join(format!("{instant}.commit.requested"))
.is_file(),
"data timeline should fence with .commit.requested"
);
assert!(
- timeline
- .join(format!("{}.inflight", result.instant))
- .is_file(),
+ timeline.join(format!("{instant}.inflight")).is_file(),
"data timeline should fence with .inflight"
);
@@ -395,7 +405,7 @@ async fn
test_create_with_metadata_and_append_updates_files_partition() {
.create()
.await
.unwrap();
- let result = table.append([sample_batch()]).await.unwrap();
+ table.append([sample_batch()]).await.unwrap();
assert!(table.is_metadata_table_enabled());
let reopened = Table::new(base_uri).await.unwrap();
@@ -407,7 +417,13 @@ async fn
test_create_with_metadata_and_append_updates_files_partition() {
.await
.unwrap();
let files = records.get("").unwrap();
- assert!(files.has_active_file(&result.base_file_path));
+ let written: Vec<String> = std::fs::read_dir(dir.path())
+ .unwrap()
+ .map(|entry| entry.unwrap().file_name().into_string().unwrap())
+ .filter(|name| name.ends_with(".parquet"))
+ .collect();
+ assert_eq!(written.len(), 1);
+ assert!(files.has_active_file(&written[0]));
}
#[tokio::test]
@@ -477,13 +493,13 @@ async fn
test_write_result_completion_time_matches_timeline() {
}
.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),
+ ("append", &append.requested_time, &append.completion_time),
+ ("upsert", &upsert.requested_time, &upsert.completion_time),
+ ("delete", &delete.requested_time, &delete.completion_time),
] {
let completed = instants
.iter()
- .find(|i| &i.timestamp == instant)
+ .find(|i| Some(&i.timestamp) == instant.as_ref())
.unwrap_or_else(|| panic!("{table_type:?}: {verb} instant not
on the timeline"));
assert!(completion_time.is_some(), "{table_type:?} {verb}");
assert_eq!(
@@ -1443,8 +1459,16 @@ async fn
test_merge_on_read_writes_parquet_logs_and_merges_snapshot() {
.append([ordered_batch(vec![("a", 1, 1), ("b", 2, 1)])])
.await
.unwrap();
- assert!(append.commit_relative_path.ends_with(".deltacommit"));
- assert!(dir.path().join(&append.base_file_path).is_file());
+ assert!(
+ dir.path()
+ .join(format!(
+ ".hoodie/timeline/{}_{}.deltacommit",
+ append.requested_time.as_deref().unwrap(),
+ append.completion_time.as_deref().unwrap()
+ ))
+ .is_file()
+ );
+ assert!(dir.path().join(only_base_file(&table).await).is_file());
let incremental = table
.read(
&ReadOptions::new()
@@ -1593,7 +1617,7 @@ async fn
test_upsert_deduplicates_keys_and_delete_missing_key_is_noop() {
.await
.unwrap();
assert_eq!(result.num_deletes, 0);
- assert!(result.instant.is_empty());
+ assert!(result.requested_time.is_none());
}
#[tokio::test]
@@ -1890,7 +1914,11 @@ async fn test_input_meta_columns_are_replaced() {
.as_any()
.downcast_ref::<StringArray>()
.unwrap();
- assert_eq!(commit_times.value(0), result.instant, "{table_type:?}");
+ assert_eq!(
+ Some(commit_times.value(0)),
+ result.requested_time.as_deref(),
+ "{table_type:?}"
+ );
let mut fields = vec![Field::new("_hoodie_record_key", DataType::Utf8,
true)];
fields.extend(
@@ -3232,7 +3260,7 @@ async fn test_mor_unsupported_verbs_error_clearly() {
.count();
let result = table.delete("value = 12345").await.unwrap();
assert_eq!(result.num_deletes, 0);
- assert_eq!(result.instant, "");
+ assert_eq!(result.requested_time, None);
let commits_after = std::fs::read_dir(dir.path().join(".hoodie/timeline"))
.unwrap()
.count();
@@ -4247,7 +4275,12 @@ async fn
upserts_into_spark_written_table_read_through_metadata_table() {
// Spark leaves the timeline timezone unset, which Java reads as LOCAL.
let local_now = ||
chrono::Local::now().format("%Y%m%d%H%M%S%3f").to_string();
let before = local_now();
- let instant = table.upsert([rows.clone()]).await.unwrap().instant;
+ let instant = table
+ .upsert([rows.clone()])
+ .await
+ .unwrap()
+ .requested_time
+ .unwrap();
let after = local_now();
assert!(
before <= instant && instant <= after,
@@ -4539,7 +4572,7 @@ async fn test_delete_keys_in_wrong_partition_is_noop() {
.await
.unwrap();
assert_eq!(result.num_deletes, 0, "{table_type:?}");
- assert!(result.instant.is_empty(), "{table_type:?}");
+ assert!(result.requested_time.is_none(), "{table_type:?}");
assert_eq!(timeline_file_count(dir.path()), before, "{table_type:?}");
}
}
diff --git a/python/src/internal.rs b/python/src/internal.rs
index 524c614b..293ef66a 100644
--- a/python/src/internal.rs
+++ b/python/src/internal.rs
@@ -50,7 +50,7 @@ use hudi::file_group::reader::FileGroupReader;
use hudi::storage::error::StorageError;
use hudi::table::CompactionResult;
use hudi::table::builder::TableBuilder;
-use hudi::table::{AppendResult, QueryType, ReadOptions, Table};
+use hudi::table::{QueryType, ReadOptions, Table};
use hudi::timeline::Timeline;
use hudi::timeline::instant::Instant;
use hudi::{HoodieKey, UpsertOptions, WriteResult};
@@ -1088,7 +1088,7 @@ impl HudiCompactionResult {
impl From<CompactionResult> for HudiCompactionResult {
fn from(r: CompactionResult) -> Self {
Self {
- instant_time: committed(r.instant),
+ instant_time: r.requested_time,
completion_time: r.completion_time,
num_file_groups: r.num_file_groups,
num_log_files: r.num_log_files,
@@ -1132,7 +1132,7 @@ impl HudiWriteResult {
impl From<WriteResult> for HudiWriteResult {
fn from(r: WriteResult) -> Self {
Self {
- instant_time: committed(r.instant),
+ instant_time: r.requested_time,
completion_time: r.completion_time,
num_writes: r.num_writes,
num_updates: r.num_updates,
@@ -1142,24 +1142,6 @@ impl From<WriteResult> for HudiWriteResult {
}
}
-impl From<AppendResult> for HudiWriteResult {
- fn from(r: AppendResult) -> Self {
- Self {
- instant_time: committed(r.instant),
- completion_time: r.completion_time,
- num_writes: r.num_rows,
- num_updates: 0,
- num_inserts: r.num_rows,
- num_deletes: 0,
- }
- }
-}
-
-/// The core reports a write that made no commit with an empty instant.
-fn committed(instant: String) -> Option<String> {
- (!instant.is_empty()).then_some(instant)
-}
-
/// A Python argument that is either one string or a list of strings.
#[cfg(not(tarpaulin_include))]
#[derive(FromPyObject)]
@@ -1821,7 +1803,6 @@ mod tests {
use arrow::datatypes::{DataType, Field, Fields, Schema, TimeUnit};
use arrow::record_batch::RecordBatch;
use hudi::WriteResult;
- use hudi::table::AppendResult;
use std::collections::HashMap;
use std::sync::Arc;
@@ -2077,7 +2058,7 @@ mod tests {
#[test]
fn write_results_convert_to_python_counts() {
let rewrite = HudiWriteResult::from(WriteResult {
- instant: "20240101000000000".to_string(),
+ requested_time: Some("20240101000000000".to_string()),
completion_time: None,
num_writes: 4,
num_updates: 1,
@@ -2094,35 +2075,6 @@ mod tests {
),
(4, 1, 2, 3)
);
-
- let append = HudiWriteResult::from(AppendResult {
- instant: "20240101000000001".to_string(),
- completion_time: None,
- commit_relative_path: String::new(),
- base_file_path: String::new(),
- num_rows: 5,
- });
- assert_eq!(append.instant_time.as_deref(), Some("20240101000000001"));
- assert_eq!(
- (
- append.num_writes,
- append.num_updates,
- append.num_inserts,
- append.num_deletes
- ),
- (5, 0, 5, 0)
- );
-
- // A write that matched nothing reports no instant.
- let noop = HudiWriteResult::from(WriteResult {
- instant: String::new(),
- completion_time: None,
- num_writes: 0,
- num_updates: 0,
- num_inserts: 0,
- num_deletes: 0,
- });
- assert_eq!(noop.instant_time, None);
}
#[test]
