Add scanner API to transaction abstraction

core/src/integration-test/java/com/scalar/db/storage/cassandra/ConsensusCommitCassandraEnv.java

@@ -1,7 +1,7 @@
 package com.scalar.db.storage.cassandra;
 
-import com.scalar.db.common.ConsensusCommitTestUtils;
 import com.scalar.db.transaction.consensuscommit.ConsensusCommitIntegrationTestUtils;
+import com.scalar.db.transaction.consensuscommit.ConsensusCommitTestUtils;
 import java.util.Properties;
 
 public final class ConsensusCommitCassandraEnv {

core/src/integration-test/java/com/scalar/db/storage/cosmos/ConsensusCommitCosmosEnv.java

@@ -1,7 +1,7 @@
 package com.scalar.db.storage.cosmos;
 
-import com.scalar.db.common.ConsensusCommitTestUtils;
 import com.scalar.db.transaction.consensuscommit.ConsensusCommitIntegrationTestUtils;
+import com.scalar.db.transaction.consensuscommit.ConsensusCommitTestUtils;
 import java.util.Map;
 import java.util.Properties;
 

core/src/integration-test/java/com/scalar/db/storage/dynamo/ConsensusCommitDynamoEnv.java

@@ -1,7 +1,7 @@
 package com.scalar.db.storage.dynamo;
 
-import com.scalar.db.common.ConsensusCommitTestUtils;
 import com.scalar.db.transaction.consensuscommit.ConsensusCommitIntegrationTestUtils;
+import com.scalar.db.transaction.consensuscommit.ConsensusCommitTestUtils;
 import java.util.Map;
 import java.util.Properties;
 

core/src/integration-test/java/com/scalar/db/storage/jdbc/ConsensusCommitJdbcEnv.java

@@ -1,7 +1,7 @@
 package com.scalar.db.storage.jdbc;
 
-import com.scalar.db.common.ConsensusCommitTestUtils;
 import com.scalar.db.transaction.consensuscommit.ConsensusCommitIntegrationTestUtils;
+import com.scalar.db.transaction.consensuscommit.ConsensusCommitTestUtils;
 import java.util.Properties;
 
 public final class ConsensusCommitJdbcEnv {

core/src/integration-test/java/com/scalar/db/storage/multistorage/ConsensusCommitNullMetadataIntegrationTestWithMultiStorage.java

@@ -1,9 +1,9 @@
 package com.scalar.db.storage.multistorage;
 
-import com.scalar.db.common.ConsensusCommitTestUtils;
 import com.scalar.db.config.DatabaseConfig;
 import com.scalar.db.storage.cassandra.CassandraAdmin;
 import com.scalar.db.transaction.consensuscommit.ConsensusCommitNullMetadataIntegrationTestBase;
+import com.scalar.db.transaction.consensuscommit.ConsensusCommitTestUtils;
 import com.scalar.db.transaction.consensuscommit.Coordinator;
 import java.util.Collections;
 import java.util.Map;

core/src/integration-test/java/com/scalar/db/storage/multistorage/ConsensusCommitSpecificIntegrationTestWithMultiStorage.java

@@ -1,9 +1,9 @@
 package com.scalar.db.storage.multistorage;
 
-import com.scalar.db.common.ConsensusCommitTestUtils;
 import com.scalar.db.config.DatabaseConfig;
 import com.scalar.db.storage.cassandra.CassandraAdmin;
 import com.scalar.db.transaction.consensuscommit.ConsensusCommitSpecificIntegrationTestBase;
+import com.scalar.db.transaction.consensuscommit.ConsensusCommitTestUtils;
 import com.scalar.db.transaction.consensuscommit.Coordinator;
 import java.util.Collections;
 import java.util.Map;

core/src/integration-test/java/com/scalar/db/transaction/jdbc/JdbcTransactionIntegrationTest.java

@@ -6,6 +6,7 @@
 import com.scalar.db.storage.jdbc.JdbcEnv;
 import java.util.Properties;
 import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
 
 public class JdbcTransactionIntegrationTest extends DistributedTransactionIntegrationTestBase {
 
@@ -24,17 +25,21 @@ protected Properties getProperties(String testName) {
 
   @Disabled("JDBC transactions don't support getState()")
   @Override
+  @Test
   public void getState_forSuccessfulTransaction_ShouldReturnCommittedState() {}
 
   @Disabled("JDBC transactions don't support getState()")
   @Override
+  @Test
   public void getState_forFailedTransaction_ShouldReturnAbortedState() {}
 
   @Disabled("JDBC transactions don't support abort()")
   @Override
+  @Test
   public void abort_forOngoingTransaction_ShouldAbortCorrectly() {}
 
   @Disabled("JDBC transactions don't support rollback()")
   @Override
+  @Test
   public void rollback_forOngoingTransaction_ShouldRollbackCorrectly() {}
 }

core/src/main/java/com/scalar/db/api/CrudOperable.java

@@ -12,6 +12,9 @@
  * An interface for transactional CRUD operations. Note that the LINEARIZABLE consistency level is
  * always used in transactional CRUD operations, so {@link Consistency} specified for CRUD
  * operations is ignored.
+ *
+ * @param <E> the type of {@link TransactionException} that the implementation throws if the
+ *     operation fails
  */
 public interface CrudOperable<E extends TransactionException> {
 
@@ -26,16 +29,37 @@ public interface CrudOperable<E extends TransactionException> {
   Optional<Result> get(Get get) throws E;
 
   /**
-   * Retrieves results from the storage through a transaction with the specified {@link Scan}
-   * command with a partition key and returns a list of {@link Result}. Results can be filtered by
-   * specifying a range of clustering keys.
+   * Retrieves results from the storage through a transaction with the specified {@link Scan} or
+   * {@link ScanAll} or {@link ScanWithIndex} command with a partition key and returns a list of
+   * {@link Result}. Results can be filtered by specifying a range of clustering keys.
+   *
+   * <ul>
+   *   <li>{@link Scan} : by specifying a partition key, it will return results within the
+   *       partition. Results can be filtered by specifying a range of clustering keys.
+   *   <li>{@link ScanAll} : for a given table, it will return all its records even if they span
+   *       several partitions.
+   *   <li>{@link ScanWithIndex} : by specifying an index key, it will return results within the
+   *       index.
+   * </ul>
    *
    * @param scan a {@code Scan} command
    * @return a list of {@link Result}
    * @throws E if the transaction CRUD operation fails
    */
   List<Result> scan(Scan scan) throws E;
 
+  /**
+   * Retrieves results from the storage through a transaction with the specified {@link Scan} or
+   * {@link ScanAll} or {@link ScanWithIndex} command with a partition key and returns a {@link
+   * Scanner} to iterate over the results. Results can be filtered by specifying a range of
+   * clustering keys.
+   *
+   * @param scan a {@code Scan} command
+   * @return a {@code Scanner} to iterate over the results
+   * @throws E if the transaction CRUD operation fails
+   */
+  Scanner<E> getScanner(Scan scan) throws E;
+
   /**
    * Inserts an entry into or updates an entry in the underlying storage through a transaction with
    * the specified {@link Put} command. If a condition is specified in the {@link Put} command, and
@@ -131,4 +155,32 @@ public interface CrudOperable<E extends TransactionException> {
    * @throws E if the transaction CRUD operation fails
    */
   void mutate(List<? extends Mutation> mutations) throws E;
+
+  /** A scanner abstraction for iterating results. */
+  interface Scanner<E extends TransactionException> extends AutoCloseable, Iterable<Result> {
+    /**
+     * Returns the next result.
+     *
+     * @return an {@code Optional} containing the next result if available, or empty if no more
+     *     results
+     * @throws E if the operation fails
+     */
+    Optional<Result> one() throws E;
+
+    /**
+     * Returns all remaining results.
+     *
+     * @return a {@code List} containing all remaining results
+     * @throws E if the operation fails
+     */
+    List<Result> all() throws E;
+
+    /**
+     * Closes the scanner.
+     *
+     * @throws E if closing the scanner fails
+     */
+    @Override
+    void close() throws E;
+  }
 }

core/src/main/java/com/scalar/db/api/Scanner.java

@@ -13,17 +13,18 @@
 public interface Scanner extends Closeable, Iterable<Result> {
 
   /**
-   * Returns the first result in the results.
+   * Returns the next result.
    *
-   * @return the first result in the results
+   * @return an {@code Optional} containing the next result if available, or empty if no more
+   *     results
    * @throws ExecutionException if the operation fails
    */
   Optional<Result> one() throws ExecutionException;
 
   /**
-   * Returns all the results.
+   * Returns all remaining results.
    *
-   * @return the list of {@code Result}s
+   * @return a {@code List} containing all remaining results
    * @throws ExecutionException if the operation fails
    */
   List<Result> all() throws ExecutionException;

core/src/main/java/com/scalar/db/api/TransactionCrudOperable.java

@@ -33,6 +33,18 @@ public interface TransactionCrudOperable extends CrudOperable<CrudException> {
   @Override
   List<Result> scan(Scan scan) throws CrudConflictException, CrudException;
 
+  /**
+   * {@inheritDoc}
+   *
+   * @throws CrudConflictException if the transaction CRUD operation fails due to transient faults
+   *     (e.g., a conflict error). You can retry the transaction from the beginning
+   * @throws CrudException if the transaction CRUD operation fails due to transient or nontransient
+   *     faults. You can try retrying the transaction from the beginning, but the transaction may
+   *     still fail if the cause is nontranient
+   */
+  @Override
+  Scanner getScanner(Scan scan) throws CrudConflictException, CrudException;
+
   /**
    * {@inheritDoc}
    *
@@ -154,4 +166,38 @@ void delete(List<Delete> deletes)
   @Override
   void mutate(List<? extends Mutation> mutations)
       throws CrudConflictException, CrudException, UnsatisfiedConditionException;
+
+  interface Scanner extends CrudOperable.Scanner<CrudException> {
+    /**
+     * {@inheritDoc}
+     *
+     * @throws CrudConflictException if the transaction CRUD operation fails due to transient faults
+     *     (e.g., a conflict error). You can retry the transaction from the beginning
+     * @throws CrudException if the transaction CRUD operation fails due to transient or
+     *     nontransient faults. You can try retrying the transaction from the beginning, but the
+     *     transaction may still fail if the cause is nontranient
+     */
+    @Override
+    Optional<Result> one() throws CrudConflictException, CrudException;
+
+    /**
+     * {@inheritDoc}
+     *
+     * @throws CrudConflictException if the transaction CRUD operation fails due to transient faults
+     *     (e.g., a conflict error). You can retry the transaction from the beginning
+     * @throws CrudException if the transaction CRUD operation fails due to transient or
+     *     nontransient faults. You can try retrying the transaction from the beginning, but the
+     *     transaction may still fail if the cause is nontranient
+     */
+    @Override
+    List<Result> all() throws CrudConflictException, CrudException;
+
+    /**
+     * {@inheritDoc}
+     *
+     * @throws CrudException if closing the scanner fails
+     */
+    @Override
+    void close() throws CrudException;
+  }
 }

core/src/main/java/com/scalar/db/api/TransactionManagerCrudOperable.java

@@ -39,6 +39,18 @@ Optional<Result> get(Get get)
   List<Result> scan(Scan scan)
       throws CrudConflictException, CrudException, UnknownTransactionStatusException;
 
+  /**
+   * {@inheritDoc}
+   *
+   * @throws CrudConflictException if the transaction CRUD operation fails due to transient faults
+   *     (e.g., a conflict error). You can retry the transaction from the beginning
+   * @throws CrudException if the transaction CRUD operation fails due to transient or nontransient
+   *     faults. You can try retrying the transaction from the beginning, but the transaction may
+   *     still fail if the cause is nontranient
+   */
+  @Override
+  Scanner getScanner(Scan scan) throws CrudConflictException, CrudException;
+
   /**
    * {@inheritDoc}
    *
@@ -177,4 +189,39 @@ void delete(List<Delete> deletes)
   void mutate(List<? extends Mutation> mutations)
       throws CrudConflictException, CrudException, UnsatisfiedConditionException,
           UnknownTransactionStatusException;
+
+  interface Scanner extends CrudOperable.Scanner<TransactionException> {
+    /**
+     * {@inheritDoc}
+     *
+     * @throws CrudConflictException if the transaction CRUD operation fails due to transient faults
+     *     (e.g., a conflict error). You can retry the transaction from the beginning
+     * @throws CrudException if the transaction CRUD operation fails due to transient or
+     *     nontransient faults. You can try retrying the transaction from the beginning, but the
+     *     transaction may still fail if the cause is nontranient
+     */
+    @Override
+    Optional<Result> one() throws CrudConflictException, CrudException;
+
+    /**
+     * {@inheritDoc}
+     *
+     * @throws CrudConflictException if the transaction CRUD operation fails due to transient faults
+     *     (e.g., a conflict error). You can retry the transaction from the beginning
+     * @throws CrudException if the transaction CRUD operation fails due to transient or
+     *     nontransient faults. You can try retrying the transaction from the beginning, but the
+     *     transaction may still fail if the cause is nontranient
+     */
+    @Override
+    List<Result> all() throws CrudConflictException, CrudException;
+
+    /**
+     * {@inheritDoc}
+     *
+     * @throws CrudException if closing the scanner fails
+     * @throws UnknownTransactionStatusException if the status of the commit is unknown
+     */
+    @Override
+    void close() throws CrudException, UnknownTransactionStatusException;
+  }
 }

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

@@ -0,0 +1,61 @@
+package com.scalar.db.common;
+
+import com.google.errorprone.annotations.concurrent.LazyInit;
+import com.scalar.db.api.CrudOperable;
+import com.scalar.db.api.Result;
+import com.scalar.db.exception.transaction.TransactionException;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+import java.util.Objects;
+import javax.annotation.Nonnull;
+import javax.annotation.concurrent.NotThreadSafe;
+
+public abstract class AbstractCrudOperableScanner<E extends TransactionException>
+    implements CrudOperable.Scanner<E> {
+
+  @LazyInit private ScannerIterator scannerIterator;
+
+  @Override
+  @Nonnull
+  public Iterator<Result> iterator() {
+    if (scannerIterator == null) {
+      scannerIterator = new ScannerIterator(this);
+    }
+    return scannerIterator;
+  }
+
+  @NotThreadSafe
+  public class ScannerIterator implements Iterator<Result> {
+
+    private final CrudOperable.Scanner<E> scanner;
+    private Result next;
+
+    public ScannerIterator(CrudOperable.Scanner<E> scanner) {
+      this.scanner = Objects.requireNonNull(scanner);
+    }
+
+    @Override
+    public boolean hasNext() {
+      if (next != null) {
+        return true;
+      }
+
+      try {
+        return (next = scanner.one().orElse(null)) != null;
+      } catch (TransactionException e) {
+        throw new RuntimeException(e.getMessage(), e);
+      }
+    }
+
+    @Override
+    public Result next() {
+      if (!hasNext()) {
+        throw new NoSuchElementException();
+      }
+
+      Result ret = next;
+      next = null;
+      return ret;
+    }
+  }
+}

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

@@ -0,0 +1,7 @@
+package com.scalar.db.common;
+
+import com.scalar.db.api.TransactionCrudOperable;
+import com.scalar.db.exception.transaction.CrudException;
+
+public abstract class AbstractTransactionCrudOperableScanner
+    extends AbstractCrudOperableScanner<CrudException> implements TransactionCrudOperable.Scanner {}

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

@@ -0,0 +1,8 @@
+package com.scalar.db.common;
+
+import com.scalar.db.api.TransactionManagerCrudOperable;
+import com.scalar.db.exception.transaction.TransactionException;
+
+public abstract class AbstractTransactionManagerCrudOperableScanner
+    extends AbstractCrudOperableScanner<TransactionException>
+    implements TransactionManagerCrudOperable.Scanner {}