czy006 commented on code in PR #4167:
URL: https://github.com/apache/amoro/pull/4167#discussion_r3036642128
##########
amoro-format-iceberg/src/main/java/org/apache/amoro/io/writer/SortedPosDeleteWriter.java:
##########
@@ -49,6 +50,25 @@
*/
public class SortedPosDeleteWriter<T> implements Closeable {
private static final long DEFAULT_RECORDS_NUM_THRESHOLD = Long.MAX_VALUE;
+ private static final Runtime RUNTIME = Runtime.getRuntime();
+ private static final HeapUsageProvider RUNTIME_HEAP_USAGE =
+ new HeapUsageProvider() {
+ @Override
+ public long maxMemory() {
+ return RUNTIME.maxMemory();
+ }
+
+ @Override
+ public long usedMemory() {
+ return RUNTIME.totalMemory() - RUNTIME.freeMemory();
Review Comment:
**[Improvement] `usedMemory()` 计算方式的局限性**
`Runtime.totalMemory() - Runtime.freeMemory()` 是 JVM 堆使用量的粗略估算：
- `totalMemory()` 是 JVM 当前已分配的堆大小（非最大堆），JVM 启动初期远小于 `maxMemory()`
- `freeMemory()` 是已分配堆中的空闲部分
这意味着在 JVM 堆尚未完全扩展时，`usedMemory()` 返回值偏小，heap flush 不会及时触发——而这恰恰是最需要保护的时刻。
**建议**: 在注释中明确说明此计算方式的局限性，或考虑使用 `MemoryPoolMXBean` 获取更精确的 Eden/Old 区使用数据。
##########
amoro-format-iceberg/src/main/java/org/apache/amoro/io/writer/SortedPosDeleteWriter.java:
##########
@@ -98,7 +174,9 @@ public SortedPosDeleteWriter(
mask,
index,
partitionKey,
- DEFAULT_RECORDS_NUM_THRESHOLD);
+ DEFAULT_RECORDS_NUM_THRESHOLD,
+ TableProperties.POS_DELETE_FLUSH_HEAP_RATIO_DEFAULT,
Review Comment:
**[Critical] 默认行为静默变更**
向后兼容的构造函数默认传入了 `POS_DELETE_FLUSH_HEAP_RATIO_DEFAULT =
0.8d`，导致所有使用旧构造函数的代码路径隐式启用了 heap-based flush。
原来 `recordsNumThreshold = Long.MAX_VALUE` 意味着 flush 永远不会由记录数触发。现在默认
ratio=0.8 意味着堆使用率超过80%就会触发 flush。对于 JVM 堆使用率常态在 80% 附近的数据处理任务，这可能导致：
- 产生大量小 delete 文件
- 查询性能退化
**建议**: 向后兼容的构造函数应传 `0d`（禁用 heap flush），仅当用户显式配置时才启用。例如：
```java
// 保持旧行为：不启用 heap flush
this(..., recordsNumThreshold, 0d, 0);
```
受影响的旧代码路径包括：
- `MixedTreeNodePosDeleteWriter` 的无 properties 构造函数
- `TestTableTracer`, `TestTaskWriter`, `MixedDataTestHelpers`,
`TestOrphanFileCleanIceberg` 等测试文件
##########
amoro-format-iceberg/src/main/java/org/apache/amoro/io/writer/SortedPosDeleteWriter.java:
##########
@@ -132,9 +234,7 @@ public void delete(CharSequence path, long pos, T row) {
records += 1;
- // TODO Flush buffer based on the policy that checking whether whole heap
memory size exceed the
- // threshold.
- if (records >= recordsNumThreshold) {
+ if (records >= recordsNumThreshold || shouldFlushByHeap()) {
Review Comment:
**[Critical] 每次 `delete()` 都调用 `shouldFlushByHeap()` 的性能开销**
当前实现在每次 `delete()` 调用时都会执行 `shouldFlushByHeap()`，该方法调用
`Runtime.totalMemory()` 和 `Runtime.freeMemory()`。在 JDK 14 之前的版本中，这些方法可能触发
safepoint 操作，高频调用会引入不可忽视的性能开销。
**建议**: 添加采样机制，避免每次都查询 JVM 内存指标：
```java
// 仅在 records 达到 heapFlushMinRecords 的倍数时检查堆
if (records >= recordsNumThreshold ||
(records % Math.max(1, heapFlushMinRecords) == 0 &&
shouldFlushByHeap())) {
flushDeletes();
}
```
或者使用一个计数器，每 N 次 delete 才检查一次堆使用率。
##########
amoro-format-iceberg/src/test/java/org/apache/amoro/io/writer/TestSortedPosDeleteWriterHeapFlush.java:
##########
@@ -0,0 +1,175 @@
+/*
+ * 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.io.writer;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import org.apache.amoro.BasicTableTestHelper;
+import org.apache.amoro.TableFormat;
+import org.apache.amoro.catalog.BasicCatalogTestHelper;
+import org.apache.amoro.catalog.TableTestBase;
+import org.apache.amoro.table.UnkeyedTable;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.MetadataColumns;
+import org.apache.iceberg.MetricsModes;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.TestHelpers;
+import org.apache.iceberg.data.GenericAppenderFactory;
+import org.apache.iceberg.data.Record;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class TestSortedPosDeleteWriterHeapFlush extends TableTestBase {
+
+ public TestSortedPosDeleteWriterHeapFlush() {
+ super(
+ new BasicCatalogTestHelper(TableFormat.MIXED_ICEBERG),
+ new BasicTableTestHelper(false, false));
+ }
+
+ @BeforeAll
+ public static void startAms() throws Exception {
+ TEST_AMS.before();
+ }
+
+ @AfterAll
+ public static void stopAms() {
+ TEST_AMS.after();
+ }
+
+ @BeforeEach
+ public void setup() throws Exception {
+ temp.create();
+ setupCatalog();
+ setupTable();
+ }
+
+ @AfterEach
+ public void tearDown() {
+ if (getCatalogMeta() != null) {
+ dropTable();
+ dropCatalog();
+ }
+ temp.delete();
+ }
+
+ @Test
Review Comment:
**[Improvement] 测试覆盖不足**
当前只有 2 个测试用例，缺少以下边界条件测试：
1. **`heapUsageRatioThreshold = 1.0`**: PR 文档说 `>=1` 禁用，但没有测试验证
2. **`maxMemory() <= 0`**: 应验证不会触发 flush
3. **`heapFlushMinRecords = 0`**: 边界行为
4. **多次 flush 循环**: flush 后 `records` 重置为0，再积累到 `heapFlushMinRecords` 再次触发
5. **`recordsNumThreshold` 与 heap flush 同时触发**: 验证优先级
6. **`heapUsageRatioThreshold` 为负数**: 验证禁用逻辑
建议补充以上测试用例，确保 `shouldFlushByHeap()` 的所有分支都有覆盖。
##########
amoro-format-mixed/amoro-mixed-spark/v3.5/amoro-mixed-spark-3.5/src/main/java/org/apache/amoro/spark/io/UnkeyedUpsertSparkWriter.java:
##########
@@ -73,6 +78,21 @@ public UnkeyedUpsertSparkWriter(
this.format = format;
this.schema = schema;
this.writer = writer;
+ this.heapUsageRatioThreshold =
Review Comment:
**[Nitpick] 三个 Spark 版本中完全重复的 property 读取代码**
这 15 行 property 读取逻辑在 `UnkeyedUpsertSparkWriter` 的 v3.3/v3.4/v3.5
三个版本中完全相同（同样的代码在 3 个文件中各出现一次）。
虽然这是 Amoro 多 Spark 版本架构的固有模式，但可以考虑将此逻辑提取到公共工具方法中，减少未来维护成本。例如：
```java
// 在公共模块中
public class FlushConfig {
public static FlushConfig fromProperties(Map<String, String> properties)
{ ... }
}
```
这不是阻塞项，但值得考虑。
##########
amoro-format-iceberg/src/main/java/org/apache/amoro/io/writer/SortedPosDeleteWriter.java:
##########
@@ -49,6 +50,25 @@
*/
public class SortedPosDeleteWriter<T> implements Closeable {
private static final long DEFAULT_RECORDS_NUM_THRESHOLD = Long.MAX_VALUE;
+ private static final Runtime RUNTIME = Runtime.getRuntime();
+ private static final HeapUsageProvider RUNTIME_HEAP_USAGE =
+ new HeapUsageProvider() {
+ @Override
+ public long maxMemory() {
+ return RUNTIME.maxMemory();
+ }
+
+ @Override
+ public long usedMemory() {
+ return RUNTIME.totalMemory() - RUNTIME.freeMemory();
+ }
+ };
+
+ interface HeapUsageProvider {
Review Comment:
**[Nitpick] `HeapUsageProvider` 接口可见性**
当前 `HeapUsageProvider` 是 package-private（无修饰符）。如果未来需要在其他包中使用（如 Spark
模块自定义实现），可能需要改为 `public`。
不过当前设计作为 package-private 是合理的——限制了 API 表面积，测试通过同一包访问。这只是一个提醒。
--
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]
