fix(utilities): Re-inject _corrupt_record in ErrorTableAwareChainedTransformer after each transformer by raghusrhyme · Pull Request #18914 · apache/hudi

raghusrhyme · 2026-06-04T17:06:44Z

Describe the issue this Pull Request addresses

Summary and Changelog

Custom JAR transformers that do column projection (e.g. ColumnFilter with mode=include) drop _corrupt_record since they are unaware of the error-table contract. ErrorTableAwareChainedTransformer called validate() after every transformer, causing HoodieValidationException for any transformer that projects the column away.

Replace validate() with addNullValueErrorTableCorruptRecordColumn() after each transformer.apply() so the column is re-injected if dropped
Remove the now-unused validate() method and HoodieValidationException import from ErrorTableUtils

Impact

No public API change. Behavior change: transformers that drop _corrupt_record no longer crash the pipeline — the column is silently re-added as null.

Risk Level

low — the fix is a one-line swap in the transformer chain loop, and addNullValueErrorTableCorruptRecordColumn() is already used at the start of the same method.

Documentation Update

none

Contributor's checklist

Read through contributor's guide
Enough context is provided in the sections above
Adequate tests were added if applicable

…former after each transformer Custom JAR transformers that do column projection (e.g. ColumnFilter with mode=include) drop _corrupt_record since they are unaware of the error-table contract. Replace the validate() call with addNullValueErrorTableCorruptRecordColumn() after each transformer.apply(), so any transformer — built-in or custom — can safely project columns without breaking the quarantine pipeline.

The validate() method is no longer called after replacing it with addNullValueErrorTableCorruptRecordColumn() in the previous commit. Remove the dead code and its HoodieValidationException import.

hudi-agent

🤖 This review was generated by an AI agent and may contain mistakes. Please verify any suggestions before applying.

Thanks for the contribution! This PR loosens ErrorTableAwareChainedTransformer so that custom column-projecting transformers no longer crash the chain by dropping _corrupt_record — instead the column is re-injected as null after each stage. One semantic concern worth confirming in the inline comment around silent loss of populated error data. Please take a look, and this should be ready for a Hudi committer or PMC member to take it from here. One minor naming inconsistency in the new tests — the lambda parameter in testTransformerPreservingCorruptRecordIsNoOp could use props to match the adjacent test and avoid visual confusion with the outer TypedProperties properties variable.

hudi-agent · 2026-06-07T05:06:44Z

-      ErrorTableUtils.validate(dataset);
+      // Re-inject _corrupt_record if the transformer dropped it (e.g. custom JAR transformers
+      // that do column projection like ColumnFilter with mode=include).
+      dataset = ErrorTableUtils.addNullValueErrorTableCorruptRecordColumn(dataset);


🤖 If an earlier transformer in the chain populates _corrupt_record with actual error rows and a later transformer drops it, this silently re-injects it as a null column — the previously captured error data is lost without warning. Previously validate() would have surfaced this as an exception. Is that an acceptable tradeoff for the column-projection use case, or worth at least logging a warn when the column was dropped after being non-null? @yihua

_{- AI-generated; verify before applying. React 👍/👎 to flag quality.}

Confirmed reachable in production: StreamSync applies the chain then calls processErrorEvents with CUSTOM_TRANSFORMER_FAILURE, and that extraction in SourceFormatAdapter keys off _corrupt_record being non-null. If an earlier transformer marks rows and a later one projects the column away, re-injecting as null here makes every row match the isNull filter in processErrorEvents, so the marked rows flow into the main write path - not just dropped from the error table but silently written to the target table. The new test testCorruptRecordReInjectedAfterTransformerDropsIt sets up exactly this populate-then-drop case (t1 marks, t2 drops) yet only asserts column presence and count, so it passes whether the marked data survives or not. Safer to extract the column before re-injecting, or at minimum WARN when a non-null column is dropped and assert the row outcome in that test.

Made changes, Testing them.

hudi-agent · 2026-06-07T05:06:44Z

+
+    Transformer keepAll = (jsc, sparkSession, dataset, properties) -> dataset.withColumn("foo",
+        dataset.col("foo").cast(IntegerType));
+    TypedProperties properties = new TypedProperties();


🤖 nit: the lambda parameter here is named properties, while the sibling test testCustomColumnProjectionPreservesCorruptRecord uses props for the same position — could you use props here too for consistency? It also avoids the visual confusion with TypedProperties properties declared two lines below.

_{- AI-generated; verify before applying. React 👍/👎 to flag quality.}

Addressing this one as well.

codecov-commenter · 2026-07-10T09:55:35Z

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 68.80%. Comparing base (cb14ca7) to head (fbcbd05).
⚠️ Report is 151 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #18914      +/-   ##
============================================
+ Coverage     68.77%   68.80%   +0.02%     
- Complexity    29073    29099      +26     
============================================
  Files          2517     2517              
  Lines        139807   139820      +13     
  Branches      17210    17205       -5     
============================================
+ Hits          96148    96197      +49     
+ Misses        35885    35851      -34     
+ Partials       7774     7772       -2

Flag	Coverage Δ
common-and-other-modules	`44.50% <0.00%> (+0.24%)`	⬆️
hadoop-mr-java-client	`44.76% <ø> (+0.07%)`	⬆️
spark-client-hadoop-common	`48.04% <ø> (-0.01%)`	⬇️
spark-java-tests	`49.32% <0.00%> (+0.02%)`	⬆️
spark-scala-tests	`45.22% <0.00%> (+<0.01%)`	⬆️
utilities	`37.28% <100.00%> (+<0.01%)`	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
...pache/hudi/utilities/streamer/ErrorTableUtils.java	`95.00% <ø> (+3.69%)`	⬆️
...s/transform/ErrorTableAwareChainedTransformer.java	`92.30% <100.00%> (ø)`

... and 29 files with indirect coverage changes

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

Previously re-injected _corrupt_record as null after each transformer, silently losing error data populated by earlier transformers. Stash values before each transformer and restore via positional RDD zip if the column was dropped. Falls back to null re-injection with a WARN if the transformer changed row count or partitioning. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

hudi-agent

⚠️ 🤖 This review was generated by an AI agent and may contain mistakes. Please verify any suggestions before applying.

Thanks for the follow-up! This revision replaces the null re-injection with a stash-and-restore approach: _corrupt_record values are cached before each transformer and re-attached via positional RDD zip if a transformer drops the column, preserving populated error markers. Both of my prior comments are addressed — the silent-data-loss concern is resolved by the new restore path, and the properties/props naming nit is fixed. One correctness concern remains on the new restoreCorruptRecordColumn: because the zip is built lazily, the try/catch fallback likely won't catch zip failures at the point it expects (see the inline comment). Please take a look at the inline comment, and this should be ready for a Hudi committer or PMC member to take it from here.

hudi-agent · 2026-07-22T19:19:34Z

+   * than the null-re-injection alternative, but callers should be aware.
+   *
+   * @param sparkSession the active Spark session
+   * @param transformed  the dataset after the transformer ran (missing {@code _corrupt_record})


🤖 The zip/map/createDataFrame chain here is all lazy — no Spark action runs inside this try, so the zip's failure (unequal partition counts throw IllegalArgumentException from getPartitions, unequal per-partition element counts throw SparkException from the iterator) is raised later at action time in the caller, not here. That means the catch + WARN fallback to addNullValueErrorTableCorruptRecordColumn won't trigger for the filter/repartition case it's meant to cover, and the pipeline crashes downstream instead. Could you force the alignment eagerly (e.g. compare transformed.rdd().getNumPartitions() and cached counts, or trigger the zip within the try) so the fallback actually kicks in?

_{⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.}

The zip/map/createDataFrame chain is entirely lazy — no Spark action runs inside the try block, so partition or element count mismatches surface downstream instead of hitting the catch + WARN fallback. Add result.count() to force eager evaluation. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

hudi-agent

⚠️ 🤖 This review was generated by an AI agent and may contain mistakes. Please verify any suggestions before applying.

Thanks for working on this! The PR replaces the hard validate() check with a stash-and-restore approach so column-projecting transformers no longer crash the error-table pipeline — the _corrupt_record values are stashed before each transformer and re-attached via a positional RDD zip if dropped. The direction is reasonable, but the correctness of the positional zip hinges on both branches seeing identical row ordering; one edge case worth double-checking is flagged inline. Please take a look at any inline comments, and this should be ready for a Hudi committer or PMC member to take it from here.

hudi-agent · 2026-07-28T00:16:18Z

+      Dataset<Row> corruptRecordStash = null;
+      if (ErrorTableUtils.isErrorTableCorruptRecordColumnPresent(dataset)) {
+        corruptRecordStash = dataset.select(new Column(ERROR_TABLE_CURRUPT_RECORD_COL_NAME));
+        corruptRecordStash.cache();


🤖 The stash is cache()d but never materialized with an action before transformer.apply runs, so both stash and transformed recompute the shared dataset lineage independently at zip time. If that upstream lineage is non-deterministic in row order (e.g. an earlier transformer did a shuffle/repartition, or the source doesn't guarantee stable ordering), the two recomputations can order rows differently within a partition — the zip still succeeds and result.count() still matches, but _corrupt_record values get attached to the wrong rows. Could you force materialization of the stash (e.g. a .count() after .cache()) — or better, restore via a monotonic join key rather than positional zip? @nsivabalan could you sanity-check the alignment assumption here?

_{⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.}

hudi-bot · 2026-07-28T01:25:53Z

CI report:

be2d9ba Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

@hudi-bot run azure re-run the last Azure build

raghusrhyme added 2 commits June 4, 2026 10:06

Remove unused validate() method from ErrorTableUtils

99653cd

The validate() method is no longer called after replacing it with addNullValueErrorTableCorruptRecordColumn() in the previous commit. Remove the dead code and its HoodieValidationException import.

raghusrhyme changed the title ~~[HUDI-XXXXX] Re-inject _corrupt_record in ErrorTableAwareChainedTransformer after each transformer~~ fix(utilities): Re-inject _corrupt_record in ErrorTableAwareChainedTransformer after each transformer Jun 4, 2026

github-actions Bot added the size:S PR with lines of changes in (10, 100] label Jun 4, 2026

raghusrhyme added 2 commits June 4, 2026 13:43

Update Javadoc to reflect re-injection behavior instead of validation

f069a12

Remove public modifier from JUnit 5 test methods

fbcbd05

hudi-agent reviewed Jun 7, 2026

View reviewed changes

voonhous mentioned this pull request Jul 16, 2026

[TRACKING] Open PR backlog triage: verdicts for open PRs (close / convert-to-issue / keep) #19306

Open

github-actions Bot added size:M PR with lines of changes in (100, 300] and removed size:S PR with lines of changes in (10, 100] labels Jul 17, 2026

hudi-agent reviewed Jul 22, 2026

View reviewed changes

hudi-agent reviewed Jul 28, 2026

View reviewed changes

Uh oh!

Conversation

raghusrhyme commented Jun 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Describe the issue this Pull Request addresses

Summary and Changelog

Impact

Risk Level

Documentation Update

Contributor's checklist

Uh oh!

hudi-agent left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

codecov-commenter commented Jul 10, 2026

Codecov Report

Uh oh!

hudi-agent left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hudi-agent left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hudi-bot commented Jul 28, 2026

CI report:

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

raghusrhyme commented Jun 4, 2026 •

edited

Loading