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 54ce1fc0 feat(python): expose commit completion time on write results
(#887)
54ce1fc0 is described below
commit 54ce1fc0c9e29a26a0da78230d76adc7596116e7
Author: Y Ethan Guo <[email protected]>
AuthorDate: Fri Oct 30 02:09:07 2026 -0700
feat(python): expose commit completion time on write results (#887)
HudiWriteResult now carries completion_time, the core write result's
completion time of the committed instant. It is None when no commit is
written or the timeline records no completion times. Passing it to
with_start_timestamp reads the commits after the write; passing
instant_time would include the write itself, since incremental bounds
compare against completion times on timeline layout v2.
---
python/hudi/_internal.pyi | 6 ++++++
python/src/internal.rs | 20 ++++++++++++++------
python/tests/test_table_write.py | 31 ++++++++++++++++++++++++++++++-
3 files changed, 50 insertions(+), 7 deletions(-)
diff --git a/python/hudi/_internal.pyi b/python/hudi/_internal.pyi
index c5d3d9df..dfb7ae24 100644
--- a/python/hudi/_internal.pyi
+++ b/python/hudi/_internal.pyi
@@ -627,6 +627,11 @@ class HudiWriteResult:
Attributes:
instant_time (Optional[str]): Instant time of the completed commit, or
``None``
when no commit was made.
+ completion_time (Optional[str]): When the commit completed, or
``None`` when no
+ commit was made or the table's timeline records no completion
times (table
+ version 6). When set, incremental reads compare their start bound
against
+ completion times, so ``with_start_timestamp(completion_time)``
returns the
+ commits after this one while passing ``instant_time`` would
include this commit.
num_writes (int): Rows written to data files.
num_updates (int): Input rows that replaced existing records.
num_inserts (int): Input rows that were new records.
@@ -634,6 +639,7 @@ class HudiWriteResult:
"""
instant_time: Optional[str]
+ completion_time: Optional[str]
num_writes: int
num_updates: int
num_inserts: int
diff --git a/python/src/internal.rs b/python/src/internal.rs
index 8c3adbf7..9d807362 100644
--- a/python/src/internal.rs
+++ b/python/src/internal.rs
@@ -1043,6 +1043,7 @@ impl HudiTable {
#[pyclass(frozen, get_all)]
pub struct HudiWriteResult {
instant_time: Option<String>,
+ completion_time: Option<String>,
num_writes: usize,
num_updates: usize,
num_inserts: usize,
@@ -1053,13 +1054,18 @@ pub struct HudiWriteResult {
#[pymethods]
impl HudiWriteResult {
fn __repr__(&self) -> String {
- let instant_time = self
- .instant_time
- .as_ref()
- .map_or_else(|| "None".to_string(), |t| format!("'{t}'"));
+ let quoted = |t: &Option<String>| {
+ t.as_ref()
+ .map_or_else(|| "None".to_string(), |t| format!("'{t}'"))
+ };
format!(
- "HudiWriteResult(instant_time={instant_time}, num_writes={},
num_updates={}, num_inserts={}, num_deletes={})",
- self.num_writes, self.num_updates, self.num_inserts,
self.num_deletes,
+ "HudiWriteResult(instant_time={}, completion_time={},
num_writes={}, num_updates={}, num_inserts={}, num_deletes={})",
+ quoted(&self.instant_time),
+ quoted(&self.completion_time),
+ self.num_writes,
+ self.num_updates,
+ self.num_inserts,
+ self.num_deletes,
)
}
}
@@ -1068,6 +1074,7 @@ impl From<WriteResult> for HudiWriteResult {
fn from(r: WriteResult) -> Self {
Self {
instant_time: committed(r.instant),
+ completion_time: r.completion_time,
num_writes: r.num_writes,
num_updates: r.num_updates,
num_inserts: r.num_inserts,
@@ -1080,6 +1087,7 @@ 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,
diff --git a/python/tests/test_table_write.py b/python/tests/test_table_write.py
index 64da9426..7ce5b1a5 100644
--- a/python/tests/test_table_write.py
+++ b/python/tests/test_table_write.py
@@ -24,7 +24,7 @@ from decimal import Decimal
import pyarrow as pa
import pytest
-from hudi import HudiTable, HudiWriteResult
+from hudi import HudiQueryType, HudiReadOptions, HudiTable, HudiWriteResult
SCHEMA = pa.schema(
[
@@ -210,6 +210,7 @@ def test_append_then_read(tmp_path, table_type):
)
assert "num_inserts=2" in repr(res)
assert f"instant_time='{res.instant_time}'" in repr(res)
+ assert f"completion_time='{res.completion_time}'" in repr(res)
with pytest.raises(AttributeError):
res.num_inserts = 0
assert snapshot(table) == {"t1": ("sf", 1, 30.0), "t2": ("nyc", 1, 45.0)}
@@ -243,6 +244,34 @@ def test_upsert_update_columns(tmp_path):
assert snapshot(table)["t1"] == ("sf", 1, 31.0)
+def test_completion_time_bounds_incremental_reads(tmp_path):
+ table = HudiTable.create(str(tmp_path / "t"), record_key="id",
ordering_fields="ts")
+ base = table.append(trips(("t1", "sf", 1, 30.0), ("t2", "sf", 1, 45.0)))
+ assert base.completion_time is not None
+ assert base.completion_time > base.instant_time
+ upsert = table.upsert(trips(("t2", "sf", 2, 50.0), ("t3", "nyc", 2, 15.0)))
+ assert upsert.completion_time > base.completion_time
+ no_op = table.delete_keys(["missing"])
+ assert (no_op.instant_time, no_op.completion_time) == (None, None)
+ assert "completion_time=None" in repr(no_op)
+
+ def changed_after(bound):
+ batches = table.read(
+ HudiReadOptions()
+ .with_query_type(HudiQueryType.Incremental)
+ .with_start_timestamp(bound)
+ )
+ return sorted(
+ id_ for batch in batches for id_ in batch.column("id").to_pylist()
+ )
+
+ # The bound is compared against completion times: the requested time of
+ # the base commit admits the base commit itself.
+ assert changed_after(base.completion_time) == ["t2", "t3"]
+ assert changed_after(base.instant_time) == ["t1", "t2", "t3"]
+ assert changed_after(upsert.completion_time) == []
+
+
@pytest.mark.parametrize("table_type", ["cow", "mor"])
def test_update_with_dict(tmp_path, table_type):
table = new_table(tmp_path, table_type)
