diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/BaseOptimizingInput.java b/amoro-common/src/main/java/org/apache/amoro/optimizing/BaseOptimizingInput.java similarity index 100% rename from amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/BaseOptimizingInput.java rename to amoro-common/src/main/java/org/apache/amoro/optimizing/BaseOptimizingInput.java diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/OptimizingCommitterFactory.java b/amoro-common/src/main/java/org/apache/amoro/optimizing/OptimizingCommitterFactory.java similarity index 100% rename from amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/OptimizingCommitterFactory.java rename to amoro-common/src/main/java/org/apache/amoro/optimizing/OptimizingCommitterFactory.java diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/OptimizingExecutor.java b/amoro-common/src/main/java/org/apache/amoro/optimizing/OptimizingExecutor.java similarity index 100% rename from amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/OptimizingExecutor.java rename to amoro-common/src/main/java/org/apache/amoro/optimizing/OptimizingExecutor.java diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/OptimizingExecutorFactory.java b/amoro-common/src/main/java/org/apache/amoro/optimizing/OptimizingExecutorFactory.java similarity index 100% rename from amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/OptimizingExecutorFactory.java rename to amoro-common/src/main/java/org/apache/amoro/optimizing/OptimizingExecutorFactory.java diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/TableOptimizing.java b/amoro-common/src/main/java/org/apache/amoro/optimizing/TableOptimizing.java similarity index 100% rename from amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/TableOptimizing.java rename to amoro-common/src/main/java/org/apache/amoro/optimizing/TableOptimizing.java diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/TaskProperties.java b/amoro-common/src/main/java/org/apache/amoro/optimizing/TaskProperties.java similarity index 76% rename from amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/TaskProperties.java rename to amoro-common/src/main/java/org/apache/amoro/optimizing/TaskProperties.java index 6a88da6b3f..23f12562f3 100644 --- a/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/TaskProperties.java +++ b/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 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 properties) { String processId = properties.get(PROCESS_ID); if (processId == null || processId.trim().isEmpty()) { diff --git a/amoro-common/src/test/java/org/apache/amoro/optimizing/TestOptimizingSpiInterfaces.java b/amoro-common/src/test/java/org/apache/amoro/optimizing/TestOptimizingSpiInterfaces.java new file mode 100644 index 0000000000..e9ae2fd4ee --- /dev/null +++ b/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 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 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; + } +} diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/AbstractRewriteFilesExecutor.java b/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/AbstractRewriteFilesExecutor.java index 4446619f92..7df3f085d3 100644 --- a/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/AbstractRewriteFilesExecutor.java +++ b/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 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, DeleteWriteResult> posWriter(); diff --git a/amoro-format-iceberg/src/test/java/org/apache/amoro/optimizing/TestTaskPropertiesSplit.java b/amoro-format-iceberg/src/test/java/org/apache/amoro/optimizing/TestTaskPropertiesSplit.java new file mode 100644 index 0000000000..2914a19783 --- /dev/null +++ b/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 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 properties = new HashMap<>(); + assertEquals(TaskProperties.UNKNOWN_PROCESS_ID, TaskProperties.getProcessId(properties)); + } + + @Test + @DisplayName("getProcessId should return UNKNOWN when property is empty") + void testGetProcessIdWithEmpty() { + Map 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); + } +} diff --git a/amoro-optimizer/amoro-optimizer-common/src/main/java/org/apache/amoro/optimizer/common/OptimizerExecutor.java b/amoro-optimizer/amoro-optimizer-common/src/main/java/org/apache/amoro/optimizer/common/OptimizerExecutor.java index 26eda902cc..f36d1c38c2 100644 --- a/amoro-optimizer/amoro-optimizer-common/src/main/java/org/apache/amoro/optimizer/common/OptimizerExecutor.java +++ b/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; diff --git a/amoro-optimizer/amoro-optimizer-common/src/test/java/org/apache/amoro/optimizer/common/TestDynConstructorsReplacement.java b/amoro-optimizer/amoro-optimizer-common/src/test/java/org/apache/amoro/optimizer/common/TestDynConstructorsReplacement.java new file mode 100644 index 0000000000..f569f2c4d1 --- /dev/null +++ b/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 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 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); + } +}