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

fix isDeterministic + CqlReader

This commit is contained in:
belliottsmith 2014-02-13 16:39:16 +00:00
parent 84f2b8908e 79c6ecccc0
commit c869087267
34 changed files with 122 additions and 338 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
CHANGES.txt
View File

@ -25,13 +25,12 @@
* CF id is changed to be non-deterministic. Data dir/key cache are created
uniquely for CF id (CASSANDRA-5202)
* New counters implementation (CASSANDRA-6504)
* Replace UnsortedColumns and TreeMapBackedSortedColumns with rewritten
ArrayBackedSortedColumns (CASSANDRA-6630, CASSANDRA-6662)
* Replace UnsortedColumns, EmptyColumns, TreeMapBackedSortedColumns with new
ArrayBackedSortedColumns (CASSANDRA-6630, CASSANDRA-6662, CASSANDRA-6690)
* Add option to use row cache with a given amount of rows (CASSANDRA-5357)
* Avoid repairing already repaired data (CASSANDRA-5351)
* Reject counter updates with USING TTL/TIMESTAMP (CASSANDRA-6649)
2.0.6
* Add compatibility for Hadoop 0.2.x (CASSANDRA-5201)
* Fix EstimatedHistogram races (CASSANDRA-6682)
@ -42,6 +41,9 @@
* Correctly handle null with IF conditions and TTL (CASSANDRA-6623)
* Account for range/row tombstones in tombstone drop
time histogram (CASSANDRA-6522)
* Stop CommitLogSegment.close() from calling sync() (CASSANDRA-6652)
* Make commitlog failure handling configurable (CASSANDRA-6364)
* Avoid overlaps in LCS (CASSANDRA-6688)
Merged from 1.2:
* Fix broken streams when replacing with same IP (CASSANDRA-6622)
* Fix upgradesstables NPE for non-CF-based indexes (CASSANDRA-6645)
@ -49,6 +51,7 @@ Merged from 1.2:
* Fix mean cells and mean row size per sstable calculations (CASSANDRA-6667)
* Compact hints after partial replay to clean out tombstones (CASSANDRA-6666)
* Log USING TTL/TIMESTAMP in a counter update warning (CASSANDRA-6649)
* Don't exchange schema between nodes with different versions (CASSANDRA-6695)
2.0.5

70
src/java/org/apache/cassandra/db/AbstractThreadUnsafeSortedColumns.java
View File

@ -1,70 +0,0 @@
/*
* 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.db;
import org.apache.cassandra.config.CFMetaData;
public abstract class AbstractThreadUnsafeSortedColumns extends ColumnFamily
{
protected DeletionInfo deletionInfo;
public AbstractThreadUnsafeSortedColumns(CFMetaData metadata)
{
this(metadata, DeletionInfo.live());
}
protected AbstractThreadUnsafeSortedColumns(CFMetaData metadata, DeletionInfo deletionInfo)
{
super(metadata);
this.deletionInfo = deletionInfo;
}
public DeletionInfo deletionInfo()
{
return deletionInfo;
}
public void delete(DeletionTime delTime)
{
deletionInfo.add(delTime);
}
public void delete(DeletionInfo newInfo)
{
deletionInfo.add(newInfo);
}
protected void delete(RangeTombstone tombstone)
{
deletionInfo.add(tombstone, getComparator());
}
public void setDeletionInfo(DeletionInfo newInfo)
{
deletionInfo = newInfo;
}
/**
* Purges any tombstones with a local deletion time before gcBefore.
* @param gcBefore a timestamp (in seconds) before which tombstones should be purged
*/
public void purgeTombstones(int gcBefore)
{
deletionInfo.purge(gcBefore);
}
}

50
src/java/org/apache/cassandra/db/ArrayBackedSortedColumns.java
View File

@ -37,13 +37,13 @@ import org.apache.cassandra.db.filter.ColumnSlice;
* main operations performed are iterating over the cells and adding cells
* (especially if insertion is in sorted order).
*/
public class ArrayBackedSortedColumns extends AbstractThreadUnsafeSortedColumns
public class ArrayBackedSortedColumns extends ColumnFamily
{
private static final int INITIAL_CAPACITY = 10;
private static final Cell[] EMPTY_ARRAY = new Cell[0];
private static final int MINIMAL_CAPACITY = 10;
private final boolean reversed;
private DeletionInfo deletionInfo;
private Cell[] cells;
private int size;
@ -61,7 +61,8 @@ public class ArrayBackedSortedColumns extends AbstractThreadUnsafeSortedColumns
{
super(metadata);
this.reversed = reversed;
this.cells = new Cell[INITIAL_CAPACITY];
this.deletionInfo = DeletionInfo.live();
this.cells = EMPTY_ARRAY;
this.size = 0;
this.sortedSize = 0;
}
@ -70,6 +71,7 @@ public class ArrayBackedSortedColumns extends AbstractThreadUnsafeSortedColumns
{
super(original.metadata);
this.reversed = original.reversed;
this.deletionInfo = DeletionInfo.live(); // this is INTENTIONALLY not set to original.deletionInfo.
this.cells = Arrays.copyOf(original.cells, original.size);
this.size = original.size;
this.sortedSize = original.sortedSize;
@ -211,9 +213,11 @@ public class ArrayBackedSortedColumns extends AbstractThreadUnsafeSortedColumns
*/
private void internalAdd(Cell cell)
{
// Resize the backing array if we hit the capacity
if (cells.length == size)
if (cells == EMPTY_ARRAY)
cells = new Cell[MINIMAL_CAPACITY];
else if (cells.length == size)
cells = Arrays.copyOf(cells, size * 3 / 2 + 1);
cells[size++] = cell;
}
@ -318,6 +322,40 @@ public class ArrayBackedSortedColumns extends AbstractThreadUnsafeSortedColumns
size = sortedSize = 0;
}
public DeletionInfo deletionInfo()
{
return deletionInfo;
}
public void delete(DeletionTime delTime)
{
deletionInfo.add(delTime);
}
public void delete(DeletionInfo newInfo)
{
deletionInfo.add(newInfo);
}
protected void delete(RangeTombstone tombstone)
{
deletionInfo.add(tombstone, getComparator());
}
public void setDeletionInfo(DeletionInfo newInfo)
{
deletionInfo = newInfo;
}
/**
* Purges any tombstones with a local deletion time before gcBefore.
* @param gcBefore a timestamp (in seconds) before which tombstones should be purged
*/
public void purgeTombstones(int gcBefore)
{
deletionInfo.purge(gcBefore);
}
public Iterable<CellName> getColumnNames()
{
maybeSortCells();

8
src/java/org/apache/cassandra/db/ColumnFamily.java
View File

@ -389,14 +389,6 @@ public abstract class ColumnFamily implements Iterable<Cell>, IRowCacheEntry
return cf1.diff(cf2);
}
public void resolve(ColumnFamily cf)
{
// Row _does_ allow null CF objects :( seems a necessary evil for efficiency
if (cf == null)
return;
addAll(cf);
}
public ColumnStats getColumnStats()
{
long minTimestampSeen = deletionInfo().isLive() ? Long.MAX_VALUE : deletionInfo().minTimestamp();

112
src/java/org/apache/cassandra/db/EmptyColumns.java
View File

@ -1,112 +0,0 @@
/*
*
* 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.db;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import com.google.common.collect.Iterators;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.db.composites.CellName;
import org.apache.cassandra.db.filter.ColumnSlice;
public class EmptyColumns extends AbstractThreadUnsafeSortedColumns
{
public static final Factory<EmptyColumns> factory = new Factory<EmptyColumns>()
{
public EmptyColumns create(CFMetaData metadata, boolean insertReversed)
{
assert !insertReversed;
return new EmptyColumns(metadata, DeletionInfo.live());
}
};
public EmptyColumns(CFMetaData metadata, DeletionInfo info)
{
super(metadata, info);
}
public ColumnFamily cloneMe()
{
return new EmptyColumns(metadata, deletionInfo);
}
public void clear()
{
}
public Factory<EmptyColumns> getFactory()
{
return factory;
}
public void addColumn(Cell cell)
{
throw new UnsupportedOperationException();
}
public void addAll(ColumnFamily cm)
{
throw new UnsupportedOperationException();
}
public Cell getColumn(CellName name)
{
throw new UnsupportedOperationException();
}
public Iterable<CellName> getColumnNames()
{
return Collections.emptyList();
}
public Collection<Cell> getSortedColumns()
{
return Collections.emptyList();
}
public Collection<Cell> getReverseSortedColumns()
{
return Collections.emptyList();
}
public int getColumnCount()
{
return 0;
}
public Iterator<Cell> iterator(ColumnSlice[] slices)
{
return Iterators.emptyIterator();
}
public Iterator<Cell> reverseIterator(ColumnSlice[] slices)
{
return Iterators.emptyIterator();
}
public boolean isInsertReversed()
{
return false;
}
}

2
src/java/org/apache/cassandra/db/Mutation.java
View File

@ -191,7 +191,7 @@ public class Mutation implements IMutation
// not in the case where it wasn't there indeed.
ColumnFamily cf = modifications.put(entry.getKey(), entry.getValue());
if (cf != null)
entry.getValue().resolve(cf);
entry.getValue().addAll(cf);
}
}

2
src/java/org/apache/cassandra/db/SystemKeyspace.java
View File

@ -810,7 +810,7 @@ public class SystemKeyspace
return new PaxosState(key, metadata);
UntypedResultSet.Row row = results.one();
Commit promised = row.has("in_progress_ballot")
? new Commit(key, row.getUUID("in_progress_ballot"), EmptyColumns.factory.create(metadata))
? new Commit(key, row.getUUID("in_progress_ballot"), ArrayBackedSortedColumns.factory.create(metadata))
: Commit.emptyCommit(key, metadata);
// either we have both a recently accepted ballot and update or we have neither
Commit accepted = row.has("proposal")

2
src/java/org/apache/cassandra/db/columniterator/IndexedSliceReader.java
View File

@ -77,7 +77,7 @@ class IndexedSliceReader extends AbstractIterator<OnDiskAtom> implements OnDiskA
try
{
this.indexes = indexEntry.columnsIndex();
emptyColumnFamily = EmptyColumns.factory.create(sstable.metadata);
emptyColumnFamily = ArrayBackedSortedColumns.factory.create(sstable.metadata);
if (indexes.isEmpty())
{
setToRowStart(indexEntry, input);

2
src/java/org/apache/cassandra/db/columniterator/SimpleSliceReader.java
View File

@ -71,7 +71,7 @@ class SimpleSliceReader extends AbstractIterator<OnDiskAtom> implements OnDiskAt
if (version.hasRowSizeAndColumnCount)
file.readLong();
emptyColumnFamily = EmptyColumns.factory.create(sstable.metadata);
emptyColumnFamily = ArrayBackedSortedColumns.factory.create(sstable.metadata);
emptyColumnFamily.delete(DeletionTime.serializer.deserialize(file));
int columnCount = version.hasRowSizeAndColumnCount ? file.readInt() : Integer.MAX_VALUE;
atomIterator = emptyColumnFamily.metadata().getOnDiskIterator(file, columnCount, sstable.descriptor.version);

2
src/java/org/apache/cassandra/db/compaction/LazilyCompactedRow.java
View File

@ -81,7 +81,7 @@ public class LazilyCompactedRow extends AbstractCompactedRow
// containing `key` outside of the set of sstables involved in this compaction.
maxPurgeableTimestamp = controller.maxPurgeableTimestamp(key);
emptyColumnFamily = EmptyColumns.factory.create(controller.cfs.metadata);
emptyColumnFamily = ArrayBackedSortedColumns.factory.create(controller.cfs.metadata);
emptyColumnFamily.delete(maxRowTombstone);
if (maxRowTombstone.markedForDeleteAt < maxPurgeableTimestamp)
emptyColumnFamily.purgeTombstones(controller.gcBefore);

23
src/java/org/apache/cassandra/db/compaction/LeveledManifest.java
View File

@ -173,23 +173,6 @@ public class LeveledManifest
}
}
/**
* if the number of SSTables in the current compacted set *by itself* exceeds the target level's
* (regardless of the level's current contents), find an empty level instead
*/
private int skipLevels(int newLevel, Iterable<SSTableReader> added)
{
// Note that we now check if the sstables included in the compaction, *before* the compaction, fit in the next level.
// This is needed since we need to decide before the actual compaction what level they will be in.
// This should be safe, we might skip levels where the compacted data could have fit but that should be ok.
while (maxBytesForLevel(newLevel) < SSTableReader.getTotalBytes(added)
&& getLevel(newLevel + 1).isEmpty())
{
newLevel++;
}
return newLevel;
}
public synchronized void replace(Collection<SSTableReader> removed, Collection<SSTableReader> added)
{
assert !removed.isEmpty(); // use add() instead of promote when adding new sstables
@ -560,7 +543,10 @@ public class LeveledManifest
for (SSTableReader newCandidate : overlappedL0)
{
candidates.add(newCandidate);
// overlappedL0 could contain sstables that are not in compactingL0, but do overlap
// other sstables that are
if (overlapping(newCandidate, compactingL0).isEmpty())
candidates.add(newCandidate);
remaining.remove(newCandidate);
}
@ -683,7 +669,6 @@ public class LeveledManifest
else
{
newLevel = minimumLevel == maximumLevel ? maximumLevel + 1 : maximumLevel;
newLevel = skipLevels(newLevel, sstables);
assert newLevel > 0;
}
return newLevel;

49
src/java/org/apache/cassandra/db/filter/ColumnSlice.java
View File

@ -138,55 +138,6 @@ public class ColumnSlice
}
}
public static class NavigableMapIterator extends AbstractIterator<Cell>
{
private final NavigableMap<CellName, Cell> map;
private final ColumnSlice[] slices;
private int idx = 0;
private Iterator<Cell> currentSlice;
public NavigableMapIterator(NavigableMap<CellName, Cell> map, ColumnSlice[] slices)
{
this.map = map;
this.slices = slices;
}
protected Cell computeNext()
{
if (currentSlice == null)
{
if (idx >= slices.length)
return endOfData();
ColumnSlice slice = slices[idx++];
// Note: we specialize the case of start == "" and finish = "" because it is slightly more efficient, but also they have a specific
// meaning (namely, they always extend to the beginning/end of the range).
if (slice.start.isEmpty())
{
if (slice.finish.isEmpty())
currentSlice = map.values().iterator();
else
currentSlice = map.headMap(new FakeCellName(slice.finish), true).values().iterator();
}
else if (slice.finish.isEmpty())
{
currentSlice = map.tailMap(new FakeCellName(slice.start), true).values().iterator();
}
else
{
currentSlice = map.subMap(new FakeCellName(slice.start), true, new FakeCellName(slice.finish), true).values().iterator();
}
}
if (currentSlice.hasNext())
return currentSlice.next();
currentSlice = null;
return computeNext();
}
}
public static class NavigableSetIterator extends AbstractIterator<Cell>
{
private final NavigableSet<Cell> set;

2
src/java/org/apache/cassandra/db/index/composites/CompositesSearcher.java
View File

@ -274,7 +274,7 @@ public class CompositesSearcher extends SecondaryIndexSearcher
if (data == null)
data = ArrayBackedSortedColumns.factory.create(baseCfs.metadata);
data.resolve(newData);
data.addAll(newData);
columnsCount += dataFilter.lastCounted();
}
}

2
src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java
View File

@ -99,7 +99,7 @@ public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterat
try
{
columnFamily = EmptyColumns.factory.create(metadata);
columnFamily = ArrayBackedSortedColumns.factory.create(metadata);
columnFamily.delete(DeletionTime.serializer.deserialize(in));
columnCount = dataVersion.hasRowSizeAndColumnCount ? in.readInt() : Integer.MAX_VALUE;
atomIterator = columnFamily.metadata().getOnDiskIterator(in, columnCount, dataVersion);

7
src/java/org/apache/cassandra/service/CassandraDaemon.java
View File

@ -56,6 +56,7 @@ import org.apache.cassandra.metrics.StorageMetrics;
import org.apache.cassandra.thrift.ThriftServer;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.CLibrary;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Mx4jTool;
import org.apache.cassandra.utils.Pair;
@ -331,7 +332,8 @@ public class CassandraDaemon
}
}
waitForGossipToSettle();
if (!FBUtilities.getBroadcastAddress().equals(InetAddress.getLoopbackAddress()))
waitForGossipToSettle();
// Thift
InetAddress rpcAddr = DatabaseDescriptor.getRpcAddress();
@ -458,7 +460,6 @@ public class CassandraDaemon
destroy();
}
private void waitForGossipToSettle()
{
int forceAfter = Integer.getInteger("cassandra.skip_wait_for_gossip_to_settle", -1);
@ -502,7 +503,7 @@ public class CassandraDaemon
if (totalPolls > GOSSIP_SETTLE_POLL_SUCCESSES_REQUIRED)
logger.info("Gossip settled after {} extra polls; proceeding", totalPolls - GOSSIP_SETTLE_POLL_SUCCESSES_REQUIRED);
else
logger.debug("Gossip settled after {} extra polls; proceeding", totalPolls - GOSSIP_SETTLE_POLL_SUCCESSES_REQUIRED);
logger.info("No gossip backlog; proceeding");
}
public static void stop(String[] args)

18
src/java/org/apache/cassandra/service/MigrationManager.java
View File

@ -132,11 +132,11 @@ public class MigrationManager
private static boolean shouldPullSchemaFrom(InetAddress endpoint)
{
/*
* Don't request schema from nodes with a higher major (may have incompatible schema)
* Don't request schema from nodes with a differnt or unknonw major version (may have incompatible schema)
* Don't request schema from fat clients
*/
return MessagingService.instance().knowsVersion(endpoint)
&& MessagingService.instance().getVersion(endpoint) <= MessagingService.current_version
&& MessagingService.instance().getVersion(endpoint) == MessagingService.current_version
&& !Gossiper.instance.isFatClient(endpoint);
}
@ -302,15 +302,13 @@ public class MigrationManager
for (InetAddress endpoint : Gossiper.instance.getLiveMembers())
{
if (endpoint.equals(FBUtilities.getBroadcastAddress()))
continue; // we've dealt with localhost already
// don't send schema to the nodes with the versions older than current major
if (MessagingService.instance().getVersion(endpoint) < MessagingService.current_version)
continue;
pushSchemaMutation(endpoint, schema);
// only push schema to nodes with known and equal versions
if (!endpoint.equals(FBUtilities.getBroadcastAddress()) &&
MessagingService.instance().knowsVersion(endpoint) &&
MessagingService.instance().getVersion(endpoint) == MessagingService.current_version)
pushSchemaMutation(endpoint, schema);
}
return f;
}

4
src/java/org/apache/cassandra/service/ReadCallback.java
View File

@ -98,10 +98,8 @@ public class ReadCallback<TMessage, TResolved> implements IAsyncCallback<TMessag
if (!await(command.getTimeout(), TimeUnit.MILLISECONDS))
{
// Same as for writes, see AbstractWriteResponseHandler
int acks = received;
if (resolver.isDataPresent() && acks >= blockfor)
acks = blockfor - 1;
ReadTimeoutException ex = new ReadTimeoutException(consistencyLevel, received, blockfor, resolver.isDataPresent());
if (logger.isDebugEnabled())
logger.debug("Read timeout: {}", ex.toString());
throw ex;

4
src/java/org/apache/cassandra/service/StorageProxy.java
View File

@ -232,7 +232,7 @@ public class StorageProxy implements StorageProxyMBean
{
Tracing.trace("CAS precondition {} does not match current values {}", conditions, current);
// We should not return null as this means success
return current == null ? EmptyColumns.factory.create(metadata) : current;
return current == null ? ArrayBackedSortedColumns.factory.create(metadata) : current;
}
// finish the paxos round w/ the desired updates
@ -603,7 +603,7 @@ public class StorageProxy implements StorageProxyMBean
private static void asyncRemoveFromBatchlog(Collection<InetAddress> endpoints, UUID uuid)
{
ColumnFamily cf = EmptyColumns.factory.create(Schema.instance.getCFMetaData(Keyspace.SYSTEM_KS, SystemKeyspace.BATCHLOG_CF));
ColumnFamily cf = ArrayBackedSortedColumns.factory.create(Schema.instance.getCFMetaData(Keyspace.SYSTEM_KS, SystemKeyspace.BATCHLOG_CF));
cf.delete(new DeletionInfo(FBUtilities.timestampMicros(), (int) (System.currentTimeMillis() / 1000)));
AbstractWriteResponseHandler handler = new WriteResponseHandler(endpoints,
Collections.<InetAddress>emptyList(),

1
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -843,7 +843,6 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
if (!current.isEmpty())
for (InetAddress existing : current)
Gossiper.instance.replacedEndpoint(existing);
logger.info("Startup completed! Now serving reads.");
assert tokenMetadata.sortedTokens().size() > 0;
Auth.setup();

2
src/java/org/apache/cassandra/service/pager/QueryPagers.java
View File

@ -147,7 +147,7 @@ public class QueryPagers
{
List<Row> rows = pager.fetchPage(pageSize);
ColumnFamily cf = rows.isEmpty() ? null : rows.get(0).cf;
return cf == null ? EmptyColumns.factory.create(cfs.metadata) : cf;
return cf == null ? ArrayBackedSortedColumns.factory.create(cfs.metadata) : cf;
}
catch (Exception e)
{

4
src/java/org/apache/cassandra/service/paxos/Commit.java
View File

@ -57,7 +57,7 @@ public class Commit
public static Commit newPrepare(ByteBuffer key, CFMetaData metadata, UUID ballot)
{
return new Commit(key, ballot, EmptyColumns.factory.create(metadata));
return new Commit(key, ballot, ArrayBackedSortedColumns.factory.create(metadata));
}
public static Commit newProposal(ByteBuffer key, UUID ballot, ColumnFamily update)
@ -67,7 +67,7 @@ public class Commit
public static Commit emptyCommit(ByteBuffer key, CFMetaData metadata)
{
return new Commit(key, UUIDGen.minTimeUUID(0), EmptyColumns.factory.create(metadata));
return new Commit(key, UUIDGen.minTimeUUID(0), ArrayBackedSortedColumns.factory.create(metadata));
}
public boolean isAfter(Commit other)

6
test/unit/org/apache/cassandra/db/RangeTombstoneListTest.java
View File

@ -195,12 +195,6 @@ public class RangeTombstoneListTest
@Test
public void addAllTest()
{
//addAllTest(false);
addAllTest(true);
}
private void addAllTest(boolean doMerge)
{
RangeTombstoneList l1 = new RangeTombstoneList(cmp, 0);
l1.add(rt(0, 4, 5));

2
test/unit/org/apache/cassandra/db/RowTest.java
View File

@ -59,7 +59,7 @@ public class RowTest extends SchemaLoader
cf2.addColumn(column("one", "B", 1));
cf2.addColumn(column("two", "C", 1));
cf1.resolve(cf2);
cf1.addAll(cf2);
assert Arrays.equals(cf1.getColumn(CellNames.simpleDense(ByteBufferUtil.bytes("one"))).value().array(), "B".getBytes());
assert Arrays.equals(cf1.getColumn(CellNames.simpleDense(ByteBufferUtil.bytes("two"))).value().array(), "C".getBytes());
}

12
test/unit/org/apache/cassandra/locator/OldNetworkTopologyStrategyTest.java
View File

@ -181,7 +181,7 @@ public class OldNetworkTopologyStrategyTest extends SchemaLoader
BigIntegerToken newToken = new BigIntegerToken("21267647932558653966460912964485513216");
BigIntegerToken[] tokens = initTokens();
BigIntegerToken[] tokensAfterMove = initTokensAfterMove(tokens, movingNodeIdx, newToken);
Pair<Set<Range<Token>>, Set<Range<Token>>> ranges = calculateStreamAndFetchRanges(tokens, tokensAfterMove, movingNodeIdx, newToken);
Pair<Set<Range<Token>>, Set<Range<Token>>> ranges = calculateStreamAndFetchRanges(tokens, tokensAfterMove, movingNodeIdx);
assertEquals(ranges.left.iterator().next().left, tokensAfterMove[movingNodeIdx]);
assertEquals(ranges.left.iterator().next().right, tokens[movingNodeIdx]);
@ -198,7 +198,7 @@ public class OldNetworkTopologyStrategyTest extends SchemaLoader
BigIntegerToken newToken = new BigIntegerToken("35267647932558653966460912964485513216");
BigIntegerToken[] tokens = initTokens();
BigIntegerToken[] tokensAfterMove = initTokensAfterMove(tokens, movingNodeIdx, newToken);
Pair<Set<Range<Token>>, Set<Range<Token>>> ranges = calculateStreamAndFetchRanges(tokens, tokensAfterMove, movingNodeIdx, newToken);
Pair<Set<Range<Token>>, Set<Range<Token>>> ranges = calculateStreamAndFetchRanges(tokens, tokensAfterMove, movingNodeIdx);
assertEquals("No data should be streamed", ranges.left.size(), 0);
assertEquals(ranges.right.iterator().next().left, tokens[movingNodeIdx]);
@ -216,7 +216,7 @@ public class OldNetworkTopologyStrategyTest extends SchemaLoader
BigIntegerToken newToken = new BigIntegerToken("90070591730234615865843651857942052864");
BigIntegerToken[] tokens = initTokens();
BigIntegerToken[] tokensAfterMove = initTokensAfterMove(tokens, movingNodeIdx, newToken);
Pair<Set<Range<Token>>, Set<Range<Token>>> ranges = calculateStreamAndFetchRanges(tokens, tokensAfterMove, movingNodeIdx, newToken);
Pair<Set<Range<Token>>, Set<Range<Token>>> ranges = calculateStreamAndFetchRanges(tokens, tokensAfterMove, movingNodeIdx);
// sort the results, so they can be compared
Range[] toStream = ranges.left.toArray(new Range[0]);
@ -248,7 +248,7 @@ public class OldNetworkTopologyStrategyTest extends SchemaLoader
BigIntegerToken newToken = new BigIntegerToken("52535295865117307932921825928971026432");
BigIntegerToken[] tokens = initTokens();
BigIntegerToken[] tokensAfterMove = initTokensAfterMove(tokens, movingNodeIdx, newToken);
Pair<Set<Range<Token>>, Set<Range<Token>>> ranges = calculateStreamAndFetchRanges(tokens, tokensAfterMove, movingNodeIdx, newToken);
Pair<Set<Range<Token>>, Set<Range<Token>>> ranges = calculateStreamAndFetchRanges(tokens, tokensAfterMove, movingNodeIdx);
// sort the results, so they can be compared
@ -280,7 +280,7 @@ public class OldNetworkTopologyStrategyTest extends SchemaLoader
BigIntegerToken newToken = new BigIntegerToken("158873535527910577765226390751398592512");
BigIntegerToken[] tokens = initTokens();
BigIntegerToken[] tokensAfterMove = initTokensAfterMove(tokens, movingNodeIdx, newToken);
Pair<Set<Range<Token>>, Set<Range<Token>>> ranges = calculateStreamAndFetchRanges(tokens, tokensAfterMove, movingNodeIdx, newToken);
Pair<Set<Range<Token>>, Set<Range<Token>>> ranges = calculateStreamAndFetchRanges(tokens, tokensAfterMove, movingNodeIdx);
Range[] toStream = ranges.left.toArray(new Range[0]);
Range[] toFetch = ranges.right.toArray(new Range[0]);
@ -350,7 +350,7 @@ public class OldNetworkTopologyStrategyTest extends SchemaLoader
}
private Pair<Set<Range<Token>>, Set<Range<Token>>> calculateStreamAndFetchRanges(BigIntegerToken[] tokens, BigIntegerToken[] tokensAfterMove, int movingNodeIdx, BigIntegerToken newToken) throws UnknownHostException
private Pair<Set<Range<Token>>, Set<Range<Token>>> calculateStreamAndFetchRanges(BigIntegerToken[] tokens, BigIntegerToken[] tokensAfterMove, int movingNodeIdx) throws UnknownHostException
{
RackInferringSnitch endpointSnitch = new RackInferringSnitch();

BIN
tools/lib/cassandra-driver-core-2.0.0-rc2-SNAPSHOT-jar-with-dependencies.jar
View File

Binary file not shown.

BIN
tools/lib/cassandra-driver-core-2.0.0-rc2-SNAPSHOT.jar
View File

Binary file not shown.

BIN
tools/lib/cassandra-driver-core-2.0.0-rc3-SNAPSHOT.jar Normal file
View File

Binary file not shown.

6
tools/stress/src/org/apache/cassandra/stress/StressAction.java
View File

@ -155,9 +155,7 @@ public class StressAction implements Runnable
private boolean hasAverageImprovement(List<StressMetrics> results, int count, double minImprovement)
{
if (results.size() < count + 1)
return true;
return averageImprovement(results, count) >= minImprovement;
return results.size() < count + 1 || averageImprovement(results, count) >= minImprovement;
}
private double averageImprovement(List<StressMetrics> results, int count)
@ -389,7 +387,7 @@ public class StressAction implements Runnable
int batchSize = (int) (operations / (1 << 19));
if (batchSize < 20)
batchSize = 20;
ArrayBlockingQueue<Work> work = new ArrayBlockingQueue<Work>(
ArrayBlockingQueue<Work> work = new ArrayBlockingQueue<>(
(int) ((operations / batchSize)
+ (operations % batchSize == 0 ? 0 : 1))
);

6
tools/stress/src/org/apache/cassandra/stress/generatedata/DataGenStringDictionary.java
View File

@ -27,17 +27,17 @@ public class DataGenStringDictionary extends DataGen
@Override
public void generate(ByteBuffer fill, long index, ByteBuffer seed)
{
fill(fill, 0);
fill(fill);
}
@Override
public void generate(List<ByteBuffer> fills, long index, ByteBuffer seed)
{
for (int i = 0 ; i < fills.size() ; i++)
fill(fills.get(0), i);
fill(fills.get(0));
}
private void fill(ByteBuffer fill, int column)
private void fill(ByteBuffer fill)
{
fill.clear();
byte[] trg = fill.array();

2
tools/stress/src/org/apache/cassandra/stress/generatedata/DataGenStringRepeats.java
View File

@ -50,7 +50,7 @@ public class DataGenStringRepeats extends DataGen
private byte[] getData(long index, int column, ByteBuffer seed)
{
final long key = (column * repeatFrequency) + ((seed == null ? index : Math.abs(seed.hashCode())) % repeatFrequency);
final long key = ((long)column * repeatFrequency) + ((seed == null ? index : Math.abs(seed.hashCode())) % repeatFrequency);
byte[] r = cache.get(key);
if (r != null)
return r;

6
tools/stress/src/org/apache/cassandra/stress/generatedata/RowGenDistributedSize.java
View File

@ -21,6 +21,8 @@ public class RowGenDistributedSize extends RowGen
final ByteBuffer[] ret;
final int[] sizes;
final boolean isDeterministic;
public RowGenDistributedSize(DataGen dataGenerator, Distribution countDistribution, Distribution sizeDistribution)
{
super(dataGenerator);
@ -28,6 +30,8 @@ public class RowGenDistributedSize extends RowGen
this.sizeDistribution = sizeDistribution;
ret = new ByteBuffer[(int) countDistribution.maxValue()];
sizes = new int[ret.length];
this.isDeterministic = dataGen.isDeterministic() && countDistribution.maxValue() == countDistribution.minValue()
&& sizeDistribution.minValue() == sizeDistribution.maxValue();
}
ByteBuffer getBuffer(int size)
@ -78,7 +82,7 @@ public class RowGenDistributedSize extends RowGen
@Override
public boolean isDeterministic()
{
return false;
return isDeterministic;
}
}

29
tools/stress/src/org/apache/cassandra/stress/operations/CqlOperation.java
View File

@ -192,8 +192,12 @@ public abstract class CqlOperation<V> extends Operation
if (result.length != expect.size())
return false;
for (int i = 0 ; i < result.length ; i++)
if (!expect.get(i).equals(Arrays.asList(result[i])))
{
List<ByteBuffer> resultRow = Arrays.asList(result[i]);
resultRow = resultRow.subList(1, resultRow.size());
if (expect.get(i) != null && !expect.get(i).equals(resultRow))
return false;
}
return true;
}
}
@ -521,20 +525,19 @@ public abstract class CqlOperation<V> extends Operation
@Override
public ByteBuffer[][] apply(ResultMessage result)
{
if (result instanceof ResultMessage.Rows)
if (!(result instanceof ResultMessage.Rows))
return new ByteBuffer[0][];
ResultMessage.Rows rows = ((ResultMessage.Rows) result);
ByteBuffer[][] r = new ByteBuffer[rows.result.size()][];
for (int i = 0 ; i < r.length ; i++)
{
ResultMessage.Rows rows = ((ResultMessage.Rows) result);
ByteBuffer[][] r = new ByteBuffer[rows.result.size()][];
for (int i = 0 ; i < r.length ; i++)
{
List<ByteBuffer> row = rows.result.rows.get(i);
r[i] = new ByteBuffer[row.size()];
for (int j = 0 ; j < row.size() ; j++)
r[i][j] = row.get(j);
}
return r;
List<ByteBuffer> row = rows.result.rows.get(i);
r[i] = new ByteBuffer[row.size()];
for (int j = 0 ; j < row.size() ; j++)
r[i][j] = row.get(j);
}
return new ByteBuffer[0][];
return r;
}
};
}

7
tools/stress/src/org/apache/cassandra/stress/settings/SettingsKey.java
View File

@ -73,9 +73,10 @@ public class SettingsKey implements Serializable
public KeyGen newKeyGen()
{
if (range != null)
return new KeyGen(new DataGenHexFromOpIndex(range[0], range[1]), keySize);
return new KeyGen(new DataGenHexFromDistribution(distribution.get()), keySize);
return new KeyGen(range == null
? new DataGenHexFromDistribution(distribution.get())
: new DataGenHexFromOpIndex(range[0], range[1]),
keySize);
}
// CLI Utility Methods

9
tools/stress/src/org/apache/cassandra/stress/util/JavaDriverClient.java
View File

@ -62,7 +62,9 @@ public class JavaDriverClient
public void connect(ProtocolOptions.Compression compression) throws Exception
{
Cluster.Builder clusterBuilder = Cluster.builder()
.addContactPoint(host).withPort(port);
.addContactPoint(host)
.withPort(port)
.withoutMetrics(); // The driver uses metrics 3 with conflict with our version
clusterBuilder.withCompression(compression);
if (encryptionOptions.enabled)
{
@ -142,7 +144,6 @@ public class JavaDriverClient
public void disconnect()
{
FBUtilities.waitOnFuture(cluster.shutdown());
cluster.close();
}
}
}