hudi-agent commented on code in PR #19801:
URL: https://github.com/apache/hudi/pull/19801#discussion_r3900949060
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkStreamingMetadataWriteHandler.java:
##########
@@ -42,26 +42,32 @@ public HoodieData<WriteStatus>
streamWriteToMetadataTable(HoodieTable table, Hoo
Option<HoodieTableMetadataWriter> metadataWriterOpt =
getMetadataWriter(instantTime, table);
ValidationUtils.checkState(metadataWriterOpt.isPresent(),
"Cannot instantiate metadata writer for the table of interest " +
table.getMetaClient().getBasePath());
- return streamWriteToMetadataTable(dataTableWriteStatuses,
metadataWriterOpt.get(), table, instantTime, coalesceDivisorForDataTableWrites);
+ return streamWriteToMetadataTable(dataTableWriteStatuses,
metadataWriterOpt.get(), table, instantTime,
+ coalesceDivisorForDataTableWrites);
}
private HoodieData<WriteStatus>
streamWriteToMetadataTable(HoodieData<WriteStatus> dataTableWriteStatuses,
HoodieTableMetadataWriter metadataWriter,
HoodieTable table,
String
instantTime,
int
coalesceDivisorForDataTableWrites) {
- HoodieData<WriteStatus> mdtWriteStatuses =
metadataWriter.streamWriteToMetadataPartitions(dataTableWriteStatuses,
instantTime);
- mdtWriteStatuses.persist("MEMORY_AND_DISK_SER", table.getContext(),
HoodieData.HoodieDataCacheKey.of(table.getMetaClient().getBasePath().toString(),
instantTime));
- HoodieData<WriteStatus> coalescedDataWriteStatuses;
- int coalesceParallelism = Math.max(1,
dataTableWriteStatuses.getNumPartitions() / coalesceDivisorForDataTableWrites);
- // lets coalesce to lesser number of spark tasks so that, when unioned
along with metadata table write status,
- // we only allocate very less number of tasks for data table write
statuses.
- // In fact, data table writes should have triggered in previous stage
before coalesce (partition by below forces the writes
- // to data table is triggered in previous stage and with the coalesced
stage)
- coalescedDataWriteStatuses =
HoodieJavaRDD.of(HoodieJavaRDD.getJavaRDD(dataTableWriteStatuses)
- .mapToPair((PairFunction<WriteStatus, String, WriteStatus>)
writeStatus -> new Tuple2(writeStatus.getStat().getPath(), writeStatus))
+ int coalesceParallelism = Math.max(1,
+ dataTableWriteStatuses.getNumPartitions() /
coalesceDivisorForDataTableWrites);
+
+ // Materialize data-table statuses behind a shuffle before deriving
metadata records. Spark's
+ // scheduler exposes only the successful attempt for each shuffle
partition, so failed or losing
+ // speculative attempts cannot contribute stale RLI/SI locations to the
metadata write.
+ HoodieData<WriteStatus> successfulDataWriteStatuses = HoodieJavaRDD.of(
Review Comment:
🤖 This won't compile as written — the local is declared
`successfulDataWriteStatuses` here, but it's referenced as
`committedDataWriteStatuses` on lines 68 and 71, which is never defined. Looks
like a rename that got left half-applied; could you reconcile the two names?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkStreamingMetadataWriteHandler.java:
##########
@@ -42,26 +42,32 @@ public HoodieData<WriteStatus>
streamWriteToMetadataTable(HoodieTable table, Hoo
Option<HoodieTableMetadataWriter> metadataWriterOpt =
getMetadataWriter(instantTime, table);
ValidationUtils.checkState(metadataWriterOpt.isPresent(),
"Cannot instantiate metadata writer for the table of interest " +
table.getMetaClient().getBasePath());
- return streamWriteToMetadataTable(dataTableWriteStatuses,
metadataWriterOpt.get(), table, instantTime, coalesceDivisorForDataTableWrites);
+ return streamWriteToMetadataTable(dataTableWriteStatuses,
metadataWriterOpt.get(), table, instantTime,
+ coalesceDivisorForDataTableWrites);
}
private HoodieData<WriteStatus>
streamWriteToMetadataTable(HoodieData<WriteStatus> dataTableWriteStatuses,
HoodieTableMetadataWriter metadataWriter,
HoodieTable table,
String
instantTime,
int
coalesceDivisorForDataTableWrites) {
- HoodieData<WriteStatus> mdtWriteStatuses =
metadataWriter.streamWriteToMetadataPartitions(dataTableWriteStatuses,
instantTime);
- mdtWriteStatuses.persist("MEMORY_AND_DISK_SER", table.getContext(),
HoodieData.HoodieDataCacheKey.of(table.getMetaClient().getBasePath().toString(),
instantTime));
- HoodieData<WriteStatus> coalescedDataWriteStatuses;
- int coalesceParallelism = Math.max(1,
dataTableWriteStatuses.getNumPartitions() / coalesceDivisorForDataTableWrites);
- // lets coalesce to lesser number of spark tasks so that, when unioned
along with metadata table write status,
- // we only allocate very less number of tasks for data table write
statuses.
- // In fact, data table writes should have triggered in previous stage
before coalesce (partition by below forces the writes
- // to data table is triggered in previous stage and with the coalesced
stage)
- coalescedDataWriteStatuses =
HoodieJavaRDD.of(HoodieJavaRDD.getJavaRDD(dataTableWriteStatuses)
- .mapToPair((PairFunction<WriteStatus, String, WriteStatus>)
writeStatus -> new Tuple2(writeStatus.getStat().getPath(), writeStatus))
+ int coalesceParallelism = Math.max(1,
+ dataTableWriteStatuses.getNumPartitions() /
coalesceDivisorForDataTableWrites);
+
+ // Materialize data-table statuses behind a shuffle before deriving
metadata records. Spark's
+ // scheduler exposes only the successful attempt for each shuffle
partition, so failed or losing
+ // speculative attempts cannot contribute stale RLI/SI locations to the
metadata write.
+ HoodieData<WriteStatus> successfulDataWriteStatuses = HoodieJavaRDD.of(
+ HoodieJavaRDD.getJavaRDD(dataTableWriteStatuses)
+ .mapToPair((PairFunction<WriteStatus, String, WriteStatus>)
writeStatus ->
+ new Tuple2<>(writeStatus.getStat().getPath(), writeStatus))
.partitionBy(new CoalescingPartitioner(coalesceParallelism))
.map((Function<Tuple2<String, WriteStatus>, WriteStatus>) entry ->
entry._2));
- return coalescedDataWriteStatuses.union(mdtWriteStatuses);
+
+ HoodieData<WriteStatus> metadataWriteStatuses =
+
metadataWriter.streamWriteToMetadataPartitions(committedDataWriteStatuses,
instantTime);
Review Comment:
🤖 Deriving metadata from the coalesced RDD ties RLI/SI generation to
`coalesceParallelism = max(1, numPartitions/coalesceDivisor)`. With the default
divisor of 5000 that's 1 partition for nearly all writes, so the whole
`writeStatus.flatMap(indexMapper)` derivation is funneled through a single
reducer, whereas before it ran at full data-write parallelism. For large writes
that looks like a serious throughput/OOM regression. @nsivabalan @yihua is the
metadata derivation intended to run at coalesced parallelism? Could we instead
persist the pre-coalesce statuses (so both branches consume the same committed
output) and coalesce only the union branch, keeping derivation parallelism
intact?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestSparkStreamingMetadataWriteHandler.java:
##########
@@ -60,27 +74,93 @@ void setUp() {
private static Stream<Arguments> coalesceDivisorTestArgs() {
return Arrays.stream(new Object[][] {
- {100, 20, 1000, true},
- {1, 1, 1000, true},
- {10000, 100, 5000, true},
- {10000, 100, 5000, true},
- {10000, 100, 20000, true},
- {10000, 100, 20000, true}
+ {100, 1000},
+ {1, 1},
+ {10000, 1},
+ {10000, 5000},
+ {10001, 5000},
+ {10000, 20000}
}).map(Arguments::of);
}
@ParameterizedTest
@MethodSource("coalesceDivisorTestArgs")
- public void testCoalesceDividentConfig(int numDataTableWriteStatuses, int
numMdtWriteStatus, int coalesceDividentForDataTableWrites) {
+ public void testCoalesceDividentConfig(int numDataTableWriteStatuses, int
coalesceDividentForDataTableWrites) {
HoodieData<WriteStatus> dataTableWriteStatus =
mockWriteStatuses(numDataTableWriteStatuses);
- HoodieData<WriteStatus> mdtWriteStatus =
mockWriteStatuses(numMdtWriteStatus);
- HoodieTableMetadataWriter mdtWriter =
mock(HoodieTableMetadataWriter.class);
- when(mdtWriter.streamWriteToMetadataPartitions(any(),
any())).thenReturn(mdtWriteStatus);
+ HoodieTableMetadataWriter mdtWriter =
metadataWriterReturningEmptyStatuses();
SparkStreamingMetadataWriteHandler metadataWriteHandler = new
MockSparkStreamingMetadataWriteHandler(mdtWriter);
HoodieData<WriteStatus> allWriteStatuses =
metadataWriteHandler.streamWriteToMetadataTable(mockHoodieTable,
dataTableWriteStatus, "00001",
coalesceDividentForDataTableWrites);
- assertEquals(Math.max(1, numDataTableWriteStatuses /
coalesceDividentForDataTableWrites) + numMdtWriteStatus,
allWriteStatuses.getNumPartitions());
+ assertEquals(Math.max(1, numDataTableWriteStatuses /
coalesceDividentForDataTableWrites),
+ allWriteStatuses.getNumPartitions());
+ verify(mdtWriter).streamWriteToMetadataPartitions(any(), any());
+ }
+
+ @Test
+ void testSparkCollectsOnlySuccessfulRetriedTaskAttempt() {
+ HoodieData<WriteStatus> retryingDataWriteStatus =
HoodieJavaRDD.of(jsc().parallelize(Arrays.asList(0, 1), 2)
+ .map(partition -> {
+ int attempt = TaskContext.get().attemptNumber();
+ if (partition == 0 && attempt == 0) {
+ throw new IllegalStateException("intentional first-attempt
failure");
+ }
+ return writeStatus("file-" + partition + "-attempt-" + attempt,
false);
+ }));
+ HoodieTableMetadataWriter mdtWriter =
mock(HoodieTableMetadataWriter.class);
+ AtomicReference<HoodieData<WriteStatus>> metadataInput = new
AtomicReference<>();
+ when(mdtWriter.streamWriteToMetadataPartitions(any(),
any())).thenAnswer(invocation -> {
+ metadataInput.set(invocation.getArgument(0));
+ return HoodieJavaRDD.of(jsc().emptyRDD());
+ });
+
+ List<WriteStatus> statuses = new
MockSparkStreamingMetadataWriteHandler(mdtWriter)
+ .streamWriteToMetadataTable(mockHoodieTable, retryingDataWriteStatus,
"00001", 1000)
+ .collectAsList();
+ List<String> paths = statuses.stream().map(status ->
status.getStat().getPath()).collect(java.util.stream.Collectors.toList());
+
+ List<String> metadataPaths = metadataInput.get().collectAsList().stream()
+ .map(status -> status.getStat().getPath())
+ .collect(java.util.stream.Collectors.toList());
+ assertTrue(paths.contains("file-0-attempt-1"), "Collected output must come
from the successful attempt");
+ assertTrue(paths.stream().noneMatch(path ->
path.equals("file-0-attempt-0")));
+ assertTrue(metadataPaths.contains("file-0-attempt-1"),
+ "Metadata generation must consume the successful attempt");
+ assertTrue(metadataPaths.stream().noneMatch(path ->
path.equals("file-0-attempt-0")));
Review Comment:
🤖 Since attempt 0 throws before returning a WriteStatus, `file-0-attempt-0`
is never emitted by any attempt, so the `noneMatch("file-0-attempt-0")` checks
hold trivially — this test would pass on master too and doesn't actually
exercise the stale-lineage path. The bug you describe is a losing/slower
speculative attempt that *does* emit output; could the test be shaped to
reproduce that so it genuinely guards the regression?
<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]
