From 022208bf0f3099b91143a76e8b57d5fa63671269 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Wed, 11 May 2011 14:19:20 +0000 Subject: [PATCH 1/5] avoid attempting to replay mutationsfrom dropped keyspaces patch by jbellis; reviewed by slebresne for CASSANDRA-2631 git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1101893 13f79535-47bb-0310-9956-ffa450edef68 --- CHANGES.txt | 1 + .../org/apache/cassandra/db/commitlog/CommitLog.java | 12 +++++++----- 2 files changed, 8 insertions(+), 5 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 83109da9cc..033cdb74a8 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -18,6 +18,7 @@ * 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) 0.7.5 diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLog.java b/src/java/org/apache/cassandra/db/commitlog/CommitLog.java index 5498153307..c02c9d55b8 100644 --- a/src/java/org/apache/cassandra/db/commitlog/CommitLog.java +++ b/src/java/org/apache/cassandra/db/commitlog/CommitLog.java @@ -165,7 +165,7 @@ public class CommitLog public static void 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(); @@ -270,9 +270,7 @@ 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 CommitLogHeader finalHeader = clHeader; final RowMutation frm = rm; @@ -280,12 +278,15 @@ public class CommitLog { 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 @@ -297,6 +298,7 @@ public class CommitLog if (!newRm.isEmpty()) { Table.open(newRm.getTable()).apply(newRm, null, false); + tablesRecovered.add(table); } } }; From d2c98b024d85bf72a6ce3e03d0b8d2994199c254 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Wed, 11 May 2011 16:03:12 +0000 Subject: [PATCH 2/5] avoid using cached position of a keywhen GTis requested patch by Richard Low and slebresne; reviewed by jbellis for CASSANDRA-2633 git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1101933 13f79535-47bb-0310-9956-ffa450edef68 --- CHANGES.txt | 1 + .../cassandra/io/sstable/SSTableReader.java | 11 +++-- .../io/sstable/SSTableReaderTest.java | 45 +++++++++++++++++++ 3 files changed, 53 insertions(+), 4 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 033cdb74a8..c553e6998b 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -19,6 +19,7 @@ (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) 0.7.5 diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java index ea19ef421f..8b4f2d0560 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java @@ -451,10 +451,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); 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))); + } } From 61adae4fdcb34e303a07ec34c8ff05726ed5de13 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Wed, 11 May 2011 16:07:13 +0000 Subject: [PATCH 3/5] fix counting bloom filter true positives patch by jbellis; reviewed by slebresne for CASSANDRA-2637 git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1101940 13f79535-47bb-0310-9956-ffa450edef68 --- CHANGES.txt | 1 + src/java/org/apache/cassandra/io/sstable/SSTableReader.java | 4 ++-- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index c553e6998b..e2dcb65fe1 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -20,6 +20,7 @@ * 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) 0.7.5 diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java index 8b4f2d0560..a0b45a6b0d 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java @@ -488,12 +488,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) From 054404ba43d2e673adbd3da0317ed17b5c91701e Mon Sep 17 00:00:00 2001 From: Gary Dusbabek Date: Wed, 11 May 2011 19:23:22 +0000 Subject: [PATCH 4/5] initialize local ep state prior to gossip startup if needed. patch by gdusbabek, reviewed by brandonwilliams. CASSANDRA-2638 git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1102038 13f79535-47bb-0310-9956-ffa450edef68 --- CHANGES.txt | 3 ++- .../org/apache/cassandra/gms/Gossiper.java | 24 ++++++++++++------- .../service/AbstractCassandraDaemon.java | 2 ++ 3 files changed, 20 insertions(+), 9 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index e2dcb65fe1..c753213bf5 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -21,7 +21,8 @@ * 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) + 0.7.5 * improvements/fixes to PIG driver (CASSANDRA-1618, CASSANDRA-2387, diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java index de9201e2a9..057b560564 100644 --- a/src/java/org/apache/cassandra/gms/Gossiper.java +++ b/src/java/org/apache/cassandra/gms/Gossiper.java @@ -898,15 +898,8 @@ public class Gossiper implements IFailureDetectionEventListener } /* initialize the heartbeat state for this localEndpoint */ + maybeInitializeLocalState(generationNbr); EndpointState localState = endpointStateMap_.get(localEndpoint_); - if ( localState == null ) - { - HeartBeatState hbState = new HeartBeatState(generationNbr); - localState = new EndpointState(hbState); - localState.isAlive(true); - localState.isAGossiper(true); - endpointStateMap_.put(localEndpoint_, localState); - } //notify snitches that Gossiper is about to start DatabaseDescriptor.getEndpointSnitch().gossiperStarting(); @@ -916,6 +909,21 @@ 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.isAlive(true); + localState.isAGossiper(true); + endpointStateMap_.put(localEndpoint_, localState); + } + } + /** * Add an endpoint we knew about previously, but whose state is unknown diff --git a/src/java/org/apache/cassandra/service/AbstractCassandraDaemon.java b/src/java/org/apache/cassandra/service/AbstractCassandraDaemon.java index e36b70ba90..84f600ae87 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; @@ -181,6 +182,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); } From 4a604fbebc1d06ee569749f69a5cd608e26e4c89 Mon Sep 17 00:00:00 2001 From: Sylvain Lebresne Date: Thu, 12 May 2011 15:06:48 +0000 Subject: [PATCH 5/5] Avoid wrong empty result with secondary index when limit=1 patch by slebresne; reviewed by jbellis for CASSANDRA-2628 git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1102337 13f79535-47bb-0310-9956-ffa450edef68 --- CHANGES.txt | 3 +- .../cassandra/db/ColumnFamilyStore.java | 5 ++- .../cassandra/db/ColumnFamilyStoreTest.java | 39 +++++++++++++++++++ 3 files changed, 45 insertions(+), 2 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index c753213bf5..610e8b473a 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -22,7 +22,8 @@ * 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 empty Result with secondary index when limit=1 (CASSANDRA-2628) + 0.7.5 * improvements/fixes to PIG driver (CASSANDRA-1618, CASSANDRA-2387, diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 9d60fe0ab1..69ca242599 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -1569,12 +1569,15 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean if (logger.isDebugEnabled()) logger.debug(String.format("Scanning index %s starting with %s", expressionString(primary), indexCFS.getComparator().getString(startKey))); + + // We shouldn't fetch only 1 row as this provides buggy paging in case the first row doesn't satisfy all clauses + int count = Math.max(clause.count, 2); QueryFilter indexFilter = QueryFilter.getSliceFilter(indexKey, new QueryPath(indexCFS.getColumnFamilyName()), startKey, ByteBufferUtil.EMPTY_BYTE_BUFFER, false, - clause.count); + count); ColumnFamily indexRow = indexCFS.getColumnFamily(indexFilter); logger.debug("fetched {}", indexRow); if (indexRow == null) diff --git a/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java b/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java index 8f9db8141d..7f5765ec72 100644 --- a/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java +++ b/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java @@ -355,6 +355,45 @@ public class ColumnFamilyStoreTest extends CleanupHelper } + // See CASSANDRA-2628 + @Test + public void testIndexScanWithLimitOne() throws IOException + { + RowMutation rm; + + rm = new RowMutation("Keyspace1", ByteBufferUtil.bytes("kk1")); + rm.add(new QueryPath("Indexed1", null, ByteBufferUtil.bytes("notbirthdate")), ByteBufferUtil.bytes(1L), 0); + rm.add(new QueryPath("Indexed1", null, ByteBufferUtil.bytes("birthdate")), ByteBufferUtil.bytes(1L), 0); + rm.apply(); + + rm = new RowMutation("Keyspace1", ByteBufferUtil.bytes("kk2")); + rm.add(new QueryPath("Indexed1", null, ByteBufferUtil.bytes("notbirthdate")), ByteBufferUtil.bytes(2L), 0); + rm.add(new QueryPath("Indexed1", null, ByteBufferUtil.bytes("birthdate")), ByteBufferUtil.bytes(1L), 0); + rm.apply(); + + rm = new RowMutation("Keyspace1", ByteBufferUtil.bytes("kk3")); + rm.add(new QueryPath("Indexed1", null, ByteBufferUtil.bytes("notbirthdate")), ByteBufferUtil.bytes(2L), 0); + rm.add(new QueryPath("Indexed1", null, ByteBufferUtil.bytes("birthdate")), ByteBufferUtil.bytes(1L), 0); + rm.apply(); + + rm = new RowMutation("Keyspace1", ByteBufferUtil.bytes("kk4")); + rm.add(new QueryPath("Indexed1", null, ByteBufferUtil.bytes("notbirthdate")), ByteBufferUtil.bytes(2L), 0); + rm.add(new QueryPath("Indexed1", null, ByteBufferUtil.bytes("birthdate")), ByteBufferUtil.bytes(1L), 0); + rm.apply(); + + // basic single-expression query + IndexExpression expr1 = new IndexExpression(ByteBufferUtil.bytes("birthdate"), IndexOperator.EQ, ByteBufferUtil.bytes(1L)); + IndexExpression expr2 = new IndexExpression(ByteBufferUtil.bytes("notbirthdate"), IndexOperator.GT, ByteBufferUtil.bytes(1L)); + IndexClause clause = new IndexClause(Arrays.asList(new IndexExpression[]{ expr1, expr2 }), ByteBufferUtil.EMPTY_BYTE_BUFFER, 1); + IFilter filter = new IdentityQueryFilter(); + IPartitioner p = StorageService.getPartitioner(); + Range range = new Range(p.getMinimumToken(), p.getMinimumToken()); + List rows = Table.open("Keyspace1").getColumnFamilyStore("Indexed1").scan(clause, range, filter); + + assert rows != null; + assert rows.size() == 1 : StringUtils.join(rows, ","); + } + @Test public void testIndexCreate() throws IOException, ConfigurationException, InterruptedException, ExecutionException {