deletedOrNotFound() {
+ return deletedOrNotFound;
+ }
+}
diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/BatchFileDeleter.java b/paimon-common/src/main/java/org/apache/paimon/fs/BatchFileDeleter.java
new file mode 100644
index 000000000000..bfc0996ea81f
--- /dev/null
+++ b/paimon-common/src/main/java/org/apache/paimon/fs/BatchFileDeleter.java
@@ -0,0 +1,49 @@
+/*
+ * 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.paimon.fs;
+
+import org.apache.paimon.annotation.Public;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * Deletes files in one provider request without falling back to individual deletes.
+ *
+ * A successful invocation confirms every requested file as deleted or not found. A failure or
+ * timeout only means that the complete batch was not confirmed; the provider may already have
+ * deleted some files. If a caller retries, it must retry the same complete batch. Implementations
+ * must validate the complete request before accessing storage.
+ *
+ * @since 2.1
+ */
+@Public
+public interface BatchFileDeleter {
+
+ /** Maximum number of files accepted by one {@link #delete(List)} invocation. */
+ int maxBatchSize();
+
+ /**
+ * Deletes one non-empty batch.
+ *
+ * @return files confirmed deleted or not found
+ * @throws IOException if any requested file cannot be confirmed
+ */
+ BatchDeleteResult delete(List files) throws IOException;
+}
diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java b/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java
index 2b0dcec3f760..98d386c702a4 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java
@@ -217,6 +217,18 @@ default FileStatus[] listDirectories(Path path) throws IOException {
*/
boolean exists(Path path) throws IOException;
+ /**
+ * Returns a strict batch-delete capability for the provider serving the given path.
+ *
+ * An empty result is the only signal that callers may use individual deletes instead. The
+ * default performs no storage access and preserves compatibility with existing providers.
+ *
+ * @since 2.1
+ */
+ default Optional batchFileDeleter(Path path) throws IOException {
+ return Optional.empty();
+ }
+
/**
* Delete a file.
*
diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/PluginFileIO.java b/paimon-common/src/main/java/org/apache/paimon/fs/PluginFileIO.java
index 587c1f2d4423..24d8868d1eec 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fs/PluginFileIO.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fs/PluginFileIO.java
@@ -24,6 +24,9 @@
import java.io.IOException;
import java.time.Duration;
+import java.util.List;
+import java.util.Optional;
+import java.util.function.Supplier;
/**
* A {@link FileIO} for plugin jar. {@link FileIO} is serializable, so plugin FileIO should be
@@ -78,6 +81,28 @@ public boolean exists(Path path) throws IOException {
return wrap(() -> fileIO(path).exists(path));
}
+ @Override
+ public Optional batchFileDeleter(Path path) throws IOException {
+ Optional capability = wrap(() -> fileIO(path).batchFileDeleter(path));
+ if (!capability.isPresent()) {
+ return Optional.empty();
+ }
+
+ BatchFileDeleter delegate = capability.get();
+ return Optional.of(
+ new BatchFileDeleter() {
+ @Override
+ public int maxBatchSize() {
+ return wrapUnchecked(delegate::maxBatchSize);
+ }
+
+ @Override
+ public BatchDeleteResult delete(List files) throws IOException {
+ return wrap(() -> delegate.delete(files));
+ }
+ });
+ }
+
@Override
public boolean delete(Path path, boolean recursive) throws IOException {
return wrap(() -> fileIO(path).delete(path, recursive));
@@ -132,6 +157,16 @@ private T wrap(Func func) throws IOException {
}
}
+ private T wrapUnchecked(Supplier supplier) {
+ ClassLoader cl = Thread.currentThread().getContextClassLoader();
+ try {
+ Thread.currentThread().setContextClassLoader(pluginClassLoader());
+ return supplier.get();
+ } finally {
+ Thread.currentThread().setContextClassLoader(cl);
+ }
+ }
+
/** Apply function with wrapping classloader. */
@FunctionalInterface
protected interface Func {
diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/ResolvingFileIO.java b/paimon-common/src/main/java/org/apache/paimon/fs/ResolvingFileIO.java
index 5568ba896cb3..f2475d985534 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fs/ResolvingFileIO.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fs/ResolvingFileIO.java
@@ -26,10 +26,14 @@
import java.io.IOException;
import java.io.Serializable;
+import java.net.URI;
import java.time.Duration;
+import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Supplier;
import static org.apache.paimon.options.CatalogOptions.RESOLVING_FILE_IO_ENABLED;
@@ -94,6 +98,30 @@ public boolean exists(Path path) throws IOException {
return wrap(() -> fileIO(path).exists(path));
}
+ @Override
+ public Optional batchFileDeleter(Path path) throws IOException {
+ Optional capability = wrap(() -> fileIO(path).batchFileDeleter(path));
+ if (!capability.isPresent()) {
+ return Optional.empty();
+ }
+
+ URI provider = path.toUri();
+ BatchFileDeleter delegate = capability.get();
+ return Optional.of(
+ new BatchFileDeleter() {
+ @Override
+ public int maxBatchSize() {
+ return wrapUnchecked(delegate::maxBatchSize);
+ }
+
+ @Override
+ public BatchDeleteResult delete(List files) throws IOException {
+ validateProvider(files, provider);
+ return wrap(() -> delegate.delete(files));
+ }
+ });
+ }
+
@Override
public boolean delete(Path path, boolean recursive) throws IOException {
return wrap(() -> fileIO(path).delete(path, recursive));
@@ -149,6 +177,30 @@ private T wrap(Func func) throws IOException {
}
}
+ private T wrapUnchecked(Supplier supplier) {
+ ClassLoader cl = Thread.currentThread().getContextClassLoader();
+ try {
+ Thread.currentThread().setContextClassLoader(ResolvingFileIO.class.getClassLoader());
+ return supplier.get();
+ } finally {
+ Thread.currentThread().setContextClassLoader(cl);
+ }
+ }
+
+ private static void validateProvider(List files, URI provider) {
+ if (files == null) {
+ throw new IllegalArgumentException("Batch delete files must not be null.");
+ }
+ for (Path file : files) {
+ if (file == null
+ || !Objects.equals(provider.getScheme(), file.toUri().getScheme())
+ || !Objects.equals(provider.getAuthority(), file.toUri().getAuthority())) {
+ throw new IllegalArgumentException(
+ "Batch delete files must use the capability provider's scheme and authority.");
+ }
+ }
+ }
+
/** Apply function with wrapping classloader. */
@FunctionalInterface
protected interface Func {
diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/cache/CachingFileIO.java b/paimon-common/src/main/java/org/apache/paimon/fs/cache/CachingFileIO.java
index 65eeaa3ebfd7..e21e237769ec 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fs/cache/CachingFileIO.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fs/cache/CachingFileIO.java
@@ -20,6 +20,7 @@
import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.data.BlobDescriptor;
+import org.apache.paimon.fs.BatchFileDeleter;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.FileStatus;
import org.apache.paimon.fs.Path;
@@ -40,6 +41,7 @@
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
@@ -165,6 +167,11 @@ public boolean exists(Path path) throws IOException {
return delegate.exists(path);
}
+ @Override
+ public Optional batchFileDeleter(Path path) throws IOException {
+ return delegate.batchFileDeleter(path);
+ }
+
@Override
public boolean delete(Path path, boolean recursive) throws IOException {
return delegate.delete(path, recursive);
diff --git a/paimon-common/src/main/java/org/apache/paimon/rest/RESTTokenFileIO.java b/paimon-common/src/main/java/org/apache/paimon/rest/RESTTokenFileIO.java
index fb210dda435f..982cad5818f0 100644
--- a/paimon-common/src/main/java/org/apache/paimon/rest/RESTTokenFileIO.java
+++ b/paimon-common/src/main/java/org/apache/paimon/rest/RESTTokenFileIO.java
@@ -21,6 +21,8 @@
import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.BlobDescriptor;
+import org.apache.paimon.fs.BatchDeleteResult;
+import org.apache.paimon.fs.BatchFileDeleter;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.FileStatus;
import org.apache.paimon.fs.Path;
@@ -47,7 +49,9 @@
import java.io.IOException;
import java.io.UncheckedIOException;
import java.time.Duration;
+import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
@@ -145,6 +149,33 @@ public boolean exists(Path path) throws IOException {
return fileIO().exists(path);
}
+ @Override
+ public Optional batchFileDeleter(Path path) throws IOException {
+ Optional capability = fileIO().batchFileDeleter(path);
+ if (!capability.isPresent()) {
+ return Optional.empty();
+ }
+
+ int maxBatchSize = capability.get().maxBatchSize();
+ return Optional.of(
+ new BatchFileDeleter() {
+ @Override
+ public int maxBatchSize() {
+ return maxBatchSize;
+ }
+
+ @Override
+ public BatchDeleteResult delete(List files) throws IOException {
+ Optional current = fileIO().batchFileDeleter(path);
+ if (!current.isPresent()) {
+ throw new IOException(
+ "Batch delete capability is unavailable after refreshing credentials.");
+ }
+ return current.get().delete(files);
+ }
+ });
+ }
+
@Override
public boolean delete(Path path, boolean recursive) throws IOException {
return fileIO().delete(path, recursive);
diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/FileIOBatchDeleteContractTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/FileIOBatchDeleteContractTest.java
new file mode 100644
index 000000000000..8f3d5cd2815d
--- /dev/null
+++ b/paimon-common/src/test/java/org/apache/paimon/fs/FileIOBatchDeleteContractTest.java
@@ -0,0 +1,202 @@
+/*
+ * 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.paimon.fs;
+
+import org.apache.paimon.catalog.CatalogContext;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import javax.tools.JavaCompiler;
+import javax.tools.ToolProvider;
+
+import java.io.IOException;
+import java.net.URLClassLoader;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Public contract and binary compatibility tests for strict batch delete. */
+class FileIOBatchDeleteContractTest {
+
+ private static final Path FIRST = new Path("oss://bucket/table/a.parquet");
+ private static final Path SECOND = new Path("oss://bucket/table/b.parquet");
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ void testLegacyImplementationUsesDefaultUnsupportedWithoutStorageAccess() throws Exception {
+ LegacyFileIO legacy = new LegacyFileIO();
+
+ Optional capability = legacy.batchFileDeleter(FIRST);
+
+ assertThat(capability).isEmpty();
+ assertThat(legacy.storageCalls).hasValue(0);
+ }
+
+ @Test
+ void testBatchDeleteResultDefensivelyCopiesAndDoesNotExposeMutableState() {
+ List callerOwned = new ArrayList<>(Arrays.asList(FIRST, SECOND));
+
+ BatchDeleteResult result = new BatchDeleteResult(callerOwned);
+ callerOwned.clear();
+
+ assertThat(result.deletedOrNotFound()).containsExactly(FIRST, SECOND);
+ assertThatThrownBy(() -> result.deletedOrNotFound().add(FIRST))
+ .isInstanceOf(UnsupportedOperationException.class);
+ assertThatThrownBy(() -> result.deletedOrNotFound().set(0, SECOND))
+ .isInstanceOf(UnsupportedOperationException.class);
+ assertThat(result.deletedOrNotFound()).containsExactly(FIRST, SECOND);
+ }
+
+ @Test
+ void testProviderCompiledAgainstOldInterfaceLoadsAndUsesNewDefaultMethod() throws Exception {
+ java.nio.file.Path sources = Files.createDirectories(tempDir.resolve("sources"));
+ java.nio.file.Path oldApiClasses = Files.createDirectories(tempDir.resolve("old-api"));
+ java.nio.file.Path providerClasses = Files.createDirectories(tempDir.resolve("provider"));
+ java.nio.file.Path oldInterface =
+ writeSource(
+ sources,
+ "org/apache/paimon/fs/FileIO.java",
+ "package org.apache.paimon.fs;\n"
+ + "public interface FileIO extends java.io.Serializable {}\n");
+ java.nio.file.Path oldProvider =
+ writeSource(
+ sources,
+ "fixture/LegacyProvider.java",
+ "package fixture;\n"
+ + "public final class LegacyProvider "
+ + "implements org.apache.paimon.fs.FileIO {\n"
+ + " public LegacyProvider() {}\n"
+ + "}\n");
+ JavaCompiler compiler = ToolProvider.getSystemJavaCompiler();
+ assertThat(compiler).as("Maven tests must run on a JDK").isNotNull();
+ assertThat(
+ compiler.run(
+ null,
+ null,
+ null,
+ "-d",
+ oldApiClasses.toString(),
+ oldInterface.toString()))
+ .isZero();
+ assertThat(
+ compiler.run(
+ null,
+ null,
+ null,
+ "-classpath",
+ oldApiClasses.toString(),
+ "-d",
+ providerClasses.toString(),
+ oldProvider.toString()))
+ .isZero();
+
+ // Parent-first loading replaces the compile-time interface with the current FileIO while
+ // retaining provider bytecode compiled without the new method.
+ try (URLClassLoader loader =
+ new URLClassLoader(
+ new java.net.URL[] {providerClasses.toUri().toURL()},
+ FileIO.class.getClassLoader())) {
+ Class> providerClass = Class.forName("fixture.LegacyProvider", true, loader);
+ assertThat(providerClass.getInterfaces()).containsExactly(FileIO.class);
+ FileIO provider = (FileIO) providerClass.getDeclaredConstructor().newInstance();
+
+ assertThat(FileIO.class.getMethod("batchFileDeleter", Path.class).isDefault()).isTrue();
+ assertThat(provider.batchFileDeleter(FIRST)).isEmpty();
+ }
+ }
+
+ private static java.nio.file.Path writeSource(
+ java.nio.file.Path root, String relative, String source) throws IOException {
+ java.nio.file.Path file = root.resolve(relative);
+ Files.createDirectories(file.getParent());
+ Files.write(file, source.getBytes(StandardCharsets.UTF_8));
+ return file;
+ }
+
+ /**
+ * This fixture intentionally does not override batchFileDeleter. Every observable storage
+ * method fails, so even a harmless-looking capability probe has causal evidence.
+ */
+ private static class LegacyFileIO implements FileIO {
+
+ private final AtomicInteger storageCalls = new AtomicInteger();
+
+ @Override
+ public boolean isObjectStore() {
+ return true;
+ }
+
+ @Override
+ public void configure(CatalogContext context) {}
+
+ @Override
+ public SeekableInputStream newInputStream(Path path) {
+ return unexpectedStorageCall("newInputStream");
+ }
+
+ @Override
+ public PositionOutputStream newOutputStream(Path path, boolean overwrite) {
+ return unexpectedStorageCall("newOutputStream");
+ }
+
+ @Override
+ public FileStatus getFileStatus(Path path) {
+ return unexpectedStorageCall("getFileStatus");
+ }
+
+ @Override
+ public FileStatus[] listStatus(Path path) {
+ return unexpectedStorageCall("listStatus");
+ }
+
+ @Override
+ public boolean exists(Path path) {
+ return unexpectedStorageCall("exists");
+ }
+
+ @Override
+ public boolean delete(Path path, boolean recursive) {
+ return unexpectedStorageCall("delete");
+ }
+
+ @Override
+ public boolean mkdirs(Path path) {
+ return unexpectedStorageCall("mkdirs");
+ }
+
+ @Override
+ public boolean rename(Path src, Path dst) {
+ return unexpectedStorageCall("rename");
+ }
+
+ private T unexpectedStorageCall(String operation) {
+ storageCalls.incrementAndGet();
+ throw new AssertionError("Default capability accessed storage through " + operation);
+ }
+ }
+}
diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/FileIOBatchDeleteForwardingTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/FileIOBatchDeleteForwardingTest.java
new file mode 100644
index 000000000000..abb5f25cf555
--- /dev/null
+++ b/paimon-common/src/test/java/org/apache/paimon/fs/FileIOBatchDeleteForwardingTest.java
@@ -0,0 +1,843 @@
+/*
+ * 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.paimon.fs;
+
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.fs.cache.CachingFileIO;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.rest.RESTApi;
+import org.apache.paimon.rest.RESTTokenFileIO;
+import org.apache.paimon.rest.responses.GetTableTokenResponse;
+import org.apache.paimon.utils.FileType;
+import org.apache.paimon.utils.InstantiationUtil;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.List;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/** Contract tests for forwarding strict batch-delete capabilities through FileIO wrappers. */
+class FileIOBatchDeleteForwardingTest {
+
+ private static final Path FIRST = new Path("oss://bucket/table/a.parquet");
+ private static final Path SECOND = new Path("oss://bucket/table/b.parquet");
+ private static final List FILES = Arrays.asList(FIRST, SECOND);
+
+ @Test
+ void testPluginForwardsSupportedCapabilityUnderPluginClassLoader() throws Exception {
+ FileIO delegate = mock(FileIO.class);
+ ClassLoader pluginClassLoader = new ClassLoader() {};
+ ClassLoader original = Thread.currentThread().getContextClassLoader();
+ BatchDeleteResult expected = result(FILES);
+ BatchFileDeleter inner =
+ new BatchFileDeleter() {
+ @Override
+ public int maxBatchSize() {
+ assertThat(Thread.currentThread().getContextClassLoader())
+ .isSameAs(pluginClassLoader);
+ return 1000;
+ }
+
+ @Override
+ public BatchDeleteResult delete(List files) {
+ assertThat(Thread.currentThread().getContextClassLoader())
+ .isSameAs(pluginClassLoader);
+ assertThat(files).containsExactlyElementsOf(FILES);
+ return expected;
+ }
+ };
+ when(delegate.batchFileDeleter(FIRST))
+ .thenAnswer(
+ ignored -> {
+ assertThat(Thread.currentThread().getContextClassLoader())
+ .isSameAs(pluginClassLoader);
+ return Optional.of(inner);
+ });
+ TestPluginFileIO plugin = new TestPluginFileIO(delegate, pluginClassLoader);
+
+ // A broken TCCL restore can poison later ServiceLoader tests in the same worker, so the
+ // fixture restores the caller loader independently of the production finally block.
+ try {
+ BatchFileDeleter forwarded =
+ plugin.batchFileDeleter(FIRST).orElseThrow(AssertionError::new);
+
+ assertThat(forwarded.maxBatchSize()).isEqualTo(1000);
+ assertThat(forwarded.delete(FILES)).isSameAs(expected);
+ assertThat(Thread.currentThread().getContextClassLoader()).isSameAs(original);
+ verify(delegate).batchFileDeleter(FIRST);
+ verify(delegate, never()).delete(any(), anyBoolean());
+ } finally {
+ Thread.currentThread().setContextClassLoader(original);
+ }
+ }
+
+ @Test
+ void testPluginForwardsUnsupportedCapability() throws Exception {
+ FileIO delegate = mock(FileIO.class);
+ ClassLoader pluginClassLoader = new ClassLoader() {};
+ TestPluginFileIO plugin = new TestPluginFileIO(delegate, pluginClassLoader);
+ when(delegate.batchFileDeleter(FIRST)).thenReturn(Optional.empty());
+ ClassLoader previous = Thread.currentThread().getContextClassLoader();
+
+ try {
+ assertThat(plugin.batchFileDeleter(FIRST)).isEmpty();
+ assertThat(Thread.currentThread().getContextClassLoader()).isSameAs(previous);
+ verify(delegate, never()).delete(any(), anyBoolean());
+ } finally {
+ Thread.currentThread().setContextClassLoader(previous);
+ }
+ }
+
+ @Test
+ void testPluginPropagatesDiscoveryFailureAndRestoresCallerClassLoader() throws Exception {
+ FileIO delegate = mock(FileIO.class);
+ ClassLoader pluginClassLoader = new ClassLoader() {};
+ ClassLoader previous = Thread.currentThread().getContextClassLoader();
+ ClassLoader callerClassLoader = new ClassLoader(previous) {};
+ IOException failure = new IOException("plugin discovery failed");
+ when(delegate.batchFileDeleter(FIRST))
+ .thenAnswer(
+ ignored -> {
+ assertThat(Thread.currentThread().getContextClassLoader())
+ .isSameAs(pluginClassLoader);
+ throw failure;
+ });
+ TestPluginFileIO plugin = new TestPluginFileIO(delegate, pluginClassLoader);
+
+ Thread.currentThread().setContextClassLoader(callerClassLoader);
+ try {
+ assertThatThrownBy(() -> plugin.batchFileDeleter(FIRST)).isSameAs(failure);
+ assertThat(Thread.currentThread().getContextClassLoader()).isSameAs(callerClassLoader);
+ verify(delegate, never()).delete(any(), anyBoolean());
+ } finally {
+ Thread.currentThread().setContextClassLoader(previous);
+ }
+ }
+
+ @Test
+ void testPluginPropagatesDeleteFailureAndRestoresCallerClassLoader() throws Exception {
+ FileIO delegate = mock(FileIO.class);
+ ClassLoader pluginClassLoader = new ClassLoader() {};
+ ClassLoader previous = Thread.currentThread().getContextClassLoader();
+ ClassLoader callerClassLoader = new ClassLoader(previous) {};
+ IOException failure = new IOException("plugin batch failed");
+ when(delegate.batchFileDeleter(FIRST))
+ .thenReturn(
+ Optional.of(
+ new BatchFileDeleter() {
+ @Override
+ public int maxBatchSize() {
+ assertThat(Thread.currentThread().getContextClassLoader())
+ .isSameAs(pluginClassLoader);
+ return 1000;
+ }
+
+ @Override
+ public BatchDeleteResult delete(List files)
+ throws IOException {
+ assertThat(Thread.currentThread().getContextClassLoader())
+ .isSameAs(pluginClassLoader);
+ throw failure;
+ }
+ }));
+ TestPluginFileIO plugin = new TestPluginFileIO(delegate, pluginClassLoader);
+
+ Thread.currentThread().setContextClassLoader(callerClassLoader);
+ try {
+ BatchFileDeleter forwarded =
+ plugin.batchFileDeleter(FIRST).orElseThrow(AssertionError::new);
+ assertThat(forwarded.maxBatchSize()).isEqualTo(1000);
+ assertThatThrownBy(() -> forwarded.delete(FILES)).isSameAs(failure);
+ assertThat(Thread.currentThread().getContextClassLoader()).isSameAs(callerClassLoader);
+ verify(delegate, never()).delete(any(), anyBoolean());
+ } finally {
+ Thread.currentThread().setContextClassLoader(previous);
+ }
+ }
+
+ @Test
+ void testPluginPropagatesMaxBatchSizeFailureAndRestoresCallerClassLoader() throws Exception {
+ FileIO delegate = mock(FileIO.class);
+ ClassLoader pluginClassLoader = new ClassLoader() {};
+ ClassLoader previous = Thread.currentThread().getContextClassLoader();
+ ClassLoader callerClassLoader = new ClassLoader(previous) {};
+ RuntimeException failure = new RuntimeException("plugin max batch size failed");
+ when(delegate.batchFileDeleter(FIRST))
+ .thenReturn(
+ Optional.of(
+ new BatchFileDeleter() {
+ @Override
+ public int maxBatchSize() {
+ assertThat(Thread.currentThread().getContextClassLoader())
+ .isSameAs(pluginClassLoader);
+ throw failure;
+ }
+
+ @Override
+ public BatchDeleteResult delete(List files) {
+ throw new AssertionError("delete must not be called");
+ }
+ }));
+ TestPluginFileIO plugin = new TestPluginFileIO(delegate, pluginClassLoader);
+
+ Thread.currentThread().setContextClassLoader(callerClassLoader);
+ try {
+ BatchFileDeleter forwarded =
+ plugin.batchFileDeleter(FIRST).orElseThrow(AssertionError::new);
+ assertThatThrownBy(forwarded::maxBatchSize).isSameAs(failure);
+ assertThat(Thread.currentThread().getContextClassLoader()).isSameAs(callerClassLoader);
+ verify(delegate, never()).delete(any(), anyBoolean());
+ } finally {
+ Thread.currentThread().setContextClassLoader(previous);
+ }
+ }
+
+ @Test
+ void testResolvingForwardsSupportedAndUnsupportedCapabilities() throws Exception {
+ FileIO supportedDelegate = mock(FileIO.class);
+ BatchDeleteResult expected = result(FILES);
+ BatchFileDeleter inner = deleter(1000, files -> expected);
+ when(supportedDelegate.batchFileDeleter(FIRST)).thenReturn(Optional.of(inner));
+ ResolvingFileIO supported = resolving(supportedDelegate);
+
+ BatchFileDeleter forwarded =
+ supported.batchFileDeleter(FIRST).orElseThrow(AssertionError::new);
+ assertThat(forwarded.maxBatchSize()).isEqualTo(1000);
+ assertThat(forwarded.delete(FILES)).isSameAs(expected);
+
+ FileIO unsupportedDelegate = mock(FileIO.class);
+ when(unsupportedDelegate.batchFileDeleter(FIRST)).thenReturn(Optional.empty());
+ assertThat(resolving(unsupportedDelegate).batchFileDeleter(FIRST)).isEmpty();
+ }
+
+ @Test
+ void testResolvingRejectsMixedAuthorityBeforeProviderInvocation() throws Exception {
+ FileIO delegate = mock(FileIO.class);
+ AtomicInteger providerCalls = new AtomicInteger();
+ when(delegate.batchFileDeleter(FIRST))
+ .thenReturn(
+ Optional.of(
+ deleter(
+ 1000,
+ files -> {
+ providerCalls.incrementAndGet();
+ return result(files);
+ })));
+ BatchFileDeleter forwarded =
+ resolving(delegate).batchFileDeleter(FIRST).orElseThrow(AssertionError::new);
+
+ assertThatThrownBy(
+ () ->
+ forwarded.delete(
+ Arrays.asList(
+ FIRST,
+ new Path("oss://other-bucket/table/b.parquet"))))
+ .isInstanceOfAny(IllegalArgumentException.class, IOException.class);
+ assertThat(providerCalls).hasValue(0);
+ verify(delegate, never()).delete(any(), anyBoolean());
+ }
+
+ @Test
+ void testResolvingRejectsMixedSchemeBeforeProviderInvocation() throws Exception {
+ FileIO delegate = mock(FileIO.class);
+ AtomicInteger providerCalls = new AtomicInteger();
+ when(delegate.batchFileDeleter(FIRST))
+ .thenReturn(
+ Optional.of(
+ deleter(
+ 1000,
+ files -> {
+ providerCalls.incrementAndGet();
+ return result(files);
+ })));
+ BatchFileDeleter forwarded =
+ resolving(delegate).batchFileDeleter(FIRST).orElseThrow(AssertionError::new);
+
+ assertThatThrownBy(
+ () ->
+ forwarded.delete(
+ Arrays.asList(
+ FIRST, new Path("s3://bucket/table/b.parquet"))))
+ .isInstanceOfAny(IllegalArgumentException.class, IOException.class);
+ assertThat(providerCalls).hasValue(0);
+ verify(delegate, never()).delete(any(), anyBoolean());
+ }
+
+ @Test
+ void testResolvingPropagatesProviderFailureWithoutFallback() throws Exception {
+ FileIO delegate = mock(FileIO.class);
+ IOException failure = new IOException("resolved batch failed");
+ when(delegate.batchFileDeleter(FIRST))
+ .thenReturn(Optional.of(deleter(1000, files -> raise(failure))));
+ BatchFileDeleter forwarded =
+ resolving(delegate).batchFileDeleter(FIRST).orElseThrow(AssertionError::new);
+
+ assertThatThrownBy(() -> forwarded.delete(FILES)).isSameAs(failure);
+ verify(delegate, never()).delete(any(), anyBoolean());
+ }
+
+ @Test
+ void testResolvingPropagatesDiscoveryFailureWithoutFallback() throws Exception {
+ FileIO delegate = mock(FileIO.class);
+ IOException failure = new IOException("resolved discovery failed");
+ when(delegate.batchFileDeleter(FIRST)).thenThrow(failure);
+
+ assertThatThrownBy(() -> resolving(delegate).batchFileDeleter(FIRST)).isSameAs(failure);
+ verify(delegate, never()).delete(any(), anyBoolean());
+ }
+
+ @Test
+ void testCachingForwardsSupportedAndUnsupportedCapabilities() throws Exception {
+ FileIO supportedDelegate = mock(FileIO.class);
+ BatchDeleteResult expected = result(FILES);
+ when(supportedDelegate.batchFileDeleter(FIRST))
+ .thenReturn(Optional.of(deleter(1000, files -> expected)));
+ CachingFileIO supported = caching(supportedDelegate);
+
+ BatchFileDeleter forwarded =
+ supported.batchFileDeleter(FIRST).orElseThrow(AssertionError::new);
+ assertThat(forwarded.maxBatchSize()).isEqualTo(1000);
+ assertThat(forwarded.delete(FILES)).isSameAs(expected);
+ verify(supportedDelegate, times(1)).batchFileDeleter(FIRST);
+
+ FileIO unsupportedDelegate = mock(FileIO.class);
+ when(unsupportedDelegate.batchFileDeleter(FIRST)).thenReturn(Optional.empty());
+ assertThat(caching(unsupportedDelegate).batchFileDeleter(FIRST)).isEmpty();
+ }
+
+ @Test
+ void testCachingPropagatesProviderFailureWithoutFallback() throws Exception {
+ FileIO delegate = mock(FileIO.class);
+ IOException failure = new IOException("cached batch failed");
+ when(delegate.batchFileDeleter(FIRST))
+ .thenReturn(Optional.of(deleter(1000, files -> raise(failure))));
+ BatchFileDeleter forwarded =
+ caching(delegate).batchFileDeleter(FIRST).orElseThrow(AssertionError::new);
+
+ assertThatThrownBy(() -> forwarded.delete(FILES)).isSameAs(failure);
+ verify(delegate, never()).delete(any(), anyBoolean());
+ }
+
+ @Test
+ void testCachingPropagatesDiscoveryFailureWithoutFallback() throws Exception {
+ FileIO delegate = mock(FileIO.class);
+ IOException failure = new IOException("cached discovery failed");
+ when(delegate.batchFileDeleter(FIRST)).thenThrow(failure);
+
+ assertThatThrownBy(() -> caching(delegate).batchFileDeleter(FIRST)).isSameAs(failure);
+ verify(delegate, times(1)).batchFileDeleter(FIRST);
+ verify(delegate, never()).delete(any(), anyBoolean());
+ }
+
+ @Test
+ void testRestTokenForwardsSupportedAndUnsupportedCapabilities() throws Exception {
+ FileIO supportedDelegate = mock(FileIO.class);
+ BatchDeleteResult expected = result(FILES);
+ when(supportedDelegate.batchFileDeleter(FIRST))
+ .thenReturn(Optional.of(deleter(1000, files -> expected)));
+ RESTTokenFileIO supported = restFileIO(supportedDelegate);
+
+ BatchFileDeleter forwarded =
+ supported.batchFileDeleter(FIRST).orElseThrow(AssertionError::new);
+ assertThat(forwarded.maxBatchSize()).isEqualTo(1000);
+ assertThat(forwarded.delete(FILES)).isSameAs(expected);
+
+ FileIO unsupportedDelegate = mock(FileIO.class);
+ when(unsupportedDelegate.batchFileDeleter(FIRST)).thenReturn(Optional.empty());
+ assertThat(restFileIO(unsupportedDelegate).batchFileDeleter(FIRST)).isEmpty();
+ }
+
+ @Test
+ void testRestTokenRefreshDoesNotInvokeStaleDeleter() throws Exception {
+ FileIO staleDelegate = mock(FileIO.class);
+ FileIO currentDelegate = mock(FileIO.class);
+ AtomicInteger staleCalls = new AtomicInteger();
+ AtomicInteger currentCalls = new AtomicInteger();
+ when(staleDelegate.batchFileDeleter(FIRST))
+ .thenReturn(
+ Optional.of(
+ deleter(
+ 1000,
+ files -> {
+ staleCalls.incrementAndGet();
+ return result(files);
+ })));
+ BatchDeleteResult expected = result(FILES);
+ when(currentDelegate.batchFileDeleter(FIRST))
+ .thenReturn(
+ Optional.of(
+ deleter(
+ 1000,
+ files -> {
+ currentCalls.incrementAndGet();
+ return expected;
+ })));
+ FileIOLoader loader =
+ loader(staleDelegate, staleDelegate, currentDelegate, currentDelegate);
+ RESTApi api = mock(RESTApi.class);
+ Identifier identifier = Identifier.create("db", "table");
+ when(api.loadTableToken(identifier)).thenReturn(token(0L), token(Long.MAX_VALUE));
+ RESTTokenFileIO rest =
+ new RESTTokenFileIO(
+ CatalogContext.create(new Options(), loader, null), api, identifier, FIRST);
+
+ BatchFileDeleter forwarded = rest.batchFileDeleter(FIRST).orElseThrow(AssertionError::new);
+ assertThat(forwarded.delete(FILES)).isSameAs(expected);
+
+ assertThat(staleCalls).hasValue(0);
+ assertThat(currentCalls).hasValue(1);
+ verify(api, times(2)).loadTableToken(identifier);
+ verify(staleDelegate, never()).delete(any(), anyBoolean());
+ verify(currentDelegate, never()).delete(any(), anyBoolean());
+ }
+
+ @Test
+ void testRestTokenCurrentUnsupportedIsHardFailureWithoutUsingStaleCapability()
+ throws Exception {
+ FileIO staleDelegate = mock(FileIO.class);
+ FileIO currentDelegate = mock(FileIO.class);
+ AtomicInteger staleCalls = new AtomicInteger();
+ when(staleDelegate.batchFileDeleter(FIRST))
+ .thenReturn(
+ Optional.of(
+ deleter(
+ 1000,
+ files -> {
+ staleCalls.incrementAndGet();
+ return result(files);
+ })));
+ when(currentDelegate.batchFileDeleter(FIRST)).thenReturn(Optional.empty());
+ RestRefreshFixture fixture = refreshingRest(staleDelegate, currentDelegate);
+
+ BatchFileDeleter forwarded =
+ fixture.fileIO.batchFileDeleter(FIRST).orElseThrow(AssertionError::new);
+ assertThat(forwarded.maxBatchSize()).isEqualTo(1000);
+ assertThatThrownBy(() -> forwarded.delete(FILES)).isInstanceOf(IOException.class);
+
+ assertThat(staleCalls).hasValue(0);
+ verify(fixture.api, times(2)).loadTableToken(fixture.identifier);
+ verify(staleDelegate, never()).delete(any(), anyBoolean());
+ verify(currentDelegate, never()).delete(any(), anyBoolean());
+ }
+
+ @Test
+ void testRestTokenCurrentDiscoveryFailureIsHardFailureWithoutUsingStaleCapability()
+ throws Exception {
+ FileIO staleDelegate = mock(FileIO.class);
+ FileIO currentDelegate = mock(FileIO.class);
+ AtomicInteger staleCalls = new AtomicInteger();
+ when(staleDelegate.batchFileDeleter(FIRST))
+ .thenReturn(
+ Optional.of(
+ deleter(
+ 1000,
+ files -> {
+ staleCalls.incrementAndGet();
+ return result(files);
+ })));
+ IOException failure = new IOException("refreshed capability discovery failed");
+ when(currentDelegate.batchFileDeleter(FIRST)).thenThrow(failure);
+ RestRefreshFixture fixture = refreshingRest(staleDelegate, currentDelegate);
+
+ BatchFileDeleter forwarded =
+ fixture.fileIO.batchFileDeleter(FIRST).orElseThrow(AssertionError::new);
+ assertThatThrownBy(() -> forwarded.delete(FILES)).isSameAs(failure);
+
+ assertThat(staleCalls).hasValue(0);
+ verify(fixture.api, times(2)).loadTableToken(fixture.identifier);
+ verify(staleDelegate, never()).delete(any(), anyBoolean());
+ verify(currentDelegate, never()).delete(any(), anyBoolean());
+ }
+
+ @Test
+ void testRestTokenMaxBatchSizeIsDiscoverySnapshotButDeleteUsesCurrentCapability()
+ throws Exception {
+ FileIO staleDelegate = mock(FileIO.class);
+ FileIO currentDelegate = mock(FileIO.class);
+ AtomicInteger staleCalls = new AtomicInteger();
+ AtomicInteger currentCalls = new AtomicInteger();
+ when(staleDelegate.batchFileDeleter(FIRST))
+ .thenReturn(
+ Optional.of(
+ deleter(
+ 1000,
+ files -> {
+ staleCalls.incrementAndGet();
+ return result(files);
+ })));
+ BatchDeleteResult expected = result(FILES);
+ when(currentDelegate.batchFileDeleter(FIRST))
+ .thenReturn(
+ Optional.of(
+ deleter(
+ 7,
+ files -> {
+ currentCalls.incrementAndGet();
+ return expected;
+ })));
+ RestRefreshFixture fixture = refreshingRest(staleDelegate, currentDelegate);
+
+ BatchFileDeleter forwarded =
+ fixture.fileIO.batchFileDeleter(FIRST).orElseThrow(AssertionError::new);
+
+ // The scheduler-facing limit is a discovery snapshot. It does not authorize use of the
+ // captured deleter: invocation still refreshes and lets the current provider validate.
+ assertThat(forwarded.maxBatchSize()).isEqualTo(1000);
+ assertThat(forwarded.delete(FILES)).isSameAs(expected);
+ assertThat(staleCalls).hasValue(0);
+ assertThat(currentCalls).hasValue(1);
+ }
+
+ @Test
+ void testRestTokenPropagatesProviderFailureWithoutFallback() throws Exception {
+ FileIO delegate = mock(FileIO.class);
+ IOException failure = new IOException("REST batch failed");
+ when(delegate.batchFileDeleter(FIRST))
+ .thenReturn(Optional.of(deleter(1000, files -> raise(failure))));
+ BatchFileDeleter forwarded =
+ restFileIO(delegate).batchFileDeleter(FIRST).orElseThrow(AssertionError::new);
+
+ assertThatThrownBy(() -> forwarded.delete(FILES)).isSameAs(failure);
+ verify(delegate, never()).delete(any(), anyBoolean());
+ }
+
+ @Test
+ void testRestTokenPropagatesDiscoveryFailureWithoutFallback() throws Exception {
+ FileIO delegate = mock(FileIO.class);
+ IOException failure = new IOException("REST discovery failed");
+ when(delegate.batchFileDeleter(FIRST)).thenThrow(failure);
+
+ assertThatThrownBy(() -> restFileIO(delegate).batchFileDeleter(FIRST)).isSameAs(failure);
+ verify(delegate, never()).delete(any(), anyBoolean());
+ }
+
+ @Test
+ void testPluginSerializationForcesCapabilityRediscovery() throws Exception {
+ FileIO staleDelegate = mock(FileIO.class);
+ FileIO currentDelegate = mock(FileIO.class);
+ AtomicInteger staleCalls = new AtomicInteger();
+ AtomicInteger currentCalls = new AtomicInteger();
+ when(staleDelegate.batchFileDeleter(FIRST))
+ .thenReturn(
+ Optional.of(
+ deleter(
+ 1000,
+ files -> {
+ staleCalls.incrementAndGet();
+ return result(files);
+ })));
+ BatchDeleteResult expected = result(FILES);
+ when(currentDelegate.batchFileDeleter(FIRST))
+ .thenReturn(
+ Optional.of(
+ deleter(
+ 1000,
+ files -> {
+ currentCalls.incrementAndGet();
+ return expected;
+ })));
+ SerializablePluginFileIO.reset(staleDelegate);
+ try {
+ SerializablePluginFileIO original = new SerializablePluginFileIO();
+ assertThat(original.batchFileDeleter(FIRST)).isPresent();
+
+ SerializablePluginFileIO restored = InstantiationUtil.clone(original);
+ SerializablePluginFileIO.activeDelegate.set(currentDelegate);
+ BatchFileDeleter rediscovered =
+ restored.batchFileDeleter(FIRST).orElseThrow(AssertionError::new);
+
+ assertThat(rediscovered.delete(FILES)).isSameAs(expected);
+ assertThat(SerializablePluginFileIO.discoveryCalls.get()).hasValue(2);
+ assertThat(staleCalls).hasValue(0);
+ assertThat(currentCalls).hasValue(1);
+ } finally {
+ SerializablePluginFileIO.clear();
+ }
+ }
+
+ @Test
+ void testRestCachingResolvingPluginChainPreservesRefreshAndStrictFailure() throws Exception {
+ FileIO staleProvider = mock(FileIO.class);
+ FileIO currentProvider = mock(FileIO.class);
+ when(staleProvider.exists(any())).thenReturn(true);
+ when(currentProvider.exists(any())).thenReturn(true);
+ AtomicInteger staleCalls = new AtomicInteger();
+ AtomicInteger currentCalls = new AtomicInteger();
+ ClassLoader stalePluginClassLoader = new ClassLoader() {};
+ ClassLoader currentPluginClassLoader = new ClassLoader() {};
+ when(staleProvider.batchFileDeleter(FIRST))
+ .thenAnswer(
+ ignored -> {
+ assertThat(Thread.currentThread().getContextClassLoader())
+ .isSameAs(stalePluginClassLoader);
+ return Optional.of(
+ deleter(
+ 1000,
+ files -> {
+ staleCalls.incrementAndGet();
+ return result(files);
+ }));
+ });
+ BatchDeleteResult expected = result(FILES);
+ when(currentProvider.batchFileDeleter(FIRST))
+ .thenAnswer(
+ ignored -> {
+ assertThat(Thread.currentThread().getContextClassLoader())
+ .isSameAs(currentPluginClassLoader);
+ return Optional.of(
+ deleter(
+ 1000,
+ files -> {
+ assertThat(
+ Thread.currentThread()
+ .getContextClassLoader())
+ .isSameAs(currentPluginClassLoader);
+ currentCalls.incrementAndGet();
+ return expected;
+ }));
+ });
+ FileIO staleChain =
+ frozenResolving(new TestPluginFileIO(staleProvider, stalePluginClassLoader));
+ FileIO currentChain =
+ frozenResolving(new TestPluginFileIO(currentProvider, currentPluginClassLoader));
+ FileIOLoader outerLoader = loader(staleChain, staleChain, currentChain, currentChain);
+ RESTApi api = mock(RESTApi.class);
+ Identifier identifier = Identifier.create("db", "table");
+ when(api.loadTableToken(identifier)).thenReturn(token(0L), token(Long.MAX_VALUE));
+ RESTTokenFileIO rest =
+ new RESTTokenFileIO(
+ CatalogContext.create(new Options(), outerLoader, null),
+ api,
+ identifier,
+ FIRST);
+ CachingFileIO chainRoot = caching(rest);
+ ClassLoader previous = Thread.currentThread().getContextClassLoader();
+ ClassLoader callerClassLoader = new ClassLoader(previous) {};
+
+ Thread.currentThread().setContextClassLoader(callerClassLoader);
+ try {
+ BatchFileDeleter chain =
+ chainRoot.batchFileDeleter(FIRST).orElseThrow(AssertionError::new);
+ assertThatThrownBy(
+ () ->
+ chain.delete(
+ Arrays.asList(
+ FIRST,
+ new Path(
+ "oss://other-bucket/table/b.parquet"))))
+ .isInstanceOfAny(IllegalArgumentException.class, IOException.class);
+ assertThat(staleCalls).hasValue(0);
+ assertThat(currentCalls).hasValue(0);
+
+ assertThat(chain.delete(FILES)).isSameAs(expected);
+ assertThat(staleCalls).hasValue(0);
+ assertThat(currentCalls).hasValue(1);
+ assertThat(Thread.currentThread().getContextClassLoader()).isSameAs(callerClassLoader);
+ verify(staleProvider, never()).delete(any(), anyBoolean());
+ verify(currentProvider, never()).delete(any(), anyBoolean());
+ } finally {
+ Thread.currentThread().setContextClassLoader(previous);
+ }
+ }
+
+ private static ResolvingFileIO resolving(FileIO delegate) throws IOException {
+ FileIOLoader loader = loader(delegate, delegate);
+ ResolvingFileIO resolving = new ResolvingFileIO();
+ resolving.configure(CatalogContext.create(new Options(), loader, null));
+ return resolving;
+ }
+
+ private static ResolvingFileIO frozenResolving(FileIO delegate) throws IOException {
+ FileIOLoader loader = loader(delegate, delegate);
+ FrozenResolvingFileIO resolving = new FrozenResolvingFileIO();
+ resolving.configure(CatalogContext.create(new Options(), loader, null));
+ return resolving;
+ }
+
+ private static CachingFileIO caching(FileIO delegate) {
+ return new CachingFileIO(
+ delegate,
+ mock(org.apache.paimon.fs.cache.LocalCacheManager.class),
+ EnumSet.of(FileType.DATA));
+ }
+
+ private static RESTTokenFileIO restFileIO(FileIO delegate) {
+ FileIOLoader loader = loader(delegate, delegate);
+ RESTApi api = mock(RESTApi.class);
+ Identifier identifier = Identifier.create("db", "table");
+ when(api.loadTableToken(identifier)).thenReturn(token(Long.MAX_VALUE));
+ return new RESTTokenFileIO(
+ CatalogContext.create(new Options(), loader, null), api, identifier, FIRST);
+ }
+
+ private static RestRefreshFixture refreshingRest(FileIO staleDelegate, FileIO currentDelegate) {
+ FileIOLoader loader =
+ loader(staleDelegate, staleDelegate, currentDelegate, currentDelegate);
+ RESTApi api = mock(RESTApi.class);
+ Identifier identifier = Identifier.create("db", "table");
+ when(api.loadTableToken(identifier)).thenReturn(token(0L), token(Long.MAX_VALUE));
+ return new RestRefreshFixture(
+ new RESTTokenFileIO(
+ CatalogContext.create(new Options(), loader, null), api, identifier, FIRST),
+ api,
+ identifier);
+ }
+
+ private static FileIOLoader loader(FileIO first, FileIO... remaining) {
+ FileIOLoader loader = mock(FileIOLoader.class);
+ when(loader.getScheme()).thenReturn("oss");
+ when(loader.load(any())).thenReturn(first, remaining);
+ return loader;
+ }
+
+ private static GetTableTokenResponse token(long expiresAtMillis) {
+ return new GetTableTokenResponse(
+ Collections.singletonMap("token", UUID.randomUUID().toString()), expiresAtMillis);
+ }
+
+ private static BatchDeleteResult result(List files) {
+ return new BatchDeleteResult(files);
+ }
+
+ private static BatchDeleteResult raise(IOException failure) throws IOException {
+ throw failure;
+ }
+
+ private static BatchFileDeleter deleter(int maxBatchSize, DeleteAction action) {
+ return new BatchFileDeleter() {
+ @Override
+ public int maxBatchSize() {
+ return maxBatchSize;
+ }
+
+ @Override
+ public BatchDeleteResult delete(List files) throws IOException {
+ return action.delete(files);
+ }
+ };
+ }
+
+ @FunctionalInterface
+ private interface DeleteAction {
+ BatchDeleteResult delete(List files) throws IOException;
+ }
+
+ private static class RestRefreshFixture {
+
+ private final RESTTokenFileIO fileIO;
+ private final RESTApi api;
+ private final Identifier identifier;
+
+ private RestRefreshFixture(RESTTokenFileIO fileIO, RESTApi api, Identifier identifier) {
+ this.fileIO = fileIO;
+ this.api = api;
+ this.identifier = identifier;
+ }
+ }
+
+ private static class FrozenResolvingFileIO extends ResolvingFileIO {
+
+ private boolean initialized;
+
+ @Override
+ public void configure(CatalogContext context) {
+ if (!initialized) {
+ super.configure(context);
+ initialized = true;
+ }
+ }
+ }
+
+ private static class SerializablePluginFileIO extends PluginFileIO {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final ThreadLocal activeDelegate = new ThreadLocal<>();
+ private static final ThreadLocal discoveryCalls = new ThreadLocal<>();
+
+ private static void reset(FileIO delegate) {
+ activeDelegate.set(delegate);
+ discoveryCalls.set(new AtomicInteger());
+ }
+
+ private static void clear() {
+ activeDelegate.remove();
+ discoveryCalls.remove();
+ }
+
+ @Override
+ public boolean isObjectStore() {
+ return true;
+ }
+
+ @Override
+ protected FileIO createFileIO(Path path) {
+ discoveryCalls.get().incrementAndGet();
+ return activeDelegate.get();
+ }
+
+ @Override
+ protected ClassLoader pluginClassLoader() {
+ return SerializablePluginFileIO.class.getClassLoader();
+ }
+ }
+
+ private static class TestPluginFileIO extends PluginFileIO {
+
+ private final FileIO delegate;
+ private final ClassLoader classLoader;
+
+ private TestPluginFileIO(FileIO delegate, ClassLoader classLoader) {
+ this.delegate = delegate;
+ this.classLoader = classLoader;
+ }
+
+ @Override
+ public boolean isObjectStore() {
+ return true;
+ }
+
+ @Override
+ protected FileIO createFileIO(Path path) {
+ return delegate;
+ }
+
+ @Override
+ protected ClassLoader pluginClassLoader() {
+ return classLoader;
+ }
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java
index 73c11d774166..f9d08a5c1bfb 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java
@@ -78,6 +78,14 @@ public BatchTableCommit newCommit() {
CoreOptions options = new CoreOptions(table.options());
boolean formatTablePartitionOnlyValueInPath = options.formatTablePartitionOnlyValueInPath();
String syncHiveUri = options.formatTableCommitSyncPartitionHiveUri();
+ int cleanupThreadNum =
+ table.partitionManager() != null && !table.partitionKeys().isEmpty()
+ ? options.formatTableCommitCleanupThreadNum()
+ : 1;
+ int publishThreadNum =
+ table.partitionManager() != null && !table.partitionKeys().isEmpty()
+ ? options.formatTableCommitPublishThreadNum()
+ : 1;
return new FormatTableCommit(
table.location(),
table.partitionKeys(),
@@ -90,7 +98,9 @@ public BatchTableCommit newCommit() {
syncHiveUri,
table.catalogContext(),
table.partitionManager(),
- options.dynamicPartitionOverwrite());
+ options.dynamicPartitionOverwrite(),
+ cleanupThreadNum,
+ publishThreadNum);
}
@Override
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java
index be2105930a57..b4a7a1c350f6 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java
@@ -18,11 +18,14 @@
package org.apache.paimon.table.format;
+import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.catalog.CatalogFactory;
import org.apache.paimon.catalog.DelegateCatalog;
import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.fs.BatchDeleteResult;
+import org.apache.paimon.fs.BatchFileDeleter;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.FileStatus;
import org.apache.paimon.fs.Path;
@@ -38,6 +41,7 @@
import org.apache.paimon.table.sink.TableCommit;
import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.PartitionPathUtils;
+import org.apache.paimon.utils.ThreadPoolUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -47,14 +51,23 @@
import java.io.FileNotFoundException;
import java.io.IOException;
import java.lang.reflect.Method;
+import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.Deque;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.Set;
+import java.util.TreeMap;
+import java.util.concurrent.CompletionService;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorCompletionService;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
import java.util.stream.Collectors;
import static org.apache.paimon.table.format.FormatBatchWriteBuilder.validateStaticPartition;
@@ -64,6 +77,18 @@ public class FormatTableCommit implements BatchTableCommit {
private static final Logger LOG = LoggerFactory.getLogger(FormatTableCommit.class);
+ private static final int MAX_CLEANUP_THREAD_NUM = 64;
+
+ private static final int MAX_PUBLISH_THREAD_NUM = 64;
+
+ private static final ExecutorService CLEANUP_EXECUTOR =
+ ThreadPoolUtils.createCachedThreadPool(
+ MAX_CLEANUP_THREAD_NUM, "FORMAT-TABLE-COMMIT-CLEANUP-THREAD-POOL");
+
+ private static final ExecutorService PUBLISH_EXECUTOR =
+ ThreadPoolUtils.createCachedThreadPool(
+ MAX_PUBLISH_THREAD_NUM, "FORMAT-TABLE-COMMIT-PUBLISH-THREAD-POOL");
+
private String location;
private final boolean formatTablePartitionOnlyValueInPath;
private final String defaultPartName;
@@ -75,6 +100,10 @@ public class FormatTableCommit implements BatchTableCommit {
private Identifier tableIdentifier;
@Nullable private final FormatTablePartitionManager partitionManager;
private final boolean dynamicPartitionOverwrite;
+ private final int cleanupThreadNum;
+ private final ExecutorService cleanupExecutor;
+ private final int publishThreadNum;
+ private final ExecutorService publishExecutor;
public FormatTableCommit(
String location,
@@ -89,6 +118,190 @@ public FormatTableCommit(
CatalogContext catalogContext,
@Nullable FormatTablePartitionManager partitionManager,
boolean dynamicPartitionOverwrite) {
+ this(
+ location,
+ partitionKeys,
+ fileIO,
+ formatTablePartitionOnlyValueInPath,
+ defaultPartName,
+ overwrite,
+ tableIdentifier,
+ staticPartitions,
+ syncHiveUri,
+ catalogContext,
+ partitionManager,
+ dynamicPartitionOverwrite,
+ 1,
+ 1,
+ CLEANUP_EXECUTOR,
+ PUBLISH_EXECUTOR);
+ }
+
+ FormatTableCommit(
+ String location,
+ List partitionKeys,
+ FileIO fileIO,
+ boolean formatTablePartitionOnlyValueInPath,
+ String defaultPartName,
+ boolean overwrite,
+ Identifier tableIdentifier,
+ @Nullable Map staticPartitions,
+ @Nullable String syncHiveUri,
+ CatalogContext catalogContext,
+ @Nullable FormatTablePartitionManager partitionManager,
+ boolean dynamicPartitionOverwrite,
+ int cleanupThreadNum) {
+ this(
+ location,
+ partitionKeys,
+ fileIO,
+ formatTablePartitionOnlyValueInPath,
+ defaultPartName,
+ overwrite,
+ tableIdentifier,
+ staticPartitions,
+ syncHiveUri,
+ catalogContext,
+ partitionManager,
+ dynamicPartitionOverwrite,
+ cleanupThreadNum,
+ 1,
+ CLEANUP_EXECUTOR,
+ PUBLISH_EXECUTOR);
+ }
+
+ FormatTableCommit(
+ String location,
+ List partitionKeys,
+ FileIO fileIO,
+ boolean formatTablePartitionOnlyValueInPath,
+ String defaultPartName,
+ boolean overwrite,
+ Identifier tableIdentifier,
+ @Nullable Map staticPartitions,
+ @Nullable String syncHiveUri,
+ CatalogContext catalogContext,
+ @Nullable FormatTablePartitionManager partitionManager,
+ boolean dynamicPartitionOverwrite,
+ int cleanupThreadNum,
+ int publishThreadNum) {
+ this(
+ location,
+ partitionKeys,
+ fileIO,
+ formatTablePartitionOnlyValueInPath,
+ defaultPartName,
+ overwrite,
+ tableIdentifier,
+ staticPartitions,
+ syncHiveUri,
+ catalogContext,
+ partitionManager,
+ dynamicPartitionOverwrite,
+ cleanupThreadNum,
+ publishThreadNum,
+ CLEANUP_EXECUTOR,
+ PUBLISH_EXECUTOR);
+ }
+
+ FormatTableCommit(
+ String location,
+ List partitionKeys,
+ FileIO fileIO,
+ boolean formatTablePartitionOnlyValueInPath,
+ String defaultPartName,
+ boolean overwrite,
+ Identifier tableIdentifier,
+ @Nullable Map staticPartitions,
+ @Nullable String syncHiveUri,
+ CatalogContext catalogContext,
+ @Nullable FormatTablePartitionManager partitionManager,
+ boolean dynamicPartitionOverwrite,
+ int cleanupThreadNum,
+ ExecutorService cleanupExecutor) {
+ this(
+ location,
+ partitionKeys,
+ fileIO,
+ formatTablePartitionOnlyValueInPath,
+ defaultPartName,
+ overwrite,
+ tableIdentifier,
+ staticPartitions,
+ syncHiveUri,
+ catalogContext,
+ partitionManager,
+ dynamicPartitionOverwrite,
+ cleanupThreadNum,
+ 1,
+ cleanupExecutor,
+ PUBLISH_EXECUTOR);
+ }
+
+ FormatTableCommit(
+ String location,
+ List partitionKeys,
+ FileIO fileIO,
+ boolean formatTablePartitionOnlyValueInPath,
+ String defaultPartName,
+ boolean overwrite,
+ Identifier tableIdentifier,
+ @Nullable Map staticPartitions,
+ @Nullable String syncHiveUri,
+ CatalogContext catalogContext,
+ @Nullable FormatTablePartitionManager partitionManager,
+ boolean dynamicPartitionOverwrite,
+ int cleanupThreadNum,
+ int publishThreadNum,
+ ExecutorService publishExecutor) {
+ this(
+ location,
+ partitionKeys,
+ fileIO,
+ formatTablePartitionOnlyValueInPath,
+ defaultPartName,
+ overwrite,
+ tableIdentifier,
+ staticPartitions,
+ syncHiveUri,
+ catalogContext,
+ partitionManager,
+ dynamicPartitionOverwrite,
+ cleanupThreadNum,
+ publishThreadNum,
+ CLEANUP_EXECUTOR,
+ publishExecutor);
+ }
+
+ private FormatTableCommit(
+ String location,
+ List partitionKeys,
+ FileIO fileIO,
+ boolean formatTablePartitionOnlyValueInPath,
+ String defaultPartName,
+ boolean overwrite,
+ Identifier tableIdentifier,
+ @Nullable Map staticPartitions,
+ @Nullable String syncHiveUri,
+ CatalogContext catalogContext,
+ @Nullable FormatTablePartitionManager partitionManager,
+ boolean dynamicPartitionOverwrite,
+ int cleanupThreadNum,
+ int publishThreadNum,
+ ExecutorService cleanupExecutor,
+ ExecutorService publishExecutor) {
+ if (cleanupThreadNum < 1 || cleanupThreadNum > MAX_CLEANUP_THREAD_NUM) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Format Table cleanup thread number must be between 1 and %s, but was %s.",
+ MAX_CLEANUP_THREAD_NUM, cleanupThreadNum));
+ }
+ if (publishThreadNum < 1 || publishThreadNum > MAX_PUBLISH_THREAD_NUM) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Format Table publish thread number must be between 1 and %s, but was %s.",
+ MAX_PUBLISH_THREAD_NUM, publishThreadNum));
+ }
this.location = location;
this.fileIO = fileIO;
this.formatTablePartitionOnlyValueInPath = formatTablePartitionOnlyValueInPath;
@@ -100,6 +313,10 @@ public FormatTableCommit(
this.tableIdentifier = tableIdentifier;
this.partitionManager = partitionManager;
this.dynamicPartitionOverwrite = dynamicPartitionOverwrite;
+ this.cleanupThreadNum = cleanupThreadNum;
+ this.cleanupExecutor = cleanupExecutor;
+ this.publishThreadNum = publishThreadNum;
+ this.publishExecutor = publishExecutor;
if (syncHiveUri != null) {
try {
Options options = new Options();
@@ -151,32 +368,37 @@ public void commit(List commitMessages) {
// A static partition may name only the leading keys, in which case the path
// is a prefix and the partition directories of the remaining keys sit below.
clearedPartitionPaths.addAll(
- deletePreviousDataFile(
- partitionPath, partitionKeys.size() - staticPartitions.size()));
+ deletePreviousDataFiles(
+ Collections.singletonList(partitionPath),
+ partitionKeys.size() - staticPartitions.size(),
+ cleanupThreadNum));
}
if (!fileIO.exists(partitionPath)) {
fileIO.mkdirs(partitionPath);
}
} else if (overwrite) {
if (replacesOnlyWrittenPartitions()) {
- Set partitionPaths = new HashSet<>();
+ Set partitionPaths = new LinkedHashSet<>();
for (TwoPhaseCommitMessage message : messages) {
partitionPaths.add(message.getCommitter().targetPath().getParent());
}
- for (Path p : partitionPaths) {
- // The parent of a written file is a complete partition directory - the
- // table directory itself when the table is unpartitioned - so there is no
- // partition level below it to descend, and it is a partition this commit
- // writes anyway.
- deletePreviousDataFile(p, 0);
+ // The parent of a written file is a complete partition directory - the table
+ // directory itself when the table is unpartitioned - so there is no partition
+ // level below it to descend. Collect every selected directory before deleting
+ // so many small partitions can share the same cleanup concurrency window.
+ if (partitionManager != null && cleanupThreadNum > 1) {
+ deletePreviousDynamicDataFiles(
+ new ArrayList<>(partitionPaths), cleanupThreadNum);
+ } else {
+ deletePreviousDataFiles(
+ new ArrayList<>(partitionPaths), 0, cleanupThreadNum);
}
} else {
// Overwriting without naming a partition replaces the table, so what has to go
// is everything the table holds rather than the files this commit happens to
// write: a statement whose query returns nothing still empties the table.
- for (Path dataDirectory : tableDataDirectories()) {
- clearedPartitionPaths.addAll(deletePreviousDataFile(dataDirectory, 0));
- }
+ clearedPartitionPaths.addAll(
+ deletePreviousDataFiles(tableDataDirectories(), 0, cleanupThreadNum));
}
}
@@ -187,9 +409,9 @@ public void commit(List commitMessages) {
boolean reportsStatistics = registersPartitions && partitionManager != null;
Map