cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jbel...@apache.org
Subject svn commit: r1089920 - in /cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service: RangeSliceResponseResolver.java StorageProxy.java
Date Thu, 07 Apr 2011 15:53:09 GMT
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);
         }



Mime
View raw message