docs: @Override
      public RowQuery<K, C> getKey(final K rowKey) {
          return new AbstractRowQueryImpl<K, C>(columnFamily.getColumnSerializer()) {
              private boolean firstPage = true;
  
              @Override
              public ColumnQuery<C> getColumn(final C column) {
                  return new ColumnQuery<C>() {
                      @Override
                      public OperationResult<Column<C>> execute() throws ConnectionException {
                          return connectionPool.executeWithFailover(new AbstractKeyspaceOperationImpl<Column<C>>(
                                  tracerFactory.newTracer(CassandraOperationType.GET_COLUMN, columnFamily), pinnedHost,
                                  keyspace.getKeyspaceName()) {
                              @Override
                              public Column<C> internalExecute(Client client, ConnectionContext context) throws Exception {
                                  ColumnOrSuperColumn cosc = client.get(
                                          columnFamily.getKeySerializer().toByteBuffer(rowKey),
                                          new org.apache.cassandra.thrift.ColumnPath().setColumn_family(
                                                  columnFamily.getName()).setColumn(
                                                  columnFamily.getColumnSerializer().toByteBuffer(column)),
                                          ThriftConverter.ToThriftConsistencyLevel(consistencyLevel));
                                  if (cosc.isSetColumn()) {
                                      org.apache.cassandra.thrift.Column c = cosc.getColumn();
                                      return new ThriftColumnImpl<C>(columnFamily.getColumnSerializer().fromBytes(
                                              c.getName()), c);
                                  }
                                  else if (cosc.isSetSuper_column()) {
                                      // TODO: Super columns
                                      // should be deprecated
                                      SuperColumn sc = cosc.getSuper_column();
                                      return new ThriftSuperColumnImpl<C>(columnFamily.getColumnSerializer().fromBytes(
                                              sc.getName()), sc);
                                  }
                                  else if (cosc.isSetCounter_column()) {
                                      org.apache.cassandra.thrift.CounterColumn c = cosc.getCounter_column();
                                      return new ThriftCounterColumnImpl<C>(columnFamily.getColumnSerializer().fromBytes(
                                              c.getName()), c);
                                  }
                                  else if (cosc.isSetCounter_super_column()) {
                                      // TODO: Super columns
                                      // should be deprecated
                                      CounterSuperColumn sc = cosc.getCounter_super_column();
                                      return new ThriftCounterSuperColumnImpl<C>(columnFamily.getColumnSerializer()
                                              .fromBytes(sc.getName()), sc);
                                  }
                                  else {
                                      throw new RuntimeException("Unknown column type in response");
                                  }
                              }
  
                              @Override
                              public ByteBuffer getRowKey() {
                                  return columnFamily.getKeySerializer().toByteBuffer(rowKey);
                              }
                          }, retry);
                      }
  
                      @Override
                      public ListenableFuture<OperationResult<Column<C>>> executeAsync() throws ConnectionException {
                          return executor.submit(new Callable<OperationResult<Column<C>>>() {
                              @Override
                              public OperationResult<Column<C>> call() throws Exception {
                                  return execute();
                              }
                          });
                      }
                  };
              }
  
              @Override
              public OperationResult<ColumnList<C>> execute() throws ConnectionException {
                  return connectionPool.executeWithFailover(
                          new AbstractKeyspaceOperationImpl<ColumnList<C>>(tracerFactory.newTracer(
                                  CassandraOperationType.GET_ROW, columnFamily), pinnedHost, keyspace.getKeyspaceName()) {
  
                              @Override
                              public ColumnList<C> execute(Client client, ConnectionContext context) throws ConnectionException {
                                  if (isPaginating && paginateNoMore) {
                                      return new EmptyColumnList<C>();
                                  }
  
                                  return super.execute(client, context);
                              }
  
                              @Override
                              public ColumnList<C> internalExecute(Client client, ConnectionContext context) throws Exception {
                                  List<ColumnOrSuperColumn> columnList = client.get_slice(columnFamily.getKeySerializer()
                                          .toByteBuffer(rowKey), new ColumnParent().setColumn_family(columnFamily
                                          .getName()), predicate, ThriftConverter
                                          .ToThriftConsistencyLevel(consistencyLevel));
  
                                  // Special handling for pagination
                                  if (isPaginating && predicate.isSetSlice_range()) {
                                      // Did we reach the end of the query.
                                      if (columnList.size() != predicate.getSlice_range().getCount()) {
                                          paginateNoMore = true;
                                      }
  
                                      // If this is the first page then adjust the
                                      // count so we fetch one extra column
                                      // that will later be dropped
                                      if (firstPage) {
                                          firstPage = false;
                                          if (predicate.getSlice_range().getCount() != Integer.MAX_VALUE)
                                              predicate.getSlice_range().setCount(predicate.getSlice_range().getCount() + 1);
                                      }
                                      else {
                                          if (!columnList.isEmpty())
                                              columnList.remove(0);
                                      }
  
                                      // Set the start column for the next page to
                                      // the last column of this page.
                                      // We will discard this column later.
                                      if (!columnList.isEmpty()) {
                                          ColumnOrSuperColumn last = Iterables.getLast(columnList);
                                          if (last.isSetColumn()) {
                                              predicate.getSlice_range().setStart(last.getColumn().getName());
                                          } else if (last.isSetCounter_column()) {
                                              predicate.getSlice_range().setStart(last.getCounter_column().getName());
                                          } else if (last.isSetSuper_column()) {
                                              // TODO: Super columns
                                              // should be deprecated
                                              predicate.getSlice_range().setStart(last.getSuper_column().getName());
                                          } else if (last.isSetCounter_super_column()) {
                                              // TODO: Super columns
                                              // should be deprecated
                                              predicate.getSlice_range().setStart(last.getCounter_super_column().getName());
                                          }
                                      }
                                  }
                                  ColumnList<C> result = new ThriftColumnOrSuperColumnListImpl<C>(columnList,
                                          columnFamily.getColumnSerializer());
                                  return result;
                              }
  
                              @Override
                              public ByteBuffer getRowKey() {
                                  return columnFamily.getKeySerializer().toByteBuffer(rowKey);
                              }
                          }, retry);
              }
  
              @Override
              public ColumnCountQuery getCount() {
                  return new ColumnCountQuery() {
                      @Override
                      public OperationResult<Integer> execute() throws ConnectionException {
                          return connectionPool.executeWithFailover(new AbstractKeyspaceOperationImpl<Integer>(
                                  tracerFactory.newTracer(CassandraOperationType.GET_COLUMN_COUNT, columnFamily),
                                  pinnedHost, keyspace.getKeyspaceName()) {
                              @Override
                              public Integer internalExecute(Client client, ConnectionContext context) throws Exception {
                                  return client.get_count(columnFamily.getKeySerializer().toByteBuffer(rowKey),
                                          new ColumnParent().setColumn_family(columnFamily.getName()), predicate,
                                          ThriftConverter.ToThriftConsistencyLevel(consistencyLevel));
                              }
  
                              @Override
                              public ByteBuffer getRowKey() {
                                  return columnFamily.getKeySerializer().toByteBuffer(rowKey);
                              }
                          }, retry);
                      }
  
                      @Override
                      public ListenableFuture<OperationResult<Integer>> executeAsync() throws ConnectionException {
                          return executor.submit(new Callable<OperationResult<Integer>>() {
                              @Override
                              public OperationResult<Integer> call() throws Exception {
                                  return execute();
                              }
                          });
                      }
                  };
              }
  
              @Override
              public ListenableFuture<OperationResult<ColumnList<C>>> executeAsync() throws ConnectionException {
                  return executor.submit(new Callable<OperationResult<ColumnList<C>>>() {
                      @Override
                      public OperationResult<ColumnList<C>> call() throws Exception {
                          return execute();
                      }
                  });
              }
  
              @Override
              public RowCopier<K, C> copyTo(final ColumnFamily<K, C> otherColumnFamily, final K otherRowKey) {
                  return new RowCopier<K, C>() {
                      private boolean useOriginalTimestamp = true;
  
                      @Override
                      public OperationResult<Void> execute() throws ConnectionException {
                          return connectionPool.executeWithFailover(
                                  new AbstractKeyspaceOperationImpl<Void>(tracerFactory.newTracer(
                                          CassandraOperationType.COPY_TO, columnFamily), pinnedHost, keyspace
                                          .getKeyspaceName()) {
                                      @Override
                                      public Void internalExecute(Client client, ConnectionContext context) throws Exception {
  
                                          long currentTime = keyspace.getConfig().getClock().getCurrentTime();
  
                                          List<ColumnOrSuperColumn> columnList = client.get_slice(columnFamily
                                                  .getKeySerializer().toByteBuffer(rowKey), new ColumnParent()
                                                  .setColumn_family(columnFamily.getName()), predicate, ThriftConverter
                                                  .ToThriftConsistencyLevel(consistencyLevel));
  
                                          // Create mutation list from columns in
                                          // the response
                                          List<Mutation> mutationList = new ArrayList<Mutation>();
                                          for (ColumnOrSuperColumn sosc : columnList) {
                                              ColumnOrSuperColumn cosc;
  
                                              if (sosc.isSetColumn()) {
                                                  cosc = new ColumnOrSuperColumn().setColumn(sosc.getColumn());
                                                  if (!useOriginalTimestamp)
                                                      cosc.getColumn().setTimestamp(currentTime);
                                              }
                                              else if (sosc.isSetSuper_column()) {
                                                  cosc = new ColumnOrSuperColumn().setSuper_column(sosc.getSuper_column());
                                                  if (!useOriginalTimestamp) {
                                                      for (org.apache.cassandra.thrift.Column subColumn : sosc.getSuper_column().getColumns()) {
                                                          subColumn.setTimestamp(currentTime);
                                                          subColumn.setTimestamp(currentTime);
                                                      }
                                                  }
                                              }
                                              else if (sosc.isSetCounter_column()) {
                                                  cosc = new ColumnOrSuperColumn().setCounter_column(sosc.getCounter_column());
                                              }
                                              else if (sosc.isSetCounter_super_column()) {
                                                  cosc = new ColumnOrSuperColumn().setCounter_super_column(sosc.getCounter_super_column());
                                              }
                                              else {
                                                  continue;
                                              }
  
                                              mutationList.add(new Mutation().setColumn_or_supercolumn(cosc));
                                          }
  
                                          // Create mutation map
                                          Map<ByteBuffer, Map<String, List<Mutation>>> mutationMap = new HashMap<ByteBuffer, Map<String, List<Mutation>>>();
                                          HashMap<String, List<Mutation>> cfmap = new HashMap<String, List<Mutation>>();
                                          cfmap.put(otherColumnFamily.getName(), mutationList);
                                          mutationMap.put(columnFamily.getKeySerializer().toByteBuffer(otherRowKey),
                                                  cfmap);
  
                                          // Execute the mutation
                                          client.batch_mutate(mutationMap,
                                                  ThriftConverter.ToThriftConsistencyLevel(consistencyLevel));
                                          return null;
                                      }
                                  }, retry);
                      }
  
                      @Override
                      public ListenableFuture<OperationResult<Void>> executeAsync() throws ConnectionException {
                          return executor.submit(new Callable<OperationResult<Void>>() {
                              @Override
                              public OperationResult<Void> call() throws Exception {
                                  return execute();
                              }
                          });
                      }
  
                      @Override
                      public RowCopier<K, C> withOriginalTimestamp(boolean useOriginalTimestamp) {
                          this.useOriginalTimestamp = useOriginalTimestamp;
                          return this;
                      }
                  };
              }
          };
      }