Fix secondary index behavior in Consensus Commit (#3133)

Author: brfrn169

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

@@ -684,6 +684,36 @@ public enum CoreError implements ScalarDbError {
       "The namespace has non-ScalarDB tables and cannot be dropped. Namespace: %s; Tables in the namespace: %s",
       "",
       ""),
+  CONSENSUS_COMMIT_SPECIFYING_TRANSACTION_METADATA_COLUMNS_IN_PROJECTION_NOT_ALLOWED(
+      Category.USER_ERROR,
+      "0258",
+      "Specifying transaction metadata columns in the projection is not allowed. Table: %s; Column: %s",
+      "",
+      ""),
+  CONSENSUS_COMMIT_SPECIFYING_TRANSACTION_METADATA_COLUMNS_IN_ORDERING_NOT_ALLOWED(
+      Category.USER_ERROR,
+      "0259",
+      "Specifying transaction metadata columns in the ordering is not allowed. Table: %s; Column: %s",
+      "",
+      ""),
+  CONSENSUS_COMMIT_INDEX_GET_NOT_ALLOWED_IN_SERIALIZABLE(
+      Category.USER_ERROR,
+      "0260",
+      "Get operations by using an index is not allowed in the SERIALIZABLE isolation level",
+      "",
+      ""),
+  CONSENSUS_COMMIT_INDEX_SCAN_NOT_ALLOWED_IN_SERIALIZABLE(
+      Category.USER_ERROR,
+      "0261",
+      "Scan operations by using an index is not allowed in the SERIALIZABLE isolation level",
+      "",
+      ""),
+  CONSENSUS_COMMIT_CONDITION_ON_INDEXED_COLUMNS_NOT_ALLOWED_IN_CROSS_PARTITION_SCAN_IN_SERIALIZABLE(
+      Category.USER_ERROR,
+      "0262",
+      "Conditions on indexed columns in cross-partition scan operations are not allowed in the SERIALIZABLE isolation level",
+      "",
+      ""),
 
   //
   // Errors for the concurrency error category

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

@@ -50,20 +50,20 @@ public class ConsensusCommit extends AbstractDistributedTransaction {
   private final TransactionContext context;
   private final CrudHandler crud;
   private final CommitHandler commit;
-  private final ConsensusCommitMutationOperationChecker mutationOperationChecker;
+  private final ConsensusCommitOperationChecker operationChecker;
   @Nullable private final CoordinatorGroupCommitter groupCommitter;
 
   @SuppressFBWarnings("EI_EXPOSE_REP2")
   public ConsensusCommit(
       TransactionContext context,
       CrudHandler crud,
       CommitHandler commit,
-      ConsensusCommitMutationOperationChecker mutationOperationChecker,
+      ConsensusCommitOperationChecker operationChecker,
       @Nullable CoordinatorGroupCommitter groupCommitter) {
     this.context = checkNotNull(context);
     this.crud = checkNotNull(crud);
     this.commit = checkNotNull(commit);
-    this.mutationOperationChecker = mutationOperationChecker;
+    this.operationChecker = operationChecker;
     this.groupCommitter = groupCommitter;
   }
 
@@ -74,17 +74,40 @@ public String getId() {
 
   @Override
   public Optional<Result> get(Get get) throws CrudException {
-    return crud.get(copyAndSetTargetToIfNot(get), context);
+    get = copyAndSetTargetToIfNot(get);
+
+    try {
+      operationChecker.check(get, context);
+    } catch (ExecutionException e) {
+      throw new CrudException(e.getMessage(), e, getId());
+    }
+
+    return crud.get(get, context);
   }
 
   @Override
   public List<Result> scan(Scan scan) throws CrudException {
-    return crud.scan(copyAndSetTargetToIfNot(scan), context);
+    scan = copyAndSetTargetToIfNot(scan);
+
+    try {
+      operationChecker.check(scan, context);
+    } catch (ExecutionException e) {
+      throw new CrudException(e.getMessage(), e, getId());
+    }
+
+    return crud.scan(scan, context);
   }
 
   @Override
   public Scanner getScanner(Scan scan) throws CrudException {
     scan = copyAndSetTargetToIfNot(scan);
+
+    try {
+      operationChecker.check(scan, context);
+    } catch (ExecutionException e) {
+      throw new CrudException(e.getMessage(), e, getId());
+    }
+
     return crud.getScanner(scan, context);
   }
 
@@ -245,7 +268,7 @@ void waitForRecoveryCompletion() throws CrudException {
 
   private void checkMutation(Mutation mutation) throws CrudException {
     try {
-      mutationOperationChecker.check(mutation);
+      operationChecker.check(mutation);
     } catch (ExecutionException e) {
       throw new CrudException(e.getMessage(), e, getId());
     }

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

@@ -56,7 +56,7 @@ public class ConsensusCommitManager extends AbstractDistributedTransactionManage
   private final CrudHandler crud;
   protected final CommitHandler commit;
   private final Isolation isolation;
-  private final ConsensusCommitMutationOperationChecker mutationOperationChecker;
+  private final ConsensusCommitOperationChecker operationChecker;
   @Nullable private final CoordinatorGroupCommitter groupCommitter;
 
   @SuppressFBWarnings("EI_EXPOSE_REP2")
@@ -84,7 +84,9 @@ public ConsensusCommitManager(
             parallelExecutor);
     commit = createCommitHandler(config);
     isolation = config.getIsolation();
-    mutationOperationChecker = new ConsensusCommitMutationOperationChecker(tableMetadataManager);
+    operationChecker =
+        new ConsensusCommitOperationChecker(
+            tableMetadataManager, config.isIncludeMetadataEnabled());
   }
 
   protected ConsensusCommitManager(DatabaseConfig databaseConfig) {
@@ -111,14 +113,17 @@ protected ConsensusCommitManager(DatabaseConfig databaseConfig) {
             parallelExecutor);
     commit = createCommitHandler(config);
     isolation = config.getIsolation();
-    mutationOperationChecker = new ConsensusCommitMutationOperationChecker(tableMetadataManager);
+    operationChecker =
+        new ConsensusCommitOperationChecker(
+            tableMetadataManager, config.isIncludeMetadataEnabled());
   }
 
   @SuppressFBWarnings("EI_EXPOSE_REP2")
   @VisibleForTesting
   ConsensusCommitManager(
       DistributedStorage storage,
       DistributedStorageAdmin admin,
+      ConsensusCommitConfig config,
       DatabaseConfig databaseConfig,
       Coordinator coordinator,
       ParallelExecutor parallelExecutor,
@@ -140,8 +145,9 @@ protected ConsensusCommitManager(DatabaseConfig databaseConfig) {
     this.commit = commit;
     this.groupCommitter = groupCommitter;
     this.isolation = isolation;
-    this.mutationOperationChecker =
-        new ConsensusCommitMutationOperationChecker(tableMetadataManager);
+    this.operationChecker =
+        new ConsensusCommitOperationChecker(
+            tableMetadataManager, config.isIncludeMetadataEnabled());
   }
 
   // `groupCommitter` must be set before calling this method.
@@ -259,7 +265,7 @@ DistributedTransaction begin(
     TransactionContext context =
         new TransactionContext(txId, snapshot, isolation, readOnly, oneOperation);
     DistributedTransaction transaction =
-        new ConsensusCommit(context, crud, commit, mutationOperationChecker, groupCommitter);
+        new ConsensusCommit(context, crud, commit, operationChecker, groupCommitter);
     if (readOnly) {
       transaction = new ReadOnlyDistributedTransaction(transaction);
     }