XuQianJin-Stars commented on code in PR #3518:
URL: https://github.com/apache/fluss/pull/3518#discussion_r3465663443
##########
fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiLakeCommitter.java:
##########
@@ -67,6 +68,7 @@ public HudiLakeCommitter(
this.hudiTableInfo = HudiWriteTableInfo.create(hudiCatalogProvider,
tablePath);
this.writeClient = hudiTableInfo.getWriteClient();
this.ckpMetadata = ckpMetadataProvider.get(tablePath, hudiTableInfo);
+ this.compactionService = new HudiCompactionService(hudiTableInfo,
null, null);
Review Comment:
Constructing `HudiCompactionService` here with `tableBucket=null` and
`partition=null` couples the commit-side and writer-side roles into one class.
The constructor of `HudiCompactionService` does not validate that callers using
`executeCompaction(...)` actually have a non-null `tableBucket`, so the failure
surfaces only at runtime via `checkNotNull` in `executeCompaction`. Consider
splitting into `HudiCompactionScheduler` (used by committer) and
`HudiCompactionExecutor` (used by writer) to make the misuse impossible.
##########
fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiLakeCommitter.java:
##########
@@ -129,7 +129,11 @@ public LakeCommitResult commit(
ckpMetadata.commitInstant(instant);
LOG.info("Committed Hudi instant {} successfully.", instant);
- return
LakeCommitResult.committedIsReadable(parseSnapshotId(instant));
+ String latestCommittedInstant =
+ commitCompactionAndSchedule(
+ committable.getCompactionWriteStats(),
commitMetadata);
+ return LakeCommitResult.committedIsReadable(
+ parseSnapshotId(getLatestCommittedInstant(instant,
latestCommittedInstant)));
Review Comment:
The Fluss snapshot id is now `max(dataInstant, compactionInstant)`.
Compaction instants are minted *after* the data instant (see
`commitCompactionAndSchedule`), so this id will normally be the compaction
instant. That value is later parsed back with `parseSnapshotId` and used by
Fluss to track the lake snapshot mapping (see line 190
`getMissingLakeSnapshot`). The next tiering round will read the data instant
from the timeline and compare against the previously-stored compaction instant,
breaking the idempotency check.
The Fluss snapshot id should remain the data commit instant; compaction
commits should be transparent to the Fluss layer. If there is a reason this
needs to be `max(...)` please add a comment explaining how
`getMissingLakeSnapshot` still works correctly.
--
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]
