This is an automated email from the ASF dual-hosted git repository.
wombatu-kun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 42913d08459c fix(utilities): Re-add _corrupt_record when a transformer
drops it in ErrorTableAwareChainedTransformer (#20336)
42913d08459c is described below
commit 42913d08459c53edad0aea674234358bed6be9d3
Author: Raghuramchadalawada
<[email protected]>
AuthorDate: Thu Oct 22 18:44:07 2026 -0700
fix(utilities): Re-add _corrupt_record when a transformer drops it in
ErrorTableAwareChainedTransformer (#20336)
* fix(utilities): Re-add _corrupt_record when a transformer drops it in
ErrorTableAwareChainedTransformer
Custom transformers that project columns drop _corrupt_record because they
are unaware of the error-table contract, and
ErrorTableAwareChainedTransformer
failed the pipeline with HoodieValidationException after any such
transformer.
Re-add the column as null when a transformer drops it while it holds no
values. It is known to be all null until the source supplies it or a
transformer keeps it; otherwise one Spark job checks the transformer input.
If the dropped column held values, still fail, since those error records
cannot be realigned with the transformer output and re-adding null would
write them to the target table.
Closes #18915
* Simplify ErrorTableAwareChainedTransformer loop and extract
dropped-column validation
* Simplify corrupt record flag handling and extract test helpers in
ErrorTableAwareChainedTransformer
---
.../hudi/utilities/streamer/ErrorTableUtils.java | 14 +-----
.../ErrorTableAwareChainedTransformer.java | 40 ++++++++++++++--
.../TestErrorTableAwareChainedTransformer.java | 53 ++++++++++++++++++++++
3 files changed, 89 insertions(+), 18 deletions(-)
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/ErrorTableUtils.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/ErrorTableUtils.java
index 1904346f593a..fe2e44a2440f 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/ErrorTableUtils.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/ErrorTableUtils.java
@@ -27,7 +27,6 @@ import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.config.HoodieErrorTableConfig;
import org.apache.hudi.exception.HoodieException;
-import org.apache.hudi.exception.HoodieValidationException;
import org.apache.hudi.utilities.ingestion.HoodieIngestionMetrics;
import org.apache.hadoop.fs.FileSystem;
@@ -79,17 +78,6 @@ public final class ErrorTableUtils {
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));
@@ -97,7 +85,7 @@ public final class ErrorTableUtils {
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));
}
}
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/ErrorTableAwareChainedTransformer.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/ErrorTableAwareChainedTransformer.java
index a7d2979e90f0..29dc421de34f 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/ErrorTableAwareChainedTransformer.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/ErrorTableAwareChainedTransformer.java
@@ -22,9 +22,11 @@ package org.apache.hudi.utilities.transform;
import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieValidationException;
import org.apache.hudi.utilities.streamer.ErrorTableUtils;
import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.sql.Column;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
@@ -33,10 +35,14 @@ import org.apache.spark.sql.types.StructType;
import java.util.List;
import java.util.function.Supplier;
+import static
org.apache.hudi.utilities.streamer.BaseErrorTableWriter.ERROR_TABLE_CURRUPT_RECORD_COL_NAME;
+
/**
* A {@link Transformer} to chain other {@link Transformer}s and apply
sequentially.
- * Adds errorTableCorruptRecordColumn at the beginning of transformations and
validates
- * if that column is not dropped in any of the transformations.
+ * Adds errorTableCorruptRecordColumn at the beginning of transformations and
re-adds it as null
+ * if a transformer drops it (e.g. a custom column-projecting transformer).
Dropping the column is
+ * rejected only if it held values, since those error records cannot be
realigned with the
+ * transformer output and would otherwise be written to the target table.
*/
public class ErrorTableAwareChainedTransformer extends ChainedTransformer {
public ErrorTableAwareChainedTransformer(List<String>
configuredTransformers, Supplier<Option<HoodieSchema>> sourceSchemaSupplier) {
@@ -51,16 +57,40 @@ public class ErrorTableAwareChainedTransformer extends
ChainedTransformer {
public Dataset<Row> apply(JavaSparkContext jsc, SparkSession sparkSession,
Dataset<Row> rowDataset,
TypedProperties properties) {
Dataset<Row> dataset = rowDataset;
+ // Until the source supplies the column or a transformer keeps it, it is
known to be all null.
+ boolean corruptRecordMayHaveValues =
ErrorTableUtils.isErrorTableCorruptRecordColumnPresent(dataset);
dataset =
ErrorTableUtils.addNullValueErrorTableCorruptRecordColumn(dataset);
for (TransformerInfo transformerInfo : transformers) {
Transformer transformer = transformerInfo.getTransformer();
- dataset = transformer.apply(jsc, sparkSession, dataset,
transformerInfo.getProperties(properties, transformers));
- // validate in every stage to ensure ErrorRecordColumn not dropped by
one of the transformer and added by next transformer.
- ErrorTableUtils.validate(dataset);
+ Dataset<Row> input = dataset;
+ dataset = transformer.apply(jsc, sparkSession, input,
transformerInfo.getProperties(properties, transformers));
+ if (ErrorTableUtils.isErrorTableCorruptRecordColumnPresent(dataset)) {
+ corruptRecordMayHaveValues = true;
+ } else {
+ if (corruptRecordMayHaveValues) {
+ validateNoErrorRecordsDropped(input, transformer);
+ }
+ // The column held no values, so re-adding it as null loses nothing.
+ dataset =
ErrorTableUtils.addNullValueErrorTableCorruptRecordColumn(dataset);
+ corruptRecordMayHaveValues = false;
+ }
}
return dataset;
}
+ /**
+ * Fails if {@code transformer} dropped a {@code _corrupt_record} column
that held values in its input. Those error
+ * records cannot be realigned with the transformer output, and re-adding
the column as null would write them to
+ * the target table.
+ */
+ private static void validateNoErrorRecordsDropped(Dataset<Row> input,
Transformer transformer) {
+ if (!input.filter(new
Column(ERROR_TABLE_CURRUPT_RECORD_COL_NAME).isNotNull()).isEmpty()) {
+ throw new HoodieValidationException(String.format("Transformer %s
dropped column %s, which holds error records "
+ + "marked by the source or an earlier transformer. Keep the column
in its output, or apply it before the "
+ + "transformers that mark error records.",
transformer.getClass().getName(), ERROR_TABLE_CURRUPT_RECORD_COL_NAME));
+ }
+ }
+
@Override
public StructType transformedSchema(JavaSparkContext jsc, SparkSession
sparkSession, StructType incomingStruct, TypedProperties properties) {
return super.transformedSchema(jsc, sparkSession, incomingStruct,
properties);
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/functional/TestErrorTableAwareChainedTransformer.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/functional/TestErrorTableAwareChainedTransformer.java
index b2e9414417ba..c528b6836155 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/functional/TestErrorTableAwareChainedTransformer.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/functional/TestErrorTableAwareChainedTransformer.java
@@ -52,6 +52,7 @@ import static org.apache.spark.sql.types.DataTypes.StringType;
import static org.apache.spark.sql.types.DataTypes.createStructField;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
@Tag("functional")
public class TestErrorTableAwareChainedTransformer extends
SparkClientFunctionalTestHarness {
@@ -89,6 +90,58 @@ public class TestErrorTableAwareChainedTransformer extends
SparkClientFunctional
assertThrows(HoodieValidationException.class, () ->
transformer.apply(jsc(), spark(), original, properties));
}
+ @Test
+ void testColumnProjectionReAddsNullCorruptRecordColumn() {
+ // A custom transformer projecting columns drops _corrupt_record while it
is still all null.
+ Dataset<Row> transformed = applyWithErrorTableEnabled(getTestDataset(),
getErrorRecordColumnDropTransformer());
+ assertCorruptRecordColumnAllNull(transformed);
+ }
+
+ @Test
+ void testColumnProjectionAfterTransformerKeepingNullCorruptRecordColumn() {
+ // t1 keeps _corrupt_record without marking rows, so t2 may drop it.
+ Transformer t1 = (jsc, sparkSession, dataset, props) ->
dataset.withColumn("foo", dataset.col("foo").cast(IntegerType));
+ Transformer t2 = getErrorRecordColumnDropTransformer();
+ Dataset<Row> transformed = applyWithErrorTableEnabled(getTestDataset(),
t1, t2);
+ assertCorruptRecordColumnAllNull(transformed);
+ }
+
+ @Test
+ void testErrorRecordsMarkedAfterColumnProjection() {
+ // t1 drops _corrupt_record, t2 marks error records in the re-added column.
+ Transformer t1 = getErrorRecordColumnDropTransformer();
+ Transformer t2 = getErrorEventHandlerTransformer();
+ Dataset<Row> transformed = applyWithErrorTableEnabled(getTestDataset(),
t1, t2);
+
+ assertArrayEquals(new String[]{"foo",
ERROR_TABLE_CURRUPT_RECORD_COL_NAME}, transformed.columns());
+ assertEquals(2, transformed.filter(new
Column(ERROR_TABLE_CURRUPT_RECORD_COL_NAME).equalTo("true")).count());
+ }
+
+ @Test
+ void testColumnProjectionWithSourceSuppliedCorruptRecordColumn() {
+ // The source supplies _corrupt_record: dropping it is allowed only while
it holds no values.
+ Transformer columnFilter = getErrorRecordColumnDropTransformer();
+
+ Dataset<Row> withoutErrors =
getTestDataset().withColumn(ERROR_TABLE_CURRUPT_RECORD_COL_NAME,
functions.lit(null).cast(StringType));
+ assertCorruptRecordColumnAllNull(applyWithErrorTableEnabled(withoutErrors,
columnFilter));
+
+ Dataset<Row> withErrors =
getTestDataset().withColumn(ERROR_TABLE_CURRUPT_RECORD_COL_NAME,
+ functions.when(new Column("foo").equalTo("100"), functions.lit("{bad
json")).otherwise(functions.lit(null).cast(StringType)));
+ HoodieValidationException e =
assertThrows(HoodieValidationException.class, () ->
applyWithErrorTableEnabled(withErrors, columnFilter));
+ assertTrue(e.getMessage().contains(ERROR_TABLE_CURRUPT_RECORD_COL_NAME));
+ }
+
+ private Dataset<Row> applyWithErrorTableEnabled(Dataset<Row> input,
Transformer... transformers) {
+ TypedProperties properties = new TypedProperties();
+ properties.setProperty(ERROR_TABLE_ENABLED.key(), "true");
+ return new
ErrorTableAwareChainedTransformer(Arrays.asList(transformers)).apply(jsc(),
spark(), input, properties);
+ }
+
+ private static void assertCorruptRecordColumnAllNull(Dataset<Row>
transformed) {
+ assertArrayEquals(new String[]{"foo",
ERROR_TABLE_CURRUPT_RECORD_COL_NAME}, transformed.columns());
+ assertEquals(2, transformed.filter(new
Column(ERROR_TABLE_CURRUPT_RECORD_COL_NAME).isNull()).count());
+ }
+
private Dataset<Row> getTestDataset() {
StructType schema = DataTypes.createStructType(
new StructField[]{
