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..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 @@ -217,6 +217,20 @@ default FileStatus[] listDirectories(Path path) throws IOException { */ boolean exists(Path path) throws IOException; + /** + * Deletes files in provider batches when supported. + * + *

{@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 boolean deleteFilesInBatch(List files) throws IOException { + return files.isEmpty(); + } + /** * 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..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 @@ -24,6 +24,7 @@ import java.io.IOException; import java.time.Duration; +import java.util.List; /** * A {@link FileIO} for plugin jar. {@link FileIO} is serializable, so plugin FileIO should be @@ -78,6 +79,14 @@ public boolean exists(Path path) throws IOException { return wrap(() -> fileIO(path).exists(path)); } + @Override + public boolean deleteFilesInBatch(List files) throws IOException { + if (files.isEmpty()) { + return true; + } + return wrap(() -> fileIO(files.get(0)).deleteFilesInBatch(files)); + } + @Override public boolean delete(Path path, boolean recursive) throws IOException { return wrap(() -> fileIO(path).delete(path, recursive)); 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..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 @@ -26,12 +26,15 @@ 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.concurrent.ConcurrentHashMap; 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 @@ -94,6 +97,26 @@ public boolean exists(Path path) throws IOException { return wrap(() -> fileIO(path).exists(path)); } + @Override + public boolean deleteFilesInBatch(List files) throws IOException { + checkArgument(files != null, "Batch delete files must not be null."); + if (files.isEmpty()) { + return true; + } + + 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 public boolean delete(Path path, boolean recursive) throws IOException { return wrap(() -> fileIO(path).delete(path, recursive)); 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..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 @@ -38,6 +38,7 @@ 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.Set; @@ -165,6 +166,11 @@ public boolean exists(Path path) throws IOException { return delegate.exists(path); } + @Override + public boolean deleteFilesInBatch(List files) throws IOException { + return delegate.deleteFilesInBatch(files); + } + @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..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 @@ -47,6 +47,7 @@ import java.io.IOException; import java.io.UncheckedIOException; import java.time.Duration; +import java.util.List; import java.util.Map; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; @@ -145,6 +146,11 @@ public boolean exists(Path path) throws IOException { return fileIO().exists(path); } + @Override + public boolean deleteFilesInBatch(List files) throws IOException { + return fileIO().deleteFilesInBatch(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/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 5a3b48a5c54b..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 @@ -37,8 +37,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 +60,13 @@ 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.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.function.Supplier; @@ -73,6 +80,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 +147,38 @@ public boolean isObjectStore() { return true; } + @Override + public boolean deleteFilesInBatch(List files) throws IOException { + checkArgument(files != null, "Batch delete files must not be null."); + if (files.isEmpty()) { + return true; + } + + 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 public void configure(CatalogContext context) { allowCache = context.options().get(FILE_IO_ALLOW_CACHE); @@ -285,6 +326,55 @@ OSSClient ossClient(Path path) throws Exception { return getOssClient((AliyunOSSFileSystem) getFileSystem(path(path))); } + private static List validateBatch(List files) { + List keys = new ArrayList<>(files.size()); + 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); + OSSUtils.ensureObjectKeyValid(key); + checkArgument( + uniqueKeys.add(key), "Batch delete object keys must not contain duplicates."); + keys.add(key); + } + return 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."); + } + + if (!new HashSet<>(requestedKeys).equals(new HashSet<>(deletedObjects))) { + throw new IOException("OSS batch delete returned an invalid acknowledgement."); + } + } + @Override public void close() { if (!allowCache) { 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..a8301a9ece4d --- /dev/null +++ b/paimon-filesystems/paimon-oss-impl/src/test/java/org/apache/paimon/oss/OSSFileIOBatchDeleteTest.java @@ -0,0 +1,173 @@ +/* + * 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.Path; + +import com.aliyun.oss.OSSClient; +import com.aliyun.oss.model.DeleteObjectsRequest; +import com.aliyun.oss.model.DeleteObjectsResult; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +import java.io.IOException; +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 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.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; + +/** Tests for OSS batch deletion. */ +class OSSFileIOBatchDeleteTest { + + private static final Path FIRST = new Path("oss://bucket/table/file-0.parquet"); + + @Test + void testDeletesInProviderSizedBatches() throws Exception { + OSSClient client = mock(OSSClient.class); + TestOSSFileIO fileIO = new TestOSSFileIO(client); + List files = files(1001, "bucket"); + when(client.deleteObjects(any(DeleteObjectsRequest.class))) + .thenAnswer( + invocation -> { + DeleteObjectsRequest request = invocation.getArgument(0); + return new DeleteObjectsResult(new ArrayList<>(request.getKeys())); + }); + + assertThat(fileIO.deleteFilesInBatch(files)).isTrue(); + + ArgumentCaptor requests = + ArgumentCaptor.forClass(DeleteObjectsRequest.class); + 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); + } + + @Test + void testValidatesWholeRequestBeforeAccessingStorage() { + TestOSSFileIO fileIO = new TestOSSFileIO(mock(OSSClient.class)); + + assertThatThrownBy( + () -> + fileIO.deleteFilesInBatch( + Arrays.asList( + FIRST, + new Path( + "oss://other-bucket/table/file-1.parquet")))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("same OSS bucket"); + + assertThat(fileIO.ossClientCalls).hasValue(0); + } + + @Test + void testValidatesEveryKeyBeforeAccessingStorage() { + OSSClient client = mock(OSSClient.class); + TestOSSFileIO fileIO = new TestOSSFileIO(client); + List files = files(1000, "bucket"); + files.add(new Path("oss://bucket/" + String.join("", Collections.nCopies(1024, "a")))); + + assertThatThrownBy(() -> fileIO.deleteFilesInBatch(files)) + .isInstanceOf(IllegalArgumentException.class); + + assertThat(fileIO.ossClientCalls).hasValue(0); + verify(client, never()).deleteObjects(any(DeleteObjectsRequest.class)); + } + + @Test + void testIncompleteResponseFails() throws Exception { + OSSClient client = mock(OSSClient.class); + TestOSSFileIO fileIO = new TestOSSFileIO(client); + when(client.deleteObjects(any(DeleteObjectsRequest.class))) + .thenReturn(new DeleteObjectsResult(Collections.singletonList(key(FIRST)))); + + assertThatThrownBy( + () -> + fileIO.deleteFilesInBatch( + Arrays.asList( + FIRST, + new Path("oss://bucket/table/file-1.parquet")))) + .isInstanceOf(IOException.class) + .hasMessageContaining("incomplete acknowledgement"); + } + + @Test + void testWrongResponseKeysFail() throws Exception { + OSSClient client = mock(OSSClient.class); + TestOSSFileIO fileIO = new TestOSSFileIO(client); + when(client.deleteObjects(any(DeleteObjectsRequest.class))) + .thenReturn( + new DeleteObjectsResult( + Arrays.asList(key(FIRST), "table/different.parquet"))); + + 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) { + 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 String key(Path path) { + return path.toUri().getPath().substring(1); + } + + private static class TestOSSFileIO extends OSSFileIO { + + private final OSSClient client; + private final AtomicInteger ossClientCalls = new AtomicInteger(); + + private TestOSSFileIO(OSSClient client) { + this.client = client; + } + + @Override + OSSClient ossClient(Path path) { + ossClientCalls.incrementAndGet(); + return client; + } + } +}