diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroManagementConf.java b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroManagementConf.java index f38f6f5acc..824ee080d0 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroManagementConf.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroManagementConf.java @@ -121,6 +121,13 @@ public class AmoroManagementConf { "Sets the size of the worker pool. The worker pool limits the number of tasks concurrently processing " + "manifests in the base table implementation across all concurrent planning or commit operations."); + public static final ConfigOption TABLE_MANIFEST_IO_MAINTENANCE_THREAD_COUNT = + ConfigOptions.key("table-manifest-io.maintenance-thread-count") + .intType() + .defaultValue(10) + .withDescription( + "Sets the size of the worker pool used for manifest I/O across best-effort table maintenance operations."); + public static final ConfigOption TABLE_MANIFEST_IO_PLANNING_THREAD_COUNT = ConfigOptions.key("self-optimizing.plan-manifest-io-thread-count") .intType() diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java index e48961d9e9..2e72cef20b 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java @@ -610,9 +610,7 @@ private Map initEnvConfig() { return ConfigHelpers.convertConfigurationKeys(prefix, System.getenv()); } - /** - * Configures Iceberg's global worker pool and initializes self-optimizing Iceberg I/O pools. - */ + /** Configures Iceberg's global worker pool and initializes Amoro Iceberg I/O pools. */ private void initIcebergThreadPools() { int workerThreadPoolSize = Math.max( @@ -630,7 +628,10 @@ private void initIcebergThreadPools() { Math.max( Runtime.getRuntime().availableProcessors() / 2, serviceConfig.getInteger(AmoroManagementConf.TABLE_MANIFEST_IO_COMMIT_THREAD_COUNT)); + int maintenanceThreadPoolSize = + serviceConfig.getInteger(AmoroManagementConf.TABLE_MANIFEST_IO_MAINTENANCE_THREAD_COUNT); IcebergThreadPools.init(planningThreadPoolSize, commitThreadPoolSize); + IcebergThreadPools.initMaintenanceThreadPool(maintenanceThreadPoolSize); } private void initContainerConfig() { diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/TestAmoroManagementConf.java b/amoro-ams/src/test/java/org/apache/amoro/server/TestAmoroManagementConf.java index 414f760c2d..aff2a17084 100644 --- a/amoro-ams/src/test/java/org/apache/amoro/server/TestAmoroManagementConf.java +++ b/amoro-ams/src/test/java/org/apache/amoro/server/TestAmoroManagementConf.java @@ -88,6 +88,14 @@ void testNewDurationConfigDefaults() { Duration.ofDays(7), serviceConfig.get(AmoroManagementConf.PROCESS_HISTORY_DATA_KEEP_TIME)); } + @Test + void testMaintenanceManifestIoThreadCountDefault() { + Configurations serviceConfig = new Configurations(); + Assertions.assertEquals( + 10, + serviceConfig.getInteger(AmoroManagementConf.TABLE_MANIFEST_IO_MAINTENANCE_THREAD_COUNT)); + } + @Test void testDeprecatedIntegerConfigDefaults() { Configurations serviceConfig = new Configurations(); diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/maintainer/IcebergTableMaintainer.java b/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/maintainer/IcebergTableMaintainer.java index 68d90e3a17..5333856cea 100644 --- a/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/maintainer/IcebergTableMaintainer.java +++ b/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/maintainer/IcebergTableMaintainer.java @@ -41,6 +41,7 @@ import org.apache.amoro.shade.guava32.com.google.common.collect.Maps; import org.apache.amoro.shade.guava32.com.google.common.collect.Sets; import org.apache.amoro.table.TableIdentifier; +import org.apache.amoro.utils.IcebergThreadPools; import org.apache.amoro.utils.TableFileUtil; import org.apache.iceberg.ContentFile; import org.apache.iceberg.ContentScanTask; @@ -223,6 +224,7 @@ protected int expireSnapshots(long olderThan, int minCount, Set exclude) .retainLast(Math.max(minCount, 1)) .expireOlderThan(olderThan) .deleteWith(expiredFileCleaner::addFile) + .planWith(IcebergThreadPools.getMaintenanceExecutor()) .cleanExpiredFiles( true) /* enable clean only for collecting the expired files, will delete them later */; // iceberg auto-selects IncrementalFileCleanup for single-ref tables. That strategy walks the diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/utils/IcebergThreadPools.java b/amoro-format-iceberg/src/main/java/org/apache/amoro/utils/IcebergThreadPools.java index 96c0f8fe50..1a93173b30 100644 --- a/amoro-format-iceberg/src/main/java/org/apache/amoro/utils/IcebergThreadPools.java +++ b/amoro-format-iceberg/src/main/java/org/apache/amoro/utils/IcebergThreadPools.java @@ -33,6 +33,7 @@ public class IcebergThreadPools { private static final String PLANNING_POOL_NAME_PREFIX = "iceberg-planning-pool"; private static final String COMMIT_POOL_NAME_PREFIX = "iceberg-commit-pool"; + private static final String MAINTENANCE_POOL_NAME_PREFIX = "iceberg-maintenance-pool"; private static final Map POOL_SIZES = new ConcurrentHashMap<>(); private static final Map POOLS = new ConcurrentHashMap<>(); @@ -90,6 +91,35 @@ public static ExecutorService getCommitExecutor() { return getThreadPool(COMMIT_POOL_NAME_PREFIX); } + /** + * Initializes the process-wide Iceberg pool for best-effort table maintenance. + * + *

Repeated initialization is ignored because existing pools cannot be resized. + * + * @param poolSize number of worker threads + */ + public static void initMaintenanceThreadPool(int poolSize) { + newThreadPool(MAINTENANCE_POOL_NAME_PREFIX, poolSize); + } + + /** + * Return an {@link ExecutorService} for best-effort Iceberg table maintenance. + * + *

Snapshot expiration planning is the first consumer of this shared maintenance pool. Other + * maintenance operations can migrate to it as their Iceberg APIs expose executor hooks. + * + *

The size of this pool is controlled by the AMS configuration {@code + * table-manifest-io.maintenance-thread-count}. + * + *

Before the dedicated pool is initialized, this returns Iceberg's global worker pool. + * + * @return the maintenance pool, or Iceberg's global worker pool if the dedicated pool has not + * been initialized + */ + public static ExecutorService getMaintenanceExecutor() { + return getThreadPool(MAINTENANCE_POOL_NAME_PREFIX); + } + /** * Returns the registered Iceberg thread pool for the given name prefix. * diff --git a/amoro-format-iceberg/src/test/java/org/apache/amoro/formats/iceberg/maintainer/TestIcebergTableMaintainer.java b/amoro-format-iceberg/src/test/java/org/apache/amoro/formats/iceberg/maintainer/TestIcebergTableMaintainer.java new file mode 100644 index 0000000000..da5639af00 --- /dev/null +++ b/amoro-format-iceberg/src/test/java/org/apache/amoro/formats/iceberg/maintainer/TestIcebergTableMaintainer.java @@ -0,0 +1,62 @@ +/* + * 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.formats.iceberg.maintainer; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import org.apache.amoro.io.AuthenticatedFileIO; +import org.apache.amoro.maintainer.TableMaintainerContext; +import org.apache.amoro.table.TableIdentifier; +import org.apache.amoro.utils.IcebergThreadPools; +import org.apache.iceberg.ExpireSnapshots; +import org.apache.iceberg.Table; +import org.junit.jupiter.api.Test; + +class TestIcebergTableMaintainer { + + @Test + void testExpireSnapshotsUsesMaintenancePool() { + IcebergThreadPools.initMaintenanceThreadPool(1); + + Table table = mock(Table.class); + ExpireSnapshots expireSnapshots = mock(ExpireSnapshots.class); + when(table.name()).thenReturn("test_table"); + when(table.io()).thenReturn(mock(AuthenticatedFileIO.class)); + when(table.expireSnapshots()).thenReturn(expireSnapshots); + when(expireSnapshots.retainLast(1)).thenReturn(expireSnapshots); + when(expireSnapshots.expireOlderThan(100L)).thenReturn(expireSnapshots); + when(expireSnapshots.deleteWith(any())).thenReturn(expireSnapshots); + when(expireSnapshots.planWith(any())).thenReturn(expireSnapshots); + when(expireSnapshots.cleanExpiredFiles(true)).thenReturn(expireSnapshots); + + IcebergTableMaintainer tableMaintainer = + new IcebergTableMaintainer( + table, + TableIdentifier.of("test_catalog", "test_database", "test_table"), + mock(TableMaintainerContext.class)); + + tableMaintainer.expireSnapshots(100L, 1); + + verify(expireSnapshots).planWith(IcebergThreadPools.getMaintenanceExecutor()); + verify(expireSnapshots).commit(); + } +} diff --git a/amoro-format-iceberg/src/test/java/org/apache/amoro/utils/TestIcebergThreadPools.java b/amoro-format-iceberg/src/test/java/org/apache/amoro/utils/TestIcebergThreadPools.java index a5bfb75c1a..f1f3eafbf6 100644 --- a/amoro-format-iceberg/src/test/java/org/apache/amoro/utils/TestIcebergThreadPools.java +++ b/amoro-format-iceberg/src/test/java/org/apache/amoro/utils/TestIcebergThreadPools.java @@ -41,12 +41,17 @@ public void testDefaultPoolsAndInitialization() throws Exception { Assert.assertSame(registeredPool, IcebergThreadPools.getThreadPool("registered-test-pool")); IcebergThreadPools.init(1, 1); + IcebergThreadPools.initMaintenanceThreadPool(1); ExecutorService planningPool = IcebergThreadPools.getPlanningExecutor(); ExecutorService commitPool = IcebergThreadPools.getCommitExecutor(); + ExecutorService maintenancePool = IcebergThreadPools.getMaintenanceExecutor(); Assert.assertNotSame(workerPool, planningPool); Assert.assertNotSame(workerPool, commitPool); + Assert.assertNotSame(workerPool, maintenancePool); Assert.assertNotSame(planningPool, commitPool); + Assert.assertNotSame(planningPool, maintenancePool); + Assert.assertNotSame(commitPool, maintenancePool); Assert.assertTrue( planningPool .submit(() -> Thread.currentThread().getName()) @@ -57,9 +62,18 @@ public void testDefaultPoolsAndInitialization() throws Exception { .submit(() -> Thread.currentThread().getName()) .get(10, TimeUnit.SECONDS) .startsWith("iceberg-commit-pool-")); + Assert.assertTrue( + maintenancePool + .submit(() -> Thread.currentThread().getName()) + .get(10, TimeUnit.SECONDS) + .startsWith("iceberg-maintenance-pool-")); IcebergThreadPools.init(2, 2); + IcebergThreadPools.initMaintenanceThreadPool(2); Assert.assertSame(planningPool, IcebergThreadPools.getPlanningExecutor()); Assert.assertSame(commitPool, IcebergThreadPools.getCommitExecutor()); + Assert.assertSame(maintenancePool, IcebergThreadPools.getMaintenanceExecutor()); + Assert.assertThrows( + IllegalArgumentException.class, () -> IcebergThreadPools.initMaintenanceThreadPool(0)); } } diff --git a/charts/amoro/templates/amoro-configmap.yaml b/charts/amoro/templates/amoro-configmap.yaml index 7a9f14ba59..26e690f0f3 100644 --- a/charts/amoro/templates/amoro-configmap.yaml +++ b/charts/amoro/templates/amoro-configmap.yaml @@ -101,6 +101,7 @@ data: # optional features table-manifest-io: thread-count: 20 + maintenance-thread-count: 10 database: type: {{ .Values.amoroConf.database.type }} diff --git a/dist/src/main/amoro-bin/conf/config.yaml b/dist/src/main/amoro-bin/conf/config.yaml index f4cdaaac8f..d4e60010ac 100644 --- a/dist/src/main/amoro-bin/conf/config.yaml +++ b/dist/src/main/amoro-bin/conf/config.yaml @@ -110,6 +110,7 @@ ams: # optional features table-manifest-io: thread-count: 20 + maintenance-thread-count: 10 catalog-meta-cache: expiration-interval: 60s diff --git a/docs/configuration/ams-config.md b/docs/configuration/ams-config.md index 2a6c6b6e8e..e2746d79b6 100644 --- a/docs/configuration/ams-config.md +++ b/docs/configuration/ams-config.md @@ -116,6 +116,7 @@ table td:last-child, table th:last-child { width: 40%; word-break: break-all; } | self-optimizing.runtime-data-keep-time | 30 d | Duration that self-optimizing runtime data is retained. | | server-bind-host | 0.0.0.0 | The host bound to the server. | | server-expose-host | | The exposed host of the server. | +| table-manifest-io.maintenance-thread-count | 10 | Sets the size of the worker pool used for manifest I/O across best-effort table maintenance operations. | | table-manifest-io.thread-count | 20 | Sets the size of the worker pool. The worker pool limits the number of tasks concurrently processing manifests in the base table implementation across all concurrent planning or commit operations. | | terminal.backend | local | Terminal backend implementation. local, kyuubi and custom are valid values. | | terminal.factory | <undefined> | Session factory implement of terminal, `terminal.backend` must be `custom` if this is set. |