class DefaultIndexer extends Indexer {

  @Override
  public void addIndexMutations(ColumnMutationCache mutationCache) throws IOException {
    for (Entry<byte[], Map<byte[], RowMutation>> tableMutation : mutationCache.getMutations()) {
      byte[] tableName = tableMutation.getKey();
      for (Entry<byte[], RowMutation> rowMutation : tableMutation.getValue().entrySet()) {
        byte[] row = rowMutation.getKey();
        for (ColumnMutation columnMuation : rowMutation.getValue().mutationList()) {
          if (columnMuation.getType().equals(Type.Put)) {
            IndexColumn indexColumn = new IndexColumn(tableName, columnMuation.getFamily(),
                columnMuation.getQualifier());
            if (columnIndexes.containsKey(indexColumn)) {
              String indexTableName = columnIndexes.get(indexColumn);
              KeyValue indexKv = constructIndexKv(row, columnMuation.getValue());
              mutationCache.addMutation(Bytes.toBytes(indexTableName), indexKv);
            }
          }
        }
      }
    }
  }

  public  DefaultIndexer(Configuration conf);

}