wombatu-kun commented on code in PR #18816:
URL: https://github.com/apache/hudi/pull/18816#discussion_r3947974553
##########
.github/workflows/bot.yml:
##########
@@ -143,7 +149,7 @@ jobs:
SPARK_PROFILE: ${{ matrix.sparkProfile }}
FLINK_PROFILE: ${{ matrix.flinkProfile }}
run:
- mvn clean install -T 2 -D"$SCALA_PROFILE" -D"$SPARK_PROFILE"
-D"$FLINK_PROFILE" -DskipTests=true -Phudi-platform-service $MVN_ARGS -am -pl
hudi-client/hudi-spark-client
+ mvn clean install -T 2 -D"$SCALA_PROFILE" -D"$SPARK_PROFILE"
-D"$FLINK_PROFILE" -DskipTests=true -Phudi-platform-service $MVN_ARGS -am -pl
hudi-client/hudi-spark-client,hudi-cli
Review Comment:
Adding hudi-cli here takes this job's -am reactor from 10 modules to 23,
since hudi-cli compile-depends on hudi-utilities-bundle, so the build grows
along with the tests - the job runs 42m53s on this head against 24-29m on the
three most recent master runs. Does the description's "the only new cost is the
tests" still hold, or is the catch-all job, which needs no build change at all,
worth another look?
##########
hudi-cli/src/main/java/org/apache/hudi/cli/commands/ArchivedCommitsCommand.java:
##########
@@ -107,9 +120,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
+ * opened and the payloads outside it never held.
+ */
+ private static void loadInstantDetails(HoodieArchivedTimeline
archivedTimeline, List<HoodieInstant> instants,
+ boolean subset) {
+ if (instants.isEmpty()) {
+ return;
+ }
+ if (subset) {
+ archivedTimeline.loadCompletedInstantDetailsInMemory(
+ instants.get(0).requestedTime(), instants.get(instants.size() -
1).requestedTime());
+ } else {
+ archivedTimeline.loadCompletedInstantDetailsInMemory();
+ }
+ }
+
+ /**
+ * Reads the write stats of the archived commit and delta commit instants
through the
+ * archived timeline, the LSM timeline that table version 8 and above are
written with.
+ * <p>
+ * Without a sort field the printer keeps the timeline order and cuts at the
limit, so the
+ * rows come from a leading run of the write instants. Each of them
contributes a row per
+ * write stat, so their details are loaded a limit-sized window of instants
at a time until
+ * the rows reach the limit, instead of materializing every archived
payload. With a sort
+ * field, or no limit, every row takes part and everything has to be loaded.
+ */
+ @VisibleForTesting
+ static List<Comparable[]>
readCommitStatsFromArchivedTimeline(HoodieArchivedTimeline archivedTimeline,
+ String
sortByField, int limit) {
+ List<HoodieInstant> writes = archivedTimeline.getInstants().stream()
+ .filter(HoodieInstant::isCompleted)
+ .filter(instant ->
HoodieTimeline.COMMIT_ACTION.equals(instant.getAction())
+ || HoodieTimeline.DELTA_COMMIT_ACTION.equals(instant.getAction()))
+ .collect(Collectors.toList());
+ if (!sortByField.isEmpty() || limit <= 0) {
+ archivedTimeline.loadCompletedInstantDetailsInMemory();
+ return writes.stream()
+ .flatMap(instant -> readWriteStatRows(archivedTimeline, instant))
+ .collect(Collectors.toList());
+ }
+ List<Comparable[]> rows = new ArrayList<>();
+ for (int from = 0; from < writes.size() && rows.size() < limit; from +=
limit) {
+ List<HoodieInstant> window = writes.subList(from, Math.min(from + limit,
writes.size()));
+ loadInstantDetails(archivedTimeline, window, window.size() <
writes.size());
+ window.forEach(instant -> readWriteStatRows(archivedTimeline,
instant).forEach(rows::add));
+ }
+ return rows;
+ }
+
+ /**
+ * Returns whether the table keeps its archived instants in the legacy log
format, the archive
+ * layout of the timeline layout version 1 that table versions before eight
are written with.
+ */
+ private static boolean isLegacyArchive(HoodieTableMetaClient metaClient) {
+ return
metaClient.getTimelineLayoutVersion().compareTo(TimelineLayoutVersion.LAYOUT_VERSION_2)
< 0;
+ }
+
+ /**
+ * Builds an archived timeline that bypasses the meta client cache, so that
an archival
+ * triggered earlier in the same CLI session is visible. The instance is
discarded with the
+ * command, so the instant details it loads need no explicit eviction.
+ */
+ private static HoodieArchivedTimeline
newArchivedTimeline(HoodieTableMetaClient metaClient) {
+ return metaClient.getArchivedTimeline(StringUtils.EMPTY_STRING, false);
+ }
+
+ private static Stream<Comparable[]> readWriteStatRows(HoodieArchivedTimeline
archivedTimeline, HoodieInstant instant) {
+ HoodieCommitMetadata metadata;
+ try {
+ metadata = archivedTimeline.readCommitMetadataToAvro(instant);
+ } catch (IOException e) {
+ throw new HoodieException("Failed to read the archived commit metadata
of instant " + instant, e);
+ }
+ if (metadata == null || metadata.getPartitionToWriteStats() == null) {
+ return Stream.empty();
+ }
+ final String action = instant.getAction();
+ final String instantTime = instant.requestedTime();
+ return sortByKey(metadata.getPartitionToWriteStats()).values().stream()
+ .flatMap(List::stream)
+ .map(writeStat -> new Comparable[] {action, instantTime,
writeStat.getPartitionPath(),
+ writeStat.getFileId(), writeStat.getPrevCommit(),
writeStat.getNumWrites(),
+ writeStat.getNumInserts(), writeStat.getNumDeletes(),
writeStat.getNumUpdateWrites(),
+ writeStat.getTotalLogFiles(), writeStat.getTotalLogBlocks(),
writeStat.getTotalCorruptLogBlock(),
+ writeStat.getTotalRollbackBlocks(), writeStat.getTotalLogRecords(),
+ writeStat.getTotalUpdatedRecordsCompacted(),
writeStat.getTotalWriteBytes(),
+ writeStat.getTotalWriteErrors()});
+ }
+
+ /**
+ * Reads the write stats of the archived commit and delta commit instants
from the given
+ * folder of archive files in the legacy log format written before table
version 8.
+ */
+ private List<Comparable[]>
readCommitStatsFromLegacyArchive(HoodieTableMetaClient metaClient, StoragePath
archivePath) throws IOException {
Review Comment:
readWriteStatRows wraps the partition map in sortByKey but
readCommitStatsFromLegacyArchive still iterates
getPartitionToWriteStats().values() raw, so the legacy rows keep the order of
the HashMap that SpecificData.deepCopy builds.
testShowArchivedCommitsOnLegacyArchive reuses the sorted writeStatRows
expectation for that path and only passes because deepCopy sizes the map to
capacity 4 - could the legacy path go through sortByKey too?
##########
hudi-cli/src/main/java/org/apache/hudi/cli/commands/ArchivedCommitsCommand.java:
##########
@@ -156,94 +345,77 @@ public String showArchivedCommits(
allStats.addAll(readCommits);
}
}
- 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);
+ return 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)
- throws IOException {
-
- System.out.println("===============> Showing only " + limit + " archived
commits <===============");
- HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
- StoragePath archivePath =
- new StoragePath(metaClient.getArchivePath(), ".commits_.archive*");
- HoodieStorage storage = metaClient.getStorage();
- List<StoragePathInfo> pathInfoList = storage.globEntries(archivePath);
- List<Comparable[]> allCommits = new ArrayList<>();
- for (StoragePathInfo pathInfo : pathInfoList) {
- // read the archived file
- try (HoodieLogFormat.Reader reader =
HoodieLogFormat.newReader(metaClient,
- new HoodieLogFile(pathInfo.getPath()),
HoodieSchema.fromAvroSchema(HoodieArchivedMetaEntry.getClassSchema()))) {
- List<IndexedRecord> readRecords = new ArrayList<>();
- // read the avro blocks
- while (reader.hasNext()) {
- HoodieAvroDataBlock blk = (HoodieAvroDataBlock) reader.next();
- try (ClosableIterator<HoodieRecord<IndexedRecord>> recordItr =
blk.getRecordIterator(HoodieRecordType.AVRO)) {
- recordItr.forEachRemaining(r -> readRecords.add(r.getData()));
- }
- }
- List<Comparable[]> readCommits = readRecords.stream().map(r ->
(GenericRecord) r)
- .map(r -> readCommit(r,
skipMetadata)).collect(Collectors.toList());
- allCommits.addAll(readCommits);
- }
- }
-
- 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);
+ private static boolean isCompletedEntry(GenericRecord archivedEntry) {
+ // entries written before the state was recorded are completed ones
+ Object actionState = archivedEntry.get("actionState");
+ return actionState == null ||
HoodieInstant.State.COMPLETED.name().equals(actionState.toString());
}
- private Comparable[] commitDetail(GenericRecord record, String metadataName,
boolean skipMetadata) {
- List<Object> commitDetails = new ArrayList<>();
- commitDetails.add(record.get("commitTime"));
- commitDetails.add(record.get("actionType").toString());
+ private static Comparable[] readArchivedCommit(HoodieArchivedTimeline
archivedTimeline, HoodieInstant instant,
+ boolean skipMetadata, boolean
legacyArchive) {
+ List<Comparable> commitDetails = new ArrayList<>();
+ commitDetails.add(instant.requestedTime());
+ commitDetails.add(instant.getAction());
if (!skipMetadata) {
-
commitDetails.add(Option.ofNullable(record.get(metadataName)).orElse("{}").toString());
+ commitDetails.add(readArchivedMetadataString(archivedTimeline, instant,
legacyArchive));
}
return commitDetails.toArray(new Comparable[commitDetails.size()]);
}
- private Comparable[] readCommit(GenericRecord record, boolean skipMetadata) {
- String actionType = record.get("actionType").toString();
- switch (actionType) {
- case HoodieTimeline.CLEAN_ACTION:
- return commitDetail(record, "hoodieCleanMetadata", skipMetadata);
- case HoodieTimeline.COMMIT_ACTION:
- case HoodieTimeline.DELTA_COMMIT_ACTION:
- return commitDetail(record, "hoodieCommitMetadata", skipMetadata);
- case HoodieTimeline.ROLLBACK_ACTION:
- return commitDetail(record, "hoodieRollbackMetadata", skipMetadata);
- case HoodieTimeline.SAVEPOINT_ACTION:
- return commitDetail(record, "hoodieSavePointMetadata", skipMetadata);
- case HoodieTimeline.COMPACTION_ACTION:
- return commitDetail(record, "hoodieCompactionMetadata", skipMetadata);
- case HoodieTimeline.REPLACE_COMMIT_ACTION:
- case HoodieTimeline.CLUSTERING_ACTION:
- return commitDetail(record, "hoodieReplaceCommitMetadata",
skipMetadata);
- default: {
- throw new HoodieException("Unexpected action type: " + actionType);
+ private static String readArchivedMetadataString(HoodieArchivedTimeline
archivedTimeline, HoodieInstant instant,
+ boolean legacyArchive) {
+ Option<byte[]> details = archivedTimeline.getInstantDetails(instant);
+ if (!details.isPresent() || details.get().length == 0) {
+ // instants can be archived with no metadata, e.g. from an empty
completed
+ // meta file that a writer failure left behind
+ return "{}";
+ }
+ if (legacyArchive) {
+ // ArchivedTimelineV1 caches the JSON rendering of each archived entry,
which is what the
+ // legacy reader printed, and the v1 serde cannot decode it back into
the Avro classes
+ return new String(details.get(), StandardCharsets.UTF_8);
+ }
+ try {
+ // only the actions TimelineArchiverV2 archives can reach here, and it
archives completed
+ // compaction as commit, completed log compaction as deltacommit and
clustering as
+ // replacecommit; savepoints are never archived
+ switch (instant.getAction()) {
+ case HoodieTimeline.CLEAN_ACTION:
+ return archivedTimeline.readCleanMetadata(instant).toString();
+ case HoodieTimeline.COMMIT_ACTION:
+ case HoodieTimeline.DELTA_COMMIT_ACTION:
+ return
sortPartitions(archivedTimeline.readCommitMetadataToAvro(instant)).toString();
+ case HoodieTimeline.ROLLBACK_ACTION:
+ return archivedTimeline.readRollbackMetadata(instant).toString();
+ case HoodieTimeline.REPLACE_COMMIT_ACTION:
+ return
sortPartitions(archivedTimeline.readReplaceCommitMetadataToAvro(instant)).toString();
+ default:
+ throw new HoodieException("Unexpected action type: " +
instant.getAction());
Review Comment:
The savepoint, compaction and clustering arms that the old reader rendered
are gone, and TimelineArchiverV2 is not the only writer of the LSM timeline:
SevenToEightUpgradeHandler.upgradeToLSMTimeline copies a legacy archive across
through LegacyArchivedMetaEntryReader, which filters no action and defaults an
entry with no actionState to COMPLETED. Should the default arm render "{}"
instead of failing the whole listing?
##########
hudi-cli/src/main/java/org/apache/hudi/cli/commands/RepairsCommand.java:
##########
@@ -189,29 +193,52 @@ public String overwriteHoodieProperties(
@ShellMethod(key = "repair corrupted clean files", value = "repair corrupted
clean files")
public void removeCorruptedPendingCleanAction() {
+ removeCorruptedPendingCleanAction(HoodieCLI.getTableMetaClient());
+ }
- HoodieTableMetaClient client = HoodieCLI.getTableMetaClient();
- HoodieTimeline cleanerTimeline =
HoodieCLI.getTableMetaClient().getActiveTimeline().getCleanerTimeline();
+ /**
+ * Removes the pending clean instants whose plan is verifiably empty or
corrupt.
+ * <p>
+ * The plan bytes are read in full before anything is judged. The timeline
serde wraps every
+ * exception raised while it streams an instant file, a transient read
failure included, in
+ * the same "unable to read commit metadata" IOException that an empty or
truncated file
+ * raises, so the message cannot tell a storage outage from corruption. A
failure of the
+ * read itself is therefore propagated, and only the in-memory decode, which
no I/O can
+ * disturb, decides that the plan is corrupt.
+ */
+ static void removeCorruptedPendingCleanAction(HoodieTableMetaClient client) {
+ HoodieActiveTimeline activeTimeline = client.getActiveTimeline();
+ HoodieTimeline cleanerTimeline = activeTimeline.getCleanerTimeline();
log.info("Inspecting pending clean metadata in timeline for corrupted
files");
cleanerTimeline.filterInflightsAndRequested().getInstants().forEach(instant -> {
- try {
- CleanerUtils.getCleanerPlan(client, instant);
- } catch (AvroRuntimeException e) {
- log.warn("Corruption found. Trying to remove corrupted clean instant
file: {}", instant);
- TimelineUtils.deleteInstantFile(client.getStorage(),
client.getTimelinePath(),
- instant, client.getInstantFileNameGenerator());
- } catch (IOException ioe) {
- if (ioe.getMessage().contains("Not an Avro data file")) {
- log.warn("Corruption found. Trying to remove corrupted clean instant
file: {}", instant);
- TimelineUtils.deleteInstantFile(client.getStorage(),
client.getTimelinePath(),
- instant, client.getInstantFileNameGenerator());
- } else {
- throw new HoodieIOException(ioe.getMessage(), ioe);
- }
+ HoodieInstant planInstant = CleanerUtils.getCleanRequestInstant(client,
instant);
+ byte[] plan;
+ try (InputStream in =
activeTimeline.getInstantContentStream(planInstant)) {
+ plan = FileIOUtils.readAsByteArray(in);
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to read the plan of pending clean
instant " + instant, e);
+ }
+ if (plan.length > 0 && isReadableCleanerPlan(client, plan)) {
+ return;
}
+ log.warn("Corruption found. Trying to remove corrupted clean instant
file: {}", instant);
+ TimelineUtils.deleteInstantFile(client.getStorage(),
client.getTimelinePath(),
+ instant, client.getInstantFileNameGenerator());
});
}
+ /**
+ * Decodes a clean plan held in memory, which fails only on the content
itself.
+ */
+ private static boolean isReadableCleanerPlan(HoodieTableMetaClient client,
byte[] plan) {
+ try {
+ CleanerUtils.getCleanerPlan(client, new ByteArrayInputStream(plan));
+ return true;
+ } catch (IOException | AvroRuntimeException e) {
Review Comment:
A plan file that is a valid Avro container with no record - what a writer
killed between DataFileWriter.create() and close() leaves behind - makes
deserializeAvroMetadata raise IllegalArgumentException from
ValidationUtils.checkArgument, which this catch misses, so it escapes the
forEach and aborts the repair instead of removing the instant. Add
IllegalArgumentException to the multi-catch, and truncate the fixture plan just
past the Avro header rather than at half its bytes, which lands inside the
header.
--
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]
