diff --git a/NEWS.txt b/NEWS.txt index da80684b97..e1e76dec39 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -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. diff --git a/src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java index 7a7b1c1d85..b1091ce070 100644 --- a/src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java +++ b/src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java @@ -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; diff --git a/src/java/org/apache/cassandra/db/compaction/LeveledManifest.java b/src/java/org/apache/cassandra/db/compaction/LeveledManifest.java index 013f0b4a22..291973f90b 100644 --- a/src/java/org/apache/cassandra/db/compaction/LeveledManifest.java +++ b/src/java/org/apache/cassandra/db/compaction/LeveledManifest.java @@ -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; } diff --git a/test/unit/org/apache/cassandra/db/TableCQLHelperTest.java b/test/unit/org/apache/cassandra/db/TableCQLHelperTest.java index 1e465b3e1e..701afbb7ef 100644 --- a/test/unit/org/apache/cassandra/db/TableCQLHelperTest.java +++ b/test/unit/org/apache/cassandra/db/TableCQLHelperTest.java @@ -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 };" diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionsCQLTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionsCQLTest.java index ca420dabd5..c91d2fed83 100644 --- a/test/unit/org/apache/cassandra/db/compaction/CompactionsCQLTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/CompactionsCQLTest.java @@ -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))");