hudi-agent commented on code in PR #19839:
URL: https://github.com/apache/hudi/pull/19839#discussion_r4161260763
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamerErrorTableWriteFlow.java:
##########
@@ -19,75 +19,223 @@
package org.apache.hudi.utilities.deltastreamer;
+import org.apache.hudi.client.WriteStatus;
import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.util.collection.Tuple3;
import org.apache.hudi.hadoop.fs.HadoopFSUtils;
+import org.apache.hudi.utilities.streamer.ErrorEvent;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.hadoop.fs.Path;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.functions;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.StructType;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Predicate;
+import java.util.function.UnaryOperator;
import java.util.stream.Stream;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
class TestHoodieDeltaStreamerErrorTableWriteFlow extends
TestHoodieDeltaStreamerSchemaEvolutionBase {
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+ /**
+ * Bad records a source batch can carry, each built from valid generated
records by one column mutation. A case
+ * declares, per write operation, whether the streamer must route the record
to the error table or write it to the
+ * base table, and the reason it is routed there, so every case is exercised
by the same streamer sync. Cover a new
+ * failure mode by adding a case here rather than a new test, since each
streamer sync is expensive.
+ */
+ enum BadRecordCase {
+ // The original record key is kept in driver, so each quarantined record
stays attributable.
+ EMPTY_RECORD_KEY(df -> df.withColumn("driver",
functions.col("_row_key")).withColumn("_row_key", functions.lit("")),
+ operation -> true, ErrorEvent.ErrorReason.RECORD_CREATION),
+ // Only an upsert combines records by their ordering value; the other
operations write a null one through.
+ NULL_ORDERING_VALUE(df -> df.withColumn("timestamp",
functions.lit(null).cast(DataTypes.LongType)),
+ operation -> operation == WriteOperationType.UPSERT,
ErrorEvent.ErrorReason.RECORD_CREATION);
+
+ private final UnaryOperator<Dataset<Row>> mutation;
+ private final Predicate<WriteOperationType> routedToErrorTable;
+ private final ErrorEvent.ErrorReason errorReason;
+
+ BadRecordCase(UnaryOperator<Dataset<Row>> mutation,
Predicate<WriteOperationType> routedToErrorTable,
+ ErrorEvent.ErrorReason errorReason) {
+ this.mutation = mutation;
+ this.routedToErrorTable = routedToErrorTable;
+ this.errorReason = errorReason;
+ }
+ }
+
protected void testBase(Tuple3<Integer, Integer, Integer> sourceGenInfo)
throws Exception {
int totalRecords = sourceGenInfo.f0;
int errorRecords = sourceGenInfo.f1;
int numFiles = sourceGenInfo.f2;
boolean shouldCreateMultipleSourceFiles = numFiles > 1;
+ // Expected outcome, kept up to date as each batch is built: the base
table as record key to rider, and the records
+ // routed to the error table as (record, reason), where a record is named
by its key, or by driver when the key is empty.
+ Map<String, String> expectedRiders = new HashMap<>();
+ List<String> expectedErrorRecords = new ArrayList<>();
PARQUET_SOURCE_ROOT = basePath + "parquetFilesDfs" + testNum++;
+ List<Row> validRecords = Collections.emptyList();
+ StructType sourceSchema = null;
if (totalRecords > 0) {
if (shouldCreateMultipleSourceFiles) {
prepareParquetDFSMultiFiles(totalRecords - errorRecords,
PARQUET_SOURCE_ROOT, numFiles);
} else {
prepareParquetDFSFiles(totalRecords - errorRecords,
PARQUET_SOURCE_ROOT);
}
+ Dataset<Row> validRecordsDf =
sparkSession.read().parquet(PARQUET_SOURCE_ROOT);
+ sourceSchema = validRecordsDf.schema();
+ validRecords = validRecordsDf.orderBy("_row_key").collectAsList();
+ validRecords.forEach(row -> expectedRiders.put(row.getAs("_row_key"),
row.getAs("rider")));
- // Add error data to the source
+ // Add errorRecords records of each bad-record case to the source
if (errorRecords > 0) {
- String errorDataSourceRoot = basePath + "parquetErrorFilesDfs" +
testNum++;
- prepareParquetDFSFiles(errorRecords, errorDataSourceRoot);
- Dataset<Row> df = sparkSession.read().parquet(errorDataSourceRoot);
- df = df.withColumn("_row_key", functions.lit(""));
- // add error records to PARQUET_SOURCE_ROOT
- addParquetData(df, false);
+ for (BadRecordCase badRecordCase : BadRecordCase.values()) {
+ String errorDataSourceRoot = basePath + "parquetErrorFilesDfs" +
testNum++;
+ prepareParquetDFSFiles(errorRecords, errorDataSourceRoot);
+ addBadRecords(badRecordCase,
sparkSession.read().parquet(errorDataSourceRoot), expectedRiders,
expectedErrorRecords);
+ }
}
} else {
fs.mkdirs(new Path(PARQUET_SOURCE_ROOT));
}
tableName = "test_parquet_table" + testNum;
tableBasePath = basePath + tableName;
- this.deltaStreamer = new HoodieDeltaStreamer(getDeltaStreamerConfig(),
jsc);
+ HoodieDeltaStreamer.Config cfg = getDeltaStreamerConfig();
+ cfg.operation = writeOperationType;
+ this.deltaStreamer = new HoodieDeltaStreamer(cfg, jsc);
this.deltaStreamer.sync();
+ int expectedInstants = 1;
+ int lastSyncErrorRecords = expectedErrorRecords.size();
+
+ // An upsert also updates stored records, in a second sync: a valid update
must be applied, every bad record in the
+ // batch must reach the error table, and a null ordering value on a stored
record must leave that record unchanged.
+ // Without a guard at record creation, that null ordering value reaches
the record merger. Every check runs after
+ // the last sync, so a failure in the first one does not stop the second
from reaching the merger.
+ if (writeOperationType == WriteOperationType.UPSERT && errorRecords > 0) {
+ assertTrue(validRecords.size() >= (BadRecordCase.values().length + 1) *
errorRecords,
+ "the second sync updates errorRecords stored records per bad-record
case, plus as many valid updates");
+ int firstSyncErrorRecords = expectedErrorRecords.size();
+ List<Row> validUpdates = updatesOf(validRecords.subList(0,
errorRecords), sourceSchema);
+ addParquetData(validUpdates, sourceSchema);
+ validUpdates.forEach(row -> expectedRiders.put(row.getAs("_row_key"),
row.getAs("rider")));
+ BadRecordCase[] badRecordCases = BadRecordCase.values();
+ for (int i = 0; i < badRecordCases.length; i++) {
+ List<Row> storedRecords = validRecords.subList((i + 1) * errorRecords,
(i + 2) * errorRecords);
+ addBadRecords(badRecordCases[i],
sparkSession.createDataFrame(updatesOf(storedRecords, sourceSchema),
sourceSchema),
+ expectedRiders, expectedErrorRecords);
+ }
+ this.deltaStreamer = new HoodieDeltaStreamer(cfg, jsc);
+ this.deltaStreamer.sync();
+ expectedInstants++;
+ lastSyncErrorRecords = expectedErrorRecords.size() -
firstSyncErrorRecords;
+ }
// base table validation
Dataset<Row> baseDf =
sparkSession.read().format("hudi").load(tableBasePath);
- assertEquals(totalRecords - errorRecords, baseDf.count());
+ if (expectedRiders.isEmpty()) {
+ assertEquals(0, baseDf.count());
+ } else {
+ Map<String, String> actualRiders = new HashMap<>();
+ baseDf.select("_row_key", "rider").collectAsList().forEach(row ->
actualRiders.put(row.getString(0), row.getString(1)));
+ assertEquals(expectedRiders, actualRiders);
+ }
HoodieTableMetaClient metaClient = HoodieTableMetaClient.builder()
.setConf(HadoopFSUtils.getStorageConfWithCopy(jsc.hadoopConfiguration()))
.setBasePath(tableBasePath).build();
- assertEquals(1, metaClient.getActiveTimeline().getInstants().size());
+ assertEquals(expectedInstants,
metaClient.getActiveTimeline().getInstants().size());
// error table validation
if (withErrorTable) {
-
TestHoodieDeltaStreamerSchemaEvolutionExtensive.validateErrorTable(errorRecords,
this.writeErrorTableInParallelWithBaseTable);
+ Collections.sort(expectedErrorRecords);
+ List<Object> receivedErrorEvents = new ArrayList<>();
+ TestErrorTable.receivedErrorEvents.forEach(errorEvents ->
receivedErrorEvents.addAll(errorEvents.collect()));
+ assertEquals(expectedErrorRecords, errorRecordsOf(receivedErrorEvents));
+ // What the error table writer committed, not only what it was handed
+ List<Object> committed = new ArrayList<>();
+ TestErrorTable.commited.values().forEach(errors -> errors.ifPresent(rdd
-> committed.addAll(rdd.collect())));
+ if (this.writeErrorTableInParallelWithBaseTable) {
+ // The unified write commits write statuses, and each commit replaces
the previous one
Review Comment:
🤖 nit: `addParquetData` is now overloaded with unrelated signatures
(Dataset+boolean vs List+schema that just wraps into a DataFrame) — might be
worth a distinct name like `addParquetRows` to avoid confusing call sites.
<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]
