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 7cc171b4 refactor(python): name the result instant requested_time and
document the remaining stub items (#939)
7cc171b4 is described below
commit 7cc171b467b4ec75f75287aa1aa3b99ebf70bf53
Author: Y Ethan Guo <[email protected]>
AuthorDate: Fri Oct 30 13:44:03 2026 -0700
refactor(python): name the result instant requested_time and document the
remaining stub items (#939)
Adds docstrings to the 29 public items in `python/hudi/_internal.pyi` that
had none: the `HudiReadOptions` constructor, builders and accessors,
`HudiQueryType`, `HudiInstant`, `HudiTimeline`, and `HudiDataFusionDataSource`.
Each statement was checked against the binding and the core it calls, for
example that the incremental window bounds completion time on table version 8
and later, and that the timeline getters read the active timeline only. The
package and stub module docstrings li [...]
`HudiWriteResult.instant_time` and `HudiCompactionResult.instant_time`
become `requested_time`, matching the Rust `WriteResult` and `CompactionResult`
field and pairing with `completion_time`; the README quick start and the upsert
benchmark read the new name. Neither class has shipped in a release.
---
README.md | 4 +-
benchmark/upsert/bench.py | 6 +-
python/hudi/__init__.py | 22 +++
python/hudi/_internal.pyi | 283 ++++++++++++++++++++++++++++++++++-----
python/src/internal.rs | 18 +--
python/tests/test_table_write.py | 26 ++--
6 files changed, 297 insertions(+), 62 deletions(-)
diff --git a/README.md b/README.md
index b4cd1fa6..c2414d14 100644
--- a/README.md
+++ b/README.md
@@ -452,7 +452,7 @@ groups.
```python
result = table.upsert(batch)
-print(f"instant={result.instant_time} updates={result.num_updates}
inserts={result.num_inserts}")
+print(f"requested_time={result.requested_time} updates={result.num_updates}
inserts={result.num_inserts}")
# Partial upsert (copy-on-write only): matched rows take only these columns
from the input.
table.upsert(batch, update_columns=["fare"])
@@ -463,7 +463,7 @@ table.upsert(batch, update_columns=["fare"])
```rust
let result = table.upsert([batch]).await?;
println!(
- "instant={} updates={} inserts={}",
+ "requested_time={} updates={} inserts={}",
result.requested_time.as_deref().unwrap_or("none"),
result.num_updates,
result.num_inserts
diff --git a/benchmark/upsert/bench.py b/benchmark/upsert/bench.py
index bb186362..1d945c9a 100755
--- a/benchmark/upsert/bench.py
+++ b/benchmark/upsert/bench.py
@@ -259,7 +259,7 @@ class HudiCase(Case):
def _marker(r):
# Time travel resolves requested instants; incremental bounds on table
# version 9 are completion times.
- return {"instant": r.instant_time, "completion": r.completion_time}
+ return {"instant": r.requested_time, "completion": r.completion_time}
def upsert(self, batch):
r = self.t.upsert(batch)
@@ -267,7 +267,7 @@ class HudiCase(Case):
return {
"num_updated": r.num_updates,
"num_inserted": r.num_inserts,
- "instant": r.instant_time,
+ "instant": r.requested_time,
}
def record_keys(self, keys):
@@ -283,7 +283,7 @@ class HudiCase(Case):
def delete(self, keys):
r = self.t.delete_keys(self.record_keys(keys))
self.last = self._marker(r)
- return {"num_deleted": r.num_deletes, "instant": r.instant_time}
+ return {"num_deleted": r.num_deletes, "instant": r.requested_time}
def _read(self, opts=None):
batches = self.t.read(opts)
diff --git a/python/hudi/__init__.py b/python/hudi/__init__.py
index 0f8dd4b8..a28570ee 100644
--- a/python/hudi/__init__.py
+++ b/python/hudi/__init__.py
@@ -15,6 +15,28 @@
# specific language governing permissions and limitations
# under the License.
+"""Apache Hudi for Python, backed by the native Rust implementation hudi-rs.
+
+Start with :class:`HudiTable`: ``HudiTable.create`` makes a table and
``HudiTable(base_uri)``
+opens an existing one; both return a handle that reads, writes, and, on a
merge-on-read table,
+compacts (:meth:`HudiTable.compact`, experimental).
+
+Resources:
+
+- Quick start: https://hudi.apache.org/docs/python-rust-quick-start-guide, and
the shorter
+ README quick start at https://github.com/apache/hudi-rs#quick-start.
+- Python API reference: the docstrings in the bundled type stubs,
``hudi/_internal.pyi``
+ (https://github.com/apache/hudi-rs/blob/main/python/hudi/_internal.pyi).
+- Rust API reference: https://docs.rs/hudi (also https://docs.rs/hudi-core and
+ https://docs.rs/hudi-datafusion).
+- Runnable Rust example:
+
https://github.com/apache/hudi-rs/blob/main/crates/core/examples/table_writes.rs.
+- Reader semantics:
https://github.com/apache/hudi-rs/blob/main/docs/reader-spec.md.
+ Writer design:
https://github.com/apache/hudi-rs/blob/main/docs/writer-design.md.
+- Packages: https://pypi.org/project/hudi/ and https://crates.io/crates/hudi.
+ Source: https://github.com/apache/hudi-rs.
+- Hudi concepts (timeline, table types, query types):
https://hudi.apache.org/docs/overview.
+"""
from hudi._config import HudiPlanConfig, HudiReadConfig, HudiTableConfig
from hudi._internal import (
diff --git a/python/hudi/_internal.pyi b/python/hudi/_internal.pyi
index 2ef6aab0..fbb314e5 100644
--- a/python/hudi/_internal.pyi
+++ b/python/hudi/_internal.pyi
@@ -14,6 +14,24 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+"""Type stubs and API reference for the ``hudi`` package.
+
+Resources:
+
+- Quick start: https://hudi.apache.org/docs/python-rust-quick-start-guide, and
the shorter
+ README quick start at https://github.com/apache/hudi-rs#quick-start.
+- Python API reference: the docstrings in this file.
+- Rust API reference: https://docs.rs/hudi (also https://docs.rs/hudi-core and
+ https://docs.rs/hudi-datafusion).
+- Runnable Rust example:
+
https://github.com/apache/hudi-rs/blob/main/crates/core/examples/table_writes.rs.
+- Reader semantics:
https://github.com/apache/hudi-rs/blob/main/docs/reader-spec.md.
+ Writer design:
https://github.com/apache/hudi-rs/blob/main/docs/writer-design.md.
+- Packages: https://pypi.org/project/hudi/ and https://crates.io/crates/hudi.
+ Source: https://github.com/apache/hudi-rs.
+- Hudi concepts (timeline, table types, query types):
https://hudi.apache.org/docs/overview.
+"""
+
from dataclasses import dataclass
from typing import Any, ClassVar, Dict, List, Optional, Protocol, Tuple, Union
@@ -45,9 +63,13 @@ class HudiQueryType:
Incremental: ClassVar["HudiQueryType"]
@property
- def name(self) -> str: ...
+ def name(self) -> str:
+ """The member name, ``"Snapshot"`` or ``"Incremental"``."""
+ ...
@property
- def value(self) -> str: ...
+ def value(self) -> str:
+ """The config value stored under ``hoodie.read.query.type``,
``"snapshot"`` or ``"incremental"``."""
+ ...
@dataclass(init=False)
class HudiReadOptions:
@@ -76,28 +98,85 @@ class HudiReadOptions:
filters: Optional[List[Tuple[str, str, str]]] = None,
projection: Optional[List[str]] = None,
hudi_options: Optional[Dict[str, str]] = None,
- ): ...
+ ):
+ """
+ Initializes the HudiReadOptions.
+
+ Parameters:
+ filters (Optional[List[Tuple[str, str, str]]]): Column filters as
``(field, op, value)``
+ tuples. Parsed and cardinality-validated immediately; an
unrecognized operator
+ or empty ``IN`` / ``NOT IN`` value list raises
``HudiCoreError``.
+ projection (Optional[List[str]]): Column names to read. If None,
all columns are read.
+ hudi_options (Optional[Dict[str, str]]): Per-read Hudi configs
(``hoodie.*``).
+ """
+ ...
# Builders return a new HudiReadOptions for chaining.
- def with_query_type(self, query_type: HudiQueryType) -> "HudiReadOptions":
...
- def with_as_of_timestamp(self, timestamp: str) -> "HudiReadOptions": ...
- def with_start_timestamp(self, timestamp: str) -> "HudiReadOptions": ...
- def with_end_timestamp(self, timestamp: str) -> "HudiReadOptions": ...
+ def with_query_type(self, query_type: HudiQueryType) -> "HudiReadOptions":
+ """Set the query type. Stored under ``hoodie.read.query.type``;
defaults to snapshot when unset."""
+ ...
+ def with_as_of_timestamp(self, timestamp: str) -> "HudiReadOptions":
+ """
+ Set the timestamp for a snapshot (time-travel) read. Stored under
+ ``hoodie.read.as.of.timestamp``; a snapshot read without it reads the
latest commit.
+
+ The value is parsed when the read runs, not here. Accepted formats are
the Hudi
+ timeline format (``yyyyMMddHHmmssSSS`` or ``yyyyMMddHHmmss``), epoch
time in seconds,
+ milliseconds, microseconds or nanoseconds, and RFC 3339
(``2024-03-15T14:25:30Z``).
+ Epoch and RFC 3339 values are converted to the table's timeline
timezone
+ (``hoodie.table.timeline.timezone``).
+ """
+ ...
+ def with_start_timestamp(self, timestamp: str) -> "HudiReadOptions":
+ """
+ Set the exclusive start of an incremental read window. Stored under
+ ``hoodie.read.start.timestamp``; when unset, the window has no lower
bound.
+
+ On table version 8 and later this bounds each commit's completion
time, so pass
+ :attr:`HudiWriteResult.completion_time` rather than its
``requested_time`` to read
+ the commits after a given write. On earlier table versions it bounds
the requested
+ time. Accepts the same formats as :meth:`with_as_of_timestamp`.
+ """
+ ...
+ def with_end_timestamp(self, timestamp: str) -> "HudiReadOptions":
+ """
+ Set the inclusive end of an incremental read window. Stored under
+ ``hoodie.read.end.timestamp``; when unset, the window covers
everything committed
+ so far.
+
+ Bounds the same timestamp as :meth:`with_start_timestamp` and accepts
the same
+ formats as :meth:`with_as_of_timestamp`.
+ """
+ ...
def with_batch_size(self, size: int) -> "HudiReadOptions":
"""Target rows per batch for streaming reads. Raises ``HudiCoreError``
if ``size == 0``."""
...
def with_filters(self, filters: List[Tuple[str, str, str]]) ->
"HudiReadOptions":
"""Set column filters as ``(field, op, value)`` tuples. Parses and
cardinality-validates immediately; an unrecognized operator or empty ``IN`` /
``NOT IN`` value list raises ``HudiCoreError``."""
...
- def with_projection(self, columns: List[str]) -> "HudiReadOptions": ...
- def with_hudi_option(self, key: str, value: str) -> "HudiReadOptions": ...
- def with_hudi_options(self, opts: Dict[str, str]) -> "HudiReadOptions": ...
+ def with_projection(self, columns: List[str]) -> "HudiReadOptions":
+ """Set the column names to read, replacing any previous projection."""
+ ...
+ def with_hudi_option(self, key: str, value: str) -> "HudiReadOptions":
+ """Set one Hudi config (``hoodie.*``) for this read only, replacing
any existing value for ``key``."""
+ ...
+ def with_hudi_options(self, opts: Dict[str, str]) -> "HudiReadOptions":
+ """Set several Hudi configs (``hoodie.*``) for this read only. Keys
already set and not in ``opts`` are kept."""
+ ...
# Typed accessors read from hudi_options.
- def query_type(self) -> HudiQueryType: ...
- def as_of_timestamp(self) -> Optional[str]: ...
- def start_timestamp(self) -> Optional[str]: ...
- def end_timestamp(self) -> Optional[str]: ...
+ def query_type(self) -> HudiQueryType:
+ """The query type, ``HudiQueryType.Snapshot`` if unset. Raises
``HudiCoreError`` if the stored value is not a recognized query type."""
+ ...
+ def as_of_timestamp(self) -> Optional[str]:
+ """The snapshot (time-travel) timestamp as stored, or ``None`` if
unset."""
+ ...
+ def start_timestamp(self) -> Optional[str]:
+ """The exclusive start of the incremental window as stored, or
``None`` if unset."""
+ ...
+ def end_timestamp(self) -> Optional[str]:
+ """The inclusive end of the incremental window as stored, or ``None``
if unset."""
+ ...
def batch_size(self) -> Optional[int]:
"""Target batch size for streaming reads, or ``None`` if unset. Raises
``HudiCoreError`` if the stored value is malformed."""
...
@@ -256,14 +335,49 @@ class HudiFileSlice:
@dataclass(init=False)
class HudiInstant:
+ """
+ An action on a table's timeline at one point in time, as returned by
:class:`HudiTimeline`.
+ """
+
@property
- def timestamp(self) -> str: ...
+ def timestamp(self) -> str:
+ """
+ The instant time, i.e. when the action was requested, in the Hudi
timeline format
+ (``yyyyMMddHHmmssSSS`` or ``yyyyMMddHHmmss``).
+
+ Returns:
+ str: The requested timestamp of the instant.
+ """
+ ...
@property
- def action(self) -> str: ...
+ def action(self) -> str:
+ """
+ The timeline action of the instant.
+
+ Returns:
+ str: ``"commit"``, ``"deltacommit"`` or ``"replacecommit"``.
+ """
+ ...
@property
- def state(self) -> str: ...
+ def state(self) -> str:
+ """
+ The state of the instant.
+
+ Returns:
+ str: ``"requested"``, ``"inflight"``, or an empty string for a
completed instant.
+ Instants returned by :class:`HudiTimeline` are all completed.
+ """
+ ...
@property
- def epoch_mills(self) -> int: ...
+ def epoch_mills(self) -> int:
+ """
+ The instant time (:attr:`timestamp`) as milliseconds since the Unix
epoch, reading it in
+ the table's timeline timezone (``hoodie.table.timeline.timezone``, UTC
when unset).
+
+ Returns:
+ int: Milliseconds since the Unix epoch.
+ """
+ ...
@dataclass(init=False)
class HudiTable:
@@ -404,6 +518,9 @@ class HudiTable:
- Snapshot: slices visible at ``options.as_of_timestamp`` (defaults to
the
latest commit). ``options.filters`` drive partition pruning;
file-level
stats pruning applies only for COW tables or MOR read-optimized mode.
+ With the record-level index, ``=`` or ``IN`` filters on the record
key
+ keep only the file groups the index names (latest snapshot only, not
in
+ read-optimized mode).
When footer stats are loaded for pruning, exact ``num_records``,
``byte_size``, and column stats are attached to the returned slices.
- Incremental: slices changed in ``(options.start_timestamp,
@@ -623,8 +740,8 @@ class HudiTable:
...
def compact(self) -> "HudiCompactionResult":
"""
- Compact a merge-on-read table: every file group whose latest file slice
- carries log files gets a new base file holding its merged rows, under
one
+ Compact a merge-on-read table (experimental): every file group whose
latest
+ file slice carries log files gets a new base file holding its merged
rows, under one
compaction instant that completes as a commit. Rows keep their commit
time,
so incremental reads see no new rows from a compaction.
@@ -662,20 +779,20 @@ class HudiWriteResult:
Outcome of a write on :class:`HudiTable`.
Attributes:
- instant_time (Optional[str]): Instant time of the completed commit, or
``None``
+ requested_time (Optional[str]): Requested 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.
+ commits after this one while passing ``requested_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.
num_deletes (int): Records deleted.
"""
- instant_time: Optional[str]
+ requested_time: Optional[str]
completion_time: Optional[str]
num_writes: int
num_updates: int
@@ -688,7 +805,7 @@ class HudiCompactionResult:
Outcome of :meth:`HudiTable.compact`.
Attributes:
- instant_time (Optional[str]): Instant time of the completed compaction
commit,
+ requested_time (Optional[str]): Requested time of the completed
compaction commit,
or ``None`` when no file group had log files and nothing was
written.
completion_time (Optional[str]): When the commit completed, or
``None`` when
nothing was written or the timeline records no completion times.
@@ -697,7 +814,7 @@ class HudiCompactionResult:
num_writes (int): Rows written to the new base files.
"""
- instant_time: Optional[str]
+ requested_time: Optional[str]
completion_time: Optional[str]
num_file_groups: int
num_log_files: int
@@ -705,16 +822,105 @@ class HudiCompactionResult:
@dataclass(init=False)
class HudiTimeline:
- def get_completed_commits(self, desc: bool = False) -> List[HudiInstant]:
...
- def get_completed_deltacommits(self, desc: bool = False) ->
List[HudiInstant]: ...
- def get_completed_replacecommits(self, desc: bool = False) ->
List[HudiInstant]: ...
- def get_completed_clustering_commits(
- self, desc: bool = False
- ) -> List[HudiInstant]: ...
- def get_instant_metadata_in_json(self, instant: HudiInstant) -> str: ...
- def get_latest_commit_timestamp(self) -> str: ...
- def get_latest_avro_schema(self) -> str: ...
- def get_latest_schema(self) -> "pyarrow.Schema": ...
+ """
+ The timeline of a Hudi table, obtained from :meth:`HudiTable.get_timeline`.
+
+ Only the active timeline is read; archived instants are not returned.
+ """
+
+ def get_completed_commits(self, desc: bool = False) -> List[HudiInstant]:
+ """
+ Returns the completed ``commit`` instants. On a MOR table, compactions
complete as commits.
+
+ Parameters:
+ desc (bool): Sort by instant time in descending order instead of
ascending.
+
+ Returns:
+ List[HudiInstant]: The completed commit instants.
+ """
+ ...
+ def get_completed_deltacommits(self, desc: bool = False) ->
List[HudiInstant]:
+ """
+ Returns the completed ``deltacommit`` instants. Empty for a COW table.
+
+ Parameters:
+ desc (bool): Sort by instant time in descending order instead of
ascending.
+
+ Returns:
+ List[HudiInstant]: The completed delta commit instants.
+ """
+ ...
+ def get_completed_replacecommits(self, desc: bool = False) ->
List[HudiInstant]:
+ """
+ Returns the completed ``replacecommit`` instants, such as overwrites
and clustering.
+
+ Parameters:
+ desc (bool): Sort by instant time in descending order instead of
ascending.
+
+ Returns:
+ List[HudiInstant]: The completed replace commit instants.
+ """
+ ...
+ def get_completed_clustering_commits(self, desc: bool = False) ->
List[HudiInstant]:
+ """
+ Returns the completed ``replacecommit`` instants whose commit metadata
has
+ ``operationType`` ``cluster``.
+
+ Reads the commit metadata of every completed replace commit, and raises
+ ``HudiCoreError`` if one of them has no operation type.
+
+ Parameters:
+ desc (bool): Sort by instant time in descending order instead of
ascending.
+
+ Returns:
+ List[HudiInstant]: The completed clustering instants.
+ """
+ ...
+ def get_instant_metadata_in_json(self, instant: HudiInstant) -> str:
+ """
+ Returns the commit metadata stored in the instant's timeline file, as
JSON.
+
+ Parameters:
+ instant (HudiInstant): A completed instant from this timeline.
+
+ Returns:
+ str: The commit metadata as a JSON string.
+ """
+ ...
+ def get_latest_commit_timestamp(self) -> str:
+ """
+ Returns the instant time of the latest completed commit, delta commit,
or replace
+ commit. Raises ``HudiCoreError`` if the table has no completed commit.
+
+ Returns:
+ str: The instant time in the Hudi timeline format.
+ """
+ ...
+ def get_latest_avro_schema(self) -> str:
+ """
+ Returns the Avro schema recorded in the latest completed commit's
metadata.
+
+ Raises ``HudiCoreError`` if the table has no completed commit or the
commit metadata
+ holds no schema. :meth:`HudiTable.get_schema_in_avro_str` instead
falls back to the
+ schema in the table config when the table has no completed commit.
+
+ Returns:
+ str: The Avro schema as a JSON string.
+ """
+ ...
+ def get_latest_schema(self) -> "pyarrow.Schema":
+ """
+ Returns the data schema, without Hudi meta fields, of the latest
completed commit.
+
+ Taken from the commit metadata, or, when the metadata holds no schema,
from a base
+ file referenced by the commit's write stats. Raises ``HudiCoreError``
if the table
+ has no completed commit; :meth:`HudiTable.get_schema` instead falls
back to the
+ schema in the table config.
+
+ Returns:
+ pyarrow.Schema: The schema of the latest commit.
+ """
+ ...
def build_hudi_table(
base_uri: str,
@@ -738,6 +944,13 @@ def build_hudi_table(
@dataclass(init=False)
class HudiDataFusionDataSource:
+ """
+ A Hudi table as a DataFusion table provider.
+
+ Register it on a ``datafusion.SessionContext`` with ``register_table`` and
query it with
+ SQL. Requires the ``datafusion`` package, installed with the
``hudi[datafusion]`` extra.
+ """
+
def __init__(
self,
base_uri: str,
diff --git a/python/src/internal.rs b/python/src/internal.rs
index 293ef66a..9ce85568 100644
--- a/python/src/internal.rs
+++ b/python/src/internal.rs
@@ -1059,7 +1059,7 @@ impl HudiTable {
#[cfg(not(tarpaulin_include))]
#[pyclass(frozen, get_all)]
pub struct HudiCompactionResult {
- instant_time: Option<String>,
+ requested_time: Option<String>,
completion_time: Option<String>,
num_file_groups: usize,
num_log_files: usize,
@@ -1075,8 +1075,8 @@ impl HudiCompactionResult {
.map_or_else(|| "None".to_string(), |t| format!("'{t}'"))
};
format!(
- "HudiCompactionResult(instant_time={}, completion_time={},
num_file_groups={}, num_log_files={}, num_writes={})",
- quoted(&self.instant_time),
+ "HudiCompactionResult(requested_time={}, completion_time={},
num_file_groups={}, num_log_files={}, num_writes={})",
+ quoted(&self.requested_time),
quoted(&self.completion_time),
self.num_file_groups,
self.num_log_files,
@@ -1088,7 +1088,7 @@ impl HudiCompactionResult {
impl From<CompactionResult> for HudiCompactionResult {
fn from(r: CompactionResult) -> Self {
Self {
- instant_time: r.requested_time,
+ requested_time: r.requested_time,
completion_time: r.completion_time,
num_file_groups: r.num_file_groups,
num_log_files: r.num_log_files,
@@ -1101,7 +1101,7 @@ impl From<CompactionResult> for HudiCompactionResult {
#[cfg(not(tarpaulin_include))]
#[pyclass(frozen, get_all)]
pub struct HudiWriteResult {
- instant_time: Option<String>,
+ requested_time: Option<String>,
completion_time: Option<String>,
num_writes: usize,
num_updates: usize,
@@ -1118,8 +1118,8 @@ impl HudiWriteResult {
.map_or_else(|| "None".to_string(), |t| format!("'{t}'"))
};
format!(
- "HudiWriteResult(instant_time={}, completion_time={},
num_writes={}, num_updates={}, num_inserts={}, num_deletes={})",
- quoted(&self.instant_time),
+ "HudiWriteResult(requested_time={}, completion_time={},
num_writes={}, num_updates={}, num_inserts={}, num_deletes={})",
+ quoted(&self.requested_time),
quoted(&self.completion_time),
self.num_writes,
self.num_updates,
@@ -1132,7 +1132,7 @@ impl HudiWriteResult {
impl From<WriteResult> for HudiWriteResult {
fn from(r: WriteResult) -> Self {
Self {
- instant_time: r.requested_time,
+ requested_time: r.requested_time,
completion_time: r.completion_time,
num_writes: r.num_writes,
num_updates: r.num_updates,
@@ -2065,7 +2065,7 @@ mod tests {
num_inserts: 2,
num_deletes: 3,
});
- assert_eq!(rewrite.instant_time.as_deref(), Some("20240101000000000"));
+ assert_eq!(rewrite.requested_time.as_deref(),
Some("20240101000000000"));
assert_eq!(
(
rewrite.num_writes,
diff --git a/python/tests/test_table_write.py b/python/tests/test_table_write.py
index 5e178786..c36e83df 100644
--- a/python/tests/test_table_write.py
+++ b/python/tests/test_table_write.py
@@ -211,7 +211,7 @@ def test_append_then_read(tmp_path, table_type):
)
res = table.append(trips(("t1", "sf", 1, 30.0), ("t2", "nyc", 1, 45.0)))
assert isinstance(res, HudiWriteResult)
- assert res.instant_time
+ assert res.requested_time
assert (res.num_writes, res.num_inserts, res.num_updates, res.num_deletes)
== (
2,
2,
@@ -219,7 +219,7 @@ def test_append_then_read(tmp_path, table_type):
0,
)
assert "num_inserts=2" in repr(res)
- assert f"instant_time='{res.instant_time}'" in repr(res)
+ assert f"requested_time='{res.requested_time}'" in repr(res)
assert f"completion_time='{res.completion_time}'" in repr(res)
with pytest.raises(AttributeError):
res.num_inserts = 0
@@ -258,11 +258,11 @@ 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
+ assert base.completion_time > base.requested_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 (no_op.requested_time, no_op.completion_time) == (None, None)
assert "completion_time=None" in repr(no_op)
def changed_after(bound):
@@ -278,7 +278,7 @@ def test_completion_time_bounds_incremental_reads(tmp_path):
# 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(base.requested_time) == ["t1", "t2", "t3"]
assert changed_after(upsert.completion_time) == []
@@ -328,8 +328,8 @@ def test_update_no_match_creates_no_commit(tmp_path):
table = new_table(tmp_path)
commits = len(table.get_timeline().get_completed_commits())
res = table.update("city = 'la'", {"fare": 1.0})
- assert (res.instant_time, res.num_updates) == (None, 0)
- assert "instant_time=None" in repr(res)
+ assert (res.requested_time, res.num_updates) == (None, 0)
+ assert "requested_time=None" in repr(res)
assert len(table.get_timeline().get_completed_commits()) == commits
@@ -526,7 +526,7 @@ def test_delete_by_non_key_predicate(tmp_path):
assert res.num_deletes == 1
assert set(snapshot(table)) == {"t1", "t3"}
res = table.delete("fare > 1000")
- assert (res.instant_time, res.num_deletes) == (None, 0)
+ assert (res.requested_time, res.num_deletes) == (None, 0)
@pytest.mark.parametrize("table_type", ["cow", "mor"])
@@ -536,7 +536,7 @@ def test_delete_keys(tmp_path, table_type):
assert res.num_deletes == 2
assert set(snapshot(table)) == {"t2"}
res = table.delete_keys(["missing"])
- assert (res.instant_time, res.num_deletes) == (None, 0)
+ assert (res.requested_time, res.num_deletes) == (None, 0)
res = table.delete_keys(k for k in ("t2",))
assert res.num_deletes == 1
assert snapshot(table) == {}
@@ -556,7 +556,7 @@ def test_delete_keys_rejects_bad_input(tmp_path):
def test_overwrite(tmp_path, table_type):
table = new_table(tmp_path, table_type)
res = table.overwrite(trips(("t9", "sf", 9, 10.0)))
- assert res.instant_time
+ assert res.requested_time
assert snapshot(table) == {"t9": ("sf", 9, 10.0)}
@@ -564,7 +564,7 @@ def test_overwrite(tmp_path, table_type):
def test_dynamic_partition_overwrite(tmp_path, table_type):
table = new_table(tmp_path, table_type)
res = table.dynamic_partition_overwrite(trips(("t8", "sf", 9, 11.0)))
- assert res.instant_time
+ assert res.requested_time
assert snapshot(table) == {"t8": ("sf", 9, 11.0), "t3": ("nyc", 1, 20.0)}
@@ -819,12 +819,12 @@ def test_compact_folds_logs_and_keeps_rows(tmp_path):
before = snapshot(table)
result = table.compact()
assert isinstance(result, HudiCompactionResult)
- assert result.instant_time is not None and result.completion_time is not
None
+ assert result.requested_time is not None and result.completion_time is not
None
assert (result.num_file_groups, result.num_log_files) == (1, 1)
assert result.num_writes == 2
assert snapshot(table) == before
nothing = table.compact()
- assert nothing.instant_time is None and nothing.num_file_groups == 0
+ assert nothing.requested_time is None and nothing.num_file_groups == 0
assert table.await_compaction() is None
