You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by si...@apache.org on 2026/04/03 21:18:26 UTC
(hudi) branch master updated: feat(common): Add API to fetch log files created on or before given instant time (#18142)
This is an automated email from the ASF dual-hosted git repository.
sivabalan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 0baabba65f36 feat(common): Add API to fetch log files created on or before given instant time (#18142)
0baabba65f36 is described below
commit 0baabba65f363752fb38892add452f60a8420613
Author: Nada <na...@uber.com>
AuthorDate: Fri Apr 3 17:18:19 2026 -0400
feat(common): Add API to fetch log files created on or before given instant time (#18142)
- Adds a new utility method to fetch log files filtered by commit time, along with the commit times associated with each block in those log files.
---
.../hudi/common/table/log/TestLogReaderUtils.java | 245 +++++++++++++++++++++
.../hudi/common/table/log/LogReaderUtils.java | 79 +++++++
.../common/table/timeline/BaseHoodieTimeline.java | 11 +
.../hudi/common/table/timeline/HoodieTimeline.java | 5 +
4 files changed, 340 insertions(+)
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/common/table/log/TestLogReaderUtils.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/common/table/log/TestLogReaderUtils.java
new file mode 100644
index 000000000000..2ddffc27d875
--- /dev/null
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/common/table/log/TestLogReaderUtils.java
@@ -0,0 +1,245 @@
+/*
+ * 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.log;
+
+import org.apache.hudi.client.SparkRDDWriteClient;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.client.timeline.versioning.v2.TimelineArchiverV2;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
+import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
+import org.apache.hudi.config.HoodieArchivalConfig;
+import org.apache.hudi.config.HoodieCleanConfig;
+import org.apache.hudi.config.HoodieCompactionConfig;
+import org.apache.hudi.config.HoodieIndexConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.index.HoodieIndex;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.table.HoodieSparkTable;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.client.WriteClientTestUtils;
+import org.apache.hudi.testutils.SparkClientFunctionalTestHarness;
+
+import org.apache.spark.api.java.JavaRDD;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.apache.hudi.common.testutils.HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA;
+import static org.apache.hudi.testutils.Assertions.assertNoWriteErrors;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests for {@link LogReaderUtils}.
+ */
+public class TestLogReaderUtils extends SparkClientFunctionalTestHarness {
+
+ @Test
+ public void testGetAllLogFilesWithMaxCommit() throws Exception {
+ HoodieTableMetaClient metaClient = getHoodieMetaClient(HoodieTableType.MERGE_ON_READ, new Properties());
+
+ HoodieWriteConfig config = getConfigBuilder(true)
+ .withPath(basePath())
+ .withCompactionConfig(HoodieCompactionConfig.newBuilder()
+ .withInlineCompaction(false)
+ .compactionSmallFileSize(0)
+ .build())
+ .build();
+
+ HoodieTestDataGenerator dataGen = new HoodieTestDataGenerator();
+
+ try (SparkRDDWriteClient client = getHoodieWriteClient(config)) {
+ // First commit - insert data
+ String firstCommit = "001";
+ WriteClientTestUtils.startCommitWithTime(client, firstCommit);
+ JavaRDD<WriteStatus> insertRdd = client.insert(jsc().parallelize(dataGen.generateInserts(firstCommit, 100), 1), firstCommit);
+ assertNoWriteErrors(insertRdd.collect());
+ client.commit(firstCommit, insertRdd);
+
+ // Second and third commits - upserts to create log files
+ String[] upsertCommits = {"002", "003"};
+ String secondCommit = upsertCommits[0];
+ String thirdCommit = upsertCommits[1];
+ for (String commitTime : upsertCommits) {
+ WriteClientTestUtils.startCommitWithTime(client, commitTime);
+ JavaRDD<WriteStatus> upsertRdd = client.upsert(jsc().parallelize(dataGen.generateUpdates(commitTime, 50), 1), commitTime);
+ List<WriteStatus> statuses = upsertRdd.collect();
+ assertNoWriteErrors(statuses);
+ assertLogFilesProduced(statuses);
+ client.commit(commitTime, upsertRdd);
+ }
+
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+
+ HoodieTableFileSystemView fsView = HoodieTableFileSystemView.fileListingBasedFileSystemView(
+ context(),
+ metaClient,
+ metaClient.getActiveTimeline().filterCompletedInstants());
+
+ List<String> partitions = Arrays.asList(dataGen.getPartitionPaths());
+
+ // Test: Get all log files up to the second commit
+ Map<HoodieLogFile, List<String>> logFilesWithMaxCommit = LogReaderUtils.getAllLogFilesWithMaxCommit(
+ metaClient, fsView, partitions, secondCommit, context());
+
+ assertNotNull(logFilesWithMaxCommit);
+ assertFalse(logFilesWithMaxCommit.isEmpty());
+
+ for (Map.Entry<HoodieLogFile, List<String>> entry : logFilesWithMaxCommit.entrySet()) {
+ assertNotNull(entry.getKey());
+ List<String> commitTimes = entry.getValue();
+ assertFalse(commitTimes.isEmpty(), "Each log file should have at least one associated commit time");
+ for (String commitTime : commitTimes) {
+ assertTrue(commitTime.compareTo(thirdCommit) < 0,
+ "Commit time should be before third commit: " + commitTime);
+ }
+ }
+
+ // Test with third commit - should include same or more log files
+ Map<HoodieLogFile, List<String>> allLogFiles = LogReaderUtils.getAllLogFilesWithMaxCommit(
+ metaClient, fsView, partitions, thirdCommit, context());
+
+ assertNotNull(allLogFiles);
+ assertTrue(allLogFiles.size() >= logFilesWithMaxCommit.size(),
+ "Should have same or more log files when including third commit");
+ }
+ }
+
+ @Test
+ public void testGetAllLogFilesWithMaxCommitEmptyPartitions() throws Exception {
+ HoodieTableMetaClient metaClient = getHoodieMetaClient(HoodieTableType.MERGE_ON_READ, new Properties());
+
+ HoodieTableFileSystemView fsView = HoodieTableFileSystemView.fileListingBasedFileSystemView(
+ context(),
+ metaClient,
+ metaClient.getActiveTimeline().filterCompletedInstants());
+
+ Map<HoodieLogFile, List<String>> result = LogReaderUtils.getAllLogFilesWithMaxCommit(
+ metaClient, fsView, Arrays.asList(), "000", context());
+
+ assertNotNull(result);
+ assertTrue(result.isEmpty(), "Should return empty map for empty partitions");
+ }
+
+ @Test
+ public void testGetAllLogFilesWithArchivedCommit() throws Exception {
+ HoodieTableMetaClient metaClient = getHoodieMetaClient(HoodieTableType.MERGE_ON_READ, new Properties());
+
+ HoodieWriteConfig config = HoodieWriteConfig.newBuilder()
+ .withPath(basePath())
+ .withSchema(TRIP_EXAMPLE_SCHEMA)
+ .withParallelism(2, 2)
+ .withCompactionConfig(HoodieCompactionConfig.newBuilder()
+ .withInlineCompaction(false)
+ .compactionSmallFileSize(0)
+ .withMaxNumDeltaCommitsBeforeCompaction(1)
+ .build())
+ .withCleanConfig(HoodieCleanConfig.newBuilder()
+ .retainCommits(1)
+ .build())
+ .withArchivalConfig(HoodieArchivalConfig.newBuilder()
+ .archiveCommitsWith(2, 4)
+ .build())
+ .withIndexConfig(HoodieIndexConfig.newBuilder()
+ .withIndexType(HoodieIndex.IndexType.BLOOM)
+ .build())
+ .withMetadataConfig(HoodieMetadataConfig.newBuilder()
+ .enable(false)
+ .build())
+ .withEmbeddedTimelineServerEnabled(true)
+ .forTable("test-trip-table")
+ .build();
+
+ HoodieTestDataGenerator dataGen = new HoodieTestDataGenerator();
+
+ try (SparkRDDWriteClient client = getHoodieWriteClient(config)) {
+ // First commit - insert data (creates base files)
+ String firstCommit = "001";
+ WriteClientTestUtils.startCommitWithTime(client, firstCommit);
+ JavaRDD<WriteStatus> insertRdd = client.insert(jsc().parallelize(dataGen.generateInserts(firstCommit, 100), 1), firstCommit);
+ assertNoWriteErrors(insertRdd.collect());
+ client.commit(firstCommit, insertRdd);
+
+ // Upsert commits to create log files and trigger archival (002-006)
+ String[] upsertCommits = {"002", "003", "004", "005", "006"};
+ for (String commitTime : upsertCommits) {
+ WriteClientTestUtils.startCommitWithTime(client, commitTime);
+ JavaRDD<WriteStatus> rdd = client.upsert(jsc().parallelize(dataGen.generateUpdates(commitTime, 20), 1), commitTime);
+ List<WriteStatus> statuses = rdd.collect();
+ assertNoWriteErrors(statuses);
+ assertLogFilesProduced(statuses);
+ client.commit(commitTime, rdd);
+ }
+
+ String sixthCommit = upsertCommits[upsertCommits.length - 1];
+
+ // Trigger archival
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ HoodieTable table = HoodieSparkTable.create(config, context(), metaClient);
+ new TimelineArchiverV2(config, table).archiveIfRequired(context());
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+
+ assertFalse(metaClient.getActiveTimeline().containsInstant(upsertCommits[0]),
+ "Second commit should be archived");
+
+ HoodieTableFileSystemView fsView = HoodieTableFileSystemView.fileListingBasedFileSystemView(
+ context(),
+ metaClient,
+ metaClient.getActiveTimeline().filterCompletedInstants());
+
+ List<String> partitions = Arrays.asList(dataGen.getPartitionPaths());
+
+ Map<HoodieLogFile, List<String>> logFilesWithMaxCommit = LogReaderUtils.getAllLogFilesWithMaxCommit(
+ metaClient, fsView, partitions, sixthCommit, context());
+
+ assertNotNull(logFilesWithMaxCommit);
+ assertFalse(logFilesWithMaxCommit.isEmpty(),
+ "Should return log files even though some commits are archived");
+
+ boolean hasArchivedCommitTime = false;
+ for (List<String> commitTimes : logFilesWithMaxCommit.values()) {
+ assertFalse(commitTimes.isEmpty(), "Each log file should have at least one associated commit time");
+ for (String commitTime : commitTimes) {
+ if (metaClient.getActiveTimeline().isBeforeTimelineStarts(commitTime)) {
+ hasArchivedCommitTime = true;
+ }
+ }
+ }
+
+ assertTrue(hasArchivedCommitTime,
+ "Should have log files with commit times from archived timeline");
+ }
+ }
+
+ private void assertLogFilesProduced(List<WriteStatus> statuses) {
+ assertTrue(statuses.stream()
+ .anyMatch(status -> FSUtils.isLogFile(new StoragePath(status.getStat().getPath()))),
+ "Expected log files to be produced from upsert");
+ }
+}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/log/LogReaderUtils.java b/hudi-common/src/main/java/org/apache/hudi/common/table/log/LogReaderUtils.java
index a837bc9b88b9..6f5c60d49bb9 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/log/LogReaderUtils.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/log/LogReaderUtils.java
@@ -18,8 +18,16 @@
package org.apache.hudi.common.table.log;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.log.block.HoodieLogBlock;
import org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.table.view.AbstractTableFileSystemView;
import org.apache.hudi.common.util.Base64CodecUtil;
+import org.apache.hudi.common.util.collection.Pair;
import org.roaringbitmap.longlong.Roaring64NavigableMap;
@@ -28,12 +36,83 @@ import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
import java.util.Set;
+import static org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType.INSTANT_TIME;
+
/**
* Utils class for performing various log file reading operations.
*/
public class LogReaderUtils {
+
+ /**
+ * Gets a map of log files which were created by commits with instant timestamps that are less than or equal to the
+ * maxCommitInstantTime. All other log files will be filtered out. For each log file maps to a list of commit instant
+ * times that are associated with each block that is found in the log file.
+ *
+ * @param metaClient Hoodie table meta client
+ * @param fsView Hoodie file system view
+ * @param partitionPaths list of partition paths to fetch log files from. for MDT, this should be "files" or
+ * MetadataPartitionType.FILES.partitionPath()
+ * @param maxCommitInstantTime the max commit which created the log files returned
+ * @param engineContext Engine context
+ * @return map of log file -> associated commit time for each block in the log file
+ */
+ public static Map<HoodieLogFile, List<String>> getAllLogFilesWithMaxCommit(HoodieTableMetaClient metaClient,
+ AbstractTableFileSystemView fsView,
+ List<String> partitionPaths,
+ String maxCommitInstantTime,
+ HoodieEngineContext engineContext) {
+ engineContext.setJobStatus("LogReaderUtils",
+ String.format("Getting list of log files in %s partition(s)", partitionPaths.size()));
+
+ List<HoodieLogFile> logFiles = engineContext.flatMap(partitionPaths, partitionPath -> fsView
+ .getLatestMergedFileSlicesBeforeOrOn(partitionPath, maxCommitInstantTime)
+ .flatMap(FileSlice::getLogFiles),
+ Math.max(partitionPaths.size(), 1));
+
+ // get completion time of the max commit instant
+ String maxCommitCompletionTime = metaClient.getActiveTimeline().filterCompletedInstants()
+ .findInstantsAfterOrEquals(maxCommitInstantTime, 1)
+ .filter(instant -> instant.requestedTime().equals(maxCommitInstantTime))
+ .getInstants()
+ .stream()
+ .map(instant -> instant.getCompletionTime())
+ .findFirst().orElse(maxCommitInstantTime);
+
+ // filter out all commits completed after the max commit completion time
+ HoodieTimeline filteredTimeline = fsView.getTimeline()
+ .findInstantsModifiedBeforeOrEqualsByCompletionTime(maxCommitCompletionTime);
+
+
+ engineContext.setJobStatus("LogReaderUtils",
+ String.format("Getting log file map for %s partition(s)", partitionPaths.size()));
+ Map<HoodieLogFile, List<String>> logFilesWithMaxCommit = engineContext.mapToPair(logFiles, logFile -> {
+ // read all blocks within the log file and find the commit associated with each log block
+ List<String> blocksWithinLogFile = new ArrayList<>();
+ try (HoodieLogFormat.Reader reader = HoodieLogFormat.newReader(metaClient.getStorage(), logFile, null)) {
+ while (reader.hasNext()) {
+ HoodieLogBlock block = reader.next();
+ String logBlockInstantTime = block.getLogBlockHeader().get(INSTANT_TIME);
+ // check if the log file contains a block created by a commit that is older than or equal to max commit
+ // use containsOrBeforeTimelineStarts to handle cases where the log block's instant time is in archived timeline
+ if (filteredTimeline.containsOrBeforeTimelineStarts(logBlockInstantTime)) {
+ blocksWithinLogFile.add(logBlockInstantTime);
+ }
+ }
+ }
+
+ return Pair.of(logFile, blocksWithinLogFile);
+ }, Math.max(logFiles.size(), 1));
+
+ logFilesWithMaxCommit.values().removeIf(List::isEmpty);
+
+ return logFilesWithMaxCommit;
+ }
+
/**
* Encodes a list of record positions in long type.
* <p>
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/BaseHoodieTimeline.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/BaseHoodieTimeline.java
index d3b9818199a1..f32cce52c451 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/BaseHoodieTimeline.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/BaseHoodieTimeline.java
@@ -292,6 +292,17 @@ public abstract class BaseHoodieTimeline implements HoodieTimeline {
getInstantReader());
}
+ @Override
+ public HoodieTimeline findInstantsModifiedBeforeOrEqualsByCompletionTime(String instantTime) {
+ return factory.createDefaultTimeline(instants.stream()
+ // completed instants with completionTime <= instantTime or matching requestedTime,
+ // or pending instants with requestedTime <= instantTime
+ .filter(s -> (s.getCompletionTime() != null
+ && (compareTimestamps(s.getCompletionTime(), LESSER_THAN_OR_EQUALS, instantTime) || s.requestedTime().equals(instantTime)))
+ || (s.getCompletionTime() == null && compareTimestamps(s.requestedTime(), LESSER_THAN_OR_EQUALS, instantTime))),
+ getInstantReader());
+ }
+
@Override
public HoodieTimeline findInstantsAfter(String instantTime, int numCommits) {
return factory.createDefaultTimeline(getInstantsAsStream()
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieTimeline.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieTimeline.java
index 77dc91d4dff0..f9a59c6ab623 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieTimeline.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieTimeline.java
@@ -487,6 +487,11 @@ public interface HoodieTimeline extends HoodieInstantReader, Serializable {
*/
HoodieTimeline findInstantsModifiedAfterByCompletionTime(String instantTime);
+ /**
+ * Create new timeline with all instants that were modified before or at the specified time.
+ */
+ HoodieTimeline findInstantsModifiedBeforeOrEqualsByCompletionTime(String instantTime);
+
/**
* Create a new Timeline with all the instants after startTs.
*/