From 345b110f73109b1cce99b9d65aa1496f77af33c8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dapeng=20Sun=28=E5=AD=99=E5=A4=A7=E9=B9=8F=29?= Date: Wed, 26 Aug 2026 06:07:22 +0800 Subject: [PATCH 1/2] [common][oss] Add strict batch delete capability Let a FileIO expose a batch deleter for the provider serving a path, forwarded by the plugin, resolving, caching and REST token wrappers. A provider that has started a batch request never falls back to individual deletes, so a partial success cannot be hidden by deleting the remaining files one by one. OSS deletes at most 1000 keys of one bucket per request and verifies the per-object result. --- .../apache/paimon/fs/BatchDeleteResult.java | 45 + .../apache/paimon/fs/BatchFileDeleter.java | 49 + .../java/org/apache/paimon/fs/FileIO.java | 12 + .../org/apache/paimon/fs/PluginFileIO.java | 35 + .../org/apache/paimon/fs/ResolvingFileIO.java | 52 ++ .../apache/paimon/fs/cache/CachingFileIO.java | 7 + .../apache/paimon/rest/RESTTokenFileIO.java | 31 + .../fs/FileIOBatchDeleteContractTest.java | 202 +++++ .../fs/FileIOBatchDeleteForwardingTest.java | 843 ++++++++++++++++++ .../java/org/apache/paimon/oss/OSSFileIO.java | 120 +++ .../paimon/oss/OSSFileIOBatchDeleteTest.java | 531 +++++++++++ 11 files changed, 1927 insertions(+) create mode 100644 paimon-common/src/main/java/org/apache/paimon/fs/BatchDeleteResult.java create mode 100644 paimon-common/src/main/java/org/apache/paimon/fs/BatchFileDeleter.java create mode 100644 paimon-common/src/test/java/org/apache/paimon/fs/FileIOBatchDeleteContractTest.java create mode 100644 paimon-common/src/test/java/org/apache/paimon/fs/FileIOBatchDeleteForwardingTest.java create mode 100644 paimon-filesystems/paimon-oss-impl/src/test/java/org/apache/paimon/oss/OSSFileIOBatchDeleteTest.java diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/BatchDeleteResult.java b/paimon-common/src/main/java/org/apache/paimon/fs/BatchDeleteResult.java new file mode 100644 index 000000000000..74a0f5223d59 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/fs/BatchDeleteResult.java @@ -0,0 +1,45 @@ +/* + * 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.util.ArrayList; +import java.util.Collections; +import java.util.List; + +/** + * Immutable result of a successful strict batch delete. + * + * @since 2.1 + */ +@Public +public final class BatchDeleteResult { + + private final List deletedOrNotFound; + + public BatchDeleteResult(List deletedOrNotFound) { + this.deletedOrNotFound = Collections.unmodifiableList(new ArrayList<>(deletedOrNotFound)); + } + + /** Files confirmed deleted or not found, in request order. */ + public List 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-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSFileIO.java b/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSFileIO.java index 5a3b48a5c54b..4b834f99b5dc 100644 --- a/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSFileIO.java +++ b/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSFileIO.java @@ -20,6 +20,8 @@ import org.apache.paimon.catalog.CatalogContext; 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.HadoopOptionsProvider; import org.apache.paimon.fs.Path; @@ -37,8 +39,11 @@ import com.aliyun.oss.internal.OSSHeaders; import com.aliyun.oss.internal.OSSMultipartOperation; import com.aliyun.oss.internal.OSSObjectOperation; +import com.aliyun.oss.internal.OSSUtils; import com.aliyun.oss.model.CopyObjectRequest; import com.aliyun.oss.model.CopyObjectResult; +import com.aliyun.oss.model.DeleteObjectsRequest; +import com.aliyun.oss.model.DeleteObjectsResult; import com.aliyun.oss.model.InitiateMultipartUploadRequest; import com.aliyun.oss.model.InitiateMultipartUploadResult; import com.aliyun.oss.model.ObjectMetadata; @@ -57,9 +62,14 @@ import java.net.URI; import java.nio.charset.StandardCharsets; import java.time.Duration; +import java.util.ArrayList; import java.util.HashMap; +import java.util.HashSet; +import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Optional; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.function.Supplier; @@ -73,6 +83,8 @@ public class OSSFileIO extends HadoopCompliantFileIO implements HadoopOptionsPro private static final Logger LOG = LoggerFactory.getLogger(OSSFileIO.class); + private static final int MAX_BATCH_DELETE_SIZE = 1000; + /** * In order to simplify, we make paimon oss configuration keys same with hadoop oss module. So, * we add all configuration key with prefix `fs.oss` in paimon conf to hadoop conf. @@ -138,6 +150,22 @@ public boolean isObjectStore() { return true; } + @Override + public Optional batchFileDeleter(Path path) { + return Optional.of( + new BatchFileDeleter() { + @Override + public int maxBatchSize() { + return MAX_BATCH_DELETE_SIZE; + } + + @Override + public BatchDeleteResult delete(List files) throws IOException { + return deleteBatch(files); + } + }); + } + @Override public void configure(CatalogContext context) { allowCache = context.options().get(FILE_IO_ALLOW_CACHE); @@ -285,6 +313,85 @@ OSSClient ossClient(Path path) throws Exception { return getOssClient((AliyunOSSFileSystem) getFileSystem(path(path))); } + private BatchDeleteResult deleteBatch(List files) throws IOException { + ValidatedBatch batch = validateBatch(files); + DeleteObjectsRequest request = + new DeleteObjectsRequest(batch.bucket).withKeys(batch.keys).withQuiet(false); + + DeleteObjectsResult response; + try { + response = ossClient(batch.files.get(0)).deleteObjects(request); + } catch (Exception e) { + throw new IOException("Failed to delete OSS object batch.", e); + } + + validateResponse(batch.keys, response); + return new BatchDeleteResult(batch.files); + } + + private static ValidatedBatch validateBatch(List files) { + checkArgument(files != null, "Batch delete files must not be null."); + checkArgument( + !files.isEmpty() && files.size() <= MAX_BATCH_DELETE_SIZE, + "Batch delete requires between 1 and %s files, but got %s.", + MAX_BATCH_DELETE_SIZE, + files.size()); + + List validatedFiles = new ArrayList<>(files.size()); + List keys = new ArrayList<>(files.size()); + Set uniqueFiles = new HashSet<>(); + Set uniqueKeys = new HashSet<>(); + String bucket = null; + for (Path file : files) { + checkArgument(file != null, "Batch delete file must not be null."); + URI uri = file.toUri(); + checkArgument("oss".equals(uri.getScheme()), "Batch delete only supports OSS paths."); + String host = uri.getHost(); + checkArgument( + host != null && host.equals(uri.getAuthority()), + "Batch delete OSS authority must contain only a bucket."); + OSSUtils.ensureBucketNameValid(host); + if (bucket == null) { + bucket = host; + } else { + checkArgument( + bucket.equals(host), "Batch delete files must use the same OSS bucket."); + } + + String path = uri.getPath(); + checkArgument( + path != null && path.length() > 1, + "Batch delete OSS object key must not be empty."); + String key = path.substring(1); + checkArgument(uniqueFiles.add(file), "Batch delete files must not contain duplicates."); + checkArgument( + uniqueKeys.add(key), "Batch delete object keys must not contain duplicates."); + validatedFiles.add(file); + keys.add(key); + } + return new ValidatedBatch(bucket, validatedFiles, keys); + } + + private static void validateResponse(List requestedKeys, DeleteObjectsResult response) + throws IOException { + if (response == null || response.getDeletedObjects() == null) { + throw new IOException("OSS batch delete returned no acknowledgement."); + } + + List deletedObjects = response.getDeletedObjects(); + if (deletedObjects.size() != requestedKeys.size()) { + throw new IOException("OSS batch delete returned an incomplete acknowledgement."); + } + + Set requested = new HashSet<>(requestedKeys); + Set acknowledged = new HashSet<>(); + for (String key : deletedObjects) { + if (key == null || !requested.contains(key) || !acknowledged.add(key)) { + throw new IOException("OSS batch delete returned an invalid acknowledgement."); + } + } + } + @Override public void close() { if (!allowCache) { @@ -565,4 +672,17 @@ public int hashCode() { return Objects.hash(options, scheme, authority); } } + + private static class ValidatedBatch { + + private final String bucket; + private final List files; + private final List keys; + + private ValidatedBatch(String bucket, List files, List keys) { + this.bucket = bucket; + this.files = files; + this.keys = keys; + } + } } diff --git a/paimon-filesystems/paimon-oss-impl/src/test/java/org/apache/paimon/oss/OSSFileIOBatchDeleteTest.java b/paimon-filesystems/paimon-oss-impl/src/test/java/org/apache/paimon/oss/OSSFileIOBatchDeleteTest.java new file mode 100644 index 000000000000..2dac09186701 --- /dev/null +++ b/paimon-filesystems/paimon-oss-impl/src/test/java/org/apache/paimon/oss/OSSFileIOBatchDeleteTest.java @@ -0,0 +1,531 @@ +/* + * 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.oss; + +import org.apache.paimon.fs.BatchDeleteResult; +import org.apache.paimon.fs.BatchFileDeleter; +import org.apache.paimon.fs.Path; + +import com.aliyun.oss.ClientException; +import com.aliyun.oss.OSSClient; +import com.aliyun.oss.OSSException; +import com.aliyun.oss.internal.OSSUtils; +import com.aliyun.oss.model.DeleteObjectsRequest; +import com.aliyun.oss.model.DeleteObjectsResult; +import com.aliyun.oss.model.GenericRequest; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; +import org.mockito.ArgumentCaptor; + +import java.io.IOException; +import java.net.URI; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.IntStream; +import java.util.stream.Stream; + +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.anyString; +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; + +/** Strict batch-delete contract tests for {@link OSSFileIO}. */ +class OSSFileIOBatchDeleteTest { + + private static final Path FIRST = new Path("oss://bucket/table/file-0.parquet"); + + @Test + void testDeletesOneObjectWithVerboseResponseValidation() throws Exception { + OSSClient client = mock(OSSClient.class); + StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); + when(client.deleteObjects(any(DeleteObjectsRequest.class))) + .thenReturn(new DeleteObjectsResult(Collections.singletonList(key(FIRST)))); + + BatchFileDeleter deleter = capability(fileIO); + BatchDeleteResult result = deleter.delete(Collections.singletonList(FIRST)); + + assertThat(deleter.maxBatchSize()).isEqualTo(1000); + assertThat(result.deletedOrNotFound()).containsExactly(FIRST); + ArgumentCaptor request = + ArgumentCaptor.forClass(DeleteObjectsRequest.class); + verify(client).deleteObjects(request.capture()); + assertThat(request.getValue().getBucketName()).isEqualTo("bucket"); + assertThat(request.getValue().getKeys()).containsExactly("table/file-0.parquet"); + assertThat(request.getValue().isQuiet()).isFalse(); + assertThat(fileIO.ossClientCalls).hasValue(1); + assertNoSingleDeleteFallback(fileIO); + } + + @Test + void testDeletesExactlyOneThousandObjectsInOneRequest() throws Exception { + OSSClient client = mock(OSSClient.class); + StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); + List files = files(1000, "bucket"); + List keys = keys(files); + when(client.deleteObjects(any(DeleteObjectsRequest.class))) + .thenReturn(new DeleteObjectsResult(keys)); + + BatchDeleteResult result = capability(fileIO).delete(files); + + assertThat(result.deletedOrNotFound()).containsExactlyElementsOf(files); + ArgumentCaptor request = + ArgumentCaptor.forClass(DeleteObjectsRequest.class); + verify(client, times(1)).deleteObjects(request.capture()); + assertThat(request.getValue().getKeys()).hasSize(1000).containsExactlyElementsOf(keys); + assertThat(request.getValue().isQuiet()).isFalse(); + assertThat(fileIO.ossClientCalls).hasValue(1); + assertNoSingleDeleteFallback(fileIO); + } + + @Test + void testReverseOrderAcknowledgementReturnsInputOrder() throws Exception { + OSSClient client = mock(OSSClient.class); + StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); + List files = Arrays.asList(FIRST, new Path("oss://bucket/table/file-1.parquet")); + when(client.deleteObjects(any(DeleteObjectsRequest.class))) + .thenReturn( + new DeleteObjectsResult( + Arrays.asList("table/file-1.parquet", "table/file-0.parquet"))); + + BatchDeleteResult result = capability(fileIO).delete(files); + + assertThat(result.deletedOrNotFound()).containsExactlyElementsOf(files); + ArgumentCaptor request = + ArgumentCaptor.forClass(DeleteObjectsRequest.class); + verify(client).deleteObjects(request.capture()); + assertThat(request.getValue().getKeys()) + .containsExactly("table/file-0.parquet", "table/file-1.parquet"); + assertNoSingleDeleteFallback(fileIO); + } + + @Test + void testRejectsEmptyBatchBeforeObtainingClient() throws Exception { + OSSClient client = mock(OSSClient.class); + StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); + + assertThatThrownBy(() -> capability(fileIO).delete(Collections.emptyList())) + .isInstanceOfAny(IllegalArgumentException.class, IOException.class); + + assertNoRemoteRequest(fileIO, client); + } + + @Test + void testRejectsOneThousandAndOneObjectsBeforeObtainingClient() throws Exception { + OSSClient client = mock(OSSClient.class); + StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); + + assertThatThrownBy(() -> capability(fileIO).delete(files(1001, "bucket"))) + .isInstanceOfAny(IllegalArgumentException.class, IOException.class); + + assertNoRemoteRequest(fileIO, client); + } + + @Test + void testRejectsMixedBucketsBeforeObtainingClient() throws Exception { + OSSClient client = mock(OSSClient.class); + StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); + List files = + Arrays.asList(FIRST, new Path("oss://other-bucket/table/file-1.parquet")); + + assertThatThrownBy(() -> capability(fileIO).delete(files)) + .isInstanceOfAny(IllegalArgumentException.class, IOException.class); + + assertNoRemoteRequest(fileIO, client); + } + + @ParameterizedTest(name = "rejects invalid OSS bucket: {0}") + @MethodSource("invalidBuckets") + void testRejectsInvalidBucketBeforeObtainingClient(String description, String bucket) + throws Exception { + OSSClient client = mock(OSSClient.class); + StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); + Path file = new Path("oss://" + bucket + "/table/file.parquet"); + assertThat(OSSUtils.validateBucketName(bucket)).as(description).isFalse(); + + assertThatThrownBy(() -> capability(fileIO).delete(Collections.singletonList(file))) + .as(description) + .isInstanceOfAny(IllegalArgumentException.class, IOException.class); + + assertNoRemoteRequest(fileIO, client); + } + + @ParameterizedTest(name = "rejects non-bucket OSS authority: {0}") + @MethodSource("invalidAuthorities") + void testRejectsNonBucketAuthorityBeforeObtainingClient(String description, String location) + throws Exception { + OSSClient client = mock(OSSClient.class); + StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); + Path file = new Path(location); + assertThat(file.toUri().getHost()).isEqualTo("bucket"); + assertThat(file.toUri().getAuthority()).as(description).isNotEqualTo("bucket"); + + assertThatThrownBy(() -> capability(fileIO).delete(Collections.singletonList(file))) + .as(description) + .isInstanceOfAny(IllegalArgumentException.class, IOException.class); + + assertNoRemoteRequest(fileIO, client); + } + + @Test + void testRejectsDuplicatePathsBeforeObtainingClient() throws Exception { + OSSClient client = mock(OSSClient.class); + StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); + + assertThatThrownBy(() -> capability(fileIO).delete(Arrays.asList(FIRST, FIRST))) + .isInstanceOfAny(IllegalArgumentException.class, IOException.class); + + assertNoRemoteRequest(fileIO, client); + } + + @Test + void testRejectsWrongSchemeBeforeObtainingClient() throws Exception { + OSSClient client = mock(OSSClient.class); + StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); + + assertThatThrownBy( + () -> + capability(fileIO) + .delete( + Collections.singletonList( + new Path( + "s3://bucket/table/file-0.parquet")))) + .isInstanceOfAny(IllegalArgumentException.class, IOException.class); + + assertNoRemoteRequest(fileIO, client); + } + + @Test + void testRejectsEmptyObjectKeyBeforeObtainingClient() throws Exception { + OSSClient client = mock(OSSClient.class); + StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); + + assertThatThrownBy( + () -> + capability(fileIO) + .delete( + Collections.singletonList( + new Path("oss://bucket/")))) + .isInstanceOfAny(IllegalArgumentException.class, IOException.class); + + assertNoRemoteRequest(fileIO, client); + } + + @Test + void testRejectsDistinctPathsWithSameObjectKeyBeforeObtainingClient() throws Exception { + OSSClient client = mock(OSSClient.class); + StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); + Path first = new Path(URI.create("oss://bucket/table/same.parquet#first")); + Path second = new Path(URI.create("oss://bucket/table/same.parquet#second")); + assertThat(first).isNotEqualTo(second); + assertThat(key(first)).isEqualTo(key(second)); + + assertThatThrownBy(() -> capability(fileIO).delete(Arrays.asList(first, second))) + .isInstanceOfAny(IllegalArgumentException.class, IOException.class); + + assertNoRemoteRequest(fileIO, client); + } + + @Test + void testRejectsNullBatchBeforeObtainingClient() throws Exception { + OSSClient client = mock(OSSClient.class); + StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); + + assertThatThrownBy(() -> capability(fileIO).delete(null)) + .isInstanceOfAny( + NullPointerException.class, + IllegalArgumentException.class, + IOException.class); + + assertNoRemoteRequest(fileIO, client); + } + + @Test + void testRejectsNullPathBeforeObtainingClient() throws Exception { + OSSClient client = mock(OSSClient.class); + StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); + + assertThatThrownBy(() -> capability(fileIO).delete(Arrays.asList(FIRST, null))) + .isInstanceOfAny( + NullPointerException.class, + IllegalArgumentException.class, + IOException.class); + + assertNoRemoteRequest(fileIO, client); + } + + @Test + void testSdkExceptionIsHardFailureWithoutSingleDeleteFallback() throws Exception { + OSSClient client = mock(OSSClient.class); + StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); + OSSException failure = new OSSException("service failed"); + when(client.deleteObjects(any(DeleteObjectsRequest.class))).thenThrow(failure); + + assertThatThrownBy(() -> capability(fileIO).delete(Collections.singletonList(FIRST))) + .isInstanceOf(IOException.class) + .hasCause(failure); + + verify(client).deleteObjects(any(DeleteObjectsRequest.class)); + assertNoSingleDeleteFallback(fileIO); + } + + @Test + void testClientExceptionIsHardFailureWithoutSingleDeleteFallback() throws Exception { + OSSClient client = mock(OSSClient.class); + StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); + ClientException failure = new ClientException("client failed"); + when(client.deleteObjects(any(DeleteObjectsRequest.class))).thenThrow(failure); + + assertThatThrownBy(() -> capability(fileIO).delete(Collections.singletonList(FIRST))) + .isInstanceOf(IOException.class) + .hasCause(failure); + + verify(client).deleteObjects(any(DeleteObjectsRequest.class)); + assertNoSingleDeleteFallback(fileIO); + } + + @Test + void testClientAcquisitionFailureIsHardFailureWithoutRemoteRequestOrFallback() + throws Exception { + OSSClient client = mock(OSSClient.class); + ClientException failure = new ClientException("client acquisition failed"); + StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client, failure); + + assertThatThrownBy(() -> capability(fileIO).delete(Collections.singletonList(FIRST))) + .isInstanceOf(IOException.class) + .hasCause(failure); + + assertThat(fileIO.ossClientCalls).hasValue(1); + verify(client, never()).deleteObjects(any(DeleteObjectsRequest.class)); + assertNoSingleDeleteFallback(fileIO); + } + + @Test + void testRetryAfterIndeterminatePartialSuccessResubmitsCompleteBatch() throws Exception { + OSSClient client = mock(OSSClient.class); + StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); + List files = Arrays.asList(FIRST, new Path("oss://bucket/table/file-1.parquet")); + List expectedKeys = keys(files); + List> submittedKeys = new ArrayList<>(); + List simulatedRemoteDeleted = new ArrayList<>(); + AtomicInteger attempts = new AtomicInteger(); + ClientException failure = new ClientException("response lost after partial success"); + when(client.deleteObjects(any(DeleteObjectsRequest.class))) + .thenAnswer( + invocation -> { + DeleteObjectsRequest request = invocation.getArgument(0); + submittedKeys.add(new ArrayList<>(request.getKeys())); + if (attempts.getAndIncrement() == 0) { + simulatedRemoteDeleted.add(request.getKeys().get(0)); + throw failure; + } + simulatedRemoteDeleted.clear(); + simulatedRemoteDeleted.addAll(request.getKeys()); + return new DeleteObjectsResult(new ArrayList<>(request.getKeys())); + }); + BatchFileDeleter deleter = capability(fileIO); + + assertThatThrownBy(() -> deleter.delete(files)) + .isInstanceOf(IOException.class) + .hasCause(failure); + assertThat(simulatedRemoteDeleted).containsExactly(expectedKeys.get(0)); + assertNoSingleDeleteFallback(fileIO); + + BatchDeleteResult result = deleter.delete(files); + + assertThat(result.deletedOrNotFound()).containsExactlyElementsOf(files); + assertThat(submittedKeys).containsExactly(expectedKeys, expectedKeys); + assertThat(simulatedRemoteDeleted).containsExactlyElementsOf(expectedKeys); + verify(client, times(2)).deleteObjects(any(DeleteObjectsRequest.class)); + assertNoSingleDeleteFallback(fileIO); + } + + @Test + void testNullSdkResponseIsHardFailureWithoutFallback() throws Exception { + OSSClient client = mock(OSSClient.class); + StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); + when(client.deleteObjects(any(DeleteObjectsRequest.class))).thenReturn(null); + + assertThatThrownBy(() -> capability(fileIO).delete(Collections.singletonList(FIRST))) + .isInstanceOf(IOException.class); + + verify(client).deleteObjects(any(DeleteObjectsRequest.class)); + assertNoSingleDeleteFallback(fileIO); + } + + @Test + void testNullDeletedObjectsIsHardFailureWithoutFallback() throws Exception { + OSSClient client = mock(OSSClient.class); + StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); + DeleteObjectsResult response = mock(DeleteObjectsResult.class); + when(response.getDeletedObjects()).thenReturn(null); + when(client.deleteObjects(any(DeleteObjectsRequest.class))).thenReturn(response); + + assertThatThrownBy(() -> capability(fileIO).delete(Collections.singletonList(FIRST))) + .isInstanceOf(IOException.class); + + verify(client).deleteObjects(any(DeleteObjectsRequest.class)); + assertNoSingleDeleteFallback(fileIO); + } + + @ParameterizedTest(name = "rejects malformed response: {0}") + @MethodSource("malformedResponses") + void testRejectsMalformedAcknowledgementWithoutFallback( + String description, List responseKeys) throws Exception { + OSSClient client = mock(OSSClient.class); + StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); + List files = Arrays.asList(FIRST, new Path("oss://bucket/table/file-1.parquet")); + when(client.deleteObjects(any(DeleteObjectsRequest.class))) + .thenReturn(new DeleteObjectsResult(responseKeys)); + + assertThatThrownBy(() -> capability(fileIO).delete(files)) + .as(description) + .isInstanceOf(IOException.class); + + verify(client).deleteObjects(any(DeleteObjectsRequest.class)); + assertNoSingleDeleteFallback(fileIO); + } + + private static Stream malformedResponses() { + return Stream.of( + Arguments.of("missing key", Collections.singletonList("table/file-0.parquet")), + Arguments.of( + "same-length replacement", + Arrays.asList("table/file-0.parquet", "table/unrequested.parquet")), + Arguments.of( + "same-length duplicate", + Arrays.asList("table/file-0.parquet", "table/file-0.parquet")), + Arguments.of( + "extra key", + Arrays.asList( + "table/file-0.parquet", + "table/file-1.parquet", + "table/unrequested.parquet")), + Arguments.of( + "duplicate key", + Arrays.asList( + "table/file-0.parquet", + "table/file-1.parquet", + "table/file-1.parquet")), + Arguments.of("null acknowledgement", Arrays.asList("table/file-0.parquet", null))); + } + + private static Stream invalidBuckets() { + return Stream.of( + Arguments.of("uppercase", "Bucket"), + Arguments.of("shorter than three characters", "ab"), + Arguments.of( + "longer than sixty-three characters", + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), + Arguments.of("leading hyphen", "-bucket"), + Arguments.of("trailing hyphen", "bucket-")); + } + + private static Stream invalidAuthorities() { + return Stream.of( + Arguments.of("userinfo", "oss://user@bucket/table/file.parquet"), + Arguments.of("port", "oss://bucket:123/table/file.parquet")); + } + + private static BatchFileDeleter capability(OSSFileIO fileIO) throws IOException { + return fileIO.batchFileDeleter(FIRST).orElseThrow(AssertionError::new); + } + + private static List files(int count, String bucket) { + List files = new ArrayList<>(count); + IntStream.range(0, count) + .mapToObj(i -> new Path("oss://" + bucket + "/table/file-" + i + ".parquet")) + .forEach(files::add); + return files; + } + + private static List keys(List files) { + List keys = new ArrayList<>(files.size()); + for (Path file : files) { + keys.add(key(file)); + } + return keys; + } + + private static String key(Path path) { + return path.toUri().getPath().substring(1); + } + + private static void assertNoRemoteRequest(StrictTestOSSFileIO fileIO, OSSClient client) { + assertThat(fileIO.ossClientCalls).hasValue(0); + verify(client, never()).deleteObjects(any(DeleteObjectsRequest.class)); + assertNoSingleDeleteFallback(fileIO); + } + + private static void assertNoSingleDeleteFallback(StrictTestOSSFileIO fileIO) { + assertThat(fileIO.singleDeleteCalls).hasValue(0); + assertThat(fileIO.hadoopFileSystemCalls).hasValue(0); + verify(fileIO.client, never()).deleteObject(anyString(), anyString()); + verify(fileIO.client, never()).deleteObject(any(GenericRequest.class)); + } + + private static class StrictTestOSSFileIO extends OSSFileIO { + + private final OSSClient client; + private final RuntimeException clientFailure; + private final AtomicInteger ossClientCalls = new AtomicInteger(); + private final AtomicInteger singleDeleteCalls = new AtomicInteger(); + private final AtomicInteger hadoopFileSystemCalls = new AtomicInteger(); + + private StrictTestOSSFileIO(OSSClient client) { + this(client, null); + } + + private StrictTestOSSFileIO(OSSClient client, RuntimeException clientFailure) { + this.client = client; + this.clientFailure = clientFailure; + } + + @Override + OSSClient ossClient(Path path) { + ossClientCalls.incrementAndGet(); + if (clientFailure != null) { + throw clientFailure; + } + return client; + } + + @Override + public boolean delete(Path path, boolean recursive) { + singleDeleteCalls.incrementAndGet(); + return false; + } + + @Override + protected org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem createFileSystem( + org.apache.hadoop.fs.Path path) { + hadoopFileSystemCalls.incrementAndGet(); + throw new AssertionError("Strict batch delete attempted Hadoop single-file fallback"); + } + } +} From 391b19763948d1ce94a5c8c74b5f655577e5a31d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dapeng=20Sun=28=E5=AD=99=E5=A4=A7=E9=B9=8F=29?= Date: Thu, 27 Aug 2026 01:15:14 +0800 Subject: [PATCH 2/2] [common][oss] Simplify strict batch deletion Replace the capability object and result type with one FileIO method. Let OSS chunk and validate requests internally, including full input validation before I/O and exact acknowledgements. --- .../apache/paimon/fs/BatchDeleteResult.java | 45 - .../apache/paimon/fs/BatchFileDeleter.java | 49 - .../java/org/apache/paimon/fs/FileIO.java | 12 +- .../org/apache/paimon/fs/PluginFileIO.java | 34 +- .../org/apache/paimon/fs/ResolvingFileIO.java | 61 +- .../apache/paimon/fs/cache/CachingFileIO.java | 7 +- .../apache/paimon/rest/RESTTokenFileIO.java | 29 +- .../fs/FileIOBatchDeleteContractTest.java | 202 ----- .../fs/FileIOBatchDeleteForwardingTest.java | 843 ------------------ .../apache/paimon/fs/ResolvingFileIOTest.java | 12 + .../java/org/apache/paimon/oss/OSSFileIO.java | 96 +- .../paimon/oss/OSSFileIOBatchDeleteTest.java | 482 ++-------- 12 files changed, 139 insertions(+), 1733 deletions(-) delete mode 100644 paimon-common/src/main/java/org/apache/paimon/fs/BatchDeleteResult.java delete mode 100644 paimon-common/src/main/java/org/apache/paimon/fs/BatchFileDeleter.java delete mode 100644 paimon-common/src/test/java/org/apache/paimon/fs/FileIOBatchDeleteContractTest.java delete mode 100644 paimon-common/src/test/java/org/apache/paimon/fs/FileIOBatchDeleteForwardingTest.java diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/BatchDeleteResult.java b/paimon-common/src/main/java/org/apache/paimon/fs/BatchDeleteResult.java deleted file mode 100644 index 74a0f5223d59..000000000000 --- a/paimon-common/src/main/java/org/apache/paimon/fs/BatchDeleteResult.java +++ /dev/null @@ -1,45 +0,0 @@ -/* - * 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.util.ArrayList; -import java.util.Collections; -import java.util.List; - -/** - * Immutable result of a successful strict batch delete. - * - * @since 2.1 - */ -@Public -public final class BatchDeleteResult { - - private final List deletedOrNotFound; - - public BatchDeleteResult(List deletedOrNotFound) { - this.deletedOrNotFound = Collections.unmodifiableList(new ArrayList<>(deletedOrNotFound)); - } - - /** Files confirmed deleted or not found, in request order. */ - public List 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 deleted file mode 100644 index bfc0996ea81f..000000000000 --- a/paimon-common/src/main/java/org/apache/paimon/fs/BatchFileDeleter.java +++ /dev/null @@ -1,49 +0,0 @@ -/* - * 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 98d386c702a4..5138686252a3 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 @@ -218,15 +218,17 @@ 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. + * Deletes files in provider batches when supported. * - *

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. + *

{@code false} means that no storage access was made and callers may fall back to + * individual deletes. Once an implementation accesses storage, it must either delete every file + * (missing files count as deleted) and return {@code true}, or throw an exception. * + * @param files files from the same URI scheme and authority * @since 2.1 */ - default Optional batchFileDeleter(Path path) throws IOException { - return Optional.empty(); + default boolean deleteFilesInBatch(List files) throws IOException { + return files.isEmpty(); } /** 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 24d8868d1eec..80e932eee499 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 @@ -25,8 +25,6 @@ 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 @@ -82,25 +80,11 @@ public boolean exists(Path path) throws IOException { } @Override - public Optional batchFileDeleter(Path path) throws IOException { - Optional capability = wrap(() -> fileIO(path).batchFileDeleter(path)); - if (!capability.isPresent()) { - return Optional.empty(); + public boolean deleteFilesInBatch(List files) throws IOException { + if (files.isEmpty()) { + return true; } - - 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)); - } - }); + return wrap(() -> fileIO(files.get(0)).deleteFilesInBatch(files)); } @Override @@ -157,16 +141,6 @@ 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 f2475d985534..635842c10382 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 @@ -31,11 +31,10 @@ 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; +import static org.apache.paimon.utils.Preconditions.checkArgument; /** * An implementation of {@link FileIO} that supports multiple file system schemas. It dynamically @@ -99,27 +98,23 @@ public boolean exists(Path path) throws IOException { } @Override - public Optional batchFileDeleter(Path path) throws IOException { - Optional capability = wrap(() -> fileIO(path).batchFileDeleter(path)); - if (!capability.isPresent()) { - return Optional.empty(); + public boolean deleteFilesInBatch(List files) throws IOException { + checkArgument(files != null, "Batch delete files must not be null."); + if (files.isEmpty()) { + return true; } - 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)); - } - }); + Path first = files.get(0); + checkArgument(first != null, "Batch delete file must not be null."); + URI provider = first.toUri(); + for (Path file : files) { + checkArgument( + file != null + && Objects.equals(provider.getScheme(), file.toUri().getScheme()) + && Objects.equals(provider.getAuthority(), file.toUri().getAuthority()), + "Batch delete files must use the same URI scheme and authority."); + } + return wrap(() -> fileIO(first).deleteFilesInBatch(files)); } @Override @@ -177,30 +172,6 @@ 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 e21e237769ec..74b352de206e 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,7 +20,6 @@ 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; @@ -39,9 +38,9 @@ import java.time.Duration; import java.util.EnumSet; import java.util.HashMap; +import java.util.List; 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; @@ -168,8 +167,8 @@ public boolean exists(Path path) throws IOException { } @Override - public Optional batchFileDeleter(Path path) throws IOException { - return delegate.batchFileDeleter(path); + public boolean deleteFilesInBatch(List files) throws IOException { + return delegate.deleteFilesInBatch(files); } @Override 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 982cad5818f0..d8fb59bb1b16 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,8 +21,6 @@ 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; @@ -51,7 +49,6 @@ 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; @@ -150,30 +147,8 @@ public boolean exists(Path path) throws IOException { } @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); - } - }); + public boolean deleteFilesInBatch(List files) throws IOException { + return fileIO().deleteFilesInBatch(files); } @Override 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 deleted file mode 100644 index 8f3d5cd2815d..000000000000 --- a/paimon-common/src/test/java/org/apache/paimon/fs/FileIOBatchDeleteContractTest.java +++ /dev/null @@ -1,202 +0,0 @@ -/* - * 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 deleted file mode 100644 index abb5f25cf555..000000000000 --- a/paimon-common/src/test/java/org/apache/paimon/fs/FileIOBatchDeleteForwardingTest.java +++ /dev/null @@ -1,843 +0,0 @@ -/* - * 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-common/src/test/java/org/apache/paimon/fs/ResolvingFileIOTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/ResolvingFileIOTest.java index 067c7da649aa..3e38894bb356 100644 --- a/paimon-common/src/test/java/org/apache/paimon/fs/ResolvingFileIOTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/fs/ResolvingFileIOTest.java @@ -29,6 +29,7 @@ import java.io.IOException; import java.time.Duration; +import java.util.Arrays; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; @@ -36,6 +37,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; @@ -184,4 +186,14 @@ public void testTryToWriteAtomicReachesResolvedOverride() throws IOException { // the interface default would have written a temp file and renamed it instead verify(delegate, never()).rename(any(), any()); } + + @Test + public void testBatchDeleteRejectsMixedProviders() { + assertThrows( + IllegalArgumentException.class, + () -> + resolvingFileIO.deleteFilesInBatch( + Arrays.asList( + new Path("file:///table/a"), new Path("hdfs:///table/b")))); + } } diff --git a/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSFileIO.java b/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSFileIO.java index 4b834f99b5dc..083d8c5deb7b 100644 --- a/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSFileIO.java +++ b/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSFileIO.java @@ -20,8 +20,6 @@ import org.apache.paimon.catalog.CatalogContext; 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.HadoopOptionsProvider; import org.apache.paimon.fs.Path; @@ -68,7 +66,6 @@ import java.util.List; import java.util.Map; import java.util.Objects; -import java.util.Optional; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.function.Supplier; @@ -151,19 +148,35 @@ public boolean isObjectStore() { } @Override - public Optional batchFileDeleter(Path path) { - return Optional.of( - new BatchFileDeleter() { - @Override - public int maxBatchSize() { - return MAX_BATCH_DELETE_SIZE; - } + public boolean deleteFilesInBatch(List files) throws IOException { + checkArgument(files != null, "Batch delete files must not be null."); + if (files.isEmpty()) { + return true; + } - @Override - public BatchDeleteResult delete(List files) throws IOException { - return deleteBatch(files); - } - }); + List keys = validateBatch(files); + String bucket = files.get(0).toUri().getHost(); + OSSClient client; + try { + client = ossClient(files.get(0)); + } catch (Exception e) { + throw new IOException("Failed to create OSS client for batch delete.", e); + } + + for (int start = 0; start < keys.size(); start += MAX_BATCH_DELETE_SIZE) { + List batch = + keys.subList(start, Math.min(start + MAX_BATCH_DELETE_SIZE, keys.size())); + DeleteObjectsRequest request = + new DeleteObjectsRequest(bucket).withKeys(batch).withQuiet(false); + DeleteObjectsResult response; + try { + response = client.deleteObjects(request); + } catch (Exception e) { + throw new IOException("Failed to delete OSS object batch.", e); + } + validateResponse(batch, response); + } + return true; } @Override @@ -313,33 +326,8 @@ OSSClient ossClient(Path path) throws Exception { return getOssClient((AliyunOSSFileSystem) getFileSystem(path(path))); } - private BatchDeleteResult deleteBatch(List files) throws IOException { - ValidatedBatch batch = validateBatch(files); - DeleteObjectsRequest request = - new DeleteObjectsRequest(batch.bucket).withKeys(batch.keys).withQuiet(false); - - DeleteObjectsResult response; - try { - response = ossClient(batch.files.get(0)).deleteObjects(request); - } catch (Exception e) { - throw new IOException("Failed to delete OSS object batch.", e); - } - - validateResponse(batch.keys, response); - return new BatchDeleteResult(batch.files); - } - - private static ValidatedBatch validateBatch(List files) { - checkArgument(files != null, "Batch delete files must not be null."); - checkArgument( - !files.isEmpty() && files.size() <= MAX_BATCH_DELETE_SIZE, - "Batch delete requires between 1 and %s files, but got %s.", - MAX_BATCH_DELETE_SIZE, - files.size()); - - List validatedFiles = new ArrayList<>(files.size()); + private static List validateBatch(List files) { List keys = new ArrayList<>(files.size()); - Set uniqueFiles = new HashSet<>(); Set uniqueKeys = new HashSet<>(); String bucket = null; for (Path file : files) { @@ -363,13 +351,12 @@ private static ValidatedBatch validateBatch(List files) { path != null && path.length() > 1, "Batch delete OSS object key must not be empty."); String key = path.substring(1); - checkArgument(uniqueFiles.add(file), "Batch delete files must not contain duplicates."); + OSSUtils.ensureObjectKeyValid(key); checkArgument( uniqueKeys.add(key), "Batch delete object keys must not contain duplicates."); - validatedFiles.add(file); keys.add(key); } - return new ValidatedBatch(bucket, validatedFiles, keys); + return keys; } private static void validateResponse(List requestedKeys, DeleteObjectsResult response) @@ -383,12 +370,8 @@ private static void validateResponse(List requestedKeys, DeleteObjectsRe throw new IOException("OSS batch delete returned an incomplete acknowledgement."); } - Set requested = new HashSet<>(requestedKeys); - Set acknowledged = new HashSet<>(); - for (String key : deletedObjects) { - if (key == null || !requested.contains(key) || !acknowledged.add(key)) { - throw new IOException("OSS batch delete returned an invalid acknowledgement."); - } + if (!new HashSet<>(requestedKeys).equals(new HashSet<>(deletedObjects))) { + throw new IOException("OSS batch delete returned an invalid acknowledgement."); } } @@ -672,17 +655,4 @@ public int hashCode() { return Objects.hash(options, scheme, authority); } } - - private static class ValidatedBatch { - - private final String bucket; - private final List files; - private final List keys; - - private ValidatedBatch(String bucket, List files, List keys) { - this.bucket = bucket; - this.files = files; - this.keys = keys; - } - } } diff --git a/paimon-filesystems/paimon-oss-impl/src/test/java/org/apache/paimon/oss/OSSFileIOBatchDeleteTest.java b/paimon-filesystems/paimon-oss-impl/src/test/java/org/apache/paimon/oss/OSSFileIOBatchDeleteTest.java index 2dac09186701..a8301a9ece4d 100644 --- a/paimon-filesystems/paimon-oss-impl/src/test/java/org/apache/paimon/oss/OSSFileIOBatchDeleteTest.java +++ b/paimon-filesystems/paimon-oss-impl/src/test/java/org/apache/paimon/oss/OSSFileIOBatchDeleteTest.java @@ -18,442 +18,129 @@ package org.apache.paimon.oss; -import org.apache.paimon.fs.BatchDeleteResult; -import org.apache.paimon.fs.BatchFileDeleter; import org.apache.paimon.fs.Path; -import com.aliyun.oss.ClientException; import com.aliyun.oss.OSSClient; -import com.aliyun.oss.OSSException; -import com.aliyun.oss.internal.OSSUtils; import com.aliyun.oss.model.DeleteObjectsRequest; import com.aliyun.oss.model.DeleteObjectsResult; -import com.aliyun.oss.model.GenericRequest; import org.junit.jupiter.api.Test; -import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.Arguments; -import org.junit.jupiter.params.provider.MethodSource; import org.mockito.ArgumentCaptor; import java.io.IOException; -import java.net.URI; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.IntStream; -import java.util.stream.Stream; 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.anyString; 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; -/** Strict batch-delete contract tests for {@link OSSFileIO}. */ +/** Tests for OSS batch deletion. */ class OSSFileIOBatchDeleteTest { private static final Path FIRST = new Path("oss://bucket/table/file-0.parquet"); @Test - void testDeletesOneObjectWithVerboseResponseValidation() throws Exception { + void testDeletesInProviderSizedBatches() throws Exception { OSSClient client = mock(OSSClient.class); - StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); + TestOSSFileIO fileIO = new TestOSSFileIO(client); + List files = files(1001, "bucket"); when(client.deleteObjects(any(DeleteObjectsRequest.class))) - .thenReturn(new DeleteObjectsResult(Collections.singletonList(key(FIRST)))); - - BatchFileDeleter deleter = capability(fileIO); - BatchDeleteResult result = deleter.delete(Collections.singletonList(FIRST)); - - assertThat(deleter.maxBatchSize()).isEqualTo(1000); - assertThat(result.deletedOrNotFound()).containsExactly(FIRST); - ArgumentCaptor request = - ArgumentCaptor.forClass(DeleteObjectsRequest.class); - verify(client).deleteObjects(request.capture()); - assertThat(request.getValue().getBucketName()).isEqualTo("bucket"); - assertThat(request.getValue().getKeys()).containsExactly("table/file-0.parquet"); - assertThat(request.getValue().isQuiet()).isFalse(); - assertThat(fileIO.ossClientCalls).hasValue(1); - assertNoSingleDeleteFallback(fileIO); - } - - @Test - void testDeletesExactlyOneThousandObjectsInOneRequest() throws Exception { - OSSClient client = mock(OSSClient.class); - StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); - List files = files(1000, "bucket"); - List keys = keys(files); - when(client.deleteObjects(any(DeleteObjectsRequest.class))) - .thenReturn(new DeleteObjectsResult(keys)); + .thenAnswer( + invocation -> { + DeleteObjectsRequest request = invocation.getArgument(0); + return new DeleteObjectsResult(new ArrayList<>(request.getKeys())); + }); - BatchDeleteResult result = capability(fileIO).delete(files); + assertThat(fileIO.deleteFilesInBatch(files)).isTrue(); - assertThat(result.deletedOrNotFound()).containsExactlyElementsOf(files); - ArgumentCaptor request = + ArgumentCaptor requests = ArgumentCaptor.forClass(DeleteObjectsRequest.class); - verify(client, times(1)).deleteObjects(request.capture()); - assertThat(request.getValue().getKeys()).hasSize(1000).containsExactlyElementsOf(keys); - assertThat(request.getValue().isQuiet()).isFalse(); + verify(client, times(2)).deleteObjects(requests.capture()); + assertThat(requests.getAllValues().get(0).getKeys()).hasSize(1000); + assertThat(requests.getAllValues().get(1).getKeys()).hasSize(1); + assertThat(requests.getAllValues()) + .allSatisfy( + request -> { + assertThat(request.getBucketName()).isEqualTo("bucket"); + assertThat(request.isQuiet()).isFalse(); + }); assertThat(fileIO.ossClientCalls).hasValue(1); - assertNoSingleDeleteFallback(fileIO); - } - - @Test - void testReverseOrderAcknowledgementReturnsInputOrder() throws Exception { - OSSClient client = mock(OSSClient.class); - StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); - List files = Arrays.asList(FIRST, new Path("oss://bucket/table/file-1.parquet")); - when(client.deleteObjects(any(DeleteObjectsRequest.class))) - .thenReturn( - new DeleteObjectsResult( - Arrays.asList("table/file-1.parquet", "table/file-0.parquet"))); - - BatchDeleteResult result = capability(fileIO).delete(files); - - assertThat(result.deletedOrNotFound()).containsExactlyElementsOf(files); - ArgumentCaptor request = - ArgumentCaptor.forClass(DeleteObjectsRequest.class); - verify(client).deleteObjects(request.capture()); - assertThat(request.getValue().getKeys()) - .containsExactly("table/file-0.parquet", "table/file-1.parquet"); - assertNoSingleDeleteFallback(fileIO); - } - - @Test - void testRejectsEmptyBatchBeforeObtainingClient() throws Exception { - OSSClient client = mock(OSSClient.class); - StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); - - assertThatThrownBy(() -> capability(fileIO).delete(Collections.emptyList())) - .isInstanceOfAny(IllegalArgumentException.class, IOException.class); - - assertNoRemoteRequest(fileIO, client); - } - - @Test - void testRejectsOneThousandAndOneObjectsBeforeObtainingClient() throws Exception { - OSSClient client = mock(OSSClient.class); - StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); - - assertThatThrownBy(() -> capability(fileIO).delete(files(1001, "bucket"))) - .isInstanceOfAny(IllegalArgumentException.class, IOException.class); - - assertNoRemoteRequest(fileIO, client); - } - - @Test - void testRejectsMixedBucketsBeforeObtainingClient() throws Exception { - OSSClient client = mock(OSSClient.class); - StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); - List files = - Arrays.asList(FIRST, new Path("oss://other-bucket/table/file-1.parquet")); - - assertThatThrownBy(() -> capability(fileIO).delete(files)) - .isInstanceOfAny(IllegalArgumentException.class, IOException.class); - - assertNoRemoteRequest(fileIO, client); - } - - @ParameterizedTest(name = "rejects invalid OSS bucket: {0}") - @MethodSource("invalidBuckets") - void testRejectsInvalidBucketBeforeObtainingClient(String description, String bucket) - throws Exception { - OSSClient client = mock(OSSClient.class); - StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); - Path file = new Path("oss://" + bucket + "/table/file.parquet"); - assertThat(OSSUtils.validateBucketName(bucket)).as(description).isFalse(); - - assertThatThrownBy(() -> capability(fileIO).delete(Collections.singletonList(file))) - .as(description) - .isInstanceOfAny(IllegalArgumentException.class, IOException.class); - - assertNoRemoteRequest(fileIO, client); - } - - @ParameterizedTest(name = "rejects non-bucket OSS authority: {0}") - @MethodSource("invalidAuthorities") - void testRejectsNonBucketAuthorityBeforeObtainingClient(String description, String location) - throws Exception { - OSSClient client = mock(OSSClient.class); - StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); - Path file = new Path(location); - assertThat(file.toUri().getHost()).isEqualTo("bucket"); - assertThat(file.toUri().getAuthority()).as(description).isNotEqualTo("bucket"); - - assertThatThrownBy(() -> capability(fileIO).delete(Collections.singletonList(file))) - .as(description) - .isInstanceOfAny(IllegalArgumentException.class, IOException.class); - - assertNoRemoteRequest(fileIO, client); - } - - @Test - void testRejectsDuplicatePathsBeforeObtainingClient() throws Exception { - OSSClient client = mock(OSSClient.class); - StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); - - assertThatThrownBy(() -> capability(fileIO).delete(Arrays.asList(FIRST, FIRST))) - .isInstanceOfAny(IllegalArgumentException.class, IOException.class); - - assertNoRemoteRequest(fileIO, client); - } - - @Test - void testRejectsWrongSchemeBeforeObtainingClient() throws Exception { - OSSClient client = mock(OSSClient.class); - StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); - - assertThatThrownBy( - () -> - capability(fileIO) - .delete( - Collections.singletonList( - new Path( - "s3://bucket/table/file-0.parquet")))) - .isInstanceOfAny(IllegalArgumentException.class, IOException.class); - - assertNoRemoteRequest(fileIO, client); } @Test - void testRejectsEmptyObjectKeyBeforeObtainingClient() throws Exception { - OSSClient client = mock(OSSClient.class); - StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); + void testValidatesWholeRequestBeforeAccessingStorage() { + TestOSSFileIO fileIO = new TestOSSFileIO(mock(OSSClient.class)); assertThatThrownBy( () -> - capability(fileIO) - .delete( - Collections.singletonList( - new Path("oss://bucket/")))) - .isInstanceOfAny(IllegalArgumentException.class, IOException.class); + fileIO.deleteFilesInBatch( + Arrays.asList( + FIRST, + new Path( + "oss://other-bucket/table/file-1.parquet")))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("same OSS bucket"); - assertNoRemoteRequest(fileIO, client); - } - - @Test - void testRejectsDistinctPathsWithSameObjectKeyBeforeObtainingClient() throws Exception { - OSSClient client = mock(OSSClient.class); - StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); - Path first = new Path(URI.create("oss://bucket/table/same.parquet#first")); - Path second = new Path(URI.create("oss://bucket/table/same.parquet#second")); - assertThat(first).isNotEqualTo(second); - assertThat(key(first)).isEqualTo(key(second)); - - assertThatThrownBy(() -> capability(fileIO).delete(Arrays.asList(first, second))) - .isInstanceOfAny(IllegalArgumentException.class, IOException.class); - - assertNoRemoteRequest(fileIO, client); - } - - @Test - void testRejectsNullBatchBeforeObtainingClient() throws Exception { - OSSClient client = mock(OSSClient.class); - StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); - - assertThatThrownBy(() -> capability(fileIO).delete(null)) - .isInstanceOfAny( - NullPointerException.class, - IllegalArgumentException.class, - IOException.class); - - assertNoRemoteRequest(fileIO, client); - } - - @Test - void testRejectsNullPathBeforeObtainingClient() throws Exception { - OSSClient client = mock(OSSClient.class); - StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); - - assertThatThrownBy(() -> capability(fileIO).delete(Arrays.asList(FIRST, null))) - .isInstanceOfAny( - NullPointerException.class, - IllegalArgumentException.class, - IOException.class); - - assertNoRemoteRequest(fileIO, client); - } - - @Test - void testSdkExceptionIsHardFailureWithoutSingleDeleteFallback() throws Exception { - OSSClient client = mock(OSSClient.class); - StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); - OSSException failure = new OSSException("service failed"); - when(client.deleteObjects(any(DeleteObjectsRequest.class))).thenThrow(failure); - - assertThatThrownBy(() -> capability(fileIO).delete(Collections.singletonList(FIRST))) - .isInstanceOf(IOException.class) - .hasCause(failure); - - verify(client).deleteObjects(any(DeleteObjectsRequest.class)); - assertNoSingleDeleteFallback(fileIO); - } - - @Test - void testClientExceptionIsHardFailureWithoutSingleDeleteFallback() throws Exception { - OSSClient client = mock(OSSClient.class); - StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); - ClientException failure = new ClientException("client failed"); - when(client.deleteObjects(any(DeleteObjectsRequest.class))).thenThrow(failure); - - assertThatThrownBy(() -> capability(fileIO).delete(Collections.singletonList(FIRST))) - .isInstanceOf(IOException.class) - .hasCause(failure); - - verify(client).deleteObjects(any(DeleteObjectsRequest.class)); - assertNoSingleDeleteFallback(fileIO); + assertThat(fileIO.ossClientCalls).hasValue(0); } @Test - void testClientAcquisitionFailureIsHardFailureWithoutRemoteRequestOrFallback() - throws Exception { + void testValidatesEveryKeyBeforeAccessingStorage() { OSSClient client = mock(OSSClient.class); - ClientException failure = new ClientException("client acquisition failed"); - StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client, failure); + TestOSSFileIO fileIO = new TestOSSFileIO(client); + List files = files(1000, "bucket"); + files.add(new Path("oss://bucket/" + String.join("", Collections.nCopies(1024, "a")))); - assertThatThrownBy(() -> capability(fileIO).delete(Collections.singletonList(FIRST))) - .isInstanceOf(IOException.class) - .hasCause(failure); + assertThatThrownBy(() -> fileIO.deleteFilesInBatch(files)) + .isInstanceOf(IllegalArgumentException.class); - assertThat(fileIO.ossClientCalls).hasValue(1); + assertThat(fileIO.ossClientCalls).hasValue(0); verify(client, never()).deleteObjects(any(DeleteObjectsRequest.class)); - assertNoSingleDeleteFallback(fileIO); } @Test - void testRetryAfterIndeterminatePartialSuccessResubmitsCompleteBatch() throws Exception { + void testIncompleteResponseFails() throws Exception { OSSClient client = mock(OSSClient.class); - StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); - List files = Arrays.asList(FIRST, new Path("oss://bucket/table/file-1.parquet")); - List expectedKeys = keys(files); - List> submittedKeys = new ArrayList<>(); - List simulatedRemoteDeleted = new ArrayList<>(); - AtomicInteger attempts = new AtomicInteger(); - ClientException failure = new ClientException("response lost after partial success"); + TestOSSFileIO fileIO = new TestOSSFileIO(client); when(client.deleteObjects(any(DeleteObjectsRequest.class))) - .thenAnswer( - invocation -> { - DeleteObjectsRequest request = invocation.getArgument(0); - submittedKeys.add(new ArrayList<>(request.getKeys())); - if (attempts.getAndIncrement() == 0) { - simulatedRemoteDeleted.add(request.getKeys().get(0)); - throw failure; - } - simulatedRemoteDeleted.clear(); - simulatedRemoteDeleted.addAll(request.getKeys()); - return new DeleteObjectsResult(new ArrayList<>(request.getKeys())); - }); - BatchFileDeleter deleter = capability(fileIO); + .thenReturn(new DeleteObjectsResult(Collections.singletonList(key(FIRST)))); - assertThatThrownBy(() -> deleter.delete(files)) + assertThatThrownBy( + () -> + fileIO.deleteFilesInBatch( + Arrays.asList( + FIRST, + new Path("oss://bucket/table/file-1.parquet")))) .isInstanceOf(IOException.class) - .hasCause(failure); - assertThat(simulatedRemoteDeleted).containsExactly(expectedKeys.get(0)); - assertNoSingleDeleteFallback(fileIO); - - BatchDeleteResult result = deleter.delete(files); - - assertThat(result.deletedOrNotFound()).containsExactlyElementsOf(files); - assertThat(submittedKeys).containsExactly(expectedKeys, expectedKeys); - assertThat(simulatedRemoteDeleted).containsExactlyElementsOf(expectedKeys); - verify(client, times(2)).deleteObjects(any(DeleteObjectsRequest.class)); - assertNoSingleDeleteFallback(fileIO); + .hasMessageContaining("incomplete acknowledgement"); } @Test - void testNullSdkResponseIsHardFailureWithoutFallback() throws Exception { + void testWrongResponseKeysFail() throws Exception { OSSClient client = mock(OSSClient.class); - StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); - when(client.deleteObjects(any(DeleteObjectsRequest.class))).thenReturn(null); - - assertThatThrownBy(() -> capability(fileIO).delete(Collections.singletonList(FIRST))) - .isInstanceOf(IOException.class); - - verify(client).deleteObjects(any(DeleteObjectsRequest.class)); - assertNoSingleDeleteFallback(fileIO); - } - - @Test - void testNullDeletedObjectsIsHardFailureWithoutFallback() throws Exception { - OSSClient client = mock(OSSClient.class); - StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); - DeleteObjectsResult response = mock(DeleteObjectsResult.class); - when(response.getDeletedObjects()).thenReturn(null); - when(client.deleteObjects(any(DeleteObjectsRequest.class))).thenReturn(response); - - assertThatThrownBy(() -> capability(fileIO).delete(Collections.singletonList(FIRST))) - .isInstanceOf(IOException.class); - - verify(client).deleteObjects(any(DeleteObjectsRequest.class)); - assertNoSingleDeleteFallback(fileIO); - } - - @ParameterizedTest(name = "rejects malformed response: {0}") - @MethodSource("malformedResponses") - void testRejectsMalformedAcknowledgementWithoutFallback( - String description, List responseKeys) throws Exception { - OSSClient client = mock(OSSClient.class); - StrictTestOSSFileIO fileIO = new StrictTestOSSFileIO(client); - List files = Arrays.asList(FIRST, new Path("oss://bucket/table/file-1.parquet")); + TestOSSFileIO fileIO = new TestOSSFileIO(client); when(client.deleteObjects(any(DeleteObjectsRequest.class))) - .thenReturn(new DeleteObjectsResult(responseKeys)); - - assertThatThrownBy(() -> capability(fileIO).delete(files)) - .as(description) - .isInstanceOf(IOException.class); - - verify(client).deleteObjects(any(DeleteObjectsRequest.class)); - assertNoSingleDeleteFallback(fileIO); - } - - private static Stream malformedResponses() { - return Stream.of( - Arguments.of("missing key", Collections.singletonList("table/file-0.parquet")), - Arguments.of( - "same-length replacement", - Arrays.asList("table/file-0.parquet", "table/unrequested.parquet")), - Arguments.of( - "same-length duplicate", - Arrays.asList("table/file-0.parquet", "table/file-0.parquet")), - Arguments.of( - "extra key", - Arrays.asList( - "table/file-0.parquet", - "table/file-1.parquet", - "table/unrequested.parquet")), - Arguments.of( - "duplicate key", - Arrays.asList( - "table/file-0.parquet", - "table/file-1.parquet", - "table/file-1.parquet")), - Arguments.of("null acknowledgement", Arrays.asList("table/file-0.parquet", null))); - } - - private static Stream invalidBuckets() { - return Stream.of( - Arguments.of("uppercase", "Bucket"), - Arguments.of("shorter than three characters", "ab"), - Arguments.of( - "longer than sixty-three characters", - "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), - Arguments.of("leading hyphen", "-bucket"), - Arguments.of("trailing hyphen", "bucket-")); - } - - private static Stream invalidAuthorities() { - return Stream.of( - Arguments.of("userinfo", "oss://user@bucket/table/file.parquet"), - Arguments.of("port", "oss://bucket:123/table/file.parquet")); - } + .thenReturn( + new DeleteObjectsResult( + Arrays.asList(key(FIRST), "table/different.parquet"))); - private static BatchFileDeleter capability(OSSFileIO fileIO) throws IOException { - return fileIO.batchFileDeleter(FIRST).orElseThrow(AssertionError::new); + assertThatThrownBy( + () -> + fileIO.deleteFilesInBatch( + Arrays.asList( + FIRST, + new Path("oss://bucket/table/file-1.parquet")))) + .isInstanceOf(IOException.class) + .hasMessageContaining("invalid acknowledgement"); } private static List files(int count, String bucket) { @@ -464,68 +151,23 @@ private static List files(int count, String bucket) { return files; } - private static List keys(List files) { - List keys = new ArrayList<>(files.size()); - for (Path file : files) { - keys.add(key(file)); - } - return keys; - } - private static String key(Path path) { return path.toUri().getPath().substring(1); } - private static void assertNoRemoteRequest(StrictTestOSSFileIO fileIO, OSSClient client) { - assertThat(fileIO.ossClientCalls).hasValue(0); - verify(client, never()).deleteObjects(any(DeleteObjectsRequest.class)); - assertNoSingleDeleteFallback(fileIO); - } - - private static void assertNoSingleDeleteFallback(StrictTestOSSFileIO fileIO) { - assertThat(fileIO.singleDeleteCalls).hasValue(0); - assertThat(fileIO.hadoopFileSystemCalls).hasValue(0); - verify(fileIO.client, never()).deleteObject(anyString(), anyString()); - verify(fileIO.client, never()).deleteObject(any(GenericRequest.class)); - } - - private static class StrictTestOSSFileIO extends OSSFileIO { + private static class TestOSSFileIO extends OSSFileIO { private final OSSClient client; - private final RuntimeException clientFailure; private final AtomicInteger ossClientCalls = new AtomicInteger(); - private final AtomicInteger singleDeleteCalls = new AtomicInteger(); - private final AtomicInteger hadoopFileSystemCalls = new AtomicInteger(); - - private StrictTestOSSFileIO(OSSClient client) { - this(client, null); - } - private StrictTestOSSFileIO(OSSClient client, RuntimeException clientFailure) { + private TestOSSFileIO(OSSClient client) { this.client = client; - this.clientFailure = clientFailure; } @Override OSSClient ossClient(Path path) { ossClientCalls.incrementAndGet(); - if (clientFailure != null) { - throw clientFailure; - } return client; } - - @Override - public boolean delete(Path path, boolean recursive) { - singleDeleteCalls.incrementAndGet(); - return false; - } - - @Override - protected org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem createFileSystem( - org.apache.hadoop.fs.Path path) { - hadoopFileSystemCalls.incrementAndGet(); - throw new AssertionError("Strict batch delete attempted Hadoop single-file fallback"); - } } }