[BUG] MOR + BUCKET + NBCC: when a bucket's earliest write is an uncommitted (file-written-but-not-committed) failed write, the entire bucket's later committed data is silently dropped from snapshot queries

[zhaoyudi-creator]

Bug Description

What happened:
With MOR + BUCKET index + NBCC, if the first write into a bucket (FileGroup) is a failed write whose "file has already been produced but the commit did not finish", then all subsequently committed data in that bucket is entirely filtered out of snapshot queries — the query does not fail, it silently returns an incomplete result.

The BUCKET index acts as an amplifier here: it forces the failed write and the later successful writes to share the same FileGroup, so a single failed write drags down all later data in the whole bucket.
What you expected:

Steps to reproduce:
Steps to reproduce the behavior:

1. Create the table:

   CREATE TABLE test_mor (
     id BIGINT, ts BIGINT, v STRING
   ) USING hudi
   TBLPROPERTIES (
     primaryKey = 'id',
     preCombineField = 'ts',
     type = 'mor',
     hoodie.index.type = 'BUCKET',
     hoodie.bucket.index.num.buckets = '2',
     hoodie.metadata.enable = 'false'
   );

2. Enable NBCC (see the concurrency config under Environment Description).
3. Normal write (lands in bucket A): INSERT INTO test_mor VALUES (1, 1000, 'a');
4. Produce one failed write (lands in bucket B, id=2): write (2, 1002, 'b') so that its log file is already produced but the corresponding deltacommit did not finish (the instant stays inflight/requested with no completed marker). For example, kill the job before the commit finishes; under the LAZY policy this failed log is not rolled back immediately and remains in the bucket B directory.
5. Two more normal writes to the same bucket B:

   INSERT INTO test_mor VALUES (2, 1003, 'b');
   INSERT INTO test_mor VALUES (2, 1004, 'b');

6. Run SELECT * FROM test_mor; → only id=1 is returned; the entire id=2 row is missing.

Environment

Hudi version: 1.1.1
Query engine: (Spark/Flink/Trino etc)
Relevant configs:

Logs and Stack Trace

Additional context

Root cause (in hudi-common's HoodieFileGroup file-slice attribution logic; line numbers are from the 1.1 branch and may differ slightly in the community release):

(1) Bucketing puts the failed write and later successful writes into the same FileGroup.
With the BUCKET index, all logs in a bucket share the same fileId (the bucket number). AbstractTableFileSystemView.buildFileGroups groups by HoodieLogFile::getFileId (AbstractTableFileSystemView.java:287), so bucket B's failed-write log (t2) and later successful-write logs (t3, t4) all fall into one FileGroup.

(2) With no base file, the earliest log's deltacommit becomes the slice base instant — without checking whether it is committed.
Logs are processed in ascending deltacommit order. HoodieFileGroup.getBaseInstantTime (HoodieFileGroup.java:192-194):

if (fileSlices.isEmpty()) {
  // no base file in the file group, use the log file delta commit time.
  return logFile.getDeltaCommitTime();   // uses the earliest log's deltacommit directly, no completed check
}

Bucket B's earliest log is exactly the failed t2 (instant time t2 < t3 < t4), so the slice's base instant is anchored on the uncommitted t2.

(3) The successful t3, t4 are routed by completion time into the same t2 slice.
completion(t3) >= t2 holds (HoodieFileGroup.java:196-201), so t3 and t4 are attached to the slice whose base instant = t2. Bucket B ends up with a single slice anchored on the uncommitted t2.

(4) getAllFileSlices filters out that whole slice.

// HoodieFileGroup.java:224-230  isFileSliceCommitted
return timeline.containsOrBeforeTimelineStarts(slice.getBaseInstantTime()); // baseInstant = t2

containsOrBeforeTimelineStarts(t2) = containsInstant(t2) || isBeforeTimelineStarts(t2) (BaseHoodieTimeline.java:480-482). t2 never completed → both are false → returns false. getAllFileSlices (HoodieFileGroup.java:249-251) therefore discards the entire slice, including the already-committed t3/t4 data attached under it.

(5) filterUncommittedFiles cannot recover it.
The filterUncommittedFiles in getLatestFileSlices (AbstractTableFileSystemView.java:750-767) runs after getAllFileSlices; the slice is already gone by then. It also reuses the original base instant when constructing the new slice, so even if it ran it would not change the anchor.

Trigger condition (key): the earliest log in a bucket comes from an uncommitted failed write — it seizes the slice's base-instant anchor. If any successfully committed log/base file exists in the bucket before the failed write, the anchor lands on a committed instant and the failed file is only dropped at file granularity by filterUncommittedFiles, without taking down the whole slice.

[danny0405]

I reproduced this and agree the fix should happen while the file group is constructed, rather than by weakening isFileSliceCommitted or treating a slice as committed merely because it contains a committed log.

The key issue is that addLogFile cannot choose the correct initial base instant when it sees only the failed leading log. The filesystem view already has the complete set of logs for the file group, so the proposed fix adds them as a batch:

1. Sort all logs by delta-commit time inside HoodieFileGroup.
2. If there is no existing base/compaction slice, find the earliest completed log and establish its instant as the initial slice base.
3. Add every log in delta-commit order. Earlier pending logs then follow the existing pending-log rule and attach to the latest slice.
4. Keep isFileSliceCommitted unchanged; it still validates the slice anchor, while the V8+ read-view filtering removes individual uncommitted files from that valid slice.

For the reported sequence, the result becomes:

raw slice t3:     [t2 (failed), t3, t4]
visible slice t3: [t3, t4]

There is no separate slice anchored at the uncommitted t2.

Patch

Download hudi-19774.patch · View the patch on GitHub

The patch includes the production changes and regression tests. The unit regression deliberately supplies logs out of order to verify that sorting is owned by addLogFiles. The filesystem-view regression constructs an MOR file group with t2 inflight and t3/t4 completed, verifies that the raw view has exactly one slice based at t3 containing all three logs, and verifies that the V8+ visible view filters only t2 while retaining t3/t4.

Compatibility notes:

• Existing base files and pending-compaction slices remain authoritative because the initial-base lookup runs only when fileSlices is empty.
• If every log is pending, the existing provisional pending-slice behavior is retained.
• Table-version-6/pre-V8 log naming continues to use the encoded base instant; the existing version-6 slicing tests pass through the new batch API.

Focused result: TestHoodieFileGroup passes all 8 tests, including both pre-V8 and V8+ slicing cases.

[zhaoyudi-creator]

Thanks for the reply, and I agree with your approach — I've actually made the fix along the same lines and verified it: during file group construction, use the instant of the earliest completed (committed) log file as the base instant of the file slice. Next I'll compare your patch against my own solution and confirm it passes the regression on TestHoodieFileGroup. I'll follow up once I have the results. Thanks again!

[danny0405]

An alternative—and currently my preferred fix—is to make the minimal change in isFileSliceCommitted, without changing file-group construction.

A file slice's base instant acts as a slicing barrier; it does not necessarily have to be committed itself. This is analogous to a pending compaction instant. If the first physical file is a pending log at t2, the existing attribution creates a slice at t2, and later completed logs t3/t4 correctly remain in that slice. The actual bug is that slice visibility currently checks only the barrier instant and discards the entire slice before V8+ file-level filtering can run.

The slice should be considered committed/visible when either its base instant is on the timeline or at least one contained log belongs to a committed delta instant:

private boolean isFileSliceCommitted(FileSlice slice) {
  if (!compareTimestamps(
      slice.getBaseInstantTime(),
      LESSER_THAN_OR_EQUALS,
      lastInstant.get().requestedTime())) {
    return false;
  }

  return timeline.containsOrBeforeTimelineStarts(slice.getBaseInstantTime())
      || slice.getLogFiles()
          .anyMatch(logFile -> timeline.containsOrBeforeTimelineStarts(logFile.getDeltaCommitTime()));
}

For the reported sequence:

raw slice t2:     [t2 (pending), t3, t4]
visible slice t2: [t3, t4]

Once the slice passes this check, the existing V8+ filterUncommittedFiles removes the pending t2 log and retains the committed t3/t4 logs. A slice containing only pending files remains hidden from committed APIs and remains accessible through the explicit including-inflight APIs.

This also keeps pre-V8/0.x compatibility: legacy writers query the latest valid base instant before writing and encode the selected base instant in the log filename. No new batch API, sorting relocation, or visibility change to addLogFile is needed.

Alternative minimal patch

Download hudi-19774-minimal.patch · View the patch on GitHub

The patch now includes two regression tests:

• TestHoodieFileGroup verifies that a slice anchored at an uncommitted first-log instant remains visible when it contains later committed logs.
• TestHoodieTableFileSystemView reproduces the V8 filesystem-view path and verifies that the raw slice remains anchored at t2, while the reader view filters t2 and retains t3/t4.

Validation: TestHoodieFileGroup and TestHoodieTableFileSystemView pass through the reactor—59 tests, 0 failures, 0 errors.