From 16984b85693c471ea6e895dbd692cf91ef1996b8 Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Sat, 5 Sep 2026 10:08:07 +0800 Subject: [PATCH 1/2] [common] Forward two-phase output stream creation to the resolved FileIO MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ResolvingFileIO overrode tryToWriteAtomic so atomic small writes reach the resolved FileIO, but not newTwoPhaseOutputStream: with resolving-file-io.enabled the interface default wrapped the resolver itself in a rename-based committer, bypassing the resolved OSS/S3 FileIO's native multipart two-phase commit — the exact bypass the class guards against elsewhere. Forward the call like the sibling methods. Adds a regression test asserting the delegate's stream is returned and no rename fallback runs (verified red on the original code). Assisted-by: GLM-5.3 --- .../org/apache/paimon/fs/ResolvingFileIO.java | 9 +++++++++ .../apache/paimon/fs/ResolvingFileIOTest.java | 18 ++++++++++++++++++ 2 files changed, 27 insertions(+) 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..ae06febd3b49 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 @@ -116,6 +116,15 @@ public boolean tryToWriteAtomic(Path path, String content) throws IOException { return wrap(() -> fileIO(path).tryToWriteAtomic(path, content)); } + @Override + public TwoPhaseOutputStream newTwoPhaseOutputStream(Path path, boolean overwrite) + throws IOException { + // Forward to the resolved FileIO so implementations with native multipart + // commits (object storage) keep them; the interface default would wrap this + // resolver in a rename-based committer instead. + return wrap(() -> fileIO(path).newTwoPhaseOutputStream(path, overwrite)); + } + @Override public String createBlobPresignedUrl( Path tableRoot, BlobDescriptor descriptor, Duration validity) throws IOException { 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..e5203833aa27 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 @@ -184,4 +184,22 @@ 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 testNewTwoPhaseOutputStreamReachesResolvedOverride() throws IOException { + FileIO delegate = mock(FileIO.class); + FileIOLoader loader = mock(FileIOLoader.class); + when(loader.load(any())).thenReturn(delegate); + when(loader.getScheme()).thenReturn("oss"); + resolvingFileIO.configure(CatalogContext.create(new Options(), loader, null)); + + Path target = new Path("oss://bucket/table/data.parquet"); + TwoPhaseOutputStream mockStream = mock(TwoPhaseOutputStream.class); + when(delegate.newTwoPhaseOutputStream(target, false)).thenReturn(mockStream); + + assertEquals(mockStream, resolvingFileIO.newTwoPhaseOutputStream(target, false)); + verify(delegate).newTwoPhaseOutputStream(target, false); + // the interface default would have renamed a temp file on the resolver instead + verify(delegate, never()).rename(any(), any()); + } } From 2e23646b1d8211e2ba425cb60509185bd89973a5 Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Sat, 5 Sep 2026 20:57:03 +0800 Subject: [PATCH 2/2] [common] Resolve a ResolvingFileIO before building the multipart upload store --- .../fs/BaseMultiPartUploadCommitter.java | 5 + .../org/apache/paimon/fs/ResolvingFileIO.java | 2 - .../fs/BaseMultiPartUploadCommitterTest.java | 97 +++++++++++++++++++ 3 files changed, 102 insertions(+), 2 deletions(-) create mode 100644 paimon-common/src/test/java/org/apache/paimon/fs/BaseMultiPartUploadCommitterTest.java diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/BaseMultiPartUploadCommitter.java b/paimon-common/src/main/java/org/apache/paimon/fs/BaseMultiPartUploadCommitter.java index 5245dcc161d4..2734b4324868 100644 --- a/paimon-common/src/main/java/org/apache/paimon/fs/BaseMultiPartUploadCommitter.java +++ b/paimon-common/src/main/java/org/apache/paimon/fs/BaseMultiPartUploadCommitter.java @@ -110,6 +110,11 @@ private MultiPartUploadStore multiPartUploadStore(FileIO fileIO) throws IO RESTTokenFileIO restTokenFileIO = (RESTTokenFileIO) fileIO; fileIO = restTokenFileIO.fileIO(); } + if (fileIO instanceof ResolvingFileIO) { + // The upload was started on the FileIO this resolver resolved to, and + // multiPartUploadStore casts to that concrete type, so resolve again here. + fileIO = ((ResolvingFileIO) fileIO).fileIO(targetPath()); + } return multiPartUploadStore(fileIO, targetPath()); } } 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 ae06febd3b49..cc3aba497e65 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 @@ -18,7 +18,6 @@ package org.apache.paimon.fs; -import org.apache.paimon.annotation.VisibleForTesting; import org.apache.paimon.catalog.CatalogContext; import org.apache.paimon.data.BlobDescriptor; import org.apache.paimon.options.CatalogOptions; @@ -134,7 +133,6 @@ public String createBlobPresignedUrl( .createBlobPresignedUrl(tableRoot, descriptor, validity)); } - @VisibleForTesting public FileIO fileIO(Path path) throws IOException { CacheKey cacheKey = new CacheKey(path.toUri().getScheme(), path.toUri().getAuthority()); return fileIOMap.computeIfAbsent( diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/BaseMultiPartUploadCommitterTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/BaseMultiPartUploadCommitterTest.java new file mode 100644 index 000000000000..c23fa5896f23 --- /dev/null +++ b/paimon-common/src/test/java/org/apache/paimon/fs/BaseMultiPartUploadCommitterTest.java @@ -0,0 +1,97 @@ +/* + * 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.options.Options; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.util.Collections; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** Tests for {@link BaseMultiPartUploadCommitter}. */ +public class BaseMultiPartUploadCommitterTest { + + private static final Path TARGET = new Path("oss://bucket/table/data-0.parquet"); + + private FileIO resolved; + private ResolvingFileIO resolvingFileIO; + + @BeforeEach + public void setUp() throws IOException { + resolved = mock(FileIO.class); + FileIOLoader loader = mock(FileIOLoader.class); + when(loader.getScheme()).thenReturn("oss"); + when(loader.load(any())).thenReturn(resolved); + resolvingFileIO = new ResolvingFileIO(); + resolvingFileIO.configure(CatalogContext.create(new Options(), loader, null)); + } + + @Test + public void testCommitResolvesResolvingFileIO() throws IOException { + RecordingCommitter committer = new RecordingCommitter(); + committer.commit(resolvingFileIO); + // the subclasses cast this to their own concrete FileIO, so the resolver itself + // reaching them would be a ClassCastException at commit time + assertThat(committer.received).isSameAs(resolved); + } + + @Test + public void testDiscardStagingResolvesResolvingFileIO() throws IOException { + RecordingCommitter committer = new RecordingCommitter(); + committer.discardStaging(resolvingFileIO); + assertThat(committer.received).isSameAs(resolved); + } + + @Test + public void testConcreteFileIOIsPassedThroughUnchanged() throws IOException { + RecordingCommitter committer = new RecordingCommitter(); + committer.commit(resolved); + assertThat(committer.received).isSameAs(resolved); + } + + private static class RecordingCommitter extends BaseMultiPartUploadCommitter { + + private FileIO received; + + private RecordingCommitter() { + super( + "upload-id", + Collections.singletonList("part-1"), + "table/data-0.parquet", + 1L, + TARGET); + } + + @Override + @SuppressWarnings("unchecked") + protected MultiPartUploadStore multiPartUploadStore( + FileIO fileIO, Path targetPath) { + this.received = fileIO; + return mock(MultiPartUploadStore.class); + } + } +}