GitHub user ravjot28 added a comment to the discussion: MoR support for the
standalone Java client (new HoodieJavaReadClient)
Following up on the `RecordMerger` direction, here is the reader design I plan
to open as a draft PR. I would like feedback on the API, and on whether you
would prefer this as an RFC first.
**Shape.** A reader class in `hudi-java-client`, built on what master already
has:
- each read call is planned from one loaded timeline: file listing, the latest
instant or the time-travel instant, the schema, and one
`TableState.snapshotOf(...)` that is passed to every file group reader of that
call;
- listing goes through `BaseHoodieTableFileIndex` (a small Java subclass, like
`HiveHoodieTableFileIndex`), so metadata-table listing, read-optimized listing
and time-travel validation are the code Spark and Hive use;
- each file slice is read by a `HoodieFileGroupReader` with the
`AvroReaderContextFactory`, so merging follows the table's merge mode:
event-time ordering, commit-time ordering, or a configured custom
`HoodieRecordMerger`. There is no Java-specific merge logic;
- the reader opens one file slice at a time, and each merge buffer can spill to
disk.
**API sketch** (signatures only):
```java
@PublicAPIClass(maturity = ApiMaturityLevel.EVOLVING)
public class HoodieJavaTableReader {
public HoodieJavaTableReader(HoodieEngineContext context, String basePath,
TypedProperties props)
public HoodieJavaTableReader(HoodieEngineContext context,
HoodieTableMetaClient metaClient, TypedProperties props)
// latest snapshot: base files merged with log files on merge-on-read tables
public ClosableIterator<IndexedRecord> readSnapshot()
public ClosableIterator<IndexedRecord> readSnapshot(ReadOptions options)
// base files only
public ClosableIterator<IndexedRecord> readOptimized(ReadOptions options)
}
// public, built with a builder; applies to both reads
ReadOptions.builder()
.asOf("20261101093000000") // time travel; the file index
takes a query instant for either read
.columns(Arrays.asList("id", "ts")) // projection; default is the full
table schema
.partitions(Arrays.asList("2026/11/01")) // partition paths relative to the
base path; default is all
.build()
```
- Records are `IndexedRecord` in the requested schema. The caller closes each
returned iterator: it holds the open files and the planning state of its call,
and a failure before the iterator is returned releases them. The reader object
owns no open read resources between calls, so it is not `AutoCloseable`. A
reader built from a meta client reloads its own copy of the timeline for each
call and does not change the one passed in.
- Time travel uses Spark's file-index settings and
`TimelineUtils#validateTimestampAsOf`. The timestamp must be earlier than the
first incomplete commit that is not a clustering commit, when there is one.
When the active timeline has a completed clean, the timestamp must also not be
earlier than that clean's earliest commit to retain. If that clean records no
earliest commit to retain, the check falls back to the first completed write
instant of the active timeline, and this fallback can refuse an archived
timestamp even if its files remain. With non-blocking concurrency control, an
inflight writer therefore blocks time travel at or after its requested instant,
as it does on Spark.
**Incremental reads (second PR, after this is agreed).**
`readIncremental(IncrementalOptions)` by completion time, with Spark's
incremental defaults: the start is exclusive and the end inclusive
(`OPEN_CLOSED`, the range type of Spark's incremental relation), compaction
commits are not skipped (as `hoodie.datasource.read.incr.skip_compact` defaults
to `false`), and clustering commits are kept (Spark's incremental relation
keeps them; skipping them is disabled there because of a duplication problem).
The start is required: `"earliest"` must be given explicitly, and `null` is
rejected. The end is optional. I plan to use `IncrementalQueryAnalyzer` to
select instants by completion time, archived instants included, and the shared
file index for listing. Spark's incremental relation adds more on top:
affected-file collection from commit metadata, a check for missing files, and a
configurable fallback to a full table scan. The second PR will specify and test
that behavior against Spark rath
er than assume it.
Filtering log records by instant range before merging can give incomplete or
wrong results: for partial updates, for event-time ordering when an in-range
write with a lower ordering value displaces an earlier higher value, and
possibly for custom mergers that need earlier records. #18943 reports the
partial-update case, and #19005 proposes merging whole file groups before
filtering the output. The second PR will settle these semantics, and document
any case it does not handle.
**Limitations of the first version:** no key or predicate filters (the Avro
reader context has key-filter overloads, so that would be a follow-up); no CDC
output (CDC-enabled tables read as merged rows); tables with the LSM storage
layout (`hoodie.table.storage.layout=lsm_tree`) are rejected unless the Java
writer path for them is tested end to end in the same PR; tables without meta
fields can be read as snapshots, and incremental reads of them are refused.
**Questions.**
1. RFC or not? Related read API designs have been proposed as RFCs: RFC-64
(#7080) for query engine integrations and RFC-98 for the Spark DataSource V2
read path. I am happy to write this up as an RFC if you prefer; otherwise I
would open the first PR as a draft linked here.
2. Package: `org.apache.hudi.client`, next to `HoodieJavaWriteClient`, or under
`org.apache.hudi.core`? Master already has `org.apache.hudi.core.read`, where
`BaseHoodieTableFileIndex` lives, and the draft #19214 moves more reader code
under `org.apache.hudi.core`.
3. Name: this is the class the first post called `HoodieJavaReadClient`.
`HoodieJavaTableReader` keeps this general table reader apart from Spark's
index-oriented `SparkRDDReadClient` / `HoodieReadClient`, which also provide
key-based read-optimized reads. `HoodieJavaReadClient` would match
`HoodieJavaWriteClient` instead. Which would you prefer?
The write-side PRs (simple bucket index on the Java engine, delete partitions
and partition TTL, log compaction, and a few small fixes) are separate and do
not depend on this reader.
GitHub link:
https://github.com/apache/hudi/discussions/19798#discussioncomment-19545215
----
This is an automatically sent email for [email protected].
To unsubscribe, please send an email to: [email protected]
