This is an automated email from the ASF dual-hosted git repository.
hansva pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new 6344943282 Add append support for GCS, fixes #7619 (#7620)
6344943282 is described below
commit 634494328264c742d0e023359315794808743025
Author: Hans Van Akelyen <[email protected]>
AuthorDate: Fri Jul 24 17:24:44 2026 +0200
Add append support for GCS, fixes #7619 (#7620)
---
.../hop/vfs/gs/ComposeAppendOutputStream.java | 121 ++++++++++++++++
.../apache/hop/vfs/gs/GoogleStorageFileObject.java | 28 +++-
.../hop/vfs/gs/GoogleStorageFileProvider.java | 6 +-
.../hop/vfs/gs/ComposeAppendOutputStreamTest.java | 160 +++++++++++++++++++++
.../GoogleStorageFileProviderCapabilitiesTest.java | 40 ++++++
5 files changed, 352 insertions(+), 3 deletions(-)
diff --git
a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/ComposeAppendOutputStream.java
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/ComposeAppendOutputStream.java
new file mode 100644
index 0000000000..283630911a
--- /dev/null
+++
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/ComposeAppendOutputStream.java
@@ -0,0 +1,121 @@
+/*
+ * 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.hop.vfs.gs;
+
+import com.google.cloud.WriteChannel;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BlobTargetOption;
+import com.google.cloud.storage.Storage.ComposeRequest;
+import java.io.IOException;
+import java.io.OutputStream;
+import org.apache.hop.core.logging.LogChannel;
+
+/**
+ * Append output stream for Google Cloud Storage, built on <a
+ *
href="https://cloud.google.com/storage/docs/composite-objects#appends";>composite
objects</a>.
+ *
+ * <p>GCS objects are immutable, so there is no true in-place append. Instead
the bytes written to
+ * this stream are streamed into a short-lived <em>temporary</em> object, and
on {@link #close()}
+ * the target and the temporary object are concatenated back onto the target
with a single {@code
+ * compose} call ({@code target = target + temp}). The temporary object is
then deleted.
+ *
+ * <p>The compose is guarded with an {@code ifGenerationMatch} precondition
captured when the stream
+ * was opened, so a concurrent modification of the target fails the append
fast instead of silently
+ * dropping data. One open/close cycle performs exactly one compose regardless
of how many times
+ * {@code write(...)} is called, so a transform that appends many rows still
costs a single compose.
+ *
+ * <p>Note that composite objects have no MD5 metadata (CRC32C is still
maintained and validated by
+ * the compose operation).
+ */
+public class ComposeAppendOutputStream extends OutputStream {
+
+ private final Storage storage;
+ private final String bucketName;
+ private final String targetName;
+ private final String tempName;
+ private final long targetGeneration;
+
+ /** Streams the appended bytes into the temporary object; reused for its
robust write/close. */
+ private final WriteChannelOutputStream tempStream;
+
+ private boolean closed = false;
+
+ public ComposeAppendOutputStream(
+ Storage storage,
+ String bucketName,
+ String targetName,
+ String tempName,
+ long targetGeneration,
+ WriteChannel tempChannel) {
+ this.storage = storage;
+ this.bucketName = bucketName;
+ this.targetName = targetName;
+ this.tempName = tempName;
+ this.targetGeneration = targetGeneration;
+ this.tempStream = new WriteChannelOutputStream(tempChannel);
+ }
+
+ @Override
+ public void write(int b) throws IOException {
+ tempStream.write(b);
+ }
+
+ @Override
+ public void write(byte[] buf, int off, int len) throws IOException {
+ tempStream.write(buf, off, len);
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ try {
+ // Finish streaming the appended bytes into the temporary object.
+ tempStream.close();
+ // Concatenate: target = target + temp. The generation precondition
makes a concurrent
+ // modification of the target fail here rather than silently lose the
appended data.
+ storage.compose(
+ ComposeRequest.newBuilder()
+ .addSource(targetName)
+ .addSource(tempName)
+ .setTarget(BlobInfo.newBuilder(bucketName, targetName).build())
+
.setTargetOptions(BlobTargetOption.generationMatch(targetGeneration))
+ .build());
+ } catch (RuntimeException e) {
+ throw new IOException(
+ "Unable to append to gs://" + bucketName + "/" + targetName + "
using compose", e);
+ } finally {
+ deleteTempQuietly();
+ }
+ }
+
+ /** The temporary object is throwaway; a failed delete only leaves a stray
object, so log it. */
+ private void deleteTempQuietly() {
+ try {
+ storage.delete(BlobId.of(bucketName, tempName));
+ } catch (RuntimeException e) {
+ LogChannel.GENERAL.logError(
+ "Unable to delete temporary append object gs://" + bucketName + "/"
+ tempName, e);
+ }
+ }
+}
diff --git
a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileObject.java
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileObject.java
index f487dd96c7..c43b8e61ae 100644
---
a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileObject.java
+++
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileObject.java
@@ -20,6 +20,7 @@ package org.apache.hop.vfs.gs;
import com.google.api.gax.paging.Page;
import com.google.api.gax.rpc.NotFoundException;
+import com.google.cloud.WriteChannel;
import com.google.cloud.storage.Blob;
import com.google.cloud.storage.BlobId;
import com.google.cloud.storage.BlobInfo;
@@ -42,6 +43,7 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.UUID;
import org.apache.commons.vfs2.FileObject;
import org.apache.commons.vfs2.FileType;
import org.apache.commons.vfs2.provider.AbstractFileName;
@@ -342,14 +344,36 @@ public class GoogleStorageFileObject extends
AbstractFileObject<GoogleStorageFil
throw new IOException("Object needs a path within the bucket");
}
Storage storage = getAbstractFileSystem().setupStorage();
+ String objectName = stripTrailingSlash(bucketPath);
+
+ if (bAppend) {
+ Blob current = storage.get(BlobId.of(bucketName, objectName));
+ if (current != null && current.getSize() != null && current.getSize() >
0) {
+ return openComposeAppendStream(storage, current, objectName);
+ }
+ }
+
if (!hasObject()) {
- this.blob =
- storage.create(BlobInfo.newBuilder(bucket,
stripTrailingSlash(bucketPath)).build());
+ this.blob = storage.create(BlobInfo.newBuilder(bucket,
objectName).build());
}
getAbstractFileSystem().invalidateListCacheForParentOf(bucketName,
bucketPath);
return new WriteChannelOutputStream(storage.writer(blob));
}
+ /**
+ * Open a {@link ComposeAppendOutputStream} that streams the appended bytes
into a throwaway
+ * temporary object next to the target and, on close, composes {@code target
+ temp} back onto the
+ * target. See <a
href="https://cloud.google.com/storage/docs/composite-objects#appends";>GCS
+ * appends</a>.
+ */
+ private OutputStream openComposeAppendStream(Storage storage, Blob current,
String objectName) {
+ String tempName = objectName + ".hop-append-" + UUID.randomUUID() + ".tmp";
+ WriteChannel tempChannel = storage.writer(BlobInfo.newBuilder(bucketName,
tempName).build());
+ getAbstractFileSystem().invalidateListCacheForParentOf(bucketName,
bucketPath);
+ return new ComposeAppendOutputStream(
+ storage, bucketName, objectName, tempName, current.getGeneration(),
tempChannel);
+ }
+
@Override
protected FileObject[] doListChildrenResolved() throws Exception {
if (!isFolder()) {
diff --git
a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileProvider.java
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileProvider.java
index 7e6f97d381..5ccde11bf5 100644
---
a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileProvider.java
+++
b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileProvider.java
@@ -52,6 +52,9 @@ public class GoogleStorageFileProvider extends
AbstractOriginatingFileProvider {
setServiceAccountCredentials(variables, googleStorageMetadataType);
}
+ // APPEND_CONTENT is required for append: without it commons-vfs2 rejects
getOutputStream(true)
+ // with "does not support append mode". Append itself is emulated with
composite objects, see
+ // ComposeAppendOutputStream.
public static final Collection<Capability> capabilities =
Set.of(
Capability.CREATE,
@@ -63,7 +66,8 @@ public class GoogleStorageFileProvider extends
AbstractOriginatingFileProvider {
Capability.LIST_CHILDREN,
Capability.READ_CONTENT,
Capability.URI,
- Capability.WRITE_CONTENT);
+ Capability.WRITE_CONTENT,
+ Capability.APPEND_CONTENT);
@Override
public Collection<Capability> getCapabilities() {
diff --git
a/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/ComposeAppendOutputStreamTest.java
b/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/ComposeAppendOutputStreamTest.java
new file mode 100644
index 0000000000..79b734940e
--- /dev/null
+++
b/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/ComposeAppendOutputStreamTest.java
@@ -0,0 +1,160 @@
+/*
+ * 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.hop.vfs.gs;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+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.ArgumentMatchers.eq;
+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;
+
+import com.google.cloud.WriteChannel;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BlobTargetOption;
+import com.google.cloud.storage.Storage.ComposeRequest;
+import com.google.cloud.storage.StorageException;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.InOrder;
+import org.mockito.Mockito;
+
+/**
+ * Covers the composite-object append behaviour of {@link
ComposeAppendOutputStream}: bytes are
+ * streamed into a temporary object and, on close, the target and temp object
are concatenated with
+ * a single generation-guarded {@code compose}, after which the temp object is
deleted.
+ */
+class ComposeAppendOutputStreamTest {
+
+ private static final String BUCKET = "my-bucket";
+ private static final String TARGET = "folder/data.txt";
+ private static final String TEMP = "folder/data.txt.hop-append-abc.tmp";
+ private static final long GENERATION = 42L;
+
+ /** A WriteChannel mock that consumes every byte offered, like a healthy
upload. */
+ private static WriteChannel consumingChannel() {
+ WriteChannel channel = mock(WriteChannel.class);
+ try {
+ when(channel.write(any(ByteBuffer.class)))
+ .thenAnswer(
+ inv -> {
+ ByteBuffer b = inv.getArgument(0);
+ int remaining = b.remaining();
+ b.position(b.limit());
+ return remaining;
+ });
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ return channel;
+ }
+
+ @Test
+ void closeStreamsToTempThenComposesThenDeletesTemp() throws Exception {
+ Storage storage = mock(Storage.class);
+ WriteChannel tempChannel = consumingChannel();
+
+ ComposeAppendOutputStream out =
+ new ComposeAppendOutputStream(storage, BUCKET, TARGET, TEMP,
GENERATION, tempChannel);
+ out.write("appended".getBytes(StandardCharsets.UTF_8));
+ out.close();
+
+ // The appended bytes were streamed into the temp object and the channel
closed.
+ verify(tempChannel).close();
+
+ ArgumentCaptor<ComposeRequest> captor =
ArgumentCaptor.forClass(ComposeRequest.class);
+ verify(storage).compose(captor.capture());
+ ComposeRequest request = captor.getValue();
+
+ // Sources are, in order, the existing target then the temp object (target
= target + temp).
+ assertEquals(2, request.getSourceBlobs().size());
+ assertEquals(TARGET, request.getSourceBlobs().get(0).getName());
+ assertEquals(TEMP, request.getSourceBlobs().get(1).getName());
+
+ // The composite is written back onto the target.
+ assertEquals(BUCKET, request.getTarget().getBucket());
+ assertEquals(TARGET, request.getTarget().getName());
+
+ // Guarded by the generation captured when the stream opened.
+ assertTrue(
+
request.getTargetOptions().contains(BlobTargetOption.generationMatch(GENERATION)),
+ "append must be guarded by ifGenerationMatch to fail fast on
concurrent modification");
+
+ // And the throwaway temp object is cleaned up, after the compose.
+ InOrder inOrder = Mockito.inOrder(storage);
+ inOrder.verify(storage).compose(any(ComposeRequest.class));
+ inOrder.verify(storage).delete(BlobId.of(BUCKET, TEMP));
+ }
+
+ @Test
+ void composeFailurePropagatesButTempIsStillDeleted() throws Exception {
+ Storage storage = mock(Storage.class);
+ when(storage.compose(any(ComposeRequest.class)))
+ .thenThrow(new StorageException(412, "precondition failed"));
+ WriteChannel tempChannel = consumingChannel();
+
+ ComposeAppendOutputStream out =
+ new ComposeAppendOutputStream(storage, BUCKET, TARGET, TEMP,
GENERATION, tempChannel);
+ out.write("x".getBytes(StandardCharsets.UTF_8));
+
+ IOException thrown = assertThrows(IOException.class, out::close);
+ assertTrue(thrown.getMessage().contains(TARGET), "the error should name
the target object");
+ // Even on a failed compose the temp object must not be leaked.
+ verify(storage).delete(BlobId.of(BUCKET, TEMP));
+ }
+
+ @Test
+ void closeIsIdempotent() throws Exception {
+ Storage storage = mock(Storage.class);
+ WriteChannel tempChannel = consumingChannel();
+
+ ComposeAppendOutputStream out =
+ new ComposeAppendOutputStream(storage, BUCKET, TARGET, TEMP,
GENERATION, tempChannel);
+ out.close();
+ out.close();
+
+ // A second close must not compose (or delete) again.
+ verify(storage, times(1)).compose(any(ComposeRequest.class));
+ verify(storage, times(1)).delete(eq(BlobId.of(BUCKET, TEMP)));
+ }
+
+ @Test
+ void nothingIsComposedBeforeClose() throws Exception {
+ Storage storage = mock(Storage.class);
+ WriteChannel tempChannel = consumingChannel();
+
+ ComposeAppendOutputStream out =
+ new ComposeAppendOutputStream(storage, BUCKET, TARGET, TEMP,
GENERATION, tempChannel);
+ out.write("streaming".getBytes(StandardCharsets.UTF_8));
+
+ // Writing alone must not compose; the concatenation happens exactly once,
at close.
+ verify(storage, never()).compose(any(ComposeRequest.class));
+
+ out.close();
+ verify(storage, times(1)).compose(any(ComposeRequest.class));
+ }
+}
diff --git
a/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/GoogleStorageFileProviderCapabilitiesTest.java
b/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/GoogleStorageFileProviderCapabilitiesTest.java
new file mode 100644
index 0000000000..b400d6af3d
--- /dev/null
+++
b/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/GoogleStorageFileProviderCapabilitiesTest.java
@@ -0,0 +1,40 @@
+/*
+ * 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.hop.vfs.gs;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.apache.commons.vfs2.Capability;
+import org.junit.jupiter.api.Test;
+
+class GoogleStorageFileProviderCapabilitiesTest {
+
+ /**
+ * commons-vfs2 rejects {@code getOutputStream(true)} with "does not support
append mode" unless
+ * the file system advertises {@link Capability#APPEND_CONTENT}. Append is
implemented with
+ * composite objects (see {@link ComposeAppendOutputStream}), so the
capability must be present or
+ * every append is blocked before reaching the provider.
+ */
+ @Test
+ void appendContentIsAdvertised() {
+ assertTrue(
+
GoogleStorageFileProvider.capabilities.contains(Capability.APPEND_CONTENT),
+ "GCS VFS must advertise APPEND_CONTENT so commons-vfs2 allows append
mode");
+ }
+}
