voonhous commented on code in PR #19877:
URL: https://github.com/apache/hudi/pull/19877#discussion_r3976339041
##########
hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestCompactionCommand.java:
##########
@@ -146,6 +172,209 @@ public void testCompactionShow() throws IOException {
assertNotNull(result);
}
+ /**
+ * Test case of the compaction validation entry point of {@link SparkMain},
which the
+ * 'compaction validate' command reaches through a spark-submit of its own.
+ */
+ @Test
+ public void testSparkMainCompactValidate() throws Exception {
+ createPendingCompactions();
+ String outputPath = outputPath("validate");
+
+ SparkMain.doCompactValidate(jsc(), tablePath, PENDING_COMPACTION_INSTANT,
outputPath, 2);
+
+ List<ValidationOpResult> results = readOperationResults(outputPath);
+ assertEquals(operationsOf(PENDING_COMPACTION_INSTANT).size(),
results.size());
+ assertTrue(results.stream().allMatch(ValidationOpResult::isSuccess),
results.toString());
+ assertEquals(fileIdsOf(PENDING_COMPACTION_INSTANT),
+ results.stream().map(result ->
result.getOperation().getFileId()).collect(Collectors.toSet()));
+ }
+
+ @Test
+ public void testSparkMainCompactValidateReportsMissingLogFile() throws
Exception {
+ createPendingCompactions();
+ HoodieCompactionOperation broken =
operationsOf(PENDING_COMPACTION_INSTANT).get(0);
+ // a log file the plan reads is gone, so that operation can no longer be
compacted
+ Files.delete(Paths.get(tablePath, broken.getPartitionPath(),
broken.getDeltaFilePaths().get(0)));
+ String outputPath = outputPath("validate-broken");
+
+ SparkMain.doCompactValidate(jsc(), tablePath, PENDING_COMPACTION_INSTANT,
outputPath, 2);
+
+ List<ValidationOpResult> results = readOperationResults(outputPath);
+ assertEquals(operationsOf(PENDING_COMPACTION_INSTANT).size(),
results.size());
+ List<ValidationOpResult> failed = results.stream().filter(result ->
!result.isSuccess()).collect(Collectors.toList());
+ assertEquals(1, failed.size(), results.toString());
+ assertEquals(broken.getFileId(), failed.get(0).getOperation().getFileId());
+ assertTrue(failed.get(0).getException().isPresent());
+ }
+
+ /**
+ * Repair runs the plan validation and returns an empty result: the log file
renaming it was
+ * written for is gone from the admin client, which leaves the plan
untouched and never reads
+ * the dry run flag, so there is only one arm to exercise. See
+ * https://github.com/apache/hudi/issues/19881.
+ */
+ @Test
+ public void testSparkMainCompactRepair() throws Exception {
+ createPendingCompactions();
+ Set<String> fileIdsBefore = fileIdsOf(PENDING_COMPACTION_INSTANT);
+ String outputPath = outputPath("repair");
+
+ SparkMain.doCompactRepair(jsc(), tablePath, PENDING_COMPACTION_INSTANT,
outputPath, 2, false);
+
+ assertTrue(readOperationResults(outputPath).isEmpty());
+
assertTrue(pendingCompactionInstants().contains(PENDING_COMPACTION_INSTANT));
+ assertEquals(fileIdsBefore, fileIdsOf(PENDING_COMPACTION_INSTANT));
+ }
+
+ /**
+ * Unscheduling a plan takes the requested compaction instant off the
timeline, unless this is a
+ * dry run. The other pending plans are left alone either way. Skip
validation is held at false:
+ * the admin client takes the flag but never reads it, so toggling it
repeats the same run.
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testSparkMainCompactUnschedulePlan(boolean dryRun) throws
Exception {
+ createPendingCompactions();
+ Set<String> pendingBefore = pendingCompactionInstants();
+ String outputPath = outputPath("unschedule-" + dryRun);
+
+ SparkMain.doCompactUnschedule(jsc(), tablePath,
PENDING_COMPACTION_INSTANT, outputPath, 2, false, dryRun);
Review Comment:
Refreshed: unschedule is described over `dryRun` alone, and the counts read
26 executions across seven classes.
##########
hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestCompactionCommand.java:
##########
@@ -146,6 +172,209 @@ public void testCompactionShow() throws IOException {
assertNotNull(result);
}
+ /**
+ * Test case of the compaction validation entry point of {@link SparkMain},
which the
+ * 'compaction validate' command reaches through a spark-submit of its own.
+ */
+ @Test
+ public void testSparkMainCompactValidate() throws Exception {
+ createPendingCompactions();
+ String outputPath = outputPath("validate");
+
+ SparkMain.doCompactValidate(jsc(), tablePath, PENDING_COMPACTION_INSTANT,
outputPath, 2);
+
+ List<ValidationOpResult> results = readOperationResults(outputPath);
+ assertEquals(operationsOf(PENDING_COMPACTION_INSTANT).size(),
results.size());
+ assertTrue(results.stream().allMatch(ValidationOpResult::isSuccess),
results.toString());
+ assertEquals(fileIdsOf(PENDING_COMPACTION_INSTANT),
+ results.stream().map(result ->
result.getOperation().getFileId()).collect(Collectors.toSet()));
+ }
+
+ @Test
+ public void testSparkMainCompactValidateReportsMissingLogFile() throws
Exception {
+ createPendingCompactions();
+ HoodieCompactionOperation broken =
operationsOf(PENDING_COMPACTION_INSTANT).get(0);
+ // a log file the plan reads is gone, so that operation can no longer be
compacted
+ Files.delete(Paths.get(tablePath, broken.getPartitionPath(),
broken.getDeltaFilePaths().get(0)));
+ String outputPath = outputPath("validate-broken");
+
+ SparkMain.doCompactValidate(jsc(), tablePath, PENDING_COMPACTION_INSTANT,
outputPath, 2);
+
+ List<ValidationOpResult> results = readOperationResults(outputPath);
+ assertEquals(operationsOf(PENDING_COMPACTION_INSTANT).size(),
results.size());
+ List<ValidationOpResult> failed = results.stream().filter(result ->
!result.isSuccess()).collect(Collectors.toList());
+ assertEquals(1, failed.size(), results.toString());
+ assertEquals(broken.getFileId(), failed.get(0).getOperation().getFileId());
+ assertTrue(failed.get(0).getException().isPresent());
+ }
+
+ /**
+ * Repair runs the plan validation and returns an empty result: the log file
renaming it was
+ * written for is gone from the admin client, which leaves the plan
untouched and never reads
+ * the dry run flag, so there is only one arm to exercise. See
+ * https://github.com/apache/hudi/issues/19881.
+ */
+ @Test
+ public void testSparkMainCompactRepair() throws Exception {
+ createPendingCompactions();
+ Set<String> fileIdsBefore = fileIdsOf(PENDING_COMPACTION_INSTANT);
+ String outputPath = outputPath("repair");
+
+ SparkMain.doCompactRepair(jsc(), tablePath, PENDING_COMPACTION_INSTANT,
outputPath, 2, false);
+
+ assertTrue(readOperationResults(outputPath).isEmpty());
+
assertTrue(pendingCompactionInstants().contains(PENDING_COMPACTION_INSTANT));
+ assertEquals(fileIdsBefore, fileIdsOf(PENDING_COMPACTION_INSTANT));
+ }
+
+ /**
+ * Unscheduling a plan takes the requested compaction instant off the
timeline, unless this is a
+ * dry run. The other pending plans are left alone either way. Skip
validation is held at false:
+ * the admin client takes the flag but never reads it, so toggling it
repeats the same run.
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testSparkMainCompactUnschedulePlan(boolean dryRun) throws
Exception {
+ createPendingCompactions();
+ Set<String> pendingBefore = pendingCompactionInstants();
+ String outputPath = outputPath("unschedule-" + dryRun);
+
+ SparkMain.doCompactUnschedule(jsc(), tablePath,
PENDING_COMPACTION_INSTANT, outputPath, 2, false, dryRun);
+
+ assertTrue(readOperationResults(outputPath).isEmpty());
+ Set<String> pendingAfter = pendingCompactionInstants();
+ if (dryRun) {
+ assertEquals(pendingBefore, pendingAfter);
+ } else {
+ assertFalse(pendingAfter.contains(PENDING_COMPACTION_INSTANT),
pendingAfter.toString());
+ pendingBefore.remove(PENDING_COMPACTION_INSTANT);
+ assertEquals(pendingBefore, pendingAfter);
+ }
+ }
+
+ /**
+ * Unscheduling a single file group rewrites the plan without it, unless
this is a dry run. Skip
+ * validation is held at false: the admin client takes the flag but never
reads it, so toggling
+ * it repeats the same run.
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testSparkMainCompactUnscheduleFile(boolean dryRun) throws
Exception {
+ Map<HoodieFileGroupId, Pair<String, HoodieCompactionOperation>>
pendingOperations = createPendingCompactions();
+ HoodieFileGroupId unscheduled = pendingOperations.entrySet().stream()
+ .filter(entry ->
entry.getValue().getKey().equals(PENDING_COMPACTION_INSTANT))
+ .map(Map.Entry::getKey).findFirst().get();
+ String outputPath = outputPath("unschedule-file-" + dryRun);
+
+ SparkMain.doCompactUnscheduleFile(jsc(), tablePath,
unscheduled.getFileId(), unscheduled.getPartitionPath(),
+ outputPath, 2, false, dryRun);
+
+ assertTrue(readOperationResults(outputPath).isEmpty());
+ // the plan itself stays pending either way, only its operations change.
Only the target file
+ // group is asserted: the admin client currently drops the other
operations of the same
+ // partition as well (https://github.com/apache/hudi/issues/19881); assert
they survive once fixed.
+
assertTrue(pendingCompactionInstants().contains(PENDING_COMPACTION_INSTANT));
+ if (dryRun) {
+
assertTrue(fileIdsOf(PENDING_COMPACTION_INSTANT).contains(unscheduled.getFileId()));
+ } else {
+
assertFalse(fileIdsOf(PENDING_COMPACTION_INSTANT).contains(unscheduled.getFileId()));
Review Comment:
Done: the non-dry-run arm now asserts the whole surviving set, empty today
because of #19881, with the comment saying the fix must flip it to "the other
operation survives". Done in a3c0e32b9cba.
--
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]
