This is an automated email from the ASF dual-hosted git repository.
danny0405 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 1d3a0affad87 fix(common): anchor a file group's first slice on a
committed log (#19785)
1d3a0affad87 is described below
commit 1d3a0affad873868e74df7207013ae8eeb4dbb8c
Author: zhaoyudi-creator <[email protected]>
AuthorDate: Fri Sep 11 14:48:18 2026 +0800
fix(common): anchor a file group's first slice on a committed log (#19785)
A MOR file group can end up with its latest file slice keyed on a base
instant that never committed. For example, under NBCC with the bucket
index the earliest delta commit on a file group fails (or is rolled back)
while later delta commits on the same group succeed. Log files are
attributed to a slice by completion time, so when the group is first
built the failed instant's log arrives first and opens a slice keyed on
that uncommitted instant; every later committed log is then attributed to
the same slice.
isFileSliceCommitted only checks whether the slice's base instant itself
committed, so the whole slice -- including its committed log files -- is
treated as uncommitted and dropped from the reader view, silently losing
committed data.
Fix this at file-group construction instead of weakening the visibility
gate. When the group has no slice yet, the earliest completed log now
establishes the initial slice, so the slice is anchored on a real
committed instant. An earlier pending log then follows the normal
pending-log rule and attaches to that committed slice, instead of
creating its own uncommitted slice that would also hide every later
committed log in the group. isFileSliceCommitted keeps its original
single invariant, so all existing and future view APIs benefit without
per-call trimming.
addLogFiles now owns the sort and the re-anchoring for a batch of log
files; addLogFile becomes private and the file-system view calls the
batch API. Sorting is done once and reused by both the re-anchor scan and
the attach loop.
Tests:
- TestHoodieFileGroup#testUncommittedFirstLogDoesNotAnchorCommittedLogs
verifies an uncommitted earliest log does not anchor the slice; the
first committed log becomes the base instant and the pending log
attaches to it.
-
TestHoodieFileGroup#testUncommittedBaseFileSliceStaysHiddenDespiteCommittedLogs
verifies a slice with an uncommitted base file stays hidden even when
later committed logs land in the same file group.
-
TestHoodieTableFileSystemView#testUncommittedFirstLogUsesFirstCommittedLogAsBaseInstant
verifies the file-system view anchors the raw slice on the first
committed log and surfaces only the committed logs.
Closes #19774
---
.../apache/hudi/common/model/HoodieFileGroup.java | 23 +++++-
.../table/view/AbstractTableFileSystemView.java | 3 +-
.../hudi/common/model/TestHoodieFileGroup.java | 92 +++++++++++++++++++---
.../table/view/TestHoodieTableFileSystemView.java | 44 +++++++++++
4 files changed, 149 insertions(+), 13 deletions(-)
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieFileGroup.java
b/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieFileGroup.java
index 89088cb14190..5b31db849504 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieFileGroup.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieFileGroup.java
@@ -32,6 +32,7 @@ import java.io.Serializable;
import java.util.Comparator;
import java.util.List;
import java.util.TreeMap;
+import java.util.stream.Collectors;
import java.util.stream.Stream;
import static
org.apache.hudi.common.table.timeline.InstantComparison.GREATER_THAN_OR_EQUALS;
@@ -117,7 +118,7 @@ public class HoodieFileGroup implements Serializable {
*
* <p>CAUTION: the log file must be added in sequence of the delta commit
time.
*/
- public void addLogFile(CompletionTimeQueryView completionTimeQueryView,
HoodieLogFile logFile) {
+ private void addLogFile(CompletionTimeQueryView completionTimeQueryView,
HoodieLogFile logFile) {
String baseInstantTime = getBaseInstantTime(completionTimeQueryView,
logFile);
if (!fileSlices.containsKey(baseInstantTime)) {
fileSlices.put(baseInstantTime, new FileSlice(fileGroupId,
baseInstantTime));
@@ -125,6 +126,26 @@ public class HoodieFileGroup implements Serializable {
fileSlices.get(baseInstantTime).addLogFile(logFile);
}
+ /**
+ * Add a batch of log files into the group, sorted by the delta commit time.
+ *
+ * <p>When the group has no existing slice yet, the earliest completed log
establishes the initial
+ * slice before any log is added. An earlier pending log then follows the
normal pending-log rule
+ * and attaches to that slice, instead of creating its own uncommitted slice
that would also hide
+ * every later committed log in the group.
+ */
+ public void addLogFiles(CompletionTimeQueryView completionTimeQueryView,
List<HoodieLogFile> logFiles) {
+ List<HoodieLogFile> sortedLogFiles = logFiles.stream()
+
.sorted(HoodieLogFile.getLogFileComparator()).collect(Collectors.toList());
+ if (fileSlices.isEmpty()) {
+ sortedLogFiles.stream()
+ .filter(logFile ->
completionTimeQueryView.isCompleted(logFile.getDeltaCommitTime()))
+ .findFirst()
+ .ifPresent(logFile ->
addNewFileSliceAtInstant(logFile.getDeltaCommitTime()));
+ }
+ sortedLogFiles.forEach(logFile -> addLogFile(completionTimeQueryView,
logFile));
+ }
+
@VisibleForTesting
public String getBaseInstantTime(CompletionTimeQueryView
completionTimeQueryView, HoodieLogFile logFile) {
if (fileSlices.isEmpty()) {
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java
index 75118b90bb2c..f768c6880def 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java
@@ -256,8 +256,7 @@ public abstract class AbstractTableFileSystemView
implements SyncableFileSystemV
}
if (logFiles.containsKey(fileId)) {
// this should work for both table versions >= 8 and lower.
-
logFiles.get(fileId).stream().sorted(HoodieLogFile.getLogFileComparator())
- .forEach(logFile -> group.addLogFile(completionTimeQueryView,
logFile));
+ group.addLogFiles(completionTimeQueryView, logFiles.get(fileId));
}
fileGroups.add(group);
});
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/model/TestHoodieFileGroup.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/model/TestHoodieFileGroup.java
index 0e37280adfb9..1b62ae4c534f 100644
---
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/model/TestHoodieFileGroup.java
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/model/TestHoodieFileGroup.java
@@ -86,7 +86,8 @@ public class TestHoodieFileGroup {
for (int i = 0; i < 3; i++) {
HoodieBaseFile baseFile = new HoodieBaseFile("data_1_00" + i +
".parquet");
fileGroup.addBaseFile(baseFile);
- fileGroup.addLogFile(queryView, new HoodieLogFile(new
StoragePath(FileCreateUtilsLegacy.logFileName(preTableVersion8 ? "001" : "00" +
i, "data", i))));
+ addLogFile(fileGroup, queryView,
+ new HoodieLogFile(new
StoragePath(FileCreateUtilsLegacy.logFileName(preTableVersion8 ? "001" : "00" +
i, "data", i))));
}
assertEquals(2, fileGroup.getAllFileSlices().count());
@@ -125,16 +126,22 @@ public class TestHoodieFileGroup {
// when: building a file group with file slices like table version 6.
HoodieFileGroup fileGroup = new HoodieFileGroup("", "f1", activeTimeline);
- fileGroup.addLogFile(queryView, new HoodieLogFile(new
StoragePath(FileCreateUtilsLegacy.logFileName("001", "f1", 0))));
- fileGroup.addLogFile(queryView, new HoodieLogFile(new
StoragePath(FileCreateUtilsLegacy.logFileName(useBaseInstantTime ? "001" :
"002", "f1", 1))));
+ addLogFile(fileGroup, queryView,
+ new HoodieLogFile(new
StoragePath(FileCreateUtilsLegacy.logFileName("001", "f1", 0))));
+ addLogFile(fileGroup, queryView,
+ new HoodieLogFile(new
StoragePath(FileCreateUtilsLegacy.logFileName(useBaseInstantTime ? "001" :
"002", "f1", 1))));
fileGroup.addBaseFile(new
HoodieBaseFile(FileCreateUtilsLegacy.baseFileName("003", "f1")));
- fileGroup.addLogFile(queryView, new HoodieLogFile(new
StoragePath(FileCreateUtilsLegacy.logFileName(useBaseInstantTime ? "003" :
"004", "f1", 0))));
- fileGroup.addLogFile(queryView, new HoodieLogFile(new
StoragePath(FileCreateUtilsLegacy.logFileName(useBaseInstantTime ? "003" :
"005", "f1", 1))));
+ addLogFile(fileGroup, queryView,
+ new HoodieLogFile(new
StoragePath(FileCreateUtilsLegacy.logFileName(useBaseInstantTime ? "003" :
"004", "f1", 0))));
+ addLogFile(fileGroup, queryView,
+ new HoodieLogFile(new
StoragePath(FileCreateUtilsLegacy.logFileName(useBaseInstantTime ? "003" :
"005", "f1", 1))));
fileGroup.addBaseFile(new
HoodieBaseFile(FileCreateUtilsLegacy.baseFileName("006", "f1")));
- fileGroup.addLogFile(queryView, new HoodieLogFile(new
StoragePath(FileCreateUtilsLegacy.logFileName(useBaseInstantTime ? "006" :
"007", "f1", 0))));
- fileGroup.addLogFile(queryView, new HoodieLogFile(new
StoragePath(FileCreateUtilsLegacy.logFileName(useBaseInstantTime ? "006" :
"008", "f1", 1))));
+ addLogFile(fileGroup, queryView,
+ new HoodieLogFile(new
StoragePath(FileCreateUtilsLegacy.logFileName(useBaseInstantTime ? "006" :
"007", "f1", 0))));
+ addLogFile(fileGroup, queryView,
+ new HoodieLogFile(new
StoragePath(FileCreateUtilsLegacy.logFileName(useBaseInstantTime ? "006" :
"008", "f1", 1))));
// then: assert that the file slices are in-tact.
assertEquals(3, fileGroup.getAllFileSlices().count());
@@ -161,6 +168,63 @@ public class TestHoodieFileGroup {
testFileSlicingForTableVersion(false);
}
+ @Test
+ public void testUncommittedFirstLogDoesNotAnchorCommittedLogs() {
+ // The earliest log "001" is still pending while "002"/"003" are
committed. The group must not
+ // create an uncommitted slice anchored at "001" (which would hide the
committed logs); the
+ // earliest completed log "002" establishes the slice and "001" attaches
to it as a pending log.
+ MockHoodieTimeline activeTimeline = new MockHoodieTimeline(Stream.of(
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.COMMIT_ACTION, "000", "000"),
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.INFLIGHT,
HoodieTimeline.DELTA_COMMIT_ACTION, "001"),
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.DELTA_COMMIT_ACTION, "002", "004"),
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.DELTA_COMMIT_ACTION, "003", "005")
+ ).collect(Collectors.toList()));
+ CompletionTimeQueryView queryView =
getMockCompletionTimeQueryView(activeTimeline);
+ HoodieFileGroup fileGroup = new HoodieFileGroup("", "data",
activeTimeline.filterCompletedAndCompactionInstants());
+
+ // logs are supplied out of order on purpose: sorting is owned by
addLogFiles.
+ fileGroup.addLogFiles(queryView, Stream.of("003", "001", "002")
+ .map(instant -> new HoodieLogFile(new
StoragePath(getLogFileName(instant))))
+ .collect(Collectors.toList()));
+
+ assertEquals(CollectionUtils.createImmutableList("002"),
+
fileGroup.getAllFileSlicesIncludingInflight().map(FileSlice::getBaseInstantTime).collect(Collectors.toList()));
+ assertEquals(CollectionUtils.createImmutableList("002"),
+
fileGroup.getAllFileSlices().map(FileSlice::getBaseInstantTime).collect(Collectors.toList()));
+ FileSlice committedSlice = fileGroup.getLatestFileSlice().get();
+ assertEquals(CollectionUtils.createImmutableList("001", "002", "003"),
+
committedSlice.getLogFiles().map(HoodieLogFile::getDeltaCommitTime).sorted().collect(Collectors.toList()));
+ }
+
+ @Test
+ public void testUncommittedBaseFileSliceStaysHiddenDespiteCommittedLogs() {
+ // An uncommitted base file at "001" (e.g. a failed compaction/bulk_insert
under LAZY) shares the
+ // file group with later committed logs "002"/"003". The slice must stay
hidden - the committed
+ // logs cannot revive it, otherwise the uncommitted base file would be
read in full.
+ MockHoodieTimeline activeTimeline = new MockHoodieTimeline(Stream.of(
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.COMMIT_ACTION, "000", "000"),
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.INFLIGHT,
HoodieTimeline.COMMIT_ACTION, "001"),
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.DELTA_COMMIT_ACTION, "002", "004"),
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.DELTA_COMMIT_ACTION, "003", "005")
+ ).collect(Collectors.toList()));
+ CompletionTimeQueryView queryView =
getMockCompletionTimeQueryView(activeTimeline);
+ HoodieFileGroup fileGroup = new HoodieFileGroup("", "data",
activeTimeline.filterCompletedAndCompactionInstants());
+
+ // uncommitted base file at "001"
+ fileGroup.addBaseFile(new HoodieBaseFile(getBaseFileName("001")));
+ // later committed logs land in the same file group
+ fileGroup.addLogFiles(queryView, Stream.of("002", "003")
+ .map(instant -> new HoodieLogFile(new
StoragePath(getLogFileName(instant))))
+ .collect(Collectors.toList()));
+
+ // the slice exists on disk but stays hidden because its base file is
uncommitted
+ assertEquals(CollectionUtils.createImmutableList("001"),
+
fileGroup.getAllFileSlicesIncludingInflight().map(FileSlice::getBaseInstantTime).collect(Collectors.toList()));
+ assertEquals(0, fileGroup.getAllFileSlices().count());
+ assertTrue(fileGroup.getLatestFileSlice().isEmpty());
+ assertTrue(fileGroup.getLatestDataFile().isEmpty());
+ }
+
@Test
public void testGetBaseInstantTime() {
MockHoodieTimeline activeTimeline = new MockHoodieTimeline(Stream.of(
@@ -176,13 +240,13 @@ public class TestHoodieFileGroup {
HoodieFileGroup fileGroup = new HoodieFileGroup("", "data",
activeTimeline.filterCompletedAndCompactionInstants());
HoodieLogFile logFile1 = new HoodieLogFile(new
StoragePath(getLogFileName("001")));
- fileGroup.addLogFile(queryView, logFile1);
+ addLogFile(fileGroup, queryView, logFile1);
assertThat("no base file in the file group, returns the delta commit
instant itself",
fileGroup.getBaseInstantTime(queryView, logFile1), is("001"));
assertThat(collectFileSlices(fileGroup), is("001"));
HoodieLogFile logFile2 = new HoodieLogFile(new
StoragePath(getLogFileName("002")));
- fileGroup.addLogFile(queryView, logFile2);
+ addLogFile(fileGroup, queryView, logFile2);
assertThat("no base file in the file group, returns the earliest delta
commit instant",
fileGroup.getBaseInstantTime(queryView, logFile2), is("001"));
assertThat(collectFileSlices(fileGroup), is("001"));
@@ -192,7 +256,7 @@ public class TestHoodieFileGroup {
collectFileSlices(fileGroup), is("001,003"));
HoodieLogFile logFile3 = new HoodieLogFile(new
StoragePath(getLogFileName("004")));
- fileGroup.addLogFile(queryView, logFile3);
+ addLogFile(fileGroup, queryView, logFile3);
assertThat("Assign the log file to maximum base instant time that less
than or equals its completion time",
fileGroup.getBaseInstantTime(queryView, logFile2), is("003"));
assertThat(collectFileSlices(fileGroup), is("001,003"));
@@ -217,9 +281,17 @@ public class TestHoodieFileGroup {
String instantTime = invocationOnMock.getArgument(1);
return Option.ofNullable(completionTimeMap.get(instantTime));
});
+ when(queryView.isCompleted(any(String.class)))
+ .thenAnswer((InvocationOnMock invocationOnMock) ->
completionTimeMap.containsKey(invocationOnMock.getArgument(0)));
return queryView;
}
+ private static void addLogFile(HoodieFileGroup fileGroup,
+ CompletionTimeQueryView
completionTimeQueryView,
+ HoodieLogFile logFile) {
+ fileGroup.addLogFiles(completionTimeQueryView,
CollectionUtils.createImmutableList(logFile));
+ }
+
private static String collectFileSlices(HoodieFileGroup fileGroup) {
return
fileGroup.getAllFileSlices().map(FileSlice::getBaseInstantTime).sorted().collect(Collectors.joining(","));
}
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestHoodieTableFileSystemView.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestHoodieTableFileSystemView.java
index ac068ec15a65..e4decc0d1c14 100644
---
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestHoodieTableFileSystemView.java
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestHoodieTableFileSystemView.java
@@ -325,6 +325,50 @@ public class TestHoodieTableFileSystemView extends
HoodieCommonTestHarness {
"Total number of file-groups in view matches expected");
}
+ @Test
+ public void testUncommittedFirstLogUsesFirstCommittedLogAsBaseInstant()
throws Exception {
+ String partitionPath = "2016/05/01";
+ Paths.get(basePath, partitionPath).toFile().mkdirs();
+ String fileId = UUID.randomUUID().toString();
+
+ String failedInstant = "2";
+ String firstCommittedInstant = "3";
+ String secondCommittedInstant = "4";
+ for (String instant : Arrays.asList(failedInstant, firstCommittedInstant,
secondCommittedInstant)) {
+ String fileName = FSUtils.makeInlineLogFileName(
+ fileId, HoodieLogFile.DELTA_EXTENSION, instant,
Integer.parseInt(instant), TEST_WRITE_TOKEN);
+ Paths.get(basePath, partitionPath, fileName).toFile().createNewFile();
+ }
+
+ HoodieActiveTimeline commitTimeline = metaClient.getActiveTimeline();
+ saveAsComplete(commitTimeline,
+ INSTANT_GENERATOR.createNewInstant(State.INFLIGHT,
HoodieTimeline.COMMIT_ACTION, "1"),
+ new HoodieCommitMetadata());
+ HoodieInstant failedRequested =
+ INSTANT_GENERATOR.createNewInstant(State.REQUESTED,
HoodieTimeline.DELTA_COMMIT_ACTION, failedInstant);
+ commitTimeline.createNewInstant(failedRequested);
+ commitTimeline.transitionRequestedToInflight(failedRequested,
Option.empty());
+ saveAsComplete(commitTimeline,
+ INSTANT_GENERATOR.createNewInstant(State.INFLIGHT,
HoodieTimeline.DELTA_COMMIT_ACTION, firstCommittedInstant),
+ new HoodieCommitMetadata());
+ saveAsComplete(commitTimeline,
+ INSTANT_GENERATOR.createNewInstant(State.INFLIGHT,
HoodieTimeline.DELTA_COMMIT_ACTION, secondCommittedInstant),
+ new HoodieCommitMetadata());
+
+ refreshFsView();
+
+ HoodieFileGroup fileGroup =
fsView.getAllFileGroups(partitionPath).findFirst().get();
+ List<FileSlice> rawSlices =
fileGroup.getAllFileSlicesIncludingInflight().collect(Collectors.toList());
+ assertEquals(1, rawSlices.size());
+ assertEquals(firstCommittedInstant, rawSlices.get(0).getBaseInstantTime());
+ assertEquals(3, rawSlices.get(0).getLogFileCnt());
+
+ FileSlice visibleSlice =
rtView.getLatestFileSlices(partitionPath).findFirst().get();
+ assertEquals(firstCommittedInstant, visibleSlice.getBaseInstantTime());
+ assertEquals(Arrays.asList(firstCommittedInstant, secondCommittedInstant),
+
visibleSlice.getLogFiles().map(HoodieLogFile::getDeltaCommitTime).sorted().collect(Collectors.toList()));
+ }
+
@ParameterizedTest(name = TEST_NAME_WITH_PARAMS_2)
@MethodSource("configParams2x2")
public void
testViewForFileSlicesWithNoBaseFileAndRequestedCompaction(boolean
testBootstrap, boolean preTableVersion8) throws Exception {
