hudi-agent commented on code in PR #18914:
URL: https://github.com/apache/hudi/pull/18914#discussion_r3633140640
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/ErrorTableUtils.java:
##########
@@ -79,25 +85,56 @@ public static
HoodieErrorTableConfig.ErrorWriteFailureStrategy getErrorWriteFail
return
HoodieErrorTableConfig.ErrorWriteFailureStrategy.valueOf(writeFailureStrategy);
}
- /**
- * validates for constraints on ErrorRecordColumn when ErrorTable enabled
configs are set.
- * @param dataset
- */
- public static void validate(Dataset<Row> dataset) {
- if (!isErrorTableCorruptRecordColumnPresent(dataset)) {
- throw new HoodieValidationException(String.format("Invalid condition,
columnName=%s "
- + "is not present in transformer " + "output schema",
ERROR_TABLE_CURRUPT_RECORD_COL_NAME));
- }
- }
-
public static Dataset<Row>
addNullValueErrorTableCorruptRecordColumn(Dataset<Row> dataset) {
if (!isErrorTableCorruptRecordColumnPresent(dataset)) {
dataset = dataset.withColumn(ERROR_TABLE_CURRUPT_RECORD_COL_NAME,
lit(null));
}
return dataset;
}
- private static boolean isErrorTableCorruptRecordColumnPresent(Dataset<Row>
dataset) {
+ public static boolean isErrorTableCorruptRecordColumnPresent(Dataset<Row>
dataset) {
return Arrays.stream(dataset.columns()).anyMatch(col ->
col.equals(ERROR_TABLE_CURRUPT_RECORD_COL_NAME));
}
+
+ /**
+ * Restores stashed {@code _corrupt_record} values onto a transformed
dataset via positional
+ * RDD zip. Works when the transformer only projects columns (row count and
partition layout
+ * unchanged). Falls back to {@link
#addNullValueErrorTableCorruptRecordColumn} with a WARN
+ * if the zip fails (e.g. because the transformer filtered or repartitioned
rows).
+ *
+ * <p><b>Limitation:</b> a transformer that reorders rows (e.g. {@code
orderBy}) without
+ * changing the count will produce a successful zip with misaligned values.
This is no worse
+ * 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})
Review Comment:
🤖 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?
<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]
