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

Checking if an unlogged batch is local is inefficient 

patch by Stefania Alborghetti; reviewed by Paulo Motta for CASSANDRA-11529
This commit is contained in:
Stefania Alborghetti 2016-04-08 11:52:17 +08:00 committed by Aleksey Yeschenko
parent 2dd244b439
commit c1b1d3bccf
5 changed files with 21 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
2.1.14
* Checking if an unlogged batch is local is inefficient (CASSANDRA-11529)
* Fix paging for COMPACT tables without clustering columns (CASSANDRA-11467)
* Fix out-of-space error treatment in memtable flushing (CASSANDRA-11448)
* Backport CASSANDRA-10859 (CASSANDRA-11415)

4
conf/cassandra.yaml
View File

@ -555,6 +555,10 @@ column_index_size_in_kb: 64
# Caution should be taken on increasing the size of this threshold as it can lead to node instability.
batch_size_warn_threshold_in_kb: 5
# Log WARN on any batches not of type LOGGED than span across more partitions than this limit
unlogged_batch_across_partitions_warn_threshold: 10
# Number of simultaneous compactions to allow, NOT including
# validation "compactions" for anti-entropy repair. Simultaneous
# compactions can help preserve read performance in a mixed read/write

1
src/java/org/apache/cassandra/config/Config.java
View File

@ -144,6 +144,7 @@ public class Config
/* if the size of columns or super-columns are more than this, indexing will kick in */
public Integer column_index_size_in_kb = 64;
public Integer batch_size_warn_threshold_in_kb = 5;
public Integer unlogged_batch_across_partitions_warn_threshold = 10;
public Integer concurrent_compactors;
public volatile Integer compaction_throughput_mb_per_sec = 16;
public volatile Integer compaction_large_partition_warning_threshold_mb = 100;

5
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -860,6 +860,11 @@ public class DatabaseDescriptor
return conf.batch_size_warn_threshold_in_kb * 1024;
}
public static int getUnloggedBatchAcrossPartitionsWarnThreshold()
{
return conf.unlogged_batch_across_partitions_warn_threshold;
}
public static Collection<String> getInitialTokens()
{
return tokensFromString(System.getProperty("cassandra.initial_token", conf.initial_token));

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

@ -33,13 +33,10 @@ import org.apache.cassandra.config.ColumnDefinition;
import org.apache.cassandra.cql3.*;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.composites.Composite;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.*;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.QueryState;
import org.apache.cassandra.service.StorageProxy;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.transport.messages.ResultMessage;
import org.apache.cassandra.utils.NoSpamLogger;
import org.apache.cassandra.utils.Pair;
@ -60,7 +57,8 @@ 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.";
private static final String unloggedBatchWarning = "Unlogged batch covering {} partitions 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
@ -263,40 +261,24 @@ public class BatchStatement implements CQLStatement
Set<String> ksCfPairs = new HashSet<>();
Set<ByteBuffer> keySet = new HashSet<>();
Map<String, Collection<Range<Token>>> localTokensByKs = new HashMap<>();
boolean localMutationsOnly = true;
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));
if (localMutationsOnly)
localMutationsOnly &= isMutationLocal(localTokensByKs, im);
}
// CASSANDRA-9303: If we only have local mutations we do not warn
if (localMutationsOnly)
return;
NoSpamLogger.log(logger, NoSpamLogger.Level.WARN, 1, TimeUnit.MINUTES, unloggedBatchWarning,
keySet.size(), keySet.size() == 1 ? "" : "s",
ksCfPairs.size() == 1 ? "" : "s", ksCfPairs);
// CASSANDRA-11529: log only if we have more than a threshold of keys, this was also suggested in the
// original ticket that introduced this warning, CASSANDRA-9282
if (keySet.size() > DatabaseDescriptor.getUnloggedBatchAcrossPartitionsWarnThreshold())
{
NoSpamLogger.log(logger, NoSpamLogger.Level.WARN, 1, TimeUnit.MINUTES, unloggedBatchWarning,
keySet.size(), ksCfPairs.size() == 1 ? "" : "s", ksCfPairs);
}
}
}
private boolean isMutationLocal(Map<String, Collection<Range<Token>>> localTokensByKs, IMutation mutation)
{
Collection<Range<Token>> localRanges = localTokensByKs.get(mutation.getKeyspaceName());
if (localRanges == null)
{
localRanges = StorageService.instance.getLocalRanges(mutation.getKeyspaceName());
localTokensByKs.put(mutation.getKeyspaceName(), localRanges);
}
return Range.isInRanges(StorageService.getPartitioner().getToken(mutation.key()), localRanges);
}
public ResultMessage execute(QueryState queryState, QueryOptions options) throws RequestExecutionException, RequestValidationException
{