Support SERIALIZABLE isolation for index-based operations with before-image index validation (#3463)

Author: brfrn169

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

@@ -83,11 +83,11 @@ protected Properties getProperties(String testName) {
 
   @Override
   @Disabled("Object Storage does not support index-related operations")
-  public void scan_ScanWithIndexGiven_WithSerializable_ShouldThrowIllegalArgumentException() {}
+  public void scan_ScanWithIndexGiven_WithSerializable_ShouldScan() {}
 
   @Override
   @Disabled("Object Storage does not support index-related operations")
-  public void get_GetWithIndexGiven_WithSerializable_ShouldThrowIllegalArgumentException() {}
+  public void get_GetWithIndexGiven_WithSerializable_ShouldGet() {}
 
   @Override
   @Disabled("Object Storage does not support index-related operations")
@@ -274,4 +274,29 @@ public void get_GetWithIndexForDeletedWhenCoordinatorStateAborted_ShouldRollBack
   public void
       getScanner_ScanAllWithIndexConditionForDeletedWhenCoordinatorStateAborted_ShouldRollBackAndReturnAllRecords(
           Isolation isolation) {}
+
+  @Override
+  @Disabled("Object Storage does not support index-related operations")
+  public void
+      commit_GetWithIndexInSerializable_WhenBeforeIndexHasPreparedRecordFromOtherTransaction_ShouldThrowCommitConflictException() {}
+
+  @Override
+  @Disabled("Object Storage does not support index-related operations")
+  public void
+      commit_ScanWithIndexInSerializable_WhenBeforeIndexHasPreparedRecordFromOtherTransaction_ShouldThrowCommitConflictException() {}
+
+  @Override
+  @Disabled("Object Storage does not support index-related operations")
+  public void
+      commit_ScanAllWithIndexConditionInSerializable_WhenBeforeIndexHasPreparedRecordFromOtherTransaction_ShouldThrowCommitConflictException() {}
+
+  @Override
+  @Disabled("Object Storage does not support index-related operations")
+  public void
+      commit_GetScannerWithIndexInSerializable_WhenBeforeIndexHasPreparedRecordFromOtherTransaction_ShouldThrowCommitConflictException() {}
+
+  @Override
+  @Disabled("Object Storage does not support index-related operations")
+  public void
+      commit_GetScannerWithScanAllIndexConditionInSerializable_WhenBeforeIndexHasPreparedRecordFromOtherTransaction_ShouldThrowCommitConflictException() {}
 }

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

@@ -914,19 +914,22 @@ public enum CoreError implements ScalarDbError {
   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",
+      "Get operations using a secondary index are not allowed in the SERIALIZABLE isolation level without before-image indexes. "
+          + "Run repairTable() to create before-image indexes for the table, which will enable index-based Get operations 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",
+      "Scan operations using a secondary index are not allowed in the SERIALIZABLE isolation level without before-image indexes. "
+          + "Run repairTable() to create before-image indexes for the table, which will enable index-based Scan operations 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",
+      "Conditions on indexed columns in cross-partition scan operations are not allowed in the SERIALIZABLE isolation level without before-image indexes. "
+          + "Run repairTable() to create before-image indexes for the table, which will enable conditions on indexed columns in cross-partition scan operations in the SERIALIZABLE isolation level",
       "",
       ""),
   OBJECT_STORAGE_CLOUD_STORAGE_SERVICE_ACCOUNT_KEY_NOT_FOUND(

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

@@ -96,7 +96,8 @@ public void check(Get get, TransactionContext context) throws ExecutionException
         // If the index column is part of the primary key, it's allowed
         String indexKeyColumnName = get.getPartitionKey().getColumns().get(0).getName();
         if (!tableMetadata.getPartitionKeyNames().contains(indexKeyColumnName)
-            && !tableMetadata.getClusteringKeyNames().contains(indexKeyColumnName)) {
+            && !tableMetadata.getClusteringKeyNames().contains(indexKeyColumnName)
+            && !metadata.hasBeforeImageSecondaryIndex(indexKeyColumnName)) {
           throw new IllegalArgumentException(
               CoreError.CONSENSUS_COMMIT_INDEX_GET_NOT_ALLOWED_IN_SERIALIZABLE.buildMessage());
         }
@@ -163,7 +164,8 @@ public void check(Scan scan, TransactionContext context) throws ExecutionExcepti
         // If the index column is part of the primary key, it's allowed
         String indexKeyColumnName = scan.getPartitionKey().getColumns().get(0).getName();
         if (!tableMetadata.getPartitionKeyNames().contains(indexKeyColumnName)
-            && !tableMetadata.getClusteringKeyNames().contains(indexKeyColumnName)) {
+            && !tableMetadata.getClusteringKeyNames().contains(indexKeyColumnName)
+            && !metadata.hasBeforeImageSecondaryIndex(indexKeyColumnName)) {
           throw new IllegalArgumentException(
               CoreError.CONSENSUS_COMMIT_INDEX_SCAN_NOT_ALLOWED_IN_SERIALIZABLE.buildMessage());
         }
@@ -177,7 +179,8 @@ public void check(Scan scan, TransactionContext context) throws ExecutionExcepti
             // If the column is an indexed column but is part of the primary key, it's allowed
             if (tableMetadata.getSecondaryIndexNames().contains(column)
                 && !tableMetadata.getPartitionKeyNames().contains(column)
-                && !tableMetadata.getClusteringKeyNames().contains(column)) {
+                && !tableMetadata.getClusteringKeyNames().contains(column)
+                && !metadata.hasBeforeImageSecondaryIndex(column)) {
               throw new IllegalArgumentException(
                   CoreError
                       .CONSENSUS_COMMIT_CONDITION_ON_INDEXED_COLUMNS_NOT_ALLOWED_IN_CROSS_PARTITION_SCAN_IN_SERIALIZABLE

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

@@ -612,8 +612,20 @@ private TransactionTableMetadata getTransactionTableMetadata(
   }
 
   /**
-   * Returns whether the given selection requires a before-image index check. This is true when the
-   * selection uses a secondary index that has a corresponding before-image secondary index.
+   * Returns whether the given selection requires a before-image index check.
+   *
+   * <p>For index-based selections (Get with index, Scan with index), this returns true when the
+   * index column has a corresponding before-image secondary index. For ScanAll, this returns true
+   * when any conjunction condition is on a column that has both a secondary index and a
+   * corresponding before-image secondary index.
+   *
+   * <p>If the before-image index does not exist (e.g., for tables created before the before-image
+   * index check feature was introduced), the check is skipped. In SNAPSHOT and READ_COMMITTED
+   * isolation, this means index-based reads may return eventually consistent results, which is a
+   * known limitation (a warning is logged at startup via {@code
+   * warnIfBeforeImageIndexesAreMissing}). In SERIALIZABLE isolation, this case does not occur
+   * because {@link ConsensusCommitOperationChecker} rejects index-based operations on tables
+   * without before-image indexes.
    *
    * @param selection the selection operation
    * @param metadata the transaction table metadata
@@ -626,11 +638,11 @@ boolean requiresBeforeIndexCheck(Selection selection, TransactionTableMetadata m
     }
 
     if (selection instanceof ScanAll) {
-      // For ScanAll, check if any conjunction condition is on a column that has a before-image
-      // secondary index
       for (Selection.Conjunction conjunction : selection.getConjunctions()) {
         for (ConditionalExpression condition : conjunction.getConditions()) {
-          if (metadata.hasBeforeImageSecondaryIndex(condition.getColumn().getName())) {
+          String columnName = condition.getColumn().getName();
+          if (metadata.getTableMetadata().getSecondaryIndexNames().contains(columnName)
+              && metadata.hasBeforeImageSecondaryIndex(columnName)) {
             return true;
           }
         }
@@ -710,6 +722,18 @@ boolean checkAndRecoverBeforeIndexRecords(
           }
         }
       }
+    } catch (RuntimeException e) {
+      Exception exception;
+      if (e.getCause() instanceof ExecutionException) {
+        exception = (ExecutionException) e.getCause();
+      } else {
+        exception = e;
+      }
+      throw new CrudException(
+          CoreError.CONSENSUS_COMMIT_SCANNING_RECORDS_FROM_STORAGE_FAILED.buildMessage(
+              exception.getMessage()),
+          exception,
+          context.transactionId);
     } catch (ExecutionException e) {
       throw new CrudException(
           CoreError.CONSENSUS_COMMIT_SCANNING_RECORDS_FROM_STORAGE_FAILED.buildMessage(

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

@@ -2,13 +2,13 @@
 
 import static com.scalar.db.transaction.consensuscommit.ConsensusCommitOperationAttributes.isImplicitPreReadEnabled;
 import static com.scalar.db.transaction.consensuscommit.ConsensusCommitOperationAttributes.isInsertModeEnabled;
-import static com.scalar.db.transaction.consensuscommit.ConsensusCommitUtils.getTransactionTableMetadata;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.MoreObjects;
 import com.google.common.collect.ComparisonChain;
 import com.google.common.collect.Iterators;
 import com.scalar.db.api.ConditionSetBuilder;
+import com.scalar.db.api.ConditionalExpression;
 import com.scalar.db.api.Delete;
 import com.scalar.db.api.DistributedStorage;
 import com.scalar.db.api.Get;
@@ -20,6 +20,7 @@
 import com.scalar.db.api.ScanAll;
 import com.scalar.db.api.ScanWithIndex;
 import com.scalar.db.api.Scanner;
+import com.scalar.db.api.Selection;
 import com.scalar.db.api.Selection.Conjunction;
 import com.scalar.db.api.TableMetadata;
 import com.scalar.db.common.CoreError;
@@ -560,22 +561,43 @@ void toSerializable(DistributedStorage storage)
 
     // Scan set is re-validated to check if there is no anti-dependency
     for (Map.Entry<Scan, LinkedHashMap<Key, TransactionResult>> entry : scanSet.entrySet()) {
-      tasks.add(() -> validateScanResults(storage, entry.getKey(), entry.getValue(), false));
+      tasks.add(
+          () -> {
+            TransactionTableMetadata txMetadata = getTransactionTableMetadata(entry.getKey());
+            validateScanResults(
+                storage, entry.getKey(), entry.getValue(), false, txMetadata.getTableMetadata());
+            validateBeforeIndex(storage, entry.getKey(), txMetadata);
+          });
     }
 
     // Scanner set is re-validated to check if there is no anti-dependency
     for (ScannerInfo scannerInfo : scannerSet) {
-      tasks.add(() -> validateScanResults(storage, scannerInfo.scan, scannerInfo.results, true));
+      tasks.add(
+          () -> {
+            TransactionTableMetadata txMetadata = getTransactionTableMetadata(scannerInfo.scan);
+            validateScanResults(
+                storage,
+                scannerInfo.scan,
+                scannerInfo.results,
+                true,
+                txMetadata.getTableMetadata());
+            validateBeforeIndex(storage, scannerInfo.scan, txMetadata);
+          });
     }
 
     // Get set is re-validated to check if there is no anti-dependency
     for (Map.Entry<Get, Optional<TransactionResult>> entry : getSet.entrySet()) {
       Get get = entry.getKey();
-      TableMetadata metadata = getTableMetadata(get);
+      TransactionTableMetadata txMetadata = getTransactionTableMetadata(get);
+      TableMetadata metadata = txMetadata.getTableMetadata();
 
       if (ScalarDbUtils.isSecondaryIndexSpecified(get, metadata)) {
         // For Get with index
-        tasks.add(() -> validateGetWithIndexResult(storage, get, entry.getValue(), metadata));
+        tasks.add(
+            () -> {
+              validateGetWithIndexResult(storage, get, entry.getValue(), metadata);
+              validateBeforeIndex(storage, get, txMetadata);
+            });
       } else {
         // For other Get
 
@@ -610,20 +632,19 @@ void toSerializable(DistributedStorage storage)
    * @param results the results of the scan
    * @param notFullyScannedScanner if this is a validation for a scanner that has not been fully
    *     scanned
+   * @param metadata the table metadata for the scanned table
    * @throws ExecutionException if a storage operation fails
    * @throws ValidationConflictException if the scan results are changed by another transaction
    */
   private void validateScanResults(
       DistributedStorage storage,
       Scan scan,
       LinkedHashMap<Key, TransactionResult> results,
-      boolean notFullyScannedScanner)
+      boolean notFullyScannedScanner,
+      TableMetadata metadata)
       throws ExecutionException, ValidationConflictException {
-    Scanner scanner = null;
-    try {
-      TableMetadata metadata = getTableMetadata(scan);
-
-      scanner = storage.scan(ConsensusCommitUtils.prepareScanForStorage(scan, metadata));
+    try (Scanner scanner =
+        storage.scan(ConsensusCommitUtils.prepareScanForStorage(scan, metadata))) {
 
       // Initialize the iterator for the latest scan results
       Optional<Result> latestResult = getNextResult(scanner, scan);
@@ -722,14 +743,8 @@ private void validateScanResults(
           throwExceptionDueToAntiDependency();
         }
       }
-    } finally {
-      if (scanner != null) {
-        try {
-          scanner.close();
-        } catch (IOException e) {
-          logger.warn("Failed to close the scanner. Transaction ID: {}", id, e);
-        }
-      }
+    } catch (IOException e) {
+      logger.warn("Failed to close the scanner. Transaction ID: {}", id, e);
     }
   }
 
@@ -780,7 +795,7 @@ private void validateGetWithIndexResult(
     originalResult.ifPresent(r -> results.put(new Snapshot.Key(scanWithIndex, r, metadata), r));
 
     // Validate the result to check if there is no anti-dependency
-    validateScanResults(storage, scanWithIndex, results, false);
+    validateScanResults(storage, scanWithIndex, results, false, metadata);
   }
 
   private void validateGetResult(
@@ -810,10 +825,99 @@ private void validateGetResult(
     }
   }
 
-  private TableMetadata getTableMetadata(Operation operation) throws ExecutionException {
-    TransactionTableMetadata transactionTableMetadata =
-        getTransactionTableMetadata(tableMetadataManager, operation);
-    return transactionTableMetadata.getTableMetadata();
+  /**
+   * Validates that there are no uncommitted records on the before-image index that could cause
+   * phantom reads.
+   *
+   * <p>This is needed because when another transaction updates a record's indexed column (e.g.,
+   * from 10 to 20) and is in PREPARED/DELETED state, the regular index scan (e.g., index_col=10)
+   * won't find that record since its current value is 20. However, the record's committed
+   * (before-image) value is still 10. Without this check, a phantom could go undetected: a record
+   * committed with index_col=10 but updated to 20 by another PREPARED transaction would be
+   * invisible to both the original scan and the validation re-scan.
+   *
+   * <p>This method is only called in the SERIALIZABLE extra-read validation phase. In SERIALIZABLE,
+   * {@link ConsensusCommitOperationChecker} rejects index-based operations on tables without
+   * before-image indexes, so the existence of before-image indexes is guaranteed when this method
+   * is called. Therefore, this method only needs to check whether the selection is an index-based
+   * operation (Get with index, Scan with index, or ScanAll with indexed column conditions), without
+   * checking for the existence of before-image indexes.
+   *
+   * @param storage a distributed storage
+   * @param selection the original selection operation (Get with index, ScanWithIndex, or ScanAll)
+   * @throws ExecutionException if a storage operation fails
+   * @throws ValidationConflictException if uncommitted records are found on the before-image index
+   */
+  private void validateBeforeIndex(
+      DistributedStorage storage, Selection selection, TransactionTableMetadata txMetadata)
+      throws ExecutionException, ValidationConflictException {
+    if (!isIndexBasedOperation(selection, txMetadata.getTableMetadata())) {
+      return;
+    }
+
+    Scan beforeIndexScan;
+    if (selection instanceof ScanAll) {
+      beforeIndexScan =
+          ConsensusCommitUtils.createBeforeIndexScanAll(
+              (ScanAll) selection, txMetadata.getTableMetadata());
+    } else {
+      beforeIndexScan = ConsensusCommitUtils.createBeforeIndexScan(selection);
+    }
+
+    try (Scanner scanner = storage.scan(beforeIndexScan)) {
+      for (Result result : scanner) {
+        TransactionResult txResult = new TransactionResult(result);
+        // Conservatively fail if any uncommitted record from another transaction is found on the
+        // before-image index. This may cause false positives (e.g., when the record will be
+        // rolled forward and its committed index value won't actually match the scan condition),
+        // but it guarantees correctness. On retry, the record should be committed, so the retry
+        // will succeed.
+        if (!txResult.isCommitted() && !id.equals(txResult.getId())) {
+          throwExceptionDueToAntiDependency();
+        }
+      }
+    } catch (RuntimeException e) {
+      if (e.getCause() instanceof ExecutionException) {
+        throw (ExecutionException) e.getCause();
+      }
+      throw e;
+    } catch (IOException e) {
+      logger.warn("Failed to close the scanner. Transaction ID: {}", id, e);
+    }
+  }
+
+  /**
+   * Checks if the given selection is an index-based operation that requires before-image index
+   * validation. This includes Get with index, Scan with index, and ScanAll with conditions on
+   * indexed columns.
+   *
+   * <p>For ScanAll, whether the underlying storage actually uses the index depends on the storage
+   * implementation. However, this method considers ScanAll with conditions on indexed columns as an
+   * index-based operation regardless.
+   *
+   * @param selection the selection operation to check
+   * @param metadata the table metadata
+   * @return true if the selection is an index-based operation
+   */
+  private boolean isIndexBasedOperation(Selection selection, TableMetadata metadata) {
+    if (ScalarDbUtils.isSecondaryIndexSpecified(selection, metadata)) {
+      return true;
+    }
+    if (selection instanceof ScanAll) {
+      for (Selection.Conjunction conjunction : selection.getConjunctions()) {
+        for (ConditionalExpression condition : conjunction.getConditions()) {
+          if (metadata.getSecondaryIndexNames().contains(condition.getColumn().getName())) {
+            return true;
+          }
+        }
+      }
+    }
+    return false;
+  }
+
+  private TransactionTableMetadata getTransactionTableMetadata(Operation operation)
+      throws ExecutionException {
+    return ConsensusCommitUtils.getTransactionTableMetadata(tableMetadataManager, operation);
   }
 
   private boolean isChanged(