code: @Override
      public void purge(AstyanaxStorage storage, Runnable progress) {
          DeltaPlacement placement = (DeltaPlacement) storage.getPlacement();
          CassandraKeyspace keyspace = placement.getKeyspace();
  
          // Scan all the shards and delete all the rows we find.
          MutationBatch mutation = keyspace.prepareMutationBatch(SorConsistencies.toAstyanax(WriteConsistency.STRONG));
          Iterator<String> keyIter = _keyScanner.scanKeys(storage, ReadConsistency.STRONG);
          while (keyIter.hasNext()) {
              ByteBuffer rowKey = storage.getRowKey(keyIter.next());
              mutation.withRow(placement.getDeltaColumnFamily(), rowKey).delete();
              mutation.withRow(placement.getBlockedDeltaColumnFamily(), rowKey).delete();
              if (mutation.getRowCount() >= 100) {
                  progress.run();
                  execute(mutation, "purge %d records from placement %s", mutation.getRowCount(), placement.getName());
                  mutation.discardMutations();
              }
          }
          if (!mutation.isEmpty()) {
              progress.run();
              execute(mutation, "purge %d records from placement %s", mutation.getRowCount(), placement.getName());
          }
      }
docs: DataPurgeDAO