From 4bed59ec79990db3bcc7084d596e8e3f2e85a111 Mon Sep 17 00:00:00 2001 From: Power John Date: Sat, 8 Aug 2026 13:53:06 +0800 Subject: [PATCH 1/2] [AMORO-4264][AMS] Isolate snapshot expiration planning --- .../iceberg/IcebergProcessFactory.java | 6 ++ .../iceberg/TestIcebergProcessFactory.java | 33 ++++++++++ .../maintainer/IcebergTableMaintainer.java | 2 + .../amoro/utils/IcebergThreadPools.java | 31 ++++++++++ .../TestIcebergTableMaintainer.java | 62 +++++++++++++++++++ .../amoro/utils/TestIcebergThreadPools.java | 12 ++++ .../conf/plugins/process-factories.yaml | 1 + 7 files changed, 147 insertions(+) create mode 100644 amoro-format-iceberg/src/test/java/org/apache/amoro/formats/iceberg/maintainer/TestIcebergTableMaintainer.java diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/process/iceberg/IcebergProcessFactory.java b/amoro-ams/src/main/java/org/apache/amoro/server/process/iceberg/IcebergProcessFactory.java index 6df589cc94..4950b60711 100755 --- a/amoro-ams/src/main/java/org/apache/amoro/server/process/iceberg/IcebergProcessFactory.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/process/iceberg/IcebergProcessFactory.java @@ -35,6 +35,7 @@ import org.apache.amoro.server.table.DefaultTableRuntime; import org.apache.amoro.shade.guava32.com.google.common.collect.Lists; import org.apache.amoro.shade.guava32.com.google.common.collect.Maps; +import org.apache.amoro.utils.IcebergThreadPools; import org.apache.commons.lang3.tuple.Pair; import java.time.Duration; @@ -57,6 +58,9 @@ public class IcebergProcessFactory implements ProcessFactory { .durationType() .defaultValue(Duration.ofHours(1)); + public static final ConfigOption SNAPSHOT_EXPIRE_PLAN_THREAD_COUNT = + ConfigOptions.key("expire-snapshots.plan-thread-count").intType().defaultValue(10); + public static final ConfigOption ORPHAN_FILES_CLEANING_ENABLED = ConfigOptions.key("clean-orphan-files.enabled").booleanType().defaultValue(true); @@ -183,6 +187,8 @@ public void open(Map properties) { } Configurations configs = Configurations.fromMap(properties); if (configs.getBoolean(SNAPSHOT_EXPIRE_ENABLED)) { + IcebergThreadPools.initSnapshotExpirationThreadPool( + configs.getInteger(SNAPSHOT_EXPIRE_PLAN_THREAD_COUNT)); Duration interval = configs.getDuration(SNAPSHOT_EXPIRE_INTERVAL); this.actions.put( IcebergActions.EXPIRE_SNAPSHOTS, ProcessTriggerStrategy.triggerAtFixRate(interval)); diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/process/iceberg/TestIcebergProcessFactory.java b/amoro-ams/src/test/java/org/apache/amoro/server/process/iceberg/TestIcebergProcessFactory.java index 45aa9e9bfd..117e41a698 100644 --- a/amoro-ams/src/test/java/org/apache/amoro/server/process/iceberg/TestIcebergProcessFactory.java +++ b/amoro-ams/src/test/java/org/apache/amoro/server/process/iceberg/TestIcebergProcessFactory.java @@ -35,6 +35,8 @@ import org.apache.amoro.process.TableProcessStore; import org.apache.amoro.server.table.DefaultTableRuntime; import org.apache.amoro.server.table.cleanup.TableRuntimeCleanupState; +import org.apache.amoro.utils.IcebergThreadPools; +import org.apache.iceberg.util.ThreadPools; import org.junit.Assert; import org.junit.Test; @@ -44,6 +46,8 @@ import java.util.Map; import java.util.Optional; import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; public class TestIcebergProcessFactory { @@ -57,6 +61,35 @@ public void testOpenAndSupportedActions() { assertSupportedAction("sync-hive-tables", IcebergActions.SYNC_HIVE_TABLES, Duration.ofHours(1)); } + @Test + public void testOpenInitializesSnapshotExpirationThreadPool() throws Exception { + IcebergProcessFactory factory = new IcebergProcessFactory(); + Map properties = new HashMap<>(); + properties.put("expire-snapshots.enabled", "true"); + properties.put("expire-snapshots.interval", "1h"); + properties.put("expire-snapshots.plan-thread-count", "10"); + + factory.open(properties); + + ExecutorService snapshotExpirationPool = IcebergThreadPools.getSnapshotExpirationExecutor(); + Assert.assertNotSame(ThreadPools.getWorkerPool(), snapshotExpirationPool); + Assert.assertTrue( + snapshotExpirationPool + .submit(() -> Thread.currentThread().getName()) + .get(10, TimeUnit.SECONDS) + .startsWith("iceberg-snapshot-expiration-planning-pool-")); + } + + @Test + public void testOpenRejectsInvalidSnapshotExpirationThreadCount() { + IcebergProcessFactory factory = new IcebergProcessFactory(); + Map properties = new HashMap<>(); + properties.put("expire-snapshots.enabled", "true"); + properties.put("expire-snapshots.plan-thread-count", "0"); + + Assert.assertThrows(IllegalArgumentException.class, () -> factory.open(properties)); + } + @Test public void testTriggerActionWhenDue() { assertTriggerWhenDue( 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..b1fce12aaf 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.getSnapshotExpirationExecutor()) .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..fa6bb27bce 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,8 @@ 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 SNAPSHOT_EXPIRATION_POOL_NAME_PREFIX = + "iceberg-snapshot-expiration-planning-pool"; private static final Map POOL_SIZES = new ConcurrentHashMap<>(); private static final Map POOLS = new ConcurrentHashMap<>(); @@ -90,6 +92,35 @@ public static ExecutorService getCommitExecutor() { return getThreadPool(COMMIT_POOL_NAME_PREFIX); } + /** + * Initializes the process-wide Iceberg planning pool used by snapshot expiration. + * + *

Repeated initialization is ignored because existing pools cannot be resized. + * + * @param poolSize number of worker threads + */ + public static void initSnapshotExpirationThreadPool(int poolSize) { + newThreadPool(SNAPSHOT_EXPIRATION_POOL_NAME_PREFIX, poolSize); + } + + /** + * Return an {@link ExecutorService} that plans snapshot expiration file cleanup. + * + *

This pool isolates snapshot expiration from Iceberg's global worker pool and the + * self-optimizing planning and commit pools. + * + *

The size of this pool is controlled by the Iceberg process configuration {@code + * expire-snapshots.plan-thread-count}. + * + *

Before the dedicated pool is initialized, this returns Iceberg's global worker pool. + * + * @return the snapshot expiration planning pool, or Iceberg's global worker pool if the dedicated + * pool has not been initialized + */ + public static ExecutorService getSnapshotExpirationExecutor() { + return getThreadPool(SNAPSHOT_EXPIRATION_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..2dd8c0e0ca --- /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 testExpireSnapshotsUsesDedicatedPlanningPool() { + IcebergThreadPools.initSnapshotExpirationThreadPool(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.getSnapshotExpirationExecutor()); + 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..c8976106ae 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 @@ -44,9 +44,14 @@ public void testDefaultPoolsAndInitialization() throws Exception { ExecutorService planningPool = IcebergThreadPools.getPlanningExecutor(); ExecutorService commitPool = IcebergThreadPools.getCommitExecutor(); + IcebergThreadPools.initSnapshotExpirationThreadPool(1); + ExecutorService snapshotExpirationPool = IcebergThreadPools.getSnapshotExpirationExecutor(); Assert.assertNotSame(workerPool, planningPool); Assert.assertNotSame(workerPool, commitPool); + Assert.assertNotSame(workerPool, snapshotExpirationPool); Assert.assertNotSame(planningPool, commitPool); + Assert.assertNotSame(planningPool, snapshotExpirationPool); + Assert.assertNotSame(commitPool, snapshotExpirationPool); Assert.assertTrue( planningPool .submit(() -> Thread.currentThread().getName()) @@ -57,9 +62,16 @@ public void testDefaultPoolsAndInitialization() throws Exception { .submit(() -> Thread.currentThread().getName()) .get(10, TimeUnit.SECONDS) .startsWith("iceberg-commit-pool-")); + Assert.assertTrue( + snapshotExpirationPool + .submit(() -> Thread.currentThread().getName()) + .get(10, TimeUnit.SECONDS) + .startsWith("iceberg-snapshot-expiration-planning-pool-")); IcebergThreadPools.init(2, 2); + IcebergThreadPools.initSnapshotExpirationThreadPool(2); Assert.assertSame(planningPool, IcebergThreadPools.getPlanningExecutor()); Assert.assertSame(commitPool, IcebergThreadPools.getCommitExecutor()); + Assert.assertSame(snapshotExpirationPool, IcebergThreadPools.getSnapshotExpirationExecutor()); } } diff --git a/dist/src/main/amoro-bin/conf/plugins/process-factories.yaml b/dist/src/main/amoro-bin/conf/plugins/process-factories.yaml index 1085d364fe..e87dfe522a 100755 --- a/dist/src/main/amoro-bin/conf/plugins/process-factories.yaml +++ b/dist/src/main/amoro-bin/conf/plugins/process-factories.yaml @@ -22,6 +22,7 @@ process-factories: properties: expire-snapshots.enabled: "true" expire-snapshots.interval: "1h" + expire-snapshots.plan-thread-count: "10" clean-orphan-files.enabled: "true" clean-orphan-files.interval: "1d" clean-dangling-delete-files.enabled: "true" From 2f35a375f5b2aaa2a61fd6b7b1ac8fa5a99f300e Mon Sep 17 00:00:00 2001 From: Power John Date: Tue, 18 Aug 2026 13:53:05 +0800 Subject: [PATCH 2/2] [AMORO-4264][AMS] Use a shared pool for Iceberg maintenance planning --- .../amoro/server/AmoroManagementConf.java | 7 ++++ .../amoro/server/AmoroServiceContainer.java | 7 ++-- .../iceberg/IcebergProcessFactory.java | 6 ---- .../amoro/server/TestAmoroManagementConf.java | 8 +++++ .../iceberg/TestIcebergProcessFactory.java | 33 ------------------- .../maintainer/IcebergTableMaintainer.java | 2 +- .../amoro/utils/IcebergThreadPools.java | 27 ++++++++------- .../TestIcebergTableMaintainer.java | 6 ++-- .../amoro/utils/TestIcebergThreadPools.java | 20 ++++++----- charts/amoro/templates/amoro-configmap.yaml | 1 + dist/src/main/amoro-bin/conf/config.yaml | 1 + .../conf/plugins/process-factories.yaml | 1 - docs/configuration/ams-config.md | 1 + 13 files changed, 50 insertions(+), 70 deletions(-) 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/main/java/org/apache/amoro/server/process/iceberg/IcebergProcessFactory.java b/amoro-ams/src/main/java/org/apache/amoro/server/process/iceberg/IcebergProcessFactory.java index 4950b60711..6df589cc94 100755 --- a/amoro-ams/src/main/java/org/apache/amoro/server/process/iceberg/IcebergProcessFactory.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/process/iceberg/IcebergProcessFactory.java @@ -35,7 +35,6 @@ import org.apache.amoro.server.table.DefaultTableRuntime; import org.apache.amoro.shade.guava32.com.google.common.collect.Lists; import org.apache.amoro.shade.guava32.com.google.common.collect.Maps; -import org.apache.amoro.utils.IcebergThreadPools; import org.apache.commons.lang3.tuple.Pair; import java.time.Duration; @@ -58,9 +57,6 @@ public class IcebergProcessFactory implements ProcessFactory { .durationType() .defaultValue(Duration.ofHours(1)); - public static final ConfigOption SNAPSHOT_EXPIRE_PLAN_THREAD_COUNT = - ConfigOptions.key("expire-snapshots.plan-thread-count").intType().defaultValue(10); - public static final ConfigOption ORPHAN_FILES_CLEANING_ENABLED = ConfigOptions.key("clean-orphan-files.enabled").booleanType().defaultValue(true); @@ -187,8 +183,6 @@ public void open(Map properties) { } Configurations configs = Configurations.fromMap(properties); if (configs.getBoolean(SNAPSHOT_EXPIRE_ENABLED)) { - IcebergThreadPools.initSnapshotExpirationThreadPool( - configs.getInteger(SNAPSHOT_EXPIRE_PLAN_THREAD_COUNT)); Duration interval = configs.getDuration(SNAPSHOT_EXPIRE_INTERVAL); this.actions.put( IcebergActions.EXPIRE_SNAPSHOTS, ProcessTriggerStrategy.triggerAtFixRate(interval)); 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-ams/src/test/java/org/apache/amoro/server/process/iceberg/TestIcebergProcessFactory.java b/amoro-ams/src/test/java/org/apache/amoro/server/process/iceberg/TestIcebergProcessFactory.java index 117e41a698..45aa9e9bfd 100644 --- a/amoro-ams/src/test/java/org/apache/amoro/server/process/iceberg/TestIcebergProcessFactory.java +++ b/amoro-ams/src/test/java/org/apache/amoro/server/process/iceberg/TestIcebergProcessFactory.java @@ -35,8 +35,6 @@ import org.apache.amoro.process.TableProcessStore; import org.apache.amoro.server.table.DefaultTableRuntime; import org.apache.amoro.server.table.cleanup.TableRuntimeCleanupState; -import org.apache.amoro.utils.IcebergThreadPools; -import org.apache.iceberg.util.ThreadPools; import org.junit.Assert; import org.junit.Test; @@ -46,8 +44,6 @@ import java.util.Map; import java.util.Optional; import java.util.Set; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.TimeUnit; public class TestIcebergProcessFactory { @@ -61,35 +57,6 @@ public void testOpenAndSupportedActions() { assertSupportedAction("sync-hive-tables", IcebergActions.SYNC_HIVE_TABLES, Duration.ofHours(1)); } - @Test - public void testOpenInitializesSnapshotExpirationThreadPool() throws Exception { - IcebergProcessFactory factory = new IcebergProcessFactory(); - Map properties = new HashMap<>(); - properties.put("expire-snapshots.enabled", "true"); - properties.put("expire-snapshots.interval", "1h"); - properties.put("expire-snapshots.plan-thread-count", "10"); - - factory.open(properties); - - ExecutorService snapshotExpirationPool = IcebergThreadPools.getSnapshotExpirationExecutor(); - Assert.assertNotSame(ThreadPools.getWorkerPool(), snapshotExpirationPool); - Assert.assertTrue( - snapshotExpirationPool - .submit(() -> Thread.currentThread().getName()) - .get(10, TimeUnit.SECONDS) - .startsWith("iceberg-snapshot-expiration-planning-pool-")); - } - - @Test - public void testOpenRejectsInvalidSnapshotExpirationThreadCount() { - IcebergProcessFactory factory = new IcebergProcessFactory(); - Map properties = new HashMap<>(); - properties.put("expire-snapshots.enabled", "true"); - properties.put("expire-snapshots.plan-thread-count", "0"); - - Assert.assertThrows(IllegalArgumentException.class, () -> factory.open(properties)); - } - @Test public void testTriggerActionWhenDue() { assertTriggerWhenDue( 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 b1fce12aaf..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 @@ -224,7 +224,7 @@ protected int expireSnapshots(long olderThan, int minCount, Set exclude) .retainLast(Math.max(minCount, 1)) .expireOlderThan(olderThan) .deleteWith(expiredFileCleaner::addFile) - .planWith(IcebergThreadPools.getSnapshotExpirationExecutor()) + .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 fa6bb27bce..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,8 +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 SNAPSHOT_EXPIRATION_POOL_NAME_PREFIX = - "iceberg-snapshot-expiration-planning-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<>(); @@ -93,32 +92,32 @@ public static ExecutorService getCommitExecutor() { } /** - * Initializes the process-wide Iceberg planning pool used by snapshot expiration. + * 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 initSnapshotExpirationThreadPool(int poolSize) { - newThreadPool(SNAPSHOT_EXPIRATION_POOL_NAME_PREFIX, poolSize); + public static void initMaintenanceThreadPool(int poolSize) { + newThreadPool(MAINTENANCE_POOL_NAME_PREFIX, poolSize); } /** - * Return an {@link ExecutorService} that plans snapshot expiration file cleanup. + * Return an {@link ExecutorService} for best-effort Iceberg table maintenance. * - *

This pool isolates snapshot expiration from Iceberg's global worker pool and the - * self-optimizing planning and commit pools. + *

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 Iceberg process configuration {@code - * expire-snapshots.plan-thread-count}. + *

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 snapshot expiration planning pool, or Iceberg's global worker pool if the dedicated - * pool has not been initialized + * @return the maintenance pool, or Iceberg's global worker pool if the dedicated pool has not + * been initialized */ - public static ExecutorService getSnapshotExpirationExecutor() { - return getThreadPool(SNAPSHOT_EXPIRATION_POOL_NAME_PREFIX); + public static ExecutorService getMaintenanceExecutor() { + return getThreadPool(MAINTENANCE_POOL_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 index 2dd8c0e0ca..da5639af00 100644 --- 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 @@ -34,8 +34,8 @@ class TestIcebergTableMaintainer { @Test - void testExpireSnapshotsUsesDedicatedPlanningPool() { - IcebergThreadPools.initSnapshotExpirationThreadPool(1); + void testExpireSnapshotsUsesMaintenancePool() { + IcebergThreadPools.initMaintenanceThreadPool(1); Table table = mock(Table.class); ExpireSnapshots expireSnapshots = mock(ExpireSnapshots.class); @@ -56,7 +56,7 @@ void testExpireSnapshotsUsesDedicatedPlanningPool() { tableMaintainer.expireSnapshots(100L, 1); - verify(expireSnapshots).planWith(IcebergThreadPools.getSnapshotExpirationExecutor()); + 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 c8976106ae..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,17 +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(); - IcebergThreadPools.initSnapshotExpirationThreadPool(1); - ExecutorService snapshotExpirationPool = IcebergThreadPools.getSnapshotExpirationExecutor(); + ExecutorService maintenancePool = IcebergThreadPools.getMaintenanceExecutor(); Assert.assertNotSame(workerPool, planningPool); Assert.assertNotSame(workerPool, commitPool); - Assert.assertNotSame(workerPool, snapshotExpirationPool); + Assert.assertNotSame(workerPool, maintenancePool); Assert.assertNotSame(planningPool, commitPool); - Assert.assertNotSame(planningPool, snapshotExpirationPool); - Assert.assertNotSame(commitPool, snapshotExpirationPool); + Assert.assertNotSame(planningPool, maintenancePool); + Assert.assertNotSame(commitPool, maintenancePool); Assert.assertTrue( planningPool .submit(() -> Thread.currentThread().getName()) @@ -63,15 +63,17 @@ public void testDefaultPoolsAndInitialization() throws Exception { .get(10, TimeUnit.SECONDS) .startsWith("iceberg-commit-pool-")); Assert.assertTrue( - snapshotExpirationPool + maintenancePool .submit(() -> Thread.currentThread().getName()) .get(10, TimeUnit.SECONDS) - .startsWith("iceberg-snapshot-expiration-planning-pool-")); + .startsWith("iceberg-maintenance-pool-")); IcebergThreadPools.init(2, 2); - IcebergThreadPools.initSnapshotExpirationThreadPool(2); + IcebergThreadPools.initMaintenanceThreadPool(2); Assert.assertSame(planningPool, IcebergThreadPools.getPlanningExecutor()); Assert.assertSame(commitPool, IcebergThreadPools.getCommitExecutor()); - Assert.assertSame(snapshotExpirationPool, IcebergThreadPools.getSnapshotExpirationExecutor()); + 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/dist/src/main/amoro-bin/conf/plugins/process-factories.yaml b/dist/src/main/amoro-bin/conf/plugins/process-factories.yaml index e87dfe522a..1085d364fe 100755 --- a/dist/src/main/amoro-bin/conf/plugins/process-factories.yaml +++ b/dist/src/main/amoro-bin/conf/plugins/process-factories.yaml @@ -22,7 +22,6 @@ process-factories: properties: expire-snapshots.enabled: "true" expire-snapshots.interval: "1h" - expire-snapshots.plan-thread-count: "10" clean-orphan-files.enabled: "true" clean-orphan-files.interval: "1d" clean-dangling-delete-files.enabled: "true" 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. |