zhaomin1423 commented on code in PR #2365:
URL: https://github.com/apache/fluss/pull/2365#discussion_r2702014078
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java:
##########
@@ -224,6 +224,48 @@ public void removeLakeTable(long tableId) {
});
}
+ /**
+ * Update the lake freshness for a table. This method should be called
when the table's datalake
+ * freshness property is changed via ALTER TABLE.
+ *
+ * @param tableId the table id
+ * @param newFreshnessMs the new freshness interval in milliseconds
+ */
+ public void updateTableLakeFreshness(long tableId, long newFreshnessMs) {
+ inLock(
+ lock,
+ () -> {
+ Long currentFreshness = tableLakeFreshness.get(tableId);
+ if (currentFreshness == null) {
+ // the table is not a lake table or has been dropped,
skip update
+ LOG.warn(
+ "Cannot update lake freshness for table {} as
it's not tracked by lake tiering manager.",
+ tableId);
+ return;
+ }
+
+ if (currentFreshness.equals(newFreshnessMs)) {
+ // no change, skip update
+ return;
+ }
+
+ tableLakeFreshness.put(tableId, newFreshnessMs);
+ LOG.info(
+ "Updated lake freshness for table {} from {} ms to
{} ms.",
+ tableId,
+ currentFreshness,
+ newFreshnessMs);
+
+ // If the table is in Scheduled state, we need to
reschedule it with the new
+ // freshness
+ TieringState currentState = tieringStates.get(tableId);
+ if (currentState == TieringState.Scheduled) {
+ // Reschedule the table tiering with the new freshness
interval
Review Comment:
Thanks, updated
--
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]
