natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'cassandra-3.11' into trunk

This commit is contained in:
Jason Brown 2017-09-29 05:24:27 -07:00
parent 289f6b157a 2a24acfa99
commit 6b4c8a9733
3 changed files with 225 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -153,6 +153,7 @@
* Duplicate the buffer before passing it to analyser in SASI operation (CASSANDRA-13512)
* Properly evict pstmts from prepared statements cache (CASSANDRA-13641)
Merged from 3.0:
* AssertionError prepending to a list (CASSANDRA-13149)
* Handle limit correctly on tables with strict liveness (CASSANDRA-13883)
* Remove non-rpc-ready nodes from counter leader candidates (CASSANDRA-13043)
* Improve short read protection performance (CASSANDRA-13794)

75
src/java/org/apache/cassandra/cql3/Lists.java
View File

@ -29,6 +29,7 @@ import java.util.stream.Collectors;
import java.util.stream.StreamSupport;
import org.apache.cassandra.schema.ColumnMetadata;
import com.google.common.annotations.VisibleForTesting;
import org.apache.cassandra.cql3.functions.Function;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.rows.*;
@ -305,18 +306,17 @@ public abstract class Lists
}
}
/*
/**
* For prepend, we need to be able to generate unique but decreasing time
* UUID, which is a bit challenging. To do that, given a time in milliseconds,
* we adds a number representing the 100-nanoseconds precision and make sure
* that within the same millisecond, that number is always decreasing. We
* do rely on the fact that the user will only provide decreasing
* milliseconds timestamp for that purpose.
* UUIDs, which is a bit challenging. To do that, given a time in milliseconds,
* we add a number representing the 100-nanoseconds precision and make sure
* that within the same millisecond, that number is always decreasing.
*/
private static class PrecisionTime
static class PrecisionTime
{
// Our reference time (1 jan 2010, 00:00:00) in milliseconds.
private static final long REFERENCE_TIME = 1262304000000L;
static final int MAX_NANOS = 9999;
private static final AtomicReference<PrecisionTime> last = new AtomicReference<>(new PrecisionTime(Long.MAX_VALUE, 0));
public final long millis;
@ -328,21 +328,52 @@ public abstract class Lists
this.nanos = nanos;
}
static PrecisionTime getNext(long millis)
static PrecisionTime getNext(long millis, int count)
{
if (count == 0)
return last.get();
while (true)
{
PrecisionTime current = last.get();
assert millis <= current.millis;
PrecisionTime next = millis < current.millis
? new PrecisionTime(millis, 9999)
: new PrecisionTime(millis, Math.max(0, current.nanos - 1));
final PrecisionTime next;
if (millis < current.millis)
{
next = new PrecisionTime(millis, MAX_NANOS - count);
}
else
{
// in addition to being at the same millisecond, we handle the unexpected case of the millis parameter
// being in the past. That could happen if the System.currentTimeMillis() not operating montonically
// or if one thread is just a really big loser in the compareAndSet game of life.
long millisToUse = millis <= current.millis ? millis : current.millis;
// if we will go below zero on the nanos, decrement the millis by one
final int nanosToUse;
if (current.nanos - count >= 0)
{
nanosToUse = current.nanos - count;
}
else
{
nanosToUse = MAX_NANOS - count;
millisToUse -= 1;
}
next = new PrecisionTime(millisToUse, nanosToUse);
}
if (last.compareAndSet(current, next))
return next;
}
}
@VisibleForTesting
static void set(long millis, int nanos)
{
last.set(new PrecisionTime(millis, nanos));
}
}
public static class Setter extends Operation
@ -480,13 +511,23 @@ public abstract class Lists
if (value == null || value == UNSET_VALUE)
return;
long time = PrecisionTime.REFERENCE_TIME - (System.currentTimeMillis() - PrecisionTime.REFERENCE_TIME);
List<ByteBuffer> toAdd = ((Value) value).elements;
for (int i = toAdd.size() - 1; i >= 0; i--)
final int totalCount = toAdd.size();
// we have to obey MAX_NANOS per batch - in the unlikely event a client has decided to prepend a list with
// an insane number of entries.
PrecisionTime pt = null;
int remainingInBatch = 0;
for (int i = totalCount - 1; i >= 0; i--)
{
PrecisionTime pt = PrecisionTime.getNext(time);
ByteBuffer uuid = ByteBuffer.wrap(UUIDGen.getTimeUUIDBytes(pt.millis, pt.nanos));
if (remainingInBatch == 0)
{
long time = PrecisionTime.REFERENCE_TIME - (System.currentTimeMillis() - PrecisionTime.REFERENCE_TIME);
remainingInBatch = Math.min(PrecisionTime.MAX_NANOS, i) + 1;
pt = PrecisionTime.getNext(time, remainingInBatch);
}
ByteBuffer uuid = ByteBuffer.wrap(UUIDGen.getTimeUUIDBytes(pt.millis, (pt.nanos + remainingInBatch--)));
params.addCell(column, CellPath.create(uuid), toAdd.get(i));
}
}

166
test/unit/org/apache/cassandra/cql3/ListsTest.java Normal file
View File

@ -0,0 +1,166 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.cql3;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.UUID;
import com.google.common.collect.Iterators;
import org.junit.Assert;
import org.junit.Test;
import org.apache.cassandra.cql3.Lists.PrecisionTime;
import org.apache.cassandra.db.Clustering;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.rows.Cell;
import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.UUIDGen;
public class ListsTest extends CQLTester
{
private static final long DEFAULT_MILLIS = 424242424242L;
private static final int DEFAULT_NANOS = PrecisionTime.MAX_NANOS;
@Test
public void testPrecisionTime_getNext_simple()
{
PrecisionTime.set(DEFAULT_MILLIS, DEFAULT_NANOS);
long millis = DEFAULT_MILLIS - 100;
int count = 1;
PrecisionTime next = PrecisionTime.getNext(millis, count);
Assert.assertEquals(millis, next.millis);
Assert.assertEquals(DEFAULT_NANOS - count, next.nanos);
next = PrecisionTime.getNext(millis, count);
Assert.assertEquals(millis, next.millis);
Assert.assertEquals(DEFAULT_NANOS - (count * 2), next.nanos);
}
@Test
public void testPrecisionTime_getNext_Mulitple()
{
PrecisionTime.set(DEFAULT_MILLIS, DEFAULT_NANOS);
long millis = DEFAULT_MILLIS - 100;
int count = DEFAULT_NANOS / 2;
PrecisionTime next = PrecisionTime.getNext(millis, count);
Assert.assertEquals(millis, next.millis);
Assert.assertEquals(DEFAULT_NANOS - count, next.nanos);
}
@Test
public void testPrecisionTime_getNext_RollOverNanos()
{
final int remainingNanos = 0;
PrecisionTime.set(DEFAULT_MILLIS, remainingNanos);
long millis = DEFAULT_MILLIS;
int count = 1;
PrecisionTime next = PrecisionTime.getNext(millis, count);
Assert.assertEquals(millis - 1, next.millis);
Assert.assertEquals(DEFAULT_NANOS - count, next.nanos);
next = PrecisionTime.getNext(millis, count);
Assert.assertEquals(millis - 1, next.millis);
Assert.assertEquals(DEFAULT_NANOS - (count * 2), next.nanos);
}
@Test
public void testPrecisionTime_getNext_BorkedClock()
{
final int remainingNanos = 1;
PrecisionTime.set(DEFAULT_MILLIS, remainingNanos);
long millis = DEFAULT_MILLIS + 100;
int count = 1;
PrecisionTime next = PrecisionTime.getNext(millis, count);
Assert.assertEquals(DEFAULT_MILLIS, next.millis);
Assert.assertEquals(remainingNanos - count, next.nanos);
// this should roll the clock
next = PrecisionTime.getNext(millis, count);
Assert.assertEquals(DEFAULT_MILLIS - 1, next.millis);
Assert.assertEquals(DEFAULT_NANOS - count, next.nanos);
}
@Test
public void testPrepender_SmallList()
{
List<ByteBuffer> terms = new ArrayList<>();
terms.add(ByteBufferUtil.bytes(1));
terms.add(ByteBufferUtil.bytes(2));
terms.add(ByteBufferUtil.bytes(3));
terms.add(ByteBufferUtil.bytes(4));
terms.add(ByteBufferUtil.bytes(5));
testPrepender_execute(terms);
}
@Test
public void testPrepender_HugeList()
{
List<ByteBuffer> terms = new ArrayList<>();
// create a large enough array, then remove some off the end, just to make it an odd size
for (int i = 0; i < PrecisionTime.MAX_NANOS * 4 - 287; i++)
terms.add(ByteBufferUtil.bytes(i));
testPrepender_execute(terms);
}
private void testPrepender_execute(List<ByteBuffer> terms)
{
createTable("CREATE TABLE %s (k int PRIMARY KEY, l list<text>)");
TableMetadata metaData = currentTableMetadata();
ColumnMetadata columnMetadata = metaData.getColumn(ByteBufferUtil.bytes("l"));
Term term = new Lists.Value(terms);
Lists.Prepender prepender = new Lists.Prepender(columnMetadata, term);
ByteBuffer keyBuf = ByteBufferUtil.bytes("key");
DecoratedKey key = Murmur3Partitioner.instance.decorateKey(keyBuf);
UpdateParameters parameters = new UpdateParameters(metaData, null, null, System.currentTimeMillis(), 1000, Collections.emptyMap());
Clustering clustering = Clustering.make(ByteBufferUtil.bytes(1));
parameters.newRow(clustering);
prepender.execute(key, parameters);
Row row = parameters.buildRow();
Assert.assertEquals(terms.size(), Iterators.size(row.cells().iterator()));
int idx = 0;
UUID last = null;
for (Cell cell : row.cells())
{
UUID uuid = UUIDGen.getUUID(cell.path().get(0));
if (last != null)
Assert.assertTrue(last.compareTo(uuid) < 0);
last = uuid;
Assert.assertEquals(String.format("different values found: expected: '%d', found '%d'", ByteBufferUtil.toInt(terms.get(idx)), ByteBufferUtil.toInt(cell.value())),
terms.get(idx), cell.value());
idx++;
}
}
}