import org.apache.cassandra.db.IColumn; //导入依赖的package包/类
public void map(ByteBuffer key, SortedMap<ByteBuffer, IColumn> columns, Context context) throws IOException, InterruptedException
{
    for (IColumn column : columns.values())
    {
        String name  = ByteBufferUtil.string(column.name());
        String value = null;
        
        if (name.contains("int"))
            value = String.valueOf(ByteBufferUtil.toInt(column.value()));
        else
            value = ByteBufferUtil.string(column.value());
                       
        logger.debug("read {}:{}={} from {}",
                     new Object[] {ByteBufferUtil.string(key), name, value, context.getInputSplit()});

        StringTokenizer itr = new StringTokenizer(value);
        while (itr.hasMoreTokens())
        {
            word.set(itr.nextToken());
            context.write(word, one);
        }
    }
}
 

import org.apache.cassandra.db.IColumn; //导入依赖的package包/类
public void add(IColumn c)
{
    if (c.isMarkedForDelete())
        return;

    ByteBuffer[] current = composite.split(c.name());

    if (currentGroup == null)
    {
        currentGroup = new ColumnGroupMap(current);
        currentGroup.add(current, idx, c);
        previous = current;
        return;
    }

    if (!isSameGroup(current))
    {
        groups.add(currentGroup);
        currentGroup = new ColumnGroupMap(current);
    }
    currentGroup.add(current, idx, c);
    previous = current;
}
 

import org.apache.cassandra.db.IColumn; //导入依赖的package包/类
protected Pair<ByteBuffer, SortedMap<ByteBuffer, IColumn>> computeNext()
{
    maybeInit();
    if (rows == null)
        return endOfData();

    totalRead++;
    KeySlice ks = rows.get(i++);
    SortedMap<ByteBuffer, IColumn> map = new TreeMap<ByteBuffer, IColumn>(comparator);
    for (ColumnOrSuperColumn cosc : ks.columns)
    {
        IColumn column = unthriftify(cosc);
        map.put(column.name(), column);
    }
    return Pair.create(ks.key, map);
}
 

import org.apache.cassandra.db.IColumn; //导入依赖的package包/类
public Entry<ByteBuffer, IColumn> next()
{
    if (!hasNext())
    {
        throw new IllegalStateException("end of column iterator");
    }

    final IColumn column = deserializeNext();
    return new Entry<ByteBuffer, IColumn>()
    {
        public IColumn setValue(IColumn value)
        {
            throw new UnsupportedOperationException();
        }

        public IColumn getValue()
        {
            return column;
        }

        public ByteBuffer getKey()
        {
            return column.name();
        }
    };
}
 

import org.apache.cassandra.db.IColumn; //导入依赖的package包/类
public void map(ByteBuffer key, SortedMap<ByteBuffer, IColumn> columns, Context context) 
throws IOException, InterruptedException {   
                        
      // Fetch columns
      IColumn column = columns.get(sourceColumn);

      if (column == null)
          return;

      String value = ByteBufferUtil.string(column.value()).toLowerCase();

      // We replace punctuations with empty string
      for (String pattern : punctuations) {
      	value = value.replaceAll(pattern, "");
      }            

      StringTokenizer itr = new StringTokenizer(value);

      while (itr.hasMoreTokens()) {
          word.set(itr.nextToken());
          context.write(word, one);
      }
  }
 

import org.apache.cassandra.db.IColumn; //导入依赖的package包/类
protected Pair<ByteBuffer, SortedMap<ByteBuffer, IColumn>> computeNext()
{
    while (true)
    {
        if (columns.hasNext())
        {
            ColumnOrSuperColumn cosc = columns.next();
            IColumn column = unthriftify(cosc);
            ImmutableSortedMap<ByteBuffer, IColumn> map = ImmutableSortedMap.of(column.name(), column);
            return Pair.<ByteBuffer, SortedMap<ByteBuffer, IColumn>>create(currentRow.key, map);
        }

        if (!rows.hasNext())
            return endOfData();

        nextRow();
    }
}
 

import org.apache.cassandra.db.IColumn; //导入依赖的package包/类
private void add(ByteBuffer[] fullName, int idx, IColumn column)
{
    ByteBuffer columnName = fullName[idx];
    if (fullName.length == idx + 2)
    {
        // It's a collection
        Value v = map.get(columnName);
        if (v == null)
        {
            v = new Collection();
            map.put(columnName, v);
        }
        assert v instanceof Collection;

        ((Collection)v).add(Pair.create(fullName[idx + 1], column));
    }
    else
    {
        assert !map.containsKey(columnName);
        map.put(columnName, new Simple(column));
    }
}
 

import org.apache.cassandra.db.IColumn; //导入依赖的package包/类
protected List<Pair<ByteBuffer, IColumn>> enforceLimit(List<Pair<ByteBuffer, IColumn>> columns)
{
    if (columns.size() <= MAX_ELEMENTS)
        return columns;

    logger.error("Detected collection with {} elements, more than the {} limit. Only the first {} elements will be returned to the client. "
               + "Please see http://cassandra.apache.org/doc/cql3/CQL.html#collections for more details.", columns.size(), MAX_ELEMENTS, MAX_ELEMENTS);
    return columns.subList(0, MAX_ELEMENTS);
}
 

import org.apache.cassandra.db.IColumn; //导入依赖的package包/类
public IColumn getSimple(ByteBuffer key)
{
    Value v = map.get(key);
    if (v == null)
        return null;

    assert v instanceof Simple;
    return ((Simple)v).column;
}
 

import org.apache.cassandra.db.IColumn; //导入依赖的package包/类
public List<Pair<ByteBuffer, IColumn>> getCollection(ByteBuffer key)
{
    Value v = map.get(key);
    if (v == null)
        return null;

    assert v instanceof Collection;
    return (List<Pair<ByteBuffer, IColumn>>)v;
}
 

import org.apache.cassandra.db.IColumn; //导入依赖的package包/类
private IColumn unthriftifySuper(SuperColumn super_column)
{
    org.apache.cassandra.db.SuperColumn sc = new org.apache.cassandra.db.SuperColumn(super_column.name, subComparator);
    for (Column column : super_column.columns)
    {
        sc.addColumn(unthriftifySimple(column));
    }
    return sc;
}
 

import org.apache.cassandra.db.IColumn; //导入依赖的package包/类
private IColumn unthriftifySuperCounter(CounterSuperColumn superColumn)
{
    org.apache.cassandra.db.SuperColumn sc = new org.apache.cassandra.db.SuperColumn(superColumn.name, subComparator);
    for (CounterColumn column : superColumn.columns)
        sc.addColumn(unthriftifyCounter(column));
    return sc;
}
 

import org.apache.cassandra.db.IColumn; //导入依赖的package包/类
/**
 * Increases the row counter only if we really moved to the next row.
 * @param next just fetched row slice
 */
private void maybeIncreaseRowCounter(Pair<ByteBuffer, SortedMap<ByteBuffer, IColumn>> next)
{
    ByteBuffer currentKey = next.left;
    if (!currentKey.equals(lastCountedKey))
    {
        totalRead++;
        lastCountedKey = currentKey;
    }
}
 

import org.apache.cassandra.db.IColumn; //导入依赖的package包/类
public static ColumnFamily createCF(long mfda, int ldt, IColumn... cols)
{
    ColumnFamily cf = ColumnFamily.create(TABLENAME, CFNAME);
    cf.delete(new DeletionInfo(mfda, ldt));
    for (IColumn col : cols)
        cf.addColumn(col);
    return cf;
}
 

import org.apache.cassandra.db.IColumn; //导入依赖的package包/类
public boolean next(ByteBuffer key, SortedMap<ByteBuffer, IColumn> value) throws IOException
{
    if (this.nextKeyValue())
    {
        key.clear();
        key.put(this.getCurrentKey().duplicate());
        key.flip();

        value.clear();
        value.putAll(this.getCurrentValue());

        return true;
    }
    return false;
}
 

import org.apache.cassandra.db.IColumn; //导入依赖的package包/类
/** convert a column to a tuple */
protected Tuple columnToTuple(IColumn col, CfDef cfDef, AbstractType comparator) throws IOException
{
    Tuple pair = TupleFactory.getInstance().newTuple(2);

    // name
    if(comparator instanceof AbstractCompositeType)
        setTupleValue(pair, 0, composeComposite((AbstractCompositeType)comparator,col.name()));
    else
        setTupleValue(pair, 0, comparator.compose(col.name()));

    // value
    if (col instanceof Column)
    {
        // standard
        Map<ByteBuffer,AbstractType> validators = getValidatorMap(cfDef);
        if (validators.get(col.name()) == null)
        {
            Map<MarshallerType, AbstractType> marshallers = getDefaultMarshallers(cfDef);
            setTupleValue(pair, 1, marshallers.get(MarshallerType.DEFAULT_VALIDATOR).compose(col.value()));
        }
        else
            setTupleValue(pair, 1, validators.get(col.name()).compose(col.value()));
        return pair;
    }
    else
    {
        // super
        ArrayList<Tuple> subcols = new ArrayList<Tuple>();
        for (IColumn subcol : col.getSubColumns())
            subcols.add(columnToTuple(subcol, cfDef, parseType(cfDef.getSubcomparator_type())));

        pair.set(1, new DefaultDataBag(subcols));
    }
    return pair;
}
 

import org.apache.cassandra.db.IColumn; //导入依赖的package包/类
public String getColumnsString(Collection<IColumn> columns)
{
    StringBuilder builder = new StringBuilder();
    for (IColumn column : columns)
    {
        builder.append(column.getString(this)).append(",");
    }
    return builder.toString();
}
 

import org.apache.cassandra.db.IColumn; //导入依赖的package包/类
public void count(IColumn column, IColumnContainer container)
{
    if (!isLive(column, container))
    {
        ignored++;
        return;
    }

    if (toGroup == 0)
    {
        live = 1;
        return;
    }

    ByteBuffer[] current = type.split(column.name());
    assert current.length >= toGroup;

    if (last != null)
    {
        boolean isSameGroup = true;
        for (int i = 0; i < toGroup; i++)
        {
            if (ByteBufferUtil.compareUnsigned(last[i], current[i]) != 0)
            {
                isSameGroup = false;
                break;
            }
        }

        if (isSameGroup)
            return;
    }

    live++;
    last = current;
}
 

import org.apache.cassandra.db.IColumn; //导入依赖的package包/类
public void map(ByteBuffer key, SortedMap<ByteBuffer, IColumn> columns, Context context) throws IOException, InterruptedException
{
    long sum = 0;
    for (IColumn column : columns.values())
    {
        logger.debug("read " + key + ":" + column.name() + " from " + context.getInputSplit());
        sum += ByteBufferUtil.toLong(column.value());
    }
    context.write(new Text(ByteBufferUtil.string(key)), new LongWritable(sum));
}
 

import org.apache.cassandra.db.IColumn; //导入依赖的package包/类
protected IColumn unthriftify(ColumnOrSuperColumn cosc)
{
    if (cosc.counter_column != null)
        return unthriftifyCounter(cosc.counter_column);
    if (cosc.counter_super_column != null)
        return unthriftifySuperCounter(cosc.counter_super_column);
    if (cosc.super_column != null)
        return unthriftifySuper(cosc.super_column);
    assert cosc.column != null;
    return unthriftifySimple(cosc.column);
}