This is an automated email from the ASF dual-hosted git repository.
zhoujinsong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/amoro.git
The following commit(s) were added to refs/heads/master by this push:
new 1c894691b [hotfix][Optimizer] Guard invalid fragment ratio and target
quota (#4324)
1c894691b is described below
commit 1c894691bcb761774901c11ed46f1e454bdad1d3
Author: ConradJam <[email protected]>
AuthorDate: Tue Aug 25 14:21:36 2026 +0800
[hotfix][Optimizer] Guard invalid fragment ratio and target quota (#4324)
self-optimizing.fragment-ratio and self-optimizing.quota do not enforce
positive values, so zero or negative values may already exist in persisted
table properties.
Normalize fragment ratios to at least one in
OptimizingConfig.setFragmentRatio(), keep the configured default value of 8 in
TableProperties, and make CommonPartitionEvaluator reuse
OptimizingConfig.maxFragmentSize(). This prevents division by zero in the
actual planner while leaving persisted properties unchanged.
Clamp quota limits to at least one so invalid target quotas neither corrupt
quota weights nor starve tables of schedulable slots.
Tests: TestOptimizingConfig, TestCommonPartitionEvaluator, and
TestOptimizingQueue#testQuotaSchedulePolicy.
---
.../amoro/server/optimizing/OptimizingQueue.java | 4 +-
.../amoro/server/table/DefaultTableRuntime.java | 6 +-
.../org/apache/amoro/config/OptimizingConfig.java | 2 +-
.../apache/amoro/config/TestOptimizingConfig.java | 47 +++++++++++++
.../optimizing/plan/CommonPartitionEvaluator.java | 2 +-
.../plan/TestCommonPartitionEvaluator.java | 82 ++++++++++++++++++++++
6 files changed, 139 insertions(+), 4 deletions(-)
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
index 9d579ef24..c090979a2 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
@@ -626,9 +626,11 @@ public class OptimizingQueue extends PersistentBase {
private int getQuotaLimit() {
double targetQuota = tableRuntime.getOptimizingConfig().getTargetQuota();
+ // A non-positive target quota must not starve the table to zero
schedulable slots;
+ // clamp to 1 like getAvailableCore does for the group quota.
return targetQuota > 1
? (int) targetQuota
- : (int) Math.ceil(targetQuota * getAvailableCore());
+ : (int) Math.max(1, Math.ceil(targetQuota * getAvailableCore()));
}
@Override
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java
b/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java
index e2608a1e6..a3467bcd3 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java
@@ -228,8 +228,12 @@ public class DefaultTableRuntime extends
AbstractTableRuntime {
public double calculateQuotaOccupy() {
double targetQuota = getOptimizingConfig().getTargetQuota();
+ // Guard against a non-positive target quota (misconfigured property): a
zero limit would
+ // turn the weight into Infinity/NaN and corrupt quota-based sorting.
int targetQuotaLimit =
- targetQuota > 1 ? (int) targetQuota : (int) Math.ceil(targetQuota *
getThreadCount());
+ targetQuota > 1
+ ? (int) targetQuota
+ : (int) Math.max(1, Math.ceil(targetQuota * getThreadCount()));
return (double) getQuotaTime() /
AmoroServiceConstants.QUOTA_LOOK_BACK_TIME / targetQuotaLimit;
}
diff --git
a/amoro-common/src/main/java/org/apache/amoro/config/OptimizingConfig.java
b/amoro-common/src/main/java/org/apache/amoro/config/OptimizingConfig.java
index a8efccd3b..2f3c6a419 100644
--- a/amoro-common/src/main/java/org/apache/amoro/config/OptimizingConfig.java
+++ b/amoro-common/src/main/java/org/apache/amoro/config/OptimizingConfig.java
@@ -212,7 +212,7 @@ public class OptimizingConfig {
}
public OptimizingConfig setFragmentRatio(int fragmentRatio) {
- this.fragmentRatio = fragmentRatio;
+ this.fragmentRatio = Math.max(1, fragmentRatio);
return this;
}
diff --git
a/amoro-common/src/test/java/org/apache/amoro/config/TestOptimizingConfig.java
b/amoro-common/src/test/java/org/apache/amoro/config/TestOptimizingConfig.java
new file mode 100644
index 000000000..a2ec0d40a
--- /dev/null
+++
b/amoro-common/src/test/java/org/apache/amoro/config/TestOptimizingConfig.java
@@ -0,0 +1,47 @@
+/*
+ * 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.config;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestOptimizingConfig {
+
+ @Test
+ public void maxFragmentSizeDividesTargetSizeByRatio() {
+ OptimizingConfig config = new OptimizingConfig().setTargetSize(128 <<
20).setFragmentRatio(8);
+ Assertions.assertEquals(16 << 20, config.maxFragmentSize());
+ }
+
+ @Test
+ public void nonPositiveFragmentRatioIsClampedToOne() {
+ OptimizingConfig zero = new OptimizingConfig().setTargetSize(128 <<
20).setFragmentRatio(0);
+ Assertions.assertEquals(1, zero.getFragmentRatio());
+ Assertions.assertEquals(128 << 20, zero.maxFragmentSize());
+
+ OptimizingConfig negative =
+ new OptimizingConfig()
+ .setTargetSize(128 << 20)
+ .setFragmentRatio(-2)
+ .setMajorDuplicateRatio(0.5);
+ Assertions.assertEquals(1, negative.getFragmentRatio());
+ Assertions.assertEquals(128 << 20, negative.maxFragmentSize());
+ Assertions.assertEquals(64 << 20, negative.maxDuplicateSize());
+ }
+}
diff --git
a/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/plan/CommonPartitionEvaluator.java
b/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/plan/CommonPartitionEvaluator.java
index 2cd11cbd5..ef1a04981 100644
---
a/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/plan/CommonPartitionEvaluator.java
+++
b/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/plan/CommonPartitionEvaluator.java
@@ -101,7 +101,7 @@ public class CommonPartitionEvaluator implements
PartitionEvaluator {
this.identifier = identifier;
this.config = config;
this.partition = partition;
- this.fragmentSize = config.getTargetSize() / config.getFragmentRatio();
+ this.fragmentSize = config.maxFragmentSize();
this.minTargetSize = (long) (config.getTargetSize() *
config.getMinTargetSizeRatio());
if (minTargetSize > config.getTargetSize() - fragmentSize) {
LOG.warn(
diff --git
a/amoro-format-iceberg/src/test/java/org/apache/amoro/optimizing/plan/TestCommonPartitionEvaluator.java
b/amoro-format-iceberg/src/test/java/org/apache/amoro/optimizing/plan/TestCommonPartitionEvaluator.java
new file mode 100644
index 000000000..8bb192481
--- /dev/null
+++
b/amoro-format-iceberg/src/test/java/org/apache/amoro/optimizing/plan/TestCommonPartitionEvaluator.java
@@ -0,0 +1,82 @@
+/*
+ * 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.optimizing.plan;
+
+import org.apache.amoro.ServerTableIdentifier;
+import org.apache.amoro.TableFormat;
+import org.apache.amoro.config.OptimizingConfig;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.util.Pair;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Collections;
+
+public class TestCommonPartitionEvaluator {
+
+ private static final long DEFAULT_TARGET_SIZE = 128L << 20;
+
+ @Test
+ public void nonPositiveFragmentRatioIsClampedBeforeEvaluation() {
+ assertUsesClampedFragmentThreshold(0);
+ assertUsesClampedFragmentThreshold(-2);
+ }
+
+ private void assertUsesClampedFragmentThreshold(int fragmentRatio) {
+ OptimizingConfig config =
+ new OptimizingConfig()
+ .setEnabled(true)
+ .setTargetSize(DEFAULT_TARGET_SIZE)
+ .setFragmentRatio(fragmentRatio)
+ .setMinTargetSizeRatio(0.75)
+ .setMajorDuplicateRatio(0.5)
+ .setFullTriggerInterval(-1);
+ Assert.assertEquals(1, config.getFragmentRatio());
+ Pair<Integer, StructLike> partition = Pair.of(0, null);
+ CommonPartitionEvaluator evaluator =
+ new CommonPartitionEvaluator(
+ ServerTableIdentifier.of(1L, "catalog", "database", "table",
TableFormat.ICEBERG),
+ config,
+ partition,
+ System.currentTimeMillis(),
+ 0L,
+ 0L,
+ 0L);
+
+ Assert.assertTrue(
+ evaluator.addFile(
+ dataFile("fragment-" + fragmentRatio, DEFAULT_TARGET_SIZE),
Collections.emptyList()));
+ Assert.assertFalse(
+ evaluator.addFile(
+ dataFile("non-fragment-" + fragmentRatio, DEFAULT_TARGET_SIZE + 1),
+ Collections.emptyList()));
+ Assert.assertEquals(1, evaluator.getFragmentFileCount());
+ }
+
+ private DataFile dataFile(String name, long size) {
+ return DataFiles.builder(PartitionSpec.unpartitioned())
+ .withPath("/tmp/" + name + ".parquet")
+ .withFileSizeInBytes(size)
+ .withRecordCount(1)
+ .build();
+ }
+}
