klion26 commented on code in PR #4038:
URL: https://github.com/apache/amoro/pull/4038#discussion_r2781883719
##########
amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/TableRuntimeRefreshExecutor.java:
##########
@@ -122,16 +151,148 @@ public void execute(TableRuntime tableRuntime) {
AmoroTable<?> table = loadTable(tableRuntime);
defaultTableRuntime.refresh(table);
MixedTable mixedTable = (MixedTable) table.originalTable();
+ // Check if there is any optimizing demand now.
+ boolean hasOptimizingDemand = false;
if ((mixedTable.isKeyedTable()
&& (lastOptimizedSnapshotId !=
defaultTableRuntime.getCurrentSnapshotId()
|| lastOptimizedChangeSnapshotId
!= defaultTableRuntime.getCurrentChangeSnapshotId()))
|| (mixedTable.isUnkeyedTable()
&& lastOptimizedSnapshotId !=
defaultTableRuntime.getCurrentSnapshotId())) {
- tryEvaluatingPendingInput(defaultTableRuntime, mixedTable);
+ hasOptimizingDemand = tryEvaluatingPendingInput(defaultTableRuntime,
mixedTable);
+ } else {
+ logger.debug("{} optimizing is not necessary",
defaultTableRuntime.getTableIdentifier());
+ }
+
+ // Update adaptive interval according to evaluated result.
+ if
(defaultTableRuntime.getOptimizingConfig().isRefreshTableAdaptiveEnabled()) {
+
defaultTableRuntime.setLatestEvaluatedNeedOptimizing(hasOptimizingDemand);
+ long newInterval = getAdaptiveExecutingInterval(defaultTableRuntime);
+ defaultTableRuntime.setLatestRefreshInterval(newInterval);
}
} catch (Throwable throwable) {
logger.error("Refreshing table {} failed.",
tableRuntime.getTableIdentifier(), throwable);
}
}
+
+ /**
+ * Calculate adaptive execution interval based on table optimization status.
+ *
+ * <p>Uses AIMD (Additive Increase Multiplicative Decrease) algorithm
inspired by TCP congestion
+ * control:
+ *
+ * <ul>
+ * <li>If table does not need to be optimized: additive increase -
gradually extend interval to
+ * reduce resource consumption
+ * <li>If table needs optimization: multiplicative decrease - rapidly
reduce interval for quick
+ * response
+ * </ul>
+ *
+ * <p>Interval is bounded by [interval_min, interval_max] and kept in memory
only (resets to
+ * interval_min on restart).
+ *
+ * @param tableRuntime The table runtime information containing current
status and configuration
+ * @return The next execution interval in milliseconds
+ */
+ @VisibleForTesting
+ public long getAdaptiveExecutingInterval(DefaultTableRuntime tableRuntime) {
+ final long minInterval = interval;
+ final long maxInterval =
+
tableRuntime.getOptimizingConfig().getRefreshTableAdaptiveMaxIntervalMs();
+
+ if (maxInterval <= minInterval) {
+ tableRuntime
+ .getOptimizingConfig()
+
.setRefreshTableAdaptiveMaxIntervalMs(AmoroServiceConstants.INVALID_TIME);
Review Comment:
Currently, we'll disable the adaptive refresh if `maxInterval <=
minInterval`, but we call the function `getAdaptiveExecutingInterval` in
`if(defaultTableRuntime.getOptimizingConfig().isRefreshTableAdaptiveEnabled())`
block,
The logic here seems strange.
--
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]
