zhaoyudi-creator opened a new issue, #19774:
URL: https://github.com/apache/hudi/issues/19774
### 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:
```sql
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:
```sql
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`):
```java
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.**
```java
// 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.
--
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]
