lokeshj1703 opened a new pull request, #19816: URL: https://github.com/apache/hudi/pull/19816
### Describe the issue this Pull Request addresses In the `StreamSync` write path only the `writeStatusRDD` is cached; the **input records RDD** that feeds the write is not. For index-based operations (`UPSERT`/`INSERT`) that records RDD is read more than once *inside* the write: 1. the index tag lookup (eager inside `writeClient.upsert`/`insert`), and 2. the write itself (materialization of the tagged records). Because only the tagged records are persisted, the whole upstream DAG is recomputed on each read. When the transformer chain contains `EmbeddingTransformer` (RFC-102, already on `master`) — which POSTs each chunk to a remote embedding service — every chunk is **re-embedded on each read**, doubling embedding cost and latency. `BULK_INSERT` avoids this (no index tag) and is the current workaround, but `UPSERT`/incremental still double-embed. ### Summary and Changelog Cache the input records RDD once in `StreamSync.writeToSink`, before the write, so both reads reuse the materialized records — mirroring how `writeStatusRDD` is already cached. - **No user config.** Caching is automatic, gated only on a transformer being configured (`transformer.isPresent()`) — that is when recomputing the input is expensive. - **Storage level** `MEMORY_AND_DISK_SER`, so it spills to disk under memory pressure rather than forcing an OOM. Overhead is negligible relative to executor memory (~150 MB for a 1 GB input batch). - **Timing.** `instantTime` is generated at `startCommit` (before the write, before the first action), so the cache is in place before the first read. - **Release by tag (no retained reference).** The RDD is named `hoodie-input-records-<targetBasePath>-<instantTime>` via `rdd().setName(...)`; the tag is unique per `(table, instant)` so it cannot collide with another table or a retried instant in the context-wide persistent-RDD registry. In the `finally` of `writeToSinkAndDoMetaSync` it is released by looking it up by tag through `SparkContext.getPersistentRDDs()` and unpersisting the match — no reference retained, runs on both the success and failure paths, unknown tag is a no-op. - **Why RDD-level, not `Dataset.persist()`.** `Dataset.persist()` registers in the SQL CacheManager keyed by the logical plan, and `df.rdd()` returns a fresh RDD each call, so a name set there would never appear in `getPersistentRDDs()`. We cache and name the `JavaRDD` that actually feeds `writeClient.upsert`/`insert`. Changes: - `hudi-utilities/.../streamer/StreamSync.java` — cache+name the input records RDD before the write (gated on `transformer.isPresent()`); release-by-tag in the `finally` of `writeToSinkAndDoMetaSync`; two `@VisibleForTesting static` helpers `inputRecordsCacheName(basePath, instantTime)` and `unpersistCachedInputRecords(jsc, name)`. - `hudi-utilities/.../transform/embedding/TestEmbeddingTransformer.java` — count the total number of inputs the stub embedding server is asked to embed; two new tests. - `hudi-utilities/.../streamer/TestStreamSyncInputRecordsCache.java` (new) — tag uniqueness and release-by-tag. Tests (all offline, deterministic): - `testCachedRecordsRddEmbedsEachInputOnceAcrossTwoActions` — cache the records RDD, then run `count()` then `collect()` (simulating the index-tag lookup + the write). The backend embeds each of the 6 inputs exactly once (N=6). - `testUncachedRecordsRddEmbedsEachInputTwiceAcrossTwoActions` — control: the same two actions on the un-cached dataset embed every input again on the second action (2N=12), reproducing the double-embedding this fixes and proving the N assertion is not vacuous. - `TestStreamSyncInputRecordsCache` — the tag is unique per `(base path, instant)`; the tagged RDD is found and released by tag via `getPersistentRDDs()` with no retained reference; releasing one write's tag leaves another write's cache intact; an unknown tag is a no-op. Build / checkstyle / test (JDK 11, `-Dscala-2.12 -Dspark3.5`): - `mvn install -pl hudi-utilities -am -DskipTests` — passes. - `mvn checkstyle:check -pl hudi-utilities` — passes. - `mvn test -pl hudi-utilities -Punit-tests -Dtest=TestEmbeddingTransformer,TestStreamSyncInputRecordsCache` — 7 tests, 0 failures. ### Impact A performance / cost fix; correctness is preserved. When a transformer is configured, the input records RDD is cached (`MEMORY_AND_DISK_SER`) for the duration of one write and released immediately after, eliminating the second embedding pass on `UPSERT`/`INSERT`. No new config and no public API change. When no transformer is configured the path is unchanged (no caching is added). ### Risk Level low — scoped to `StreamSync.writeToSink` / `writeToSinkAndDoMetaSync`, gated on `transformer.isPresent()`, uses a storage level that spills rather than OOMs, and releases the cache by tag on both the success and failure paths. Verified with the red/green test pair above (2N control vs. N cached). ### Documentation Update none — no new config or user-facing surface; caching is automatic and internal. ### Contributor's checklist - [x] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute) - [x] Enough context is provided in the sections above - [x] Adequate tests were added if applicable --- **Base / stacking note:** this fix depends only on the unstructured-ingestion feature (`EmbeddingTransformer` and the `StreamSync` transformer path), which is **already merged to `master`** (RFC-102). It therefore targets `master` directly as a clean, self-contained increment rather than stacking on an open feature PR. The remaining open unstructured PRs (e.g. #19677, #19678) only *harden* `EmbeddingTransformer` and this change does not depend on them. Ported from internal onehouseinc/hudi-internal#2322 (adapted to upstream: the release-by-tag `finally` sits alongside `master`'s existing `writeStatusRDD` caching in the reworked pre-commit orchestration of `writeToSinkAndDoMetaSync`). 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
