diff --git a/CHANGES.txt b/CHANGES.txt index 5e44f27402..c3469bca38 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -10,6 +10,7 @@ Merged from 2.2: * (Hadoop) fix splits calculation (CASSANDRA-10640) * (Hadoop) ensure that Cluster instances are always closed (CASSANDRA-10058) Merged from 2.1: + * Limit window size in DTCS (CASSANDRA-10280) * sstableloader does not use MAX_HEAP_SIZE env parameter (CASSANDRA-10188) * (cqlsh) Improve COPY TO performance and error handling (CASSANDRA-9304) * Create compression chunk for sending file only (CASSANDRA-10680) diff --git a/pylib/cqlshlib/cql3handling.py b/pylib/cqlshlib/cql3handling.py index 42e542ffa3..9ba41227db 100644 --- a/pylib/cqlshlib/cql3handling.py +++ b/pylib/cqlshlib/cql3handling.py @@ -508,6 +508,7 @@ def cf_prop_val_mapkey_completer(ctxt, cass): opts.add('max_sstable_age_days') opts.add('timestamp_resolution') opts.add('min_threshold') + opts.add('max_window_size_seconds') return map(escape_value, opts) return () diff --git a/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java index 65fec2b21d..50f9b71142 100644 --- a/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java +++ b/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java @@ -137,14 +137,15 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy { Iterable candidates = filterOldSSTables(Lists.newArrayList(candidateSSTables), options.maxSSTableAge, now); - List> buckets = getBuckets(createSSTableAndMinTimestampPairs(candidates), options.baseTime, base, now); - logger.trace("Compaction buckets are {}", buckets); + List> buckets = getBuckets(createSSTableAndMinTimestampPairs(candidates), options.baseTime, base, now, options.maxWindowSize); + logger.debug("Compaction buckets are {}", buckets); updateEstimatedCompactionsByTasks(buckets); List mostInteresting = newestBucket(buckets, cfs.getMinimumCompactionThreshold(), cfs.getMaximumCompactionThreshold(), now, options.baseTime, + options.maxWindowSize, stcsOptions); if (!mostInteresting.isEmpty()) return mostInteresting; @@ -221,10 +222,13 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy // A timestamp t hits the target iff t / size == divPosition. public final long divPosition; - public Target(long size, long divPosition) + public final long maxWindowSize; + + public Target(long size, long divPosition, long maxWindowSize) { this.size = size; this.divPosition = divPosition; + this.maxWindowSize = maxWindowSize; } /** @@ -254,10 +258,10 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy */ public Target nextTarget(int base) { - if (divPosition % base > 0) - return new Target(size, divPosition - 1); + if (divPosition % base > 0 || size * base > maxWindowSize) + return new Target(size, divPosition - 1, maxWindowSize); else - return new Target(size * base, divPosition / base - 1); + return new Target(size * base, divPosition / base - 1, maxWindowSize); } } @@ -274,7 +278,7 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy * Each bucket is also a list of files ordered from newest to oldest. */ @VisibleForTesting - static List> getBuckets(Collection> files, long timeUnit, int base, long now) + static List> getBuckets(Collection> files, long timeUnit, int base, long now, long maxWindowSize) { // Sort files by age. Newest first. final List> sortedFiles = Lists.newArrayList(files); @@ -287,7 +291,7 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy })); List> buckets = Lists.newArrayList(); - Target target = getInitialTarget(now, timeUnit); + Target target = getInitialTarget(now, timeUnit, maxWindowSize); PeekingIterator> it = Iterators.peekingIterator(sortedFiles.iterator()); outerLoop: @@ -306,7 +310,6 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy else // If the file is too old for the target, switch targets. target = target.nextTarget(base); } - List bucket = Lists.newArrayList(); while (target.onTarget(it.peek().right)) { @@ -322,9 +325,9 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy } @VisibleForTesting - static Target getInitialTarget(long now, long timeUnit) + static Target getInitialTarget(long now, long timeUnit, long maxWindowSize) { - return new Target(timeUnit, now / timeUnit); + return new Target(timeUnit, now / timeUnit, maxWindowSize); } @@ -333,8 +336,9 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy int n = 0; for (List bucket : tasks) { - if (bucket.size() >= cfs.getMinimumCompactionThreshold()) - n += getSTCSBuckets(bucket, stcsOptions).size(); + for (List stcsBucket : getSTCSBuckets(bucket, stcsOptions)) + if (stcsBucket.size() >= cfs.getMinimumCompactionThreshold()) + n += Math.ceil((double)stcsBucket.size() / cfs.getMaximumCompactionThreshold()); } estimatedRemainingTasks = n; } @@ -347,12 +351,12 @@ 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, SizeTieredCompactionStrategyOptions stcsOptions) + static List newestBucket(List> buckets, int minThreshold, int maxThreshold, long now, long baseTime, long maxWindowSize, SizeTieredCompactionStrategyOptions stcsOptions) { // If the "incoming window" has at least minThreshold SSTables, choose that one. // For any other bucket, at least 2 SSTables is enough. // In any case, limit to maxThreshold SSTables. - Target incomingWindow = getInitialTarget(now, baseTime); + Target incomingWindow = getInitialTarget(now, baseTime, maxWindowSize); for (List bucket : buckets) { boolean inFirstWindow = incomingWindow.onTarget(bucket.get(0).getMinTimestamp()); diff --git a/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategyOptions.java b/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategyOptions.java index 0cbf90ef9c..580311543d 100644 --- a/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategyOptions.java +++ b/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategyOptions.java @@ -25,17 +25,24 @@ import org.apache.cassandra.exceptions.ConfigurationException; public final class DateTieredCompactionStrategyOptions { protected static final TimeUnit DEFAULT_TIMESTAMP_RESOLUTION = TimeUnit.MICROSECONDS; - protected static final double DEFAULT_MAX_SSTABLE_AGE_DAYS = 365; + @Deprecated + protected static final double DEFAULT_MAX_SSTABLE_AGE_DAYS = 365*1000; protected static final long DEFAULT_BASE_TIME_SECONDS = 60; + protected static final long DEFAULT_MAX_WINDOW_SIZE_SECONDS = TimeUnit.SECONDS.convert(1, TimeUnit.DAYS); + protected static final int DEFAULT_EXPIRED_SSTABLE_CHECK_FREQUENCY_SECONDS = 60 * 10; protected static final String TIMESTAMP_RESOLUTION_KEY = "timestamp_resolution"; + @Deprecated protected static final String MAX_SSTABLE_AGE_KEY = "max_sstable_age_days"; protected static final String BASE_TIME_KEY = "base_time_seconds"; protected static final String EXPIRED_SSTABLE_CHECK_FREQUENCY_SECONDS_KEY = "expired_sstable_check_frequency_seconds"; + protected static final String MAX_WINDOW_SIZE_KEY = "max_window_size_seconds"; + @Deprecated protected final long maxSSTableAge; protected final long baseTime; protected final long expiredSSTableCheckFrequency; + protected final long maxWindowSize; public DateTieredCompactionStrategyOptions(Map options) { @@ -48,13 +55,16 @@ public final class DateTieredCompactionStrategyOptions baseTime = timestampResolution.convert(optionValue == null ? DEFAULT_BASE_TIME_SECONDS : Long.parseLong(optionValue), TimeUnit.SECONDS); optionValue = options.get(EXPIRED_SSTABLE_CHECK_FREQUENCY_SECONDS_KEY); expiredSSTableCheckFrequency = TimeUnit.MILLISECONDS.convert(optionValue == null ? DEFAULT_EXPIRED_SSTABLE_CHECK_FREQUENCY_SECONDS : Long.parseLong(optionValue), TimeUnit.SECONDS); + optionValue = options.get(MAX_WINDOW_SIZE_KEY); + maxWindowSize = timestampResolution.convert(optionValue == null ? DEFAULT_MAX_WINDOW_SIZE_SECONDS : Long.parseLong(optionValue), TimeUnit.SECONDS); } public DateTieredCompactionStrategyOptions() { - maxSSTableAge = Math.round(DEFAULT_MAX_SSTABLE_AGE_DAYS * DEFAULT_TIMESTAMP_RESOLUTION.convert(1, TimeUnit.DAYS)); + maxSSTableAge = Math.round(DEFAULT_MAX_SSTABLE_AGE_DAYS * DEFAULT_TIMESTAMP_RESOLUTION.convert((long) DEFAULT_MAX_SSTABLE_AGE_DAYS, TimeUnit.DAYS)); baseTime = DEFAULT_TIMESTAMP_RESOLUTION.convert(DEFAULT_BASE_TIME_SECONDS, TimeUnit.SECONDS); expiredSSTableCheckFrequency = TimeUnit.MILLISECONDS.convert(DEFAULT_EXPIRED_SSTABLE_CHECK_FREQUENCY_SECONDS, TimeUnit.SECONDS); + maxWindowSize = DEFAULT_TIMESTAMP_RESOLUTION.convert(1, TimeUnit.DAYS); } public static Map validateOptions(Map options, Map uncheckedOptions) throws ConfigurationException @@ -112,10 +122,26 @@ public final class DateTieredCompactionStrategyOptions throw new ConfigurationException(String.format("%s is not a parsable int (base10) for %s", optionValue, EXPIRED_SSTABLE_CHECK_FREQUENCY_SECONDS_KEY), e); } + optionValue = options.get(MAX_WINDOW_SIZE_KEY); + try + { + long maxWindowSize = optionValue == null ? DEFAULT_MAX_WINDOW_SIZE_SECONDS : Long.parseLong(optionValue); + if (maxWindowSize < 0) + { + throw new ConfigurationException(String.format("%s must not be negative, but was %d", MAX_WINDOW_SIZE_KEY, maxWindowSize)); + } + } + catch (NumberFormatException e) + { + throw new ConfigurationException(String.format("%s is not a parsable int (base10) for %s", optionValue, MAX_WINDOW_SIZE_KEY), e); + } + + uncheckedOptions.remove(MAX_SSTABLE_AGE_KEY); uncheckedOptions.remove(BASE_TIME_KEY); uncheckedOptions.remove(TIMESTAMP_RESOLUTION_KEY); uncheckedOptions.remove(EXPIRED_SSTABLE_CHECK_FREQUENCY_SECONDS_KEY); + uncheckedOptions.remove(MAX_WINDOW_SIZE_KEY); 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 01a6dfad8a..22b4829ec1 100644 --- a/test/unit/org/apache/cassandra/db/compaction/DateTieredCompactionStrategyTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/DateTieredCompactionStrategyTest.java @@ -98,6 +98,17 @@ public class DateTieredCompactionStrategyTest extends SchemaLoader options.put(DateTieredCompactionStrategyOptions.MAX_SSTABLE_AGE_KEY, "0"); } + try + { + options.put(DateTieredCompactionStrategyOptions.MAX_WINDOW_SIZE_KEY, "-1"); + validateOptions(options); + fail(String.format("Negative %s should be rejected", DateTieredCompactionStrategyOptions.MAX_WINDOW_SIZE_KEY)); + } + catch (ConfigurationException e) + { + options.put(DateTieredCompactionStrategyOptions.MAX_WINDOW_SIZE_KEY, "0"); + } + options.put("bad_option", "1.0"); unvalidated = validateOptions(options); assertTrue(unvalidated.containsKey("bad_option")); @@ -111,11 +122,11 @@ public class DateTieredCompactionStrategyTest extends SchemaLoader options.put(DateTieredCompactionStrategyOptions.TIMESTAMP_RESOLUTION_KEY, "SECONDS"); DateTieredCompactionStrategyOptions opts = new DateTieredCompactionStrategyOptions(options); - assertEquals(opts.maxSSTableAge, TimeUnit.SECONDS.convert(365, TimeUnit.DAYS)); + assertEquals(opts.maxSSTableAge, TimeUnit.SECONDS.convert(365*1000, TimeUnit.DAYS)); options.put(DateTieredCompactionStrategyOptions.TIMESTAMP_RESOLUTION_KEY, "MILLISECONDS"); opts = new DateTieredCompactionStrategyOptions(options); - assertEquals(opts.maxSSTableAge, TimeUnit.MILLISECONDS.convert(365, TimeUnit.DAYS)); + assertEquals(opts.maxSSTableAge, TimeUnit.MILLISECONDS.convert(365*1000, TimeUnit.DAYS)); options.put(DateTieredCompactionStrategyOptions.TIMESTAMP_RESOLUTION_KEY, "MICROSECONDS"); options.put(DateTieredCompactionStrategyOptions.MAX_SSTABLE_AGE_KEY, "10"); @@ -142,7 +153,7 @@ public class DateTieredCompactionStrategyTest extends SchemaLoader Pair.create("a", 1L), Pair.create("b", 201L) ); - List> buckets = getBuckets(pairs, 100L, 2, 200L); + List> buckets = getBuckets(pairs, 100L, 2, 200L, Long.MAX_VALUE); assertEquals(2, buckets.size()); for (List bucket : buckets) @@ -161,7 +172,7 @@ public class DateTieredCompactionStrategyTest extends SchemaLoader Pair.create("b", 3899L), Pair.create("c", 3900L) ); - buckets = getBuckets(pairs, 100L, 3, 4050L); + buckets = getBuckets(pairs, 100L, 3, 4050L, Long.MAX_VALUE); // targets (divPosition, size): (40, 100), (39, 100), (12, 300), (3, 900), (0, 2700) // in other words: 0 - 2699, 2700 - 3599, 3600 - 3899, 3900 - 3999, 4000 - 4099 assertEquals(3, buckets.size()); @@ -187,7 +198,7 @@ public class DateTieredCompactionStrategyTest extends SchemaLoader Pair.create("e", 3950L), Pair.create("too new", 4125L) ); - buckets = getBuckets(pairs, 100L, 1, 4050L); + buckets = getBuckets(pairs, 100L, 1, 4050L, Long.MAX_VALUE); assertEquals(5, buckets.size()); @@ -203,7 +214,6 @@ public class DateTieredCompactionStrategyTest extends SchemaLoader { Keyspace keyspace = Keyspace.open(KEYSPACE1); ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF_STANDARD1); - cfs.truncateBlocking(); cfs.disableAutoCompaction(); ByteBuffer value = ByteBuffer.wrap(new byte[100]); @@ -223,15 +233,16 @@ public class DateTieredCompactionStrategyTest extends SchemaLoader List sstrs = new ArrayList<>(cfs.getLiveSSTables()); - List newBucket = newestBucket(Collections.singletonList(sstrs.subList(0, 2)), 4, 32, 9, 10, new SizeTieredCompactionStrategyOptions()); + List newBucket = newestBucket(Collections.singletonList(sstrs.subList(0, 2)), 4, 32, 9, 10, Long.MAX_VALUE, 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, new SizeTieredCompactionStrategyOptions()); + newBucket = newestBucket(Collections.singletonList(sstrs.subList(0, 2)), 4, 32, 10, 10, Long.MAX_VALUE, 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()); + cfs.truncateBlocking(); } @Test @@ -239,7 +250,6 @@ public class DateTieredCompactionStrategyTest extends SchemaLoader { Keyspace keyspace = Keyspace.open(KEYSPACE1); ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF_STANDARD1); - cfs.truncateBlocking(); cfs.disableAutoCompaction(); ByteBuffer value = ByteBuffer.wrap(new byte[100]); @@ -271,6 +281,7 @@ public class DateTieredCompactionStrategyTest extends SchemaLoader filtered = filterOldSSTables(sstrs, 1, 4); assertEquals("no sstables should remain when all are too old", 0, Iterables.size(filtered)); + cfs.truncateBlocking(); } @@ -279,7 +290,6 @@ public class DateTieredCompactionStrategyTest extends SchemaLoader { Keyspace keyspace = Keyspace.open(KEYSPACE1); ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF_STANDARD1); - cfs.truncateBlocking(); cfs.disableAutoCompaction(); ByteBuffer value = ByteBuffer.wrap(new byte[100]); @@ -320,6 +330,7 @@ public class DateTieredCompactionStrategyTest extends SchemaLoader SSTableReader sstable = t.transaction.originals().iterator().next(); assertEquals(sstable, expiredSSTable); t.transaction.abort(); + cfs.truncateBlocking(); } @Test @@ -327,7 +338,6 @@ public class DateTieredCompactionStrategyTest extends SchemaLoader { 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]); @@ -362,7 +372,5 @@ public class DateTieredCompactionStrategyTest extends SchemaLoader AbstractCompactionTask task = dtcs.getNextBackgroundTask(0); assertEquals(20, task.transaction.originals().size()); task.transaction.abort(); - } - }