diff --git a/core/src/integration-test/java/com/scalar/db/transaction/jdbc/JdbcTransactionIntegrationTest.java b/core/src/integration-test/java/com/scalar/db/transaction/jdbc/JdbcTransactionIntegrationTest.java
index 951ea6a57a..0ca3b064b8 100644
--- a/core/src/integration-test/java/com/scalar/db/transaction/jdbc/JdbcTransactionIntegrationTest.java
+++ b/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,31 @@ 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() {}
+
+  @Disabled("Implement later")
+  @Override
+  @Test
+  public void getScanner_ScanGivenForCommittedRecord_ShouldReturnRecords() {}
+
+  @Disabled("Implement later")
+  @Override
+  @Test
+  public void manager_getScanner_ScanGivenForCommittedRecord_ShouldReturnRecords() {}
 }
diff --git a/core/src/main/java/com/scalar/db/api/CrudOperable.java b/core/src/main/java/com/scalar/db/api/CrudOperable.java
index a773f3e93a..ecbf67dd28 100644
--- a/core/src/main/java/com/scalar/db/api/CrudOperable.java
+++ b/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,9 +29,18 @@ 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}
@@ -36,6 +48,18 @@ public interface CrudOperable<E extends TransactionException> {
    */
   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;
+  }
 }
diff --git a/core/src/main/java/com/scalar/db/api/Scanner.java b/core/src/main/java/com/scalar/db/api/Scanner.java
index 21a9b3a7cc..b863b53480 100644
--- a/core/src/main/java/com/scalar/db/api/Scanner.java
+++ b/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;
diff --git a/core/src/main/java/com/scalar/db/api/TransactionCrudOperable.java b/core/src/main/java/com/scalar/db/api/TransactionCrudOperable.java
index c8303f7a90..d2be32919a 100644
--- a/core/src/main/java/com/scalar/db/api/TransactionCrudOperable.java
+++ b/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;
+  }
 }
diff --git a/core/src/main/java/com/scalar/db/api/TransactionManagerCrudOperable.java b/core/src/main/java/com/scalar/db/api/TransactionManagerCrudOperable.java
index eb285a2d54..608d80cdf3 100644
--- a/core/src/main/java/com/scalar/db/api/TransactionManagerCrudOperable.java
+++ b/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;
+  }
 }
diff --git a/core/src/main/java/com/scalar/db/common/AbstractCrudOperableScanner.java b/core/src/main/java/com/scalar/db/common/AbstractCrudOperableScanner.java
new file mode 100644
index 0000000000..1e5a6fc6be
--- /dev/null
+++ b/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;
+    }
+  }
+}
diff --git a/core/src/main/java/com/scalar/db/common/AbstractTransactionCrudOperableScanner.java b/core/src/main/java/com/scalar/db/common/AbstractTransactionCrudOperableScanner.java
new file mode 100644
index 0000000000..34e1d54568
--- /dev/null
+++ b/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 {}
diff --git a/core/src/main/java/com/scalar/db/common/AbstractTransactionManagerCrudOperableScanner.java b/core/src/main/java/com/scalar/db/common/AbstractTransactionManagerCrudOperableScanner.java
new file mode 100644
index 0000000000..5dcbdb3479
--- /dev/null
+++ b/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 {}
diff --git a/core/src/main/java/com/scalar/db/common/ActiveTransactionManagedDistributedTransactionManager.java b/core/src/main/java/com/scalar/db/common/ActiveTransactionManagedDistributedTransactionManager.java
index e3ef02c635..ea592e5b41 100644
--- a/core/src/main/java/com/scalar/db/common/ActiveTransactionManagedDistributedTransactionManager.java
+++ b/core/src/main/java/com/scalar/db/common/ActiveTransactionManagedDistributedTransactionManager.java
@@ -121,6 +121,11 @@ public synchronized List<Result> scan(Scan scan) throws CrudException {
       return super.scan(scan);
     }
 
+    @Override
+    public synchronized Scanner getScanner(Scan scan) throws CrudException {
+      return super.getScanner(scan);
+    }
+
     /** @deprecated As of release 3.13.0. Will be removed in release 5.0.0. */
     @Deprecated
     @Override
diff --git a/core/src/main/java/com/scalar/db/common/ActiveTransactionManagedTwoPhaseCommitTransactionManager.java b/core/src/main/java/com/scalar/db/common/ActiveTransactionManagedTwoPhaseCommitTransactionManager.java
index a6a4c6b9ee..b0543433d3 100644
--- a/core/src/main/java/com/scalar/db/common/ActiveTransactionManagedTwoPhaseCommitTransactionManager.java
+++ b/core/src/main/java/com/scalar/db/common/ActiveTransactionManagedTwoPhaseCommitTransactionManager.java
@@ -127,6 +127,11 @@ public synchronized List<Result> scan(Scan scan) throws CrudException {
       return super.scan(scan);
     }
 
+    @Override
+    public synchronized Scanner getScanner(Scan scan) throws CrudException {
+      return super.getScanner(scan);
+    }
+
     /** @deprecated As of release 3.13.0. Will be removed in release 5.0.0. */
     @Deprecated
     @Override
diff --git a/core/src/main/java/com/scalar/db/common/DecoratedDistributedTransaction.java b/core/src/main/java/com/scalar/db/common/DecoratedDistributedTransaction.java
index ca6d0cdd76..ca997988c6 100644
--- a/core/src/main/java/com/scalar/db/common/DecoratedDistributedTransaction.java
+++ b/core/src/main/java/com/scalar/db/common/DecoratedDistributedTransaction.java
@@ -78,6 +78,11 @@ public List<Result> scan(Scan scan) throws CrudException {
     return transaction.scan(scan);
   }
 
+  @Override
+  public Scanner getScanner(Scan scan) throws CrudException {
+    return transaction.getScanner(scan);
+  }
+
   /** @deprecated As of release 3.13.0. Will be removed in release 5.0.0. */
   @Deprecated
   @Override
diff --git a/core/src/main/java/com/scalar/db/common/DecoratedDistributedTransactionManager.java b/core/src/main/java/com/scalar/db/common/DecoratedDistributedTransactionManager.java
index c2caadd9eb..dac3cfa2c7 100644
--- a/core/src/main/java/com/scalar/db/common/DecoratedDistributedTransactionManager.java
+++ b/core/src/main/java/com/scalar/db/common/DecoratedDistributedTransactionManager.java
@@ -157,6 +157,11 @@ public List<Result> scan(Scan scan) throws CrudException, UnknownTransactionStat
     return transactionManager.scan(scan);
   }
 
+  @Override
+  public Scanner getScanner(Scan scan) throws CrudException {
+    return transactionManager.getScanner(scan);
+  }
+
   /** @deprecated As of release 3.13.0. Will be removed in release 5.0.0. */
   @Deprecated
   @Override
diff --git a/core/src/main/java/com/scalar/db/common/DecoratedTwoPhaseCommitTransaction.java b/core/src/main/java/com/scalar/db/common/DecoratedTwoPhaseCommitTransaction.java
index 097e4f032c..04a48f4314 100644
--- a/core/src/main/java/com/scalar/db/common/DecoratedTwoPhaseCommitTransaction.java
+++ b/core/src/main/java/com/scalar/db/common/DecoratedTwoPhaseCommitTransaction.java
@@ -80,6 +80,11 @@ public List<Result> scan(Scan scan) throws CrudException {
     return transaction.scan(scan);
   }
 
+  @Override
+  public Scanner getScanner(Scan scan) throws CrudException {
+    return transaction.getScanner(scan);
+  }
+
   /** @deprecated As of release 3.13.0. Will be removed in release 5.0.0. */
   @Deprecated
   @Override
diff --git a/core/src/main/java/com/scalar/db/common/DecoratedTwoPhaseCommitTransactionManager.java b/core/src/main/java/com/scalar/db/common/DecoratedTwoPhaseCommitTransactionManager.java
index edcbd6e7a6..ce479795f1 100644
--- a/core/src/main/java/com/scalar/db/common/DecoratedTwoPhaseCommitTransactionManager.java
+++ b/core/src/main/java/com/scalar/db/common/DecoratedTwoPhaseCommitTransactionManager.java
@@ -111,6 +111,11 @@ public List<Result> scan(Scan scan) throws CrudException, UnknownTransactionStat
     return transactionManager.scan(scan);
   }
 
+  @Override
+  public Scanner getScanner(Scan scan) throws CrudException {
+    return transactionManager.getScanner(scan);
+  }
+
   /** @deprecated As of release 3.13.0. Will be removed in release 5.0.0. */
   @Deprecated
   @Override
diff --git a/core/src/main/java/com/scalar/db/common/StateManagedDistributedTransactionManager.java b/core/src/main/java/com/scalar/db/common/StateManagedDistributedTransactionManager.java
index 7d7adec72e..52866cb405 100644
--- a/core/src/main/java/com/scalar/db/common/StateManagedDistributedTransactionManager.java
+++ b/core/src/main/java/com/scalar/db/common/StateManagedDistributedTransactionManager.java
@@ -70,6 +70,12 @@ public List<Result> scan(Scan scan) throws CrudException {
       return super.scan(scan);
     }
 
+    @Override
+    public Scanner getScanner(Scan scan) throws CrudException {
+      checkIfActive();
+      return super.getScanner(scan);
+    }
+
     /** @deprecated As of release 3.13.0. Will be removed in release 5.0.0. */
     @Deprecated
     @Override
diff --git a/core/src/main/java/com/scalar/db/common/StateManagedTwoPhaseCommitTransactionManager.java b/core/src/main/java/com/scalar/db/common/StateManagedTwoPhaseCommitTransactionManager.java
index 7ba79ddede..1d79240d04 100644
--- a/core/src/main/java/com/scalar/db/common/StateManagedTwoPhaseCommitTransactionManager.java
+++ b/core/src/main/java/com/scalar/db/common/StateManagedTwoPhaseCommitTransactionManager.java
@@ -76,6 +76,12 @@ public List<Result> scan(Scan scan) throws CrudException {
       return super.scan(scan);
     }
 
+    @Override
+    public Scanner getScanner(Scan scan) throws CrudException {
+      checkIfActive();
+      return super.getScanner(scan);
+    }
+
     /** @deprecated As of release 3.13.0. Will be removed in release 5.0.0. */
     @Deprecated
     @Override
diff --git a/core/src/main/java/com/scalar/db/service/TransactionService.java b/core/src/main/java/com/scalar/db/service/TransactionService.java
index fd68d2dc22..8acc748eaa 100644
--- a/core/src/main/java/com/scalar/db/service/TransactionService.java
+++ b/core/src/main/java/com/scalar/db/service/TransactionService.java
@@ -167,11 +167,20 @@ public List<Result> scan(Scan scan) throws CrudException, UnknownTransactionStat
     return manager.scan(scan);
   }
 
+  @Override
+  public Scanner getScanner(Scan scan) throws CrudException {
+    return manager.getScanner(scan);
+  }
+
+  /** @deprecated As of release 3.13.0. Will be removed in release 5.0.0. */
+  @Deprecated
   @Override
   public void put(Put put) throws CrudException, UnknownTransactionStatusException {
     manager.put(put);
   }
 
+  /** @deprecated As of release 3.13.0. Will be removed in release 5.0.0. */
+  @Deprecated
   @Override
   public void put(List<Put> puts) throws CrudException, UnknownTransactionStatusException {
     manager.put(puts);
@@ -197,6 +206,8 @@ public void delete(Delete delete) throws CrudException, UnknownTransactionStatus
     manager.delete(delete);
   }
 
+  /** @deprecated As of release 3.13.0. Will be removed in release 5.0.0. */
+  @Deprecated
   @Override
   public void delete(List<Delete> deletes) throws CrudException, UnknownTransactionStatusException {
     manager.delete(deletes);
diff --git a/core/src/main/java/com/scalar/db/service/TwoPhaseCommitTransactionService.java b/core/src/main/java/com/scalar/db/service/TwoPhaseCommitTransactionService.java
index a2a2439f2c..6cf6f68a98 100644
--- a/core/src/main/java/com/scalar/db/service/TwoPhaseCommitTransactionService.java
+++ b/core/src/main/java/com/scalar/db/service/TwoPhaseCommitTransactionService.java
@@ -124,11 +124,20 @@ public List<Result> scan(Scan scan) throws CrudException, UnknownTransactionStat
     return manager.scan(scan);
   }
 
+  @Override
+  public Scanner getScanner(Scan scan) throws CrudException {
+    return manager.getScanner(scan);
+  }
+
+  /** @deprecated As of release 3.13.0. Will be removed in release 5.0.0. */
+  @Deprecated
   @Override
   public void put(Put put) throws CrudException, UnknownTransactionStatusException {
     manager.put(put);
   }
 
+  /** @deprecated As of release 3.13.0. Will be removed in release 5.0.0. */
+  @Deprecated
   @Override
   public void put(List<Put> puts) throws CrudException, UnknownTransactionStatusException {
     manager.put(puts);
@@ -154,6 +163,8 @@ public void delete(Delete delete) throws CrudException, UnknownTransactionStatus
     manager.delete(delete);
   }
 
+  /** @deprecated As of release 3.13.0. Will be removed in release 5.0.0. */
+  @Deprecated
   @Override
   public void delete(List<Delete> deletes) throws CrudException, UnknownTransactionStatusException {
     manager.delete(deletes);
diff --git a/core/src/main/java/com/scalar/db/transaction/consensuscommit/ConsensusCommit.java b/core/src/main/java/com/scalar/db/transaction/consensuscommit/ConsensusCommit.java
index b215ccc928..cd4c2fb4a6 100644
--- a/core/src/main/java/com/scalar/db/transaction/consensuscommit/ConsensusCommit.java
+++ b/core/src/main/java/com/scalar/db/transaction/consensuscommit/ConsensusCommit.java
@@ -96,6 +96,11 @@ public List<Result> scan(Scan scan) throws CrudException {
     }
   }
 
+  @Override
+  public Scanner getScanner(Scan scan) throws CrudException {
+    throw new UnsupportedOperationException("Implement later");
+  }
+
   /** @deprecated As of release 3.13.0. Will be removed in release 5.0.0. */
   @Deprecated
   @Override
diff --git a/core/src/main/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitManager.java b/core/src/main/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitManager.java
index 93e2056065..46a13035cb 100644
--- a/core/src/main/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitManager.java
+++ b/core/src/main/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitManager.java
@@ -229,6 +229,11 @@ public List<Result> scan(Scan scan) throws CrudException, UnknownTransactionStat
     return executeTransaction(t -> t.scan(copyAndSetTargetToIfNot(scan)));
   }
 
+  @Override
+  public Scanner getScanner(Scan scan) throws CrudException {
+    throw new UnsupportedOperationException("Implement later");
+  }
+
   @Deprecated
   @Override
   public void put(Put put) throws CrudException, UnknownTransactionStatusException {
diff --git a/core/src/main/java/com/scalar/db/transaction/consensuscommit/TwoPhaseConsensusCommit.java b/core/src/main/java/com/scalar/db/transaction/consensuscommit/TwoPhaseConsensusCommit.java
index fe25742dd7..d8edb63291 100644
--- a/core/src/main/java/com/scalar/db/transaction/consensuscommit/TwoPhaseConsensusCommit.java
+++ b/core/src/main/java/com/scalar/db/transaction/consensuscommit/TwoPhaseConsensusCommit.java
@@ -86,6 +86,11 @@ public List<Result> scan(Scan scan) throws CrudException {
     }
   }
 
+  @Override
+  public Scanner getScanner(Scan scan) throws CrudException {
+    throw new UnsupportedOperationException("Implement later");
+  }
+
   /** @deprecated As of release 3.13.0. Will be removed in release 5.0.0. */
   @Deprecated
   @Override
diff --git a/core/src/main/java/com/scalar/db/transaction/consensuscommit/TwoPhaseConsensusCommitManager.java b/core/src/main/java/com/scalar/db/transaction/consensuscommit/TwoPhaseConsensusCommitManager.java
index 5afc3e98df..3c9a47aa60 100644
--- a/core/src/main/java/com/scalar/db/transaction/consensuscommit/TwoPhaseConsensusCommitManager.java
+++ b/core/src/main/java/com/scalar/db/transaction/consensuscommit/TwoPhaseConsensusCommitManager.java
@@ -186,6 +186,11 @@ public List<Result> scan(Scan scan) throws CrudException, UnknownTransactionStat
     return executeTransaction(t -> t.scan(copyAndSetTargetToIfNot(scan)));
   }
 
+  @Override
+  public Scanner getScanner(Scan scan) throws CrudException {
+    throw new UnsupportedOperationException("Implement later");
+  }
+
   @Deprecated
   @Override
   public void put(Put put) throws CrudException, UnknownTransactionStatusException {
diff --git a/core/src/main/java/com/scalar/db/transaction/jdbc/JdbcTransaction.java b/core/src/main/java/com/scalar/db/transaction/jdbc/JdbcTransaction.java
index 27e2cdc7f8..a469c47003 100644
--- a/core/src/main/java/com/scalar/db/transaction/jdbc/JdbcTransaction.java
+++ b/core/src/main/java/com/scalar/db/transaction/jdbc/JdbcTransaction.java
@@ -94,6 +94,11 @@ public List<Result> scan(Scan scan) throws CrudException {
     }
   }
 
+  @Override
+  public Scanner getScanner(Scan scan) throws CrudException {
+    throw new UnsupportedOperationException("Implement later");
+  }
+
   /** @deprecated As of release 3.13.0. Will be removed in release 5.0.0. */
   @Deprecated
   @Override
diff --git a/core/src/main/java/com/scalar/db/transaction/jdbc/JdbcTransactionManager.java b/core/src/main/java/com/scalar/db/transaction/jdbc/JdbcTransactionManager.java
index 551725b264..df7f02197d 100644
--- a/core/src/main/java/com/scalar/db/transaction/jdbc/JdbcTransactionManager.java
+++ b/core/src/main/java/com/scalar/db/transaction/jdbc/JdbcTransactionManager.java
@@ -168,6 +168,11 @@ public List<Result> scan(Scan scan) throws CrudException, UnknownTransactionStat
     return executeTransaction(t -> t.scan(copyAndSetTargetToIfNot(scan)));
   }
 
+  @Override
+  public Scanner getScanner(Scan scan) throws CrudException {
+    throw new UnsupportedOperationException("Implement later");
+  }
+
   @Deprecated
   @Override
   public void put(Put put) throws CrudException, UnknownTransactionStatusException {
diff --git a/core/src/main/java/com/scalar/db/transaction/singlecrudoperation/SingleCrudOperationTransactionManager.java b/core/src/main/java/com/scalar/db/transaction/singlecrudoperation/SingleCrudOperationTransactionManager.java
index cb488cfa55..03857cac8a 100644
--- a/core/src/main/java/com/scalar/db/transaction/singlecrudoperation/SingleCrudOperationTransactionManager.java
+++ b/core/src/main/java/com/scalar/db/transaction/singlecrudoperation/SingleCrudOperationTransactionManager.java
@@ -20,7 +20,6 @@
 import com.scalar.db.api.PutIf;
 import com.scalar.db.api.Result;
 import com.scalar.db.api.Scan;
-import com.scalar.db.api.Scanner;
 import com.scalar.db.api.SerializableStrategy;
 import com.scalar.db.api.TransactionState;
 import com.scalar.db.api.Update;
@@ -156,13 +155,19 @@ public Optional<Result> get(Get get) throws CrudException {
   public List<Result> scan(Scan scan) throws CrudException {
     scan = copyAndSetTargetToIfNot(scan);
 
-    try (Scanner scanner = storage.scan(scan.withConsistency(Consistency.LINEARIZABLE))) {
+    try (com.scalar.db.api.Scanner scanner =
+        storage.scan(scan.withConsistency(Consistency.LINEARIZABLE))) {
       return scanner.all();
     } catch (ExecutionException | IOException e) {
       throw new CrudException(e.getMessage(), e, null);
     }
   }
 
+  @Override
+  public Scanner getScanner(Scan scan) throws CrudException {
+    throw new UnsupportedOperationException("Implement later");
+  }
+
   /** @deprecated As of release 3.13.0. Will be removed in release 5.0.0. */
   @Deprecated
   @Override
diff --git a/integration-test/src/main/java/com/scalar/db/api/DistributedTransactionIntegrationTestBase.java b/integration-test/src/main/java/com/scalar/db/api/DistributedTransactionIntegrationTestBase.java
index 7779090ece..5e3c9910e0 100644
--- a/integration-test/src/main/java/com/scalar/db/api/DistributedTransactionIntegrationTestBase.java
+++ b/integration-test/src/main/java/com/scalar/db/api/DistributedTransactionIntegrationTestBase.java
@@ -652,7 +652,7 @@ public void scan_ScanAllWithProjectionsGiven_ShouldRetrieveSpecifiedValues()
   }
 
   @Test
-  public void scanAll_ScanAllGivenForNonExisting_ShouldReturnEmpty() throws TransactionException {
+  public void scan_ScanAllGivenForNonExisting_ShouldReturnEmpty() throws TransactionException {
     // Arrange
     DistributedTransaction transaction = manager.start();
     ScanAll scanAll = prepareScanAll();
@@ -1093,6 +1093,30 @@ public void rollback_forOngoingTransaction_ShouldRollbackCorrectly() throws Tran
         results, Collections.singletonList(expectedResult));
   }
 
+  @Test
+  public void getScanner_ScanGivenForCommittedRecord_ShouldReturnRecords()
+      throws TransactionException {
+    // Arrange
+    populateRecords();
+    DistributedTransaction transaction = manager.start();
+    Scan scan = prepareScan(0, 0, 2);
+
+    // Act Assert
+    TransactionCrudOperable.Scanner scanner = transaction.getScanner(scan);
+
+    Optional<Result> result1 = scanner.one();
+    assertThat(result1).isPresent();
+    assertResult(0, 0, result1.get());
+
+    Optional<Result> result2 = scanner.one();
+    assertThat(result2).isPresent();
+    assertResult(0, 1, result2.get());
+
+    scanner.close();
+
+    transaction.commit();
+  }
+
   @Test
   public void resume_WithBeginningTransaction_ShouldReturnBegunTransaction()
       throws TransactionException {
@@ -2006,6 +2030,42 @@ public void manager_scan_ScanGivenForCommittedRecord_ShouldReturnRecords()
     assertThat(results.get(2).getInt(SOME_COLUMN)).isEqualTo(2);
   }
 
+  @Test
+  public void manager_getScanner_ScanGivenForCommittedRecord_ShouldReturnRecords()
+      throws TransactionException {
+    // Arrange
+    populateRecords();
+    Scan scan = prepareScan(1, 0, 2);
+
+    // Act Assert
+    TransactionManagerCrudOperable.Scanner scanner = manager.getScanner(scan);
+
+    Optional<Result> result1 = scanner.one();
+    assertThat(result1).isPresent();
+    assertThat(result1.get().getInt(ACCOUNT_ID)).isEqualTo(1);
+    assertThat(result1.get().getInt(ACCOUNT_TYPE)).isEqualTo(0);
+    assertThat(getBalance(result1.get())).isEqualTo(INITIAL_BALANCE);
+    assertThat(result1.get().getInt(SOME_COLUMN)).isEqualTo(0);
+
+    Optional<Result> result2 = scanner.one();
+    assertThat(result2).isPresent();
+    assertThat(result2.get().getInt(ACCOUNT_ID)).isEqualTo(1);
+    assertThat(result2.get().getInt(ACCOUNT_TYPE)).isEqualTo(1);
+    assertThat(getBalance(result2.get())).isEqualTo(INITIAL_BALANCE);
+    assertThat(result2.get().getInt(SOME_COLUMN)).isEqualTo(1);
+
+    Optional<Result> result3 = scanner.one();
+    assertThat(result3).isPresent();
+    assertThat(result3.get().getInt(ACCOUNT_ID)).isEqualTo(1);
+    assertThat(result3.get().getInt(ACCOUNT_TYPE)).isEqualTo(2);
+    assertThat(getBalance(result3.get())).isEqualTo(INITIAL_BALANCE);
+    assertThat(result3.get().getInt(SOME_COLUMN)).isEqualTo(2);
+
+    assertThat(scanner.one()).isNotPresent();
+
+    scanner.close();
+  }
+
   @Test
   public void manager_put_PutGivenForNonExisting_ShouldCreateRecord() throws TransactionException {
     // Arrange
diff --git a/integration-test/src/main/java/com/scalar/db/api/TwoPhaseCommitTransactionIntegrationTestBase.java b/integration-test/src/main/java/com/scalar/db/api/TwoPhaseCommitTransactionIntegrationTestBase.java
index 6587cd79a1..f6f7febcbe 100644
--- a/integration-test/src/main/java/com/scalar/db/api/TwoPhaseCommitTransactionIntegrationTestBase.java
+++ b/integration-test/src/main/java/com/scalar/db/api/TwoPhaseCommitTransactionIntegrationTestBase.java
@@ -1326,6 +1326,32 @@ public void scan_ScanAllGivenForNonExisting_ShouldReturnEmpty() throws Transacti
         results, Collections.singletonList(expectedResult));
   }
 
+  @Test
+  public void getScanner_ScanGivenForCommittedRecord_ShouldReturnRecords()
+      throws TransactionException {
+    // Arrange
+    populateRecords(manager1, namespace1, TABLE_1);
+    TwoPhaseCommitTransaction transaction = manager1.start();
+    Scan scan = prepareScan(0, 0, 2, namespace1, TABLE_1);
+
+    // Act Assert
+    TransactionCrudOperable.Scanner scanner = transaction.getScanner(scan);
+
+    Optional<Result> result1 = scanner.one();
+    assertThat(result1).isPresent();
+    assertResult(0, 0, result1.get());
+
+    Optional<Result> result2 = scanner.one();
+    assertThat(result2).isPresent();
+    assertResult(0, 1, result2.get());
+
+    scanner.close();
+
+    transaction.prepare();
+    transaction.validate();
+    transaction.commit();
+  }
+
   @Test
   public void resume_WithBeginningTransaction_ShouldReturnBegunTransaction()
       throws TransactionException {
@@ -2279,6 +2305,42 @@ public void manager_scan_ScanGivenForCommittedRecord_ShouldReturnRecords()
     assertThat(results.get(2).getInt(SOME_COLUMN)).isEqualTo(2);
   }
 
+  @Test
+  public void manager_getScanner_ScanGivenForCommittedRecord_ShouldReturnRecords()
+      throws TransactionException {
+    // Arrange
+    populateRecords(manager1, namespace1, TABLE_1);
+    Scan scan = prepareScan(1, 0, 2, namespace1, TABLE_1);
+
+    // Act Assert
+    TransactionManagerCrudOperable.Scanner scanner = manager1.getScanner(scan);
+
+    Optional<Result> result1 = scanner.one();
+    assertThat(result1).isPresent();
+    assertThat(result1.get().getInt(ACCOUNT_ID)).isEqualTo(1);
+    assertThat(result1.get().getInt(ACCOUNT_TYPE)).isEqualTo(0);
+    assertThat(getBalance(result1.get())).isEqualTo(INITIAL_BALANCE);
+    assertThat(result1.get().getInt(SOME_COLUMN)).isEqualTo(0);
+
+    Optional<Result> result2 = scanner.one();
+    assertThat(result2).isPresent();
+    assertThat(result2.get().getInt(ACCOUNT_ID)).isEqualTo(1);
+    assertThat(result2.get().getInt(ACCOUNT_TYPE)).isEqualTo(1);
+    assertThat(getBalance(result2.get())).isEqualTo(INITIAL_BALANCE);
+    assertThat(result2.get().getInt(SOME_COLUMN)).isEqualTo(1);
+
+    Optional<Result> result3 = scanner.one();
+    assertThat(result3).isPresent();
+    assertThat(result3.get().getInt(ACCOUNT_ID)).isEqualTo(1);
+    assertThat(result3.get().getInt(ACCOUNT_TYPE)).isEqualTo(2);
+    assertThat(getBalance(result3.get())).isEqualTo(INITIAL_BALANCE);
+    assertThat(result3.get().getInt(SOME_COLUMN)).isEqualTo(2);
+
+    assertThat(scanner.one()).isNotPresent();
+
+    scanner.close();
+  }
+
   @Test
   public void manager_put_PutGivenForNonExisting_ShouldCreateRecord() throws TransactionException {
     // Arrange
diff --git a/integration-test/src/main/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitIntegrationTestBase.java b/integration-test/src/main/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitIntegrationTestBase.java
index bf5abaae05..9854894ac4 100644
--- a/integration-test/src/main/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitIntegrationTestBase.java
+++ b/integration-test/src/main/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitIntegrationTestBase.java
@@ -15,6 +15,7 @@
 import com.scalar.db.io.Key;
 import java.util.Optional;
 import java.util.Properties;
+import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
 public abstract class ConsensusCommitIntegrationTestBase
@@ -929,4 +930,14 @@ public void deleteAndDelete_forSameRecord_shouldWorkCorrectly() throws Transacti
     Optional<Result> optResult = get(prepareGet(0, 0));
     assertThat(optResult).isNotPresent();
   }
+
+  @Disabled("Implement later")
+  @Override
+  @Test
+  public void getScanner_ScanGivenForCommittedRecord_ShouldReturnRecords() {}
+
+  @Disabled("Implement later")
+  @Override
+  @Test
+  public void manager_getScanner_ScanGivenForCommittedRecord_ShouldReturnRecords() {}
 }
diff --git a/integration-test/src/main/java/com/scalar/db/transaction/consensuscommit/TwoPhaseConsensusCommitIntegrationTestBase.java b/integration-test/src/main/java/com/scalar/db/transaction/consensuscommit/TwoPhaseConsensusCommitIntegrationTestBase.java
index 781087ed79..d57c31302c 100644
--- a/integration-test/src/main/java/com/scalar/db/transaction/consensuscommit/TwoPhaseConsensusCommitIntegrationTestBase.java
+++ b/integration-test/src/main/java/com/scalar/db/transaction/consensuscommit/TwoPhaseConsensusCommitIntegrationTestBase.java
@@ -2,6 +2,8 @@
 
 import com.scalar.db.api.TwoPhaseCommitTransactionIntegrationTestBase;
 import java.util.Properties;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
 
 public abstract class TwoPhaseConsensusCommitIntegrationTestBase
     extends TwoPhaseCommitTransactionIntegrationTestBase {
@@ -38,4 +40,14 @@ protected final Properties getProperties2(String testName) {
   protected Properties getProps2(String testName) {
     return getProps1(testName);
   }
+
+  @Disabled("Implement later")
+  @Override
+  @Test
+  public void getScanner_ScanGivenForCommittedRecord_ShouldReturnRecords() {}
+
+  @Disabled("Implement later")
+  @Override
+  @Test
+  public void manager_getScanner_ScanGivenForCommittedRecord_ShouldReturnRecords() {}
 }
diff --git a/integration-test/src/main/java/com/scalar/db/transaction/singlecrudoperation/SingleCrudOperationTransactionIntegrationTestBase.java b/integration-test/src/main/java/com/scalar/db/transaction/singlecrudoperation/SingleCrudOperationTransactionIntegrationTestBase.java
index d9c27237c1..6c50954e07 100644
--- a/integration-test/src/main/java/com/scalar/db/transaction/singlecrudoperation/SingleCrudOperationTransactionIntegrationTestBase.java
+++ b/integration-test/src/main/java/com/scalar/db/transaction/singlecrudoperation/SingleCrudOperationTransactionIntegrationTestBase.java
@@ -126,7 +126,12 @@ public void scan_ScanAllWithProjectionsGiven_ShouldRetrieveSpecifiedValues() {}
   @Disabled("Single CRUD operation transactions don't support beginning a transaction")
   @Override
   @Test
-  public void scanAll_ScanAllGivenForNonExisting_ShouldReturnEmpty() {}
+  public void scan_ScanAllGivenForNonExisting_ShouldReturnEmpty() {}
+
+  @Disabled("Single CRUD operation transactions don't support beginning a transaction")
+  @Override
+  @Test
+  public void getScanner_ScanGivenForCommittedRecord_ShouldReturnRecords() {}
 
   @Disabled("Single CRUD operation transactions don't support beginning a transaction")
   @Override
@@ -405,6 +410,11 @@ public void abort_forOngoingTransaction_ShouldAbortCorrectly() {}
   @Test
   public void rollback_forOngoingTransaction_ShouldRollbackCorrectly() {}
 
+  @Disabled("Implement later")
+  @Override
+  @Test
+  public void manager_getScanner_ScanGivenForCommittedRecord_ShouldReturnRecords() {}
+
   @Disabled(
       "Single CRUD operation transactions don't support executing multiple mutations in a transaction")
   @Override