From cedcf07c542235815c023b66f151ad8c7aa9ba9a Mon Sep 17 00:00:00 2001 From: Marcus Eriksson Date: Mon, 7 Sep 2015 10:39:15 +0200 Subject: [PATCH] Do size tiered compaction in date tiered compaction windows Patch by marcuse; reviewed by Jeff Jirsa for CASSANDRA-10276 --- CHANGES.txt | 1 + .../DateTieredCompactionStrategy.java | 46 +++++++++------ .../DateTieredCompactionStrategyTest.java | 57 ++++++++++++++----- 3 files changed, 72 insertions(+), 32 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 2ca3b435e3..5b46eacd66 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 2.1.12 + * Do STCS in DTCS windows (CASSANDRA-10276) * Don't try to get ancestors from half-renamed sstables (CASSANDRA-10501) * Avoid repetition of JVM_OPTS in debian package (CASSANDRA-10251) * Fix potential NPE from handling result of SIM.highestSelectivityIndex (CASSANDRA-10550) diff --git a/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java index a8e2affbe0..ece596fe5e 100644 --- a/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java +++ b/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java @@ -18,7 +18,6 @@ package org.apache.cassandra.db.compaction; import java.util.*; -import java.util.concurrent.TimeUnit; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Predicate; @@ -40,6 +39,7 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy protected volatile int estimatedRemainingTasks; private final Set sstables = new HashSet<>(); private long lastExpiredCheck; + private final SizeTieredCompactionStrategyOptions stcsOptions; public DateTieredCompactionStrategy(ColumnFamilyStore cfs, Map options) { @@ -54,6 +54,7 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy else logger.debug("Enabling tombstone compactions for DTCS"); + this.stcsOptions = new SizeTieredCompactionStrategyOptions(options); } @Override @@ -137,7 +138,8 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy cfs.getMinimumCompactionThreshold(), cfs.getMaximumCompactionThreshold(), now, - options.baseTime); + options.baseTime, + stcsOptions); if (!mostInteresting.isEmpty()) return mostInteresting; return null; @@ -328,7 +330,7 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy for (List bucket : tasks) { if (bucket.size() >= cfs.getMinimumCompactionThreshold()) - n += Math.ceil((double)bucket.size() / cfs.getMaximumCompactionThreshold()); + n += getSTCSBuckets(bucket, stcsOptions).size(); } estimatedRemainingTasks = n; } @@ -341,7 +343,7 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy * @return a bucket (list) of sstables to compact. */ @VisibleForTesting - static List newestBucket(List> buckets, int minThreshold, int maxThreshold, long now, long baseTime) + static List newestBucket(List> buckets, int minThreshold, int maxThreshold, long now, long baseTime, SizeTieredCompactionStrategyOptions stcsOptions) { // If the "incoming window" has at least minThreshold SSTables, choose that one. // For any other bucket, at least 2 SSTables is enough. @@ -349,23 +351,31 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy Target incomingWindow = getInitialTarget(now, baseTime); for (List bucket : buckets) { - if (bucket.size() >= minThreshold || - (bucket.size() >= 2 && !incomingWindow.onTarget(bucket.get(0).getMinTimestamp()))) - return trimToThreshold(bucket, maxThreshold); + boolean inFirstWindow = incomingWindow.onTarget(bucket.get(0).getMinTimestamp()); + if (bucket.size() >= minThreshold || (bucket.size() >= 2 && !inFirstWindow)) + { + List stcsSSTables = getSSTablesForSTCS(bucket, inFirstWindow ? minThreshold : 2, maxThreshold, stcsOptions); + if (!stcsSSTables.isEmpty()) + return stcsSSTables; + } } return Collections.emptyList(); } - /** - * @param bucket list of sstables, ordered from newest to oldest by getMinTimestamp(). - * @param maxThreshold maximum number of sstables in a single compaction task. - * @return A bucket trimmed to the maxThreshold newest sstables. - */ - @VisibleForTesting - static List trimToThreshold(List bucket, int maxThreshold) + private static List getSSTablesForSTCS(Collection sstables, int minThreshold, int maxThreshold, SizeTieredCompactionStrategyOptions stcsOptions) { - // Trim the oldest sstables off the end to meet the maxThreshold - return bucket.subList(0, Math.min(bucket.size(), maxThreshold)); + List s = SizeTieredCompactionStrategy.mostInterestingBucket(getSTCSBuckets(sstables, stcsOptions), minThreshold, maxThreshold); + logger.debug("Got sstables {} for STCS from {}", s, sstables); + return s; + } + + private static List> getSTCSBuckets(Collection sstables, SizeTieredCompactionStrategyOptions stcsOptions) + { + List> pairs = SizeTieredCompactionStrategy.createSSTableAndLengthPairs(AbstractCompactionStrategy.filterSuspectSSTables(sstables)); + return SizeTieredCompactionStrategy.getBuckets(pairs, + stcsOptions.bucketHigh, + stcsOptions.bucketLow, + stcsOptions.minSSTableSize); } @Override @@ -375,7 +385,7 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy if (sstables == null) return null; - return Arrays.asList(new CompactionTask(cfs, sstables, gcBefore, false)); + return Collections.singleton(new CompactionTask(cfs, sstables, gcBefore, false)); } @Override @@ -411,6 +421,8 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy uncheckedOptions.remove(CFPropDefs.KW_MINCOMPACTIONTHRESHOLD); uncheckedOptions.remove(CFPropDefs.KW_MAXCOMPACTIONTHRESHOLD); + uncheckedOptions = SizeTieredCompactionStrategyOptions.validateOptions(options, uncheckedOptions); + return uncheckedOptions; } diff --git a/test/unit/org/apache/cassandra/db/compaction/DateTieredCompactionStrategyTest.java b/test/unit/org/apache/cassandra/db/compaction/DateTieredCompactionStrategyTest.java index f05bf449ee..368101d013 100644 --- a/test/unit/org/apache/cassandra/db/compaction/DateTieredCompactionStrategyTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/DateTieredCompactionStrategyTest.java @@ -39,7 +39,6 @@ import org.apache.cassandra.utils.Pair; import static org.apache.cassandra.db.compaction.DateTieredCompactionStrategy.getBuckets; import static org.apache.cassandra.db.compaction.DateTieredCompactionStrategy.newestBucket; -import static org.apache.cassandra.db.compaction.DateTieredCompactionStrategy.trimToThreshold; import static org.apache.cassandra.db.compaction.DateTieredCompactionStrategy.filterOldSSTables; import static org.apache.cassandra.db.compaction.DateTieredCompactionStrategy.validateOptions; @@ -213,27 +212,15 @@ public class DateTieredCompactionStrategyTest extends SchemaLoader List sstrs = new ArrayList<>(cfs.getSSTables()); - List newBucket = newestBucket(Collections.singletonList(sstrs.subList(0, 2)), 4, 32, 9, 10); + List newBucket = newestBucket(Collections.singletonList(sstrs.subList(0, 2)), 4, 32, 9, 10, new SizeTieredCompactionStrategyOptions()); assertTrue("incoming bucket should not be accepted when it has below the min threshold SSTables", newBucket.isEmpty()); - newBucket = newestBucket(Collections.singletonList(sstrs.subList(0, 2)), 4, 32, 10, 10); + newBucket = newestBucket(Collections.singletonList(sstrs.subList(0, 2)), 4, 32, 10, 10, new SizeTieredCompactionStrategyOptions()); assertFalse("non-incoming bucket should be accepted when it has at least 2 SSTables", newBucket.isEmpty()); assertEquals("an sstable with a single value should have equal min/max timestamps", sstrs.get(0).getMinTimestamp(), sstrs.get(0).getMaxTimestamp()); assertEquals("an sstable with a single value should have equal min/max timestamps", sstrs.get(1).getMinTimestamp(), sstrs.get(1).getMaxTimestamp()); assertEquals("an sstable with a single value should have equal min/max timestamps", sstrs.get(2).getMinTimestamp(), sstrs.get(2).getMaxTimestamp()); - - // if we have more than the max threshold, the oldest should be dropped - Collections.sort(sstrs, Collections.reverseOrder(new Comparator() { - public int compare(SSTableReader o1, SSTableReader o2) { - return Long.compare(o1.getMinTimestamp(), o2.getMinTimestamp()) ; - } - })); - - List bucket = trimToThreshold(sstrs, 2); - assertEquals("one bucket should have been dropped", 2, bucket.size()); - for (SSTableReader sstr : bucket) - assertFalse("the oldest sstable should be dropped", sstr.getMinTimestamp() == 0); } @Test @@ -320,4 +307,44 @@ public class DateTieredCompactionStrategyTest extends SchemaLoader cfs.getDataTracker().unmarkCompacting(cfs.getSSTables()); } + @Test + public void testSTCSBigWindow() + { + Keyspace keyspace = Keyspace.open(KEYSPACE1); + ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF_STANDARD1); + cfs.truncateBlocking(); + cfs.disableAutoCompaction(); + ByteBuffer bigValue = ByteBuffer.wrap(new byte[10000]); + ByteBuffer value = ByteBuffer.wrap(new byte[100]); + int numSSTables = 40; + // create big sstabels out of half: + long timestamp = System.currentTimeMillis(); + for (int r = 0; r < numSSTables / 2; r++) + { + for (int i = 0; i < 10; i++) + { + DecoratedKey key = Util.dk(String.valueOf(r)); + Mutation rm = new Mutation(KEYSPACE1, key.getKey()); + rm.add(CF_STANDARD1, Util.cellname("column"), bigValue, timestamp); + rm.apply(); + } + cfs.forceBlockingFlush(); + } + // and small ones: + for (int r = 0; r < numSSTables / 2; r++) + { + DecoratedKey key = Util.dk(String.valueOf(r)); + Mutation rm = new Mutation(KEYSPACE1, key.getKey()); + rm.add(CF_STANDARD1, Util.cellname("column"), value, timestamp); + rm.apply(); + cfs.forceBlockingFlush(); + } + Map options = new HashMap<>(); + options.put(SizeTieredCompactionStrategyOptions.MIN_SSTABLE_SIZE_KEY, "1"); + DateTieredCompactionStrategy dtcs = new DateTieredCompactionStrategy(cfs, options); + for (SSTableReader sstable : cfs.getSSTables()) + dtcs.addSSTable(sstable); + assertEquals(20, dtcs.getNextBackgroundTask(0).sstables.size()); + } + }