import org.apache.cassandra.dht.Murmur3Partitioner; //导入依赖的package包/类
@Test
public void testParsingCQLOnCompoundedTable() {
   String cql = "CREATE TABLE keyspace1.compressed_bills (" +
           "     user text," +
           "     email text," +
           "     account_id text static," +
           "     balance int static," +
           "     expense_id int," +
           "     item_id int," +
           "     amount int," +
           "     name text," +
           "     PRIMARY KEY ((user, email), expense_id, item_id))";


    ParsedStatement stmt = QueryProcessor.parseStatement(cql);
    stmt.properties.properties.addProperty(CFProperties.KEYSPACE_NAME, "keyspace1");
    stmt.properties.properties.addProperty(CFProperties.PARTITIONER_CLASS,
                                           "org.apache.cassandra.dht.Murmur3Partitioner");

    ParsedStatement.Prepared preparedStmt = stmt.prepare();

    CFMetaData cfMetaData = ((CreateTableStatement) preparedStmt.statement).getCFMetaData();
    Assert.assertEquals(cfMetaData.ksName, "keyspace1");
    Assert.assertEquals(cfMetaData.partitioner, Murmur3Partitioner.instance);

}
 

import org.apache.cassandra.dht.Murmur3Partitioner; //导入依赖的package包/类
@Test
public void testMurmur3FilterHash()
{
    IPartitioner partitioner = new Murmur3Partitioner();
    Iterator<ByteBuffer> gen = new KeyGenerator.RandomStringGenerator(new Random().nextInt(), FilterTestHelper.ELEMENTS);
    long[] expected = new long[2];
    long[] actual = new long[2];
    while (gen.hasNext())
    {
        expected[0] = 1;
        expected[1] = 2;
        actual[0] = 3;
        actual[1] = 4;
        ByteBuffer key = gen.next();
        FilterKey expectedKey = FilterTestHelper.wrap(key);
        FilterKey actualKey = partitioner.decorateKey(key);
        actualKey.filterHash(actual);
        expectedKey.filterHash(expected);
        Assert.assertArrayEquals(expected, actual);
    }
}
 

import org.apache.cassandra.dht.Murmur3Partitioner; //导入依赖的package包/类
private void testShouldHint(Verb verb, ConsistencyLevel cl, boolean allowHints, boolean expectHint) throws Exception
{
    Object payload = verb == Verb.PAXOS_COMMIT
                     ? new Commit(UUID.randomUUID(), new PartitionUpdate(MockSchema.newCFMetaData("", ""), ByteBufferUtil.EMPTY_BYTE_BUFFER, PartitionColumns.NONE, 1))
                     : new Mutation("", new BufferDecoratedKey(new Murmur3Partitioner.LongToken(0), ByteBufferUtil.EMPTY_BYTE_BUFFER));

    WriteCallbackInfo wcbi = new WriteCallbackInfo(InetAddress.getByName("192.168.1.1"), null, new MessageOut(verb, payload, null), null, cl, allowHints);
    Assert.assertEquals(expectHint, wcbi.shouldHint());
    if (expectHint)
    {
        Assert.assertNotNull(wcbi.mutation());
    }
    else
    {
        boolean fail = false;
        try
        {
            wcbi.mutation();
        }
        catch (Throwable t)
        {
            fail = true;
        }
        Assert.assertTrue(fail);
    }
}
 

import org.apache.cassandra.dht.Murmur3Partitioner; //导入依赖的package包/类
@Test
public void compareMurmur3Partitioner() throws IOException {
  conf.set(CassandraParams.SCRUB_CASSANDRACLUSTER_PARTITIONER_CONFIG,
      Murmur3Partitioner.class.getName());
  comparator.setConf(conf);
  // murmur3_128("foo")[0] = -2129773440516405919
  // murmur3_128("bar")[0] = -7911037993560119804
  // murmur3_128("baz")[0] = 8295379539955784970
  checkOrder("bar", "foo");
  checkOrder("foo", "baz");
  checkOrder("bar", "baz");

  // Murmur3Partitioner maps empty string to Long.MIN_VALUE
  checkOrder("", "foo");
  checkOrder("", "bar");
}
 

import org.apache.cassandra.dht.Murmur3Partitioner; //导入依赖的package包/类
public void init()
{
    if (!(StorageService.getPartitioner() instanceof Murmur3Partitioner))
        throw new UnsupportedOperationException("SASI supported only with Murmur3Partitioner.");

    isInitialized = true;

    metrics = new IndexMetrics(baseCfs);

    // init() is called by SIM only on the instance that it will keep around, but will call addColumnDef on any instance
    // that it happens to create (and subsequently/immediately throw away)
    track(columnDefs);

    baseCfs.getDataTracker().subscribe(this);
}
 

import org.apache.cassandra.dht.Murmur3Partitioner; //导入依赖的package包/类
@Before
public void beforeTest()
{
    ExtendedFilter filter = ExtendedFilter.create(BACKEND.getBaseCfs(),
                                                  DataRange.allData(new Murmur3Partitioner()),
                                                  null,
                                                  Integer.MAX_VALUE,
                                                  false,
                                                  System.currentTimeMillis());

    controller = new QueryController(BACKEND, filter, TimeUnit.SECONDS.toMillis(10));
}
 

import org.apache.cassandra.dht.Murmur3Partitioner; //导入依赖的package包/类
@BeforeClass
public static void setUpClass()
{
    if (ROW_CACHE_SIZE_IN_MB > 0)
        DatabaseDescriptor.setRowCacheSizeInMB(ROW_CACHE_SIZE_IN_MB);

    StorageService.instance.setPartitionerUnsafe(Murmur3Partitioner.instance);
}
 

import org.apache.cassandra.dht.Murmur3Partitioner; //导入依赖的package包/类
public static CFMetaData newCFMetaData(String ksname, String cfname)
{
    CFMetaData metadata = CFMetaData.Builder.create(ksname, cfname)
                                            .addPartitionKey("key", UTF8Type.instance)
                                            .addClusteringColumn("col", UTF8Type.instance)
                                            .addRegularColumn("value", UTF8Type.instance)
                                            .withPartitioner(Murmur3Partitioner.instance)
                                            .build();
    metadata.caching(CachingParams.CACHE_NOTHING);
    return metadata;
}
 

import org.apache.cassandra.dht.Murmur3Partitioner; //导入依赖的package包/类
@Test
public void testSessionSuccess()
{
    int cmd = 321;
    String message = String.format("Repair session %s for range %s finished", UUID.randomUUID(),
                                   new Range<Token>(new Murmur3Partitioner.LongToken(3), new Murmur3Partitioner.LongToken(4)));
    Optional<int[]> result = LegacyJMXProgressSupport.getLegacyUserdata(String.format("repair:%d", cmd),
                                                        new ProgressEvent(ProgressEventType.PROGRESS, 2, 10, message));
    assertTrue(result.isPresent());
    assertArrayEquals(new int[]{ cmd, ActiveRepairService.Status.SESSION_SUCCESS.ordinal() }, result.get());
}
 

import org.apache.cassandra.dht.Murmur3Partitioner; //导入依赖的package包/类
@Test
public void testSessionFailed()
{
    int cmd = 321;
    String message = String.format("Repair session %s for range %s failed with error %s", UUID.randomUUID(),
                                   new Range<Token>(new Murmur3Partitioner.LongToken(3), new Murmur3Partitioner.LongToken(4)).toString(),
                                   new RuntimeException("error"));
    Optional<int[]> result = LegacyJMXProgressSupport.getLegacyUserdata(String.format("repair:%d", cmd),
                                                                        new ProgressEvent(ProgressEventType.PROGRESS, 2, 10, message));
    assertTrue(result.isPresent());
    assertArrayEquals(new int[]{ cmd, ActiveRepairService.Status.SESSION_FAILED.ordinal() }, result.get());
}
 

import org.apache.cassandra.dht.Murmur3Partitioner; //导入依赖的package包/类
@Test
public void testConviction() throws Exception
{
    InetAddress remote = InetAddress.getByName("127.0.0.2");
    Gossiper.instance.initializeNodeUnsafe(remote, UUID.randomUUID(), 1);

    // Set up RepairSession
    UUID parentSessionId = UUIDGen.getTimeUUID();
    UUID sessionId = UUID.randomUUID();
    IPartitioner p = Murmur3Partitioner.instance;
    Range<Token> repairRange = new Range<>(p.getToken(ByteBufferUtil.bytes(0)), p.getToken(ByteBufferUtil.bytes(100)));
    Set<InetAddress> endpoints = Sets.newHashSet(remote);
    RepairSession session = new RepairSession(parentSessionId, sessionId, Arrays.asList(repairRange), "Keyspace1", RepairParallelism.SEQUENTIAL, endpoints, ActiveRepairService.UNREPAIRED_SSTABLE, "Standard1");

    // perform convict
    session.convict(remote, Double.MAX_VALUE);

    // RepairSession should throw ExecutorException with the cause of IOException when getting its value
    try
    {
        session.get();
        fail();
    }
    catch (ExecutionException ex)
    {
        assertEquals(IOException.class, ex.getCause().getClass());
    }
}
 

import org.apache.cassandra.dht.Murmur3Partitioner; //导入依赖的package包/类
@Test
public void testIncrementalRepairWithSubrangesIsNotGlobal() throws Exception
{
    RepairOption ro = RepairOption.parse(ImmutableMap.of(RepairOption.INCREMENTAL_KEY, "true", RepairOption.RANGES_KEY, "42:42"),
                       Murmur3Partitioner.instance);
    assertFalse(ro.isGlobal());
    ro = RepairOption.parse(ImmutableMap.of(RepairOption.INCREMENTAL_KEY, "true", RepairOption.RANGES_KEY, ""),
            Murmur3Partitioner.instance);
    assertTrue(ro.isGlobal());
}
 

import org.apache.cassandra.dht.Murmur3Partitioner; //导入依赖的package包/类
@BeforeClass
public static void defineSchema() throws ConfigurationException
{
    sw = Util.switchPartitioner(Murmur3Partitioner.instance);
    SchemaLoader.prepareServer();
    SchemaLoader.createKeyspace(KEYSPACE1,
                                KeyspaceParams.simple(1),
                                SchemaLoader.standardCFMD(KEYSPACE1, CF_STANDARD1, 1, BytesType.instance),
                                SchemaLoader.standardCFMD(KEYSPACE1, CF_STANDARD2, 1, BytesType.instance),
                                SchemaLoader.standardCFMD(KEYSPACE1, CF_STANDARD3, 1, BytesType.instance),
                                SchemaLoader.standardCFMD(KEYSPACE1, CF_STANDARD4, 1, BytesType.instance),
                                SchemaLoader.standardCFMD(KEYSPACE1, CF_STANDARD5, 1, BytesType.instance));
}
 

import org.apache.cassandra.dht.Murmur3Partitioner; //导入依赖的package包/类
/**
 * Returns a new {@link TokenMapper} instance for the current partitioner using the specified column family
 * metadata.
 *
 * @param metadata The column family metadata.
 * @return A new {@link TokenMapper} instance for the current partitioner.
 */
public static TokenMapper instance(CFMetaData metadata) {
    IPartitioner partitioner = DatabaseDescriptor.getPartitioner();
    if (partitioner instanceof Murmur3Partitioner) {
        return new TokenMapperMurmur(metadata);
    } else {
        return new TokenMapperGeneric(metadata);
    }
}
 

import org.apache.cassandra.dht.Murmur3Partitioner; //导入依赖的package包/类
@Before
public void setup() {
	partitioner = new Murmur3Partitioner();
	rateLimiter = RateLimiter.create(1.0d * 1024.0d * 1024.0d);
	metadata = new ColumnFamilyMetadata(keyspace, columnFamily, keyAliasesString, columnAliasesString, keyValidatorString, comparatorString);
	sstableFile = new File(this.getClass().getResource(sstableFilePath).getPath());
}
 

import org.apache.cassandra.dht.Murmur3Partitioner; //导入依赖的package包/类
public static CFMetaData compile(String cql, String keyspace)
{
    return compile(cql, keyspace, Murmur3Partitioner.instance);
}
 

import org.apache.cassandra.dht.Murmur3Partitioner; //导入依赖的package包/类
public static IPartitioner getDefaultPartitioner()
{
    return Murmur3Partitioner.instance;
}
 

import org.apache.cassandra.dht.Murmur3Partitioner; //导入依赖的package包/类
/**
 * Test creating sstable files using SSTableTxnWriter.
 * @throws IOException
 */
@Test
public void testCreatingSSTableWithTnx() throws IOException {
    final String inputSSTableFullPathFileName = CASS3_DATA_DIR + "keyspace1/bills_compress/mc-6-big-Data.db";

    final Descriptor descriptor = Descriptor.fromFilename(inputSSTableFullPathFileName,
                                                          TestBaseSSTableFunSuite.HADOOP_CONF);
    final CFMetaData inputCFMetaData =
            SSTableUtils.metaDataFromSSTable(inputSSTableFullPathFileName,
                                                    "casspactor",
                                                    "bills_compress",
                                                    Collections.<String>emptyList(),
                                                    Collections.<String>emptyList(),
                                                    TestBaseSSTableFunSuite.HADOOP_CONF);

    final CFMetaData outputCFMetaData = SSTableUtils.createNewCFMetaData(descriptor, inputCFMetaData);
    final SerializationHeader header = new SerializationHeader(true, outputCFMetaData,
        inputCFMetaData.partitionColumns(),
        EncodingStats.NO_STATS);

    final Descriptor outDescriptor = new Descriptor(
        SSTableFormat.Type.BIG.info.getLatestVersion().getVersion(),
        "/tmp",
        "casspactor",
        "bills_compress",
        9,
        SSTableFormat.Type.BIG, TestBaseSSTableFunSuite.HADOOP_CONF);

    final SSTableTxnWriter writer = SSTableTxnWriter.create(outputCFMetaData,
                                                            outDescriptor,
                                                            4,
                                                            -1,
                                                            1,
                                                            header);

    final ColumnDefinition staticCollDef =
        ColumnDefinition.staticDef(inputCFMetaData, ByteBuffer.wrap("balance".getBytes()), Int32Type.instance);
    final ColumnDefinition regCollDef1 =
        ColumnDefinition.regularDef(inputCFMetaData, ByteBuffer.wrap("amount".getBytes()), Int32Type.instance);
    final ColumnDefinition regCollDef2 =
        ColumnDefinition.regularDef(inputCFMetaData, ByteBuffer.wrap("name".getBytes()), UTF8Type.instance);

    final DecoratedKey key = Murmur3Partitioner.instance.decorateKey(ByteBuffer.wrap("user1".getBytes()));
    final long now = System.currentTimeMillis();

    final Row.Builder builder = BTreeRow.sortedBuilder();
    builder.newRow(Clustering.STATIC_CLUSTERING);
    builder.addCell(BufferCell.live(staticCollDef, now, Int32Type.instance.decompose(123)));
    final PartitionUpdate partitionUpdate = PartitionUpdate.singleRowUpdate(inputCFMetaData,
        key, builder.build());
    final Row.Builder builder2 = BTreeRow.sortedBuilder();
    final Clustering clustering2 = new BufferClustering(Int32Type.instance.decompose(10000));
    builder2.newRow(clustering2);
    builder2.addCell(BufferCell.live(regCollDef1, now, Int32Type.instance.decompose(5)));
    builder2.addCell(BufferCell.live(regCollDef2, now, UTF8Type.instance.decompose("minh1")));

    final PartitionUpdate partitionUpdate2 = PartitionUpdate.singleRowUpdate(inputCFMetaData,
        key, builder2.build());

    final List<PartitionUpdate> partitionUpdates = new ArrayList<PartitionUpdate>() {
        private static final long serialVersionUID = 1L;
        {
            add(partitionUpdate);
            add(partitionUpdate2);
        }
    };

    final PartitionUpdate mergedUpdate = PartitionUpdate.merge(partitionUpdates);

    writer.append(mergedUpdate.unfilteredIterator());
    writer.finish(false);
}
 

import org.apache.cassandra.dht.Murmur3Partitioner; //导入依赖的package包/类
public static BufferDecoratedKey readerBounds(int generation)
{
    return new BufferDecoratedKey(new Murmur3Partitioner.LongToken(generation), ByteBufferUtil.EMPTY_BYTE_BUFFER);
}
 

import org.apache.cassandra.dht.Murmur3Partitioner; //导入依赖的package包/类
@Test
public void testParseOptions()
{
    IPartitioner partitioner = Murmur3Partitioner.instance;
    Token.TokenFactory tokenFactory = partitioner.getTokenFactory();

    // parse with empty options
    RepairOption option = RepairOption.parse(new HashMap<String, String>(), partitioner);

    if (FBUtilities.isWindows() && (DatabaseDescriptor.getDiskAccessMode() != Config.DiskAccessMode.standard || DatabaseDescriptor.getIndexAccessMode() != Config.DiskAccessMode.standard))
        assertTrue(option.getParallelism() == RepairParallelism.PARALLEL);
    else
        assertTrue(option.getParallelism() == RepairParallelism.SEQUENTIAL);

    assertFalse(option.isPrimaryRange());
    assertFalse(option.isIncremental());

    // parse everything
    Map<String, String> options = new HashMap<>();
    options.put(RepairOption.PARALLELISM_KEY, "parallel");
    options.put(RepairOption.PRIMARY_RANGE_KEY, "false");
    options.put(RepairOption.INCREMENTAL_KEY, "false");
    options.put(RepairOption.RANGES_KEY, "0:10,11:20,21:30");
    options.put(RepairOption.COLUMNFAMILIES_KEY, "cf1,cf2,cf3");
    options.put(RepairOption.DATACENTERS_KEY, "dc1,dc2,dc3");
    options.put(RepairOption.HOSTS_KEY, "127.0.0.1,127.0.0.2,127.0.0.3");

    option = RepairOption.parse(options, partitioner);
    assertTrue(option.getParallelism() == RepairParallelism.PARALLEL);
    assertFalse(option.isPrimaryRange());
    assertFalse(option.isIncremental());

    Set<Range<Token>> expectedRanges = new HashSet<>(3);
    expectedRanges.add(new Range<>(tokenFactory.fromString("0"), tokenFactory.fromString("10")));
    expectedRanges.add(new Range<>(tokenFactory.fromString("11"), tokenFactory.fromString("20")));
    expectedRanges.add(new Range<>(tokenFactory.fromString("21"), tokenFactory.fromString("30")));
    assertEquals(expectedRanges, option.getRanges());

    Set<String> expectedCFs = new HashSet<>(3);
    expectedCFs.add("cf1");
    expectedCFs.add("cf2");
    expectedCFs.add("cf3");
    assertEquals(expectedCFs, option.getColumnFamilies());

    Set<String> expectedDCs = new HashSet<>(3);
    expectedDCs.add("dc1");
    expectedDCs.add("dc2");
    expectedDCs.add("dc3");
    assertEquals(expectedDCs, option.getDataCenters());

    Set<String> expectedHosts = new HashSet<>(3);
    expectedHosts.add("127.0.0.1");
    expectedHosts.add("127.0.0.2");
    expectedHosts.add("127.0.0.3");
    assertEquals(expectedHosts, option.getHosts());
}