wombatu-kun commented on code in PR #18816:
URL: https://github.com/apache/hudi/pull/18816#discussion_r3956514257
##########
hudi-cli/src/main/java/org/apache/hudi/cli/commands/ArchivedCommitsCommand.java:
##########
@@ -107,9 +123,182 @@ public String showArchivedCommits(
throws IOException {
System.out.println("===============> Showing only " + limit + " archived
commits <===============");
HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
- StoragePath archivePath = folder != null && !folder.isEmpty()
- ? new StoragePath(metaClient.getMetaPath(), folder)
- : new StoragePath(metaClient.getArchivePath(), ".commits_.archive*");
+ List<Comparable[]> allStats;
+ if (folder != null && !folder.isEmpty()) {
+ allStats = readCommitStatsFromLegacyArchive(metaClient, new
StoragePath(metaClient.getMetaPath(), folder));
+ } else if (isLegacyArchive(metaClient)) {
+ allStats = readCommitStatsFromLegacyArchive(
+ metaClient, new StoragePath(metaClient.getArchivePath(),
".commits_.archive*"));
+ } else {
+ allStats =
readCommitStatsFromArchivedTimeline(newArchivedTimeline(metaClient),
sortByField, limit);
+ }
+ TableHeader header = new
TableHeader().addTableHeaderField("action").addTableHeaderField("instant")
+
.addTableHeaderField("partition").addTableHeaderField("file_id").addTableHeaderField("prev_instant")
+
.addTableHeaderField("num_writes").addTableHeaderField("num_inserts").addTableHeaderField("num_deletes")
+
.addTableHeaderField("num_update_writes").addTableHeaderField("total_log_files")
+
.addTableHeaderField("total_log_blocks").addTableHeaderField("total_corrupt_log_blocks")
+
.addTableHeaderField("total_rollback_blocks").addTableHeaderField("total_log_records")
+
.addTableHeaderField("total_updated_records_compacted").addTableHeaderField("total_write_bytes")
+ .addTableHeaderField("total_write_errors");
+
+ return HoodiePrintHelper.print(header, new HashMap<>(), sortByField,
descending, limit, headerOnly, allStats);
+ }
+
+ @ShellMethod(key = "show archived commits", value = "Read commits from
archived files and show details")
+ public String showCommits(
+ @ShellOption(value = {"--skipMetadata"}, help = "Skip displaying commit
metadata",
+ defaultValue = "true") boolean skipMetadata,
+ @ShellOption(value = {"--limit"}, help = "Limit commits", defaultValue =
"10") final Integer limit,
+ @ShellOption(value = {"--sortBy"}, help = "Sorting Field", defaultValue
= "") final String sortByField,
+ @ShellOption(value = {"--desc"}, help = "Ordering", defaultValue =
"false") final boolean descending,
+ @ShellOption(value = {"--headeronly"}, help = "Print Header Only",
+ defaultValue = "false") final boolean headerOnly) {
+
+ System.out.println("===============> Showing only " + limit + " archived
commits <===============");
+ HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
+ List<Comparable[]> allCommits = readArchivedCommits(
+ newArchivedTimeline(metaClient), skipMetadata,
isLegacyArchive(metaClient), sortByField, limit);
+
+ TableHeader header = new
TableHeader().addTableHeaderField("CommitTime").addTableHeaderField("CommitType");
+
+ if (!skipMetadata) {
+ header = header.addTableHeaderField("CommitDetails");
+ }
+
+ return HoodiePrintHelper.print(header, new HashMap<>(), sortByField,
descending, limit, headerOnly, allCommits);
+ }
+
+ /**
+ * Renders the completed archived instants as rows, loading metadata only
for the rows that
+ * can reach the output.
+ * <p>
+ * Without a sort field the printer keeps the timeline order and cuts at the
limit, so only
+ * the leading instants can be shown, and an archive that has grown for
years holds far more
+ * payload than those few rows need. With a sort field, or no limit, every
row takes part
+ * and everything has to be loaded.
+ */
+ @VisibleForTesting
+ static List<Comparable[]> readArchivedCommits(HoodieArchivedTimeline
archivedTimeline, boolean skipMetadata,
+ boolean legacyArchive, String
sortByField, int limit) {
+ List<HoodieInstant> completed = archivedTimeline.getInstants().stream()
+ .filter(HoodieInstant::isCompleted)
+ .collect(Collectors.toList());
+ List<HoodieInstant> shown = boundedByLimit(completed, sortByField, limit);
+ if (!skipMetadata) {
+ loadInstantDetails(archivedTimeline, shown, shown.size() <
completed.size());
+ }
+ return shown.stream()
+ .map(instant -> readArchivedCommit(archivedTimeline, instant,
skipMetadata, legacyArchive))
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Returns the leading instants that the printer can render, or all of them
when a sort field
+ * or a non-positive limit makes every row a candidate.
+ */
+ private static List<HoodieInstant> boundedByLimit(List<HoodieInstant>
instants, String sortByField, int limit) {
+ if (!sortByField.isEmpty() || limit <= 0 || limit >= instants.size()) {
+ return instants;
+ }
+ return instants.subList(0, limit);
+ }
+
+ /**
+ * Loads the details of the given instants, through the closed time range
they span when they
+ * are a strict subset of the archive, so that the archive files outside the
range are never
Review Comment:
loadInstantDetails also serves pre-v8 tables, and there ArchivedTimelineV1
passes no log-file filter, so its loader still globs and scans every archive
file - the claim that files outside the range are never opened holds only on
the LSM timeline. Could that sentence be narrowed to the v8+ layout?
##########
hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestCleansCommand.java:
##########
@@ -179,12 +179,10 @@ public void testShowCleanPartitions() {
// There should be two partition path
List<Comparable[]> rows = new ArrayList<>();
- rows.add(new Comparable[]
{HoodieTestCommitMetadataGenerator.DEFAULT_SECOND_PARTITION_PATH,
- HoodieCleaningPolicy.KEEP_LATEST_COMMITS, "1", "0"});
- rows.add(new Comparable[]
{HoodieTestCommitMetadataGenerator.DEFAULT_THIRD_PARTITION_PATH,
- HoodieCleaningPolicy.KEEP_LATEST_COMMITS, "0", "0"});
rows.add(new Comparable[]
{HoodieTestCommitMetadataGenerator.DEFAULT_FIRST_PARTITION_PATH,
HoodieCleaningPolicy.KEEP_LATEST_COMMITS, "1", "0"});
+ rows.add(new Comparable[]
{HoodieTestCommitMetadataGenerator.DEFAULT_SECOND_PARTITION_PATH,
Review Comment:
testShowCleanPartitions pins a new row order that is neither lexicographic
nor one the test builds - CleansCommand renders the clean metadata's partition
map straight from entrySet and the printer does not sort without --sortBy. Sort
those rows the way readWriteStatRows now does, or compare order-insensitively
here?
##########
hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestRepairsCommand.java:
##########
@@ -256,49 +279,220 @@ public void testRemoveCorruptedPendingCleanAction()
throws IOException {
// Create four requested files
for (int i = 100; i < 104; i++) {
String timestamp = String.valueOf(i);
- // Write corrupted requested Clean File
+ // Write an empty requested Clean File
HoodieTestCommitMetadataGenerator.createEmptyCleanRequestedFile(tablePath,
timestamp, conf);
}
+ // A plan cut short mid-write, so that its bytes stop inside the Avro
header
+ FileCreateUtils.createRequestedCleanFile(metaClient, "104",
validCleanerPlan());
+ truncateInHalf(metaClient, requestedCleanPath(metaClient, "104"));
+
+ // A plan whose writer was killed between opening and closing the Avro
container, which leaves a
+ // complete header and no record behind
+ try (DataFileWriter<HoodieCleanerPlan> writer =
+ new DataFileWriter<>(new
SpecificDatumWriter<>(HoodieCleanerPlan.class))) {
+ writer.create(HoodieCleanerPlan.getClassSchema(),
+ metaClient.getStorage().create(requestedCleanPath(metaClient,
"105"), true));
+ }
+
+ // A plan that decodes, which the command has to leave in place
+ FileCreateUtils.createRequestedCleanFile(metaClient, "106",
validCleanerPlan());
+
// reload meta client
metaClient = HoodieTableMetaClient.reload(metaClient);
- // first, there are four instants
- assertEquals(4,
metaClient.getActiveTimeline().filterInflightsAndRequested().countInstants());
+ // first, there are seven pending instants
+ assertEquals(7,
metaClient.getActiveTimeline().filterInflightsAndRequested().countInstants());
- Object result = shell.evaluate(() -> "repair corrupted clean files");
- assertTrue(ShellEvaluationResultUtil.isSuccess(result));
+ Object cleanResult = shell.evaluate(() -> "repair corrupted clean files");
+ assertTrue(ShellEvaluationResultUtil.isSuccess(cleanResult));
// reload meta client
metaClient = HoodieTableMetaClient.reload(metaClient);
- assertEquals(0,
metaClient.getActiveTimeline().filterInflightsAndRequested().countInstants());
+ // the empty, the truncated and the record-less plans are gone and the
readable one is untouched
+ List<HoodieInstant> remaining =
+
metaClient.getActiveTimeline().filterInflightsAndRequested().getInstants();
+ assertEquals(1, remaining.size());
+ assertEquals("106", remaining.get(0).requestedTime());
+ }
+
+ /**
+ * A transient read failure on a valid pending clean plan must not be taken
for corruption:
+ * the timeline serde wraps any exception raised while it streams the plan
file in the same
+ * "unable to read commit metadata" IOException that an empty or truncated
file raises.
+ */
+ @Test
+ public void testRemoveCorruptedPendingCleanActionKeepsPlanOnReadFailure()
throws IOException {
+ HoodieCLI.conf = storageConf();
+ HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
+ FileCreateUtils.createRequestedCleanFile(metaClient, "100",
validCleanerPlan());
+ StoragePath planPath = requestedCleanPath(metaClient, "100");
+
+ HoodieTableMetaClient timingOutClient = HoodieTableMetaClient.builder()
+ .setStorage(new TimingOutStorage(fs,
planPath)).setBasePath(tablePath).build();
+ HoodieIOException thrown = assertThrows(HoodieIOException.class,
+ () ->
RepairsCommand.removeCorruptedPendingCleanAction(timingOutClient));
+ assertInstanceOf(SocketTimeoutException.class, thrown.getCause());
+ assertTrue(metaClient.getStorage().exists(planPath));
+
+ // the same plan read through a healthy storage is left alone as well
+
RepairsCommand.removeCorruptedPendingCleanAction(HoodieTableMetaClient.reload(metaClient));
+ assertEquals(1,
HoodieTableMetaClient.reload(metaClient).getActiveTimeline()
+ .filterInflightsAndRequested().countInstants());
+ }
+
+ /**
+ * A clean that reached inflight keeps its plan in the requested file, so a
corrupt plan has to
+ * take both files with it. The timeline collapses the two states into the
inflight instant
+ * alone, so removing only the instant it listed would leave the corrupt
plan behind for the
+ * next clean to fail on, and take a second run of this command to clear.
+ */
+ @Test
+ public void testRemoveCorruptedPendingCleanActionRemovesInflightAndItsPlan()
throws IOException {
+ HoodieCLI.conf = storageConf();
+ HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
+
+ // a clean that was scheduled and started, whose plan was then cut short
+ FileCreateUtils.createRequestedCleanFile(metaClient, "100",
validCleanerPlan());
+ // the inflight file a clean leaves behind carries no plan of its own
+ FileCreateUtils.createInflightCleanFile(metaClient, "100", null, true);
+ truncateInHalf(metaClient, requestedCleanPath(metaClient, "100"));
+
+ StoragePath inflightPath = new StoragePath(metaClient.getTimelinePath(),
+
metaClient.getInstantFileNameGenerator().makeInflightCleanerFileName("100"));
+ assertTrue(metaClient.getStorage().exists(inflightPath));
+
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ List<HoodieInstant> pending =
metaClient.getActiveTimeline().filterInflightsAndRequested().getInstants();
+ assertEquals(1, pending.size());
+ assertTrue(pending.get(0).isInflight());
+
+ RepairsCommand.removeCorruptedPendingCleanAction(metaClient);
+
+ // one pass takes the whole action, not just the file the timeline listed
+ assertFalse(metaClient.getStorage().exists(inflightPath));
+ assertFalse(metaClient.getStorage().exists(requestedCleanPath(metaClient,
"100")));
+ assertEquals(0,
HoodieTableMetaClient.reload(metaClient).getActiveTimeline()
+ .filterInflightsAndRequested().countInstants());
+ }
+
+ /**
+ * Corrupt bytes do not always reach the decoder as an Avro failure: a plan
that decodes with no
+ * version leaves the migrator to unbox a null, and a length that survives
as far as Avro's own
+ * ceiling raises an {@code UnsupportedOperationException}. Such an instant
has to be judged
+ * corrupt like any other, and the instants behind it still repaired, rather
than the failure
+ * escaping and abandoning the rest of the timeline.
+ */
+ @Test
+ public void
testRemoveCorruptedPendingCleanActionRepairsPastAnUndecodablePlan() throws
IOException {
+ HoodieCLI.conf = storageConf();
+ HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
+
+ HoodieCleanerPlan versionless = validCleanerPlan();
+ versionless.setVersion(null);
+ writeCleanerPlan(metaClient, "100", versionless);
+
+ // an ordinary corruption behind it, which is only reached if the first
one does not escape
+ HoodieTestCommitMetadataGenerator.createEmptyCleanRequestedFile(tablePath,
"101", HoodieCLI.conf);
+ // and a readable plan the command has to leave in place
+ FileCreateUtils.createRequestedCleanFile(metaClient, "102",
validCleanerPlan());
+
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ assertEquals(3,
metaClient.getActiveTimeline().filterInflightsAndRequested().countInstants());
+
+ RepairsCommand.removeCorruptedPendingCleanAction(metaClient);
+
+ List<HoodieInstant> remaining = HoodieTableMetaClient.reload(metaClient)
+ .getActiveTimeline().filterInflightsAndRequested().getInstants();
+ assertEquals(1, remaining.size());
+ assertEquals("102", remaining.get(0).requestedTime());
+ }
+
+ /**
+ * Cuts a file's bytes in half in place. The file is overwritten rather than
deleted and
+ * rewritten, because both the emptiness check and the file name generator
resolve the file
+ * from the instant itself.
+ */
+ private static void truncateInHalf(HoodieTableMetaClient metaClient,
StoragePath path) throws IOException {
+ byte[] bytes;
+ try (InputStream in = metaClient.getStorage().open(path)) {
+ bytes = FileIOUtils.readAsByteArray(in);
+ }
+ try (OutputStream out = metaClient.getStorage().create(path, true)) {
+ out.write(bytes, 0, bytes.length / 2);
+ }
+ }
+
+ /**
+ * Writes a clean plan straight into the requested file of an instant, so
that a plan the
+ * timeline's own writer would not produce still reaches the command.
+ */
+ private static void writeCleanerPlan(HoodieTableMetaClient metaClient,
String instantTime,
+ HoodieCleanerPlan plan) throws
IOException {
+ try (DataFileWriter<HoodieCleanerPlan> writer =
+ new DataFileWriter<>(new
SpecificDatumWriter<>(HoodieCleanerPlan.class))) {
+ writer.create(HoodieCleanerPlan.getClassSchema(),
+ metaClient.getStorage().create(requestedCleanPath(metaClient,
instantTime), true));
+ writer.append(plan);
+ }
+ }
+
+ /**
+ * The path of the requested file of a clean instant.
+ */
+ private static StoragePath requestedCleanPath(HoodieTableMetaClient
metaClient, String instantTime) {
+ return new StoragePath(metaClient.getTimelinePath(),
+
metaClient.getInstantFileNameGenerator().makeRequestedCleanerFileName(instantTime));
+ }
+
+ /**
+ * A clean plan that decodes into the latest plan version.
+ */
+ private static HoodieCleanerPlan validCleanerPlan() {
+ return HoodieCleanerPlan.newBuilder()
+ .setEarliestInstantToRetain(HoodieActionInstant.newBuilder()
+ .setAction(HoodieTimeline.COMMIT_ACTION).setTimestamp("001")
+ .setState(HoodieInstant.State.COMPLETED.name()).build())
+
.setFilesToBeDeletedPerPartition(Collections.singletonMap("partition1",
Collections.singletonList("file1")))
+
.setFilePathsToBeDeletedPerPartition(Collections.singletonMap("partition1",
+
Collections.singletonList(HoodieCleanFileInfo.newBuilder().setFilePath("file1").build())))
+ .setLastCompletedCommitTimestamp("002")
+ .setPolicy(HoodieCleaningPolicy.KEEP_LATEST_COMMITS.name())
+ .setVersion(TimelineLayoutVersion.CURR_VERSION)
Review Comment:
validCleanerPlan sets the plan's version from
TimelineLayoutVersion.CURR_VERSION, but that field is the clean plan schema
version CleanPlanMigrator dispatches on, and the two are both 2 only by
coincidence. Could it take CleanPlanV2MigrationHandler.VERSION, the way
addClean in TestArchivedCommitsCommand does?
--
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]
