Add support for Cloud Storage

.github/workflows/object-storage-adapter-check.yaml

@@ -44,6 +44,9 @@ env:
   AWS_SECRET_ACCESS_KEY: ${{ secrets.S3_SECRET_ACCESS_KEY }}
   S3_REGION: ap-northeast-1
   S3_BUCKET_NAME: scalardb-test-bucket
+  CLOUD_STORAGE_PROJECT_ID: ${{ secrets.CLOUD_STORAGE_PROJECT_ID }}
+  CLOUD_STORAGE_SERVICE_ACCOUNT_KEY: ${{ secrets.CLOUD_STORAGE_SERVICE_ACCOUNT_KEY }}
+  CLOUD_STORAGE_BUCKET_NAME: scalardb-test-bucket
 
 jobs:
   integration-test-s3:
@@ -98,5 +101,59 @@ jobs:
         if: always()
         uses: actions/upload-artifact@v5
         with:
-          name: cassandra_3.0_integration_test_reports_${{ matrix.mode.label }}
+          name: s3_integration_test_reports_${{ matrix.mode.label }}
+          path: core/build/reports/tests/integrationTestObjectStorage
+  integration-test-cloud-storage:
+    name: Cloud Storage integration test (${{ matrix.mode.label }})
+    runs-on: ubuntu-latest
+
+    strategy:
+      fail-fast: false
+      matrix:
+        mode:
+          - label: default
+            group_commit_enabled: false
+          - label: with_group_commit
+            group_commit_enabled: true
+
+    steps:
+      - uses: actions/checkout@v5
+
+      - name: Set up JDK ${{ env.JAVA_VERSION }} (${{ env.JAVA_VENDOR }})
+        uses: actions/setup-java@v5
+        with:
+          java-version: ${{ env.JAVA_VERSION }}
+          distribution: ${{ env.JAVA_VENDOR }}
+
+      - name: Set up JDK ${{ env.INT_TEST_JAVA_RUNTIME_VERSION }} (${{ env.INT_TEST_JAVA_RUNTIME_VENDOR }}) to run integration test
+        uses: actions/setup-java@v5
+        if: ${{ env.SET_UP_INT_TEST_RUNTIME_NON_ORACLE_JDK == 'true'}}
+        with:
+          java-version: ${{ env.INT_TEST_JAVA_RUNTIME_VERSION }}
+          distribution: ${{ env.INT_TEST_JAVA_RUNTIME_VENDOR }}
+
+      - name: Login to Oracle container registry
+        uses: docker/login-action@v3
+        if: ${{ env.INT_TEST_JAVA_RUNTIME_VENDOR == 'oracle' }}
+        with:
+          registry: container-registry.oracle.com
+          username: ${{ secrets.OCR_USERNAME }}
+          password: ${{ secrets.OCR_TOKEN }}
+
+      - name: Set up JDK ${{ env.INT_TEST_JAVA_RUNTIME_VERSION }} (oracle) to run the integration test
+        if: ${{ env.INT_TEST_JAVA_RUNTIME_VENDOR == 'oracle' }}
+        run: |
+          container_id=$(docker create "container-registry.oracle.com/java/jdk:${{ env.INT_TEST_JAVA_RUNTIME_VERSION }}")
+          docker cp -L "$container_id:/usr/java/default" /usr/lib/jvm/oracle-jdk && docker rm "$container_id"
+      - name: Setup Gradle
+        uses: gradle/actions/setup-gradle@v5
+
+      - name: Execute Gradle 'integrationTestObjectStorage' task
+        run: ./gradlew integrationTestObjectStorage -Dscalardb.object_storage.storage=cloud-storage -Dscalardb.object_storage.endpoint=scalardb-test-bucket -Dscalardb.object_storage.username=${{ env.CLOUD_STORAGE_PROJECT_ID }} -Dscalardb.object_storage.password=${{ env.CLOUD_STORAGE_SERVICE_ACCOUNT_KEY }} ${{ matrix.mode.group_commit_enabled && env.INT_TEST_GRADLE_OPTIONS_FOR_GROUP_COMMIT || '' }}
+
+      - name: Upload Gradle test reports
+        if: always()
+        uses: actions/upload-artifact@v5
+        with:
+          name: cloud_storage_integration_test_reports_${{ matrix.mode.label }}
           path: core/build/reports/tests/integrationTestObjectStorage

build.gradle

@@ -41,6 +41,7 @@ subprojects {
         db2DriverVersion = '12.1.3.0'
         mariadDbDriverVersion = '3.5.6'
         alloyDbJdbcConnectorVersion = '1.2.8'
+        googleCloudStorageVersion = '2.60.0'
         picocliVersion = '4.7.7'
         commonsTextVersion = '1.14.0'
         junitVersion = '5.14.1'

core/build.gradle

@@ -193,6 +193,9 @@ dependencies {
     implementation("com.google.cloud:alloydb-jdbc-connector:${alloyDbJdbcConnectorVersion}") {
         exclude group: 'org.slf4j', module: 'slf4j-api'
     }
+    implementation("com.google.cloud:google-cloud-storage:${googleCloudStorageVersion}") {
+        exclude group: 'org.slf4j', module: 'slf4j-api'
+    }
     implementation "org.apache.commons:commons-text:${commonsTextVersion}"
     testImplementation platform("org.junit:junit-bom:${junitVersion}")
     testImplementation 'org.junit.jupiter:junit-jupiter'
@@ -226,7 +229,7 @@ task integrationTestCassandra(type: Test) {
     classpath = sourceSets.integrationTestCassandra.runtimeClasspath
     outputs.upToDateWhen { false }  // ensures integration tests are run every time when called
     options {
-        systemProperties(System.getProperties().findAll{it.key.toString().startsWith("scalardb")})
+        systemProperties(System.getProperties().findAll { it.key.toString().startsWith("scalardb") })
     }
 }
 
@@ -237,7 +240,7 @@ task integrationTestCosmos(type: Test) {
     classpath = sourceSets.integrationTestCosmos.runtimeClasspath
     outputs.upToDateWhen { false }  // ensures integration tests are run every time when called
     options {
-        systemProperties(System.getProperties().findAll{it.key.toString().startsWith("scalardb")})
+        systemProperties(System.getProperties().findAll { it.key.toString().startsWith("scalardb") })
     }
     jvmArgs '-XX:MaxDirectMemorySize=4g', '-Xmx6g',
             // INFO com.azure.cosmos.implementation.RxDocumentClientImpl - Initializing DocumentClient [3] with serviceEndpoint [https://localhost:8081/], ...
@@ -255,7 +258,7 @@ task integrationTestDynamo(type: Test) {
     classpath = sourceSets.integrationTestDynamo.runtimeClasspath
     outputs.upToDateWhen { false }  // ensures integration tests are run every time when called
     options {
-        systemProperties(System.getProperties().findAll{it.key.toString().startsWith("scalardb")})
+        systemProperties(System.getProperties().findAll { it.key.toString().startsWith("scalardb") })
     }
     maxParallelForks = 10
 }
@@ -267,7 +270,7 @@ task integrationTestJdbc(type: Test) {
     classpath = sourceSets.integrationTestJdbc.runtimeClasspath
     outputs.upToDateWhen { false }  // ensures integration tests are run every time when called
     options {
-        systemProperties(System.getProperties().findAll{it.key.toString().startsWith("scalardb")})
+        systemProperties(System.getProperties().findAll { it.key.toString().startsWith("scalardb") })
     }
     maxHeapSize = "4g"
 }
@@ -279,7 +282,7 @@ task integrationTestObjectStorage(type: Test) {
     classpath = sourceSets.integrationTestObjectStorage.runtimeClasspath
     outputs.upToDateWhen { false }  // ensures integration tests are run every time when called
     options {
-        systemProperties(System.getProperties().findAll{it.key.toString().startsWith("scalardb")})
+        systemProperties(System.getProperties().findAll { it.key.toString().startsWith("scalardb") })
     }
 }
 
@@ -290,7 +293,7 @@ task integrationTestMultiStorage(type: Test) {
     classpath = sourceSets.integrationTestMultiStorage.runtimeClasspath
     outputs.upToDateWhen { false }  // ensures integration tests are run every time when called
     options {
-        systemProperties(System.getProperties().findAll{it.key.toString().startsWith("scalardb")})
+        systemProperties(System.getProperties().findAll { it.key.toString().startsWith("scalardb") })
     }
 }
 

core/src/integration-test/java/com/scalar/db/storage/objectstorage/ObjectStorageEnv.java

@@ -2,6 +2,7 @@
 
 import com.scalar.db.config.DatabaseConfig;
 import com.scalar.db.storage.objectstorage.blobstorage.BlobStorageConfig;
+import com.scalar.db.storage.objectstorage.cloudstorage.CloudStorageConfig;
 import com.scalar.db.storage.objectstorage.s3.S3Config;
 import java.util.Collections;
 import java.util.Map;
@@ -76,6 +77,11 @@ public static boolean isBlobStorage() {
         .equals(BlobStorageConfig.STORAGE_NAME);
   }
 
+  public static boolean isCloudStorage() {
+    return System.getProperty(PROP_OBJECT_STORAGE_STORAGE, DEFAULT_OBJECT_STORAGE_STORAGE)
+        .equals(CloudStorageConfig.STORAGE_NAME);
+  }
+
   public static boolean isS3() {
     return System.getProperty(PROP_OBJECT_STORAGE_STORAGE, DEFAULT_OBJECT_STORAGE_STORAGE)
         .equals(S3Config.STORAGE_NAME);

core/src/integration-test/java/com/scalar/db/storage/objectstorage/ObjectStorageWrapperIntegrationTest.java

@@ -28,8 +28,11 @@ public class ObjectStorageWrapperIntegrationTest {
   private static final String TEST_OBJECT2 = "test-object2";
   private static final String TEST_OBJECT3 = "test-object3";
   private static final int BLOB_STORAGE_LIST_MAX_KEYS = 5000;
+  private static final int CLOUD_STORAGE_LIST_MAX_KEYS = 1000;
+  private static final int S3_LIST_MAX_KEYS = 1000;
 
   private ObjectStorageWrapper wrapper;
+  private int listMaxKeys;
 
   @BeforeAll
   public void beforeAll() throws ObjectStorageWrapperException {
@@ -38,6 +41,16 @@ public void beforeAll() throws ObjectStorageWrapperException {
         ObjectStorageUtils.getObjectStorageConfig(new DatabaseConfig(properties));
     wrapper = ObjectStorageWrapperFactory.create(objectStorageConfig);
     createObjects();
+
+    if (ObjectStorageEnv.isBlobStorage()) {
+      listMaxKeys = BLOB_STORAGE_LIST_MAX_KEYS;
+    } else if (ObjectStorageEnv.isCloudStorage()) {
+      listMaxKeys = CLOUD_STORAGE_LIST_MAX_KEYS;
+    } else if (ObjectStorageEnv.isS3()) {
+      listMaxKeys = S3_LIST_MAX_KEYS;
+    } else {
+      throw new AssertionError();
+    }
   }
 
   @AfterAll
@@ -152,14 +165,14 @@ public void update_NonExistingObjectKeyGiven_ShouldThrowPreconditionFailedExcept
     String objectKey = "non-existing-key";
 
     // Act Assert
-    assertThatCode(() -> wrapper.update(objectKey, "some-object", "some-version"))
+    assertThatCode(() -> wrapper.update(objectKey, "some-object", "123456789"))
         .isInstanceOf(PreconditionFailedException.class);
   }
 
   @Test
   public void update_WrongVersionGiven_ShouldThrowPreconditionFailedException() {
     // Arrange
-    String wrongVersion = "wrong-version";
+    String wrongVersion = "123456789";
 
     // Act Assert
     assertThatCode(() -> wrapper.update(TEST_KEY2, "another-object", wrongVersion))
@@ -219,7 +232,7 @@ public void delete_ExistingObjectKeyWithWrongVersionGiven_ShouldThrowPreconditio
     // Arrange
     Optional<ObjectStorageWrapperResponse> response1 = wrapper.get(TEST_KEY1);
     assertThat(response1.isPresent()).isTrue();
-    String wrongVersion = "wrong-version";
+    String wrongVersion = "123456789";
 
     // Act Assert
     assertThatCode(() -> wrapper.delete(TEST_KEY1, wrongVersion))
@@ -253,7 +266,7 @@ public void getKeys_WithNonExistingPrefix_ShouldReturnEmptySet() throws Exceptio
   public void getKeys_WithPrefixForTheNumberOfObjectsExceedingTheListLimit_ShouldReturnAllKeys()
       throws Exception {
     String prefix = "prefix-";
-    int numberOfObjects = BLOB_STORAGE_LIST_MAX_KEYS + 1;
+    int numberOfObjects = listMaxKeys + 1;
     try {
       // Arrange
       for (int i = 0; i < numberOfObjects; i++) {
@@ -313,7 +326,7 @@ public void deleteByPrefix_WithNonExistingPrefix_ShouldDoNothing() throws Except
       deleteByPrefix_WithPrefixForTheNumberOfObjectsExceedingTheListLimit_ShouldDeleteAllObjects()
           throws Exception {
     String prefix = "prefix-";
-    int numberOfObjects = BLOB_STORAGE_LIST_MAX_KEYS + 1;
+    int numberOfObjects = listMaxKeys + 1;
     try {
       // Arrange
       for (int i = 0; i < numberOfObjects; i++) {

core/src/integration-test/java/com/scalar/db/storage/objectstorage/ObjectStorageWrapperLargeObjectWriteIntegrationTest.java

@@ -5,6 +5,7 @@
 
 import com.scalar.db.config.DatabaseConfig;
 import com.scalar.db.storage.objectstorage.blobstorage.BlobStorageConfig;
+import com.scalar.db.storage.objectstorage.cloudstorage.CloudStorageConfig;
 import com.scalar.db.storage.objectstorage.s3.S3Config;
 import java.util.Arrays;
 import java.util.Optional;
@@ -47,6 +48,13 @@ public void beforeAll() throws ObjectStorageWrapperException {
           BlobStorageConfig.PARALLEL_UPLOAD_THRESHOLD_IN_BYTES,
           String.valueOf(parallelUploadUnit * 2));
       parallelUploadThresholdInBytes = parallelUploadUnit * 2;
+    } else if (ObjectStorageEnv.isCloudStorage()) {
+      // Minimum block size must be greater than or equal to 256KB for Cloud Storage
+      Long parallelUploadUnit = 256 * 1024L; // 256KB
+      properties.setProperty(
+          CloudStorageConfig.PARALLEL_UPLOAD_BLOCK_SIZE_IN_BYTES,
+          String.valueOf(parallelUploadUnit));
+      parallelUploadThresholdInBytes = parallelUploadUnit * 2;
     } else if (ObjectStorageEnv.isS3()) {
       // Minimum part size must be greater than or equal to 5MB for S3
       Long parallelUploadUnit = 5 * 1024 * 1024L; // 5MB
@@ -59,7 +67,7 @@ public void beforeAll() throws ObjectStorageWrapperException {
       throw new AssertionError();
     }
 
-    char[] charArray = new char[(int) parallelUploadThresholdInBytes];
+    char[] charArray = new char[(int) parallelUploadThresholdInBytes + 1];
     Arrays.fill(charArray, 'a');
     testObject1 = new String(charArray);
     Arrays.fill(charArray, 'b');
@@ -163,14 +171,14 @@ public void update_NonExistingObjectKeyGiven_ShouldThrowPreconditionFailedExcept
     String objectKey = "non-existing-key";
 
     // Act Assert
-    assertThatCode(() -> wrapper.update(objectKey, "some-object", "some-version"))
+    assertThatCode(() -> wrapper.update(objectKey, "some-object", "123456789"))
         .isInstanceOf(PreconditionFailedException.class);
   }
 
   @Test
   public void update_WrongVersionGiven_ShouldThrowPreconditionFailedException() {
     // Arrange
-    String wrongVersion = "wrong-version";
+    String wrongVersion = "123456789";
 
     // Act Assert
     assertThatCode(() -> wrapper.update(TEST_KEY2, "another-object", wrongVersion))

core/src/main/java/com/scalar/db/common/CoreError.java

@@ -931,6 +931,18 @@ public enum CoreError implements ScalarDbError {
       "Conditions on indexed columns in cross-partition scan operations are not allowed in the SERIALIZABLE isolation level",
       "",
       ""),
+  OBJECT_STORAGE_CLOUD_STORAGE_SERVICE_ACCOUNT_KEY_NOT_FOUND(
+      Category.USER_ERROR,
+      "0263",
+      "The service account key for Cloud Storage is not found.",
+      "",
+      ""),
+  OBJECT_STORAGE_CLOUD_STORAGE_SERVICE_ACCOUNT_KEY_LOAD_FAILED(
+      Category.USER_ERROR,
+      "0264",
+      "Failed to load the service account key for Cloud Storage.",
+      "",
+      ""),
 
   //
   // Errors for the concurrency error category

core/src/main/java/com/scalar/db/storage/objectstorage/ObjectStorage.java

@@ -154,6 +154,10 @@ public void mutate(List<? extends Mutation> mutations) throws ExecutionException
 
   @Override
   public void close() {
-    wrapper.close();
+    try {
+      wrapper.close();
+    } catch (ObjectStorageWrapperException e) {
+      logger.warn("Failed to close the ObjectStorageWrapper", e);
+    }
   }
 }

core/src/main/java/com/scalar/db/storage/objectstorage/ObjectStorageAdmin.java

@@ -21,12 +21,15 @@
 import java.util.stream.Collectors;
 import javax.annotation.Nullable;
 import javax.annotation.concurrent.ThreadSafe;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 @ThreadSafe
 public class ObjectStorageAdmin implements DistributedStorageAdmin {
   public static final String NAMESPACE_METADATA_TABLE = "namespaces";
   public static final String TABLE_METADATA_TABLE = "metadata";
 
+  private static final Logger logger = LoggerFactory.getLogger(ObjectStorageAdmin.class);
   private static final StorageInfo STORAGE_INFO =
       new StorageInfoImpl(
           "object_storage",
@@ -80,7 +83,11 @@ public StorageInfo getStorageInfo(String namespace) throws ExecutionException {
 
   @Override
   public void close() {
-    wrapper.close();
+    try {
+      wrapper.close();
+    } catch (ObjectStorageWrapperException e) {
+      logger.warn("Failed to close the ObjectStorageWrapper", e);
+    }
   }
 
   @Override

core/src/main/java/com/scalar/db/storage/objectstorage/ObjectStorageConfig.java

@@ -9,13 +9,6 @@ public interface ObjectStorageConfig {
    */
   String getStorageName();
 
-  /**
-   * Returns the username for authentication.
-   *
-   * @return the username
-   */
-  String getUsername();
-
   /**
    * Returns the password for authentication.
    *

core/src/main/java/com/scalar/db/storage/objectstorage/ObjectStorageUtils.java

@@ -2,6 +2,7 @@
 
 import com.scalar.db.config.DatabaseConfig;
 import com.scalar.db.storage.objectstorage.blobstorage.BlobStorageConfig;
+import com.scalar.db.storage.objectstorage.cloudstorage.CloudStorageConfig;
 import com.scalar.db.storage.objectstorage.s3.S3Config;
 import java.util.Objects;
 
@@ -26,6 +27,8 @@ public static ObjectStorageConfig getObjectStorageConfig(DatabaseConfig database
       return new BlobStorageConfig(databaseConfig);
     } else if (Objects.equals(databaseConfig.getStorage(), S3Config.STORAGE_NAME)) {
       return new S3Config(databaseConfig);
+    } else if (Objects.equals(databaseConfig.getStorage(), CloudStorageConfig.STORAGE_NAME)) {
+      return new CloudStorageConfig(databaseConfig);
     } else {
       throw new IllegalArgumentException(
           "Unsupported Object Storage: " + databaseConfig.getStorage());

core/src/main/java/com/scalar/db/storage/objectstorage/ObjectStorageWrapper.java

@@ -66,13 +66,16 @@ public interface ObjectStorageWrapper {
   void delete(String key, String version) throws ObjectStorageWrapperException;
 
   /**
-   * Delete objects with the specified prefix from the storage.
+   * Delete objects with the specified prefix from the storage. <br>
+   * <br>
+   * <strong>Attention:</strong> This method does not guarantee atomicity and is assumed to be used
+   * where concurrent operations do not occur.
    *
    * @param prefix the prefix of the objects to delete
    * @throws ObjectStorageWrapperException if an error occurs
    */
   void deleteByPrefix(String prefix) throws ObjectStorageWrapperException;
 
   /** Close the storage wrapper. */
-  void close();
+  void close() throws ObjectStorageWrapperException;
 }