code: private Iterator<Record> rowQuery(final List<Map.Entry<ByteBuffer, Key>> rowKeys, final ReadConsistency consistency,
                                        final DeltaPlacement placement) {
          List<ByteBuffer> keys = Lists.newArrayListWithCapacity(rowKeys.size());
          final Map<ByteBuffer, Key> rawKeyMap = Maps.newHashMap();
          for (Map.Entry<ByteBuffer, Key> entry : rowKeys) {
              keys.add(entry.getKey());
              rawKeyMap.put(entry.getKey(), entry.getValue());
          }
  
          BlockedDeltaTableDDL tableDDL = placement.getBlockedDeltaTableDDL();
  
          Statement statement = selectDeltaFrom(tableDDL)
                  .where(in(tableDDL.getRowKeyColumnName(), keys))
                  .setConsistencyLevel(SorConsistencies.toCql(consistency));
  
          Iterator<Iterable<Row>> rowGroups = deltaQueryAsync(placement, statement, false, "Failed to read records %s", rawKeyMap.values());
  
          return Iterators.concat(
                  // First iterator reads the row groups found and transforms them to Records
                  Iterators.transform(rowGroups, rows -> {
                      ByteBuffer keyBytes = getRawKeyFromRowGroup(rows);
                      Key key = rawKeyMap.remove(keyBytes);
                      assert key != null : "Query returned row with a key out of bound";
                      return newRecordFromCql(key, rows, placement, ByteBufferUtil.bytesToHex(keyBytes));
                  }),
                  // Second iterator returns an empty Record for each key queried but not found.
                  new AbstractIterator<Record>() {
                      private Iterator<Key> _nonExistentKeyIterator;
  
                      @Override
                      protected Record computeNext() {
                          // Lazily return an empty record for each key not found in the previous iterator.
                          // rawKeyMap.iterator() must not be called until the first iterator is completely spent.
                          if (_nonExistentKeyIterator == null) {
                              _nonExistentKeyIterator = rawKeyMap.values().iterator();
                          }
                          return _nonExistentKeyIterator.hasNext() ?
                                  emptyRecord(_nonExistentKeyIterator.next()) :
                                  endOfData();
                      }
                  });
      }
docs: Returns an iterator for the Records keyed by the provided row keys.  An empty record is returned for any
  key which does not have a corresponding row in C*.