diff --git a/CHANGES.txt b/CHANGES.txt index d8328f41af..4430de6f84 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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) diff --git a/src/java/org/apache/cassandra/cql/AbstractModification.java b/src/java/org/apache/cassandra/cql/AbstractModification.java index e14355f3bb..e1c72d2ea3 100644 --- a/src/java/org/apache/cassandra/cql/AbstractModification.java +++ b/src/java/org/apache/cassandra/cql/AbstractModification.java @@ -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 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 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 getKeys(); } diff --git a/src/java/org/apache/cassandra/cql/BatchStatement.java b/src/java/org/apache/cassandra/cql/BatchStatement.java index c5e1418ce0..2f84c83985 100644 --- a/src/java/org/apache/cassandra/cql/BatchStatement.java +++ b/src/java/org/apache/cassandra/cql/BatchStatement.java @@ -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 BATCH statement parsed from a CQL query. * @@ -78,11 +87,39 @@ public class BatchStatement public List getMutations(String keyspace, ClientState clientState) throws InvalidRequestException { + // To avoid unnecessary authorizations. + List seenColumnFamilies = new ArrayList(); + List batch = new LinkedList(); 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 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 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); diff --git a/src/java/org/apache/cassandra/cql/DeleteStatement.java b/src/java/org/apache/cassandra/cql/DeleteStatement.java index 80ecc90587..120cb4475c 100644 --- a/src/java/org/apache/cassandra/cql/DeleteStatement.java +++ b/src/java/org/apache/cassandra/cql/DeleteStatement.java @@ -59,6 +59,7 @@ public class DeleteStatement extends AbstractModification return columns; } + /** {@inheritDoc} */ public List getKeys() { return keys; @@ -74,35 +75,47 @@ public class DeleteStatement extends AbstractModification public List 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 rowMutations = new ArrayList(); 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)", diff --git a/src/java/org/apache/cassandra/cql/UpdateStatement.java b/src/java/org/apache/cassandra/cql/UpdateStatement.java index 8ee23b2071..f2e6de890b 100644 --- a/src/java/org/apache/cassandra/cql/UpdateStatement.java +++ b/src/java/org/apache/cassandra/cql/UpdateStatement.java @@ -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 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 getKeys() { return keys; } - + public Map getColumns() throws InvalidRequestException { // Created from an UPDATE diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLog.java b/src/java/org/apache/cassandra/db/commitlog/CommitLog.java index 65de2604af..9f6077e947 100644 --- a/src/java/org/apache/cassandra/db/commitlog/CommitLog.java +++ b/src/java/org/apache/cassandra/db/commitlog/CommitLog.java @@ -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 tablesRecovered = new HashSet
(); + final Set
tablesRecovered = new HashSet
(); List> futures = new ArrayList>(); byte[] bytes = new byte[4096]; Map invalidMutations = new HashMap(); @@ -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 columnFamilies = new ArrayList(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); } } }; diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java index 37b453948c..12cbcf6e5f 100644 --- a/src/java/org/apache/cassandra/gms/Gossiper.java +++ b/src/java/org/apache/cassandra/gms/Gossiper.java @@ -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 diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java index af4fe8be5e..6fd850f508 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java @@ -539,10 +539,13 @@ public class SSTableReader extends SSTable implements Comparable } // next, the key cache - Pair unifiedKey = new Pair(descriptor, decoratedKey); - Long cachedPosition = getCachedPosition(unifiedKey); - if (cachedPosition != null) - return cachedPosition; + if (op == Operator.EQ || op == Operator.GE) + { + Pair unifiedKey = new Pair(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 { 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) diff --git a/src/java/org/apache/cassandra/service/AbstractCassandraDaemon.java b/src/java/org/apache/cassandra/service/AbstractCassandraDaemon.java index 0378edcab1..a9e343adf5 100644 --- a/src/java/org/apache/cassandra/service/AbstractCassandraDaemon.java +++ b/src/java/org/apache/cassandra/service/AbstractCassandraDaemon.java @@ -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); } diff --git a/test/unit/org/apache/cassandra/io/sstable/SSTableReaderTest.java b/test/unit/org/apache/cassandra/io/sstable/SSTableReaderTest.java index 45a50aaf96..ff899245b7 100644 --- a/test/unit/org/apache/cassandra/io/sstable/SSTableReaderTest.java +++ b/test/unit/org/apache/cassandra/io/sstable/SSTableReaderTest.java @@ -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 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 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))); + } }