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

merge from 0.8.1 

git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1102518 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2011-05-13 00:09:32 +00:00
parent c5c07bd73d 4a604fbebc
commit f4f1781a8f
10 changed files with 257 additions and 67 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
CHANGES.txt
View File

@ -1,6 +1,3 @@
1.0-dev
0.8.1
* add support for insert, delete in cql BATCH (CASSANDRA-2537)
* add support for IN to cql SELECT, UPDATE (CASSANDRA-2553)
@ -9,21 +6,17 @@
(CASSANDRA-2355)
* add CompositeType and DynamicCompositeType (CASSANDRA-2231)
* add CQL TTL support (CASSANDRA-2476)
0.8.0-?
* initialize local ep state prior to gossip startup if needed (CASSANDRA-2638)
* fix counter increment lost after restart (CASSANDRA-2642)
* optimize batches containing multiple updates to the same row
(CASSANDRA-2583)
0.8.0-rc1
* faster flushes and compaction from fixing excessively pessimistic
rebuffering in BRAF (CASSANDRA-2581)
* fix merkle tree splitting exiting early (CASSANDRA-2605)
* Disable compaction throttling during bootstrap (CASSANDRA-2612)
* fix returning null column values in the python cql driver (CASSANDRA-2593)
* fix merkle tree splitting exiting early (CASSANDRA-2605)
* snapshot_before_compaction directory name fix (CASSANDRA-2598)
* Disable compaction throttling during bootstrap (CASSANDRA-2612)
* fix CQL treatment of > and < operators in range slices (CASSANDRA-2592)
* fix potential double-application of counter updates on commitlog replay
(CASSANDRA-2419)
@ -35,6 +28,17 @@
* Don't allow {LOCAL|EACH}_QUORUM unless strategy is NTS (CASSANDRA-2627)
* validate keyspace strategy_options during CQL create (CASSANDRA-2624)
* fix empty Result with secondary index when limit=1 (CASSANDRA-2628)
* Fix regression where bootstrapping a node with no schema fails
(CASSANDRA-2625)
* Allow removing LocationInfo sstables (CASSANDRA-2632)
* avoid attempting to replay mutations from dropped keyspaces (CASSANDRA-2631)
* avoid using cached position of a key when GT is requested (CASSANDRA-2633)
* fix counting bloom filter true positives (CASSANDRA-2637)
* initialize local ep state prior to gossip startup if needed (CASSANDRA-2638)
* fix counter increment lost after restart (CASSANDRA-2642)
* add quote-escaping via backslash to CLI (CASSANDRA-2623)
* fig pig example script (CASSANDRA-2487)
* fix dynamic snitch race in adding latencies (CASSANDRA-2618)
0.8.0-beta2
@ -62,10 +66,7 @@
* improve ignoring of obsolete mutations in index maintenance (CASSANDRA-2401)
* recognize attempt to drop just the index while leaving the column
definition alone (CASSANDRA-2619)
* Fix regression where bootstrapping a node with no schema fails
(CASSANDRA-2625)
* Allow removing LocationInfo sstables (CASSANDRA-2632)
0.8.0-beta1
* remove Avro RPC support (CASSANDRA-926)

41
src/java/org/apache/cassandra/cql/AbstractModification.java
View File

@ -23,7 +23,9 @@ package org.apache.cassandra.cql;
import org.apache.cassandra.db.RowMutation;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.InvalidRequestException;
import java.nio.ByteBuffer;
import java.util.List;
public abstract class AbstractModification
@ -91,10 +93,10 @@ public abstract class AbstractModification
*
* @return list of the mutations
*
* @throws org.apache.cassandra.thrift.InvalidRequestException on the wrong request
* @throws InvalidRequestException on the wrong request
*/
public abstract List<RowMutation> prepareRowMutations(String keyspace, ClientState clientState)
throws org.apache.cassandra.thrift.InvalidRequestException;
throws InvalidRequestException;
/**
* Convert statement into a list of mutations to apply on the server
@ -105,8 +107,39 @@ public abstract class AbstractModification
*
* @return list of the mutations
*
* @throws org.apache.cassandra.thrift.InvalidRequestException on the wrong request
* @throws InvalidRequestException on the wrong request
*/
public abstract List<RowMutation> prepareRowMutations(String keyspace, ClientState clientState, Long timestamp)
throws org.apache.cassandra.thrift.InvalidRequestException;
throws InvalidRequestException;
/**
* Compute a row mutation for a single key
*
* @param key The key for mutation
* @param keyspace The keyspace
* @param timestamp The global timestamp for mutation
*
* @return row mutation
*
* @throws InvalidRequestException on the wrong request
*/
public abstract RowMutation mutationForKey(ByteBuffer key, String keyspace, Long timestamp)
throws InvalidRequestException;
/**
* Compute a row mutation for a single key and add it to the given RowMutation object
*
* @param mutation The row mutation to add computed mutation into
* @param keyspace The keyspace
* @param timestamp The global timestamp for mutation
*
* @throws InvalidRequestException on the wrong request
*/
public abstract void mutationForKey(RowMutation mutation, String keyspace, Long timestamp)
throws InvalidRequestException;
/**
* @return a list of the keys associated with the statement
*/
public abstract List<Term> getKeys();
}

67
src/java/org/apache/cassandra/cql/BatchStatement.java
View File

@ -20,14 +20,23 @@
*/
package org.apache.cassandra.cql;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedList;
import java.util.List;
import org.apache.cassandra.auth.Permission;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamily;
import org.apache.cassandra.db.RowMutation;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.InvalidRequestException;
import static org.apache.cassandra.thrift.ThriftValidation.validateColumnFamily;
/**
* A <code>BATCH</code> statement parsed from a CQL query.
*
@ -78,11 +87,39 @@ public class BatchStatement
public List<RowMutation> getMutations(String keyspace, ClientState clientState) throws InvalidRequestException
{
// To avoid unnecessary authorizations.
List<String> seenColumnFamilies = new ArrayList<String>();
List<RowMutation> batch = new LinkedList<RowMutation>();
for (AbstractModification statement : statements)
{
batch.addAll(statement.prepareRowMutations(keyspace, clientState, timestamp));
final String columnFamily = statement.getColumnFamily();
authorizeColumnFamily(keyspace, columnFamily, clientState, seenColumnFamilies);
AbstractType<?> keyValidator = getKeyType(keyspace, columnFamily);
for (Term rawKey : statement.getKeys()) // for each key of the statement
{
ByteBuffer key = rawKey.getByteBuffer(keyValidator);
boolean found = false;
for (RowMutation mutation : batch)
{
if (mutation.key().equals(key) && hasColumnFamily(mutation.getColumnFamilies(), columnFamily))
{
statement.mutationForKey(mutation, keyspace, timestamp);
found = true;
break;
}
}
if (!found) // if mutation was not found we should add a new one
batch.add(statement.mutationForKey(key, keyspace, timestamp));
}
}
return batch;
@ -93,6 +130,34 @@ public class BatchStatement
return timestamp != null;
}
private boolean hasColumnFamily(Collection<ColumnFamily> columnFamilies, String columnFamily)
{
for (ColumnFamily cf : columnFamilies)
{
if (cf.metadata().cfName.equals(columnFamily))
return true;
}
return false;
}
private void authorizeColumnFamily(String keyspace, String columnFamily, ClientState state, List<String> seenCFs)
throws InvalidRequestException
{
validateColumnFamily(keyspace, columnFamily, false);
if (!seenCFs.contains(columnFamily))
{
state.hasColumnFamilyAccess(columnFamily, Permission.WRITE);
seenCFs.add(columnFamily);
}
}
public AbstractType<?> getKeyType(String keyspace, String columnFamily)
{
return DatabaseDescriptor.getCFMetaData(keyspace, columnFamily).getKeyValidator();
}
public String toString()
{
return String.format("BatchStatement(statements=%s, consistency=%s)", statements, consistency);

49
src/java/org/apache/cassandra/cql/DeleteStatement.java
View File

@ -59,6 +59,7 @@ public class DeleteStatement extends AbstractModification
return columns;
}
/** {@inheritDoc} */
public List<Term> getKeys()
{
return keys;
@ -74,35 +75,47 @@ public class DeleteStatement extends AbstractModification
public List<RowMutation> prepareRowMutations(String keyspace, ClientState clientState, Long timestamp) throws InvalidRequestException
{
clientState.hasColumnFamilyAccess(columnFamily, Permission.WRITE);
CFMetaData metadata = validateColumnFamily(keyspace, columnFamily, false);
AbstractType comparator = metadata.getComparatorFor(null);
AbstractType<?> keyType = DatabaseDescriptor.getCFMetaData(keyspace, columnFamily).getKeyValidator();
List<RowMutation> rowMutations = new ArrayList<RowMutation>();
for (Term key : keys)
{
RowMutation rm = new RowMutation(keyspace, key.getByteBuffer(keyType));
if (columns.size() < 1) // No columns, delete the row
rm.delete(new QueryPath(columnFamily), System.currentTimeMillis());
else // Delete specific columns
{
for (Term column : columns)
{
ByteBuffer columnName = column.getByteBuffer(comparator);
validateColumnName(columnName);
rm.delete(new QueryPath(columnFamily, null, columnName), (timestamp == null) ? getTimestamp() : timestamp);
}
}
rowMutations.add(rm);
rowMutations.add(mutationForKey(key.getByteBuffer(keyType), keyspace, timestamp));
}
return rowMutations;
}
/** {@inheritDoc} */
public RowMutation mutationForKey(ByteBuffer key, String keyspace, Long timestamp) throws InvalidRequestException
{
RowMutation rm = new RowMutation(keyspace, key);
mutationForKey(rm, keyspace, timestamp);
return rm;
}
/** {@inheritDoc} */
public void mutationForKey(RowMutation mutation, String keyspace, Long timestamp) throws InvalidRequestException
{
CFMetaData metadata = validateColumnFamily(keyspace, columnFamily, false);
AbstractType comparator = metadata.getComparatorFor(null);
if (columns.size() < 1) // No columns, delete the row
mutation.delete(new QueryPath(columnFamily), System.currentTimeMillis());
else // Delete specific columns
{
for (Term column : columns)
{
ByteBuffer columnName = column.getByteBuffer(comparator);
validateColumnName(columnName);
mutation.delete(new QueryPath(columnFamily, null, columnName), (timestamp == null) ? getTimestamp() : timestamp);
}
}
}
public String toString()
{
return String.format("DeleteStatement(columns=%s, columnFamily=%s, consistency=%s keys=%s)",

39
src/java/org/apache/cassandra/cql/UpdateStatement.java
View File

@ -33,7 +33,6 @@ import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.InvalidRequestException;
import static org.apache.cassandra.cql.QueryProcessor.validateKey;
import static org.apache.cassandra.cql.QueryProcessor.validateColumn;
import static org.apache.cassandra.thrift.ThriftValidation.validateColumnFamily;
@ -157,36 +156,54 @@ public class UpdateStatement extends AbstractModification
*/
private RowMutation mutationForKey(String keyspace, ByteBuffer key, CFMetaData metadata, Long timestamp) throws InvalidRequestException
{
validateKey(key);
RowMutation rm = new RowMutation(keyspace, key);
mutationForKey(rm, keyspace, metadata, timestamp);
return rm;
}
/** {@inheritDoc} */
public RowMutation mutationForKey(ByteBuffer key, String keyspace, Long timestamp) throws InvalidRequestException
{
return mutationForKey(keyspace, key, validateColumnFamily(keyspace, columnFamily, false), timestamp);
}
/** {@inheritDoc} */
public void mutationForKey(RowMutation mutation, String keyspace, Long timestamp) throws InvalidRequestException
{
mutationForKey(mutation, keyspace, validateColumnFamily(keyspace, columnFamily, false), timestamp);
}
private void mutationForKey(RowMutation mutation, String keyspace, CFMetaData metadata, Long timestamp) throws InvalidRequestException
{
AbstractType<?> comparator = getComparator(keyspace);
RowMutation rm = new RowMutation(keyspace, key);
for (Map.Entry<Term, Term> column : getColumns().entrySet())
{
ByteBuffer colName = column.getKey().getByteBuffer(comparator);
ByteBuffer colValue = column.getValue().getByteBuffer(getValueValidator(keyspace, colName));
validateColumn(metadata, colName, colValue);
rm.add(new QueryPath(columnFamily, null, colName),
colValue,
(timestamp == null) ? getTimestamp() : timestamp,
getTimeToLive());
}
return rm;
mutation.add(new QueryPath(columnFamily, null, colName),
colValue,
(timestamp == null) ? getTimestamp() : timestamp,
getTimeToLive());
}
}
public String getColumnFamily()
{
return columnFamily;
}
/** {@inheritDoc} */
public List<Term> getKeys()
{
return keys;
}
public Map<Term, Term> getColumns() throws InvalidRequestException
{
// Created from an UPDATE

12
src/java/org/apache/cassandra/db/commitlog/CommitLog.java
View File

@ -168,7 +168,7 @@ public class CommitLog
// returns the number of replayed mutation (useful for tests in particular)
public static int recover(File[] clogs) throws IOException
{
Set<Table> tablesRecovered = new HashSet<Table>();
final Set<Table> tablesRecovered = new HashSet<Table>();
List<Future<?>> futures = new ArrayList<Future<?>>();
byte[] bytes = new byte[4096];
Map<Integer, AtomicInteger> invalidMutations = new HashMap<Integer, AtomicInteger>();
@ -288,21 +288,22 @@ public class CommitLog
rm.getTable(),
ByteBufferUtil.bytesToHex(rm.key()),
"{" + StringUtils.join(rm.getColumnFamilies(), ", ") + "}"));
final Table table = Table.open(rm.getTable());
tablesRecovered.add(table);
final Collection<ColumnFamily> columnFamilies = new ArrayList<ColumnFamily>(rm.getColumnFamilies());
final long entryLocation = reader.getFilePointer();
final RowMutation frm = rm;
Runnable runnable = new WrappedRunnable()
{
public void runMayThrow() throws IOException
{
if (DatabaseDescriptor.getKSMetaData(frm.getTable()) == null)
return;
final Table table = Table.open(frm.getTable());
RowMutation newRm = new RowMutation(frm.getTable(), frm.key());
// Rebuild the row mutation, omitting column families that a) have already been flushed,
// b) are part of a cf that was dropped. Keep in mind that the cf.name() is suspect. do every
// thing based on the cfid instead.
for (ColumnFamily columnFamily : columnFamilies)
for (ColumnFamily columnFamily : frm.getColumnFamilies())
{
if (CFMetaData.getCF(columnFamily.id()) == null)
// null means the cf has been dropped
@ -321,6 +322,7 @@ public class CommitLog
if (!newRm.isEmpty())
{
Table.open(newRm.getTable()).apply(newRm, false);
tablesRecovered.add(table);
}
}
};

23
src/java/org/apache/cassandra/gms/Gossiper.java
View File

@ -27,6 +27,7 @@ import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.*;
import org.apache.cassandra.db.SystemTable;
import org.apache.cassandra.net.MessageProducer;
import org.apache.cassandra.utils.FBUtilities;
import org.cliffc.high_scale_lib.NonBlockingHashMap;
@ -838,14 +839,8 @@ public class Gossiper implements IFailureDetectionEventListener
}
/* initialize the heartbeat state for this localEndpoint */
maybeInitializeLocalState(generationNbr);
EndpointState localState = endpointStateMap.get(FBUtilities.getLocalAddress());
if ( localState == null )
{
HeartBeatState hbState = new HeartBeatState(generationNbr);
localState = new EndpointState(hbState);
localState.markAlive();
endpointStateMap.put(FBUtilities.getLocalAddress(), localState);
}
//notify snitches that Gossiper is about to start
DatabaseDescriptor.getEndpointSnitch().gossiperStarting();
@ -857,6 +852,20 @@ public class Gossiper implements IFailureDetectionEventListener
Gossiper.intervalInMillis,
TimeUnit.MILLISECONDS);
}
// initialize local HB state if needed.
public void maybeInitializeLocalState(int generationNbr)
{
EndpointState localState = endpointStateMap.get(FBUtilities.getLocalAddress());
if ( localState == null )
{
HeartBeatState hbState = new HeartBeatState(generationNbr);
localState = new EndpointState(hbState);
localState.markAlive();
endpointStateMap.put(FBUtilities.getLocalAddress(), localState);
}
}
/**
* Add an endpoint we knew about previously, but whose state is unknown

15
src/java/org/apache/cassandra/io/sstable/SSTableReader.java
View File

@ -539,10 +539,13 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
}
// next, the key cache
Pair<Descriptor, DecoratedKey> unifiedKey = new Pair<Descriptor, DecoratedKey>(descriptor, decoratedKey);
Long cachedPosition = getCachedPosition(unifiedKey);
if (cachedPosition != null)
return cachedPosition;
if (op == Operator.EQ || op == Operator.GE)
{
Pair<Descriptor, DecoratedKey> unifiedKey = new Pair<Descriptor, DecoratedKey>(descriptor, decoratedKey);
Long cachedPosition = getCachedPosition(unifiedKey);
if (cachedPosition != null)
return cachedPosition;
}
// next, see if the sampled index says it's impossible for the key to be present
IndexSummary.KeyPosition sampledPosition = getIndexScanPosition(decoratedKey);
@ -573,12 +576,12 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
{
if (comparison == 0 && keyCache != null && keyCache.getCapacity() > 0)
{
if (op == Operator.EQ)
bloomFilterTracker.addTruePositive();
// store exact match for the key
if (decoratedKey.key != null)
cacheKey(decoratedKey, dataPosition);
}
if (op == Operator.EQ)
bloomFilterTracker.addTruePositive();
return dataPosition;
}
if (v < 0)

2
src/java/org/apache/cassandra/service/AbstractCassandraDaemon.java
View File

@ -29,6 +29,7 @@ import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.cassandra.gms.Gossiper;
import org.apache.log4j.PropertyConfigurator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -179,6 +180,7 @@ public abstract class AbstractCassandraDaemon implements CassandraDaemon
UUID lastMigration = Migration.getLastMigrationId();
if ((lastMigration != null) && (lastMigration.timestamp() > currentMigration.timestamp()))
{
Gossiper.instance.maybeInitializeLocalState(SystemTable.incrementAndGetGeneration());
MigrationManager.applyMigrations(currentMigration, lastMigration);
}

45
test/unit/org/apache/cassandra/io/sstable/SSTableReaderTest.java
View File

@ -25,6 +25,7 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.concurrent.ExecutionException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import org.junit.Test;
@ -148,4 +149,48 @@ public class SSTableReaderTest extends CleanupHelper
store.forceBlockingFlush();
assert store.getMaxRowSize() != 0;
}
@Test
public void testGetPositionsForRangesWithKeyCache() throws IOException, ExecutionException, InterruptedException
{
Table table = Table.open("Keyspace1");
ColumnFamilyStore store = table.getColumnFamilyStore("Standard2");
store.getKeyCache().setCapacity(100);
// insert data and compact to a single sstable
CompactionManager.instance.disableAutoCompaction();
for (int j = 0; j < 10; j++)
{
ByteBuffer key = ByteBufferUtil.bytes(String.valueOf(j));
RowMutation rm = new RowMutation("Keyspace1", key);
rm.add(new QueryPath("Standard2", null, ByteBufferUtil.bytes("0")), ByteBufferUtil.EMPTY_BYTE_BUFFER, j);
rm.apply();
}
store.forceBlockingFlush();
CompactionManager.instance.performMajor(store);
SSTableReader sstable = store.getSSTables().iterator().next();
long p2 = sstable.getPosition(k(2), SSTableReader.Operator.EQ);
long p3 = sstable.getPosition(k(3), SSTableReader.Operator.EQ);
long p6 = sstable.getPosition(k(6), SSTableReader.Operator.EQ);
long p7 = sstable.getPosition(k(7), SSTableReader.Operator.EQ);
Pair<Long, Long> p = sstable.getPositionsForRanges(makeRanges(t(2), t(6))).iterator().next();
// range are start exclusive so we should start at 3
assert p.left == p3;
// to capture 6 we have to stop at the start of 7
assert p.right == p7;
}
private List<Range> makeRanges(Token left, Token right)
{
return Arrays.asList(new Range[]{ new Range(left, right) });
}
private DecoratedKey k(int i)
{
return new DecoratedKey(t(i), ByteBufferUtil.bytes(String.valueOf(i)));
}
}