From fba03ce69122e7a1bd6f2d5709eaa06c1ada0bc1 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Fri, 7 Aug 2009 17:35:25 +0000 Subject: [PATCH] move getKeyRange to CFS, where it encapsulates better. patch by jbellis; reviewed by Eric Evans for CASSANDRA-345 git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@802098 13f79535-47bb-0310-9956-ffa450edef68 --- .../cassandra/db/ColumnFamilyStore.java | 113 +++++++++++++++++ src/java/org/apache/cassandra/db/Table.java | 114 ------------------ .../cassandra/service/RangeVerbHandler.java | 2 +- .../apache/cassandra/db/CompactionsTest.java | 4 +- .../cassandra/db/OneCompactionTest.java | 4 +- .../cassandra/db/RecoveryManager2Test.java | 2 +- 6 files changed, 119 insertions(+), 120 deletions(-) diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 430fe93468..c3db0ac92a 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -43,7 +43,9 @@ import org.apache.cassandra.db.filter.*; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.commons.lang.StringUtils; +import org.apache.commons.lang.ArrayUtils; import org.apache.commons.collections.IteratorUtils; +import org.apache.commons.collections.Predicate; import org.cliffc.high_scale_lib.NonBlockingHashMap; @@ -1458,6 +1460,117 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean } } + /** + * @param startWith key to start with, inclusive. empty string = start at beginning. + * @param stopAt key to stop at, inclusive. empty string = stop only when keys are exhausted. + * @param maxResults + * @return list of keys between startWith and stopAt + */ + public RangeReply getKeyRange(final String startWith, final String stopAt, int maxResults) + throws IOException, ExecutionException, InterruptedException + { + getReadLock().lock(); + try + { + return getKeyRangeUnsafe(startWith, stopAt, maxResults); + } + finally + { + getReadLock().unlock(); + } + } + + private RangeReply getKeyRangeUnsafe(final String startWith, final String stopAt, int maxResults) throws IOException, ExecutionException, InterruptedException + { + // (OPP key decoration is a no-op so using the "decorated" comparator against raw keys is fine) + final Comparator comparator = StorageService.getPartitioner().getDecoratedKeyComparator(); + + // create a CollatedIterator that will return unique keys from different sources + // (current memtable, historical memtables, and SSTables) in the correct order. + List> iterators = new ArrayList>(); + + // we iterate through memtables with a priority queue to avoid more sorting than necessary. + // this predicate throws out the keys before the start of our range. + Predicate p = new Predicate() + { + public boolean evaluate(Object key) + { + String st = (String)key; + return comparator.compare(startWith, st) <= 0 && (stopAt.isEmpty() || comparator.compare(st, stopAt) <= 0); + } + }; + + // current memtable keys. have to go through the CFS api for locking. + iterators.add(IteratorUtils.filteredIterator(memtableKeyIterator(), p)); + // historical memtables + for (Memtable memtable : ColumnFamilyStore.getUnflushedMemtables(columnFamily_)) + { + iterators.add(IteratorUtils.filteredIterator(Memtable.getKeyIterator(memtable.getKeys()), p)); + } + + // sstables + for (SSTableReader sstable : getSSTables()) + { + FileStruct fs = sstable.getFileStruct(); + fs.seekTo(startWith); + iterators.add(fs); + } + + Iterator collated = IteratorUtils.collatedIterator(comparator, iterators); + Iterable reduced = new ReducingIterator(collated) { + String current; + + public void reduce(String current) + { + this.current = current; + } + + protected String getReduced() + { + return current; + } + }; + + try + { + // pull keys out of the CollatedIterator. checking tombstone status is expensive, + // so we set an arbitrary limit on how many we'll do at once. + List keys = new ArrayList(); + boolean rangeCompletedLocally = false; + for (String current : reduced) + { + if (!stopAt.isEmpty() && comparator.compare(stopAt, current) < 0) + { + rangeCompletedLocally = true; + break; + } + // make sure there is actually non-tombstone content associated w/ this key + // TODO record the key source(s) somehow and only check that source (e.g., memtable or sstable) + QueryFilter filter = new SliceQueryFilter(current, new QueryPath(columnFamily_), ArrayUtils.EMPTY_BYTE_ARRAY, ArrayUtils.EMPTY_BYTE_ARRAY, true, 1); + if (getColumnFamily(filter, Integer.MAX_VALUE) != null) + { + keys.add(current); + } + if (keys.size() >= maxResults) + { + rangeCompletedLocally = true; + break; + } + } + return new RangeReply(keys, rangeCompletedLocally); + } + finally + { + for (Iterator iter : iterators) + { + if (iter instanceof FileStruct) + { + ((FileStruct)iter).close(); + } + } + } + } + public AbstractType getComparator() { return DatabaseDescriptor.getComparator(table_, columnFamily_); diff --git a/src/java/org/apache/cassandra/db/Table.java b/src/java/org/apache/cassandra/db/Table.java index f9068220d7..85bb104f00 100644 --- a/src/java/org/apache/cassandra/db/Table.java +++ b/src/java/org/apache/cassandra/db/Table.java @@ -664,120 +664,6 @@ public class Table return applicationColumnFamilies_; } - /** - * @param startWith key to start with, inclusive. empty string = start at beginning. - * @param stopAt key to stop at, inclusive. empty string = stop only when keys are exhausted. - * @param maxResults - * @return list of keys between startWith and stopAt - */ - public RangeReply getKeyRange(String columnFamily, final String startWith, final String stopAt, int maxResults) - throws IOException, ExecutionException, InterruptedException - { - assert getColumnFamilyStore(columnFamily) != null : columnFamily; - - getColumnFamilyStore(columnFamily).getReadLock().lock(); - try - { - return getKeyRangeUnsafe(columnFamily, startWith, stopAt, maxResults); - } - finally - { - getColumnFamilyStore(columnFamily).getReadLock().unlock(); - } - } - - private RangeReply getKeyRangeUnsafe(final String cfName, final String startWith, final String stopAt, int maxResults) throws IOException, ExecutionException, InterruptedException - { - // (OPP key decoration is a no-op so using the "decorated" comparator against raw keys is fine) - final Comparator comparator = StorageService.getPartitioner().getDecoratedKeyComparator(); - - // create a CollatedIterator that will return unique keys from different sources - // (current memtable, historical memtables, and SSTables) in the correct order. - List> iterators = new ArrayList>(); - ColumnFamilyStore cfs = getColumnFamilyStore(cfName); - - // we iterate through memtables with a priority queue to avoid more sorting than necessary. - // this predicate throws out the keys before the start of our range. - Predicate p = new Predicate() - { - public boolean evaluate(Object key) - { - String st = (String)key; - return comparator.compare(startWith, st) <= 0 && (stopAt.isEmpty() || comparator.compare(st, stopAt) <= 0); - } - }; - - // current memtable keys. have to go through the CFS api for locking. - iterators.add(IteratorUtils.filteredIterator(cfs.memtableKeyIterator(), p)); - // historical memtables - for (Memtable memtable : ColumnFamilyStore.getUnflushedMemtables(cfName)) - { - iterators.add(IteratorUtils.filteredIterator(Memtable.getKeyIterator(memtable.getKeys()), p)); - } - - // sstables - for (SSTableReader sstable : cfs.getSSTables()) - { - FileStruct fs = sstable.getFileStruct(); - fs.seekTo(startWith); - iterators.add(fs); - } - - Iterator collated = IteratorUtils.collatedIterator(comparator, iterators); - Iterable reduced = new ReducingIterator(collated) { - String current; - - public void reduce(String current) - { - this.current = current; - } - - protected String getReduced() - { - return current; - } - }; - - try - { - // pull keys out of the CollatedIterator. checking tombstone status is expensive, - // so we set an arbitrary limit on how many we'll do at once. - List keys = new ArrayList(); - boolean rangeCompletedLocally = false; - for (String current : reduced) - { - if (!stopAt.isEmpty() && comparator.compare(stopAt, current) < 0) - { - rangeCompletedLocally = true; - break; - } - // make sure there is actually non-tombstone content associated w/ this key - // TODO record the key source(s) somehow and only check that source (e.g., memtable or sstable) - QueryFilter filter = new SliceQueryFilter(current, new QueryPath(cfName), ArrayUtils.EMPTY_BYTE_ARRAY, ArrayUtils.EMPTY_BYTE_ARRAY, true, 1); - if (cfs.getColumnFamily(filter, Integer.MAX_VALUE) != null) - { - keys.add(current); - } - if (keys.size() >= maxResults) - { - rangeCompletedLocally = true; - break; - } - } - return new RangeReply(keys, rangeCompletedLocally); - } - finally - { - for (Iterator iter : iterators) - { - if (iter instanceof FileStruct) - { - ((FileStruct)iter).close(); - } - } - } - } - public static String getSnapshotPath(String dataDirPath, String tableName, String snapshotName) { return dataDirPath + File.separator + tableName + File.separator + SNAPSHOT_SUBDIR_NAME + File.separator + snapshotName; diff --git a/src/java/org/apache/cassandra/service/RangeVerbHandler.java b/src/java/org/apache/cassandra/service/RangeVerbHandler.java index 852a0bbae7..79713567a8 100644 --- a/src/java/org/apache/cassandra/service/RangeVerbHandler.java +++ b/src/java/org/apache/cassandra/service/RangeVerbHandler.java @@ -38,7 +38,7 @@ public class RangeVerbHandler implements IVerbHandler RangeCommand command = RangeCommand.read(message); Table table = Table.open(command.table); - RangeReply rangeReply = table.getKeyRange(command.columnFamily, command.startWith, command.stopAt, command.maxResults); + RangeReply rangeReply = table.getColumnFamilyStore(command.columnFamily).getKeyRange(command.startWith, command.stopAt, command.maxResults); Message response = rangeReply.getReply(message); if (logger.isDebugEnabled()) logger.debug("Sending " + rangeReply + " to " + message.getMessageId() + "@" + message.getFrom()); diff --git a/test/unit/org/apache/cassandra/db/CompactionsTest.java b/test/unit/org/apache/cassandra/db/CompactionsTest.java index 5e478c011a..fa6afd6201 100644 --- a/test/unit/org/apache/cassandra/db/CompactionsTest.java +++ b/test/unit/org/apache/cassandra/db/CompactionsTest.java @@ -52,7 +52,7 @@ public class CompactionsTest extends CleanupHelper inserted.add(key); } store.forceBlockingFlush(); - assertEquals(table.getKeyRange("Standard1", "", "", 10000).keys.size(), inserted.size()); + assertEquals(table.getColumnFamilyStore("Standard1").getKeyRange("", "", 10000).keys.size(), inserted.size()); } while (true) { @@ -64,6 +64,6 @@ public class CompactionsTest extends CleanupHelper { store.doCompaction(store.getSSTables().size()); } - assertEquals(table.getKeyRange("Standard1", "", "", 10000).keys.size(), inserted.size()); + assertEquals(table.getColumnFamilyStore("Standard1").getKeyRange("", "", 10000).keys.size(), inserted.size()); } } diff --git a/test/unit/org/apache/cassandra/db/OneCompactionTest.java b/test/unit/org/apache/cassandra/db/OneCompactionTest.java index b4bbc240c3..41fd40aad1 100644 --- a/test/unit/org/apache/cassandra/db/OneCompactionTest.java +++ b/test/unit/org/apache/cassandra/db/OneCompactionTest.java @@ -45,12 +45,12 @@ public class OneCompactionTest rm.apply(); inserted.add(key); store.forceBlockingFlush(); - assertEquals(inserted.size(), table.getKeyRange(columnFamilyName, "", "", 10000).keys.size()); + assertEquals(inserted.size(), table.getColumnFamilyStore(columnFamilyName).getKeyRange("", "", 10000).keys.size()); } Future ft = MinorCompactionManager.instance().submit(store, 2); ft.get(); assertEquals(1, store.getSSTables().size()); - assertEquals(table.getKeyRange(columnFamilyName, "", "", 10000).keys.size(), inserted.size()); + assertEquals(table.getColumnFamilyStore(columnFamilyName).getKeyRange("", "", 10000).keys.size(), inserted.size()); } @Test diff --git a/test/unit/org/apache/cassandra/db/RecoveryManager2Test.java b/test/unit/org/apache/cassandra/db/RecoveryManager2Test.java index fc2c744b57..2029498012 100644 --- a/test/unit/org/apache/cassandra/db/RecoveryManager2Test.java +++ b/test/unit/org/apache/cassandra/db/RecoveryManager2Test.java @@ -34,7 +34,7 @@ public class RecoveryManager2Test extends CleanupHelper table1.getColumnFamilyStore("Standard1").clearUnsafe(); RecoveryManager.doRecovery(); - Set foundKeys = new HashSet(table1.getKeyRange("Standard1", "", "", 1000).keys); + Set foundKeys = new HashSet(table1.getColumnFamilyStore("Standard1").getKeyRange("", "", 1000).keys); assert keys.equals(foundKeys); } }