[AMORO-4183][Improvement]: Refactor Optimizing Interface Move to amoro-common

amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/TaskProperties.java → amoro-common/src/main/java/org/apache/amoro/optimizing/TaskProperties.java

@@ -19,8 +19,6 @@
 package org.apache.amoro.optimizing;
 
 import org.apache.amoro.OptimizerProperties;
-import org.apache.amoro.utils.PropertyUtil;
-import org.apache.amoro.utils.map.StructLikeCollections;
 
 import java.util.Map;
 
@@ -45,17 +43,6 @@ public class TaskProperties {
 
   public static final String MOVE_FILE_TO_HIVE_LOCATION = "move-files-to-hive-location";
 
-  public static StructLikeCollections getStructLikeCollections(Map<String, String> properties) {
-    boolean enableSpillMap =
-        PropertyUtil.propertyAsBoolean(
-            properties, EXTEND_DISK_STORAGE, EXTEND_DISK_STORAGE_DEFAULT);
-    long maxInMemory =
-        PropertyUtil.propertyAsLong(properties, MEMORY_STORAGE_SIZE, MEMORY_STORAGE_SIZE_DEFAULT);
-    String spillMapPath = properties.get(DISK_STORAGE_PATH);
-
-    return new StructLikeCollections(enableSpillMap, maxInMemory, spillMapPath);
-  }
-
   public static String getProcessId(Map<String, String> properties) {
     String processId = properties.get(PROCESS_ID);
     if (processId == null || processId.trim().isEmpty()) {

amoro-common/src/test/java/org/apache/amoro/optimizing/TestOptimizingSpiInterfaces.java

@@ -0,0 +1,109 @@
+/*
+ * 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;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+import org.apache.amoro.utils.SerializationUtil;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+@DisplayName("Test optimizing SPI interfaces after moving to amoro-common")
+public class TestOptimizingSpiInterfaces {
+
+  @Test
+  @DisplayName("BaseOptimizingInput.option() should store and retrieve a single option")
+  void testBaseOptimizingInputOption() {
+    BaseOptimizingInput input = createTestInput();
+    input.option("key1", "value1");
+    assertEquals("value1", input.getOptions().get("key1"));
+  }
+
+  @Test
+  @DisplayName("BaseOptimizingInput.options() should merge multiple options")
+  void testBaseOptimizingInputOptions() {
+    BaseOptimizingInput input = createTestInput();
+    input.option("existing", "v0");
+
+    Map<String, String> batch = new HashMap<>();
+    batch.put("key1", "value1");
+    batch.put("key2", "value2");
+    input.options(batch);
+
+    assertEquals(3, input.getOptions().size());
+    assertEquals("v0", input.getOptions().get("existing"));
+    assertEquals("value1", input.getOptions().get("key1"));
+    assertEquals("value2", input.getOptions().get("key2"));
+  }
+
+  @Test
+  @DisplayName("OptimizingOutput.summary() should return a map")
+  void testOptimizingOutputSummary() {
+    TableOptimizing.OptimizingOutput output =
+        new TableOptimizing.OptimizingOutput() {
+          @Override
+          public Map<String, String> summary() {
+            return Collections.singletonMap("files", "10");
+          }
+        };
+    assertNotNull(output.summary());
+    assertEquals("10", output.summary().get("files"));
+  }
+
+  @Test
+  @DisplayName("TaskProperties constants should have expected values")
+  void testTaskPropertyConstants() {
+    assertEquals("task-executor-factory-impl", TaskProperties.TASK_EXECUTOR_FACTORY_IMPL);
+    assertEquals("process-id", TaskProperties.PROCESS_ID);
+    assertEquals("unknown", TaskProperties.UNKNOWN_PROCESS_ID);
+  }
+
+  @Test
+  @DisplayName("BaseOptimizingInput subclass should be serializable via SerializationUtil")
+  void testBaseOptimizingInputSerialization() {
+    TestSerializableInput input = new TestSerializableInput();
+    input.option("key1", "value1");
+    input.option("key2", "value2");
+
+    ByteBuffer buffer = SerializationUtil.simpleSerialize(input);
+    assertNotNull(buffer);
+
+    byte[] bytes = new byte[buffer.remaining()];
+    buffer.get(bytes);
+    TestSerializableInput deserialized = SerializationUtil.simpleDeserialize(bytes);
+    assertNotNull(deserialized);
+    assertEquals("value1", deserialized.getOptions().get("key1"));
+    assertEquals("value2", deserialized.getOptions().get("key2"));
+  }
+
+  private BaseOptimizingInput createTestInput() {
+    return new BaseOptimizingInput() {};
+  }
+
+  /** A concrete subclass for serialization testing. */
+  public static class TestSerializableInput extends BaseOptimizingInput {
+    private static final long serialVersionUID = 1L;
+  }
+}

amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/AbstractRewriteFilesExecutor.java

@@ -84,10 +84,25 @@ public AbstractRewriteFilesExecutor(
     this.table = table;
     this.io = table.io();
     this.properties = properties;
-    this.structLikeCollections = TaskProperties.getStructLikeCollections(properties);
+    this.structLikeCollections = getStructLikeCollections(properties);
     dataReader = dataReader();
   }
 
+  protected static StructLikeCollections getStructLikeCollections(Map<String, String> properties) {
+    boolean enableSpillMap =
+        PropertyUtil.propertyAsBoolean(
+            properties,
+            TaskProperties.EXTEND_DISK_STORAGE,
+            TaskProperties.EXTEND_DISK_STORAGE_DEFAULT);
+    long maxInMemory =
+        PropertyUtil.propertyAsLong(
+            properties,
+            TaskProperties.MEMORY_STORAGE_SIZE,
+            TaskProperties.MEMORY_STORAGE_SIZE_DEFAULT);
+    String spillMapPath = properties.get(TaskProperties.DISK_STORAGE_PATH);
+    return new StructLikeCollections(enableSpillMap, maxInMemory, spillMapPath);
+  }
+
   protected abstract OptimizingDataReader dataReader();
 
   protected abstract FileWriter<PositionDelete<Record>, DeleteWriteResult> posWriter();

amoro-format-iceberg/src/test/java/org/apache/amoro/optimizing/TestTaskPropertiesSplit.java

@@ -0,0 +1,64 @@
+/*
+ * 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;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@DisplayName("Test TaskProperties after getStructLikeCollections extraction")
+public class TestTaskPropertiesSplit {
+
+  @Test
+  @DisplayName("getProcessId should return correct process id from properties")
+  void testGetProcessId() {
+    Map<String, String> properties = new HashMap<>();
+    properties.put(TaskProperties.PROCESS_ID, "12345");
+    assertEquals("12345", TaskProperties.getProcessId(properties));
+  }
+
+  @Test
+  @DisplayName("getProcessId should return UNKNOWN when property is null")
+  void testGetProcessIdWithNull() {
+    Map<String, String> properties = new HashMap<>();
+    assertEquals(TaskProperties.UNKNOWN_PROCESS_ID, TaskProperties.getProcessId(properties));
+  }
+
+  @Test
+  @DisplayName("getProcessId should return UNKNOWN when property is empty")
+  void testGetProcessIdWithEmpty() {
+    Map<String, String> properties = new HashMap<>();
+    properties.put(TaskProperties.PROCESS_ID, "  ");
+    assertEquals(TaskProperties.UNKNOWN_PROCESS_ID, TaskProperties.getProcessId(properties));
+  }
+
+  @Test
+  @DisplayName("Task property constants should have expected values")
+  void testTaskPropertyConstants() {
+    assertEquals("task-executor-factory-impl", TaskProperties.TASK_EXECUTOR_FACTORY_IMPL);
+    assertEquals("process-id", TaskProperties.PROCESS_ID);
+    assertEquals("unknown", TaskProperties.UNKNOWN_PROCESS_ID);
+    assertEquals("output_location", TaskProperties.OUTPUT_DIR);
+    assertEquals("move-files-to-hive-location", TaskProperties.MOVE_FILE_TO_HIVE_LOCATION);
+  }
+}

amoro-optimizer/amoro-optimizer-common/src/main/java/org/apache/amoro/optimizer/common/OptimizerExecutor.java

@@ -27,9 +27,9 @@
 import org.apache.amoro.optimizing.TaskProperties;
 import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
 import org.apache.amoro.shade.thrift.org.apache.thrift.TException;
+import org.apache.amoro.utils.DynConstructors;
 import org.apache.amoro.utils.ExceptionUtil;
 import org.apache.amoro.utils.SerializationUtil;
-import org.apache.iceberg.common.DynConstructors;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

amoro-optimizer/amoro-optimizer-common/src/test/java/org/apache/amoro/optimizer/common/TestDynConstructorsReplacement.java

@@ -0,0 +1,71 @@
+/*
+ * 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.optimizer.common;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import org.apache.amoro.optimizing.OptimizingExecutor;
+import org.apache.amoro.optimizing.OptimizingExecutorFactory;
+import org.apache.amoro.utils.DynConstructors;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+
+@DisplayName("Test amoro-common DynConstructors as replacement for Iceberg DynConstructors")
+public class TestDynConstructorsReplacement {
+
+  @Test
+  @DisplayName("Should load OptimizingExecutorFactory by class name")
+  void testLoadFactoryByClassName() throws Exception {
+    String className = TestOptimizerExecutor.TestOptimizingExecutorFactory.class.getName();
+    DynConstructors.Ctor<OptimizingExecutorFactory> ctor =
+        DynConstructors.builder(OptimizingExecutorFactory.class).impl(className).buildChecked();
+    OptimizingExecutorFactory<?> factory = ctor.newInstance();
+    assertNotNull(factory);
+  }
+
+  @Test
+  @DisplayName("Should throw when loading non-existent class")
+  void testLoadNonExistentClassThrows() {
+    assertThrows(
+        NoSuchMethodException.class,
+        () ->
+            DynConstructors.builder(OptimizingExecutorFactory.class)
+                .impl("com.nonexistent.FakeFactory")
+                .buildChecked());
+  }
+
+  @Test
+  @DisplayName("Should create executor from loaded factory")
+  @SuppressWarnings({"rawtypes", "unchecked"})
+  void testLoadedFactoryCreateExecutor() throws Exception {
+    String className = TestOptimizerExecutor.TestOptimizingExecutorFactory.class.getName();
+    DynConstructors.Ctor<OptimizingExecutorFactory> ctor =
+        DynConstructors.builder(OptimizingExecutorFactory.class).impl(className).buildChecked();
+    OptimizingExecutorFactory factory = ctor.newInstance();
+    factory.initialize(new HashMap<>());
+
+    TestOptimizerExecutor.TestOptimizingInput input =
+        TestOptimizerExecutor.TestOptimizingInput.successInput(42);
+    OptimizingExecutor<?> executor = factory.createExecutor(input);
+    assertNotNull(executor);
+  }
+}