SortedTableScrubber.java
/*
* 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.io.sstable.format;
import java.io.IOError;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import javax.annotation.concurrent.NotThreadSafe;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.db.ClusteringComparator;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.LivenessInfo;
import org.apache.cassandra.db.compaction.CompactionInfo;
import org.apache.cassandra.db.compaction.CompactionManager;
import org.apache.cassandra.db.compaction.OperationType;
import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
import org.apache.cassandra.db.partitions.ImmutableBTreePartition;
import org.apache.cassandra.db.partitions.Partition;
import org.apache.cassandra.db.rows.AbstractCell;
import org.apache.cassandra.db.rows.Cell;
import org.apache.cassandra.db.rows.ColumnData;
import org.apache.cassandra.db.rows.ComplexColumnData;
import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.db.rows.Rows;
import org.apache.cassandra.db.rows.Unfiltered;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.db.rows.UnfilteredRowIterators;
import org.apache.cassandra.db.rows.WrappingUnfilteredRowIterator;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.io.sstable.Component;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.sstable.IScrubber;
import org.apache.cassandra.io.sstable.SSTableIdentityIterator;
import org.apache.cassandra.io.sstable.SSTableRewriter;
import org.apache.cassandra.io.sstable.format.SSTableFormat.Components;
import org.apache.cassandra.io.sstable.metadata.StatsMetadata;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.io.util.RandomAccessReader;
import org.apache.cassandra.service.ActiveRepairService;
import org.apache.cassandra.utils.AbstractIterator;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.OutputHandler;
import org.apache.cassandra.utils.TimeUUID;
import org.apache.cassandra.utils.concurrent.Refs;
import org.apache.cassandra.utils.memory.HeapCloner;
import static org.apache.cassandra.utils.TimeUUID.Generator.nextTimeUUID;
@NotThreadSafe
public abstract class SortedTableScrubber<R extends SSTableReaderWithFilter> implements IScrubber
{
private final static Logger logger = LoggerFactory.getLogger(SortedTableScrubber.class);
protected final ColumnFamilyStore cfs;
protected final LifecycleTransaction transaction;
protected final File destination;
protected final IScrubber.Options options;
protected final R sstable;
protected final OutputHandler outputHandler;
protected final boolean isCommutative;
protected final long expectedBloomFilterSize;
protected final ReadWriteLock fileAccessLock = new ReentrantReadWriteLock();
protected final RandomAccessReader dataFile;
protected final ScrubInfo scrubInfo;
protected final NegativeLocalDeletionInfoMetrics negativeLocalDeletionInfoMetrics = new NegativeLocalDeletionInfoMetrics();
private static final Comparator<Partition> partitionComparator = Comparator.comparing(Partition::partitionKey);
protected final SortedSet<Partition> outOfOrder = new TreeSet<>(partitionComparator);
protected int goodPartitions;
protected int badPartitions;
protected int emptyPartitions;
protected SortedTableScrubber(ColumnFamilyStore cfs,
LifecycleTransaction transaction,
OutputHandler outputHandler,
Options options)
{
this.sstable = (R) transaction.onlyOne();
Preconditions.checkNotNull(sstable.metadata());
assert sstable.metadata().keyspace.equals(cfs.getKeyspaceName());
if (!sstable.descriptor.cfname.equals(cfs.metadata().name))
{
logger.warn("Descriptor points to a different table {} than metadata {}", sstable.descriptor.cfname, cfs.metadata().name);
}
try
{
sstable.metadata().validateCompatibility(cfs.metadata());
}
catch (ConfigurationException ex)
{
logger.warn("Descriptor points to a different table {} than metadata {}", sstable.descriptor.cfname, cfs.metadata().name);
}
this.cfs = cfs;
this.transaction = transaction;
this.outputHandler = outputHandler;
this.options = options;
this.destination = cfs.getDirectories().getLocationForDisk(cfs.getDiskBoundaries().getCorrectDiskForSSTable(sstable));
this.isCommutative = cfs.metadata().isCounter();
List<SSTableReader> toScrub = Collections.singletonList(sstable);
long approximateKeyCount;
try
{
approximateKeyCount = SSTableReader.getApproximateKeyCount(toScrub);
}
catch (RuntimeException ex)
{
approximateKeyCount = 0;
}
this.expectedBloomFilterSize = Math.max(cfs.metadata().params.minIndexInterval, approximateKeyCount);
// loop through each partition, 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
// partition header (key or data size) is corrupt. (This means our position in the index file will be one
// partition "ahead" of the data file.)
this.dataFile = transaction.isOffline()
? sstable.openDataReader()
: sstable.openDataReader(CompactionManager.instance.getRateLimiter());
this.scrubInfo = new ScrubInfo(dataFile, sstable, fileAccessLock.readLock());
if (options.reinsertOverflowedTTLRows)
outputHandler.output("Starting scrub with reinsert overflowed TTL option");
}
public static void deleteOrphanedComponents(Descriptor descriptor, Set<Component> components)
{
File dataFile = descriptor.fileFor(Components.DATA);
if (components.contains(Components.DATA) && dataFile.length() > 0)
// everything appears to be in order... moving on.
return;
// missing the DATA file! all components are orphaned
logger.warn("Removing orphans for {}: {}", descriptor, components);
for (Component component : components)
{
File file = descriptor.fileFor(component);
if (file.exists())
descriptor.fileFor(component).delete();
}
}
@Override
public void scrub()
{
List<SSTableReader> finished = new ArrayList<>();
outputHandler.output("Scrubbing %s (%s)", sstable, FBUtilities.prettyPrintMemory(dataFile.length()));
try (SSTableRewriter writer = SSTableRewriter.construct(cfs, transaction, false, sstable.maxDataAge);
Refs<SSTableReader> refs = Refs.ref(Collections.singleton(sstable)))
{
StatsMetadata metadata = sstable.getSSTableMetadata();
writer.switchWriter(CompactionManager.createWriter(cfs, destination, expectedBloomFilterSize, metadata.repairedAt, metadata.pendingRepair, metadata.isTransient, sstable, transaction));
scrubInternal(writer);
if (!outOfOrder.isEmpty())
finished.add(writeOutOfOrderPartitions(metadata));
// finish obsoletes the old sstable
transaction.obsoleteOriginals();
finished.addAll(writer.setRepairedAt(badPartitions > 0 ? ActiveRepairService.UNREPAIRED_SSTABLE : sstable.getSSTableMetadata().repairedAt).finish());
}
catch (IOException ex)
{
throw new RuntimeException(ex);
}
finally
{
if (transaction.isOffline())
finished.forEach(sstable -> sstable.selfRef().release());
}
outputSummary(finished);
}
protected abstract void scrubInternal(SSTableRewriter writer) throws IOException;
private void outputSummary(List<SSTableReader> finished)
{
if (!finished.isEmpty())
{
outputHandler.output("Scrub of %s complete: %d partitions in new sstable and %d empty (tombstoned) partitions dropped", sstable, goodPartitions, emptyPartitions);
if (negativeLocalDeletionInfoMetrics.fixedRows > 0)
outputHandler.output("Fixed %d rows with overflowed local deletion time.", negativeLocalDeletionInfoMetrics.fixedRows);
if (badPartitions > 0)
outputHandler.warn("Unable to recover %d partitions 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", badPartitions);
}
else
{
if (badPartitions > 0)
outputHandler.warn("No valid partitions found while scrubbing %s; it is marked for deletion now. If you want to attempt manual recovery, you can find a copy in the pre-scrub snapshot", sstable);
else
outputHandler.output("Scrub of %s complete; looks like all %d partitions were tombstoned", sstable, emptyPartitions);
}
}
private SSTableReader writeOutOfOrderPartitions(StatsMetadata metadata)
{
// out of order partitions/rows, but no bad partition found - we can keep our repairedAt time
long repairedAt = badPartitions > 0 ? ActiveRepairService.UNREPAIRED_SSTABLE : sstable.getSSTableMetadata().repairedAt;
SSTableReader newInOrderSstable;
try (SSTableWriter inOrderWriter = CompactionManager.createWriter(cfs, destination, expectedBloomFilterSize, repairedAt, metadata.pendingRepair, metadata.isTransient, sstable, transaction))
{
for (Partition partition : outOfOrder)
inOrderWriter.append(partition.unfilteredIterator());
inOrderWriter.setRepairedAt(-1);
inOrderWriter.setMaxDataAge(sstable.maxDataAge);
newInOrderSstable = inOrderWriter.finish(true);
}
transaction.update(newInOrderSstable, false);
outputHandler.warn("%d out of order partition (or partitions without of order rows) found while scrubbing %s; " +
"Those have been written (in order) to a new sstable (%s)", outOfOrder.size(), sstable, newInOrderSstable);
return newInOrderSstable;
}
protected abstract UnfilteredRowIterator withValidation(UnfilteredRowIterator iter, String filename);
@Override
@VisibleForTesting
public ScrubResult scrubWithResult()
{
scrub();
return new ScrubResult(goodPartitions, badPartitions, emptyPartitions);
}
@Override
public CompactionInfo.Holder getScrubInfo()
{
return scrubInfo;
}
protected String keyString(DecoratedKey key)
{
if (key == null)
return "(unknown)";
try
{
return cfs.metadata().partitionKeyType.getString(key.getKey());
}
catch (Exception e)
{
return String.format("(corrupted; hex value: %s)", ByteBufferUtil.bytesToHex(key.getKey()));
}
}
protected boolean tryAppend(DecoratedKey prevKey, DecoratedKey key, SSTableRewriter writer)
{
// 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.
try (OrderCheckerIterator sstableIterator = new OrderCheckerIterator(getIterator(key), cfs.metadata().comparator);
UnfilteredRowIterator iterator = withValidation(sstableIterator, dataFile.getPath()))
{
if (prevKey != null && prevKey.compareTo(key) > 0)
{
saveOutOfOrderPartition(prevKey, key, iterator);
return false;
}
if (writer.tryAppend(iterator) == null)
emptyPartitions++;
else
goodPartitions++;
if (sstableIterator.hasRowsOutOfOrder())
{
outputHandler.warn("Out of order rows found in partition: %s", keyString(key));
outOfOrder.add(sstableIterator.getRowsOutOfOrder());
}
}
return true;
}
/**
* Only wrap with {@link FixNegativeLocalDeletionTimeIterator} if {@link IScrubber.Options#reinsertOverflowedTTLRows} option
* is specified
*/
private UnfilteredRowIterator getIterator(DecoratedKey key)
{
RowMergingSSTableIterator rowMergingIterator = new RowMergingSSTableIterator(SSTableIdentityIterator.create(sstable,
dataFile,
key),
outputHandler,
sstable.descriptor.version,
options.reinsertOverflowedTTLRows);
if (options.reinsertOverflowedTTLRows)
return new FixNegativeLocalDeletionTimeIterator(rowMergingIterator, outputHandler, negativeLocalDeletionInfoMetrics);
else
return rowMergingIterator;
}
private void saveOutOfOrderPartition(DecoratedKey prevKey, DecoratedKey key, UnfilteredRowIterator iterator)
{
// TODO bitch if the row is too large? if it is there's not much we can do ...
outputHandler.warn("Out of order partition detected (%s found after %s)", keyString(key), keyString(prevKey));
outOfOrder.add(ImmutableBTreePartition.create(iterator));
}
protected static void throwIfFatal(Throwable th)
{
if (th instanceof Error && !(th instanceof AssertionError || th instanceof IOError))
throw (Error) th;
}
protected void throwIfCannotContinue(DecoratedKey key, Throwable th)
{
if (isCommutative && !options.skipCorrupted)
{
outputHandler.warn("An error occurred while scrubbing the partition with key '%s'. Skipping corrupt " +
"data in counter tables will result in undercounts for the affected " +
"counters (see CASSANDRA-2759 for more details), so by default the scrub will " +
"stop at this point. If you would like to skip the row anyway and continue " +
"scrubbing, re-run the scrub with the --skip-corrupted option.",
keyString(key));
throw new IOError(th);
}
}
public static class ScrubInfo extends CompactionInfo.Holder
{
private final RandomAccessReader dataFile;
private final SSTableReader sstable;
private final TimeUUID scrubCompactionId;
private final Lock fileReadLock;
public ScrubInfo(RandomAccessReader dataFile, SSTableReader sstable, Lock fileReadLock)
{
this.dataFile = dataFile;
this.sstable = sstable;
this.fileReadLock = fileReadLock;
scrubCompactionId = nextTimeUUID();
}
public CompactionInfo getCompactionInfo()
{
fileReadLock.lock();
try
{
return new CompactionInfo(sstable.metadata(),
OperationType.SCRUB,
dataFile.getFilePointer(),
dataFile.length(),
scrubCompactionId,
ImmutableSet.of(sstable),
File.getPath(sstable.getFilename()).getParent().toString());
}
catch (Exception e)
{
throw new RuntimeException(e);
}
finally
{
fileReadLock.unlock();
}
}
public boolean isGlobal()
{
return false;
}
}
/**
* In some case like CASSANDRA-12127 the cells might have been stored in the wrong order. This decorator check the
* cells order and collect the out-of-order cells to correct the problem.
*/
private static final class OrderCheckerIterator extends AbstractIterator<Unfiltered> implements WrappingUnfilteredRowIterator
{
private final UnfilteredRowIterator iterator;
private final ClusteringComparator comparator;
private Unfiltered previous;
/**
* The partition containing the rows which are out of order.
*/
private Partition rowsOutOfOrder;
public OrderCheckerIterator(UnfilteredRowIterator iterator, ClusteringComparator comparator)
{
this.iterator = iterator;
this.comparator = comparator;
}
@Override
public UnfilteredRowIterator wrapped()
{
return iterator;
}
public boolean hasRowsOutOfOrder()
{
return rowsOutOfOrder != null;
}
public Partition getRowsOutOfOrder()
{
return rowsOutOfOrder;
}
@Override
protected Unfiltered computeNext()
{
if (!iterator.hasNext())
return endOfData();
Unfiltered next = iterator.next();
// If we detect that some rows are out of order we will store and sort the remaining ones to insert them
// in a separate SSTable.
if (previous != null && comparator.compare(next, previous) < 0)
{
rowsOutOfOrder = ImmutableBTreePartition.create(UnfilteredRowIterators.concat(next, iterator), false);
return endOfData();
}
previous = next;
return next;
}
}
/**
* During 2.x migration, under some circumstances rows might have gotten duplicated.
* Merging iterator merges rows with same clustering.
* <p>
* For more details, refer to CASSANDRA-12144.
*/
private static class RowMergingSSTableIterator implements WrappingUnfilteredRowIterator
{
Unfiltered nextToOffer = null;
private final OutputHandler output;
private final UnfilteredRowIterator wrapped;
private final Version sstableVersion;
private final boolean reinsertOverflowedTTLRows;
RowMergingSSTableIterator(UnfilteredRowIterator source, OutputHandler output, Version sstableVersion, boolean reinsertOverflowedTTLRows)
{
this.wrapped = source;
this.output = output;
this.sstableVersion = sstableVersion;
this.reinsertOverflowedTTLRows = reinsertOverflowedTTLRows;
}
@Override
public UnfilteredRowIterator wrapped()
{
return wrapped;
}
@Override
public boolean hasNext()
{
return nextToOffer != null || wrapped.hasNext();
}
@Override
public Unfiltered next()
{
Unfiltered next = nextToOffer != null ? nextToOffer : wrapped.next();
if (next.isRow())
{
boolean logged = false;
while (wrapped.hasNext())
{
Unfiltered peek = wrapped.next();
if (!peek.isRow() || !next.clustering().equals(peek.clustering()))
{
nextToOffer = peek; // Offer peek in next call
return computeFinalRow((Row) next);
}
// Duplicate row, merge it.
next = Rows.merge((Row) next, (Row) peek);
if (!logged)
{
String partitionKey = metadata().partitionKeyType.getString(partitionKey().getKey());
output.warn("Duplicate row detected in %s.%s: %s %s", metadata().keyspace, metadata().name, partitionKey, next.clustering().toString(metadata()));
logged = true;
}
}
}
nextToOffer = null;
return computeFinalRow((Row) next);
}
private Row computeFinalRow(Row next)
{
// If the row has overflowed let rows skip them unless we need to keep them for the overflow policy
if (hasOverflowedLocalExpirationTimeRow(next) && !reinsertOverflowedTTLRows)
return null;
else if (reinsertOverflowedTTLRows)
return rebuildTimestamptsForOverflowedRows(next);
else
return next;
}
/*
* When building ldt on deser it won't overflow now being a long as it used to.
* This causes row resurrection for old sstable formats!
* To prevent it we preserve the overflow to be backwards compatible and to feed into the overflow policy
*/
private Row rebuildTimestamptsForOverflowedRows(Row row)
{
if (sstableVersion.hasUIntDeletionTime())
return row;
LivenessInfo livenessInfo = row.primaryKeyLivenessInfo();
if (livenessInfo.isExpiring() && livenessInfo.localExpirationTime() >= 0)
{
livenessInfo = livenessInfo.withUpdatedTimestampAndLocalDeletionTime(livenessInfo.timestamp(), livenessInfo.localExpirationTime(), false);
}
return row.transformAndFilter(livenessInfo, row.deletion(), cd -> {
if (cd.column().isSimple())
{
Cell<?> cell = (Cell<?>)cd;
return cell.isExpiring() && cell.localDeletionTime() >= 0
? cell.withUpdatedTimestampAndLocalDeletionTime(cell.timestamp(), cell.localDeletionTime())
: cell;
}
else
{
ComplexColumnData complexData = (ComplexColumnData)cd;
return complexData.transformAndFilter(cell -> cell.isExpiring() && cell.localDeletionTime() >= 0
? cell.withUpdatedTimestampAndLocalDeletionTime(cell.timestamp(), cell.localDeletionTime())
: cell);
}
}).clone(HeapCloner.instance);
}
private boolean hasOverflowedLocalExpirationTimeRow(Row next)
{
if (sstableVersion.hasUIntDeletionTime())
return false;
if (next.primaryKeyLivenessInfo().isExpiring() && next.primaryKeyLivenessInfo().localExpirationTime() >= 0)
{
return true;
}
for (ColumnData cd : next)
{
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;
}
}
/**
* This iterator converts negative {@link AbstractCell#localDeletionTime()} into {@link AbstractCell#MAX_DELETION_TIME}
* <p>
* This is to recover entries with overflowed localExpirationTime due to CASSANDRA-14092
*/
private static final class FixNegativeLocalDeletionTimeIterator extends AbstractIterator<Unfiltered> implements WrappingUnfilteredRowIterator
{
/**
* 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;
}
@Override
public UnfilteredRowIterator wrapped()
{
return iterator;
}
@Override
protected Unfiltered computeNext()
{
if (!iterator.hasNext())
return endOfData();
Unfiltered next = iterator.next();
if (!next.isRow())
return next;
if (hasNegativeLocalExpirationTime((Row) next))
{
outputHandler.debug("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() == Cell.INVALID_DELETION_TIME)
{
return true;
}
for (ColumnData cd : row)
{
if (cd.column().isSimple())
{
Cell<?> cell = (Cell<?>) cd;
if (cell.isExpiring() && cell.localDeletionTime() == Cell.INVALID_DELETION_TIME)
return true;
}
else
{
ComplexColumnData complexData = (ComplexColumnData) cd;
for (Cell<?> cell : complexData)
{
if (cell.isExpiring() && cell.localDeletionTime() == Cell.INVALID_DELETION_TIME)
return true;
}
}
}
return false;
}
private Unfiltered fixNegativeLocalExpirationTime(Row row)
{
LivenessInfo livenessInfo = row.primaryKeyLivenessInfo();
if (livenessInfo.isExpiring() && livenessInfo.localExpirationTime() == Cell.INVALID_DELETION_TIME)
livenessInfo = livenessInfo.withUpdatedTimestampAndLocalDeletionTime(livenessInfo.timestamp() + 1, AbstractCell.MAX_DELETION_TIME_2038_LEGACY_CAP);
return row.transformAndFilter(livenessInfo, row.deletion(), cd -> {
if (cd.column().isSimple())
{
Cell cell = (Cell) cd;
return cell.isExpiring() && cell.localDeletionTime() == Cell.INVALID_DELETION_TIME
? cell.withUpdatedTimestampAndLocalDeletionTime(cell.timestamp() + 1, AbstractCell.MAX_DELETION_TIME_2038_LEGACY_CAP)
: cell;
}
else
{
ComplexColumnData complexData = (ComplexColumnData) cd;
return complexData.transformAndFilter(cell -> cell.isExpiring() && cell.localDeletionTime() == Cell.INVALID_DELETION_TIME
? cell.withUpdatedTimestampAndLocalDeletionTime(cell.timestamp() + 1, AbstractCell.MAX_DELETION_TIME_2038_LEGACY_CAP)
: cell);
}
}).clone(HeapCloner.instance);
}
}
private static class NegativeLocalDeletionInfoMetrics
{
public volatile int fixedRows = 0;
}
}