import org.apache.cassandra.transport.Server; //导入依赖的package包/类
@Override protected void before() throws Throwable {
  if (server != null) return;

  DatabaseDescriptor.daemonInitialization();

  // Cleanup first
  try {
    cleanupAndLeaveDirs();
  } catch (IOException e) {
    throw new RuntimeException("Failed to cleanup and recreate directories.", e);
  }

  Keyspace.setInitialized();
  SystemKeyspace.persistLocalMetadata();
  SystemKeyspace.finishStartup();
  StorageService.instance.initServer();

  server = new Server.Builder().withHost(nativeAddr).withPort(nativePort).build();
  server.start();
}
 

import org.apache.cassandra.transport.Server; //导入依赖的package包/类
/**
 * Returns the element at the given index in a list.
 * @param serializedList a serialized list
 * @param index the index to get
 * @return the serialized element at the given index, or null if the index exceeds the list size
 */
public ByteBuffer getElement(ByteBuffer serializedList, int index)
{
    try
    {
        ByteBuffer input = serializedList.duplicate();
        int n = readCollectionSize(input, Server.VERSION_3);
        if (n <= index)
            return null;

        for (int i = 0; i < index; i++)
        {
            int length = input.getInt();
            input.position(input.position() + length);
        }
        return readValue(input, Server.VERSION_3);
    }
    catch (BufferUnderflowException e)
    {
        throw new MarshalException("Not enough bytes to read a list");
    }
}
 

import org.apache.cassandra.transport.Server; //导入依赖的package包/类
/**
 * Given a serialized map, gets the value associated with a given key.
 * @param serializedMap a serialized map
 * @param serializedKey a serialized key
 * @param keyType the key type for the map
 * @return the value associated with the key if one exists, null otherwise
 */
public ByteBuffer getSerializedValue(ByteBuffer serializedMap, ByteBuffer serializedKey, AbstractType keyType)
{
    try
    {
        ByteBuffer input = serializedMap.duplicate();
        int n = readCollectionSize(input, Server.VERSION_3);
        for (int i = 0; i < n; i++)
        {
            ByteBuffer kbb = readValue(input, Server.VERSION_3);
            ByteBuffer vbb = readValue(input, Server.VERSION_3);
            int comparison = keyType.compare(kbb, serializedKey);
            if (comparison == 0)
                return vbb;
            else if (comparison > 0)
                // since the map is in sorted order, we know we've gone too far and the element doesn't exist
                return null;
        }
        return null;
    }
    catch (BufferUnderflowException e)
    {
        throw new MarshalException("Not enough bytes to read a map");
    }
}
 

import org.apache.cassandra.transport.Server; //导入依赖的package包/类
static void doAdd(Term.Terminal value, ColumnDefinition column, UpdateParameters params) throws InvalidRequestException
{
    if (column.type.isMultiCell())
    {
        if (value == null)
            return;

        for (ByteBuffer bb : ((Value) value).elements)
        {
            if (bb == ByteBufferUtil.UNSET_BYTE_BUFFER)
                continue;

            params.addCell(column, CellPath.create(bb), ByteBufferUtil.EMPTY_BYTE_BUFFER);
        }
    }
    else
    {
        // for frozen sets, we're overwriting the whole cell
        if (value == null)
            params.addTombstone(column);
        else
            params.addCell(column, value.get(Server.CURRENT_VERSION));
    }
}
 

import org.apache.cassandra.transport.Server; //导入依赖的package包/类
static void doAppend(Term.Terminal value, ColumnDefinition column, UpdateParameters params) throws InvalidRequestException
{
    if (column.type.isMultiCell())
    {
        // If we append null, do nothing. Note that for Setter, we've
        // already removed the previous value so we're good here too
        if (value == null)
            return;

        for (ByteBuffer buffer : ((Value) value).elements)
        {
            ByteBuffer uuid = ByteBuffer.wrap(UUIDGen.getTimeUUIDBytes());
            params.addCell(column, CellPath.create(uuid), buffer);
        }
    }
    else
    {
        // for frozen lists, we're overwriting the whole cell value
        if (value == null)
            params.addTombstone(column);
        else
            params.addCell(column, value.get(Server.CURRENT_VERSION));
    }
}
 

import org.apache.cassandra.transport.Server; //导入依赖的package包/类
static void doPut(Term.Terminal value, ColumnDefinition column, UpdateParameters params) throws InvalidRequestException
{
    if (column.type.isMultiCell())
    {
        if (value == null)
            return;

        Map<ByteBuffer, ByteBuffer> elements = ((Value) value).map;
        for (Map.Entry<ByteBuffer, ByteBuffer> entry : elements.entrySet())
            params.addCell(column, CellPath.create(entry.getKey()), entry.getValue());
    }
    else
    {
        // for frozen maps, we're overwriting the whole cell
        if (value == null)
            params.addTombstone(column);
        else
            params.addCell(column, value.get(Server.CURRENT_VERSION));
    }
}
 

import org.apache.cassandra.transport.Server; //导入依赖的package包/类
@Test
public void testNative()
{
    // test each native type against each supported protocol version (although it doesn't make sense to
    // iterate through all protocol versions as of C* 3.0).

    for (int version = Server.MIN_SUPPORTED_VERSION; version <= Server.CURRENT_VERSION; version++)
    {
        for (Map.Entry<CQL3Type.Native, List<Value>> entry : nativeTypeValues.entrySet())
        {
            for (Value value : entry.getValue())
            {
                compareCqlLiteral(version, value);
            }
        }
    }
}
 

import org.apache.cassandra.transport.Server; //导入依赖的package包/类
@Test
public void testNested()
{
    // This is the "nice" part of this unit test - it tests (probably) nested type structures
    // like 'tuple<map, list<user>, tuple, user>' or 'map<tuple<int, text>, set<inet>>' with
    // random types  against each supported protocol version.

    for (int version = Server.MIN_SUPPORTED_VERSION; version <= Server.CURRENT_VERSION; version++)
    {
        for (int n = 0; n < 100; n++)
        {
            Value value = randomNested(version);
            compareCqlLiteral(version, value);
        }
    }
}
 

import org.apache.cassandra.transport.Server; //导入依赖的package包/类
@Test
public void SliceQueryWithTombstoneTest() throws Exception
{
    // Testing for the bug of #6748
    String keyspace = "cql_keyspace";
    String table = "table2";
    ColumnFamilyStore cfs = Keyspace.open(keyspace).getColumnFamilyStore(table);

    // Insert rows but with a tombstone as last cell
    for (int i = 0; i < 5; i++)
        executeInternal(String.format("INSERT INTO %s.%s (k, c, v) VALUES ('k%d', 'c%d', null)", keyspace, table, 0, i));

    ReadCommand command = SinglePartitionReadCommand.create(cfs.metadata, nowInSec, Util.dk("k0"), Slice.ALL);

    QueryPager pager = command.getPager(null, Server.CURRENT_VERSION);

    for (int i = 0; i < 5; i++)
    {
        List<FilteredPartition> partitions = query(pager, 1);
        // The only live cell we should have each time is the row marker
        assertRow(partitions.get(0), "k0", "c" + i);
    }
}
 

import org.apache.cassandra.transport.Server; //导入依赖的package包/类
@Test
public void testSerializationBackwardCompatibility()
{
    /*
     * Tests that the serialized paging state for the native protocol V3 is backward compatible
     * with what old nodes generate. For that, it compares the serialized format to the hard-coded
     * value of the same state generated on a 2.1. For the curious, said hardcoded value has been
     * generated by the following code:
     *     ByteBuffer pk = ByteBufferUtil.bytes("someKey");
     *     CellName cn = CellNames.compositeSparse(new ByteBuffer[]{ ByteBufferUtil.bytes("c1"), ByteBufferUtil.bytes(42) },
     *                                             new ColumnIdentifier("myCol", false),
     *                                             false);
     *     PagingState state = new PagingState(pk, cn.toByteBuffer(), 10);
     *     System.out.println("PagingState = " + ByteBufferUtil.bytesToHex(state.serialize()));
     */
    PagingState state = makeSomePagingState(Server.VERSION_3);

    String serializedState = ByteBufferUtil.bytesToHex(state.serialize(Server.VERSION_3));
    // Note that we don't assert exact equality because we know 3.0 nodes include the "remainingInPartition" number
    // that is not present on 2.1/2.2 nodes. We know this is ok however because we know that 2.1/2.2 nodes will ignore
    // anything remaining once they have properly deserialized a paging state.
    assertTrue(serializedState.startsWith("0007736f6d654b65790014000263310000040000002a0000056d79636f6c000000000a"));
}
 

import org.apache.cassandra.transport.Server; //导入依赖的package包/类
@Test
public void testSSLOnly()
{
    // default ssl settings: client encryption enabled and default native transport port used for ssl only
    DatabaseDescriptor.getClientEncryptionOptions().enabled = true;
    DatabaseDescriptor.getClientEncryptionOptions().optional = false;

    withService((NativeTransportService service) ->
                {
                    service.initialize();
                    assertEquals(1, service.getServers().size());
                    Server server = service.getServers().iterator().next();
                    assertTrue(server.useSSL);
                    assertEquals(server.socket.getPort(), DatabaseDescriptor.getNativeTransportPort());
                }, false, 1);
}
 

import org.apache.cassandra.transport.Server; //导入依赖的package包/类
@Test
public void testSSLOptional()
{
    // default ssl settings: client encryption enabled and default native transport port used for optional ssl
    DatabaseDescriptor.getClientEncryptionOptions().enabled = true;
    DatabaseDescriptor.getClientEncryptionOptions().optional = true;

    withService((NativeTransportService service) ->
                {
                    service.initialize();
                    assertEquals(1, service.getServers().size());
                    Server server = service.getServers().iterator().next();
                    assertTrue(server.useSSL);
                    assertEquals(server.socket.getPort(), DatabaseDescriptor.getNativeTransportPort());
                }, false, 1);
}
 

import org.apache.cassandra.transport.Server; //导入依赖的package包/类
@Test
public void testSSLWithNonSSL()
{
    // ssl+non-ssl settings: client encryption enabled and additional ssl port specified
    DatabaseDescriptor.getClientEncryptionOptions().enabled = true;
    DatabaseDescriptor.setNativeTransportPortSSL(8432);

    withService((NativeTransportService service) ->
                {
                    service.initialize();
                    assertEquals(2, service.getServers().size());
                    assertEquals(
                                Sets.newHashSet(Arrays.asList(
                                                             Pair.create(true, DatabaseDescriptor.getNativeTransportPortSSL()),
                                                             Pair.create(false, DatabaseDescriptor.getNativeTransportPort())
                                                )
                                ),
                                service.getServers().stream().map((Server s) ->
                                                                  Pair.create(s.useSSL, s.socket.getPort())).collect(Collectors.toSet())
                    );
                }, false, 1);
}
 

import org.apache.cassandra.transport.Server; //导入依赖的package包/类
@Test
public void customExpressionsDisallowedInModifications() throws Throwable
{
    createTable("CREATE TABLE %s (a int, b int, c int, d int, PRIMARY KEY (a, b))");
    String indexName = currentTable() + "_custom_index";
    createIndex(String.format("CREATE CUSTOM INDEX %s ON %%s(c) USING '%s'",
                              indexName, StubIndex.class.getName()));

    assertInvalidThrowMessage(Server.CURRENT_VERSION,
                              ModificationStatement.CUSTOM_EXPRESSIONS_NOT_ALLOWED,
                              QueryValidationException.class,
                              String.format("DELETE FROM %%s WHERE expr(%s, 'foo bar baz ')", indexName));
    assertInvalidThrowMessage(Server.CURRENT_VERSION,
                              ModificationStatement.CUSTOM_EXPRESSIONS_NOT_ALLOWED,
                              QueryValidationException.class,
                              String.format("UPDATE %%s SET d=0 WHERE expr(%s, 'foo bar baz ')", indexName));
}
 

import org.apache.cassandra.transport.Server; //导入依赖的package包/类
@Test
public void testCustomExpressionValueType() throws Throwable
{
    // verify that the type of the expression value is determined by Index::customExpressionValueType
    createTable("CREATE TABLE %s (k int, v1 uuid, v2 blob, PRIMARY KEY(k))");
    createIndex(String.format("CREATE CUSTOM INDEX int_index ON %%s() USING '%s'",
                              Int32ExpressionIndex.class.getName()));
    createIndex(String.format("CREATE CUSTOM INDEX text_index ON %%s() USING '%s'",
                              UTF8ExpressionIndex.class.getName()));

    execute("SELECT * FROM %s WHERE expr(text_index, 'foo')");
    assertInvalidThrowMessage(Server.CURRENT_VERSION,
                              "Invalid INTEGER constant (99) for \"custom index expression\" of type text",
                              QueryValidationException.class,
                              "SELECT * FROM %s WHERE expr(text_index, 99)");

    execute("SELECT * FROM %s WHERE expr(int_index, 99)");
    assertInvalidThrowMessage(Server.CURRENT_VERSION,
                              "Invalid STRING constant (foo) for \"custom index expression\" of type int",
                              QueryValidationException.class,
                              "SELECT * FROM %s WHERE expr(int_index, 'foo')");
}
 

import org.apache.cassandra.transport.Server; //导入依赖的package包/类
private ByteBuffer[] bindInternal(QueryOptions options) throws InvalidRequestException
{
    int version = options.getProtocolVersion();

    ByteBuffer[] buffers = new ByteBuffer[values.size()];
    for (int i = 0; i < type.size(); i++)
    {
        buffers[i] = values.get(i).bindAndGet(options);
        // Inside UDT values, we must force the serialization of collections to v3 whatever protocol
        // version is in use since we're going to store directly that serialized value.
        if (version < Server.VERSION_3 && type.fieldType(i).isCollection() && buffers[i] != null)
            buffers[i] = ((CollectionType)type.fieldType(i)).getSerializer().reserializeToV3(buffers[i]);
    }
    return buffers;
}
 

import org.apache.cassandra.transport.Server; //导入依赖的package包/类
static void doAppend(Term t, ColumnFamily cf, Composite prefix, ColumnDefinition column, UpdateParameters params) throws InvalidRequestException
{
    Term.Terminal value = t.bind(params.options);
    Lists.Value listValue = (Lists.Value)value;
    if (column.type.isMultiCell())
    {
        // If we append null, do nothing. Note that for Setter, we've
        // already removed the previous value so we're good here too
        if (value == null)
            return;

        List<ByteBuffer> toAdd = listValue.elements;
        for (int i = 0; i < toAdd.size(); i++)
        {
            ByteBuffer uuid = ByteBuffer.wrap(UUIDGen.getTimeUUIDBytes());
            cf.addColumn(params.makeColumn(cf.getComparator().create(prefix, column, uuid), toAdd.get(i)));
        }
    }
    else
    {
        // for frozen lists, we're overwriting the whole cell value
        CellName name = cf.getComparator().create(prefix, column);
        if (value == null)
            cf.addAtom(params.makeTombstone(name));
        else
            cf.addColumn(params.makeColumn(name, listValue.getWithProtocolVersion(Server.CURRENT_VERSION)));
    }
}
 

import org.apache.cassandra.transport.Server; //导入依赖的package包/类
private static void setupVersion()
{
    String req = "INSERT INTO system.%s (key, release_version, cql_version, thrift_version, native_protocol_version, data_center, rack, partitioner) VALUES (?, ?, ?, ?, ?, ?, ?, ?)";
    IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
    executeOnceInternal(String.format(req, LOCAL_CF),
                        LOCAL_KEY,
                        FBUtilities.getReleaseVersionString(),
                        QueryProcessor.CQL_VERSION.toString(),
                        cassandraConstants.VERSION,
                        String.valueOf(Server.CURRENT_VERSION),
                        snitch.getDatacenter(FBUtilities.getBroadcastAddress()),
                        snitch.getRack(FBUtilities.getBroadcastAddress()),
                        DatabaseDescriptor.getPartitioner().getClass().getName());
}
 

import org.apache.cassandra.transport.Server; //导入依赖的package包/类
public static int compareMaps(AbstractType<?> keysComparator, AbstractType<?> valuesComparator, ByteBuffer o1, ByteBuffer o2)
{
     if (!o1.hasRemaining() || !o2.hasRemaining())
        return o1.hasRemaining() ? 1 : o2.hasRemaining() ? -1 : 0;

    ByteBuffer bb1 = o1.duplicate();
    ByteBuffer bb2 = o2.duplicate();

    int protocolVersion = Server.VERSION_3;
    int size1 = CollectionSerializer.readCollectionSize(bb1, protocolVersion);
    int size2 = CollectionSerializer.readCollectionSize(bb2, protocolVersion);

    for (int i = 0; i < Math.min(size1, size2); i++)
    {
        ByteBuffer k1 = CollectionSerializer.readValue(bb1, protocolVersion);
        ByteBuffer k2 = CollectionSerializer.readValue(bb2, protocolVersion);
        int cmp = keysComparator.compare(k1, k2);
        if (cmp != 0)
            return cmp;

        ByteBuffer v1 = CollectionSerializer.readValue(bb1, protocolVersion);
        ByteBuffer v2 = CollectionSerializer.readValue(bb2, protocolVersion);
        cmp = valuesComparator.compare(v1, v2);
        if (cmp != 0)
            return cmp;
    }

    return size1 == size2 ? 0 : (size1 < size2 ? -1 : 1);
}
 

import org.apache.cassandra.transport.Server; //导入依赖的package包/类
public T deserialize(ByteBuffer bytes)
{
    // The only cases we serialize/deserialize collections internally (i.e. not for the protocol sake),
    // is:
    //  1) when collections are frozen
    //  2) for internal calls.
    // In both case, using the protocol 3 version variant is the right thing to do.
    return deserializeForNativeProtocol(bytes, Server.VERSION_3);
}