danny0405 commented on code in PR #20099: URL: https://github.com/apache/hudi/pull/20099#discussion_r4408105232
########## hudi-common/src/main/java/org/apache/hudi/common/table/timeline/ArchivedCleanTimelineUtils.java: ########## @@ -0,0 +1,115 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hudi.common.table.timeline; + +import org.apache.hudi.common.avro.HoodieAvroUtils; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion; +import org.apache.hudi.exception.HoodieIOException; + +import org.apache.avro.generic.IndexedRecord; + +import java.io.ByteArrayInputStream; +import java.io.InputStream; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.stream.Collectors; + +import static org.apache.hudi.common.table.timeline.versioning.v2.ArchivedTimelineV2.ACTION_ARCHIVED_META_FIELD; +import static org.apache.hudi.common.table.timeline.versioning.v2.ArchivedTimelineV2.METADATA_ARCHIVED_META_FIELD; +import static org.apache.hudi.common.table.timeline.versioning.v2.ArchivedTimelineV2.PLAN_ARCHIVED_META_FIELD; +import static org.apache.hudi.common.util.ValidationUtils.checkArgument; + +/** Utilities for reading archived cleaner plans and metadata in either timeline layout. */ +public class ArchivedCleanTimelineUtils { + + private ArchivedCleanTimelineUtils() { + } + + /** + * Creates a timeline for the selected completed cleans, supplied in descending requested-time order. + * Payloads are loaded lazily in windows of at most {@code batchSize} selected instants; only the + * current window is retained. This lets callers stop after enough partition rows, including when + * some cleans have empty partition metadata, without loading the entire archived history. + */ + public static HoodieTimeline getTimeline(HoodieTableMetaClient metaClient, List<HoodieInstant> selectedInstants, + boolean loadPlans, int batchSize) { + checkArgument(batchSize > 0, "Batch size must be positive"); + List<HoodieInstant> instants = new ArrayList<>(selectedInstants); + TimelineFactory factory = metaClient.getTableFormat().getTimelineFactory(); + Map<String, Integer> positions = new HashMap<>(); + for (int i = 0; i < instants.size(); i++) { + positions.put(instants.get(i).requestedTime(), i); + } + boolean legacy = metaClient.getTimelineLayoutVersion().getVersion() < TimelineLayoutVersion.VERSION_2; + String actionField = legacy ? "actionType" : ACTION_ARCHIVED_META_FIELD; + String contentField = legacy ? (loadPlans ? "hoodieCleanerPlan" : "hoodieCleanMetadata") + : (loadPlans ? PLAN_ARCHIVED_META_FIELD : METADATA_ARCHIVED_META_FIELD); + HoodieArchivedTimeline.LoadMode loadMode = loadPlans ? HoodieArchivedTimeline.LoadMode.PLAN : HoodieArchivedTimeline.LoadMode.METADATA; + HoodieInstantReader reader = new HoodieInstantReader() { + private final Map<String, byte[]> contents = new ConcurrentHashMap<>(); + + @Override + public synchronized InputStream getContentStream(HoodieInstant instant) { + String instantTime = instant.requestedTime(); + if (!contents.containsKey(instantTime)) { + Integer position = positions.get(instantTime); + checkArgument(position != null, "Clean instant was not selected: " + instantTime); + List<HoodieInstant> batch = instants.subList(position, position + Math.min(batchSize, instants.size() - position)); + Set<String> batchTimes = new HashSet<>(batch.stream().map(HoodieInstant::requestedTime).collect(Collectors.toList())); + HoodieArchivedTimeline.TimeRangeFilter range = new HoodieArchivedTimeline.ClosedClosedTimeRangeFilter( + batch.get(batch.size() - 1).requestedTime(), batch.get(0).requestedTime()); + contents.clear(); + factory.createArchivedTimelineLoader().loadInstants(metaClient, legacy ? null : range, loadMode, Review Comment: [P2] Avoid repeatedly scanning the same V2 archive file across batches. In addition to the V1 rescan concern already discussed, V2's getFilteredFiles only prunes whole files by time range. loadInstants then iterates every record in each overlapping file and applies the range check after iterator.next(). Consequently, if a compacted file contains 320 selected cleans, consuming them in the callers' 32-instant windows reopens and traverses that file 10 times. With show_cleans_metadata(limit => 1) and many empty cleans, the one-instant windows can make this substantially worse. The mocked fixture.loaded assertions count accepted payloads and do not detect these repeated physical reads. Could we follow up with file-aware iteration/caching or reader-level range pruning, and a test that counts file opens or records visited across multiple windows in one V2 archive file? This is a performance concern inferred from the loader, not a measured benchmark. ########## hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCleansProcedure.scala: ########## @@ -256,6 +249,37 @@ object ShowCleansProcedure { val NAME = "show_cleans" def builder: Supplier[ProcedureBuilder] = () => new ShowCleansProcedure(false) + + private[procedures] def getCleanTimeline(metaClient: HoodieTableMetaClient, + showArchived: Boolean, + loadPlans: Boolean, + instantLimit: Int, + batchSize: Int): HoodieTimeline = { + val activeTimeline = metaClient.getActiveTimeline + val activeCleans = if (loadPlans) activeTimeline.getCleanerTimeline + else activeTimeline.getCleanerTimeline.filterCompletedInstants + val activeInstants = activeCleans.getInstants.asScala.toSeq + val archivedInstants = if (showArchived) { + metaClient.getArchivedTimeline.getCleanerTimeline.filterCompletedInstants.getInstants.asScala.toSeq + } else { + Seq.empty[HoodieInstant] + } + val comparator = TimelineLayout.fromVersion(metaClient.getTimelineLayoutVersion) + .getInstantComparator.requestedTimeOrderedComparator.reversed() + val selected = (activeInstants.map(_ -> true) ++ archivedInstants.map(_ -> false)) + .sortWith((a, b) => comparator.compare(a._1, b._1) < 0).take(instantLimit) + val selectedArchived = selected.collect { case (instant, false) => instant } + val archivedTimeline = ArchivedCleanTimelineUtils.getTimeline(metaClient, selectedArchived.asJava, + loadPlans, batchSize) + val activeTimes = selected.collect { case (instant, true) => instant.requestedTime() }.toSet + metaClient.getTableFormat.getTimelineFactory.createDefaultTimeline( + selected.map(_._1).reverse.asJava.stream(), new HoodieInstantReader { Review Comment: [P3] Simplify the repeated ordering and collection conversions. selected is already sorted descending, and this line stores it ascending, but both consumers then sort again (getSortedCleans for metadata and getReverseOrderedInstants for plans); the archived helper also sorts its selected subset. In particular, the partition-metadata path keeps all instant descriptors, so it repeats that work over the full clean history even for a small row limit. Could the row-building code consume a single selected descending sequence (or reverse the known ascending timeline list), retaining this timeline only for payload lookup? That would also let getSortedCleans shed its Java list -> array -> Scala list -> Java list -> ArrayList conversion chain. Non-blocking simplification. -- 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]
