diff --git a/conf/cassandra-env.sh b/conf/cassandra-env.sh index f9930e484d..517a6c442d 100644 --- a/conf/cassandra-env.sh +++ b/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" diff --git a/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java b/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java index 04a6c3ea2d..d5906f82aa 100644 --- a/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java +++ b/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() @@ -443,41 +444,46 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo private void initSchema() { - Cassandra.Client client = null; - try + 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)) { - client = createConnection(ConfigHelper.getInitialAddress(conf), ConfigHelper.getRpcPort(conf), true); - CfDef cfDef = null; - client.set_keyspace(keyspace); - KsDef ksDef = client.describe_keyspace(keyspace); - List defs = ksDef.getCf_defs(); - for (CfDef def : defs) + Cassandra.Client client = null; + try { - if (column_family.equalsIgnoreCase(def.getName())) + client = createConnection(ConfigHelper.getInitialAddress(conf), ConfigHelper.getRpcPort(conf), true); + CfDef cfDef = null; + client.set_keyspace(keyspace); + KsDef ksDef = client.describe_keyspace(keyspace); + List defs = ksDef.getCf_defs(); + for (CfDef def : defs) { - cfDef = def; - break; + if (column_family.equalsIgnoreCase(def.getName())) + { + cfDef = def; + break; + } } + property.setProperty(UDFCONTEXT_SCHEMA_KEY, cfdefToString(cfDef)); + } + catch (TException e) + { + throw new RuntimeException(e); + } + catch (InvalidRequestException e) + { + throw new RuntimeException(e); + } + catch (NotFoundException e) + { + throw new RuntimeException(e); + } + catch (IOException e) + { + throw new RuntimeException(e); } - UDFContext context = UDFContext.getUDFContext(); - Properties property = context.getUDFProperties(ResourceSchema.class); - property.setProperty(UDFCONTEXT_SCHEMA_KEY, cfdefToString(cfDef)); - } - catch (TException e) - { - throw new RuntimeException(e); - } - catch (InvalidRequestException e) - { - throw new RuntimeException(e); - } - catch (NotFoundException e) - { - throw new RuntimeException(e); - } - catch (IOException e) - { - throw new RuntimeException(e); } } diff --git a/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java b/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java index dcc0d209ca..521035c9d5 100644 --- a/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java +++ b/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> +public class RangeSliceResponseResolver implements IResponseResolver> { private static final Logger logger_ = LoggerFactory.getLogger(RangeSliceResponseResolver.class); private final String table; @@ -62,7 +64,7 @@ public class RangeSliceResponseResolver implements IResponseResolver> // 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 resolve() throws IOException + public Iterable resolve() throws IOException { CollatingIterator collator = new CollatingIterator(new Comparator>() { @@ -81,7 +83,8 @@ public class RangeSliceResponseResolver implements IResponseResolver> } // for each row, compute the combination of all different versions seen, and repair incomplete versions - ReducingIterator, Row> iter = new ReducingIterator, Row>(collator) + + return new ReducingIterator, Row>(collator) { List versions = new ArrayList(sources.size()); List versionSources = new ArrayList(sources.size()); @@ -109,12 +112,6 @@ public class RangeSliceResponseResolver implements IResponseResolver> return new Row(key, resolved); } }; - - List resolvedRows = new ArrayList(n); - while (iter.hasNext()) - resolvedRows.add(iter.next()); - - return resolvedRows; } public void preprocess(Message message) diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index 14eb84f38e..720a947073 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/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> handler = getReadCallback(resolver, command, consistency_level, liveEndpoints); + ReadCallback> handler = getReadCallback(resolver, command, consistency_level, liveEndpoints); handler.assureSufficientLiveNodes(); for (InetAddress endpoint : liveEndpoints) { @@ -700,17 +700,13 @@ 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()) { - 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) { @@ -718,6 +714,7 @@ public class StorageProxy implements StorageProxyMBean } } + // 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> handler = getReadCallback(resolver, iCommand, consistency_level, liveEndpoints); + ReadCallback> 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 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); } diff --git a/src/java/org/apache/cassandra/utils/FBUtilities.java b/src/java/org/apache/cassandra/utils/FBUtilities.java index de52ab9580..8f87ed332b 100644 --- a/src/java/org/apache/cassandra/utils/FBUtilities.java +++ b/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)