hudi-agent commented on code in PR #18914:
URL: https://github.com/apache/hudi/pull/18914#discussion_r3661824832
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/ErrorTableAwareChainedTransformer.java:
##########
@@ -54,9 +59,29 @@ public Dataset<Row> apply(JavaSparkContext jsc, SparkSession
sparkSession, Datas
dataset =
ErrorTableUtils.addNullValueErrorTableCorruptRecordColumn(dataset);
for (TransformerInfo transformerInfo : transformers) {
Transformer transformer = transformerInfo.getTransformer();
+
+ // Stash _corrupt_record values before the transformer can drop them
+ Dataset<Row> corruptRecordStash = null;
+ if (ErrorTableUtils.isErrorTableCorruptRecordColumnPresent(dataset)) {
+ corruptRecordStash = dataset.select(new
Column(ERROR_TABLE_CURRUPT_RECORD_COL_NAME));
+ corruptRecordStash.cache();
Review Comment:
🤖 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?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
