hudi-agent commented on code in PR #20336:
URL: https://github.com/apache/hudi/pull/20336#discussion_r4274331477
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/ErrorTableUtils.java:
##########
@@ -79,25 +85,61 @@ 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})
+ * @param stash single-column dataset of pre-transform {@code
_corrupt_record} values
+ * @return {@code transformed} with the {@code _corrupt_record} column
restored
+ */
+ public static Dataset<Row> restoreCorruptRecordColumn(
+ SparkSession sparkSession, Dataset<Row> transformed, Dataset<Row> stash)
{
+ try {
+ JavaRDD<Row> zipped = transformed.javaRDD()
+ .zip(stash.javaRDD())
+ .map(pair -> {
+ Row dataRow = pair._1();
+ Row stashRow = pair._2();
+ int size = dataRow.size();
+ Object[] fields = new Object[size + 1];
+ for (int i = 0; i < size; i++) {
+ fields[i] = dataRow.get(i);
+ }
+ fields[size] = stashRow.isNullAt(0) ? null : stashRow.get(0);
+ return RowFactory.create(fields);
+ });
+ StructType schema = transformed.schema()
+ .add(ERROR_TABLE_CURRUPT_RECORD_COL_NAME, DataTypes.StringType,
true);
+ Dataset<Row> result = sparkSession.createDataFrame(zipped, schema);
+ // Force eager evaluation so zip failures (partition/element count
mismatches)
+ // are caught here instead of surfacing downstream. Without this, the
entire
+ // zip/map/createDataFrame chain is lazy and the catch block never
triggers.
+ result.count();
+ return result;
+ } catch (Exception e) {
+ LOG.warn("Failed to restore {} column after transformer dropped it "
+ + "(row count or partitioning changed): {}",
+ ERROR_TABLE_CURRUPT_RECORD_COL_NAME, e.getMessage());
+ return addNullValueErrorTableCorruptRecordColumn(transformed);
Review Comment:
🤖 If an earlier transformer marked rows as corrupt and a later one filters
rows and drops the column (for example a SQL transformer with a WHERE and an
explicit projection), the zip fails and this re-adds the column as null, so
`processErrorEvents` sends those corrupt rows to the main table with only a
WARN; before this change the pipeline failed loudly. Could we rethrow here (or
fall back to null only when the stash has no non-null values) instead of
quietly writing error records as good data?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/functional/TestErrorTableAwareChainedTransformer.java:
##########
@@ -76,17 +78,90 @@ public void testForErrorTableConfig() {
}
@Test
Review Comment:
🤖 nit: could you drop the internal `ENG-41958` ticket id from this comment?
It won't mean anything to OSS readers, and the rest of the comment already
explains the regression.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/ErrorTableAwareChainedTransformer.java:
##########
@@ -51,12 +56,38 @@ public ErrorTableAwareChainedTransformer(List<Transformer>
transformers) {
public Dataset<Row> apply(JavaSparkContext jsc, SparkSession sparkSession,
Dataset<Row> rowDataset,
TypedProperties properties) {
Dataset<Row> dataset = rowDataset;
+ // _corrupt_record can only hold values if the source supplied it or a
transformer kept it;
+ // while it is known to be all-null, re-adding it as null is lossless and
needs no stash.
+ boolean corruptRecordMayHaveValues =
ErrorTableUtils.isErrorTableCorruptRecordColumnPresent(dataset);
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 (corruptRecordMayHaveValues) {
+ // Eagerly checkpoint the input so the stash and the transformer read
the same
+ // materialized rows. Caching only the stash is not enough: the
transformer would
+ // recompute the upstream lineage independently, and if that lineage
has
+ // non-deterministic row ordering (e.g. shuffle/repartition) the
positional zip
+ // silently misaligns values. The checkpoint also keeps the restored
dataset
+ // aligned when downstream actions recompute it, so it is left for
Spark's
+ // ContextCleaner to release rather than unpersisted here.
+ dataset = dataset.localCheckpoint(true);
Review Comment:
🤖 Because the source path already strips `_corrupt_record`, any transformer
that just keeps the column (e.g. `SELECT * FROM <SRC>`) flips
`corruptRecordMayHaveValues` to true even when every value is null, so each
later transformer in the chain now runs an eager `localCheckpoint` of the whole
batch. Those checkpoint blocks live only on executors, so losing an executor
(spot or dynamic allocation) can fail the sync, which affects chains that
passed fine before this PR; could the checkpoint be skipped when the column is
still all null, or the stash and checkpoint be created only after we see the
transformer dropped the column?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/ErrorTableUtils.java:
##########
@@ -79,25 +85,61 @@ 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
Review Comment:
🤖 I don't think a misaligned zip is "no worse than" the null fallback: after
an `orderBy`/`repartition` with the same per-partition counts, good rows get
another row's non-null `_corrupt_record` and are moved to the error table (so
they disappear from the target table), while the truly corrupt rows are
upserted; would it make sense to fail here, or to carry a row id through rather
than relying on position?
<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]
