Add append support for GCS, fixes #7619

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);
+    }
+  }
+}

plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileObject.java

@@ -20,6 +20,7 @@
 
 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.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 @@ protected OutputStream doGetOutputStream(boolean bAppend) throws Exception {
       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()) {

plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileProvider.java

@@ -52,6 +52,9 @@ public GoogleStorageFileProvider(
     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 GoogleStorageFileProvider(
           Capability.LIST_CHILDREN,
           Capability.READ_CONTENT,
           Capability.URI,
-          Capability.WRITE_CONTENT);
+          Capability.WRITE_CONTENT,
+          Capability.APPEND_CONTENT);
 
   @Override
   public Collection<Capability> getCapabilities() {

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));
+  }
+}

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");
+  }
+}