Author: jbellis
Date: Thu Apr 7 15:53:08 2011
New Revision: 1089920
URL: http://svn.apache.org/viewvc?rev=1089920&view=rev
Log:
remove unnecessary copy during range/index scans
patch by jbellis; reviewed by slebresne for CASSANDRA-2425
Modified:
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/StorageProxy.java
Modified:
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java
URL:
http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java?rev=1089920&r1=1089919&r2=1089920&view=diff
==============================================================================
---
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java
(original)
+++
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java
Thu Apr 7 15:53:08 2011
@@ -24,6 +24,8 @@ import java.util.*;
import java.util.concurrent.LinkedBlockingQueue;
import com.google.common.collect.AbstractIterator;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Iterators;
import org.apache.commons.collections.iterators.CollatingIterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -40,7 +42,7 @@ import org.apache.cassandra.utils.Reduci
* Turns RangeSliceReply objects into row (string -> CF) maps, resolving
* to the most recent ColumnFamily and setting up read repairs as necessary.
*/
-public class RangeSliceResponseResolver implements IResponseResolver<List<Row>>
+public class RangeSliceResponseResolver implements
IResponseResolver<Iterable<Row>>
{
private static final Logger logger_ =
LoggerFactory.getLogger(RangeSliceResponseResolver.class);
private final String table;
@@ -62,7 +64,7 @@ public class RangeSliceResponseResolver
// Note: this deserializes the response a 2nd time if getData was called
first
// (this is not currently an issue since we don't do read repair for range
queries.)
- public List<Row> resolve() throws IOException
+ public Iterable<Row> resolve() throws IOException
{
CollatingIterator collator = new CollatingIterator(new
Comparator<Pair<Row,InetAddress>>()
{
@@ -81,7 +83,8 @@ public class RangeSliceResponseResolver
}
// for each row, compute the combination of all different versions
seen, and repair incomplete versions
- ReducingIterator<Pair<Row,InetAddress>, Row> iter = new
ReducingIterator<Pair<Row,InetAddress>, Row>(collator)
+
+ return new ReducingIterator<Pair<Row,InetAddress>, Row>(collator)
{
List<ColumnFamily> versions = new
ArrayList<ColumnFamily>(sources.size());
List<InetAddress> versionSources = new
ArrayList<InetAddress>(sources.size());
@@ -109,12 +112,6 @@ public class RangeSliceResponseResolver
return new Row(key, resolved);
}
};
-
- List<Row> resolvedRows = new ArrayList<Row>(n);
- while (iter.hasNext())
- resolvedRows.add(iter.next());
-
- return resolvedRows;
}
public void preprocess(Message message)
Modified:
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/StorageProxy.java
URL:
http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/StorageProxy.java?rev=1089920&r1=1089919&r2=1089920&view=diff
==============================================================================
---
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/StorageProxy.java
(original)
+++
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/StorageProxy.java
Thu Apr 7 15:53:08 2011
@@ -512,7 +512,7 @@ public class StorageProxy implements Sto
// collect replies and resolve according to consistency
level
RangeSliceResponseResolver resolver = new
RangeSliceResponseResolver(command.keyspace, liveEndpoints);
- ReadCallback<List<Row>> handler =
getReadCallback(resolver, command, consistency_level, liveEndpoints);
+ ReadCallback<Iterable<Row>> handler =
getReadCallback(resolver, command, consistency_level, liveEndpoints);
handler.assureSufficientLiveNodes();
for (InetAddress endpoint : handler.endpoints)
{
@@ -521,24 +521,21 @@ public class StorageProxy implements Sto
logger.debug("reading " + c2 + " from " +
endpoint);
}
- // if we're done, great, otherwise, move to the next range
- try
+ try
{
- if (logger.isDebugEnabled())
+ for (Row row : handler.get())
{
- for (Row row : handler.get())
- {
- logger.debug("range slices read " + row.key);
- }
+ rows.add(row);
+ logger.debug("range slices read {}", row.key);
}
- rows.addAll(handler.get());
- }
+ }
catch (DigestMismatchException e)
{
throw new AssertionError(e); // no digests in range
slices yet
}
}
+ // if we're done, great, otherwise, move to the next range
if (rows.size() >= command.max_keys)
break;
}
@@ -771,7 +768,7 @@ public class StorageProxy implements Sto
return keyspace;
}
};
- ReadCallback<List<Row>> handler = getReadCallback(resolver,
iCommand, consistency_level, liveEndpoints);
+ ReadCallback<Iterable<Row>> handler = getReadCallback(resolver,
iCommand, consistency_level, liveEndpoints);
handler.assureSufficientLiveNodes();
IndexScanCommand command = new IndexScanCommand(keyspace,
column_family, index_clause, column_predicate, range);
@@ -783,21 +780,18 @@ public class StorageProxy implements Sto
logger.debug("reading " + command + " from " + endpoint);
}
- List<Row> theseRows;
try
{
- theseRows = handler.get();
+ for (Row row : handler.get())
+ {
+ rows.add(row);
+ logger.debug("read {}", row);
+ }
}
catch (DigestMismatchException e)
{
throw new RuntimeException(e);
}
- rows.addAll(theseRows);
- if (logger.isDebugEnabled())
- {
- for (Row row : theseRows)
- logger.debug("read " + row);
- }
if (rows.size() >= index_clause.count)
return rows.subList(0, index_clause.count);
}
