MoR support for the standalone Java client (new HoodieJavaReadClient)

[ravjot28]

Motivation

The Java engine (hudi-java-client) today ships only HoodieJavaWriteClient and HoodieJavaTableServiceClient. There is no read client, and MERGE_ON_READ tables can't be completed end-to-end from a plain Java application - no compaction execution and no MoR reads. Spark and Flink both have full MoR support; embedded Java users have none.
I have a working implementation on my fork (full CI matrix green) and would like feedback on scope and API shape before opening PRs.

What's implemented

1. New HoodieJavaReadClient (Java engine):

• Snapshot, read-optimized, time-travel, and completion-time-based incremental reads, for both MERGE_ON_READ and COPY_ON_WRITE tables.
• One HoodieFileGroupReader per file slice (base file merged with log files), exposed as a lazily-chained ClosableIterator → bounded memory.
• Fresh meta-client snapshot per call → safe under NBCC with inflight writers; inflight instants are invisible to snapshot/incremental reads.
• Documented v1 limitations: no archived-timeline incremental ranges, populate.meta.fields=false unsupported for completion-time filtering, no InternalSchema evolution reconciliation, no CDC.

2. MoR write-side completion: delta commit + compaction action executor wiring so Java writers run compaction end-to-end.

3. Bucket index for the Java engine: JavaBucketIndexPartitioner, bulk-insert partitioner, index factory + layout-config wiring.

4. BaseAvroPayload decode fix (hudi-common): getRecord(schema) now resolves by name from the true writer schema for projection schemas (fixes a pre-existing bug where merges against persisted data silently nulled data fields). Evolution-shaped schemas (renames/adds) keep the legacy positional decode the Hive/Presto MoR readers depend on; unresolvable fields with no default fail loudly with AvroTypeException. This touches the shared merge path for all engines → planned as a separate, first PR with risk called out.

Testing

• ~2,900 lines of new tests: MoR snapshot/incremental reads, NBCC concurrent reader/writer, payload schema-mismatch regressions, bucket-index partitioner.
• Full fork CI matrix green (all engine jobs), plus local runs of the Hive MoR schema-evolution functional tests and hudi-hadoop-mr realtime suites.

Proposed plan

1. File a type:feature issue (+ sub-issues) and open three PRs:

• fix(common): resolve payload decode by name for projection schemas
• feat(client): add HoodieJavaReadClient with MoR support
• feat(client): support bucket index for the Java engine

3. Land the common fix first (the read-client tests depend on it).

Questions for the community

• Issue + PRs sufficient, or is an RFC preferred given the new public read API?
• Any concerns with the iterator-based API surface (caller-owned iterators)?
• Are the v1 limitations acceptable for a first iteration?

Code: master...ravjot28:hudi:master

[vinothchandar]

Thanks for starting this @ravjot28 . Wonder if we can do this based on the new 1.x RecordMerger APIs.

[ravjot28]

@vinothchandar, thanks for suggesting the Hudi 1.x RecordMerger APIs. I assessed that direction and plan to make it the primary approach for the standalone Java MOR work.

The proposed reader already uses the shared HoodieFileGroupReader, and Java write deduplication already uses the shared buffered merge infrastructure. The follow-up work will make native Avro records, the built-in event-time/commit-time merge modes, and configured custom HoodieRecordMerger implementations the main examples and acceptance paths.

As an initial check, I restored the original payload implementation in an isolated checkout. Fifteen focused native-record cases passed across the three merge modes, with three mode-inapplicable cases skipped. These include assertions that the configured custom merger is actually invoked during snapshot reading and compaction, and that built-in partial updates preserve untouched values across compaction. This is focused feasibility evidence, not a claim of complete parity or production readiness.

My plan is to:

1. Add permanent Java-client integration coverage for custom RecordMerger configuration and invocation, with custom semantics that differ from the built-in modes. Assert the persisted merge mode and strategy so tests cannot silently exercise a default instead.
2. Extend the native-record coverage through writes, reads, duplicate reduction, ordering ties and out-of-order changes, deletes/resurrection, partial updates, schema evolution, compaction/log compaction, recovery, and spill cleanup. Distinguish null/default-value updates from partial-schema merging through partialMerge, and keep existing regression tests.
3. Update the Java examples and documentation around the modern APIs, including supported configurations, resource ownership, and remaining limitations.
4. Assess the shared payload fix in #19872 separately, rather than treating it as an unconditional prerequisite for Java MOR. The original code still reproduces a dropped-update regression for legacy payload input with a metadata-declaring schema. I will preserve that regression and evaluate a narrower fix that avoids changing the default serialization format; I am not proposing to remove legacy compatibility coverage.
5. Rework the remaining PR boundaries around these results and validate each resulting branch before submission. The intended feature split remains Java SIMPLE bucket-index/write routing, standalone reader and lifecycle improvements, and companion documentation. The Kafka test-readiness change remains independent.

Only #19872 is open so far. Its 16 focused tests passed locally at preparation. The earlier successful fork CI run covers the combined implementation and is not CI evidence for the revised PR series. I will keep the new API scope and any RFC requirement explicit for maintainer review.

Thank you also to Shiyan Xu and Nadine Farah for your guidance on the Java-client improvements and contribution process.

A correction to the original discussion and dev-list email: ordinary Java MOR delta writes and regular compaction already existed. This work adds the standalone reader and addresses specific lifecycle gaps. The current local suite covers internal-schema reconciliation and CDC-enabled writes through the common extractor; there is no standalone CDC-output API. Archived incremental ranges and LSM reads remain outside this iteration. Merge readers open one file group at a time and can spill, while planning memory still grows with the selected file groups.

I also plan to contribute further Java-client improvements, including the streaming work discussed earlier, to close more applicable gaps with other engines. This matters for teams that want Hudi embedded in their applications without the additional infrastructure cost and operational work of running a Spark deployment.