You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "lokeshj1703 (via GitHub)" <gi...@apache.org> on 2026/09/02 06:50:31 UTC
[PR] fix(utilities): cache StreamSync input records to avoid double-embedding [hudi]
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(utilities): cache StreamSync input records to avoid double-embedding [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19816:
URL: https://github.com/apache/hudi/pull/19816#issuecomment-5506832430
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e5a10ab9ca0ae596a54c02b14041d66b7270fb68",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16763",
"triggerID" : "e5a10ab9ca0ae596a54c02b14041d66b7270fb68",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* e5a10ab9ca0ae596a54c02b14041d66b7270fb68 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16763)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(utilities): cache StreamSync input records to avoid double-embedding [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19816:
URL: https://github.com/apache/hudi/pull/19816#issuecomment-5505780682
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e5a10ab9ca0ae596a54c02b14041d66b7270fb68",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16763",
"triggerID" : "e5a10ab9ca0ae596a54c02b14041d66b7270fb68",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* e5a10ab9ca0ae596a54c02b14041d66b7270fb68 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16763)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(utilities): cache StreamSync input records to avoid double-embedding [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19816:
URL: https://github.com/apache/hudi/pull/19816#issuecomment-5505715127
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e5a10ab9ca0ae596a54c02b14041d66b7270fb68",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "e5a10ab9ca0ae596a54c02b14041d66b7270fb68",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* e5a10ab9ca0ae596a54c02b14041d66b7270fb68 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(utilities): cache StreamSync input records to avoid double-embedding [hudi]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #19816:
URL: https://github.com/apache/hudi/pull/19816#issuecomment-5505948321
## [Codecov](https://app.codecov.io/gh/apache/hudi/pull/19816?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 78.30%. Comparing base ([`a2788ee`](https://app.codecov.io/gh/apache/hudi/commit/a2788eef922e2375432ead3fe50af7fe5f55faf7?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`e5a10ab`](https://app.codecov.io/gh/apache/hudi/commit/e5a10ab9ca0ae596a54c02b14041d66b7270fb68?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #19816 +/- ##
============================================
- Coverage 78.31% 78.30% -0.01%
Complexity 33885 33885
============================================
Files 2541 2541
Lines 141728 141728
Branches 17182 17182
============================================
- Hits 110988 110986 -2
- Misses 23032 23033 +1
- Partials 7708 7709 +1
```
| [Components](https://app.codecov.io/gh/apache/hudi/pull/19816/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [hudi-common](https://app.codecov.io/gh/apache/hudi/pull/19816/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.64% <ø> (ø)` | |
| [hudi-client](https://app.codecov.io/gh/apache/hudi/pull/19816/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.20% <ø> (-0.01%)` | :arrow_down: |
| [hudi-flink](https://app.codecov.io/gh/apache/hudi/pull/19816/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `85.65% <ø> (ø)` | |
| [hudi-spark-datasource](https://app.codecov.io/gh/apache/hudi/pull/19816/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.21% <ø> (ø)` | |
| [hudi-utilities](https://app.codecov.io/gh/apache/hudi/pull/19816/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `74.55% <ø> (ø)` | |
| [hudi-cli](https://app.codecov.io/gh/apache/hudi/pull/19816/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `15.06% <ø> (ø)` | |
| [hudi-hadoop](https://app.codecov.io/gh/apache/hudi/pull/19816/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.72% <ø> (ø)` | |
| [hudi-sync](https://app.codecov.io/gh/apache/hudi/pull/19816/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `75.54% <ø> (ø)` | |
| [hudi-io](https://app.codecov.io/gh/apache/hudi/pull/19816/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `79.98% <ø> (ø)` | |
| [hudi-timeline-service](https://app.codecov.io/gh/apache/hudi/pull/19816/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.44% <ø> (ø)` | |
| [hudi-cloud](https://app.codecov.io/gh/apache/hudi/pull/19816/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `65.81% <ø> (ø)` | |
| [hudi-kafka-connect](https://app.codecov.io/gh/apache/hudi/pull/19816/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `53.20% <ø> (ø)` | |
| [Flag](https://app.codecov.io/gh/apache/hudi/pull/19816/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [hadoop-mr-java-client](https://app.codecov.io/gh/apache/hudi/pull/19816/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `44.07% <ø> (-0.02%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files with missing lines](https://app.codecov.io/gh/apache/hudi/pull/19816?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...org/apache/hudi/utilities/streamer/StreamSync.java](https://app.codecov.io/gh/apache/hudi/pull/19816?src=pr&el=tree&filepath=hudi-utilities%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fhudi%2Futilities%2Fstreamer%2FStreamSync.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aHVkaS11dGlsaXRpZXMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdXRpbGl0aWVzL3N0cmVhbWVyL1N0cmVhbVN5bmMuamF2YQ==) | `77.77% <ø> (ø)` | |
... and [1 file with indirect coverage changes](https://app.codecov.io/gh/apache/hudi/pull/19816/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org