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

remove unnecessary copy during range/index scans 

patch by jbellis; reviewed by slebresne for CASSANDRA-2425

git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1089920 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2011-04-07 15:53:08 +00:00
parent cc0322fef8
commit 381401af70
2 changed files with 19 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java
View File

@ -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.ReducingIterator;
* 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 implements IResponseResolver<List<Row>>
// 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 implements IResponseResolver<List<Row>>
}
// 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 implements IResponseResolver<List<Row>>
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)

32
src/java/org/apache/cassandra/service/StorageProxy.java
View File

@ -512,7 +512,7 @@ public class StorageProxy implements StorageProxyMBean
// 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 StorageProxyMBean
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 StorageProxyMBean
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 StorageProxyMBean
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);
}