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

Merge branch 'cassandra-2.1' into trunk 

Conflicts:
	src/java/org/apache/cassandra/cql3/statements/BatchStatement.java
This commit is contained in:
T Jake Luciani 2015-05-15 09:02:44 -04:00
parent 1644e82f75 6544fb7aaa
commit bf208377a1
3 changed files with 71 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -107,6 +107,7 @@
2.1.6
* Warn on misuse of unlogged batches (CASSANDRA-9282)
* Failure detector detects and ignores local pauses (CASSANDRA-9183)
* Add utility class to support for rate limiting a given log statement (CASSANDRA-9029)
* Add missing consistency levels to cassandra-stess (CASSANDRA-9361)

43
src/java/org/apache/cassandra/cql3/statements/BatchStatement.java
View File

@ -19,6 +19,7 @@ package org.apache.cassandra.cql3.statements;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.TimeUnit;
import com.google.common.base.Function;
import com.google.common.collect.Iterables;
@ -26,6 +27,7 @@ import com.google.common.collect.Iterables;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.helpers.MessageFormatter;
import org.apache.cassandra.config.ColumnDefinition;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.*;
@ -38,10 +40,10 @@ import org.apache.cassandra.service.QueryState;
import org.apache.cassandra.service.StorageProxy;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.transport.messages.ResultMessage;
import org.apache.cassandra.utils.NoSpamLogger;
/**
* A <code>BATCH</code> statement parsed from a CQL query.
*
*/
public class BatchStatement implements CQLStatement
{
@ -56,14 +58,15 @@ public class BatchStatement implements CQLStatement
private final Attributes attrs;
private final boolean hasConditions;
private static final Logger logger = LoggerFactory.getLogger(BatchStatement.class);
private static final String unloggedBatchWarning = "Unlogged batch covering {} partition{} detected against table{} {}. You should use a logged batch for atomicity, or asynchronous writes for performance.";
/**
* Creates a new BatchStatement from a list of statements and a
* Thrift consistency level.
*
* @param type type of the batch
* @param type type of the batch
* @param statements a list of UpdateStatements
* @param attrs additional attributes for statement (CL, timestamp, timeToLive)
* @param attrs additional attributes for statement (CL, timestamp, timeToLive)
*/
public BatchStatement(int boundTerms, Type type, List<ModificationStatement> statements, Attributes attrs)
{
@ -181,13 +184,16 @@ public class BatchStatement implements CQLStatement
private Collection<? extends IMutation> unzipMutations(Map<String, Map<ByteBuffer, IMutation>> mutations)
{
// The case where all statement where on the same keyspace is pretty common
if (mutations.size() == 1)
return mutations.values().iterator().next().values();
List<IMutation> ms = new ArrayList<>();
for (Map<ByteBuffer, IMutation> ksMap : mutations.values())
ms.addAll(ksMap.values());
return ms;
}
@ -225,7 +231,7 @@ public class BatchStatement implements CQLStatement
}
else
{
mut = statement.cfm.isCounter() ? ((CounterMutation)mutation).getMutation() : (Mutation)mutation;
mut = statement.cfm.isCounter() ? ((CounterMutation) mutation).getMutation() : (Mutation) mutation;
}
statement.addUpdateForKey(mut.addOrGet(statement.cfm), key, clusteringPrefix, params);
@ -234,6 +240,7 @@ public class BatchStatement implements CQLStatement
/**
* Checks batch size to ensure threshold is met. If not, a warning is logged.
*
* @param cfs ColumnFamilies that will store the batch's mutations.
*/
public static void verifyBatchSize(Iterable<ColumnFamily> cfs) throws InvalidRequestException
@ -249,7 +256,7 @@ public class BatchStatement implements CQLStatement
{
Set<String> ksCfPairs = new HashSet<>();
for (ColumnFamily cf : cfs)
ksCfPairs.add(cf.metadata().ksName + "." + cf.metadata().cfName);
ksCfPairs.add(String.format("%s.%s", cf.metadata().ksName, cf.metadata().cfName));
String format = "Batch of prepared statements for {} is of size {}, exceeding specified threshold of {} by {}.{}";
if (size > failThreshold)
@ -266,6 +273,30 @@ public class BatchStatement implements CQLStatement
}
}
private void verifyBatchType(Collection<? extends IMutation> mutations)
{
if (type != Type.LOGGED && mutations.size() > 1)
{
Set<String> ksCfPairs = new HashSet<>();
Set<ByteBuffer> keySet = new HashSet<>();
for (IMutation im : mutations)
{
keySet.add(im.key());
for (ColumnFamily cf : im.getColumnFamilies())
ksCfPairs.add(String.format("%s.%s", cf.metadata().ksName, cf.metadata().cfName));
}
NoSpamLogger.log(logger, NoSpamLogger.Level.WARN, 1, TimeUnit.MINUTES, unloggedBatchWarning,
keySet.size(), keySet.size() == 1 ? "" : "s",
ksCfPairs.size() == 1 ? "" : "s", ksCfPairs);
ClientWarn.warn(MessageFormatter.arrayFormat(unloggedBatchWarning, new Object[]{keySet.size(), keySet.size() == 1 ? "" : "s",
ksCfPairs.size() == 1 ? "" : "s", ksCfPairs}).getMessage());
}
}
public ResultMessage execute(QueryState queryState, QueryOptions options) throws RequestExecutionException, RequestValidationException
{
return execute(queryState, BatchQueryOptions.withoutPerStatementVariables(options));
@ -301,7 +332,9 @@ public class BatchStatement implements CQLStatement
return im.getColumnFamilies();
}
}));
verifyBatchSize(cfs);
verifyBatchType(mutations);
boolean mutateAtomic = (type == Type.LOGGED && mutations.size() > 1);
StorageProxy.mutateWithTriggers(mutations, cl, mutateAtomic);

32
test/unit/org/apache/cassandra/service/ClientWarningsTest.java
View File

@ -41,6 +41,26 @@ public class ClientWarningsTest extends CQLTester
DatabaseDescriptor.setBatchSizeWarnThresholdInKB(1);
}
@Test
public void testUnloggedBatchWithProtoV4() throws Exception
{
createTable("CREATE TABLE %s (pk int PRIMARY KEY, v text)");
try (SimpleClient client = new SimpleClient(nativeAddr.getHostAddress(), nativePort, Server.VERSION_4))
{
client.connect(false);
QueryMessage query = new QueryMessage(createBatchStatement2(1), QueryOptions.DEFAULT);
Message.Response resp = client.execute(query);
assertEquals(1, resp.getWarnings().size());
query = new QueryMessage(createBatchStatement2(DatabaseDescriptor.getBatchSizeWarnThreshold()), QueryOptions.DEFAULT);
resp = client.execute(query);
assertEquals(2, resp.getWarnings().size());
}
}
@Test
public void testLargeBatchWithProtoV4() throws Exception
{
@ -78,4 +98,16 @@ public class ClientWarningsTest extends CQLTester
currentTable(),
StringUtils.repeat('1', minSize));
}
private String createBatchStatement2(int minSize)
{
return String.format("BEGIN UNLOGGED BATCH INSERT INTO %s.%s (pk, v) VALUES (1, '%s'); INSERT INTO %s.%s (pk, v) VALUES (2, '%s'); APPLY BATCH;",
KEYSPACE,
currentTable(),
StringUtils.repeat('1', minSize),
KEYSPACE,
currentTable(),
StringUtils.repeat('1', minSize));
}
}