diff --git a/CASSANDRA-14092.txt b/CASSANDRA-14092.txt new file mode 100644 index 0000000000..f95380b583 --- /dev/null +++ b/CASSANDRA-14092.txt @@ -0,0 +1,81 @@ +CASSANDRA-14092: MAXIMUM TTL EXPIRATION DATE +--------------------------------------------- + +The maximum expiration timestamp that can be represented by the storage engine is +2038-01-19T03:14:06+00:00, which means that INSERTS using TTL that would expire +after this date are not currently supported. + +# Expiration Date Overflow Policy + +We plan to lift this limitation in newer versions, but while the fix is not available, +operators can decide which policy to apply when dealing with inserts with TTL exceeding +the maximum supported expiration date: + - REJECT: this is the default policy and will reject any requests with expiration + date timestamp after 2038-01-19T03:14:06+00:00. + - CAP: any insert with TTL expiring after 2038-01-19T03:14:06+00:00 will expire on + 2038-01-19T03:14:06+00:00 and the client will receive a warning. + - CAP_NOWARN: same as previous, except that the client warning will not be emitted. + +These policies may be specified via the -Dcassandra.expiration_date_overflow_policy=POLICY +startup option in the jvm.options configuration file. + +# Potential data loss on earlier versions + +Prior to 3.0.16 (3.0.X) and 3.11.2 (3.11.x), there was no protection against +INSERTS with TTL expiring after the maximum supported date, causing the expiration +time field to overflow and the records to expire immediately. Expired records due +to overflow will not be queryable and will be permanently removed after a compaction. + +2.1.X, 2.2.X and earlier series are not subject to this bug when assertions are enabled +since an AssertionError is thrown during INSERT when the expiration time field overflows +on these versions. When assertions are disabled then it is possible to INSERT entries +with overflowed local expiration time and even the earlier versions are subject to data +loss due to this bug. + +This issue only affected INSERTs with very large TTLs, close to the maximum allowed value +of 630720000 seconds (20 years), starting from 2018-01-19T03:14:06+00:00. As time progresses, +the maximum supported TTL will be gradually reduced as the maximum expiration date approaches. +For instance, a user on an affected version on 2028-01-19T03:14:06 with a TTL of 10 years +will be affected by this bug, so we urge users of very large TTLs to upgrade to a version +where this issue is addressed as soon as possible. + +# Data Recovery + +SSTables from Cassandra versions prior to 2.1.20/2.2.12/3.0.16/3.11.2 containing entries +with overflowed expiration time that were backed up or did not go through compaction can +be recovered by reinserting overflowed entries with a valid expiration time and a higher +timestamp, since tombstones may have been generated with the original timestamp. + +To find out if an SSTable has an entry with overflowed expiration, inspect it with the +'sstablemetadata' tool and look for a negative "Minimum local deletion time" field. SSTables +in this condition should be backed up immediately, as they are subject to data loss during +compaction. + +A "--reinsert-overflowed-ttl" option was added to scrub to rewrite SSTables containing +rows with overflowed expiration time with the maximum expiration date of +2038-01-19T03:14:06+00:00 and the original timestamp + 1 (ms). Two methods are offered +for recovery of SSTables via scrub: + +- Offline scrub: + - Clone the data directory tree to another location, keeping only the folders and the + contents of the system tables. + - Clone the configuration directory to another location, setting the data_file_directories + property to the cloned data directory in the cloned cassandra.yaml. + - Copy the affected SSTables to the cloned data location of the affected table. + - Set the environment variable CASSANDRA_CONF=. + - Execute "sstablescrub --reinsert-overflowed-ttl ". + WARNING: not specifying --reinsert-overflowed-ttl is equivalent to a single-sstable + compaction, so the data with overflowed will be removed - make sure to back up + your SSTables before running scrub. + - Once the scrub is completed, copy the resulting SSTables to the original data directory. + - Execute "nodetool refresh" in a live node to load the recovered SSTables. + +- Online scrub: + - Disable compaction on the node with "nodetool disableautocompaction" - this step is crucial + as otherwise, the data may be removed permanently during compaction. + - Copy the SSTables containing entries with overflowed expiration time to the data directory. + - run "nodetool refresh" to load the SSTables. + - run "nodetool scrub --reinsert-overflowed-ttl
". + - Re-enable compactions after verifying that scrub recovered the missing entries. + +See https://issues.apache.org/jira/browse/CASSANDRA-14092 for more details about this issue. diff --git a/CHANGES.txt b/CHANGES.txt index f42f3f45cb..a492c42a04 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 3.0.16 + * Protect against overflow of local expiration time (CASSANDRA-14092) * Use the correct digest file and reload sstable metadata in nodetool verify (CASSANDRA-14217) * Handle failure when mutating repaired status in Verifier (CASSANDRA-13933) * Close socket on error during connect on OutboundTcpConnection (CASSANDRA-9630) diff --git a/NEWS.txt b/NEWS.txt index 1bbe1aa301..f574c339bc 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -1,3 +1,23 @@ +PLEASE READ: MAXIMUM TTL EXPIRATION DATE NOTICE (CASSANDRA-14092) +------------------------------------------------------------------ +(General upgrading instructions are available in the next section) + +The maximum expiration timestamp that can be represented by the storage engine is +2038-01-19T03:14:06+00:00, which means that inserts with TTL thatl expire after +this date are not currently supported. By default, INSERTS with TTL exceeding the +maximum supported date are rejected, but it's possible to choose a different + expiration overflow policy. See CASSANDRA-14092.txt for more details. + +Prior to 3.0.16 (3.0.X) and 3.11.2 (3.11.x) there was no protection against INSERTS +with TTL expiring after the maximum supported date, causing the expiration time +field to overflow and the records to expire immediately. Clusters in the 2.X and +lower series are not subject to this when assertions are enabled. Backed up SSTables +can be potentially recovered and recovery instructions can be found on the +CASSANDRA-14092.txt file. + +If you use or plan to use very large TTLS (10 to 20 years), read CASSANDRA-14092.txt +for more information. + GENERAL UPGRADING ADVICE FOR ANY VERSION ======================================== @@ -18,6 +38,7 @@ using the provided 'sstableupgrade' tool. Upgrading --------- + - See MAXIMUM TTL EXPIRATION DATE NOTICE above. - Cassandra is now relying on the JVM options to properly shutdown on OutOfMemoryError. By default it will rely on the OnOutOfMemoryError option as the ExitOnOutOfMemoryError and CrashOnOutOfMemoryError options are not supported by the older 1.7 and 1.8 JVMs. A warning will be logged at startup if none of those JVM diff --git a/conf/jvm.options b/conf/jvm.options index a7b3bd87f3..eb2ad19e43 100644 --- a/conf/jvm.options +++ b/conf/jvm.options @@ -46,6 +46,17 @@ # 100 MB per physical CPU core. #-Xmn800M +################################### +# EXPIRATION DATE OVERFLOW POLICY # +################################### + +# Defines how to handle INSERT requests with TTL exceeding the maximum supported expiration date: +# * REJECT: this is the default policy and will reject any requests with expiration date timestamp after 2038-01-19T03:14:06+00:00. +# * CAP: any insert with TTL expiring after 2038-01-19T03:14:06+00:00 will expire on 2038-01-19T03:14:06+00:00 and the client will receive a warning. +# * CAP_NOWARN: same as previous, except that the client warning will not be emitted. +# +#-Dcassandra.expiration_date_overflow_policy=REJECT + ################# # GC SETTINGS # ################# diff --git a/debian/rules b/debian/rules index 35f5a51e40..ff1d64df1b 100755 --- a/debian/rules +++ b/debian/rules @@ -63,7 +63,7 @@ binary-indep: build install dh_testroot dh_installchangelogs dh_installinit -u'start 50 2 3 4 5 . stop 50 0 1 6 .' - dh_installdocs README.asc CHANGES.txt NEWS.txt doc/cql3/CQL.css doc/cql3/CQL.html + dh_installdocs README.asc CHANGES.txt NEWS.txt doc/cql3/CQL.css doc/cql3/CQL.html CASSANDRA-14092.txt dh_installexamples tools/*.yaml dh_bash-completion dh_compress diff --git a/redhat/cassandra.spec b/redhat/cassandra.spec index 0dbe05d1d4..38ed20f144 100644 --- a/redhat/cassandra.spec +++ b/redhat/cassandra.spec @@ -118,7 +118,7 @@ exit 0 %files %defattr(0644,root,root,0755) -%doc CHANGES.txt LICENSE.txt README.asc NEWS.txt NOTICE.txt +%doc CHANGES.txt LICENSE.txt README.asc NEWS.txt NOTICE.txt CASSANDRA-14092.txt %attr(755,root,root) %{_bindir}/cassandra-stress %attr(755,root,root) %{_bindir}/cqlsh %attr(755,root,root) %{_bindir}/cqlsh.py diff --git a/src/java/org/apache/cassandra/cql3/Attributes.java b/src/java/org/apache/cassandra/cql3/Attributes.java index 4ed0f833b9..832d0a780b 100644 --- a/src/java/org/apache/cassandra/cql3/Attributes.java +++ b/src/java/org/apache/cassandra/cql3/Attributes.java @@ -18,12 +18,11 @@ package org.apache.cassandra.cql3; import java.nio.ByteBuffer; -import java.util.Collections; import java.util.List; -import com.google.common.collect.Iterables; - +import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.cql3.functions.Function; +import org.apache.cassandra.db.ExpirationDateOverflowHandling; import org.apache.cassandra.db.marshal.Int32Type; import org.apache.cassandra.db.marshal.LongType; import org.apache.cassandra.exceptions.InvalidRequestException; @@ -100,10 +99,13 @@ public class Attributes return LongType.instance.compose(tval); } - public int getTimeToLive(QueryOptions options) throws InvalidRequestException + public int getTimeToLive(QueryOptions options, CFMetaData metadata) throws InvalidRequestException { if (timeToLive == null) - return 0; + { + ExpirationDateOverflowHandling.maybeApplyExpirationDateOverflowPolicy(metadata, metadata.params.defaultTimeToLive, true); + return metadata.params.defaultTimeToLive; + } ByteBuffer tval = timeToLive.bindAndGet(options); if (tval == null) @@ -128,6 +130,8 @@ public class Attributes if (ttl > MAX_TTL) throw new InvalidRequestException(String.format("ttl is too large. requested (%d) maximum (%d)", ttl, MAX_TTL)); + ExpirationDateOverflowHandling.maybeApplyExpirationDateOverflowPolicy(metadata, ttl, false); + return ttl; } diff --git a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java index 8ae4d647b6..f4ad8b2194 100644 --- a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java @@ -200,7 +200,7 @@ public abstract class ModificationStatement implements CQLStatement public int getTimeToLive(QueryOptions options) throws InvalidRequestException { - return attrs.getTimeToLive(options); + return attrs.getTimeToLive(options, cfm); } public void checkAccess(ClientState state) throws InvalidRequestException, UnauthorizedException diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 536681f6ba..b5946bbc1b 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -1379,13 +1379,13 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean return CompactionManager.instance.performCleanup(ColumnFamilyStore.this, jobs); } - public CompactionManager.AllSSTableOpStatus scrub(boolean disableSnapshot, boolean skipCorrupted, boolean checkData, int jobs) throws ExecutionException, InterruptedException + public CompactionManager.AllSSTableOpStatus scrub(boolean disableSnapshot, boolean skipCorrupted, boolean checkData, boolean reinsertOverflowedTTL, int jobs) throws ExecutionException, InterruptedException { - return scrub(disableSnapshot, skipCorrupted, false, checkData, jobs); + return scrub(disableSnapshot, skipCorrupted, reinsertOverflowedTTL, false, checkData, jobs); } @VisibleForTesting - public CompactionManager.AllSSTableOpStatus scrub(boolean disableSnapshot, boolean skipCorrupted, boolean alwaysFail, boolean checkData, int jobs) throws ExecutionException, InterruptedException + public CompactionManager.AllSSTableOpStatus scrub(boolean disableSnapshot, boolean skipCorrupted, boolean reinsertOverflowedTTL, boolean alwaysFail, boolean checkData, int jobs) throws ExecutionException, InterruptedException { // skip snapshot creation during scrub, SEE JIRA 5891 if(!disableSnapshot) @@ -1393,7 +1393,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean try { - return CompactionManager.instance.performScrub(ColumnFamilyStore.this, skipCorrupted, checkData, jobs); + return CompactionManager.instance.performScrub(ColumnFamilyStore.this, skipCorrupted, checkData, reinsertOverflowedTTL, jobs); } catch(Throwable t) { diff --git a/src/java/org/apache/cassandra/db/ExpirationDateOverflowHandling.java b/src/java/org/apache/cassandra/db/ExpirationDateOverflowHandling.java new file mode 100644 index 0000000000..852dcb1f7d --- /dev/null +++ b/src/java/org/apache/cassandra/db/ExpirationDateOverflowHandling.java @@ -0,0 +1,121 @@ +/* + * 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; + +import java.util.concurrent.TimeUnit; + +import com.google.common.annotations.VisibleForTesting; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.slf4j.helpers.MessageFormatter; + +import org.apache.cassandra.config.CFMetaData; +import org.apache.cassandra.cql3.Attributes; +import org.apache.cassandra.db.rows.BufferCell; +import org.apache.cassandra.db.rows.Cell; +import org.apache.cassandra.exceptions.InvalidRequestException; +import org.apache.cassandra.service.ClientWarn; +import org.apache.cassandra.utils.NoSpamLogger; + +public class ExpirationDateOverflowHandling +{ + private static final Logger logger = LoggerFactory.getLogger(Attributes.class); + + private static final int EXPIRATION_OVERFLOW_WARNING_INTERVAL_MINUTES = Integer.getInteger("cassandra.expiration_overflow_warning_interval_minutes", 5); + + public enum ExpirationDateOverflowPolicy + { + REJECT, CAP_NOWARN, CAP + } + + @VisibleForTesting + public static ExpirationDateOverflowPolicy policy; + + static { + String policyAsString = System.getProperty("cassandra.expiration_date_overflow_policy", ExpirationDateOverflowPolicy.REJECT.name()); + try + { + policy = ExpirationDateOverflowPolicy.valueOf(policyAsString.toUpperCase()); + } + catch (RuntimeException e) + { + logger.warn("Invalid expiration date overflow policy: {}. Using default: {}", policyAsString, ExpirationDateOverflowPolicy.REJECT.name()); + policy = ExpirationDateOverflowPolicy.REJECT; + } + } + + public static final String MAXIMUM_EXPIRATION_DATE_EXCEEDED_WARNING = "Request on table {}.{} with {}ttl of {} seconds exceeds maximum supported expiration " + + "date of 2038-01-19T03:14:06+00:00 and will have its expiration capped to that date. " + + "In order to avoid this use a lower TTL or upgrade to a version where this limitation " + + "is fixed. See CASSANDRA-14092 for more details."; + + public static final String MAXIMUM_EXPIRATION_DATE_EXCEEDED_REJECT_MESSAGE = "Request on table %s.%s with %sttl of %d seconds exceeds maximum supported expiration " + + "date of 2038-01-19T03:14:06+00:00. In order to avoid this use a lower TTL, change " + + "the expiration date overflow policy or upgrade to a version where this limitation " + + "is fixed. See CASSANDRA-14092 for more details."; + + public static void maybeApplyExpirationDateOverflowPolicy(CFMetaData metadata, int ttl, boolean isDefaultTTL) throws InvalidRequestException + { + if (ttl == BufferCell.NO_TTL) + return; + + // Check for localExpirationTime overflow (CASSANDRA-14092) + int nowInSecs = (int)(System.currentTimeMillis() / 1000); + if (ttl + nowInSecs < 0) + { + switch (policy) + { + case CAP: + ClientWarn.instance.warn(MessageFormatter.arrayFormat(MAXIMUM_EXPIRATION_DATE_EXCEEDED_WARNING, new Object[] { metadata.ksName, + metadata.cfName, + isDefaultTTL? "default " : "", ttl }) + .getMessage()); + case CAP_NOWARN: + /** + * Capping at this stage is basically not rejecting the request. The actual capping is done + * by {@link #computeLocalExpirationTime(int, int)}, which converts the negative TTL + * to {@link org.apache.cassandra.db.BufferExpiringCell#MAX_DELETION_TIME} + */ + NoSpamLogger.log(logger, NoSpamLogger.Level.WARN, EXPIRATION_OVERFLOW_WARNING_INTERVAL_MINUTES, TimeUnit.MINUTES, MAXIMUM_EXPIRATION_DATE_EXCEEDED_WARNING, + metadata.ksName, metadata.cfName, isDefaultTTL? "default " : "", ttl); + return; + + default: + throw new InvalidRequestException(String.format(MAXIMUM_EXPIRATION_DATE_EXCEEDED_REJECT_MESSAGE, metadata.ksName, metadata.cfName, + isDefaultTTL? "default " : "", ttl)); + } + } + } + + /** + * This method computes the {@link Cell#localDeletionTime()}, maybe capping to the maximum representable value + * which is {@link Cell#MAX_DELETION_TIME}. + * + * Please note that the {@link ExpirationDateOverflowHandling.ExpirationDateOverflowPolicy} is applied + * during {@link ExpirationDateOverflowHandling#maybeApplyExpirationDateOverflowPolicy(CFMetaData, int, boolean)}, + * so if the request was not denied it means its expiration date should be capped. + * + * See CASSANDRA-14092 + */ + public static int computeLocalExpirationTime(int nowInSec, int timeToLive) + { + int localExpirationTime = nowInSec + timeToLive; + return localExpirationTime >= 0? localExpirationTime : Cell.MAX_DELETION_TIME; + } +} diff --git a/src/java/org/apache/cassandra/db/LegacyLayout.java b/src/java/org/apache/cassandra/db/LegacyLayout.java index 2117dd6e37..912d5910bf 100644 --- a/src/java/org/apache/cassandra/db/LegacyLayout.java +++ b/src/java/org/apache/cassandra/db/LegacyLayout.java @@ -25,6 +25,7 @@ import java.security.MessageDigest; import java.util.*; import org.apache.cassandra.cql3.SuperColumnCompatibility; +import org.apache.cassandra.thrift.Column; import org.apache.cassandra.utils.AbstractIterator; import com.google.common.collect.Iterators; import com.google.common.collect.Lists; @@ -1541,7 +1542,11 @@ public abstract class LegacyLayout public static LegacyCell expiring(CFMetaData metadata, ByteBuffer superColumnName, ByteBuffer name, ByteBuffer value, long timestamp, int ttl, int nowInSec) throws UnknownColumnException { - return new LegacyCell(Kind.EXPIRING, decodeCellName(metadata, superColumnName, name), value, timestamp, nowInSec + ttl, ttl); + /* + * CASSANDRA-14092: Max expiration date capping is maybe performed here, expiration overflow policy application + * is done at {@link org.apache.cassandra.thrift.ThriftValidation#validateTtl(CFMetaData, Column)} + */ + return new LegacyCell(Kind.EXPIRING, decodeCellName(metadata, superColumnName, name), value, timestamp, ExpirationDateOverflowHandling.computeLocalExpirationTime(nowInSec, ttl), ttl); } public static LegacyCell tombstone(CFMetaData metadata, ByteBuffer superColumnName, ByteBuffer name, long timestamp, int nowInSec) diff --git a/src/java/org/apache/cassandra/db/LivenessInfo.java b/src/java/org/apache/cassandra/db/LivenessInfo.java index 89e0578911..f6c9b629f8 100644 --- a/src/java/org/apache/cassandra/db/LivenessInfo.java +++ b/src/java/org/apache/cassandra/db/LivenessInfo.java @@ -21,6 +21,7 @@ import java.util.Objects; import java.security.MessageDigest; import org.apache.cassandra.config.CFMetaData; +import org.apache.cassandra.db.rows.Cell; import org.apache.cassandra.serializers.MarshalException; import org.apache.cassandra.utils.FBUtilities; @@ -40,7 +41,7 @@ import org.apache.cassandra.utils.FBUtilities; public class LivenessInfo { public static final long NO_TIMESTAMP = Long.MIN_VALUE; - public static final int NO_TTL = 0; + public static final int NO_TTL = Cell.NO_TTL; /** * Used as flag for representing an expired liveness. * @@ -48,7 +49,7 @@ public class LivenessInfo * (See {@link org.apache.cassandra.cql3.Attributes#MAX_TTL}) */ public static final int EXPIRED_LIVENESS_TTL = Integer.MAX_VALUE; - public static final int NO_EXPIRATION_TIME = Integer.MAX_VALUE; + public static final int NO_EXPIRATION_TIME = Cell.NO_DELETION_TIME; public static final LivenessInfo EMPTY = new LivenessInfo(NO_TIMESTAMP); @@ -71,7 +72,7 @@ public class LivenessInfo public static LivenessInfo expiring(long timestamp, int ttl, int nowInSec) { assert ttl != EXPIRED_LIVENESS_TTL; - return new ExpiringLivenessInfo(timestamp, ttl, nowInSec + ttl); + return new ExpiringLivenessInfo(timestamp, ttl, ExpirationDateOverflowHandling.computeLocalExpirationTime(nowInSec, ttl)); } public static LivenessInfo create(CFMetaData metadata, long timestamp, int ttl, int nowInSec) @@ -232,6 +233,11 @@ public class LivenessInfo return new LivenessInfo(newTimestamp); } + public LivenessInfo withUpdatedTimestampAndLocalDeletionTime(long newTimestamp, int newLocalDeletionTime) + { + return LivenessInfo.create(newTimestamp, ttl(), newLocalDeletionTime); + } + @Override public String toString() { diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java index eaf6dab595..1d5466795e 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java @@ -363,7 +363,15 @@ public class CompactionManager implements CompactionManagerMBean } } - public AllSSTableOpStatus performScrub(final ColumnFamilyStore cfs, final boolean skipCorrupted, final boolean checkData, int jobs) + public AllSSTableOpStatus performScrub(final ColumnFamilyStore cfs, final boolean skipCorrupted, final boolean checkData, + int jobs) + throws InterruptedException, ExecutionException + { + return performScrub(cfs, skipCorrupted, checkData, false, jobs); + } + + public AllSSTableOpStatus performScrub(final ColumnFamilyStore cfs, final boolean skipCorrupted, final boolean checkData, + final boolean reinsertOverflowedTTL, int jobs) throws InterruptedException, ExecutionException { return parallelAllSSTableOperation(cfs, new OneSSTableOperation() @@ -377,7 +385,7 @@ public class CompactionManager implements CompactionManagerMBean @Override public void execute(LifecycleTransaction input) throws IOException { - scrubOne(cfs, input, skipCorrupted, checkData); + scrubOne(cfs, input, skipCorrupted, checkData, reinsertOverflowedTTL); } }, jobs, OperationType.SCRUB); } @@ -745,11 +753,11 @@ public class CompactionManager implements CompactionManagerMBean } } - private void scrubOne(ColumnFamilyStore cfs, LifecycleTransaction modifier, boolean skipCorrupted, boolean checkData) throws IOException + private void scrubOne(ColumnFamilyStore cfs, LifecycleTransaction modifier, boolean skipCorrupted, boolean checkData, boolean reinsertOverflowedTTL) throws IOException { CompactionInfo.Holder scrubInfo = null; - try (Scrubber scrubber = new Scrubber(cfs, modifier, skipCorrupted, checkData)) + try (Scrubber scrubber = new Scrubber(cfs, modifier, skipCorrupted, checkData, reinsertOverflowedTTL)) { scrubInfo = scrubber.getScrubInfo(); metrics.beginCompaction(scrubInfo); diff --git a/src/java/org/apache/cassandra/db/compaction/Scrubber.java b/src/java/org/apache/cassandra/db/compaction/Scrubber.java index c8e0c53323..bc1150430f 100644 --- a/src/java/org/apache/cassandra/db/compaction/Scrubber.java +++ b/src/java/org/apache/cassandra/db/compaction/Scrubber.java @@ -37,6 +37,7 @@ import org.apache.cassandra.io.util.RandomAccessReader; import org.apache.cassandra.service.ActiveRepairService; import org.apache.cassandra.utils.*; import org.apache.cassandra.utils.concurrent.Refs; +import org.apache.cassandra.utils.memory.HeapAllocator; public class Scrubber implements Closeable { @@ -45,6 +46,7 @@ public class Scrubber implements Closeable private final LifecycleTransaction transaction; private final File destination; private final boolean skipCorrupted; + private final boolean reinsertOverflowedTTLRows; private final boolean isCommutative; private final boolean isIndex; @@ -65,6 +67,8 @@ public class Scrubber implements Closeable long currentRowPositionFromIndex; long nextRowPositionFromIndex; + private NegativeLocalDeletionInfoMetrics negativeLocalDeletionInfoMetrics = new NegativeLocalDeletionInfoMetrics(); + private final OutputHandler outputHandler; private static final Comparator partitionComparator = new Comparator() @@ -78,7 +82,13 @@ public class Scrubber implements Closeable public Scrubber(ColumnFamilyStore cfs, LifecycleTransaction transaction, boolean skipCorrupted, boolean checkData) throws IOException { - this(cfs, transaction, skipCorrupted, new OutputHandler.LogOutput(), checkData); + this(cfs, transaction, skipCorrupted, checkData, false); + } + + public Scrubber(ColumnFamilyStore cfs, LifecycleTransaction transaction, boolean skipCorrupted, boolean checkData, + boolean reinsertOverflowedTTLRows) throws IOException + { + this(cfs, transaction, skipCorrupted, new OutputHandler.LogOutput(), checkData, reinsertOverflowedTTLRows); } @SuppressWarnings("resource") @@ -86,17 +96,18 @@ public class Scrubber implements Closeable LifecycleTransaction transaction, boolean skipCorrupted, OutputHandler outputHandler, - boolean checkData) throws IOException + boolean checkData, + boolean reinsertOverflowedTTLRows) throws IOException { this.cfs = cfs; this.transaction = transaction; this.sstable = transaction.onlyOne(); this.outputHandler = outputHandler; this.skipCorrupted = skipCorrupted; + this.reinsertOverflowedTTLRows = reinsertOverflowedTTLRows; this.rowIndexEntrySerializer = sstable.descriptor.version.getSSTableFormat().getIndexSerializer(sstable.metadata, sstable.descriptor.version, sstable.header); - List toScrub = Collections.singletonList(sstable); // Calculate the expected compacted filesize @@ -134,6 +145,9 @@ public class Scrubber implements Closeable this.currentRowPositionFromIndex = 0; this.nextRowPositionFromIndex = 0; + + if (reinsertOverflowedTTLRows) + outputHandler.output("Starting scrub with reinsert overflowed TTL option"); } private UnfilteredRowIterator withValidation(UnfilteredRowIterator iter, String filename) @@ -290,18 +304,20 @@ public class Scrubber implements Closeable } if (completed) + { + outputHandler.output("Scrub of " + sstable + " complete: " + goodRows + " rows in new sstable and " + emptyRows + " empty (tombstoned) rows dropped"); + if (negativeLocalDeletionInfoMetrics.fixedRows > 0) + outputHandler.output("Fixed " + negativeLocalDeletionInfoMetrics.fixedRows + " rows with overflowed local deletion time."); + if (badRows > 0) + outputHandler.warn("Unable to recover " + badRows + " rows that were skipped. You can attempt manual recovery from the pre-scrub snapshot. You can also run nodetool repair to transfer the data from a healthy replica, if any"); + } + else { if (badRows > 0) outputHandler.warn("No valid rows found while scrubbing " + sstable + "; it is marked for deletion now. If you want to attempt manual recovery, you can find a copy in the pre-scrub snapshot"); else outputHandler.output("Scrub of " + sstable + " complete; looks like all " + emptyRows + " rows were tombstoned"); } - else - { - outputHandler.output("Scrub of " + sstable + " complete: " + goodRows + " rows in new sstable and " + emptyRows + " empty (tombstoned) rows dropped"); - if (badRows > 0) - outputHandler.warn("Unable to recover " + badRows + " rows that were skipped. You can attempt manual recovery from the pre-scrub snapshot. You can also run nodetool repair to transfer the data from a healthy replica, if any"); - } } @SuppressWarnings("resource") @@ -310,7 +326,7 @@ public class Scrubber implements Closeable // OrderCheckerIterator will check, at iteration time, that the rows are in the proper order. If it detects // that one row is out of order, it will stop returning them. The remaining rows will be sorted and added // to the outOfOrder set that will be later written to a new SSTable. - OrderCheckerIterator sstableIterator = new OrderCheckerIterator(new RowMergingSSTableIterator(sstable, dataFile, key), + OrderCheckerIterator sstableIterator = new OrderCheckerIterator(getIterator(key), cfs.metadata.comparator); try (UnfilteredRowIterator iterator = withValidation(sstableIterator, dataFile.getPath())) @@ -336,6 +352,18 @@ public class Scrubber implements Closeable return true; } + /** + * Only wrap with {@link FixNegativeLocalDeletionTimeIterator} if {@link #reinsertOverflowedTTLRows} option + * is specified + */ + private UnfilteredRowIterator getIterator(DecoratedKey key) + { + RowMergingSSTableIterator rowMergingIterator = new RowMergingSSTableIterator(sstable, dataFile, key); + return reinsertOverflowedTTLRows ? new FixNegativeLocalDeletionTimeIterator(rowMergingIterator, + outputHandler, + negativeLocalDeletionInfoMetrics) : rowMergingIterator; + } + private void updateIndexKey() { currentIndexKey = nextIndexKey; @@ -477,6 +505,11 @@ public class Scrubber implements Closeable } } + public class NegativeLocalDeletionInfoMetrics + { + public volatile int fixedRows = 0; + } + /** * During 2.x migration, under some circumstances rows might have gotten duplicated. * Merging iterator merges rows with same clustering. @@ -517,6 +550,7 @@ public class Scrubber implements Closeable return next; } + } /** @@ -618,6 +652,152 @@ public class Scrubber implements Closeable previous = next; return next; } + } + /** + * This iterator converts negative {@link AbstractCell#localDeletionTime()} into {@link AbstractCell#MAX_DELETION_TIME} + * + * This is to recover entries with overflowed localExpirationTime due to CASSANDRA-14092 + */ + private static final class FixNegativeLocalDeletionTimeIterator extends AbstractIterator implements UnfilteredRowIterator + { + /** + * The decorated iterator. + */ + private final UnfilteredRowIterator iterator; + + private final OutputHandler outputHandler; + private final NegativeLocalDeletionInfoMetrics negativeLocalExpirationTimeMetrics; + + public FixNegativeLocalDeletionTimeIterator(UnfilteredRowIterator iterator, OutputHandler outputHandler, + NegativeLocalDeletionInfoMetrics negativeLocalDeletionInfoMetrics) + { + this.iterator = iterator; + this.outputHandler = outputHandler; + this.negativeLocalExpirationTimeMetrics = negativeLocalDeletionInfoMetrics; + } + + public CFMetaData metadata() + { + return iterator.metadata(); + } + + public boolean isReverseOrder() + { + return iterator.isReverseOrder(); + } + + public PartitionColumns columns() + { + return iterator.columns(); + } + + public DecoratedKey partitionKey() + { + return iterator.partitionKey(); + } + + public Row staticRow() + { + return iterator.staticRow(); + } + + @Override + public boolean isEmpty() + { + return iterator.isEmpty(); + } + + public void close() + { + iterator.close(); + } + + public DeletionTime partitionLevelDeletion() + { + return iterator.partitionLevelDeletion(); + } + + public EncodingStats stats() + { + return iterator.stats(); + } + + protected Unfiltered computeNext() + { + if (!iterator.hasNext()) + return endOfData(); + + Unfiltered next = iterator.next(); + if (!next.isRow()) + return next; + + if (hasNegativeLocalExpirationTime((Row) next)) + { + outputHandler.debug(String.format("Found row with negative local expiration time: %s", next.toString(metadata(), false))); + negativeLocalExpirationTimeMetrics.fixedRows++; + return fixNegativeLocalExpirationTime((Row) next); + } + + return next; + } + + private boolean hasNegativeLocalExpirationTime(Row next) + { + Row row = next; + if (row.primaryKeyLivenessInfo().isExpiring() && row.primaryKeyLivenessInfo().localExpirationTime() < 0) + { + return true; + } + + for (ColumnData cd : row) + { + if (cd.column().isSimple()) + { + Cell cell = (Cell)cd; + if (cell.isExpiring() && cell.localDeletionTime() < 0) + return true; + } + else + { + ComplexColumnData complexData = (ComplexColumnData)cd; + for (Cell cell : complexData) + { + if (cell.isExpiring() && cell.localDeletionTime() < 0) + return true; + } + } + } + + return false; + } + + private Unfiltered fixNegativeLocalExpirationTime(Row row) + { + Row.Builder builder = HeapAllocator.instance.cloningBTreeRowBuilder(); + builder.newRow(row.clustering()); + builder.addPrimaryKeyLivenessInfo(row.primaryKeyLivenessInfo().isExpiring() && row.primaryKeyLivenessInfo().localExpirationTime() < 0 ? + row.primaryKeyLivenessInfo().withUpdatedTimestampAndLocalDeletionTime(row.primaryKeyLivenessInfo().timestamp() + 1, AbstractCell.MAX_DELETION_TIME) + :row.primaryKeyLivenessInfo()); + builder.addRowDeletion(row.deletion()); + for (ColumnData cd : row) + { + if (cd.column().isSimple()) + { + Cell cell = (Cell)cd; + builder.addCell(cell.isExpiring() && cell.localDeletionTime() < 0 ? cell.withUpdatedTimestampAndLocalDeletionTime(cell.timestamp() + 1, AbstractCell.MAX_DELETION_TIME) : cell); + } + else + { + ComplexColumnData complexData = (ComplexColumnData)cd; + builder.addComplexDeletion(complexData.column(), complexData.complexDeletion()); + for (Cell cell : complexData) + { + builder.addCell(cell.isExpiring() && cell.localDeletionTime() < 0 ? cell.withUpdatedTimestampAndLocalDeletionTime(cell.timestamp() + 1, AbstractCell.MAX_DELETION_TIME) : cell); + } + } + } + return builder.build(); + } } } diff --git a/src/java/org/apache/cassandra/db/rows/BufferCell.java b/src/java/org/apache/cassandra/db/rows/BufferCell.java index 82ae02ca68..df2619c7ce 100644 --- a/src/java/org/apache/cassandra/db/rows/BufferCell.java +++ b/src/java/org/apache/cassandra/db/rows/BufferCell.java @@ -74,7 +74,7 @@ public class BufferCell extends AbstractCell public static BufferCell expiring(ColumnDefinition column, long timestamp, int ttl, int nowInSec, ByteBuffer value, CellPath path) { assert ttl != NO_TTL; - return new BufferCell(column, timestamp, ttl, nowInSec + ttl, value, path); + return new BufferCell(column, timestamp, ttl, ExpirationDateOverflowHandling.computeLocalExpirationTime(nowInSec, ttl), value, path); } public static BufferCell tombstone(ColumnDefinition column, long timestamp, int nowInSec) @@ -142,6 +142,11 @@ public class BufferCell extends AbstractCell return new BufferCell(column, timestamp, ttl, localDeletionTime, newValue, path); } + public Cell withUpdatedTimestampAndLocalDeletionTime(long newTimestamp, int newLocalDeletionTime) + { + return new BufferCell(column, newTimestamp, ttl, newLocalDeletionTime, value, path); + } + public Cell copy(AbstractAllocator allocator) { if (!value.hasRemaining()) diff --git a/src/java/org/apache/cassandra/db/rows/Cell.java b/src/java/org/apache/cassandra/db/rows/Cell.java index d10cc74286..c69e11f65c 100644 --- a/src/java/org/apache/cassandra/db/rows/Cell.java +++ b/src/java/org/apache/cassandra/db/rows/Cell.java @@ -21,7 +21,13 @@ import java.io.IOException; import java.nio.ByteBuffer; import java.util.Comparator; +import com.google.common.annotations.VisibleForTesting; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.config.ColumnDefinition; +import org.apache.cassandra.cql3.Attributes; import org.apache.cassandra.db.*; import org.apache.cassandra.io.util.DataOutputPlus; import org.apache.cassandra.io.util.DataInputPlus; @@ -40,6 +46,7 @@ public abstract class Cell extends ColumnData { public static final int NO_TTL = 0; public static final int NO_DELETION_TIME = Integer.MAX_VALUE; + public static final int MAX_DELETION_TIME = Integer.MAX_VALUE - 1; public final static Comparator comparator = (c1, c2) -> { @@ -133,6 +140,8 @@ public abstract class Cell extends ColumnData public abstract Cell withUpdatedValue(ByteBuffer newValue); + public abstract Cell withUpdatedTimestampAndLocalDeletionTime(long newTimestamp, int newLocalDeletionTime); + public abstract Cell copy(AbstractAllocator allocator); @Override diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index e5a50dd0a6..cf8e257936 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -2716,11 +2716,16 @@ public class StorageService extends NotificationBroadcasterSupport implements IE } public int scrub(boolean disableSnapshot, boolean skipCorrupted, boolean checkData, int jobs, String keyspaceName, String... tables) throws IOException, ExecutionException, InterruptedException + { + return scrub(disableSnapshot, skipCorrupted, checkData, false, jobs, keyspaceName, tables); + } + + public int scrub(boolean disableSnapshot, boolean skipCorrupted, boolean checkData, boolean reinsertOverflowedTTL, int jobs, String keyspaceName, String... tables) throws IOException, ExecutionException, InterruptedException { CompactionManager.AllSSTableOpStatus status = CompactionManager.AllSSTableOpStatus.SUCCESSFUL; for (ColumnFamilyStore cfStore : getValidColumnFamilies(true, false, keyspaceName, tables)) { - CompactionManager.AllSSTableOpStatus oneStatus = cfStore.scrub(disableSnapshot, skipCorrupted, checkData, jobs); + CompactionManager.AllSSTableOpStatus oneStatus = cfStore.scrub(disableSnapshot, skipCorrupted, reinsertOverflowedTTL, checkData, jobs); if (oneStatus != CompactionManager.AllSSTableOpStatus.SUCCESSFUL) status = oneStatus; } diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java index 7344ca8c77..10d47f716f 100644 --- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java +++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java @@ -265,8 +265,11 @@ public interface StorageServiceMBean extends NotificationEmitter public int scrub(boolean disableSnapshot, boolean skipCorrupted, String keyspaceName, String... tableNames) throws IOException, ExecutionException, InterruptedException; @Deprecated public int scrub(boolean disableSnapshot, boolean skipCorrupted, boolean checkData, String keyspaceName, String... tableNames) throws IOException, ExecutionException, InterruptedException; + @Deprecated public int scrub(boolean disableSnapshot, boolean skipCorrupted, boolean checkData, int jobs, String keyspaceName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException; + public int scrub(boolean disableSnapshot, boolean skipCorrupted, boolean checkData, boolean reinsertOverflowedTTL, int jobs, String keyspaceName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException; + /** * Verify (checksums of) the given keyspace. * If tableNames array is empty, all CFs are verified. diff --git a/src/java/org/apache/cassandra/thrift/ThriftValidation.java b/src/java/org/apache/cassandra/thrift/ThriftValidation.java index 6ad791d888..2ab0330078 100644 --- a/src/java/org/apache/cassandra/thrift/ThriftValidation.java +++ b/src/java/org/apache/cassandra/thrift/ThriftValidation.java @@ -332,7 +332,7 @@ public class ThriftValidation if (isCommutative) throw new org.apache.cassandra.exceptions.InvalidRequestException("invalid operation for commutative table " + metadata.cfName); - validateTtl(cosc.column); + validateTtl(metadata, cosc.column); validateColumnPath(metadata, new ColumnPath(metadata.cfName).setSuper_column((ByteBuffer)null).setColumn(cosc.column.name)); validateColumnData(metadata, null, cosc.column); } @@ -367,7 +367,7 @@ public class ThriftValidation } } - private static void validateTtl(Column column) throws org.apache.cassandra.exceptions.InvalidRequestException + private static void validateTtl(CFMetaData metadata, Column column) throws org.apache.cassandra.exceptions.InvalidRequestException { if (column.isSetTtl()) { @@ -376,9 +376,11 @@ public class ThriftValidation if (column.ttl > Attributes.MAX_TTL) throw new org.apache.cassandra.exceptions.InvalidRequestException(String.format("ttl is too large. requested (%d) maximum (%d)", column.ttl, Attributes.MAX_TTL)); + ExpirationDateOverflowHandling.maybeApplyExpirationDateOverflowPolicy(metadata, column.ttl, false); } else { + ExpirationDateOverflowHandling.maybeApplyExpirationDateOverflowPolicy(metadata, metadata.params.defaultTimeToLive, true); // if it's not set, then it should be zero -- here we are just checking to make sure Thrift doesn't change that contract with us. assert column.ttl == 0; } @@ -452,7 +454,7 @@ public class ThriftValidation */ public static void validateColumnData(CFMetaData metadata, ByteBuffer scName, Column column) throws org.apache.cassandra.exceptions.InvalidRequestException { - validateTtl(column); + validateTtl(metadata, column); if (!column.isSetValue()) throw new org.apache.cassandra.exceptions.InvalidRequestException("Column value is required"); if (!column.isSetTimestamp()) diff --git a/src/java/org/apache/cassandra/tools/NodeProbe.java b/src/java/org/apache/cassandra/tools/NodeProbe.java index 172b505614..0d3c0789d3 100644 --- a/src/java/org/apache/cassandra/tools/NodeProbe.java +++ b/src/java/org/apache/cassandra/tools/NodeProbe.java @@ -251,9 +251,9 @@ public class NodeProbe implements AutoCloseable return ssProxy.forceKeyspaceCleanup(jobs, keyspaceName, tables); } - public int scrub(boolean disableSnapshot, boolean skipCorrupted, boolean checkData, int jobs, String keyspaceName, String... tables) throws IOException, ExecutionException, InterruptedException + public int scrub(boolean disableSnapshot, boolean skipCorrupted, boolean checkData, boolean reinsertOverflowedTTL, int jobs, String keyspaceName, String... tables) throws IOException, ExecutionException, InterruptedException { - return ssProxy.scrub(disableSnapshot, skipCorrupted, checkData, jobs, keyspaceName, tables); + return ssProxy.scrub(disableSnapshot, skipCorrupted, checkData, reinsertOverflowedTTL, jobs, keyspaceName, tables); } public int verify(boolean extendedVerify, String keyspaceName, String... tableNames) throws IOException, ExecutionException, InterruptedException @@ -288,10 +288,10 @@ public class NodeProbe implements AutoCloseable } } - public void scrub(PrintStream out, boolean disableSnapshot, boolean skipCorrupted, boolean checkData, int jobs, String keyspaceName, String... tables) throws IOException, ExecutionException, InterruptedException + public void scrub(PrintStream out, boolean disableSnapshot, boolean skipCorrupted, boolean checkData, boolean reinsertOverflowedTTL, int jobs, String keyspaceName, String... tables) throws IOException, ExecutionException, InterruptedException { checkJobs(out, jobs); - switch (scrub(disableSnapshot, skipCorrupted, checkData, jobs, keyspaceName, tables)) + switch (ssProxy.scrub(disableSnapshot, skipCorrupted, checkData, reinsertOverflowedTTL, jobs, keyspaceName, tables)) { case 1: failed = true; diff --git a/src/java/org/apache/cassandra/tools/SSTableMetadataViewer.java b/src/java/org/apache/cassandra/tools/SSTableMetadataViewer.java index 6076e324a8..19af957228 100644 --- a/src/java/org/apache/cassandra/tools/SSTableMetadataViewer.java +++ b/src/java/org/apache/cassandra/tools/SSTableMetadataViewer.java @@ -90,6 +90,7 @@ public class SSTableMetadataViewer { out.printf("Minimum timestamp: %s%n", stats.minTimestamp); out.printf("Maximum timestamp: %s%n", stats.maxTimestamp); + out.printf("SSTable min local deletion time: %s%n", stats.minLocalDeletionTime); out.printf("SSTable max local deletion time: %s%n", stats.maxLocalDeletionTime); out.printf("Compression ratio: %s%n", stats.compressionRatio); out.printf("Estimated droppable tombstones: %s%n", stats.getEstimatedDroppableTombstoneRatio((int) (System.currentTimeMillis() / 1000) - gcgs)); diff --git a/src/java/org/apache/cassandra/tools/StandaloneScrubber.java b/src/java/org/apache/cassandra/tools/StandaloneScrubber.java index 424943031c..4778d7263c 100644 --- a/src/java/org/apache/cassandra/tools/StandaloneScrubber.java +++ b/src/java/org/apache/cassandra/tools/StandaloneScrubber.java @@ -43,6 +43,12 @@ import static org.apache.cassandra.tools.BulkLoader.CmdLineOptions; public class StandaloneScrubber { + public static final String REINSERT_OVERFLOWED_TTL_OPTION_DESCRIPTION = "Rewrites rows with overflowed expiration date affected by CASSANDRA-14092 with " + + "the maximum supported expiration date of 2038-01-19T03:14:06+00:00. " + + "The rows are rewritten with the original timestamp incremented by one millisecond " + + "to override/supersede any potential tombstone that may have been generated " + + "during compaction of the affected rows."; + private static final String TOOL_NAME = "sstablescrub"; private static final String VERBOSE_OPTION = "verbose"; private static final String DEBUG_OPTION = "debug"; @@ -50,6 +56,7 @@ public class StandaloneScrubber private static final String MANIFEST_CHECK_OPTION = "manifest-check"; private static final String SKIP_CORRUPTED_OPTION = "skip-corrupted"; private static final String NO_VALIDATE_OPTION = "no-validate"; + private static final String REINSERT_OVERFLOWED_TTL_OPTION = "reinsert-overflowed-ttl"; public static void main(String args[]) { @@ -122,7 +129,7 @@ public class StandaloneScrubber try (LifecycleTransaction txn = LifecycleTransaction.offline(OperationType.SCRUB, sstable)) { txn.obsoleteOriginals(); // make sure originals are deleted and avoid NPE if index is missing, CASSANDRA-9591 - try (Scrubber scrubber = new Scrubber(cfs, txn, options.skipCorrupted, handler, !options.noValidate)) + try (Scrubber scrubber = new Scrubber(cfs, txn, options.skipCorrupted, handler, !options.noValidate, options.reinserOverflowedTTL)) { scrubber.scrub(); } @@ -199,6 +206,7 @@ public class StandaloneScrubber public boolean manifestCheckOnly; public boolean skipCorrupted; public boolean noValidate; + public boolean reinserOverflowedTTL; private Options(String keyspaceName, String cfName) { @@ -239,6 +247,7 @@ public class StandaloneScrubber opts.manifestCheckOnly = cmd.hasOption(MANIFEST_CHECK_OPTION); opts.skipCorrupted = cmd.hasOption(SKIP_CORRUPTED_OPTION); opts.noValidate = cmd.hasOption(NO_VALIDATE_OPTION); + opts.reinserOverflowedTTL = cmd.hasOption(REINSERT_OVERFLOWED_TTL_OPTION); return opts; } @@ -265,6 +274,7 @@ public class StandaloneScrubber options.addOption("m", MANIFEST_CHECK_OPTION, "only check and repair the leveled manifest, without actually scrubbing the sstables"); options.addOption("s", SKIP_CORRUPTED_OPTION, "skip corrupt rows in counter tables"); options.addOption("n", NO_VALIDATE_OPTION, "do not validate columns using column validator"); + options.addOption("r", REINSERT_OVERFLOWED_TTL_OPTION, REINSERT_OVERFLOWED_TTL_OPTION_DESCRIPTION); return options; } diff --git a/src/java/org/apache/cassandra/tools/nodetool/Scrub.java b/src/java/org/apache/cassandra/tools/nodetool/Scrub.java index 2345a8533f..ead2fd4c37 100644 --- a/src/java/org/apache/cassandra/tools/nodetool/Scrub.java +++ b/src/java/org/apache/cassandra/tools/nodetool/Scrub.java @@ -26,6 +26,7 @@ import java.util.List; import org.apache.cassandra.tools.NodeProbe; import org.apache.cassandra.tools.NodeTool.NodeToolCmd; +import org.apache.cassandra.tools.StandaloneScrubber; @Command(name = "scrub", description = "Scrub (rebuild sstables for) one or more tables") public class Scrub extends NodeToolCmd @@ -48,6 +49,11 @@ public class Scrub extends NodeToolCmd description = "Do not validate columns using column validator") private boolean noValidation = false; + @Option(title = "reinsert_overflowed_ttl", + name = {"r", "--reinsert-overflowed-ttl"}, + description = StandaloneScrubber.REINSERT_OVERFLOWED_TTL_OPTION_DESCRIPTION) + private boolean reinsertOverflowedTTL = false; + @Option(title = "jobs", name = {"-j", "--jobs"}, description = "Number of sstables to scrub simultanously, set to 0 to use all available compaction threads") @@ -63,7 +69,7 @@ public class Scrub extends NodeToolCmd { try { - probe.scrub(System.out, disableSnapshot, skipCorrupted, !noValidation, jobs, keyspace, tableNames); + probe.scrub(System.out, disableSnapshot, skipCorrupted, !noValidation, reinsertOverflowedTTL, jobs, keyspace, tableNames); } catch (IllegalArgumentException e) { diff --git a/test/data/negative-local-expiration-test/table1/mc-1-big-CompressionInfo.db b/test/data/negative-local-expiration-test/table1/mc-1-big-CompressionInfo.db new file mode 100644 index 0000000000..d759cec5c2 Binary files /dev/null and b/test/data/negative-local-expiration-test/table1/mc-1-big-CompressionInfo.db differ diff --git a/test/data/negative-local-expiration-test/table1/mc-1-big-Data.db b/test/data/negative-local-expiration-test/table1/mc-1-big-Data.db new file mode 100644 index 0000000000..e7a72da607 Binary files /dev/null and b/test/data/negative-local-expiration-test/table1/mc-1-big-Data.db differ diff --git a/test/data/negative-local-expiration-test/table1/mc-1-big-Digest.crc32 b/test/data/negative-local-expiration-test/table1/mc-1-big-Digest.crc32 new file mode 100644 index 0000000000..a3c633a846 --- /dev/null +++ b/test/data/negative-local-expiration-test/table1/mc-1-big-Digest.crc32 @@ -0,0 +1 @@ +203700622 \ No newline at end of file diff --git a/test/data/negative-local-expiration-test/table1/mc-1-big-Filter.db b/test/data/negative-local-expiration-test/table1/mc-1-big-Filter.db new file mode 100644 index 0000000000..a397f351d9 Binary files /dev/null and b/test/data/negative-local-expiration-test/table1/mc-1-big-Filter.db differ diff --git a/test/data/negative-local-expiration-test/table1/mc-1-big-Index.db b/test/data/negative-local-expiration-test/table1/mc-1-big-Index.db new file mode 100644 index 0000000000..d7427249d0 Binary files /dev/null and b/test/data/negative-local-expiration-test/table1/mc-1-big-Index.db differ diff --git a/test/data/negative-local-expiration-test/table1/mc-1-big-Statistics.db b/test/data/negative-local-expiration-test/table1/mc-1-big-Statistics.db new file mode 100644 index 0000000000..faf367b5fb Binary files /dev/null and b/test/data/negative-local-expiration-test/table1/mc-1-big-Statistics.db differ diff --git a/test/data/negative-local-expiration-test/table1/mc-1-big-Summary.db b/test/data/negative-local-expiration-test/table1/mc-1-big-Summary.db new file mode 100644 index 0000000000..66cf70f7c7 Binary files /dev/null and b/test/data/negative-local-expiration-test/table1/mc-1-big-Summary.db differ diff --git a/test/data/negative-local-expiration-test/table1/mc-1-big-TOC.txt b/test/data/negative-local-expiration-test/table1/mc-1-big-TOC.txt new file mode 100644 index 0000000000..45113dc798 --- /dev/null +++ b/test/data/negative-local-expiration-test/table1/mc-1-big-TOC.txt @@ -0,0 +1,8 @@ +CompressionInfo.db +Data.db +Summary.db +Filter.db +Statistics.db +TOC.txt +Digest.crc32 +Index.db diff --git a/test/data/negative-local-expiration-test/table2/mc-1-big-CompressionInfo.db b/test/data/negative-local-expiration-test/table2/mc-1-big-CompressionInfo.db new file mode 100644 index 0000000000..1759c09c63 Binary files /dev/null and b/test/data/negative-local-expiration-test/table2/mc-1-big-CompressionInfo.db differ diff --git a/test/data/negative-local-expiration-test/table2/mc-1-big-Data.db b/test/data/negative-local-expiration-test/table2/mc-1-big-Data.db new file mode 100644 index 0000000000..c1de5721bc Binary files /dev/null and b/test/data/negative-local-expiration-test/table2/mc-1-big-Data.db differ diff --git a/test/data/negative-local-expiration-test/table2/mc-1-big-Digest.crc32 b/test/data/negative-local-expiration-test/table2/mc-1-big-Digest.crc32 new file mode 100644 index 0000000000..0403b5bdcd --- /dev/null +++ b/test/data/negative-local-expiration-test/table2/mc-1-big-Digest.crc32 @@ -0,0 +1 @@ +82785930 \ No newline at end of file diff --git a/test/data/negative-local-expiration-test/table2/mc-1-big-Filter.db b/test/data/negative-local-expiration-test/table2/mc-1-big-Filter.db new file mode 100644 index 0000000000..a397f351d9 Binary files /dev/null and b/test/data/negative-local-expiration-test/table2/mc-1-big-Filter.db differ diff --git a/test/data/negative-local-expiration-test/table2/mc-1-big-Index.db b/test/data/negative-local-expiration-test/table2/mc-1-big-Index.db new file mode 100644 index 0000000000..a0477eb8ee Binary files /dev/null and b/test/data/negative-local-expiration-test/table2/mc-1-big-Index.db differ diff --git a/test/data/negative-local-expiration-test/table2/mc-1-big-Statistics.db b/test/data/negative-local-expiration-test/table2/mc-1-big-Statistics.db new file mode 100644 index 0000000000..e9d65771bf Binary files /dev/null and b/test/data/negative-local-expiration-test/table2/mc-1-big-Statistics.db differ diff --git a/test/data/negative-local-expiration-test/table2/mc-1-big-Summary.db b/test/data/negative-local-expiration-test/table2/mc-1-big-Summary.db new file mode 100644 index 0000000000..66cf70f7c7 Binary files /dev/null and b/test/data/negative-local-expiration-test/table2/mc-1-big-Summary.db differ diff --git a/test/data/negative-local-expiration-test/table2/mc-1-big-TOC.txt b/test/data/negative-local-expiration-test/table2/mc-1-big-TOC.txt new file mode 100644 index 0000000000..45113dc798 --- /dev/null +++ b/test/data/negative-local-expiration-test/table2/mc-1-big-TOC.txt @@ -0,0 +1,8 @@ +CompressionInfo.db +Data.db +Summary.db +Filter.db +Statistics.db +TOC.txt +Digest.crc32 +Index.db diff --git a/test/data/negative-local-expiration-test/table3/mc-1-big-CompressionInfo.db b/test/data/negative-local-expiration-test/table3/mc-1-big-CompressionInfo.db new file mode 100644 index 0000000000..b4de068379 Binary files /dev/null and b/test/data/negative-local-expiration-test/table3/mc-1-big-CompressionInfo.db differ diff --git a/test/data/negative-local-expiration-test/table3/mc-1-big-Data.db b/test/data/negative-local-expiration-test/table3/mc-1-big-Data.db new file mode 100644 index 0000000000..e96f77253e Binary files /dev/null and b/test/data/negative-local-expiration-test/table3/mc-1-big-Data.db differ diff --git a/test/data/negative-local-expiration-test/table3/mc-1-big-Digest.crc32 b/test/data/negative-local-expiration-test/table3/mc-1-big-Digest.crc32 new file mode 100644 index 0000000000..459804bdf3 --- /dev/null +++ b/test/data/negative-local-expiration-test/table3/mc-1-big-Digest.crc32 @@ -0,0 +1 @@ +3064924389 \ No newline at end of file diff --git a/test/data/negative-local-expiration-test/table3/mc-1-big-Filter.db b/test/data/negative-local-expiration-test/table3/mc-1-big-Filter.db new file mode 100644 index 0000000000..a397f351d9 Binary files /dev/null and b/test/data/negative-local-expiration-test/table3/mc-1-big-Filter.db differ diff --git a/test/data/negative-local-expiration-test/table3/mc-1-big-Index.db b/test/data/negative-local-expiration-test/table3/mc-1-big-Index.db new file mode 100644 index 0000000000..807a27b8bd Binary files /dev/null and b/test/data/negative-local-expiration-test/table3/mc-1-big-Index.db differ diff --git a/test/data/negative-local-expiration-test/table3/mc-1-big-Statistics.db b/test/data/negative-local-expiration-test/table3/mc-1-big-Statistics.db new file mode 100644 index 0000000000..1ee01e674a Binary files /dev/null and b/test/data/negative-local-expiration-test/table3/mc-1-big-Statistics.db differ diff --git a/test/data/negative-local-expiration-test/table3/mc-1-big-Summary.db b/test/data/negative-local-expiration-test/table3/mc-1-big-Summary.db new file mode 100644 index 0000000000..66cf70f7c7 Binary files /dev/null and b/test/data/negative-local-expiration-test/table3/mc-1-big-Summary.db differ diff --git a/test/data/negative-local-expiration-test/table3/mc-1-big-TOC.txt b/test/data/negative-local-expiration-test/table3/mc-1-big-TOC.txt new file mode 100644 index 0000000000..f4455377aa --- /dev/null +++ b/test/data/negative-local-expiration-test/table3/mc-1-big-TOC.txt @@ -0,0 +1,8 @@ +Summary.db +TOC.txt +Filter.db +Index.db +Digest.crc32 +CompressionInfo.db +Data.db +Statistics.db diff --git a/test/data/negative-local-expiration-test/table4/mc-1-big-CompressionInfo.db b/test/data/negative-local-expiration-test/table4/mc-1-big-CompressionInfo.db new file mode 100644 index 0000000000..5d22c044a4 Binary files /dev/null and b/test/data/negative-local-expiration-test/table4/mc-1-big-CompressionInfo.db differ diff --git a/test/data/negative-local-expiration-test/table4/mc-1-big-Data.db b/test/data/negative-local-expiration-test/table4/mc-1-big-Data.db new file mode 100644 index 0000000000..a22a7a3044 Binary files /dev/null and b/test/data/negative-local-expiration-test/table4/mc-1-big-Data.db differ diff --git a/test/data/negative-local-expiration-test/table4/mc-1-big-Digest.crc32 b/test/data/negative-local-expiration-test/table4/mc-1-big-Digest.crc32 new file mode 100644 index 0000000000..db7a6c765c --- /dev/null +++ b/test/data/negative-local-expiration-test/table4/mc-1-big-Digest.crc32 @@ -0,0 +1 @@ +1803989939 \ No newline at end of file diff --git a/test/data/negative-local-expiration-test/table4/mc-1-big-Filter.db b/test/data/negative-local-expiration-test/table4/mc-1-big-Filter.db new file mode 100644 index 0000000000..a397f351d9 Binary files /dev/null and b/test/data/negative-local-expiration-test/table4/mc-1-big-Filter.db differ diff --git a/test/data/negative-local-expiration-test/table4/mc-1-big-Index.db b/test/data/negative-local-expiration-test/table4/mc-1-big-Index.db new file mode 100644 index 0000000000..6397b5e0a9 Binary files /dev/null and b/test/data/negative-local-expiration-test/table4/mc-1-big-Index.db differ diff --git a/test/data/negative-local-expiration-test/table4/mc-1-big-Statistics.db b/test/data/negative-local-expiration-test/table4/mc-1-big-Statistics.db new file mode 100644 index 0000000000..4ee9294b5a Binary files /dev/null and b/test/data/negative-local-expiration-test/table4/mc-1-big-Statistics.db differ diff --git a/test/data/negative-local-expiration-test/table4/mc-1-big-Summary.db b/test/data/negative-local-expiration-test/table4/mc-1-big-Summary.db new file mode 100644 index 0000000000..66cf70f7c7 Binary files /dev/null and b/test/data/negative-local-expiration-test/table4/mc-1-big-Summary.db differ diff --git a/test/data/negative-local-expiration-test/table4/mc-1-big-TOC.txt b/test/data/negative-local-expiration-test/table4/mc-1-big-TOC.txt new file mode 100644 index 0000000000..f4455377aa --- /dev/null +++ b/test/data/negative-local-expiration-test/table4/mc-1-big-TOC.txt @@ -0,0 +1,8 @@ +Summary.db +TOC.txt +Filter.db +Index.db +Digest.crc32 +CompressionInfo.db +Data.db +Statistics.db diff --git a/test/unit/org/apache/cassandra/cql3/validation/operations/TTLTest.java b/test/unit/org/apache/cassandra/cql3/validation/operations/TTLTest.java index 9f375d43bf..fc709748f5 100644 --- a/test/unit/org/apache/cassandra/cql3/validation/operations/TTLTest.java +++ b/test/unit/org/apache/cassandra/cql3/validation/operations/TTLTest.java @@ -1,28 +1,47 @@ package org.apache.cassandra.cql3.validation.operations; +import java.io.File; +import java.io.FileInputStream; +import java.io.FileOutputStream; +import java.io.IOException; + +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import org.apache.cassandra.cql3.Attributes; import org.apache.cassandra.cql3.CQLTester; +import org.apache.cassandra.cql3.UntypedResultSet; +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.ExpirationDateOverflowHandling; +import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.db.rows.AbstractCell; import org.apache.cassandra.exceptions.InvalidRequestException; +import org.apache.cassandra.utils.FBUtilities; + import org.junit.Test; public class TTLTest extends CQLTester { + public static String NEGATIVE_LOCAL_EXPIRATION_TEST_DIR = "test/data/negative-local-expiration-test/%s"; + + public static int MAX_TTL = Attributes.MAX_TTL; + + public static final String SIMPLE_NOCLUSTERING = "table1"; + public static final String SIMPLE_CLUSTERING = "table2"; + public static final String COMPLEX_NOCLUSTERING = "table3"; + public static final String COMPLEX_CLUSTERING = "table4"; @Test public void testTTLPerRequestLimit() throws Throwable { createTable("CREATE TABLE %s (k int PRIMARY KEY, i int)"); - // insert - execute("INSERT INTO %s (k, i) VALUES (1, 1) USING TTL ?", Attributes.MAX_TTL); // max ttl - int ttl = execute("SELECT ttl(i) FROM %s").one().getInt("ttl(i)"); - assertTrue(ttl > Attributes.MAX_TTL - 10); + // insert with low TTL should not be denied + execute("INSERT INTO %s (k, i) VALUES (1, 1) USING TTL ?", 10); try { - execute("INSERT INTO %s (k, i) VALUES (1, 1) USING TTL ?", Attributes.MAX_TTL + 1); + execute("INSERT INTO %s (k, i) VALUES (1, 1) USING TTL ?", MAX_TTL + 1); fail("Expect InvalidRequestException"); } catch (InvalidRequestException e) @@ -37,18 +56,16 @@ public class TTLTest extends CQLTester } catch (InvalidRequestException e) { - assertTrue(e.getMessage().contains("A TTL must be greater or equal to 0, but was -1")); + assertTrue(e.getMessage().contains("A TTL must be greater or equal to 0")); } execute("TRUNCATE %s"); - // update - execute("UPDATE %s USING TTL ? SET i = 1 WHERE k = 2", Attributes.MAX_TTL); // max ttl - ttl = execute("SELECT ttl(i) FROM %s").one().getInt("ttl(i)"); - assertTrue(ttl > Attributes.MAX_TTL - 10); + // insert with low TTL should not be denied + execute("UPDATE %s USING TTL ? SET i = 1 WHERE k = 2", 5); try { - execute("UPDATE %s USING TTL ? SET i = 1 WHERE k = 2", Attributes.MAX_TTL + 1); + execute("UPDATE %s USING TTL ? SET i = 1 WHERE k = 2", MAX_TTL + 1); fail("Expect InvalidRequestException"); } catch (InvalidRequestException e) @@ -63,7 +80,7 @@ public class TTLTest extends CQLTester } catch (InvalidRequestException e) { - assertTrue(e.getMessage().contains("A TTL must be greater or equal to 0, but was -1")); + assertTrue(e.getMessage().contains("A TTL must be greater or equal to 0")); } } @@ -84,21 +101,295 @@ public class TTLTest extends CQLTester try { createTable("CREATE TABLE %s (k int PRIMARY KEY, i int) WITH default_time_to_live=" - + (Attributes.MAX_TTL + 1)); + + (MAX_TTL + 1)); fail("Expect Invalid schema"); } catch (RuntimeException e) { assertTrue(e.getCause() .getMessage() - .contains("default_time_to_live must be less than or equal to " + Attributes.MAX_TTL + " (got " - + (Attributes.MAX_TTL + 1) + ")")); + .contains("default_time_to_live must be less than or equal to " + MAX_TTL + " (got " + + (MAX_TTL + 1) + ")")); } - createTable("CREATE TABLE %s (k int PRIMARY KEY, i int) WITH default_time_to_live=" + Attributes.MAX_TTL); + // table with default low TTL should not be denied + createTable("CREATE TABLE %s (k int PRIMARY KEY, i int) WITH default_time_to_live=" + 5); execute("INSERT INTO %s (k, i) VALUES (1, 1)"); - int ttl = execute("SELECT ttl(i) FROM %s").one().getInt("ttl(i)"); - assertTrue(ttl > 10000 - 10); // within 10 second } + @Test + public void testCapWarnExpirationOverflowPolicy() throws Throwable + { + // We don't test that the actual warn is logged here, only on dtest + testCapExpirationDateOverflowPolicy(ExpirationDateOverflowHandling.ExpirationDateOverflowPolicy.CAP); + } + + @Test + public void testCapNoWarnExpirationOverflowPolicy() throws Throwable + { + testCapExpirationDateOverflowPolicy(ExpirationDateOverflowHandling.ExpirationDateOverflowPolicy.CAP_NOWARN); + } + + @Test + public void testCapNoWarnExpirationOverflowPolicyDefaultTTL() throws Throwable + { + ExpirationDateOverflowHandling.policy = ExpirationDateOverflowHandling.policy.CAP_NOWARN; + createTable("CREATE TABLE %s (k int PRIMARY KEY, i int) WITH default_time_to_live=" + MAX_TTL); + execute("INSERT INTO %s (k, i) VALUES (1, 1)"); + checkTTLIsCapped("i"); + ExpirationDateOverflowHandling.policy = ExpirationDateOverflowHandling.policy.REJECT; + } + + @Test + public void testRejectExpirationOverflowPolicy() throws Throwable + { + //ExpirationDateOverflowHandling.expirationDateOverflowPolicy = ExpirationDateOverflowHandling.expirationDateOverflowPolicy.REJECT; + createTable("CREATE TABLE %s (k int PRIMARY KEY, i int)"); + try + { + execute("INSERT INTO %s (k, i) VALUES (1, 1) USING TTL " + MAX_TTL); + } + catch (InvalidRequestException e) + { + assertTrue(e.getMessage().contains("exceeds maximum supported expiration date")); + } + try + { + createTable("CREATE TABLE %s (k int PRIMARY KEY, i int) WITH default_time_to_live=" + MAX_TTL); + execute("INSERT INTO %s (k, i) VALUES (1, 1)"); + } + catch (InvalidRequestException e) + { + assertTrue(e.getMessage().contains("exceeds maximum supported expiration date")); + } + } + + @Test + public void testRecoverOverflowedExpirationWithScrub() throws Throwable + { + baseTestRecoverOverflowedExpiration(false, false); + baseTestRecoverOverflowedExpiration(true, false); + baseTestRecoverOverflowedExpiration(true, true); + } + + public void testCapExpirationDateOverflowPolicy(ExpirationDateOverflowHandling.ExpirationDateOverflowPolicy policy) throws Throwable + { + ExpirationDateOverflowHandling.policy = policy; + + // simple column, clustering, flush + testCapExpirationDateOverflowPolicy(true, true, true); + // simple column, clustering, noflush + testCapExpirationDateOverflowPolicy(true, true, false); + // simple column, noclustering, flush + testCapExpirationDateOverflowPolicy(true, false, true); + // simple column, noclustering, noflush + testCapExpirationDateOverflowPolicy(true, false, false); + // complex column, clustering, flush + testCapExpirationDateOverflowPolicy(false, true, true); + // complex column, clustering, noflush + testCapExpirationDateOverflowPolicy(false, true, false); + // complex column, noclustering, flush + testCapExpirationDateOverflowPolicy(false, false, true); + // complex column, noclustering, noflush + testCapExpirationDateOverflowPolicy(false, false, false); + + // Return to previous policy + ExpirationDateOverflowHandling.policy = ExpirationDateOverflowHandling.ExpirationDateOverflowPolicy.REJECT; + } + + public void testCapExpirationDateOverflowPolicy(boolean simple, boolean clustering, boolean flush) throws Throwable + { + // Create Table + createTable(simple, clustering); + + // Insert data with INSERT and UPDATE + if (simple) + { + execute("INSERT INTO %s (k, a) VALUES (?, ?) USING TTL " + MAX_TTL, 2, 2); + if (clustering) + execute("UPDATE %s USING TTL " + MAX_TTL + " SET b = 1 WHERE k = 1 AND a = 1;"); + else + execute("UPDATE %s USING TTL " + MAX_TTL + " SET a = 1, b = 1 WHERE k = 1;"); + } + else + { + execute("INSERT INTO %s (k, a, b) VALUES (?, ?, ?) USING TTL " + MAX_TTL, 2, 2, set("v21", "v22", "v23", "v24")); + if (clustering) + execute("UPDATE %s USING TTL " + MAX_TTL + " SET b = ? WHERE k = 1 AND a = 1;", set("v11", "v12", "v13", "v14")); + else + execute("UPDATE %s USING TTL " + MAX_TTL + " SET a = 1, b = ? WHERE k = 1;", set("v11", "v12", "v13", "v14")); + } + + // Maybe Flush + Keyspace ks = Keyspace.open(keyspace()); + if (flush) + FBUtilities.waitOnFutures(ks.flush()); + + // Verify data + verifyData(simple); + + // Maybe major compact + if (flush) + { + // Major compact and check data is still present + ks.getColumnFamilyStore(currentTable()).forceMajorCompaction(); + + // Verify data again + verifyData(simple); + } + } + + public void baseTestRecoverOverflowedExpiration(boolean runScrub, boolean reinsertOverflowedTTL) throws Throwable + { + // simple column, clustering + testRecoverOverflowedExpirationWithScrub(true, true, runScrub, reinsertOverflowedTTL); + // simple column, noclustering + testRecoverOverflowedExpirationWithScrub(true, false, runScrub, reinsertOverflowedTTL); + // complex column, clustering + testRecoverOverflowedExpirationWithScrub(false, true, runScrub, reinsertOverflowedTTL); + // complex column, noclustering + testRecoverOverflowedExpirationWithScrub(false, false, runScrub, reinsertOverflowedTTL); + } + + private void createTable(boolean simple, boolean clustering) + { + if (simple) + { + if (clustering) + createTable("create table %s (k int, a int, b int, primary key(k, a))"); + else + createTable("create table %s (k int primary key, a int, b int)"); + } + else + { + if (clustering) + createTable("create table %s (k int, a int, b set, primary key(k, a))"); + else + createTable("create table %s (k int primary key, a int, b set)"); + } + } + + private void verifyData(boolean simple) throws Throwable + { + if (simple) + { + assertRows(execute("SELECT * from %s"), row(1, 1, 1), row(2, 2, null)); + } + else + { + assertRows(execute("SELECT * from %s"), row(1, 1, set("v11", "v12", "v13", "v14")), row(2, 2, set("v21", "v22", "v23", "v24"))); + } + // Cannot retrieve TTL from collections + if (simple) + checkTTLIsCapped("b"); + } + + /** + * Verify that the computed TTL is equal to the maximum allowed ttl given the + * {@link AbstractCell#localDeletionTime()} field limitation (CASSANDRA-14092) + */ + private void checkTTLIsCapped(String field) throws Throwable + { + + // TTL is computed dynamically from row expiration time, so if it is + // equal or higher to the minimum max TTL we compute before the query + // we are fine. + int minMaxTTL = computeMaxTTL(); + UntypedResultSet execute = execute("SELECT ttl(" + field + ") FROM %s WHERE k = 1"); + for (UntypedResultSet.Row row : execute) + { + int ttl = row.getInt("ttl(" + field + ")"); + assertTrue(ttl >= minMaxTTL); + } + } + + /** + * The max TTL is computed such that the TTL summed with the current time is equal to the maximum + * allowed expiration time {@link org.apache.cassandra.db.rows.Cell#MAX_DELETION_TIME} (2038-01-19T03:14:06+00:00) + */ + private int computeMaxTTL() + { + int nowInSecs = (int) (System.currentTimeMillis() / 1000); + return AbstractCell.MAX_DELETION_TIME - nowInSecs; + } + + public void testRecoverOverflowedExpirationWithScrub(boolean simple, boolean clustering, boolean runScrub, boolean reinsertOverflowedTTL) throws Throwable + { + if (reinsertOverflowedTTL) + { + assert runScrub; + } + + createTable(simple, clustering); + + Keyspace keyspace = Keyspace.open(KEYSPACE); + ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(currentTable()); + + assertEquals(0, cfs.getLiveSSTables().size()); + + copySSTablesToTableDir(currentTable(), simple, clustering); + + cfs.loadNewSSTables(); + + if (runScrub) + { + cfs.scrub(true, false, true, reinsertOverflowedTTL, 1); + } + + if (reinsertOverflowedTTL) + { + if (simple) + assertRows(execute("SELECT * from %s"), row(1, 1, 1), row(2, 2, null)); + else + assertRows(execute("SELECT * from %s"), row(1, 1, set("v11", "v12", "v13", "v14")), row(2, 2, set("v21", "v22", "v23", "v24"))); + + cfs.forceMajorCompaction(); + + if (simple) + assertRows(execute("SELECT * from %s"), row(1, 1, 1), row(2, 2, null)); + else + assertRows(execute("SELECT * from %s"), row(1, 1, set("v11", "v12", "v13", "v14")), row(2, 2, set("v21", "v22", "v23", "v24"))); + } + else + { + assertEmpty(execute("SELECT * from %s")); + } + } + + private void copySSTablesToTableDir(String table, boolean simple, boolean clustering) throws IOException + { + File destDir = Keyspace.open(keyspace()).getColumnFamilyStore(table).getDirectories().getCFDirectories().iterator().next(); + File sourceDir = getTableDir(table, simple, clustering); + for (File file : sourceDir.listFiles()) + { + copyFile(file, destDir); + } + } + + private static File getTableDir(String table, boolean simple, boolean clustering) + { + return new File(String.format(NEGATIVE_LOCAL_EXPIRATION_TEST_DIR, getTableName(simple, clustering))); + } + + private static void copyFile(File src, File dest) throws IOException + { + byte[] buf = new byte[65536]; + if (src.isFile()) + { + File target = new File(dest, src.getName()); + int rd; + FileInputStream is = new FileInputStream(src); + FileOutputStream os = new FileOutputStream(target); + while ((rd = is.read(buf)) >= 0) + os.write(buf, 0, rd); + } + } + + public static String getTableName(boolean simple, boolean clustering) + { + if (simple) + return clustering ? SIMPLE_CLUSTERING : SIMPLE_NOCLUSTERING; + else + return clustering ? COMPLEX_CLUSTERING : COMPLEX_NOCLUSTERING; + } } diff --git a/test/unit/org/apache/cassandra/db/ScrubTest.java b/test/unit/org/apache/cassandra/db/ScrubTest.java index 08336a16ad..fc2faeaa4f 100644 --- a/test/unit/org/apache/cassandra/db/ScrubTest.java +++ b/test/unit/org/apache/cassandra/db/ScrubTest.java @@ -117,7 +117,7 @@ public class ScrubTest fillCF(cfs, 1); assertOrderedAll(cfs, 1); - CompactionManager.instance.performScrub(cfs, false, true, 2); + CompactionManager.instance.performScrub(cfs, false, true, false, 2); // check data is still there assertOrderedAll(cfs, 1); @@ -619,7 +619,7 @@ public class ScrubTest { //make sure the next scrub fails overrideWithGarbage(indexCfs.getLiveSSTables().iterator().next(), ByteBufferUtil.bytes(1L), ByteBufferUtil.bytes(2L)); } - CompactionManager.AllSSTableOpStatus result = indexCfs.scrub(false, false, true, true, 0); + CompactionManager.AllSSTableOpStatus result = indexCfs.scrub(false, false, false, true, false,0); assertEquals(failure ? CompactionManager.AllSSTableOpStatus.ABORTED : CompactionManager.AllSSTableOpStatus.SUCCESSFUL, @@ -698,7 +698,7 @@ public class ScrubTest cfs.loadNewSSTables(); - cfs.scrub(true, true, true, 1); + cfs.scrub(true, true, false, false, false, 1); UntypedResultSet rs = QueryProcessor.executeInternal(String.format("SELECT * FROM \"%s\".cf_with_duplicates_3_0", KEYSPACE)); assertEquals(1, rs.size());