diff --git a/doc/cql3/CQL.textile b/doc/cql3/CQL.textile index 88d669468d..db1ec22bae 100644 --- a/doc/cql3/CQL.textile +++ b/doc/cql3/CQL.textile @@ -347,24 +347,24 @@ h4(#compactionOptions). Compaction options The @compaction@ property must at least define the @'class'@ sub-option, that defines the compaction strategy class to use. The default supported class are @'SizeTieredCompactionStrategy'@, @'LeveledCompactionStrategy'@, @'DateTieredCompactionStrategy'@ and @'TimeWindowCompactionStrategy'@. Custom strategy can be provided by specifying the full class name as a "string constant":#constants. The rest of the sub-options depends on the chosen class. The sub-options supported by the default classes are: -|_. option |_. supported compaction strategy |_. default |_. description | -| @enabled@ | _all_ | true | A boolean denoting whether compaction should be enabled or not.| -| @tombstone_threshold@ | _all_ | 0.2 | A ratio such that if a sstable has more than this ratio of gcable tombstones over all contained columns, the sstable will be compacted (with no other sstables) for the purpose of purging those tombstones. | -| @tombstone_compaction_interval@ | _all_ | 1 day | The minimum time to wait after an sstable creation time before considering it for "tombstone compaction", where "tombstone compaction" is the compaction triggered if the sstable has more gcable tombstones than @tombstone_threshold@. | -| @unchecked_tombstone_compaction@ | _all_ | false | Setting this to true enables more aggressive tombstone compactions - single sstable tombstone compactions will run without checking how likely it is that they will be successful. | -| @min_sstable_size@ | SizeTieredCompactionStrategy | 50MB | The size tiered strategy groups SSTables to compact in buckets. A bucket groups SSTables that differs from less than 50% in size. However, for small sizes, this would result in a bucketing that is too fine grained. @min_sstable_size@ defines a size threshold (in bytes) below which all SSTables belong to one unique bucket| -| @min_threshold@ | SizeTieredCompactionStrategy | 4 | Minimum number of SSTables needed to start a minor compaction.| -| @max_threshold@ | SizeTieredCompactionStrategy | 32 | Maximum number of SSTables processed by one minor compaction.| -| @bucket_low@ | SizeTieredCompactionStrategy | 0.5 | Size tiered consider sstables to be within the same bucket if their size is within [average_size * @bucket_low@, average_size * @bucket_high@ ] (i.e the default groups sstable whose sizes diverges by at most 50%)| -| @bucket_high@ | SizeTieredCompactionStrategy | 1.5 | Size tiered consider sstables to be within the same bucket if their size is within [average_size * @bucket_low@, average_size * @bucket_high@ ] (i.e the default groups sstable whose sizes diverges by at most 50%).| -| @sstable_size_in_mb@ | LeveledCompactionStrategy | 5MB | The target size (in MB) for sstables in the leveled strategy. Note that while sstable sizes should stay less or equal to @sstable_size_in_mb@, it is possible to exceptionally have a larger sstable as during compaction, data for a given partition key are never split into 2 sstables| -| @timestamp_resolution@ | DateTieredCompactionStrategy | MICROSECONDS | The timestamp resolution used when inserting data, could be MILLISECONDS, MICROSECONDS etc (should be understandable by Java TimeUnit) - don't change this unless you do mutations with USING TIMESTAMP (or equivalent directly in the client)| -| @base_time_seconds@ | DateTieredCompactionStrategy | 60 | The base size of the time windows. | -| @max_sstable_age_days@ | DateTieredCompactionStrategy | 365 | SSTables only containing data that is older than this will never be compacted. | -| @timestamp_resolution@ | TimeWindowCompactionStrategy | MICROSECONDS | The timestamp resolution used when inserting data, could be MILLISECONDS, MICROSECONDS etc (should be understandable by Java TimeUnit) - don't change this unless you do mutations with USING TIMESTAMP (or equivalent directly in the client)| -| @compaction_window_unit@ | TimeWindowCompactionStrategy | DAYS | The Java TimeUnit used for the window size, set in conjunction with @compaction_window_size@. Must be one of DAYS, HOURS, MINUTES | -| @compaction_window_size@ | TimeWindowCompactionStrategy | 1 | The number of @compaction_window_unit@ units that make up a time window. | - +|_. option |_. supported compaction strategy |_. default |_. description | +| @enabled@ | _all_ | true | A boolean denoting whether compaction should be enabled or not.| +| @tombstone_threshold@ | _all_ | 0.2 | A ratio such that if a sstable has more than this ratio of gcable tombstones over all contained columns, the sstable will be compacted (with no other sstables) for the purpose of purging those tombstones. | +| @tombstone_compaction_interval@ | _all_ | 1 day | The minimum time to wait after an sstable creation time before considering it for "tombstone compaction", where "tombstone compaction" is the compaction triggered if the sstable has more gcable tombstones than @tombstone_threshold@. | +| @unchecked_tombstone_compaction@ | _all_ | false | Setting this to true enables more aggressive tombstone compactions - single sstable tombstone compactions will run without checking how likely it is that they will be successful. | +| @min_sstable_size@ | SizeTieredCompactionStrategy | 50MB | The size tiered strategy groups SSTables to compact in buckets. A bucket groups SSTables that differs from less than 50% in size. However, for small sizes, this would result in a bucketing that is too fine grained. @min_sstable_size@ defines a size threshold (in bytes) below which all SSTables belong to one unique bucket| +| @min_threshold@ | SizeTieredCompactionStrategy | 4 | Minimum number of SSTables needed to start a minor compaction.| +| @max_threshold@ | SizeTieredCompactionStrategy | 32 | Maximum number of SSTables processed by one minor compaction.| +| @bucket_low@ | SizeTieredCompactionStrategy | 0.5 | Size tiered consider sstables to be within the same bucket if their size is within [average_size * @bucket_low@, average_size * @bucket_high@ ] (i.e the default groups sstable whose sizes diverges by at most 50%)| +| @bucket_high@ | SizeTieredCompactionStrategy | 1.5 | Size tiered consider sstables to be within the same bucket if their size is within [average_size * @bucket_low@, average_size * @bucket_high@ ] (i.e the default groups sstable whose sizes diverges by at most 50%).| +| @sstable_size_in_mb@ | LeveledCompactionStrategy | 5MB | The target size (in MB) for sstables in the leveled strategy. Note that while sstable sizes should stay less or equal to @sstable_size_in_mb@, it is possible to exceptionally have a larger sstable as during compaction, data for a given partition key are never split into 2 sstables| +| @timestamp_resolution@ | DateTieredCompactionStrategy | MICROSECONDS | The timestamp resolution used when inserting data, could be MILLISECONDS, MICROSECONDS etc (should be understandable by Java TimeUnit) - don't change this unless you do mutations with USING TIMESTAMP (or equivalent directly in the client)| +| @base_time_seconds@ | DateTieredCompactionStrategy | 60 | The base size of the time windows. | +| @max_sstable_age_days@ | DateTieredCompactionStrategy | 365 | SSTables only containing data that is older than this will never be compacted. | +| @timestamp_resolution@ | TimeWindowCompactionStrategy | MICROSECONDS | The timestamp resolution used when inserting data, could be MILLISECONDS, MICROSECONDS etc (should be understandable by Java TimeUnit) - don't change this unless you do mutations with USING TIMESTAMP (or equivalent directly in the client)| +| @compaction_window_unit@ | TimeWindowCompactionStrategy | DAYS | The Java TimeUnit used for the window size, set in conjunction with @compaction_window_size@. Must be one of DAYS, HOURS, MINUTES | +| @compaction_window_size@ | TimeWindowCompactionStrategy | 1 | The number of @compaction_window_unit@ units that make up a time window. | +| @unsafe_aggressive_sstable_expiration@ | TimeWindowCompactionStrategy | false | Expired sstables will be dropped without checking its data is shadowing other sstables. This is a potentially risky option that can lead to data loss or deleted data re-appearing, going beyond what `unchecked_tombstone_compaction` does for single sstable compaction. Due to the risk the jvm must also be started with `-Dcassandra.unsafe_aggressive_sstable_expiration=true`. | h4(#compressionOptions). Compression options diff --git a/doc/source/operating/compaction.rst b/doc/source/operating/compaction.rst index 0f39000421..cb6be458f1 100644 --- a/doc/source/operating/compaction.rst +++ b/doc/source/operating/compaction.rst @@ -160,7 +160,8 @@ compaction can drop that sstable. If you see sstables with only tombstones (note tombstones once the time to live has expired) but it is not being dropped by compaction, it is likely that other sstables contain older data. There is a tool called ``sstableexpiredblockers`` that will list which sstables are droppable and which are blocking them from being dropped. This is especially useful for time series compaction with -``TimeWindowCompactionStrategy`` (and the deprecated ``DateTieredCompactionStrategy``). +``TimeWindowCompactionStrategy`` (and the deprecated ``DateTieredCompactionStrategy``). With ``TimeWindowCompactionStrategy`` +it is possible to remove the guarantee (not check for shadowing data) by enabling ``unsafe_aggressive_sstable_expiration``. Repaired/unrepaired data ^^^^^^^^^^^^^^^^^^^^^^^^ @@ -403,6 +404,11 @@ as the combination of two primary options: A Java TimeUnit (MINUTES, HOURS, or DAYS). ``compaction_window_size`` (default: 1) The number of units that make up a window. +``unsafe_aggressive_sstable_expiration`` (default: false) + Expired sstables will be dropped without checking its data is shadowing other sstables. This is a potentially + risky option that can lead to data loss or deleted data re-appearing, going beyond what + `unchecked_tombstone_compaction` does for single sstable compaction. Due to the risk the jvm must also be + started with `-Dcassandra.unsafe_aggressive_sstable_expiration=true`. Taken together, the operator can specify windows of virtually any size, and `TimeWindowCompactionStrategy` will work to create a single sstable for writes within that window. For efficiency during writing, the newest window will be diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionController.java b/src/java/org/apache/cassandra/db/compaction/CompactionController.java index bf3647affc..84aac09052 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionController.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionController.java @@ -20,6 +20,7 @@ package org.apache.cassandra.db.compaction; import java.util.*; import java.util.function.Predicate; +import org.apache.cassandra.config.Config; import org.apache.cassandra.db.Memtable; import org.apache.cassandra.db.rows.UnfilteredRowIterator; @@ -49,7 +50,8 @@ import static org.apache.cassandra.db.lifecycle.SSTableIntervalTree.buildInterva public class CompactionController implements AutoCloseable { private static final Logger logger = LoggerFactory.getLogger(CompactionController.class); - static final boolean NEVER_PURGE_TOMBSTONES = Boolean.getBoolean("cassandra.never_purge_tombstones"); + private static final String NEVER_PURGE_TOMBSTONES_PROPERTY = Config.PROPERTY_PREFIX + "never_purge_tombstones"; + static final boolean NEVER_PURGE_TOMBSTONES = Boolean.getBoolean(NEVER_PURGE_TOMBSTONES_PROPERTY); public final ColumnFamilyStore cfs; private final boolean compactingRepaired; @@ -98,7 +100,14 @@ public class CompactionController implements AutoCloseable { if (NEVER_PURGE_TOMBSTONES) { - logger.debug("not refreshing overlaps - running with -Dcassandra.never_purge_tombstones=true"); + logger.debug("not refreshing overlaps - running with -D{}=true", + NEVER_PURGE_TOMBSTONES_PROPERTY); + return; + } + + if (ignoreOverlaps()) + { + logger.debug("not refreshing overlaps - running with ignoreOverlaps activated"); return; } @@ -120,7 +129,7 @@ public class CompactionController implements AutoCloseable if (this.overlappingSSTables != null) close(); - if (compacting == null) + if (compacting == null || ignoreOverlaps()) overlappingSSTables = Refs.tryRef(Collections.emptyList()); else overlappingSSTables = cfs.getAndReferenceOverlappingLiveSSTables(compacting); @@ -129,7 +138,7 @@ public class CompactionController implements AutoCloseable public Set getFullyExpiredSSTables() { - return getFullyExpiredSSTables(cfs, compacting, overlappingSSTables, gcBefore); + return getFullyExpiredSSTables(cfs, compacting, overlappingSSTables, gcBefore, ignoreOverlaps()); } /** @@ -146,20 +155,39 @@ public class CompactionController implements AutoCloseable * @param compacting we take the drop-candidates from this set, it is usually the sstables included in the compaction * @param overlapping the sstables that overlap the ones in compacting. * @param gcBefore + * @param ignoreOverlaps don't check if data shadows/overlaps any data in other sstables * @return */ - public static Set getFullyExpiredSSTables(ColumnFamilyStore cfStore, Iterable compacting, Iterable overlapping, int gcBefore) + public static Set getFullyExpiredSSTables(ColumnFamilyStore cfStore, + Iterable compacting, + Iterable overlapping, + int gcBefore, + boolean ignoreOverlaps) { logger.trace("Checking droppable sstables in {}", cfStore); if (NEVER_PURGE_TOMBSTONES || compacting == null) - return Collections.emptySet(); + return Collections.emptySet(); if (cfStore.getCompactionStrategyManager().onlyPurgeRepairedTombstones() && !Iterables.all(compacting, SSTableReader::isRepaired)) return Collections.emptySet(); - List candidates = new ArrayList<>(); + if (ignoreOverlaps) + { + Set fullyExpired = new HashSet<>(); + for (SSTableReader candidate : compacting) + { + if (candidate.getSSTableMetadata().maxLocalDeletionTime < gcBefore) + { + fullyExpired.add(candidate); + logger.trace("Dropping overlap ignored expired SSTable {} (maxLocalDeletionTime={}, gcBefore={})", + candidate, candidate.getSSTableMetadata().maxLocalDeletionTime, gcBefore); + } + } + return fullyExpired; + } + List candidates = new ArrayList<>(); long minTimestamp = Long.MAX_VALUE; for (SSTableReader sstable : overlapping) @@ -203,6 +231,14 @@ public class CompactionController implements AutoCloseable return new HashSet<>(candidates); } + public static Set getFullyExpiredSSTables(ColumnFamilyStore cfStore, + Iterable compacting, + Iterable overlapping, + int gcBefore) + { + return getFullyExpiredSSTables(cfStore, compacting, overlapping, gcBefore, false); + } + public String getKeyspace() { return cfs.keyspace.getName(); @@ -306,6 +342,23 @@ public class CompactionController implements AutoCloseable return reader.simpleIterator(dfile, key, position, tombstoneOnly); } + /** + * Is overlapped sstables ignored + * + * Control whether or not we are taking into account overlapping sstables when looking for fully expired sstables. + * In order to reduce the amount of work needed, we look for sstables that can be dropped instead of compacted. + * As a safeguard mechanism, for each time range of data in a sstable, we are checking globally to see if all data + * of this time range is fully expired before considering to drop the sstable. + * This strategy can retain for a long time a lot of sstables on disk (see CASSANDRA-13418) so this option + * control whether or not this check should be ignored. + * + * @return false by default + */ + protected boolean ignoreOverlaps() + { + return false; + } + private FileDataInput openDataFile(SSTableReader reader) { return limiter != null ? reader.openDataReader(limiter) : reader.openDataReader(); diff --git a/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionController.java b/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionController.java new file mode 100644 index 0000000000..cf9e0e6be6 --- /dev/null +++ b/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionController.java @@ -0,0 +1,49 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.db.compaction; + + +import java.util.Set; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.io.sstable.format.SSTableReader; + +public class TimeWindowCompactionController extends CompactionController +{ + private static final Logger logger = LoggerFactory.getLogger(TimeWindowCompactionController.class); + + private final boolean ignoreOverlaps; + + public TimeWindowCompactionController(ColumnFamilyStore cfs, Set compacting, int gcBefore, boolean ignoreOverlaps) + { + super(cfs, compacting, gcBefore); + this.ignoreOverlaps = ignoreOverlaps; + if (ignoreOverlaps) + logger.warn("You are running with sstables overlapping checks disabled, it can result in loss of data"); + } + + @Override + protected boolean ignoreOverlaps() + { + return ignoreOverlaps; + } +} diff --git a/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategy.java index 0f3c171208..5a4e88b883 100644 --- a/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategy.java +++ b/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategy.java @@ -66,7 +66,6 @@ public class TimeWindowCompactionStrategy extends AbstractCompactionStrategy } else logger.debug("Enabling tombstone compactions for TWCS"); - } @Override @@ -82,7 +81,7 @@ public class TimeWindowCompactionStrategy extends AbstractCompactionStrategy LifecycleTransaction modifier = cfs.getTracker().tryModify(latestBucket, OperationType.COMPACTION); if (modifier != null) - return new CompactionTask(cfs, modifier, gcBefore); + return new TimeWindowCompactionTask(cfs, modifier, gcBefore, options.ignoreOverlaps); } } @@ -104,7 +103,8 @@ public class TimeWindowCompactionStrategy extends AbstractCompactionStrategy if (System.currentTimeMillis() - lastExpiredCheck > options.expiredSSTableCheckFrequency) { logger.debug("TWCS expired check sufficiently far in the past, checking for fully expired SSTables"); - expired = CompactionController.getFullyExpiredSSTables(cfs, uncompacting, cfs.getOverlappingLiveSSTables(uncompacting), gcBefore); + expired = CompactionController.getFullyExpiredSSTables(cfs, uncompacting, options.ignoreOverlaps ? Collections.emptySet() : cfs.getOverlappingLiveSSTables(uncompacting), + gcBefore, options.ignoreOverlaps); lastExpiredCheck = System.currentTimeMillis(); } else @@ -336,7 +336,7 @@ public class TimeWindowCompactionStrategy extends AbstractCompactionStrategy LifecycleTransaction txn = cfs.getTracker().tryModify(filteredSSTables, OperationType.COMPACTION); if (txn == null) return null; - return Collections.singleton(new CompactionTask(cfs, txn, gcBefore)); + return Collections.singleton(new TimeWindowCompactionTask(cfs, txn, gcBefore, options.ignoreOverlaps)); } @Override @@ -352,7 +352,7 @@ public class TimeWindowCompactionStrategy extends AbstractCompactionStrategy return null; } - return new CompactionTask(cfs, modifier, gcBefore).setUserDefined(true); + return new TimeWindowCompactionTask(cfs, modifier, gcBefore, options.ignoreOverlaps).setUserDefined(true); } public int getEstimatedRemainingTasks() diff --git a/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategyOptions.java b/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategyOptions.java index 07df606049..24b4fe0503 100644 --- a/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategyOptions.java +++ b/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategyOptions.java @@ -26,6 +26,7 @@ import java.util.concurrent.TimeUnit; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.cassandra.config.Config; import org.apache.cassandra.exceptions.ConfigurationException; public final class TimeWindowCompactionStrategyOptions @@ -36,16 +37,21 @@ public final class TimeWindowCompactionStrategyOptions protected static final TimeUnit DEFAULT_COMPACTION_WINDOW_UNIT = TimeUnit.DAYS; protected static final int DEFAULT_COMPACTION_WINDOW_SIZE = 1; protected static final int DEFAULT_EXPIRED_SSTABLE_CHECK_FREQUENCY_SECONDS = 60 * 10; + protected static final Boolean DEFAULT_UNSAFE_AGGRESSIVE_SSTABLE_EXPIRATION = false; protected static final String TIMESTAMP_RESOLUTION_KEY = "timestamp_resolution"; protected static final String COMPACTION_WINDOW_UNIT_KEY = "compaction_window_unit"; protected static final String COMPACTION_WINDOW_SIZE_KEY = "compaction_window_size"; protected static final String EXPIRED_SSTABLE_CHECK_FREQUENCY_SECONDS_KEY = "expired_sstable_check_frequency_seconds"; + protected static final String UNSAFE_AGGRESSIVE_SSTABLE_EXPIRATION_KEY = "unsafe_aggressive_sstable_expiration"; + + static final String UNSAFE_AGGRESSIVE_SSTABLE_EXPIRATION_PROPERTY = Config.PROPERTY_PREFIX + "allow_unsafe_aggressive_sstable_expiration"; protected final int sstableWindowSize; protected final TimeUnit sstableWindowUnit; protected final TimeUnit timestampResolution; protected final long expiredSSTableCheckFrequency; + protected final boolean ignoreOverlaps; SizeTieredCompactionStrategyOptions stcsOptions; @@ -68,6 +74,9 @@ public final class TimeWindowCompactionStrategyOptions 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(UNSAFE_AGGRESSIVE_SSTABLE_EXPIRATION_KEY); + ignoreOverlaps = optionValue == null ? DEFAULT_UNSAFE_AGGRESSIVE_SSTABLE_EXPIRATION : (Boolean.getBoolean(UNSAFE_AGGRESSIVE_SSTABLE_EXPIRATION_PROPERTY) && Boolean.parseBoolean(optionValue)); + stcsOptions = new SizeTieredCompactionStrategyOptions(options); } @@ -77,6 +86,7 @@ public final class TimeWindowCompactionStrategyOptions timestampResolution = DEFAULT_TIMESTAMP_RESOLUTION; sstableWindowSize = DEFAULT_COMPACTION_WINDOW_SIZE; expiredSSTableCheckFrequency = TimeUnit.MILLISECONDS.convert(DEFAULT_EXPIRED_SSTABLE_CHECK_FREQUENCY_SECONDS, TimeUnit.SECONDS); + ignoreOverlaps = DEFAULT_UNSAFE_AGGRESSIVE_SSTABLE_EXPIRATION; stcsOptions = new SizeTieredCompactionStrategyOptions(); } @@ -136,10 +146,22 @@ public final class TimeWindowCompactionStrategyOptions 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(UNSAFE_AGGRESSIVE_SSTABLE_EXPIRATION_KEY); + if (optionValue != null) + { + if (!(optionValue.equalsIgnoreCase("true") || optionValue.equalsIgnoreCase("false"))) + throw new ConfigurationException(String.format("%s is not 'true' or 'false' (%s)", UNSAFE_AGGRESSIVE_SSTABLE_EXPIRATION_KEY, optionValue)); + + if(optionValue.equalsIgnoreCase("true") && !Boolean.getBoolean(UNSAFE_AGGRESSIVE_SSTABLE_EXPIRATION_PROPERTY)) + throw new ConfigurationException(String.format("%s is requested but not allowed, restart cassandra with -D%s=true to allow it", UNSAFE_AGGRESSIVE_SSTABLE_EXPIRATION_KEY, UNSAFE_AGGRESSIVE_SSTABLE_EXPIRATION_PROPERTY)); + } + uncheckedOptions.remove(COMPACTION_WINDOW_SIZE_KEY); uncheckedOptions.remove(COMPACTION_WINDOW_UNIT_KEY); uncheckedOptions.remove(TIMESTAMP_RESOLUTION_KEY); uncheckedOptions.remove(EXPIRED_SSTABLE_CHECK_FREQUENCY_SECONDS_KEY); + uncheckedOptions.remove(UNSAFE_AGGRESSIVE_SSTABLE_EXPIRATION_KEY); uncheckedOptions = SizeTieredCompactionStrategyOptions.validateOptions(options, uncheckedOptions); diff --git a/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionTask.java b/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionTask.java new file mode 100644 index 0000000000..4f1fe6a0b9 --- /dev/null +++ b/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionTask.java @@ -0,0 +1,42 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.db.compaction; + +import java.util.Set; + +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.lifecycle.LifecycleTransaction; +import org.apache.cassandra.io.sstable.format.SSTableReader; + +public class TimeWindowCompactionTask extends CompactionTask +{ + private final boolean ignoreOverlaps; + + public TimeWindowCompactionTask(ColumnFamilyStore cfs, LifecycleTransaction txn, int gcBefore, boolean ignoreOverlaps) + { + super(cfs, txn, gcBefore); + this.ignoreOverlaps = ignoreOverlaps; + } + + @Override + public CompactionController getCompactionController(Set toCompact) + { + return new TimeWindowCompactionController(cfs, toCompact, gcBefore, ignoreOverlaps); + } +} diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionControllerTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionControllerTest.java index 14dc3be077..40693fa249 100644 --- a/test/unit/org/apache/cassandra/db/compaction/CompactionControllerTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/CompactionControllerTest.java @@ -179,6 +179,11 @@ public class CompactionControllerTest extends SchemaLoader expired = CompactionController.getFullyExpiredSSTables(cfs, compacting, overlapping, gcBefore); assertNotNull(expired); assertEquals(0, expired.size()); + + // Now if we explicitly ask to ignore overlaped sstables, we should get back our expired sstable + expired = CompactionController.getFullyExpiredSSTables(cfs, compacting, overlapping, gcBefore, true); + assertNotNull(expired); + assertEquals(1, expired.size()); } private void applyMutation(TableMetadata cfm, DecoratedKey key, long timestamp) diff --git a/test/unit/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategyTest.java b/test/unit/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategyTest.java index 0df546de77..89dd2f59b1 100644 --- a/test/unit/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategyTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategyTest.java @@ -60,6 +60,7 @@ public class TimeWindowCompactionStrategyTest extends SchemaLoader { // Disable tombstone histogram rounding for tests System.setProperty("cassandra.streaminghistogram.roundseconds", "1"); + System.setProperty(TimeWindowCompactionStrategyOptions.UNSAFE_AGGRESSIVE_SSTABLE_EXPIRATION_PROPERTY, "true"); SchemaLoader.prepareServer(); @@ -100,13 +101,24 @@ public class TimeWindowCompactionStrategyTest extends SchemaLoader { options.put(TimeWindowCompactionStrategyOptions.COMPACTION_WINDOW_UNIT_KEY, "MONTHS"); validateOptions(options); - fail(String.format("Invalid time units should be rejected", TimeWindowCompactionStrategyOptions.COMPACTION_WINDOW_UNIT_KEY)); + fail(String.format("Invalid %s should be rejected", TimeWindowCompactionStrategyOptions.COMPACTION_WINDOW_UNIT_KEY)); } catch (ConfigurationException e) { options.put(TimeWindowCompactionStrategyOptions.COMPACTION_WINDOW_UNIT_KEY, "MINUTES"); } + try + { + options.put(TimeWindowCompactionStrategyOptions.UNSAFE_AGGRESSIVE_SSTABLE_EXPIRATION_KEY, "not-a-boolean"); + validateOptions(options); + fail(String.format("Invalid %s should be rejected", TimeWindowCompactionStrategyOptions.UNSAFE_AGGRESSIVE_SSTABLE_EXPIRATION_KEY)); + } + catch (ConfigurationException e) + { + options.put(TimeWindowCompactionStrategyOptions.UNSAFE_AGGRESSIVE_SSTABLE_EXPIRATION_KEY, "true"); + } + options.put("bad_option", "1.0"); unvalidated = validateOptions(options); assertTrue(unvalidated.containsKey("bad_option")); @@ -272,4 +284,64 @@ public class TimeWindowCompactionStrategyTest extends SchemaLoader t.transaction.abort(); } + @Test + public void testDropOverlappingExpiredSSTables() throws InterruptedException + { + Keyspace keyspace = Keyspace.open(KEYSPACE1); + ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF_STANDARD1); + cfs.truncateBlocking(); + cfs.disableAutoCompaction(); + + ByteBuffer value = ByteBuffer.wrap(new byte[100]); + + // create 2 sstables + DecoratedKey key = Util.dk(String.valueOf("expired")); + new RowUpdateBuilder(cfs.metadata(), System.currentTimeMillis(), 1, key.getKey()) + .clustering("column") + .add("val", value).build().applyUnsafe(); + + cfs.forceBlockingFlush(); + SSTableReader expiredSSTable = cfs.getLiveSSTables().iterator().next(); + Thread.sleep(10); + + new RowUpdateBuilder(cfs.metadata(), System.currentTimeMillis() - 1000, key.getKey()) + .clustering("column") + .add("val", value).build().applyUnsafe(); + key = Util.dk(String.valueOf("nonexpired")); + new RowUpdateBuilder(cfs.metadata(), System.currentTimeMillis(), key.getKey()) + .clustering("column") + .add("val", value).build().applyUnsafe(); + + cfs.forceBlockingFlush(); + assertEquals(cfs.getLiveSSTables().size(), 2); + + Map options = new HashMap<>(); + + options.put(TimeWindowCompactionStrategyOptions.COMPACTION_WINDOW_SIZE_KEY, "30"); + options.put(TimeWindowCompactionStrategyOptions.COMPACTION_WINDOW_UNIT_KEY, "SECONDS"); + options.put(TimeWindowCompactionStrategyOptions.TIMESTAMP_RESOLUTION_KEY, "MILLISECONDS"); + options.put(TimeWindowCompactionStrategyOptions.EXPIRED_SSTABLE_CHECK_FREQUENCY_SECONDS_KEY, "0"); + TimeWindowCompactionStrategy twcs = new TimeWindowCompactionStrategy(cfs, options); + for (SSTableReader sstable : cfs.getLiveSSTables()) + twcs.addSSTable(sstable); + + twcs.startup(); + assertNull(twcs.getNextBackgroundTask((int) (System.currentTimeMillis() / 1000))); + Thread.sleep(2000); + assertNull(twcs.getNextBackgroundTask((int) (System.currentTimeMillis()/1000))); + + options.put(TimeWindowCompactionStrategyOptions.UNSAFE_AGGRESSIVE_SSTABLE_EXPIRATION_KEY, "true"); + twcs = new TimeWindowCompactionStrategy(cfs, options); + for (SSTableReader sstable : cfs.getLiveSSTables()) + twcs.addSSTable(sstable); + + twcs.startup(); + AbstractCompactionTask t = twcs.getNextBackgroundTask((int) (System.currentTimeMillis()/1000)); + assertNotNull(t); + assertEquals(1, Iterables.size(t.transaction.originals())); + SSTableReader sstable = t.transaction.originals().iterator().next(); + assertEquals(sstable, expiredSSTable); + twcs.shutdown(); + t.transaction.abort(); + } }