mirror of https://github.com/apache/cassandra
Merge branch 'cassandra-3.0' into cassandra-3.11
This commit is contained in:
Paulo Motta
commit
0a6b6f506b
|
|
@ -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=<cloned configuration directory>.
|
||||
- Execute "sstablescrub --reinsert-overflowed-ttl <keyspace> <table>".
|
||||
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 <keyspace> <table>".
|
||||
- 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.
|
||||
|
|
@ -50,6 +50,7 @@ Merged from 2.2:
|
|||
* Rely on the JVM to handle OutOfMemoryErrors (CASSANDRA-13006)
|
||||
* Grab refs during scrub/index redistribution/cleanup (CASSANDRA-13873)
|
||||
Merged from 2.1:
|
||||
* Protect against overflow of local expiration time (CASSANDRA-14092)
|
||||
* RPM package spec: fix permissions for installed jars and config files (CASSANDRA-14181)
|
||||
* More PEP8 compiance for cqlsh (CASSANDRA-14021)
|
||||
|
||||
|
|
|
|||
21
NEWS.txt
21
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
|
||||
|
|
|
|||
|
|
@ -183,6 +183,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 #
|
||||
#################
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -20,7 +20,9 @@ package org.apache.cassandra.cql3;
|
|||
import java.nio.ByteBuffer;
|
||||
import java.util.List;
|
||||
|
||||
import org.apache.cassandra.config.CFMetaData;
|
||||
import org.apache.cassandra.cql3.functions.Function;
|
||||
import org.apache.cassandra.db.ExpirationDateOverflowHandling;
|
||||
import org.apache.cassandra.db.LivenessInfo;
|
||||
import org.apache.cassandra.db.marshal.Int32Type;
|
||||
import org.apache.cassandra.db.marshal.LongType;
|
||||
|
|
@ -98,17 +100,20 @@ public class Attributes
|
|||
return LongType.instance.compose(tval);
|
||||
}
|
||||
|
||||
public int getTimeToLive(QueryOptions options, int defaultTimeToLive) throws InvalidRequestException
|
||||
public int getTimeToLive(QueryOptions options, CFMetaData metadata) throws InvalidRequestException
|
||||
{
|
||||
if (timeToLive == null)
|
||||
return defaultTimeToLive;
|
||||
{
|
||||
ExpirationDateOverflowHandling.maybeApplyExpirationDateOverflowPolicy(metadata, metadata.params.defaultTimeToLive, true);
|
||||
return metadata.params.defaultTimeToLive;
|
||||
}
|
||||
|
||||
ByteBuffer tval = timeToLive.bindAndGet(options);
|
||||
if (tval == null)
|
||||
return 0;
|
||||
|
||||
if (tval == ByteBufferUtil.UNSET_BYTE_BUFFER)
|
||||
return defaultTimeToLive;
|
||||
return metadata.params.defaultTimeToLive;
|
||||
|
||||
try
|
||||
{
|
||||
|
|
@ -126,9 +131,11 @@ public class Attributes
|
|||
if (ttl > MAX_TTL)
|
||||
throw new InvalidRequestException(String.format("ttl is too large. requested (%d) maximum (%d)", ttl, MAX_TTL));
|
||||
|
||||
if (defaultTimeToLive != LivenessInfo.NO_TTL && ttl == LivenessInfo.NO_TTL)
|
||||
if (metadata.params.defaultTimeToLive != LivenessInfo.NO_TTL && ttl == LivenessInfo.NO_TTL)
|
||||
return LivenessInfo.NO_TTL;
|
||||
|
||||
ExpirationDateOverflowHandling.maybeApplyExpirationDateOverflowPolicy(metadata, ttl, false);
|
||||
|
||||
return ttl;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -203,7 +203,7 @@ public abstract class ModificationStatement implements CQLStatement
|
|||
|
||||
public int getTimeToLive(QueryOptions options) throws InvalidRequestException
|
||||
{
|
||||
return attrs.getTimeToLive(options, cfm.params.defaultTimeToLive);
|
||||
return attrs.getTimeToLive(options, cfm);
|
||||
}
|
||||
|
||||
public void checkAccess(ClientState state) throws InvalidRequestException, UnauthorizedException
|
||||
|
|
|
|||
|
|
@ -1510,13 +1510,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)
|
||||
|
|
@ -1524,7 +1524,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)
|
||||
{
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
}
|
||||
|
|
@ -1543,7 +1543,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)
|
||||
|
|
|
|||
|
|
@ -20,6 +20,8 @@ package org.apache.cassandra.db;
|
|||
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;
|
||||
|
||||
|
|
@ -39,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.
|
||||
*
|
||||
|
|
@ -47,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);
|
||||
|
||||
|
|
@ -66,7 +68,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(long timestamp, int ttl, int nowInSec)
|
||||
|
|
@ -227,6 +229,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()
|
||||
{
|
||||
|
|
|
|||
|
|
@ -367,7 +367,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()
|
||||
|
|
@ -381,7 +389,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);
|
||||
}
|
||||
|
|
@ -969,11 +977,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);
|
||||
|
|
|
|||
|
|
@ -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,20 +67,28 @@ public class Scrubber implements Closeable
|
|||
long currentRowPositionFromIndex;
|
||||
long nextRowPositionFromIndex;
|
||||
|
||||
private NegativeLocalDeletionInfoMetrics negativeLocalDeletionInfoMetrics = new NegativeLocalDeletionInfoMetrics();
|
||||
|
||||
private final OutputHandler outputHandler;
|
||||
|
||||
private static final Comparator<Partition> partitionComparator = new Comparator<Partition>()
|
||||
{
|
||||
public int compare(Partition r1, Partition r2)
|
||||
{
|
||||
return r1.partitionKey().compareTo(r2.partitionKey());
|
||||
}
|
||||
public int compare(Partition r1, Partition r2)
|
||||
{
|
||||
return r1.partitionKey().compareTo(r2.partitionKey());
|
||||
}
|
||||
};
|
||||
private final SortedSet<Partition> outOfOrder = new TreeSet<>(partitionComparator);
|
||||
|
||||
public Scrubber(ColumnFamilyStore cfs, LifecycleTransaction transaction, boolean skipCorrupted, boolean checkData)
|
||||
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,13 +96,15 @@ public class Scrubber implements Closeable
|
|||
LifecycleTransaction transaction,
|
||||
boolean skipCorrupted,
|
||||
OutputHandler outputHandler,
|
||||
boolean checkData)
|
||||
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);
|
||||
|
|
@ -112,8 +124,8 @@ public class Scrubber implements Closeable
|
|||
}
|
||||
this.checkData = checkData && !this.isIndex; //LocalByPartitionerType does not support validation
|
||||
this.expectedBloomFilterSize = Math.max(
|
||||
cfs.metadata.params.minIndexInterval,
|
||||
hasIndexFile ? SSTableReader.getApproximateKeyCount(toScrub) : 0);
|
||||
cfs.metadata.params.minIndexInterval,
|
||||
hasIndexFile ? SSTableReader.getApproximateKeyCount(toScrub) : 0);
|
||||
|
||||
// loop through each row, deserializing to check for damage.
|
||||
// we'll also loop through the index at the same time, using the position from the index to recover if the
|
||||
|
|
@ -124,13 +136,16 @@ public class Scrubber implements Closeable
|
|||
: sstable.openDataReader(CompactionManager.instance.getRateLimiter());
|
||||
|
||||
this.indexFile = hasIndexFile
|
||||
? RandomAccessReader.open(new File(sstable.descriptor.filenameFor(Component.PRIMARY_INDEX)))
|
||||
: null;
|
||||
? RandomAccessReader.open(new File(sstable.descriptor.filenameFor(Component.PRIMARY_INDEX)))
|
||||
: null;
|
||||
|
||||
this.scrubInfo = new ScrubInfo(dataFile, sstable);
|
||||
|
||||
this.currentRowPositionFromIndex = 0;
|
||||
this.nextRowPositionFromIndex = 0;
|
||||
|
||||
if (reinsertOverflowedTTLRows)
|
||||
outputHandler.output("Starting scrub with reinsert overflowed TTL option");
|
||||
}
|
||||
|
||||
private UnfilteredRowIterator withValidation(UnfilteredRowIterator iter, String filename)
|
||||
|
|
@ -203,8 +218,8 @@ public class Scrubber implements Closeable
|
|||
if (currentIndexKey != null && !key.getKey().equals(currentIndexKey))
|
||||
{
|
||||
throw new IOError(new IOException(String.format("Key from data file (%s) does not match key from index file (%s)",
|
||||
//ByteBufferUtil.bytesToHex(key.getKey()), ByteBufferUtil.bytesToHex(currentIndexKey))));
|
||||
"_too big_", ByteBufferUtil.bytesToHex(currentIndexKey))));
|
||||
//ByteBufferUtil.bytesToHex(key.getKey()), ByteBufferUtil.bytesToHex(currentIndexKey))));
|
||||
"_too big_", ByteBufferUtil.bytesToHex(currentIndexKey))));
|
||||
}
|
||||
|
||||
if (indexFile != null && dataSizeFromIndex > dataFile.length())
|
||||
|
|
@ -225,7 +240,7 @@ public class Scrubber implements Closeable
|
|||
&& (key == null || !key.getKey().equals(currentIndexKey) || dataStart != dataStartFromIndex))
|
||||
{
|
||||
outputHandler.output(String.format("Retrying from row index; data is %s bytes starting at %s",
|
||||
dataSizeFromIndex, dataStartFromIndex));
|
||||
dataSizeFromIndex, dataStartFromIndex));
|
||||
key = sstable.decorateKey(currentIndexKey);
|
||||
try
|
||||
{
|
||||
|
|
@ -287,18 +302,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")
|
||||
|
|
@ -307,7 +324,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(SSTableIdentityIterator.create(sstable, dataFile, key)),
|
||||
OrderCheckerIterator sstableIterator = new OrderCheckerIterator(getIterator(key),
|
||||
cfs.metadata.comparator);
|
||||
|
||||
try (UnfilteredRowIterator iterator = withValidation(sstableIterator, dataFile.getPath()))
|
||||
|
|
@ -333,6 +350,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(SSTableIdentityIterator.create(sstable, dataFile, key));
|
||||
return reinsertOverflowedTTLRows ? new FixNegativeLocalDeletionTimeIterator(rowMergingIterator,
|
||||
outputHandler,
|
||||
negativeLocalDeletionInfoMetrics) : rowMergingIterator;
|
||||
}
|
||||
|
||||
private void updateIndexKey()
|
||||
{
|
||||
currentIndexKey = nextIndexKey;
|
||||
|
|
@ -342,8 +371,8 @@ public class Scrubber implements Closeable
|
|||
nextIndexKey = !indexAvailable() ? null : ByteBufferUtil.readWithShortLength(indexFile);
|
||||
|
||||
nextRowPositionFromIndex = !indexAvailable()
|
||||
? dataFile.length()
|
||||
: rowIndexEntrySerializer.deserializePositionAndSkip(indexFile);
|
||||
? dataFile.length()
|
||||
: rowIndexEntrySerializer.deserializePositionAndSkip(indexFile);
|
||||
}
|
||||
catch (Throwable th)
|
||||
{
|
||||
|
|
@ -474,6 +503,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.
|
||||
|
|
@ -620,6 +654,153 @@ 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<Unfiltered> 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();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -20,6 +20,7 @@ package org.apache.cassandra.db.rows;
|
|||
import java.nio.ByteBuffer;
|
||||
|
||||
import org.apache.cassandra.config.ColumnDefinition;
|
||||
import org.apache.cassandra.db.ExpirationDateOverflowHandling;
|
||||
import org.apache.cassandra.db.marshal.ByteType;
|
||||
import org.apache.cassandra.utils.ByteBufferUtil;
|
||||
import org.apache.cassandra.utils.ObjectSizes;
|
||||
|
|
@ -66,7 +67,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)
|
||||
|
|
@ -114,6 +115,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())
|
||||
|
|
|
|||
|
|
@ -21,6 +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.config.*;
|
||||
import org.apache.cassandra.db.*;
|
||||
import org.apache.cassandra.io.util.DataOutputPlus;
|
||||
|
|
@ -41,6 +48,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<Cell> comparator = (c1, c2) ->
|
||||
{
|
||||
|
|
@ -134,6 +142,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
|
||||
|
|
|
|||
|
|
@ -143,6 +143,11 @@ public class NativeCell extends AbstractCell
|
|||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
public Cell withUpdatedTimestampAndLocalDeletionTime(long newTimestamp, int newLocalDeletionTime)
|
||||
{
|
||||
return new BufferCell(column, newTimestamp, ttl(), newLocalDeletionTime, value(), path());
|
||||
}
|
||||
|
||||
public Cell withUpdatedColumn(ColumnDefinition column)
|
||||
{
|
||||
return new BufferCell(column, timestamp(), ttl(), localDeletionTime(), value(), path());
|
||||
|
|
|
|||
|
|
@ -2995,11 +2995,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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -276,8 +276,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.
|
||||
|
|
|
|||
|
|
@ -333,7 +333,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);
|
||||
}
|
||||
|
|
@ -368,7 +368,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())
|
||||
{
|
||||
|
|
@ -377,9 +377,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;
|
||||
}
|
||||
|
|
@ -453,7 +455,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())
|
||||
|
|
|
|||
|
|
@ -255,9 +255,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
|
||||
|
|
@ -299,10 +299,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;
|
||||
|
|
|
|||
|
|
@ -50,6 +50,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 +123,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();
|
||||
}
|
||||
|
|
@ -200,6 +201,7 @@ public class StandaloneScrubber
|
|||
public boolean manifestCheckOnly;
|
||||
public boolean skipCorrupted;
|
||||
public boolean noValidate;
|
||||
public boolean reinserOverflowedTTL;
|
||||
|
||||
private Options(String keyspaceName, String cfName)
|
||||
{
|
||||
|
|
@ -240,6 +242,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;
|
||||
}
|
||||
|
|
@ -266,6 +269,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 found rows with overflowed TTL affected by CASSANDRA-14092");
|
||||
return options;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -48,6 +48,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 = "Reinsert found rows with overflowed TTL affected by CASSANDRA-14092")
|
||||
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 +68,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)
|
||||
{
|
||||
|
|
|
|||
Binary file not shown.
Binary file not shown.
|
|
@ -0,0 +1 @@
|
|||
4223695539
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
|
|
@ -0,0 +1,8 @@
|
|||
Digest.crc32
|
||||
CompressionInfo.db
|
||||
Index.db
|
||||
TOC.txt
|
||||
Data.db
|
||||
Statistics.db
|
||||
Filter.db
|
||||
Summary.db
|
||||
Binary file not shown.
Binary file not shown.
|
|
@ -0,0 +1 @@
|
|||
2886964045
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
|
|
@ -0,0 +1,8 @@
|
|||
Digest.crc32
|
||||
CompressionInfo.db
|
||||
Index.db
|
||||
TOC.txt
|
||||
Data.db
|
||||
Statistics.db
|
||||
Filter.db
|
||||
Summary.db
|
||||
Binary file not shown.
Binary file not shown.
|
|
@ -0,0 +1 @@
|
|||
3254141434
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
|
|
@ -0,0 +1,8 @@
|
|||
Digest.crc32
|
||||
CompressionInfo.db
|
||||
Index.db
|
||||
TOC.txt
|
||||
Data.db
|
||||
Statistics.db
|
||||
Filter.db
|
||||
Summary.db
|
||||
Binary file not shown.
Binary file not shown.
|
|
@ -0,0 +1 @@
|
|||
3231150985
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
|
|
@ -0,0 +1,8 @@
|
|||
Digest.crc32
|
||||
CompressionInfo.db
|
||||
Index.db
|
||||
TOC.txt
|
||||
Data.db
|
||||
Statistics.db
|
||||
Filter.db
|
||||
Summary.db
|
||||
|
|
@ -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<text>, primary key(k, a))");
|
||||
else
|
||||
createTable("create table %s (k int primary key, a int, b set<text>)");
|
||||
}
|
||||
}
|
||||
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -6,9 +6,9 @@
|
|||
* 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
|
||||
|
|
@ -152,9 +152,9 @@ public class CellTest
|
|||
|
||||
// Invalid ttl
|
||||
assertInvalid(BufferCell.expiring(c, 0, -4, 4, bbs(4)));
|
||||
// Invalid local deletion times
|
||||
assertInvalid(BufferCell.expiring(c, 0, 4, -5, bbs(4)));
|
||||
assertInvalid(BufferCell.expiring(c, 0, 4, Cell.NO_DELETION_TIME, bbs(4)));
|
||||
// Cells with overflowed localExpirationTime are valid after CASSANDRA-14092
|
||||
assertValid(BufferCell.expiring(c, 0, 4, -5, bbs(4)));
|
||||
assertValid(BufferCell.expiring(c, 0, 4, Cell.NO_DELETION_TIME, bbs(4)));
|
||||
|
||||
c = fakeColumn("c", MapType.getInstance(Int32Type.instance, Int32Type.instance, true));
|
||||
// Valid cell path
|
||||
|
|
@ -188,9 +188,9 @@ public class CellTest
|
|||
|
||||
// Invalid ttl
|
||||
assertInvalid(BufferCell.expiring(c, 0, -4, 4, bb(1), CellPath.create(bbs(0))));
|
||||
// Invalid local deletion times
|
||||
assertInvalid(BufferCell.expiring(c, 0, 4, -5, bb(1), CellPath.create(bbs(0))));
|
||||
assertInvalid(BufferCell.expiring(c, 0, 4, Cell.NO_DELETION_TIME, bb(1), CellPath.create(bbs(0))));
|
||||
// Cells with overflowed localExpirationTime are valid after CASSANDRA-14092
|
||||
assertValid(BufferCell.expiring(c, 0, 4, -5, bb(1), CellPath.create(bbs(0))));
|
||||
assertValid((BufferCell.expiring(c, 0, 4, Cell.NO_DELETION_TIME, bb(1), CellPath.create(bbs(0)))));
|
||||
|
||||
// Invalid cell path (int values should be 0 or 2 bytes)
|
||||
assertInvalid(BufferCell.live(c, 0, bb(1), CellPath.create(ByteBufferUtil.bytes((long)4))));
|
||||
|
|
@ -227,9 +227,9 @@ public class CellTest
|
|||
|
||||
// Invalid ttl
|
||||
assertInvalid(BufferCell.expiring(c, 0, -4, 4, val));
|
||||
// Invalid local deletion times
|
||||
assertInvalid(BufferCell.expiring(c, 0, 4, -5, val));
|
||||
assertInvalid(BufferCell.expiring(c, 0, 4, Cell.NO_DELETION_TIME, val));
|
||||
// Cells with overflowed localExpirationTime are valid after CASSANDRA-14092
|
||||
assertValid(BufferCell.expiring(c, 0, 4, -5, val));
|
||||
assertValid(BufferCell.expiring(c, 0, 4, Cell.NO_DELETION_TIME, val));
|
||||
}
|
||||
|
||||
@Test
|
||||
|
|
|
|||
|
|
@ -118,7 +118,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);
|
||||
|
|
@ -622,7 +622,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,
|
||||
|
|
@ -701,7 +701,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());
|
||||
|
|
|
|||
Loading…
Reference in New Issue