Copilot commented on code in PR #2454:
URL: https://github.com/apache/fluss/pull/2454#discussion_r2928840733
##########
fluss-server/src/test/java/org/apache/fluss/server/coordinator/LakeTableTieringManagerTest.java:
##########
@@ -278,6 +288,90 @@ void testGlobalMetrics() throws Exception {
assertThat(tableTieringManager.getPendingTablesCount()).isEqualTo(1);
// back to pending
}
+ @Test
+ void testTableLevelMetrics() {
+ long tableId1 = 1L;
+ TablePath tablePath1 = TablePath.of("db", "table1");
+ TableInfo tableInfo1 = createTableInfo(tableId1, tablePath1,
Duration.ofSeconds(10));
+ tableTieringManager.addNewLakeTable(tableInfo1);
+
+ // Initially, table is just created, lastSuccessTime is the current
time
+ Long initialTime =
tableTieringManager.getTableLastSuccessTime(tableId1);
+ assertThat(initialTime).isNotNull();
+ assertThat(initialTime).isEqualTo(manualClock.milliseconds());
+
+ // Advance time and request table
+ manualClock.advanceTime(Duration.ofSeconds(10));
+ assertRequestTable(tableId1, tablePath1, 1);
+
+ // State should be Tiering (4)
+ assertThat(tableTieringManager.getTableState(tableId1))
+ .isEqualTo(LakeTableTieringManager.TieringState.Tiering);
+
+ // Simulate tiering duration — finish with UNKNOWN stats (empty
commit).
+ // Duration should NOT be updated because no data was written.
+ manualClock.advanceTime(Duration.ofSeconds(5));
+ tableTieringManager.finishTableTiering(tableId1, 1, false,
TieringStats.UNKNOWN);
+ assertThat(tableTieringManager.getLastTieringResultField(tableId1, r
-> r.tierDuration))
+ .isEqualTo(5000L);
Review Comment:
The comment on line 311 says "Duration should NOT be updated because no data
was written", but the assertion on line 315-316 verifies that `tierDuration` IS
updated to 5000L. The comment is incorrect — `updateTableTieringResult` always
updates `tieredTime` and `tierDuration` regardless of whether stats are
available. The comment should say something like "Duration should still be
updated even with UNKNOWN stats".
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperator.java:
##########
@@ -162,28 +181,31 @@ public void
processElement(StreamRecord<TableBucketWriteResult<WriteResult>> str
}
}
- @Nullable
- private Committable commitWriteResults(
+ /**
+ * Commits the collected write results for one table to the lake and Fluss.
+ *
+ * <p>Always returns a non-null {@link CommitResult}. When all buckets
produced no data (empty
+ * commit), {@link CommitResult#committable} is {@code null} and stats are
{@link
+ * TieringStats#UNKNOWN}.
+ */
+ private CommitResult commitWriteResults(
long tableId,
TablePath tablePath,
List<TableBucketWriteResult<WriteResult>> committableWriteResults)
throws Exception {
- // filter out non-null write result
- committableWriteResults =
+ // filter down to buckets that actually produced data
+ List<TableBucketWriteResult<WriteResult>> nonEmptyResults =
committableWriteResults.stream()
- .filter(
- writeResultTableBucketWriteResult ->
-
writeResultTableBucketWriteResult.writeResult() != null)
+ .filter(r -> r.writeResult() != null)
.collect(Collectors.toList());
- // empty, means all write result is null, which is a empty commit,
- // return null to skip the empty commit
- if (committableWriteResults.isEmpty()) {
+ // all buckets were empty — nothing to commit to the lake
+ if (nonEmptyResults.isEmpty()) {
LOG.info(
"Commit tiering write results is empty for table {}, table
path {}",
tableId,
tablePath);
- return null;
+ return new CommitResult(null, null);
Review Comment:
The docstring says "stats are `TieringStats.UNKNOWN`" for empty commits, but
the code passes `null` as the stats argument. While `FinishedTieringEvent` and
`TieringFinishInfo` defensively convert `null` to `TieringStats.UNKNOWN`, it
would be cleaner and consistent with the documented contract to pass
`TieringStats.UNKNOWN` here instead of `null`.
--
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]
