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

Make it possible to set the max/min compaction thresholds 

Patch by Marcus Eriksson, reviewed by Alex Petrov for CASSANDRA-14388
This commit is contained in:
Marcus Eriksson 2018-04-16 16:14:28 +02:00 committed by Alex Petrov
parent c6570fac18
commit d52bdaefda
5 changed files with 88 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
NEWS.txt
View File

@ -38,6 +38,11 @@ using the provided 'sstableupgrade' tool.
New features
------------
- LCS now respects the max_threshold parameter when compacting - this was hard coded to 32
before, but now it is possible to do bigger compactions when compacting from L0 to L1.
This also applies to STCS-compactions in L0 - if there are more than 32 sstables in L0
we will compact at most max_threshold sstables in an L0 STCS compaction. See CASSANDRA-14388
for more information.
- There is now an option to automatically upgrade sstables after Cassandra upgrade, enable
either in `cassandra.yaml:automatic_sstable_upgrade` or via JMX during runtime. See
CASSANDRA-14197.

4
src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java
View File

@ -31,6 +31,7 @@ import org.slf4j.LoggerFactory;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.JsonNodeFactory;
import com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.cassandra.schema.CompactionParams;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.db.ColumnFamilyStore;
@ -569,6 +570,9 @@ public class LeveledCompactionStrategy extends AbstractCompactionStrategy
uncheckedOptions.remove(LEVEL_FANOUT_SIZE_OPTION);
uncheckedOptions.remove(CompactionParams.Option.MIN_THRESHOLD.toString());
uncheckedOptions.remove(CompactionParams.Option.MAX_THRESHOLD.toString());
uncheckedOptions = SizeTieredCompactionStrategyOptions.validateOptions(options, uncheckedOptions);
return uncheckedOptions;

22
src/java/org/apache/cassandra/db/compaction/LeveledManifest.java
View File

@ -49,9 +49,8 @@ public class LeveledManifest
private static final Logger logger = LoggerFactory.getLogger(LeveledManifest.class);
/**
* limit the number of L0 sstables we do at once, because compaction bloom filter creation
* uses a pessimistic estimate of how many keys overlap (none), so we risk wasting memory
* or even OOMing when compacting highly overlapping sstables
* if we have more than MAX_COMPACTING_L0 sstables in L0, we will run a round of STCS with at most
* cfs.getMaxCompactionThreshold() sstables.
*/
private static final int MAX_COMPACTING_L0 = 32;
/**
@ -163,6 +162,7 @@ public class LeveledManifest
// The add(..):ed sstable will be sent to level 0
try
{
logger.debug("Could not add sstable {} in level {} - dropping to 0", reader, reader.getSSTableLevel());
reader.descriptor.getMetadataSerializer().mutateLevel(reader.descriptor, 0);
reader.reloadSSTableMetadata();
}
@ -346,12 +346,12 @@ public class LeveledManifest
// L2: 12 [ideal: 100]
//
// The problem is that L0 has a much higher score (almost 250) than L1 (11), so what we'll
// do is compact a batch of MAX_COMPACTING_L0 sstables with all 117 L1 sstables, and put the
// result (say, 120 sstables) in L1. Then we'll compact the next batch of MAX_COMPACTING_L0,
// do is compact a batch of cfs.getMaximumCompactionThreshold() sstables with all 117 L1 sstables, and put the
// result (say, 120 sstables) in L1. Then we'll compact the next batch of cfs.getMaxCompactionThreshold(),
// and so forth. So we spend most of our i/o rewriting the L1 data with each batch.
//
// If we could just do *all* L0 a single time with L1, that would be ideal. But we can't
// -- see the javadoc for MAX_COMPACTING_L0.
// since we might run out of memory
//
// LevelDB's way around this is to simply block writes if L0 compaction falls behind.
// We don't have that luxury.
@ -667,7 +667,7 @@ public class LeveledManifest
// 1a. add sstables to the candidate set until we have at least maxSSTableSizeInMB
// 1b. prefer choosing older sstables as candidates, to newer ones
// 1c. any L0 sstables that overlap a candidate, will also become candidates
// 2. At most MAX_COMPACTING_L0 sstables from L0 will be compacted at once
// 2. At most max_threshold sstables from L0 will be compacted at once
// 3. If total candidate size is less than maxSSTableSizeInMB, we won't bother compacting with L1,
// and the result of the compaction will stay in L0 instead of being promoted (see promote())
//
@ -693,10 +693,10 @@ public class LeveledManifest
remaining.remove(newCandidate);
}
if (candidates.size() > MAX_COMPACTING_L0)
if (candidates.size() > cfs.getMaximumCompactionThreshold())
{
// limit to only the MAX_COMPACTING_L0 oldest candidates
candidates = new HashSet<>(ageSortedSSTables(candidates).subList(0, MAX_COMPACTING_L0));
// limit to only the cfs.getMaximumCompactionThreshold() oldest candidates
candidates = new HashSet<>(ageSortedSSTables(candidates).subList(0, cfs.getMaximumCompactionThreshold()));
break;
}
}
@ -820,7 +820,7 @@ public class LeveledManifest
if (!DatabaseDescriptor.getDisableSTCSInL0() && getLevel(0).size() > MAX_COMPACTING_L0)
{
int l0compactions = getLevel(0).size() / MAX_COMPACTING_L0;
int l0compactions = getLevel(0).size() / cfs.getMaximumCompactionThreshold();
tasks += l0compactions;
estimated[0] += l0compactions;
}

2
test/unit/org/apache/cassandra/db/TableCQLHelperTest.java
View File

@ -274,7 +274,7 @@ public class TableCQLHelperTest extends CQLTester
"\tAND speculative_retry = 'ALWAYS'\n" +
"\tAND comment = 'comment'\n" +
"\tAND caching = { 'keys': 'ALL', 'rows_per_partition': 'NONE' }\n" +
"\tAND compaction = { 'class': 'org.apache.cassandra.db.compaction.LeveledCompactionStrategy', 'sstable_size_in_mb': '1' }\n" +
"\tAND compaction = { 'max_threshold': '32', 'min_threshold': '4', 'sstable_size_in_mb': '1', 'class': 'org.apache.cassandra.db.compaction.LeveledCompactionStrategy' }\n" +
"\tAND compression = { 'chunk_length_in_kb': '64', 'min_compress_ratio': '2.0', 'class': 'org.apache.cassandra.io.compress.LZ4Compressor' }\n" +
"\tAND cdc = false\n" +
"\tAND extensions = { 'ext1': 0x76616c31 };"

67
test/unit/org/apache/cassandra/db/compaction/CompactionsCQLTest.java
View File

@ -17,10 +17,12 @@
*/
package org.apache.cassandra.db.compaction;
import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import org.apache.commons.lang.StringUtils;
import org.junit.After;
@ -31,11 +33,13 @@ import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.DeletionTime;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.RangeTombstone;
import org.apache.cassandra.db.RowUpdateBuilder;
import org.apache.cassandra.db.Slice;
import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.rows.Cell;
import org.apache.cassandra.db.rows.Row;
@ -388,6 +392,69 @@ public class CompactionsCQLTest extends CQLTester
DatabaseDescriptor.setColumnIndexSize(maxSizePre);
}
@Test
public void testLCSThresholdParams() throws Throwable
{
createTable("create table %s (id int, id2 int, t blob, primary key (id, id2)) with compaction = {'class':'LeveledCompactionStrategy', 'sstable_size_in_mb':'1', 'max_threshold':'60'}");
ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
cfs.disableAutoCompaction();
byte [] b = new byte[100 * 1024];
new Random().nextBytes(b);
ByteBuffer value = ByteBuffer.wrap(b);
for (int i = 0; i < 50; i++)
{
for (int j = 0; j < 10; j++)
{
execute("insert into %s (id, id2, t) values (?, ?, ?)", i, j, value);
}
cfs.forceBlockingFlush();
}
assertEquals(50, cfs.getLiveSSTables().size());
LeveledCompactionStrategy lcs = (LeveledCompactionStrategy) cfs.getCompactionStrategyManager().getUnrepaired().get(0);
AbstractCompactionTask act = lcs.getNextBackgroundTask(0);
// we should be compacting all 50 sstables:
assertEquals(50, act.transaction.originals().size());
act.execute(null);
}
@Test
public void testSTCSinL0() throws Throwable
{
createTable("create table %s (id int, id2 int, t blob, primary key (id, id2)) with compaction = {'class':'LeveledCompactionStrategy', 'sstable_size_in_mb':'1', 'max_threshold':'60'}");
ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
cfs.disableAutoCompaction();
execute("insert into %s (id, id2, t) values (?, ?, ?)", 1,1,"L1");
cfs.forceBlockingFlush();
cfs.forceMajorCompaction();
SSTableReader l1sstable = cfs.getLiveSSTables().iterator().next();
assertEquals(1, l1sstable.getSSTableLevel());
// now we have a single L1 sstable, create many L0 ones:
byte [] b = new byte[100 * 1024];
new Random().nextBytes(b);
ByteBuffer value = ByteBuffer.wrap(b);
for (int i = 0; i < 50; i++)
{
for (int j = 0; j < 10; j++)
{
execute("insert into %s (id, id2, t) values (?, ?, ?)", i, j, value);
}
cfs.forceBlockingFlush();
}
assertEquals(51, cfs.getLiveSSTables().size());
// mark the L1 sstable as compacting to make sure we trigger STCS in L0:
LifecycleTransaction txn = cfs.getTracker().tryModify(l1sstable, OperationType.COMPACTION);
LeveledCompactionStrategy lcs = (LeveledCompactionStrategy) cfs.getCompactionStrategyManager().getUnrepaired().get(0);
AbstractCompactionTask act = lcs.getNextBackgroundTask(0);
// note that max_threshold is 60 (more than the amount of L0 sstables), but MAX_COMPACTING_L0 is 32, which means we will trigger STCS with at most max_threshold sstables
assertEquals(50, act.transaction.originals().size());
assertEquals(0, ((LeveledCompactionTask)act).getLevel());
assertTrue(act.transaction.originals().stream().allMatch(s -> s.getSSTableLevel() == 0));
txn.abort(); // unmark the l1 sstable compacting
act.execute(null);
}
private void prepareWide() throws Throwable
{
createTable("CREATE TABLE %s (id int, id2 int, b text, primary key (id, id2))");