code: private Iterator<Record> readBatch(final DeltaPlacement placement, final Collection<Key> keys, final ReadConsistency consistency) {
          checkNotNull(keys, "keys");
  
          // Convert the keys to ByteBuffer Cassandra row keys
          List<Map.Entry<ByteBuffer, Key>> rowKeys = Lists.newArrayListWithCapacity(keys.size());
          for (Key key : keys) {
              AstyanaxTable table = (AstyanaxTable) key.getTable();
              AstyanaxStorage storage = table.getReadStorage();
              rowKeys.add(Maps.immutableEntry(storage.getRowKey(key.getKey()), key));
          }
  
          // Sort the keys by their byte array encoding to get some locality w/queries.
          Collections.sort(rowKeys, Ordering.natural().onResultOf(entry -> entry.getKey()));
  
          // Group them into batches.  Cassandra may have to seek each row so prefer smaller batches.
          List<List<Map.Entry<ByteBuffer, Key>>> batches = Lists.partition(rowKeys, _driverConfig.getMaxRandomRowsBatchSize());
  
          // This algorithm is arranged such that rows are return in pages with size _fetchSize.  The rows are grouped
          // into row groups by common row key.  The first RECORD_CACHE_SIZE rows are cached for the row group
          // and any remaining rows are cached using soft references.  This places an upper bound on the memory
          // requirements needed while iterating.  If at any time a soft reference is lost C* is re-queried to
          // fetch the missing columns.
  
          return Iterators.concat(Iterators.transform(batches.iterator(),
                  rowKeySubset -> {
                      Timer.Context timerCtx = _readBatchTimer.time();
                      try {
                          return rowQuery(rowKeySubset, consistency, placement);
                      } finally {
                          timerCtx.stop();
                      }
                  }));
      }
docs: Read a batch of keys that all belong to the same placement (ColumnFamily).