merge from 0.7

git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1090062 13f79535-47bb-0310-9956-ffa450edef68

conf/cassandra-env.sh

@@ -130,7 +130,7 @@ JVM_OPTS="$JVM_OPTS -XX:+UseCMSInitiatingOccupancyOnly"
 # JVM_OPTS="$JVM_OPTS -XX:+PrintClassHistogram"
 # JVM_OPTS="$JVM_OPTS -XX:+PrintTenuringDistribution"
 # JVM_OPTS="$JVM_OPTS -XX:+PrintGCApplicationStoppedTime"
-# JVM_OPTS="$JVM_OPTS -Xloggc:/var/log/cassandra/gc.log"
+# JVM_OPTS="$JVM_OPTS -Xloggc:/var/log/cassandra/gc-`date +%s`.log"
 
 # uncomment to have Cassandra JVM listen for remote debuggers/profilers on port 1414
 # JVM_OPTS="$JVM_OPTS -Xdebug -Xnoagent -Xrunjdwp:transport=dt_socket,server=y,suspend=n,address=1414"

contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java

@@ -68,7 +68,7 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo
     public final static String PIG_INITIAL_ADDRESS = "PIG_INITIAL_ADDRESS";
     public final static String PIG_PARTITIONER = "PIG_PARTITIONER";
 
-    private static String UDFCONTEXT_SCHEMA_KEY = "schema";
+    private static String UDFCONTEXT_SCHEMA_KEY = "cassandra.schema";
 
     private final static ByteBuffer BOUND = ByteBufferUtil.EMPTY_BYTE_BUFFER;
     private static final Log logger = LogFactory.getLog(CassandraStorage.class);
@@ -168,7 +168,7 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo
     private CfDef getCfDef()
     {
         UDFContext context = UDFContext.getUDFContext();
-        Properties property = context.getUDFProperties(ResourceSchema.class);
+        Properties property = context.getUDFProperties(CassandraStorage.class);
         return cfdefFromString(property.getProperty(UDFCONTEXT_SCHEMA_KEY));
     }
 
@@ -314,6 +314,7 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo
         setLocationFromUri(location);
         ConfigHelper.setOutputColumnFamily(conf, keyspace, column_family);
         setConnectionInformation();
+        initSchema();
     }
 
     public OutputFormat getOutputFormat()
@@ -442,6 +443,12 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo
     /* Methods to get the column family schema from Cassandra */
 
     private void initSchema()
+    {
+        UDFContext context = UDFContext.getUDFContext();
+        Properties property = context.getUDFProperties(CassandraStorage.class);
+        
+        // Only get the schema if we haven't already gotten it
+        if (!property.containsKey(UDFCONTEXT_SCHEMA_KEY))
         {
             Cassandra.Client client = null;
             try
@@ -459,8 +466,6 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo
                         break;
                     }
                 }
-            UDFContext context = UDFContext.getUDFContext();
-            Properties property = context.getUDFProperties(ResourceSchema.class);
                 property.setProperty(UDFCONTEXT_SCHEMA_KEY, cfdefToString(cfDef));
             }
             catch (TException e)
@@ -480,6 +485,7 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo
                 throw new RuntimeException(e);
             }
         }
+    }
 
     private static Cassandra.Client createConnection(String host, Integer port, boolean framed) throws IOException
     {

src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java

@@ -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)

src/java/org/apache/cassandra/service/StorageProxy.java

@@ -691,7 +691,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 : liveEndpoints)
                     {
@@ -700,24 +700,21 @@ public class StorageProxy implements StorageProxyMBean
                             logger.debug("reading " + c2 + " from " + endpoint);
                     }
 
-                    // if we're done, great, otherwise, move to the next range
                     try
-                    {
-                        if (logger.isDebugEnabled())
                     {
                         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;
             }
@@ -976,7 +973,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);
@@ -988,21 +985,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);
         }

src/java/org/apache/cassandra/utils/FBUtilities.java

@@ -602,7 +602,7 @@ public class FBUtilities
         T rval = null;
         try
         {
-            rval = (T) cls.getDeclaredMethod("getInstance").invoke(null, (Object) null);
+            rval = (T) cls.getDeclaredMethod("getInstance").invoke(new Object[] {null, null});
 
         }
         catch (NoSuchMethodException e)