You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@amoro.apache.org by "zhangwl9 (via GitHub)" <gi...@apache.org> on 2026/01/14 06:46:44 UTC
[PR] [AMORO-4037] Add support for dynamic refresh interval for table metadata refreshing [amoro]
zhangwl9 opened a new pull request, #4038:
URL: https://github.com/apache/amoro/pull/4038
…g in the TableRuntimeRefreshExecutor.
<!--
Thanks for sending a pull request!
Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://amoro.apache.org/how-to-contribute/
2. If the PR is related to an issue in https://github.com/apache/amoro/issues, add '[AMORO-XXXX]' in your PR title, e.g., '[AMORO-XXXX] Your PR title ...'.
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][AMORO-XXXX] Your PR title ...'.
-->
## Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you add a feature, you can talk about its use case.
2. If you fix a bug, you can clarify why it is a bug.
3. Use Fix/Resolve/Close #{ISSUE_NUMBER} to link this PR to its related issue
-->
Close #4037.
## Brief change log
<!--
Clearly describe the changes made in modules, classes, methods, etc.
-->
- The new feature involves two methods of "TableRuntimeRefreshExecutor".
- #getNextExecutingTime()
- When dynamic refresh interval is enabled，retrieving the latest refresh interval from "DefaultTableRuntime" to determine the next execution time for "TableRuntimeRefreshExecutor#execute"
- #execute()
- Depending on the evaluated results, determine whether the table requires optimization.
- When the evaluation table requires no optimization, extend the interval for its next refresh (incrementally increasing the interval by a fixed step).
- When evaluating table requiring optimization, shorten its next refresh interval (halving the interval each time).
- dynamic refresh interval in [ refresh-tables.interval(1 min)，max_interval(1 hour)]
## How was this patch tested?
- [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible
- [ ] Add screenshots for manual tests if appropriate
- [x] Run test locally before making a pull request
## Documentation
- Does this pull request introduce a new feature? (yes / no)
- If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-4037] Add support for dynamic refresh interval for table metadata refreshing in TableRuntimeRefreshExecutor [amoro]
Posted by "klion26 (via GitHub)" <gi...@apache.org>.
klion26 commented on code in PR #4038:
URL: https://github.com/apache/amoro/pull/4038#discussion_r2763432818
##########
amoro-common/src/main/java/org/apache/amoro/config/OptimizingConfig.java:
##########
@@ -318,6 +324,34 @@ public OptimizingConfig setEvaluationMseTolerance(long evaluationMseTolerance) {
return this;
}
+ public boolean isRefreshTableAdaptiveEnabled(long minInterval) {
Review Comment:
Why do we need to pass in this parameter? will compare the `refreshTableAdaptiveMaxIntervalMs` to a default value in the current class be better?
The logic here seems is different than the doc below `The maximum time interval in milliseconds to refresh table metadata. 0 means disable adaptive refresh.`?
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-4037] Add support for dynamic refresh interval for table metadata refreshing in TableRuntimeRefreshExecutor [amoro]
Posted by "klion26 (via GitHub)" <gi...@apache.org>.
klion26 commented on code in PR #4038:
URL: https://github.com/apache/amoro/pull/4038#discussion_r2731561968
##########
amoro-format-iceberg/src/main/java/org/apache/amoro/table/TableProperties.java:
##########
@@ -160,6 +160,18 @@ private TableProperties() {}
public static final String SNAPSHOT_MIN_COUNT = "snapshot.keep.min-count";
public static final int SNAPSHOT_MIN_COUNT_DEFAULT = 1;
+ public static final String SELF_OPTIMIZING_REFRESH_TABLE_ADAPTIVE_ENABLED =
+ "self-optimizing.refresh-table.adaptive.enabled";
Review Comment:
If we need to set `self-optimizing.refresh-table.adaptive.enabled` and `self-optimizing.refresh-table.adaptive.max-interval`, how about using only `self-optimizing.refresh-table.adaptive.max-interval`? Some value (like -1 or zero) means that disable it, and some other value means that to enable it.
##########
amoro-format-iceberg/src/main/java/org/apache/amoro/table/TableProperties.java:
##########
@@ -160,6 +160,18 @@ private TableProperties() {}
public static final String SNAPSHOT_MIN_COUNT = "snapshot.keep.min-count";
public static final int SNAPSHOT_MIN_COUNT_DEFAULT = 1;
+ public static final String SELF_OPTIMIZING_REFRESH_TABLE_ADAPTIVE_ENABLED =
+ "self-optimizing.refresh-table.adaptive.enabled";
+ public static final boolean SELF_OPTIMIZING_REFRESH_TABLE_ADAPTIVE_ENABLED_DEFAULT = false;
+ public static final String SELF_OPTIMIZING_REFRESH_TABLE_ADAPTIVE_MAX_INTERVAL =
+ "self-optimizing.refresh-table.adaptive.max-interval";
+ public static final long SELF_OPTIMIZING_REFRESH_TABLE_ADAPTIVE_MAX_INTERVAL_DEFAULT =
Review Comment:
How about add the time unit in the name, `SELF_OPTIMIZING_REFRESH_TABLE_ADAPTIVE_MAX_INTERVAL_DEFAULT` -> `SELF_OPTIMIZING_REFRESH_TABLE_ADAPTIVE_MAX_INTERVAL_MS_DEFAULT`
##########
amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/TableRuntimeRefreshExecutor.java:
##########
@@ -129,9 +148,124 @@ public void execute(TableRuntime tableRuntime) {
|| (mixedTable.isUnkeyedTable()
&& lastOptimizedSnapshotId != defaultTableRuntime.getCurrentSnapshotId())) {
tryEvaluatingPendingInput(defaultTableRuntime, mixedTable);
+ } else {
+ logger.debug("{} optimizing is not necessary", defaultTableRuntime.getTableIdentifier());
+ defaultTableRuntime.setLatestEvaluatedNeedOptimizing(false);
+ }
+
+ // Update adaptive interval according to evaluated result.
+ if (defaultTableRuntime.getOptimizingConfig().getRefreshTableAdaptiveEnabled()) {
+ long newInterval = getAdaptiveExecutingInterval(defaultTableRuntime);
+ defaultTableRuntime.setLatestRefreshInterval(newInterval);
Review Comment:
Maybe we can call `tableRuntime.setLatestEvaluatedNeedOptimizing(true);` and `defaultTableRuntime.setLatestRefreshInterval(newInterval);` both in this if block, so that we wouldn't miss either
##########
amoro-ams/src/test/java/org/apache/amoro/server/scheduler/inline/TestTableRuntimeRefreshExecutor.java:
##########
@@ -0,0 +1,254 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.amoro.server.scheduler.inline;
+
+import static org.mockito.Mockito.verify;
+
+import org.apache.amoro.BasicTableTestHelper;
+import org.apache.amoro.TableFormat;
+import org.apache.amoro.TableTestHelper;
+import org.apache.amoro.TestedCatalogs;
+import org.apache.amoro.catalog.CatalogTestHelper;
+import org.apache.amoro.config.OptimizingConfig;
+import org.apache.amoro.hive.catalog.HiveCatalogTestHelper;
+import org.apache.amoro.hive.catalog.HiveTableTestHelper;
+import org.apache.amoro.server.table.AMSTableTestBase;
+import org.apache.amoro.server.table.DefaultTableRuntime;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.mockito.Mockito;
+
+@RunWith(Parameterized.class)
+public class TestTableRuntimeRefreshExecutor extends AMSTableTestBase {
+ @Parameterized.Parameters(name = "{0}, {1}")
+ public static Object[] parameters() {
+ return new Object[][] {
+ {
+ TestedCatalogs.hadoopCatalog(TableFormat.MIXED_ICEBERG),
+ new BasicTableTestHelper(true, true)
+ },
+ {
+ TestedCatalogs.hadoopCatalog(TableFormat.MIXED_ICEBERG),
+ new BasicTableTestHelper(false, true)
+ },
+ {
+ TestedCatalogs.hadoopCatalog(TableFormat.MIXED_ICEBERG),
+ new BasicTableTestHelper(false, false)
+ },
+ {
+ TestedCatalogs.hadoopCatalog(TableFormat.MIXED_ICEBERG),
+ new BasicTableTestHelper(true, false)
+ },
+ {
+ new HiveCatalogTestHelper(TableFormat.ICEBERG, TEST_HMS.getHiveConf()),
+ new HiveTableTestHelper(true, true)
+ },
+ {
+ new HiveCatalogTestHelper(TableFormat.ICEBERG, TEST_HMS.getHiveConf()),
+ new HiveTableTestHelper(false, true)
+ },
+ {
+ new HiveCatalogTestHelper(TableFormat.ICEBERG, TEST_HMS.getHiveConf()),
+ new HiveTableTestHelper(false, false)
+ },
+ {
+ new HiveCatalogTestHelper(TableFormat.ICEBERG, TEST_HMS.getHiveConf()),
+ new HiveTableTestHelper(true, false)
+ },
+ };
+ }
+
+ public TestTableRuntimeRefreshExecutor(
+ CatalogTestHelper catalogTestHelper, TableTestHelper tableTestHelper) {
+ super(catalogTestHelper, tableTestHelper);
+ }
+
+ private static final long INTERVAL = 60000L; // 1 minute
+ private static final long MAX_INTERVAL = 300000L; // 5 minutes
+ private static final long STEP = 30000L; // 30s
+ private static final int MAX_PENDING_PARTITIONS = 1;
+ private static final int MINOR_LEAST_INTERVAL = 3600000; // 1h
+ private static final long INITIAL_SNAPSHOTID = 0L;
+
+ // Create mock DefaultTableRuntime for adaptive interval tests
+ private DefaultTableRuntime getMockTableRuntimeWithAdaptiveInterval(
+ DefaultTableRuntime tableRuntime,
+ boolean needOptimizing,
+ boolean adaptiveEnabled,
+ long currentInterval,
+ long step) {
+ DefaultTableRuntime mockTableRuntime = Mockito.mock(DefaultTableRuntime.class);
+ Mockito.when(mockTableRuntime.getTableIdentifier())
+ .thenReturn(tableRuntime.getTableIdentifier());
+ Mockito.when(mockTableRuntime.getLastOptimizedSnapshotId())
+ .thenReturn(tableRuntime.getLastOptimizedSnapshotId());
+ Mockito.when(mockTableRuntime.getLastOptimizedChangeSnapshotId())
+ .thenReturn(tableRuntime.getLastOptimizedChangeSnapshotId());
+ Mockito.when(mockTableRuntime.getCurrentSnapshotId()).thenReturn(INITIAL_SNAPSHOTID);
+ Mockito.when(mockTableRuntime.getCurrentChangeSnapshotId()).thenReturn(INITIAL_SNAPSHOTID);
+ Mockito.when(mockTableRuntime.getLatestRefreshInterval()).thenReturn(currentInterval);
+ Mockito.when(mockTableRuntime.getLatestEvaluatedNeedOptimizing()).thenReturn(needOptimizing);
+
+ OptimizingConfig mockOptimizingConfig = Mockito.mock(OptimizingConfig.class);
+ Mockito.when(mockOptimizingConfig.getRefreshTableAdaptiveEnabled()).thenReturn(adaptiveEnabled);
+ Mockito.when(mockOptimizingConfig.getRefreshTableAdaptiveMaxInterval())
+ .thenReturn(MAX_INTERVAL);
+ Mockito.when(mockOptimizingConfig.getMinorLeastInterval()).thenReturn(MINOR_LEAST_INTERVAL);
+ Mockito.when(mockOptimizingConfig.getRefreshTableAdaptiveIncreaseStep()).thenReturn(step);
+ Mockito.when(mockTableRuntime.getOptimizingConfig()).thenReturn(mockOptimizingConfig);
+
+ return mockTableRuntime;
+ }
+
+ // Overloaded method with default step factor
+ private DefaultTableRuntime getMockTableRuntimeWithAdaptiveInterval(
+ DefaultTableRuntime tableRuntime,
+ boolean needOptimizing,
+ boolean adaptiveEnabled,
+ long currentInterval) {
+ return getMockTableRuntimeWithAdaptiveInterval(
+ tableRuntime, needOptimizing, adaptiveEnabled, currentInterval, STEP);
+ }
+
+ @Test
+ public void testAdaptiveIntervalScenarios() {
+ createDatabase();
+ createTable();
+
+ DefaultTableRuntime tableRuntime =
+ (DefaultTableRuntime) tableService().getRuntime(serverTableIdentifier().getId());
+ TableRuntimeRefreshExecutor executor =
+ new TableRuntimeRefreshExecutor(tableService(), 1, INTERVAL, MAX_PENDING_PARTITIONS);
+
+ // Test healthy table (not need optimizing) - interval should increase
+ DefaultTableRuntime mockTableRuntime =
+ getMockTableRuntimeWithAdaptiveInterval(tableRuntime, false, true, INTERVAL);
+
+ // Initial interval is INTERVAL, should increase by 30000
+ long expectedInterval = INTERVAL + STEP;
+ executor.execute(mockTableRuntime);
+
+ // Verify that setLatestRefreshInterval was called with the expected value
+ verify(mockTableRuntime, Mockito.times(1)).setLatestRefreshInterval(expectedInterval);
Review Comment:
Instead using `verify` here to validate, is there any way we can validate the value of the `true interval` after `executor.execute` in L147.
The `true interval` is the value keeped in the table runtime.
##########
docs/user-guides/configurations.md:
##########
@@ -43,25 +43,28 @@ modified through [Alter Table](../using-tables/#modify-table) operations.
Self-optimizing configurations are applicable to both Iceberg Format and Mixed streaming Format.
-| Key | Default | Description |
-|-----------------------------------------------|------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| self-optimizing.enabled | true | Enables Self-optimizing |
-| self-optimizing.allow-partial-commit | false | Whether to allow partial commit when self-optimizing fails or process is cancelled |
-| self-optimizing.group | default | Optimizer group for Self-optimizing |
-| self-optimizing.quota | 0.5 | Quota for Self-optimizing, indicating the optimizer resources the table can take up |
-| self-optimizing.execute.num-retries | 5 | Number of retries after failure of Self-optimizing |
-| self-optimizing.target-size | 134217728(128MB) | Target size for Self-optimizing |
-| self-optimizing.max-file-count | 10000 | Maximum number of files processed by a Self-optimizing process |
-| self-optimizing.max-task-size-bytes | 134217728(128MB) | Maximum file size bytes in a single task for splitting tasks |
-| self-optimizing.fragment-ratio | 8 | The fragment file size threshold. We could divide self-optimizing.target-size by this ratio to get the actual fragment file size |
-| self-optimizing.min-target-size-ratio | 0.75 | The undersized segment file size threshold. Segment files under this threshold will be considered for rewriting |
-| self-optimizing.minor.trigger.file-count | 12 | The minimum number of files to trigger minor optimizing is determined by the sum of fragment file count and equality delete file count |
-| self-optimizing.minor.trigger.interval | 3600000(1 hour) | The time interval in milliseconds to trigger minor optimizing |
-| self-optimizing.major.trigger.duplicate-ratio | 0.1 | The ratio of duplicate data of segment files to trigger major optimizing |
-| self-optimizing.full.trigger.interval | -1(closed) | The time interval in milliseconds to trigger full optimizing |
-| self-optimizing.full.rewrite-all-files | true | Whether full optimizing rewrites all files or skips files that do not need to be optimized |
-| self-optimizing.min-plan-interval | 60000 | The minimum time interval between two self-optimizing planning action |
-| self-optimizing.filter | NULL | Filter conditions for self-optimizing, using SQL conditional expressions, without supporting any functions. For the timestamp column condition, the ISO date-time formatter must be used. For example: op_time > '2007-12-03T10:15:30'. |
+| Key | Default | Description |
+|------------------------------------------------------|------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| self-optimizing.enabled | true | Enables Self-optimizing |
+| self-optimizing.allow-partial-commit | false | Whether to allow partial commit when self-optimizing fails or process is cancelled |
+| self-optimizing.group | default | Optimizer group for Self-optimizing |
+| self-optimizing.quota | 0.5 | Quota for Self-optimizing, indicating the optimizer resources the table can take up |
+| self-optimizing.execute.num-retries | 5 | Number of retries after failure of Self-optimizing |
+| self-optimizing.target-size | 134217728(128MB) | Target size for Self-optimizing |
+| self-optimizing.max-file-count | 10000 | Maximum number of files processed by a Self-optimizing process |
+| self-optimizing.max-task-size-bytes | 134217728(128MB) | Maximum file size bytes in a single task for splitting tasks |
+| self-optimizing.fragment-ratio | 8 | The fragment file size threshold. We could divide self-optimizing.target-size by this ratio to get the actual fragment file size |
+| self-optimizing.min-target-size-ratio | 0.75 | The undersized segment file size threshold. Segment files under this threshold will be considered for rewriting |
+| self-optimizing.minor.trigger.file-count | 12 | The minimum number of files to trigger minor optimizing is determined by the sum of fragment file count and equality delete file count |
+| self-optimizing.minor.trigger.interval | 3600000(1 hour) | The time interval in milliseconds to trigger minor optimizing |
+| self-optimizing.major.trigger.duplicate-ratio | 0.1 | The ratio of duplicate data of segment files to trigger major optimizing |
+| self-optimizing.full.trigger.interval | -1(closed) | The time interval in milliseconds to trigger full optimizing |
+| self-optimizing.full.rewrite-all-files | true | Whether full optimizing rewrites all files or skips files that do not need to be optimized |
+| self-optimizing.min-plan-interval | 60000 | The minimum time interval between two self-optimizing planning action |
+| self-optimizing.filter | NULL | Filter conditions for self-optimizing, using SQL conditional expressions, without supporting any functions. For the timestamp column condition, the ISO date-time formatter must be used. For example: op_time > '2007-12-03T10:15:30'. |
+| self-optimizing.refresh-table.adaptive.enabled | false | Whether to enable adaptive refresh interval for refreshing table metadata |
+| self-optimizing.refresh-table.adaptive.max-interval | 3600000(1 hour) | The maximum time interval in milliseconds to refresh table metadata |
Review Comment:
Maybe we can set the max-interval to an unset value, so that we can keep the behavior same as before the current pr.
##########
amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/TableRuntimeRefreshExecutor.java:
##########
@@ -56,19 +56,29 @@ protected boolean enabled(TableRuntime tableRuntime) {
@Override
protected long getNextExecutingTime(TableRuntime tableRuntime) {
DefaultTableRuntime defaultTableRuntime = (DefaultTableRuntime) tableRuntime;
+
+ if (defaultTableRuntime.getOptimizingConfig().getRefreshTableAdaptiveEnabled()) {
+ long newInterval = defaultTableRuntime.getLatestRefreshInterval();
+ if (newInterval > 0) {
+ return newInterval;
+ }
+ }
+
return Math.min(
defaultTableRuntime.getOptimizingConfig().getMinorLeastInterval() * 4L / 5, interval);
}
private void tryEvaluatingPendingInput(DefaultTableRuntime tableRuntime, MixedTable table) {
// only evaluate pending input when optimizing is enabled and in idle state
OptimizingConfig optimizingConfig = tableRuntime.getOptimizingConfig();
- if (optimizingConfig.isEnabled()
- && tableRuntime.getOptimizingStatus().equals(OptimizingStatus.IDLE)) {
+ boolean optimizingEnabled = optimizingConfig.isEnabled();
+ if (optimizingEnabled && tableRuntime.getOptimizingStatus().equals(OptimizingStatus.IDLE)) {
if (optimizingConfig.isMetadataBasedTriggerEnabled()
&& !MetadataBasedEvaluationEvent.isEvaluatingNecessary(
optimizingConfig, table, tableRuntime.getLastPlanTime())) {
+ tableRuntime.setLatestEvaluatedNeedOptimizing(false);
Review Comment:
Is it better that we change the return value of `tryEvaluatingPendingInput` to `boolean` -- indicate whether we need to optimize after evaluating?
So that we can call `tableRuntime.setLatestEvaluteNeedOptimizing` in one place in `execute` below,
Please let me what do you think about it.
##########
amoro-ams/src/test/java/org/apache/amoro/server/scheduler/inline/TestTableRuntimeRefreshExecutor.java:
##########
@@ -0,0 +1,254 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.amoro.server.scheduler.inline;
+
+import static org.mockito.Mockito.verify;
+
+import org.apache.amoro.BasicTableTestHelper;
+import org.apache.amoro.TableFormat;
+import org.apache.amoro.TableTestHelper;
+import org.apache.amoro.TestedCatalogs;
+import org.apache.amoro.catalog.CatalogTestHelper;
+import org.apache.amoro.config.OptimizingConfig;
+import org.apache.amoro.hive.catalog.HiveCatalogTestHelper;
+import org.apache.amoro.hive.catalog.HiveTableTestHelper;
+import org.apache.amoro.server.table.AMSTableTestBase;
+import org.apache.amoro.server.table.DefaultTableRuntime;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.mockito.Mockito;
+
+@RunWith(Parameterized.class)
+public class TestTableRuntimeRefreshExecutor extends AMSTableTestBase {
+ @Parameterized.Parameters(name = "{0}, {1}")
+ public static Object[] parameters() {
+ return new Object[][] {
+ {
+ TestedCatalogs.hadoopCatalog(TableFormat.MIXED_ICEBERG),
+ new BasicTableTestHelper(true, true)
+ },
+ {
+ TestedCatalogs.hadoopCatalog(TableFormat.MIXED_ICEBERG),
+ new BasicTableTestHelper(false, true)
+ },
+ {
+ TestedCatalogs.hadoopCatalog(TableFormat.MIXED_ICEBERG),
+ new BasicTableTestHelper(false, false)
+ },
+ {
+ TestedCatalogs.hadoopCatalog(TableFormat.MIXED_ICEBERG),
+ new BasicTableTestHelper(true, false)
+ },
+ {
+ new HiveCatalogTestHelper(TableFormat.ICEBERG, TEST_HMS.getHiveConf()),
+ new HiveTableTestHelper(true, true)
+ },
+ {
+ new HiveCatalogTestHelper(TableFormat.ICEBERG, TEST_HMS.getHiveConf()),
+ new HiveTableTestHelper(false, true)
+ },
+ {
+ new HiveCatalogTestHelper(TableFormat.ICEBERG, TEST_HMS.getHiveConf()),
+ new HiveTableTestHelper(false, false)
+ },
+ {
+ new HiveCatalogTestHelper(TableFormat.ICEBERG, TEST_HMS.getHiveConf()),
+ new HiveTableTestHelper(true, false)
+ },
+ };
+ }
+
+ public TestTableRuntimeRefreshExecutor(
+ CatalogTestHelper catalogTestHelper, TableTestHelper tableTestHelper) {
+ super(catalogTestHelper, tableTestHelper);
+ }
+
+ private static final long INTERVAL = 60000L; // 1 minute
+ private static final long MAX_INTERVAL = 300000L; // 5 minutes
+ private static final long STEP = 30000L; // 30s
+ private static final int MAX_PENDING_PARTITIONS = 1;
+ private static final int MINOR_LEAST_INTERVAL = 3600000; // 1h
+ private static final long INITIAL_SNAPSHOTID = 0L;
+
+ // Create mock DefaultTableRuntime for adaptive interval tests
+ private DefaultTableRuntime getMockTableRuntimeWithAdaptiveInterval(
+ DefaultTableRuntime tableRuntime,
+ boolean needOptimizing,
+ boolean adaptiveEnabled,
+ long currentInterval,
+ long step) {
+ DefaultTableRuntime mockTableRuntime = Mockito.mock(DefaultTableRuntime.class);
+ Mockito.when(mockTableRuntime.getTableIdentifier())
+ .thenReturn(tableRuntime.getTableIdentifier());
+ Mockito.when(mockTableRuntime.getLastOptimizedSnapshotId())
+ .thenReturn(tableRuntime.getLastOptimizedSnapshotId());
+ Mockito.when(mockTableRuntime.getLastOptimizedChangeSnapshotId())
+ .thenReturn(tableRuntime.getLastOptimizedChangeSnapshotId());
+ Mockito.when(mockTableRuntime.getCurrentSnapshotId()).thenReturn(INITIAL_SNAPSHOTID);
+ Mockito.when(mockTableRuntime.getCurrentChangeSnapshotId()).thenReturn(INITIAL_SNAPSHOTID);
+ Mockito.when(mockTableRuntime.getLatestRefreshInterval()).thenReturn(currentInterval);
+ Mockito.when(mockTableRuntime.getLatestEvaluatedNeedOptimizing()).thenReturn(needOptimizing);
+
+ OptimizingConfig mockOptimizingConfig = Mockito.mock(OptimizingConfig.class);
+ Mockito.when(mockOptimizingConfig.getRefreshTableAdaptiveEnabled()).thenReturn(adaptiveEnabled);
+ Mockito.when(mockOptimizingConfig.getRefreshTableAdaptiveMaxInterval())
+ .thenReturn(MAX_INTERVAL);
+ Mockito.when(mockOptimizingConfig.getMinorLeastInterval()).thenReturn(MINOR_LEAST_INTERVAL);
+ Mockito.when(mockOptimizingConfig.getRefreshTableAdaptiveIncreaseStep()).thenReturn(step);
+ Mockito.when(mockTableRuntime.getOptimizingConfig()).thenReturn(mockOptimizingConfig);
+
+ return mockTableRuntime;
+ }
+
+ // Overloaded method with default step factor
+ private DefaultTableRuntime getMockTableRuntimeWithAdaptiveInterval(
+ DefaultTableRuntime tableRuntime,
+ boolean needOptimizing,
+ boolean adaptiveEnabled,
+ long currentInterval) {
+ return getMockTableRuntimeWithAdaptiveInterval(
+ tableRuntime, needOptimizing, adaptiveEnabled, currentInterval, STEP);
+ }
+
+ @Test
+ public void testAdaptiveIntervalScenarios() {
+ createDatabase();
+ createTable();
+
+ DefaultTableRuntime tableRuntime =
+ (DefaultTableRuntime) tableService().getRuntime(serverTableIdentifier().getId());
+ TableRuntimeRefreshExecutor executor =
+ new TableRuntimeRefreshExecutor(tableService(), 1, INTERVAL, MAX_PENDING_PARTITIONS);
+
+ // Test healthy table (not need optimizing) - interval should increase
+ DefaultTableRuntime mockTableRuntime =
+ getMockTableRuntimeWithAdaptiveInterval(tableRuntime, false, true, INTERVAL);
+
+ // Initial interval is INTERVAL, should increase by 30000
+ long expectedInterval = INTERVAL + STEP;
+ executor.execute(mockTableRuntime);
+
+ // Verify that setLatestRefreshInterval was called with the expected value
+ verify(mockTableRuntime, Mockito.times(1)).setLatestRefreshInterval(expectedInterval);
+
+ // Test unhealthy table (need optimizing) - interval should decrease
+ mockTableRuntime =
+ getMockTableRuntimeWithAdaptiveInterval(tableRuntime, true, true, INTERVAL * 2);
Review Comment:
Is there a way to reuse this `tableRuntime` so we can observe the change in `interval` before and after two consecutive `execute` calls? Reusing a single `tableRuntime` function would also be more suitable for production scenarios.
##########
amoro-format-iceberg/src/main/java/org/apache/amoro/table/TableProperties.java:
##########
@@ -160,6 +160,18 @@ private TableProperties() {}
public static final String SNAPSHOT_MIN_COUNT = "snapshot.keep.min-count";
public static final int SNAPSHOT_MIN_COUNT_DEFAULT = 1;
+ public static final String SELF_OPTIMIZING_REFRESH_TABLE_ADAPTIVE_ENABLED =
+ "self-optimizing.refresh-table.adaptive.enabled";
+ public static final boolean SELF_OPTIMIZING_REFRESH_TABLE_ADAPTIVE_ENABLED_DEFAULT = false;
+ public static final String SELF_OPTIMIZING_REFRESH_TABLE_ADAPTIVE_MAX_INTERVAL =
+ "self-optimizing.refresh-table.adaptive.max-interval";
+ public static final long SELF_OPTIMIZING_REFRESH_TABLE_ADAPTIVE_MAX_INTERVAL_DEFAULT =
Review Comment:
Same as the other places
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-4037] Add support for dynamic refresh interval for table metadata refreshing in TableRuntimeRefreshExecutor [amoro]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #4038:
URL: https://github.com/apache/amoro/pull/4038#issuecomment-3839966827
## [Codecov](https://app.codecov.io/gh/apache/amoro/pull/4038?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `61.79775%` with `34 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 28.68%. Comparing base ([`5427715`](https://app.codecov.io/gh/apache/amoro/commit/5427715fe1e793a231a05315c06e49af5bf8585c?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`fbbd84f`](https://app.codecov.io/gh/apache/amoro/commit/fbbd84f179e2e4246e12f65f74e9dccfd3f396ec?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 1 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/amoro/pull/4038?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...java/org/apache/amoro/config/OptimizingConfig.java](https://app.codecov.io/gh/apache/amoro/pull/4038?src=pr&el=tree&filepath=amoro-common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Famoro%2Fconfig%2FOptimizingConfig.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-YW1vcm8tY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9hbW9yby9jb25maWcvT3B0aW1pemluZ0NvbmZpZy5qYXZh) | 0.00% | [18 Missing :warning: ](https://app.codecov.io/gh/apache/amoro/pull/4038?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [.../scheduler/inline/TableRuntimeRefreshExecutor.java](https://app.codecov.io/gh/apache/amoro/pull/4038?src=pr&el=tree&filepath=amoro-ams%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Famoro%2Fserver%2Fscheduler%2Finline%2FTableRuntimeRefreshExecutor.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-YW1vcm8tYW1zL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9hbW9yby9zZXJ2ZXIvc2NoZWR1bGVyL2lubGluZS9UYWJsZVJ1bnRpbWVSZWZyZXNoRXhlY3V0b3IuamF2YQ==) | 72.88% | [12 Missing and 4 partials :warning: ](https://app.codecov.io/gh/apache/amoro/pull/4038?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #4038 +/- ##
============================================
- Coverage 28.70% 28.68% -0.02%
- Complexity 3944 3951 +7
============================================
Files 656 654 -2
Lines 52382 52385 +3
Branches 6642 6630 -12
============================================
- Hits 15035 15026 -9
- Misses 36249 36265 +16
+ Partials 1098 1094 -4
```
| [Flag](https://app.codecov.io/gh/apache/amoro/pull/4038/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [core](https://app.codecov.io/gh/apache/amoro/pull/4038/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `28.68% <61.79%> (-0.02%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/amoro/pull/4038?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-4037] Add support for dynamic refresh interval for table metadata refreshing in TableRuntimeRefreshExecutor [amoro]
Posted by "zhangwl9 (via GitHub)" <gi...@apache.org>.
zhangwl9 commented on PR #4038:
URL: https://github.com/apache/amoro/pull/4038#issuecomment-3757740459
@xxubai Could you help me review it when you are free? Thanks!
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-4037] Add support for dynamic refresh interval for table metadata refreshing in TableRuntimeRefreshExecutor [amoro]
Posted by "klion26 (via GitHub)" <gi...@apache.org>.
klion26 merged PR #4038:
URL: https://github.com/apache/amoro/pull/4038
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-4037] Add support for dynamic refresh interval for table metadata refreshing in TableRuntimeRefreshExecutor [amoro]
Posted by "klion26 (via GitHub)" <gi...@apache.org>.
klion26 commented on code in PR #4038:
URL: https://github.com/apache/amoro/pull/4038#discussion_r2745355589
##########
amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/TableRuntimeRefreshExecutor.java:
##########
@@ -122,16 +143,133 @@ public void execute(TableRuntime tableRuntime) {
AmoroTable<?> table = loadTable(tableRuntime);
defaultTableRuntime.refresh(table);
MixedTable mixedTable = (MixedTable) table.originalTable();
+ boolean needOptimizing = false;
if ((mixedTable.isKeyedTable()
&& (lastOptimizedSnapshotId != defaultTableRuntime.getCurrentSnapshotId()
|| lastOptimizedChangeSnapshotId
!= defaultTableRuntime.getCurrentChangeSnapshotId()))
|| (mixedTable.isUnkeyedTable()
&& lastOptimizedSnapshotId != defaultTableRuntime.getCurrentSnapshotId())) {
- tryEvaluatingPendingInput(defaultTableRuntime, mixedTable);
+ needOptimizing = tryEvaluatingPendingInput(defaultTableRuntime, mixedTable);
+ } else {
+ logger.debug("{} optimizing is not necessary", defaultTableRuntime.getTableIdentifier());
+ }
+
+ // Update adaptive interval according to evaluated result.
+ if (defaultTableRuntime.getOptimizingConfig().getRefreshTableAdaptiveMaxIntervalMs() > 0) {
Review Comment:
Maybe we can add a function in `OptimizingConfig` which return `true/false` show whethere we enabled this feature
##########
amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/TableRuntimeRefreshExecutor.java:
##########
@@ -88,7 +98,18 @@ private void tryEvaluatingPendingInput(DefaultTableRuntime tableRuntime, MixedTa
} else {
tableRuntime.optimizingNotNecessary();
}
+
tableRuntime.setTableSummary(evaluator.getPendingInput());
+ return evaluatorIsNecessary;
+ } else if (!optimizingEnabled) {
+ logger.debug(
+ "{} optimizing is not enabled, skip evaluating pending input",
+ tableRuntime.getTableIdentifier());
+ return false;
+ } else {
+ logger.debug(
Review Comment:
From below, the name is `needOptimizing`, looks like whethere we need to do an opitmizing, but in this branch, we don't need to optimizing in current loop? as there is an ongoing optimizing process
##########
amoro-ams/src/test/java/org/apache/amoro/server/scheduler/inline/TestTableRuntimeRefreshExecutor.java:
##########
@@ -0,0 +1,204 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.amoro.server.scheduler.inline;
+
+import org.apache.amoro.BasicTableTestHelper;
+import org.apache.amoro.TableFormat;
+import org.apache.amoro.TableTestHelper;
+import org.apache.amoro.TestedCatalogs;
+import org.apache.amoro.catalog.CatalogTestHelper;
+import org.apache.amoro.config.OptimizingConfig;
+import org.apache.amoro.hive.catalog.HiveCatalogTestHelper;
+import org.apache.amoro.hive.catalog.HiveTableTestHelper;
+import org.apache.amoro.server.table.AMSTableTestBase;
+import org.apache.amoro.server.table.DefaultTableRuntime;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(Parameterized.class)
+public class TestTableRuntimeRefreshExecutor extends AMSTableTestBase {
+ @Parameterized.Parameters(name = "{0}, {1}")
+ public static Object[] parameters() {
+ return new Object[][] {
+ {
+ TestedCatalogs.hadoopCatalog(TableFormat.MIXED_ICEBERG),
+ new BasicTableTestHelper(true, true)
+ },
+ {
+ TestedCatalogs.hadoopCatalog(TableFormat.MIXED_ICEBERG),
+ new BasicTableTestHelper(false, true)
+ },
+ {
+ TestedCatalogs.hadoopCatalog(TableFormat.MIXED_ICEBERG),
+ new BasicTableTestHelper(false, false)
+ },
+ {
+ TestedCatalogs.hadoopCatalog(TableFormat.MIXED_ICEBERG),
+ new BasicTableTestHelper(true, false)
+ },
+ {
+ new HiveCatalogTestHelper(TableFormat.ICEBERG, TEST_HMS.getHiveConf()),
+ new HiveTableTestHelper(true, true)
+ },
+ {
+ new HiveCatalogTestHelper(TableFormat.ICEBERG, TEST_HMS.getHiveConf()),
+ new HiveTableTestHelper(false, true)
+ },
+ {
+ new HiveCatalogTestHelper(TableFormat.ICEBERG, TEST_HMS.getHiveConf()),
+ new HiveTableTestHelper(false, false)
+ },
+ {
+ new HiveCatalogTestHelper(TableFormat.ICEBERG, TEST_HMS.getHiveConf()),
+ new HiveTableTestHelper(true, false)
+ },
+ };
+ }
+
+ public TestTableRuntimeRefreshExecutor(
+ CatalogTestHelper catalogTestHelper, TableTestHelper tableTestHelper) {
+ super(catalogTestHelper, tableTestHelper);
+ }
+
+ private static final long INTERVAL = 60000L; // 1 minute
+ private static final long MAX_INTERVAL = 300000L; // 5 minutes
+ private static final long DEFAULT_MAX_INTERVAL = -1L;
+ private static final long STEP = 30000L; // 30s
+ private static final int MAX_PENDING_PARTITIONS = 1;
+ private static final int MINOR_LEAST_INTERVAL = 3600000; // 1h
+
+ // Create DefaultTableRuntime for adaptive interval tests
+ private DefaultTableRuntime buildTableRuntimeWithAdaptiveRefresh(
+ DefaultTableRuntime tableRuntime, boolean needOptimizing, long interval, long step) {
+ return buildTableRuntimeWithConfig(tableRuntime, needOptimizing, interval, MAX_INTERVAL, step);
+ }
+
+ private DefaultTableRuntime buildTableRuntimeWithAdaptiveMaxInterval(
+ DefaultTableRuntime tableRuntime, long interval, long maxInterval) {
+ return buildTableRuntimeWithConfig(tableRuntime, true, interval, maxInterval, STEP);
+ }
+
+ private DefaultTableRuntime buildTableRuntimeWithConfig(
+ DefaultTableRuntime tableRuntime,
+ boolean needOptimizing,
+ long interval,
+ long maxInterval,
+ long step) {
+ OptimizingConfig optimizingConfig = new OptimizingConfig();
+ optimizingConfig.setRefreshTableAdaptiveMaxIntervalMs(maxInterval);
+ optimizingConfig.setMinorLeastInterval(MINOR_LEAST_INTERVAL);
+ optimizingConfig.setRefreshTableAdaptiveIncreaseStepMs(step);
+
+ DefaultTableRuntime newRuntime =
+ new DefaultTableRuntime(tableRuntime.store()) {
+ @Override
+ public OptimizingConfig getOptimizingConfig() {
+ return optimizingConfig;
+ }
+ };
+ newRuntime.setLatestEvaluatedNeedOptimizing(needOptimizing);
+ newRuntime.setLatestRefreshInterval(interval);
+
+ return newRuntime;
+ }
+
+ @Test
+ public void testAdaptiveRefreshIntervalScenarios() {
+ createDatabase();
+ createTable();
+
+ DefaultTableRuntime tableRuntime =
+ (DefaultTableRuntime) tableService().getRuntime(serverTableIdentifier().getId());
+ TableRuntimeRefreshExecutor executor =
+ new TableRuntimeRefreshExecutor(tableService(), 1, INTERVAL, MAX_PENDING_PARTITIONS);
+
+ // Test healthy table (not need optimizing) - interval should increase
+ DefaultTableRuntime newTableRuntime =
+ buildTableRuntimeWithAdaptiveRefresh(tableRuntime, false, 0, STEP);
+ long adaptiveExecutingInterval = executor.getAdaptiveExecutingInterval(newTableRuntime);
+ long expectedInterval = INTERVAL + STEP;
+ Assert.assertEquals(expectedInterval, adaptiveExecutingInterval);
+
+ // Test minimum boundary - interval should not below INTERVAL
+ // The unhealthy table (need optimizing) - interval should decrease half
+ // The currentInterval is INTERVAL + STEP, the latest interval is INTERVAL not (INTERVAL + STEP)
+ // / 2
Review Comment:
Place `(INTERVAL + STEP) / 2` in the same line is better than 2 lines
##########
amoro-ams/src/main/java/org/apache/amoro/server/scheduler/inline/TableRuntimeRefreshExecutor.java:
##########
@@ -88,7 +98,18 @@ private void tryEvaluatingPendingInput(DefaultTableRuntime tableRuntime, MixedTa
} else {
tableRuntime.optimizingNotNecessary();
}
+
tableRuntime.setTableSummary(evaluator.getPendingInput());
+ return evaluatorIsNecessary;
+ } else if (!optimizingEnabled) {
+ logger.debug(
+ "{} optimizing is not enabled, skip evaluating pending input",
+ tableRuntime.getTableIdentifier());
+ return false;
+ } else {
+ logger.debug(
Review Comment:
Could we add some comments to describle the "semantics" of the return value, and why would this case would be `true`?
##########
docs/user-guides/configurations.md:
##########
@@ -43,25 +43,27 @@ modified through [Alter Table](../using-tables/#modify-table) operations.
Self-optimizing configurations are applicable to both Iceberg Format and Mixed streaming Format.
-| Key | Default | Description |
-|-----------------------------------------------|------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| self-optimizing.enabled | true | Enables Self-optimizing |
-| self-optimizing.allow-partial-commit | false | Whether to allow partial commit when self-optimizing fails or process is cancelled |
-| self-optimizing.group | default | Optimizer group for Self-optimizing |
-| self-optimizing.quota | 0.5 | Quota for Self-optimizing, indicating the optimizer resources the table can take up |
-| self-optimizing.execute.num-retries | 5 | Number of retries after failure of Self-optimizing |
-| self-optimizing.target-size | 134217728(128MB) | Target size for Self-optimizing |
-| self-optimizing.max-file-count | 10000 | Maximum number of files processed by a Self-optimizing process |
-| self-optimizing.max-task-size-bytes | 134217728(128MB) | Maximum file size bytes in a single task for splitting tasks |
-| self-optimizing.fragment-ratio | 8 | The fragment file size threshold. We could divide self-optimizing.target-size by this ratio to get the actual fragment file size |
-| self-optimizing.min-target-size-ratio | 0.75 | The undersized segment file size threshold. Segment files under this threshold will be considered for rewriting |
-| self-optimizing.minor.trigger.file-count | 12 | The minimum number of files to trigger minor optimizing is determined by the sum of fragment file count and equality delete file count |
-| self-optimizing.minor.trigger.interval | 3600000(1 hour) | The time interval in milliseconds to trigger minor optimizing |
-| self-optimizing.major.trigger.duplicate-ratio | 0.1 | The ratio of duplicate data of segment files to trigger major optimizing |
-| self-optimizing.full.trigger.interval | -1(closed) | The time interval in milliseconds to trigger full optimizing |
-| self-optimizing.full.rewrite-all-files | true | Whether full optimizing rewrites all files or skips files that do not need to be optimized |
-| self-optimizing.min-plan-interval | 60000 | The minimum time interval between two self-optimizing planning action |
-| self-optimizing.filter | NULL | Filter conditions for self-optimizing, using SQL conditional expressions, without supporting any functions. For the timestamp column condition, the ISO date-time formatter must be used. For example: op_time > '2007-12-03T10:15:30'. |
+| Key | Default | Description |
+|---------------------------------------------------------|------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| self-optimizing.enabled | true | Enables Self-optimizing |
+| self-optimizing.allow-partial-commit | false | Whether to allow partial commit when self-optimizing fails or process is cancelled |
+| self-optimizing.group | default | Optimizer group for Self-optimizing |
+| self-optimizing.quota | 0.5 | Quota for Self-optimizing, indicating the optimizer resources the table can take up |
+| self-optimizing.execute.num-retries | 5 | Number of retries after failure of Self-optimizing |
+| self-optimizing.target-size | 134217728(128MB) | Target size for Self-optimizing |
+| self-optimizing.max-file-count | 10000 | Maximum number of files processed by a Self-optimizing process |
+| self-optimizing.max-task-size-bytes | 134217728(128MB) | Maximum file size bytes in a single task for splitting tasks |
+| self-optimizing.fragment-ratio | 8 | The fragment file size threshold. We could divide self-optimizing.target-size by this ratio to get the actual fragment file size |
+| self-optimizing.min-target-size-ratio | 0.75 | The undersized segment file size threshold. Segment files under this threshold will be considered for rewriting |
+| self-optimizing.minor.trigger.file-count | 12 | The minimum number of files to trigger minor optimizing is determined by the sum of fragment file count and equality delete file count |
+| self-optimizing.minor.trigger.interval | 3600000(1 hour) | The time interval in milliseconds to trigger minor optimizing |
+| self-optimizing.major.trigger.duplicate-ratio | 0.1 | The ratio of duplicate data of segment files to trigger major optimizing |
+| self-optimizing.full.trigger.interval | -1(closed) | The time interval in milliseconds to trigger full optimizing |
+| self-optimizing.full.rewrite-all-files | true | Whether full optimizing rewrites all files or skips files that do not need to be optimized |
+| self-optimizing.min-plan-interval | 60000 | The minimum time interval between two self-optimizing planning action |
+| self-optimizing.filter | NULL | Filter conditions for self-optimizing, using SQL conditional expressions, without supporting any functions. For the timestamp column condition, the ISO date-time formatter must be used. For example: op_time > '2007-12-03T10:15:30'. |
+| self-optimizing.refresh-table.adaptive.max-interval-ms | -1 | The maximum time interval in milliseconds to refresh table metadata. The default value is -1, which disables the adaptive refresh. |
Review Comment:
Maybe we need to add some description here: after this has been enabled, the interval may be larger than `defaultTableRuntime.getOptimizingConfig().getMinorLeastInterval() * 4L / 5`
##########
amoro-ams/src/test/java/org/apache/amoro/server/scheduler/inline/TestTableRuntimeRefreshExecutor.java:
##########
@@ -0,0 +1,204 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.amoro.server.scheduler.inline;
+
+import org.apache.amoro.BasicTableTestHelper;
+import org.apache.amoro.TableFormat;
+import org.apache.amoro.TableTestHelper;
+import org.apache.amoro.TestedCatalogs;
+import org.apache.amoro.catalog.CatalogTestHelper;
+import org.apache.amoro.config.OptimizingConfig;
+import org.apache.amoro.hive.catalog.HiveCatalogTestHelper;
+import org.apache.amoro.hive.catalog.HiveTableTestHelper;
+import org.apache.amoro.server.table.AMSTableTestBase;
+import org.apache.amoro.server.table.DefaultTableRuntime;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(Parameterized.class)
+public class TestTableRuntimeRefreshExecutor extends AMSTableTestBase {
+ @Parameterized.Parameters(name = "{0}, {1}")
+ public static Object[] parameters() {
+ return new Object[][] {
+ {
+ TestedCatalogs.hadoopCatalog(TableFormat.MIXED_ICEBERG),
+ new BasicTableTestHelper(true, true)
+ },
+ {
+ TestedCatalogs.hadoopCatalog(TableFormat.MIXED_ICEBERG),
+ new BasicTableTestHelper(false, true)
+ },
+ {
+ TestedCatalogs.hadoopCatalog(TableFormat.MIXED_ICEBERG),
+ new BasicTableTestHelper(false, false)
+ },
+ {
+ TestedCatalogs.hadoopCatalog(TableFormat.MIXED_ICEBERG),
+ new BasicTableTestHelper(true, false)
+ },
+ {
+ new HiveCatalogTestHelper(TableFormat.ICEBERG, TEST_HMS.getHiveConf()),
+ new HiveTableTestHelper(true, true)
+ },
+ {
+ new HiveCatalogTestHelper(TableFormat.ICEBERG, TEST_HMS.getHiveConf()),
+ new HiveTableTestHelper(false, true)
+ },
+ {
+ new HiveCatalogTestHelper(TableFormat.ICEBERG, TEST_HMS.getHiveConf()),
+ new HiveTableTestHelper(false, false)
+ },
+ {
+ new HiveCatalogTestHelper(TableFormat.ICEBERG, TEST_HMS.getHiveConf()),
+ new HiveTableTestHelper(true, false)
+ },
+ };
+ }
+
+ public TestTableRuntimeRefreshExecutor(
+ CatalogTestHelper catalogTestHelper, TableTestHelper tableTestHelper) {
+ super(catalogTestHelper, tableTestHelper);
+ }
+
+ private static final long INTERVAL = 60000L; // 1 minute
+ private static final long MAX_INTERVAL = 300000L; // 5 minutes
+ private static final long DEFAULT_MAX_INTERVAL = -1L;
+ private static final long STEP = 30000L; // 30s
+ private static final int MAX_PENDING_PARTITIONS = 1;
+ private static final int MINOR_LEAST_INTERVAL = 3600000; // 1h
+
+ // Create DefaultTableRuntime for adaptive interval tests
+ private DefaultTableRuntime buildTableRuntimeWithAdaptiveRefresh(
+ DefaultTableRuntime tableRuntime, boolean needOptimizing, long interval, long step) {
+ return buildTableRuntimeWithConfig(tableRuntime, needOptimizing, interval, MAX_INTERVAL, step);
+ }
+
+ private DefaultTableRuntime buildTableRuntimeWithAdaptiveMaxInterval(
+ DefaultTableRuntime tableRuntime, long interval, long maxInterval) {
+ return buildTableRuntimeWithConfig(tableRuntime, true, interval, maxInterval, STEP);
+ }
+
+ private DefaultTableRuntime buildTableRuntimeWithConfig(
+ DefaultTableRuntime tableRuntime,
+ boolean needOptimizing,
+ long interval,
+ long maxInterval,
+ long step) {
+ OptimizingConfig optimizingConfig = new OptimizingConfig();
+ optimizingConfig.setRefreshTableAdaptiveMaxIntervalMs(maxInterval);
+ optimizingConfig.setMinorLeastInterval(MINOR_LEAST_INTERVAL);
+ optimizingConfig.setRefreshTableAdaptiveIncreaseStepMs(step);
+
+ DefaultTableRuntime newRuntime =
+ new DefaultTableRuntime(tableRuntime.store()) {
+ @Override
+ public OptimizingConfig getOptimizingConfig() {
+ return optimizingConfig;
+ }
+ };
+ newRuntime.setLatestEvaluatedNeedOptimizing(needOptimizing);
+ newRuntime.setLatestRefreshInterval(interval);
+
+ return newRuntime;
+ }
+
+ @Test
+ public void testAdaptiveRefreshIntervalScenarios() {
+ createDatabase();
+ createTable();
+
+ DefaultTableRuntime tableRuntime =
Review Comment:
Could you please reuse the _same_ table runtime, in the current test, we'll create every _new_ table runtime in `buildTableRuntimeWithAdaptiveRefresh`
##########
amoro-ams/src/test/java/org/apache/amoro/server/scheduler/inline/TestTableRuntimeRefreshExecutor.java:
##########
@@ -0,0 +1,204 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.amoro.server.scheduler.inline;
+
+import org.apache.amoro.BasicTableTestHelper;
+import org.apache.amoro.TableFormat;
+import org.apache.amoro.TableTestHelper;
+import org.apache.amoro.TestedCatalogs;
+import org.apache.amoro.catalog.CatalogTestHelper;
+import org.apache.amoro.config.OptimizingConfig;
+import org.apache.amoro.hive.catalog.HiveCatalogTestHelper;
+import org.apache.amoro.hive.catalog.HiveTableTestHelper;
+import org.apache.amoro.server.table.AMSTableTestBase;
+import org.apache.amoro.server.table.DefaultTableRuntime;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(Parameterized.class)
+public class TestTableRuntimeRefreshExecutor extends AMSTableTestBase {
+ @Parameterized.Parameters(name = "{0}, {1}")
+ public static Object[] parameters() {
+ return new Object[][] {
+ {
+ TestedCatalogs.hadoopCatalog(TableFormat.MIXED_ICEBERG),
+ new BasicTableTestHelper(true, true)
+ },
+ {
+ TestedCatalogs.hadoopCatalog(TableFormat.MIXED_ICEBERG),
+ new BasicTableTestHelper(false, true)
+ },
+ {
+ TestedCatalogs.hadoopCatalog(TableFormat.MIXED_ICEBERG),
+ new BasicTableTestHelper(false, false)
+ },
+ {
+ TestedCatalogs.hadoopCatalog(TableFormat.MIXED_ICEBERG),
+ new BasicTableTestHelper(true, false)
+ },
+ {
+ new HiveCatalogTestHelper(TableFormat.ICEBERG, TEST_HMS.getHiveConf()),
+ new HiveTableTestHelper(true, true)
+ },
+ {
+ new HiveCatalogTestHelper(TableFormat.ICEBERG, TEST_HMS.getHiveConf()),
+ new HiveTableTestHelper(false, true)
+ },
+ {
+ new HiveCatalogTestHelper(TableFormat.ICEBERG, TEST_HMS.getHiveConf()),
+ new HiveTableTestHelper(false, false)
+ },
+ {
+ new HiveCatalogTestHelper(TableFormat.ICEBERG, TEST_HMS.getHiveConf()),
+ new HiveTableTestHelper(true, false)
+ },
+ };
+ }
+
+ public TestTableRuntimeRefreshExecutor(
+ CatalogTestHelper catalogTestHelper, TableTestHelper tableTestHelper) {
+ super(catalogTestHelper, tableTestHelper);
+ }
+
+ private static final long INTERVAL = 60000L; // 1 minute
+ private static final long MAX_INTERVAL = 300000L; // 5 minutes
+ private static final long DEFAULT_MAX_INTERVAL = -1L;
+ private static final long STEP = 30000L; // 30s
+ private static final int MAX_PENDING_PARTITIONS = 1;
+ private static final int MINOR_LEAST_INTERVAL = 3600000; // 1h
+
+ // Create DefaultTableRuntime for adaptive interval tests
+ private DefaultTableRuntime buildTableRuntimeWithAdaptiveRefresh(
+ DefaultTableRuntime tableRuntime, boolean needOptimizing, long interval, long step) {
+ return buildTableRuntimeWithConfig(tableRuntime, needOptimizing, interval, MAX_INTERVAL, step);
+ }
+
+ private DefaultTableRuntime buildTableRuntimeWithAdaptiveMaxInterval(
+ DefaultTableRuntime tableRuntime, long interval, long maxInterval) {
+ return buildTableRuntimeWithConfig(tableRuntime, true, interval, maxInterval, STEP);
+ }
+
+ private DefaultTableRuntime buildTableRuntimeWithConfig(
+ DefaultTableRuntime tableRuntime,
+ boolean needOptimizing,
+ long interval,
+ long maxInterval,
+ long step) {
+ OptimizingConfig optimizingConfig = new OptimizingConfig();
+ optimizingConfig.setRefreshTableAdaptiveMaxIntervalMs(maxInterval);
+ optimizingConfig.setMinorLeastInterval(MINOR_LEAST_INTERVAL);
+ optimizingConfig.setRefreshTableAdaptiveIncreaseStepMs(step);
+
+ DefaultTableRuntime newRuntime =
+ new DefaultTableRuntime(tableRuntime.store()) {
+ @Override
+ public OptimizingConfig getOptimizingConfig() {
+ return optimizingConfig;
+ }
+ };
+ newRuntime.setLatestEvaluatedNeedOptimizing(needOptimizing);
+ newRuntime.setLatestRefreshInterval(interval);
+
+ return newRuntime;
+ }
+
+ @Test
+ public void testAdaptiveRefreshIntervalScenarios() {
+ createDatabase();
+ createTable();
+
+ DefaultTableRuntime tableRuntime =
Review Comment:
If we need to update the `config` for an existing table runtime, maybe we can add a new class who extend `DefaultTableRuntime` and overrides the `getStore` function.
```
class TestTableRuntime extends DefaultTableRuntime {
TableRuntimeStore testStore;
TestTableRuntime(TableRuntimeStore store) {
...
this.testStore = store
}
void updateRuntimeStore(TableRuntimeStore newStore) {
...
this.testStore = newStore;
}
@Override
TableConfiguration getStore() {
this.testStore;
}
```
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-4037] Add support for dynamic refresh interval for table metadata refreshing in TableRuntimeRefreshExecutor [amoro]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #4038:
URL: https://github.com/apache/amoro/pull/4038#discussion_r2715288245
##########
docs/user-guides/configurations.md:
##########
@@ -43,25 +43,28 @@ modified through [Alter Table](../using-tables/#modify-table) operations.
Self-optimizing configurations are applicable to both Iceberg Format and Mixed streaming Format.
-| Key | Default | Description |
-|-----------------------------------------------|------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| self-optimizing.enabled | true | Enables Self-optimizing |
-| self-optimizing.allow-partial-commit | false | Whether to allow partial commit when self-optimizing fails or process is cancelled |
-| self-optimizing.group | default | Optimizer group for Self-optimizing |
-| self-optimizing.quota | 0.5 | Quota for Self-optimizing, indicating the optimizer resources the table can take up |
-| self-optimizing.execute.num-retries | 5 | Number of retries after failure of Self-optimizing |
-| self-optimizing.target-size | 134217728(128MB) | Target size for Self-optimizing |
-| self-optimizing.max-file-count | 10000 | Maximum number of files processed by a Self-optimizing process |
-| self-optimizing.max-task-size-bytes | 134217728(128MB) | Maximum file size bytes in a single task for splitting tasks |
-| self-optimizing.fragment-ratio | 8 | The fragment file size threshold. We could divide self-optimizing.target-size by this ratio to get the actual fragment file size |
-| self-optimizing.min-target-size-ratio | 0.75 | The undersized segment file size threshold. Segment files under this threshold will be considered for rewriting |
-| self-optimizing.minor.trigger.file-count | 12 | The minimum number of files to trigger minor optimizing is determined by the sum of fragment file count and equality delete file count |
-| self-optimizing.minor.trigger.interval | 3600000(1 hour) | The time interval in milliseconds to trigger minor optimizing |
-| self-optimizing.major.trigger.duplicate-ratio | 0.1 | The ratio of duplicate data of segment files to trigger major optimizing |
-| self-optimizing.full.trigger.interval | -1(closed) | The time interval in milliseconds to trigger full optimizing |
-| self-optimizing.full.rewrite-all-files | true | Whether full optimizing rewrites all files or skips files that do not need to be optimized |
-| self-optimizing.min-plan-interval | 60000 | The minimum time interval between two self-optimizing planning action |
-| self-optimizing.filter | NULL | Filter conditions for self-optimizing, using SQL conditional expressions, without supporting any functions. For the timestamp column condition, the ISO date-time formatter must be used. For example: op_time > '2007-12-03T10:15:30'. |
+| Key | Default | Description |
+|------------------------------------------------------|------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| self-optimizing.enabled | true | Enables Self-optimizing |
+| self-optimizing.allow-partial-commit | false | Whether to allow partial commit when self-optimizing fails or process is cancelled |
+| self-optimizing.group | default | Optimizer group for Self-optimizing |
+| self-optimizing.quota | 0.5 | Quota for Self-optimizing, indicating the optimizer resources the table can take up |
+| self-optimizing.execute.num-retries | 5 | Number of retries after failure of Self-optimizing |
+| self-optimizing.target-size | 134217728(128MB) | Target size for Self-optimizing |
+| self-optimizing.max-file-count | 10000 | Maximum number of files processed by a Self-optimizing process |
+| self-optimizing.max-task-size-bytes | 134217728(128MB) | Maximum file size bytes in a single task for splitting tasks |
+| self-optimizing.fragment-ratio | 8 | The fragment file size threshold. We could divide self-optimizing.target-size by this ratio to get the actual fragment file size |
+| self-optimizing.min-target-size-ratio | 0.75 | The undersized segment file size threshold. Segment files under this threshold will be considered for rewriting |
+| self-optimizing.minor.trigger.file-count | 12 | The minimum number of files to trigger minor optimizing is determined by the sum of fragment file count and equality delete file count |
+| self-optimizing.minor.trigger.interval | 3600000(1 hour) | The time interval in milliseconds to trigger minor optimizing |
+| self-optimizing.major.trigger.duplicate-ratio | 0.1 | The ratio of duplicate data of segment files to trigger major optimizing |
+| self-optimizing.full.trigger.interval | -1(closed) | The time interval in milliseconds to trigger full optimizing |
+| self-optimizing.full.rewrite-all-files | true | Whether full optimizing rewrites all files or skips files that do not need to be optimized |
+| self-optimizing.min-plan-interval | 60000 | The minimum time interval between two self-optimizing planning action |
+| self-optimizing.filter | NULL | Filter conditions for self-optimizing, using SQL conditional expressions, without supporting any functions. For the timestamp column condition, the ISO date-time formatter must be used. For example: op_time > '2007-12-03T10:15:30'. |
+| self-optimizing.refresh-table.adaptive.enabled | false | Whether to enable adaptive refresh interval for refreshing table metadata |
+| self-optimizing.refresh-table.adaptive.max-interval | 3600000(1 hour) | The maximum time interval in milliseconds to refresh table metadata |
+| self-optimizing.refresh-table.adaptive.increase-step | 30000(30s) | The time interval increase step in milliseconds to refres table metadata |
Review Comment:
Corrected spelling of 'refres' to 'refresh'.
```suggestion
| self-optimizing.refresh-table.adaptive.increase-step | 30000(30s) | The time interval increase step in milliseconds to refresh table metadata |
```
--
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AMORO-4037] Add support for dynamic refresh interval for table metadata refreshing in TableRuntimeRefreshExecutor [amoro]
Posted by "klion26 (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org