Merged changes and resolved conflicts with master branch

Author: inv-jishnu

.github/workflows/ci.yaml

@@ -67,8 +67,12 @@ jobs:
         run: |
           mkdir -p /tmp/gradle_test_reports/core
           mkdir -p /tmp/gradle_test_reports/schema-loader
+          mkdir -p /tmp/gradle_test_reports/data-loader/core
+          mkdir -p /tmp/gradle_test_reports/data-loader/cli
           cp -a core/build/reports/tests/test /tmp/gradle_test_reports/core/
           cp -a schema-loader/build/reports/tests/test /tmp/gradle_test_reports/schema-loader/
+          cp -a data-loader/core/build/reports/tests/test /tmp/gradle_test_reports/data-loader/core/
+          cp -a data-loader/cli/build/reports/tests/test /tmp/gradle_test_reports/data-loader/cli/
 
       - name: Upload Gradle test reports
         if: always()
@@ -83,9 +87,13 @@ jobs:
           mkdir -p /tmp/gradle_spotbugs_reports/core
           mkdir -p /tmp/gradle_spotbugs_reports/schema-loader
           mkdir -p /tmp/gradle_spotbugs_reports/integration-test
+          mkdir -p /tmp/gradle_spotbugs_reports/data-loader/core
+          mkdir -p /tmp/gradle_spotbugs_reports/data-loader/cli
           cp -a core/build/reports/spotbugs /tmp/gradle_spotbugs_reports/core/
           cp -a schema-loader/build/reports/spotbugs /tmp/gradle_spotbugs_reports/schema-loader/
           cp -a integration-test/build/reports/spotbugs /tmp/gradle_spotbugs_reports/integration-test/
+          cp -a data-loader/core/build/reports/spotbugs /tmp/gradle_spotbugs_reports/data-loader/core/
+          cp -a data-loader/cli/build/reports/spotbugs /tmp/gradle_spotbugs_reports/data-loader/cli/
 
       - name: Upload Spotbugs reports
         if: always()
@@ -113,6 +121,9 @@ jobs:
       - name: Dockerfile Lint for ScalarDB Schema Loader
         run: ./gradlew schema-loader:dockerfileLint
 
+      - name: Dockerfile Lint for ScalarDB Data Loader CLI
+        run: ./gradlew data-loader:cli:dockerfileLint
+
   integration-test-for-cassandra-3-0:
     name: Cassandra 3.0 integration test (${{ matrix.mode.label }})
     runs-on: ubuntu-latest

.github/workflows/create-release.yaml

@@ -77,3 +77,16 @@ jobs:
           asset_path: schema-loader/build/libs/scalardb-schema-loader-${{ steps.version.outputs.version }}.jar
           asset_name: scalardb-schema-loader-${{ steps.version.outputs.version }}.jar
           asset_content_type: application/java-archive
+
+      - name: Build scalardb-data-loader-cli jar
+        run: ./gradlew :data-loader:cli:shadowJar
+
+      - name: Upload scalardb-data-loader-cli jar
+        uses: actions/upload-release-asset@v1
+        env:
+          GITHUB_TOKEN: ${{ secrets.GH_PROJECT_ACCESS_TOKEN }}
+        with:
+          upload_url: ${{ steps.create_release.outputs.upload_url }}
+          asset_path: data-loader/cli/build/libs/scalardb-data-loader-cli-${{ steps.version.outputs.version }}.jar
+          asset_name: scalardb-data-loader-cli-${{ steps.version.outputs.version }}.jar
+          asset_content_type: application/java-archive

.github/workflows/release-snapshot.yaml

@@ -56,3 +56,4 @@ jobs:
         if: contains(steps.version.outputs.version, '-SNAPSHOT')
         run: |
           docker push ghcr.io/scalar-labs/scalardb-schema-loader:${{ steps.version.outputs.version }}
+          docker push ghcr.io/scalar-labs/scalardb-data-loader-cli:${{ steps.version.outputs.version }}

.github/workflows/remove-untagged-images.yaml

@@ -25,3 +25,9 @@ jobs:
         with:
           github-token: ${{ secrets.CR_PAT }}
           package-name: scalardb-schema-loader
+
+      - name: scalardb-data-loader-cli
+        uses: camargo/delete-untagged-action@v1
+        with:
+          github-token: ${{ secrets.CR_PAT }}
+          package-name: scalardb-data-loader-cli

.github/workflows/upload-artifacts.yaml

@@ -56,6 +56,7 @@ jobs:
       - name: Push containers
         run: |
           docker push ghcr.io/scalar-labs/scalardb-schema-loader:${{ steps.version.outputs.version }}
+          docker push ghcr.io/scalar-labs/scalardb-data-loader-cli:${{ steps.version.outputs.version }}
 
       - name: Upload scalardb, scalardb-schema-loader, scalardb-data-loader-core, and scalardb-integration-test to Maven Central Repository
         run: |

.github/workflows/vuln-check.yaml

@@ -25,7 +25,7 @@ jobs:
     with:
       target-ref: ${{ inputs.target-ref }}
       find-latest-release: ${{ inputs.find-latest-release }}
-      images: '[["ScalarDB Schema Loader", "scalardb-schema-loader"]]'
+      images: '[["ScalarDB Schema Loader", "scalardb-schema-loader"], ["ScalarDB Data Loader CLI", "scalardb-data-loader-cli"]]'
       version-command: "./gradlew :core:properties -q | grep version: | awk '{print $2}'"
     secrets:
       CR_PAT: ${{ secrets.CR_PAT }}

build.gradle

@@ -30,12 +30,12 @@ subprojects {
         awssdkVersion = '2.31.3'
         commonsDbcp2Version = '2.13.0'
         mysqlDriverVersion = '8.4.0'
-        postgresqlDriverVersion = '42.7.6'
+        postgresqlDriverVersion = '42.7.7'
         oracleDriverVersion = '23.8.0.25.04'
         sqlserverDriverVersion = '12.8.1.jre8'
         sqliteDriverVersion = '3.50.1.0'
         yugabyteDriverVersion = '42.7.3-yb-4'
-        db2DriverVersion= '12.1.0.0'
+        db2DriverVersion= '12.1.2.0'
         mariadDbDriverVersion = '3.5.3'
         picocliVersion = '4.7.7'
         commonsTextVersion = '1.13.1'

core/src/main/java/com/scalar/db/storage/jdbc/JdbcDatabase.java

@@ -117,6 +117,17 @@ public Scanner scan(Scan scan) throws ExecutionException {
       close(connection);
       throw new ExecutionException(
           CoreError.JDBC_ERROR_OCCURRED_IN_SELECTION.buildMessage(e.getMessage()), e);
+    } catch (Exception e) {
+      try {
+        if (connection != null) {
+          connection.rollback();
+        }
+      } catch (SQLException ex) {
+        e.addSuppressed(ex);
+      }
+
+      close(connection);
+      throw e;
     }
   }
 
@@ -186,6 +197,9 @@ public void mutate(List<? extends Mutation> mutations) throws ExecutionException
       close(connection);
       throw new ExecutionException(
           CoreError.JDBC_ERROR_OCCURRED_IN_MUTATION.buildMessage(e.getMessage()), e);
+    } catch (Exception e) {
+      close(connection);
+      throw e;
     }
 
     try {

core/src/main/java/com/scalar/db/transaction/consensuscommit/CommitHandler.java

@@ -36,6 +36,7 @@ public class CommitHandler {
   protected final Coordinator coordinator;
   private final TransactionTableMetadataManager tableMetadataManager;
   private final ParallelExecutor parallelExecutor;
+  protected final boolean coordinatorWriteOmissionOnReadOnlyEnabled;
 
   @LazyInit @Nullable private BeforePreparationSnapshotHook beforePreparationSnapshotHook;
 
@@ -44,11 +45,13 @@ public CommitHandler(
       DistributedStorage storage,
       Coordinator coordinator,
       TransactionTableMetadataManager tableMetadataManager,
-      ParallelExecutor parallelExecutor) {
+      ParallelExecutor parallelExecutor,
+      boolean coordinatorWriteOmissionOnReadOnlyEnabled) {
     this.storage = checkNotNull(storage);
     this.coordinator = checkNotNull(coordinator);
     this.tableMetadataManager = checkNotNull(tableMetadataManager);
     this.parallelExecutor = checkNotNull(parallelExecutor);
+    this.coordinatorWriteOmissionOnReadOnlyEnabled = coordinatorWriteOmissionOnReadOnlyEnabled;
   }
 
   /**
@@ -106,42 +109,62 @@ private void waitBeforePreparationSnapshotHookFuture(
     }
   }
 
-  public void commit(Snapshot snapshot) throws CommitException, UnknownTransactionStatusException {
+  public void commit(Snapshot snapshot, boolean readOnly)
+      throws CommitException, UnknownTransactionStatusException {
+    boolean hasWritesOrDeletesInSnapshot = !readOnly && snapshot.hasWritesOrDeletes();
+
     Optional<Future<Void>> snapshotHookFuture = invokeBeforePreparationSnapshotHook(snapshot);
-    try {
-      prepare(snapshot);
-    } catch (PreparationException e) {
-      safelyCallOnFailureBeforeCommit(snapshot);
-      abortState(snapshot.getId());
-      rollbackRecords(snapshot);
-      if (e instanceof PreparationConflictException) {
-        throw new CommitConflictException(e.getMessage(), e, e.getTransactionId().orElse(null));
+
+    if (hasWritesOrDeletesInSnapshot) {
+      try {
+        prepare(snapshot);
+      } catch (PreparationException e) {
+        safelyCallOnFailureBeforeCommit(snapshot);
+        abortState(snapshot.getId());
+        rollbackRecords(snapshot);
+        if (e instanceof PreparationConflictException) {
+          throw new CommitConflictException(e.getMessage(), e, e.getTransactionId().orElse(null));
+        }
+        throw new CommitException(e.getMessage(), e, e.getTransactionId().orElse(null));
+      } catch (Exception e) {
+        safelyCallOnFailureBeforeCommit(snapshot);
+        throw e;
       }
-      throw new CommitException(e.getMessage(), e, e.getTransactionId().orElse(null));
-    } catch (Exception e) {
-      safelyCallOnFailureBeforeCommit(snapshot);
-      throw e;
     }
 
-    try {
-      validate(snapshot);
-    } catch (ValidationException e) {
-      safelyCallOnFailureBeforeCommit(snapshot);
-      abortState(snapshot.getId());
-      rollbackRecords(snapshot);
-      if (e instanceof ValidationConflictException) {
-        throw new CommitConflictException(e.getMessage(), e, e.getTransactionId().orElse(null));
+    if (snapshot.hasReads()) {
+      try {
+        validate(snapshot);
+      } catch (ValidationException e) {
+        safelyCallOnFailureBeforeCommit(snapshot);
+
+        // If the transaction has no writes and deletes, we don't need to abort-state and
+        // rollback-records since there are no changes to be made.
+        if (hasWritesOrDeletesInSnapshot || !coordinatorWriteOmissionOnReadOnlyEnabled) {
+          abortState(snapshot.getId());
+        }
+        if (hasWritesOrDeletesInSnapshot) {
+          rollbackRecords(snapshot);
+        }
+
+        if (e instanceof ValidationConflictException) {
+          throw new CommitConflictException(e.getMessage(), e, e.getTransactionId().orElse(null));
+        }
+        throw new CommitException(e.getMessage(), e, e.getTransactionId().orElse(null));
+      } catch (Exception e) {
+        safelyCallOnFailureBeforeCommit(snapshot);
+        throw e;
       }
-      throw new CommitException(e.getMessage(), e, e.getTransactionId().orElse(null));
-    } catch (Exception e) {
-      safelyCallOnFailureBeforeCommit(snapshot);
-      throw e;
     }
 
     waitBeforePreparationSnapshotHookFuture(snapshot, snapshotHookFuture.orElse(null));
 
-    commitState(snapshot);
-    commitRecords(snapshot);
+    if (hasWritesOrDeletesInSnapshot || !coordinatorWriteOmissionOnReadOnlyEnabled) {
+      commitState(snapshot);
+    }
+    if (hasWritesOrDeletesInSnapshot) {
+      commitRecords(snapshot);
+    }
   }
 
   protected void handleCommitConflict(Snapshot snapshot, Exception cause)

core/src/main/java/com/scalar/db/transaction/consensuscommit/CommitHandlerWithGroupCommit.java

@@ -5,6 +5,7 @@
 import com.scalar.db.api.DistributedStorage;
 import com.scalar.db.api.TransactionState;
 import com.scalar.db.exception.transaction.CommitConflictException;
+import com.scalar.db.exception.transaction.CommitException;
 import com.scalar.db.exception.transaction.UnknownTransactionStatusException;
 import com.scalar.db.transaction.consensuscommit.Coordinator.State;
 import com.scalar.db.util.groupcommit.Emittable;
@@ -28,15 +29,31 @@ public CommitHandlerWithGroupCommit(
       Coordinator coordinator,
       TransactionTableMetadataManager tableMetadataManager,
       ParallelExecutor parallelExecutor,
+      boolean coordinatorWriteOmissionOnReadOnlyEnabled,
       CoordinatorGroupCommitter groupCommitter) {
-    super(storage, coordinator, tableMetadataManager, parallelExecutor);
+    super(
+        storage,
+        coordinator,
+        tableMetadataManager,
+        parallelExecutor,
+        coordinatorWriteOmissionOnReadOnlyEnabled);
 
     checkNotNull(groupCommitter);
     // The methods of this emitter will be called via GroupCommitter.ready().
     groupCommitter.setEmitter(new Emitter(coordinator));
     this.groupCommitter = groupCommitter;
   }
 
+  @Override
+  public void commit(Snapshot snapshot, boolean readOnly)
+      throws CommitException, UnknownTransactionStatusException {
+    if (!readOnly && !snapshot.hasWritesOrDeletes() && coordinatorWriteOmissionOnReadOnlyEnabled) {
+      cancelGroupCommitIfNeeded(snapshot.getId());
+    }
+
+    super.commit(snapshot, readOnly);
+  }
+
   @Override
   protected void onFailureBeforeCommit(Snapshot snapshot) {
     cancelGroupCommitIfNeeded(snapshot.getId());