import org.apache.cassandra.io.util.DataOutputPlus; //导入依赖的package包/类
private void serializeSimple(DataOutputPlus out) throws IOException
{
    List<AbstractType<?>> types = new ArrayList<>(comparator.clusteringComparator.subtypes());
    assert types.size() == 1 : types;

    for (int i = 0; i < size; i++)
    {
        LegacyBound start = starts[i];
        LegacyBound end = ends[i];

        ClusteringPrefix startClustering = start.bound.clustering();
        ClusteringPrefix endClustering = end.bound.clustering();

        assert startClustering.size() == 1;
        assert endClustering.size() == 1;

        ByteBufferUtil.writeWithShortLength(startClustering.get(0), out);
        ByteBufferUtil.writeWithShortLength(endClustering.get(0), out);

        out.writeInt(delTimes[i]);
        out.writeLong(markedAts[i]);
    }
}
 

import org.apache.cassandra.io.util.DataOutputPlus; //导入依赖的package包/类
public void serialize(CachedPartition partition, DataOutputPlus out) throws IOException
{
    int version = MessagingService.current_version;

    assert partition instanceof CachedBTreePartition;
    CachedBTreePartition p = (CachedBTreePartition)partition;

    out.writeInt(p.createdAtInSec);
    out.writeInt(p.cachedLiveRows);
    out.writeInt(p.rowsWithNonExpiringCells);
    out.writeInt(p.nonTombstoneCellCount);
    out.writeInt(p.nonExpiringLiveCells);
    CFMetaData.serializer.serialize(partition.metadata(), out, version);
    try (UnfilteredRowIterator iter = p.unfilteredIterator())
    {
        UnfilteredRowIteratorSerializer.serializer.serialize(iter, null, out, version, p.rowCount());
    }
}
 

import org.apache.cassandra.io.util.DataOutputPlus; //导入依赖的package包/类
@Override
public void serialize(DataOutputPlus out, IndexInfo.Serializer idxInfoSerializer, ByteBuffer indexInfo) throws IOException
{
    assert indexedPartSize != Integer.MIN_VALUE;

    out.writeUnsignedVInt(position);

    out.writeUnsignedVInt(serializedSize(deletionTime, headerLength, columnsIndex.length) + indexedPartSize);

    out.writeUnsignedVInt(headerLength);
    DeletionTime.serializer.serialize(deletionTime, out);
    out.writeUnsignedVInt(columnsIndex.length);
    for (IndexInfo info : columnsIndex)
        idxInfoSerializer.serialize(info, out);
    for (int offset : offsets)
        out.writeInt(offset);
}
 

import org.apache.cassandra.io.util.DataOutputPlus; //导入依赖的package包/类
void serializeValuesWithoutSize(ClusteringPrefix clustering, DataOutputPlus out, int version, List<AbstractType<?>> types) throws IOException
{
    int offset = 0;
    int clusteringSize = clustering.size();
    // serialize in batches of 32, to avoid garbage when deserializing headers
    while (offset < clusteringSize)
    {
        // we micro-batch the headers, so that we can incur fewer method calls,
        // and generate no garbage on deserialization;
        // we piggyback on vint encoding so that, typically, only 1 byte is used per 32 clustering values,
        // i.e. more than we ever expect to see
        int limit = Math.min(clusteringSize, offset + 32);
        out.writeUnsignedVInt(makeHeader(clustering, offset, limit));
        while (offset < limit)
        {
            ByteBuffer v = clustering.get(offset);
            if (v != null && v.hasRemaining())
                types.get(offset).writeValue(v, out);
            offset++;
        }
    }
}
 

import org.apache.cassandra.io.util.DataOutputPlus; //导入依赖的package包/类
public void serializeForMessaging(SerializationHeader header, ColumnFilter selection, DataOutputPlus out, boolean hasStatic) throws IOException
{
    EncodingStats.serializer.serialize(header.stats, out);

    if (selection == null)
    {
        if (hasStatic)
            Columns.serializer.serialize(header.columns.statics, out);
        Columns.serializer.serialize(header.columns.regulars, out);
    }
    else
    {
        if (hasStatic)
            Columns.serializer.serializeSubset(header.columns.statics, selection.fetchedColumns().statics, out);
        Columns.serializer.serializeSubset(header.columns.regulars, selection.fetchedColumns().regulars, out);
    }
}
 

import org.apache.cassandra.io.util.DataOutputPlus; //导入依赖的package包/类
public void serialize(ColumnFilter selection, DataOutputPlus out, int version) throws IOException
{
    out.writeByte(makeHeaderByte(selection));

    if (selection.queried != null)
    {
        Columns.serializer.serialize(selection.queried.statics, out);
        Columns.serializer.serialize(selection.queried.regulars, out);
    }

    if (selection.subSelections != null)
    {
        out.writeUnsignedVInt(selection.subSelections.size());
        for (ColumnSubselection subSel : selection.subSelections.values())
            ColumnSubselection.serializer.serialize(subSel, out, version);
    }
}
 

import org.apache.cassandra.io.util.DataOutputPlus; //导入依赖的package包/类
public void serialize(ColumnSubselection subSel, DataOutputPlus out, int version) throws IOException
{
    ColumnDefinition column = subSel.column();
    ByteBufferUtil.writeWithShortLength(column.name.bytes, out);
    out.writeByte(subSel.kind().ordinal());
    switch (subSel.kind())
    {
        case SLICE:
            Slice slice = (Slice)subSel;
            column.cellPathSerializer().serialize(slice.from, out);
            column.cellPathSerializer().serialize(slice.to, out);
            break;
        case ELEMENT:
            Element eltSelection = (Element)subSel;
            column.cellPathSerializer().serialize(eltSelection.element, out);
            break;
        default:
            throw new AssertionError();
    }
}
 

import org.apache.cassandra.io.util.DataOutputPlus; //导入依赖的package包/类
private void serialize(RangeTombstoneMarker marker, SerializationHeader header, DataOutputPlus out, long previousUnfilteredSize, int version)
throws IOException
{
    out.writeByte((byte)IS_MARKER);
    ClusteringBoundOrBoundary.serializer.serialize(marker.clustering(), out, version, header.clusteringTypes());

    if (header.isForSSTable())
    {
        out.writeUnsignedVInt(serializedMarkerBodySize(marker, header, previousUnfilteredSize, version));
        out.writeUnsignedVInt(previousUnfilteredSize);
    }

    if (marker.isBoundary())
    {
        RangeTombstoneBoundaryMarker bm = (RangeTombstoneBoundaryMarker)marker;
        header.writeDeletionTime(bm.endDeletionTime(), out);
        header.writeDeletionTime(bm.startDeletionTime(), out);
    }
    else
    {
        header.writeDeletionTime(((RangeTombstoneBoundMarker)marker).deletionTime(), out);
    }
}
 

import org.apache.cassandra.io.util.DataOutputPlus; //导入依赖的package包/类
public RowIndexEntry write(long currentPosition, DataOutputPlus out) throws IOException {
    assert !closed;

    ColumnIndex columnsIndex;
    try {
        indexBuilder = new ColumnIndex.Builder(emptyColumnFamily, key.getKey(), out);
        columnsIndex = indexBuilder.buildForCompaction(merger);

        // if there aren't any columns or tombstones, return null
        if (columnsIndex.columnsIndex.isEmpty() && !emptyColumnFamily.isMarkedForDelete())
            return null;
    } catch (IOException e) {
        throw new RuntimeException(e);
    }

    // in case no columns were ever written, we may still need to write an empty header with a top-level tombstone
    indexBuilder.maybeWriteEmptyRowHeader();

    out.writeShort(BigTableWriter.END_OF_ROW);

    close();

    return RowIndexEntry.create(currentPosition, emptyColumnFamily.deletionInfo().getTopLevelDeletion(), columnsIndex);
}
 

import org.apache.cassandra.io.util.DataOutputPlus; //导入依赖的package包/类
public void serialize(AbstractBounds<?> range, DataOutputPlus out, int version) throws IOException
{
    /*
     * The first int tells us if it's a range or bounds (depending on the value) _and_ if it's tokens or keys (depending on the
     * sign). We use negative kind for keys so as to preserve the serialization of token from older version.
     */
    out.writeInt(kindInt(range));
    if (range.left instanceof Token)
    {
        Token.serializer.serialize((Token) range.left, out);
        Token.serializer.serialize((Token) range.right, out);
    }
    else
    {
        RowPosition.serializer.serialize((RowPosition) range.left, out);
        RowPosition.serializer.serialize((RowPosition) range.right, out);
    }
}
 

import org.apache.cassandra.io.util.DataOutputPlus; //导入依赖的package包/类
public void serialize(EndpointState epState, DataOutputPlus out, int version) throws IOException
{
    /* serialize the HeartBeatState */
    HeartBeatState hbState = epState.getHeartBeatState();
    HeartBeatState.serializer.serialize(hbState, out, version);

    /* serialize the map of ApplicationState objects */
    int size = epState.applicationState.size();
    out.writeInt(size);
    for (Map.Entry<ApplicationState, VersionedValue> entry : epState.applicationState.entrySet())
    {
        VersionedValue value = entry.getValue();
        out.writeInt(entry.getKey().ordinal());
        VersionedValue.serializer.serialize(value, out, version);
    }
}
 

import org.apache.cassandra.io.util.DataOutputPlus; //导入依赖的package包/类
void serialize(DataOutputPlus out) throws IOException
{
    CRC32 crc = new CRC32();

    out.writeInt(version);
    updateChecksumInt(crc, version);

    out.writeLong(timestamp);
    updateChecksumLong(crc, timestamp);

    out.writeLong(hostId.getMostSignificantBits());
    updateChecksumLong(crc, hostId.getMostSignificantBits());
    out.writeLong(hostId.getLeastSignificantBits());
    updateChecksumLong(crc, hostId.getLeastSignificantBits());

    byte[] paramsBytes = JSONValue.toJSONString(parameters).getBytes(StandardCharsets.UTF_8);
    out.writeInt(paramsBytes.length);
    updateChecksumInt(crc, paramsBytes.length);
    out.writeInt((int) crc.getValue());

    out.write(paramsBytes);
    crc.update(paramsBytes, 0, paramsBytes.length);

    out.writeInt((int) crc.getValue());
}
 

import org.apache.cassandra.io.util.DataOutputPlus; //导入依赖的package包/类
public void serialize(Cell cell, DataOutputPlus out) throws IOException
{
    assert !cell.name().isEmpty();
    type.cellSerializer().serialize(cell.name(), out);
    try
    {
        out.writeByte(cell.serializationFlags());
        if (cell instanceof CounterCell)
        {
            out.writeLong(((CounterCell) cell).timestampOfLastDelete());
        }
        else if (cell instanceof ExpiringCell)
        {
            out.writeInt(((ExpiringCell) cell).getTimeToLive());
            out.writeInt(cell.getLocalDeletionTime());
        }
        out.writeLong(cell.timestamp());
        ByteBufferUtil.writeWithLength(cell.value(), out);
    }
    catch (IOException e)
    {
        throw new RuntimeException(e);
    }
}
 

import org.apache.cassandra.io.util.DataOutputPlus; //导入依赖的package包/类
public void serialize(DataLimits limits, DataOutputPlus out, int version) throws IOException
{
    out.writeByte(limits.kind().ordinal());
    switch (limits.kind())
    {
        case CQL_LIMIT:
        case CQL_PAGING_LIMIT:
            CQLLimits cqlLimits = (CQLLimits)limits;
            out.writeUnsignedVInt(cqlLimits.rowLimit);
            out.writeUnsignedVInt(cqlLimits.perPartitionLimit);
            out.writeBoolean(cqlLimits.isDistinct);
            if (limits.kind() == Kind.CQL_PAGING_LIMIT)
            {
                CQLPagingLimits pagingLimits = (CQLPagingLimits)cqlLimits;
                ByteBufferUtil.writeWithVIntLength(pagingLimits.lastReturnedKey, out);
                out.writeUnsignedVInt(pagingLimits.lastReturnedKeyRemaining);
            }
            break;
        case THRIFT_LIMIT:
        case SUPER_COLUMN_COUNTING_LIMIT:
            ThriftLimits thriftLimits = (ThriftLimits)limits;
            out.writeUnsignedVInt(thriftLimits.partitionLimit);
            out.writeUnsignedVInt(thriftLimits.cellPerPartitionLimit);
            break;
    }
}
 

import org.apache.cassandra.io.util.DataOutputPlus; //导入依赖的package包/类
public void serialize(ColumnFilter selection, DataOutputPlus out, int version) throws IOException
{
    out.writeByte(makeHeaderByte(selection));

    if (selection.selection != null)
    {
        Columns.serializer.serialize(selection.selection.statics, out);
        Columns.serializer.serialize(selection.selection.regulars, out);
    }

    if (selection.subSelections != null)
    {
        out.writeUnsignedVInt(selection.subSelections.size());
        for (ColumnSubselection subSel : selection.subSelections.values())
            ColumnSubselection.serializer.serialize(subSel, out, version);
    }
}
 

import org.apache.cassandra.io.util.DataOutputPlus; //导入依赖的package包/类
public void serialize(Mutation mutation, DataOutputPlus out, int version) throws IOException
{
    if (version < MessagingService.VERSION_20)
        out.writeUTF(mutation.getKeyspaceName());

    /* serialize the modifications in the mutation */
    int size = mutation.modifications.size();

    if (version < MessagingService.VERSION_30)
    {
        ByteBufferUtil.writeWithShortLength(mutation.key().getKey(), out);
        out.writeInt(size);
    }
    else
    {
        out.writeUnsignedVInt(size);
    }

    assert size > 0;
    for (Map.Entry<UUID, PartitionUpdate> entry : mutation.modifications.entrySet())
        PartitionUpdate.serializer.serialize(entry.getValue(), out, version);
}
 

import org.apache.cassandra.io.util.DataOutputPlus; //导入依赖的package包/类
public void serialize(ReadCommand command, DataOutputPlus out, int version) throws IOException
{
    assert version >= MessagingService.VERSION_30;

    out.writeByte(command.kind.ordinal());
    out.writeByte(digestFlag(command.isDigestQuery()) | thriftFlag(command.isForThrift()) | indexFlag(command.index.isPresent()));
    if (command.isDigestQuery())
        out.writeUnsignedVInt(command.digestVersion());
    CFMetaData.serializer.serialize(command.metadata(), out, version);
    out.writeInt(command.nowInSec());
    ColumnFilter.serializer.serialize(command.columnFilter(), out, version);
    RowFilter.serializer.serialize(command.rowFilter(), out, version);
    DataLimits.serializer.serialize(command.limits(), out, version);
    if (command.index.isPresent())
        IndexMetadata.serializer.serialize(command.index.get(), out, version);

    command.serializeSelection(out, version);
}
 

import org.apache.cassandra.io.util.DataOutputPlus; //导入依赖的package包/类
public void serialize(KeyCacheKey key, DataOutputPlus out, ColumnFamilyStore cfs) throws IOException
{
    //Don't serialize old format entries since we didn't bother to implement serialization of both for simplicity
    //https://issues.apache.org/jira/browse/CASSANDRA-10778
    if (!key.desc.version.storeRows()) return;

    RowIndexEntry entry = CacheService.instance.keyCache.getInternal(key);
    if (entry == null)
        return;

    out.write(cfs.metadata.ksAndCFBytes);
    ByteBufferUtil.writeWithLength(key.key, out);
    out.writeInt(key.desc.generation);
    out.writeBoolean(true);
    key.desc.getFormat().getIndexSerializer(cfs.metadata, key.desc.version, SerializationHeader.forKeyCache(cfs.metadata)).serialize(entry, out);
}
 

import org.apache.cassandra.io.util.DataOutputPlus; //导入依赖的package包/类
public static void writeWithShortLength(ByteBuffer buffer, DataOutputPlus out) throws IOException
{
    int length = buffer.remaining();
    assert 0 <= length && length <= FBUtilities.MAX_UNSIGNED_SHORT
        : String.format("Attempted serializing to buffer exceeded maximum of %s bytes: %s", FBUtilities.MAX_UNSIGNED_SHORT, length);
    out.writeShort(length);
    out.write(buffer);
}
 

import org.apache.cassandra.io.util.DataOutputPlus; //导入依赖的package包/类
public void serialize(EstimatedHistogram eh, DataOutputPlus out) throws IOException
{
    long[] offsets = eh.getBucketOffsets();
    long[] buckets = eh.getBuckets(false);
    out.writeInt(buckets.length);
    for (int i = 0; i < buckets.length; i++)
    {
        out.writeLong(offsets[i == 0 ? 0 : i - 1]);
        out.writeLong(buckets[i]);
    }
}