natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add option to only purge tombstones from repaired sstables 

Patch by marcuse; reviewed by yukim for CASSANDRA-6434
This commit is contained in:
Marcus Eriksson 2015-07-07 07:50:11 +02:00
parent 76ca69717c
commit 6f0c12f3a4
14 changed files with 437 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
3.0.0-beta1
* Add option to only purge repaired tombstones (CASSANDRA-6434)
* Change authorization handling for MVs (CASSANDRA-9927)
* Add custom JMX enabled executor for UDF sandbox (CASSANDRA-10026)
* Fix row deletion bug for Materialized Views (CASSANDRA-10014)

6
NEWS.txt
View File

@ -23,6 +23,12 @@ New features
for non-primary key queries, and perform much better for indexing high
cardinality columns.
See http://www.datastax.com/dev/blog/new-in-cassandra-3-0-materialized-views
- Option to not purge unrepaired tombstones. To avoid users having data resurrected
if repair has not been run within gc_grace_seconds, an option has been added to
only allow tombstones from repaired sstables to be purged. To enable, set the
compaction option 'only_purge_repaired_tombstones':true but keep in mind that if
you do not run repair for a long time, you will keep all tombstones around which
can cause other problems.
Upgrading

2
pylib/cqlshlib/cql3handling.py
View File

@ -69,7 +69,7 @@ class Cql3ParsingRuleSet(CqlParsingRuleSet):
# (CQL3 option name, schema_columnfamilies column name (or None if same),
# list of known map keys)
('compaction', 'compaction_strategy_options',
('class', 'max_threshold', 'tombstone_compaction_interval', 'tombstone_threshold', 'enabled', 'unchecked_tombstone_compaction')),
('class', 'max_threshold', 'tombstone_compaction_interval', 'tombstone_threshold', 'enabled', 'unchecked_tombstone_compaction', 'only_purge_repaired_tombstones')),
('compression', 'compression_parameters',
('sstable_compression', 'chunk_length_kb', 'crc_check_chance')),
('caching', None,

95
src/java/org/apache/cassandra/db/Memtable.java
View File

@ -260,7 +260,7 @@ public class Memtable implements Comparable<Memtable>
100 * allocator.onHeap().ownershipRatio(), 100 * allocator.offHeap().ownershipRatio());
}
public UnfilteredPartitionIterator makePartitionIterator(final ColumnFilter columnFilter, final DataRange dataRange, final boolean isForThrift)
public MemtableUnfilteredPartitionIterator makePartitionIterator(final ColumnFilter columnFilter, final DataRange dataRange, final boolean isForThrift)
{
AbstractBounds<PartitionPosition> keyRange = dataRange.keyRange();
@ -278,35 +278,26 @@ public class Memtable implements Comparable<Memtable>
? partitions.tailMap(keyRange.left, includeStart)
: partitions.subMap(keyRange.left, includeStart, keyRange.right, includeStop);
int minLocalDeletionTime = Integer.MAX_VALUE;
// avoid iterating over the memtable if we purge all tombstones
if (cfs.getCompactionStrategyManager().onlyPurgeRepairedTombstones())
minLocalDeletionTime = findMinLocalDeletionTime(subMap.entrySet().iterator());
final Iterator<Map.Entry<PartitionPosition, AtomicBTreePartition>> iter = subMap.entrySet().iterator();
return new AbstractUnfilteredPartitionIterator()
return new MemtableUnfilteredPartitionIterator(cfs, iter, isForThrift, minLocalDeletionTime, columnFilter, dataRange);
}
private int findMinLocalDeletionTime(Iterator<Map.Entry<PartitionPosition, AtomicBTreePartition>> iterator)
{
int minLocalDeletionTime = Integer.MAX_VALUE;
while (iterator.hasNext())
{
public boolean isForThrift()
{
return isForThrift;
}
public CFMetaData metadata()
{
return cfs.metadata;
}
public boolean hasNext()
{
return iter.hasNext();
}
public UnfilteredRowIterator next()
{
Map.Entry<PartitionPosition, AtomicBTreePartition> entry = iter.next();
// Actual stored key should be true DecoratedKey
assert entry.getKey() instanceof DecoratedKey;
DecoratedKey key = (DecoratedKey)entry.getKey();
ClusteringIndexFilter filter = dataRange.clusteringIndexFilter(key);
return filter.getUnfilteredRowIterator(columnFilter, entry.getValue());
}
};
Map.Entry<PartitionPosition, AtomicBTreePartition> entry = iterator.next();
minLocalDeletionTime = Math.min(minLocalDeletionTime, entry.getValue().stats().minLocalDeletionTime);
}
return minLocalDeletionTime;
}
public Partition getPartition(DecoratedKey key)
@ -463,6 +454,56 @@ public class Memtable implements Comparable<Memtable>
}
}
public static class MemtableUnfilteredPartitionIterator extends AbstractUnfilteredPartitionIterator
{
private final ColumnFamilyStore cfs;
private final Iterator<Map.Entry<PartitionPosition, AtomicBTreePartition>> iter;
private final boolean isForThrift;
private final int minLocalDeletionTime;
private final ColumnFilter columnFilter;
private final DataRange dataRange;
public MemtableUnfilteredPartitionIterator(ColumnFamilyStore cfs, Iterator<Map.Entry<PartitionPosition, AtomicBTreePartition>> iter, boolean isForThrift, int minLocalDeletionTime, ColumnFilter columnFilter, DataRange dataRange)
{
this.cfs = cfs;
this.iter = iter;
this.isForThrift = isForThrift;
this.minLocalDeletionTime = minLocalDeletionTime;
this.columnFilter = columnFilter;
this.dataRange = dataRange;
}
public boolean isForThrift()
{
return isForThrift;
}
public int getMinLocalDeletionTime()
{
return minLocalDeletionTime;
}
public CFMetaData metadata()
{
return cfs.metadata;
}
public boolean hasNext()
{
return iter.hasNext();
}
public UnfilteredRowIterator next()
{
Map.Entry<PartitionPosition, AtomicBTreePartition> entry = iter.next();
// Actual stored key should be true DecoratedKey
assert entry.getKey() instanceof DecoratedKey;
DecoratedKey key = (DecoratedKey)entry.getKey();
ClusteringIndexFilter filter = dataRange.clusteringIndexFilter(key);
return filter.getUnfilteredRowIterator(columnFilter, entry.getValue());
}
}
private static class ColumnsCollector
{
private final HashMap<ColumnDefinition, AtomicBoolean> predefined = new HashMap<>();

13
src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
View File

@ -53,6 +53,7 @@ public class PartitionRangeReadCommand extends ReadCommand
protected static final SelectionDeserializer selectionDeserializer = new Deserializer();
private final DataRange dataRange;
private int oldestUnrepairedTombstone = Integer.MAX_VALUE;
public PartitionRangeReadCommand(boolean isDigest,
boolean isForThrift,
@ -172,7 +173,8 @@ public class PartitionRangeReadCommand extends ReadCommand
for (Memtable memtable : view.memtables)
{
@SuppressWarnings("resource") // We close on exception and on closing the result returned by this method
UnfilteredPartitionIterator iter = memtable.makePartitionIterator(columnFilter(), dataRange(), isForThrift());
Memtable.MemtableUnfilteredPartitionIterator iter = memtable.makePartitionIterator(columnFilter(), dataRange(), isForThrift());
oldestUnrepairedTombstone = Math.min(oldestUnrepairedTombstone, iter.getMinLocalDeletionTime());
iterators.add(isForThrift() ? ThriftResultsMerger.maybeWrap(iter, metadata(), nowInSec()) : iter);
}
@ -181,8 +183,9 @@ public class PartitionRangeReadCommand extends ReadCommand
@SuppressWarnings("resource") // We close on exception and on closing the result returned by this method
UnfilteredPartitionIterator iter = sstable.getScanner(columnFilter(), dataRange(), isForThrift());
iterators.add(isForThrift() ? ThriftResultsMerger.maybeWrap(iter, metadata(), nowInSec()) : iter);
if (!sstable.isRepaired())
oldestUnrepairedTombstone = Math.min(oldestUnrepairedTombstone, sstable.getMinLocalDeletionTime());
}
return checkCacheFilter(UnfilteredPartitionIterators.mergeLazily(iterators, nowInSec()), cfs);
}
catch (RuntimeException | Error e)
@ -200,6 +203,12 @@ public class PartitionRangeReadCommand extends ReadCommand
}
}
@Override
protected int oldestUnrepairedTombstone()
{
return oldestUnrepairedTombstone;
}
private UnfilteredPartitionIterator checkCacheFilter(UnfilteredPartitionIterator iter, final ColumnFamilyStore cfs)
{
return new WrappingUnfilteredPartitionIterator(iter)

4
src/java/org/apache/cassandra/db/ReadCommand.java
View File

@ -249,6 +249,8 @@ public abstract class ReadCommand implements ReadQuery
protected abstract UnfilteredPartitionIterator queryStorage(ColumnFamilyStore cfs, ReadOrderGroup orderGroup);
protected abstract int oldestUnrepairedTombstone();
public ReadResponse createResponse(UnfilteredPartitionIterator iterator)
{
return isDigestQuery()
@ -426,7 +428,7 @@ public abstract class ReadCommand implements ReadQuery
// are to some extend an artefact of compaction lagging behind and hence counting them is somewhat unintuitive).
protected UnfilteredPartitionIterator withoutPurgeableTombstones(UnfilteredPartitionIterator iterator, ColumnFamilyStore cfs)
{
return new PurgingPartitionIterator(iterator, cfs.gcBefore(nowInSec()))
return new PurgingPartitionIterator(iterator, cfs.gcBefore(nowInSec()), oldestUnrepairedTombstone(), cfs.getCompactionStrategyManager().onlyPurgeRepairedTombstones())
{
protected long getMaxPurgeableTimestamp()
{

16
src/java/org/apache/cassandra/db/SinglePartitionNamesCommand.java
View File

@ -44,6 +44,7 @@ import org.apache.cassandra.utils.memory.HeapAllocator;
*/
public class SinglePartitionNamesCommand extends SinglePartitionReadCommand<ClusteringIndexNamesFilter>
{
private int oldestUnrepairedDeletionTime = Integer.MAX_VALUE;
protected SinglePartitionNamesCommand(boolean isDigest,
boolean isForThrift,
CFMetaData metadata,
@ -84,6 +85,12 @@ public class SinglePartitionNamesCommand extends SinglePartitionReadCommand<Clus
return new SinglePartitionNamesCommand(isDigestQuery(), isForThrift(), metadata(), nowInSec(), columnFilter(), rowFilter(), limits(), partitionKey(), clusteringIndexFilter());
}
@Override
protected int oldestUnrepairedTombstone()
{
return oldestUnrepairedDeletionTime;
}
protected UnfilteredRowIterator queryMemtableAndDiskInternal(ColumnFamilyStore cfs, boolean copyOnHeap)
{
Tracing.trace("Acquiring sstable references");
@ -107,7 +114,7 @@ public class SinglePartitionNamesCommand extends SinglePartitionReadCommand<Clus
UnfilteredRowIterator clonedFilter = copyOnHeap
? UnfilteredRowIterators.cloningIterator(iter, HeapAllocator.instance)
: iter;
result = add(isForThrift() ? ThriftResultsMerger.maybeWrap(clonedFilter, nowInSec()) : clonedFilter, result);
result = add(isForThrift() ? ThriftResultsMerger.maybeWrap(clonedFilter, nowInSec()) : clonedFilter, result, false);
}
}
@ -137,7 +144,7 @@ public class SinglePartitionNamesCommand extends SinglePartitionReadCommand<Clus
continue;
sstablesIterated++;
result = add(isForThrift() ? ThriftResultsMerger.maybeWrap(iter, nowInSec()) : iter, result);
result = add(isForThrift() ? ThriftResultsMerger.maybeWrap(iter, nowInSec()) : iter, result, sstable.isRepaired());
}
}
@ -175,8 +182,11 @@ public class SinglePartitionNamesCommand extends SinglePartitionReadCommand<Clus
return result.unfilteredIterator(columnFilter(), Slices.ALL, clusteringIndexFilter().isReversed());
}
private ArrayBackedPartition add(UnfilteredRowIterator iter, ArrayBackedPartition result)
private ArrayBackedPartition add(UnfilteredRowIterator iter, ArrayBackedPartition result, boolean isRepaired)
{
if (!isRepaired)
oldestUnrepairedDeletionTime = Math.min(oldestUnrepairedDeletionTime, iter.stats().minLocalDeletionTime);
int maxRows = Math.max(clusteringIndexFilter().requestedRows().size(), 1);
if (result == null)
return ArrayBackedPartition.create(iter, maxRows);

15
src/java/org/apache/cassandra/db/SinglePartitionSliceCommand.java
View File

@ -41,6 +41,8 @@ import org.apache.cassandra.utils.memory.HeapAllocator;
*/
public class SinglePartitionSliceCommand extends SinglePartitionReadCommand<ClusteringIndexSliceFilter>
{
private int oldestUnrepairedTombstone = Integer.MAX_VALUE;
public SinglePartitionSliceCommand(boolean isDigest,
boolean isForThrift,
CFMetaData metadata,
@ -119,6 +121,12 @@ public class SinglePartitionSliceCommand extends SinglePartitionReadCommand<Clus
return new SinglePartitionSliceCommand(isDigestQuery(), isForThrift(), metadata(), nowInSec(), columnFilter(), rowFilter(), limits(), partitionKey(), clusteringIndexFilter());
}
@Override
protected int oldestUnrepairedTombstone()
{
return oldestUnrepairedTombstone;
}
protected UnfilteredRowIterator queryMemtableAndDiskInternal(ColumnFamilyStore cfs, boolean copyOnHeap)
{
Tracing.trace("Acquiring sstable references");
@ -139,9 +147,9 @@ public class SinglePartitionSliceCommand extends SinglePartitionReadCommand<Clus
UnfilteredRowIterator iter = filter.getUnfilteredRowIterator(columnFilter(), partition);
@SuppressWarnings("resource") // same as above
UnfilteredRowIterator maybeCopied = copyOnHeap ? UnfilteredRowIterators.cloningIterator(iter, HeapAllocator.instance) : iter;
oldestUnrepairedTombstone = Math.min(oldestUnrepairedTombstone, partition.stats().minLocalDeletionTime);
iterators.add(isForThrift() ? ThriftResultsMerger.maybeWrap(maybeCopied, nowInSec()) : maybeCopied);
}
/*
* We can't eliminate full sstables based on the timestamp of what we've already read like
* in collectTimeOrderedData, but we still want to eliminate sstable whose maxTimestamp < mostRecentTombstone
@ -185,6 +193,9 @@ public class SinglePartitionSliceCommand extends SinglePartitionReadCommand<Clus
sstable.incrementReadCount();
@SuppressWarnings("resource") // 'iter' is added to iterators which is closed on exception, or through the closing of the final merged iterator
UnfilteredRowIterator iter = filter.filter(sstable.iterator(partitionKey(), columnFilter(), filter.isReversed(), isForThrift()));
if (!sstable.isRepaired())
oldestUnrepairedTombstone = Math.min(oldestUnrepairedTombstone, sstable.getMinLocalDeletionTime());
iterators.add(isForThrift() ? ThriftResultsMerger.maybeWrap(iter, nowInSec()) : iter);
mostRecentPartitionTombstone = Math.max(mostRecentPartitionTombstone, iter.partitionLevelDeletion().markedForDeleteAt());
sstablesIterated++;
@ -205,6 +216,8 @@ public class SinglePartitionSliceCommand extends SinglePartitionReadCommand<Clus
if (iter.partitionLevelDeletion().markedForDeleteAt() > minTimestamp)
{
iterators.add(iter);
if (!sstable.isRepaired())
oldestUnrepairedTombstone = Math.min(oldestUnrepairedTombstone, sstable.getMinLocalDeletionTime());
includedDueToTombstones++;
sstablesIterated++;
}

4
src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
View File

@ -26,7 +26,6 @@ import com.google.common.collect.Iterables;
import com.google.common.util.concurrent.RateLimiter;
import org.apache.cassandra.db.lifecycle.SSTableSet;
import org.apache.cassandra.db.lifecycle.View;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -39,7 +38,6 @@ import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.io.sstable.Component;
import org.apache.cassandra.io.sstable.ISSTableScanner;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.JVMStabilityInspector;
/**
@ -64,6 +62,7 @@ public abstract class AbstractCompactionStrategy
// disable range overlap check when deciding if an SSTable is candidate for tombstone compaction (CASSANDRA-6563)
protected static final String UNCHECKED_TOMBSTONE_COMPACTION_OPTION = "unchecked_tombstone_compaction";
protected static final String COMPACTION_ENABLED = "enabled";
public static final String ONLY_PURGE_REPAIRED_TOMBSTONES = "only_purge_repaired_tombstones";
protected Map<String, String> options;
@ -453,6 +452,7 @@ public abstract class AbstractCompactionStrategy
uncheckedOptions.remove(TOMBSTONE_COMPACTION_INTERVAL_OPTION);
uncheckedOptions.remove(UNCHECKED_TOMBSTONE_COMPACTION_OPTION);
uncheckedOptions.remove(COMPACTION_ENABLED);
uncheckedOptions.remove(ONLY_PURGE_REPAIRED_TOMBSTONES);
return uncheckedOptions;
}

20
src/java/org/apache/cassandra/db/compaction/CompactionController.java
View File

@ -20,14 +20,13 @@ package org.apache.cassandra.db.compaction;
import java.util.*;
import org.apache.cassandra.db.lifecycle.SSTableSet;
import org.apache.cassandra.db.lifecycle.View;
import com.google.common.collect.Iterables;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.lifecycle.SSTableIntervalTree;
import org.apache.cassandra.db.lifecycle.Tracker;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.PartitionPosition;
import org.apache.cassandra.utils.AlwaysPresentFilter;
@ -45,6 +44,7 @@ public class CompactionController implements AutoCloseable
private static final Logger logger = LoggerFactory.getLogger(CompactionController.class);
public final ColumnFamilyStore cfs;
private final boolean compactingRepaired;
private Refs<SSTableReader> overlappingSSTables;
private OverlapIterator<PartitionPosition, SSTableReader> overlapIterator;
private final Iterable<SSTableReader> compacting;
@ -56,12 +56,13 @@ public class CompactionController implements AutoCloseable
this(cfs, null, maxValue);
}
public CompactionController(ColumnFamilyStore cfs, Set<SSTableReader> compacting, int gcBefore)
public CompactionController(ColumnFamilyStore cfs, Set<SSTableReader> compacting, int gcBefore)
{
assert cfs != null;
this.cfs = cfs;
this.gcBefore = gcBefore;
this.compacting = compacting;
compactingRepaired = compacting != null && compacting.stream().allMatch(SSTableReader::isRepaired);
refreshOverlaps();
}
@ -117,6 +118,9 @@ public class CompactionController implements AutoCloseable
if (compacting == null)
return Collections.<SSTableReader>emptySet();
if (cfStore.getCompactionStrategyManager().onlyPurgeRepairedTombstones() && !Iterables.all(compacting, SSTableReader::isRepaired))
return Collections.emptySet();
List<SSTableReader> candidates = new ArrayList<>();
long minTimestamp = Long.MAX_VALUE;
@ -177,6 +181,9 @@ public class CompactionController implements AutoCloseable
*/
public long maxPurgeableTimestamp(DecoratedKey key)
{
if (!compactingRepaired())
return Long.MIN_VALUE;
long min = Long.MAX_VALUE;
overlapIterator.update(key);
for (SSTableReader sstable : overlapIterator.overlaps())
@ -201,4 +208,9 @@ public class CompactionController implements AutoCloseable
overlappingSSTables.release();
}
public boolean compactingRepaired()
{
return !cfs.getCompactionStrategyManager().onlyPurgeRepairedTombstones() || compactingRepaired;
}
}

2
src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
View File

@ -252,7 +252,7 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
private PurgeIterator(UnfilteredPartitionIterator toPurge, CompactionController controller)
{
super(toPurge, controller.gcBefore);
super(toPurge, controller.gcBefore, controller.compactingRepaired() ? Integer.MIN_VALUE : Integer.MAX_VALUE, controller.cfs.getCompactionStrategyManager().onlyPurgeRepairedTombstones());
this.controller = controller;
}

5
src/java/org/apache/cassandra/db/compaction/CompactionStrategyManager.java
View File

@ -479,4 +479,9 @@ public class CompactionStrategyManager implements INotificationConsumer
{
return params;
}
public boolean onlyPurgeRepairedTombstones()
{
return Boolean.parseBoolean(params.options().get(AbstractCompactionStrategy.ONLY_PURGE_REPAIRED_TOMBSTONES));
}
}

5
src/java/org/apache/cassandra/db/partitions/PurgingPartitionIterator.java
View File

@ -27,7 +27,7 @@ public abstract class PurgingPartitionIterator extends WrappingUnfilteredPartiti
private UnfilteredRowIterator next;
public PurgingPartitionIterator(UnfilteredPartitionIterator iterator, int gcBefore)
public PurgingPartitionIterator(UnfilteredPartitionIterator iterator, int gcBefore, int oldestUnrepairedTombstone, boolean onlyPurgeRepairedTombstones)
{
super(iterator);
this.gcBefore = gcBefore;
@ -35,6 +35,9 @@ public abstract class PurgingPartitionIterator extends WrappingUnfilteredPartiti
{
public boolean shouldPurge(long timestamp, int localDeletionTime)
{
if (onlyPurgeRepairedTombstones && localDeletionTime >= oldestUnrepairedTombstone)
return false;
return timestamp < getMaxPurgeableTimestamp() && localDeletionTime < gcBefore;
}
};

292
test/unit/org/apache/cassandra/db/RepairedDataTombstonesTest.java Normal file
View File

@ -0,0 +1,292 @@
/*
* 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.io.IOException;
import java.util.Collections;
import com.google.common.collect.Iterables;
import org.junit.Test;
import org.apache.cassandra.Util;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.db.lifecycle.SSTableSet;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.db.rows.AbstractRow;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.utils.ByteBufferUtil;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
public class RepairedDataTombstonesTest extends CQLTester
{
@Test
public void compactionTest() throws Throwable
{
createTable("create table %s (id int, id2 int, t text, primary key (id, id2)) with gc_grace_seconds=0 and compaction = {'class':'SizeTieredCompactionStrategy', 'only_purge_repaired_tombstones':true}");
// insert a live row to make sure that the sstables are not dropped (we test dropping in compactionDropExpiredSSTableTest() below)
execute("insert into %s (id, id2, t) values (999,999,'live')");
for (int i = 0; i < 10; i++)
{
execute("delete from %s where id=? and id2=?", 1, i);
}
flush();
SSTableReader repairedSSTable = getCurrentColumnFamilyStore().getSSTables(SSTableSet.LIVE).iterator().next();
repair(getCurrentColumnFamilyStore(), repairedSSTable);
Thread.sleep(2000);
execute("insert into %s (id, id2, t) values (999,999,'live')");
for (int i = 10; i < 20; i++)
{
execute("delete from %s where id=? and id2=?", 1, i);
}
flush();
Thread.sleep(1000);
// at this point we have 2 sstables, one repaired and one unrepaired. Both sstables contain expired tombstones, but we should only drop the tombstones from the repaired sstable.
getCurrentColumnFamilyStore().forceMajorCompaction();
verify();
verify2(1);
assertEquals(2, Iterables.size(getCurrentColumnFamilyStore().getSSTables(SSTableSet.LIVE)));
}
@Test
public void compactionDropExpiredSSTableTest() throws Throwable
{
createTable("create table %s (id int, id2 int, t text, primary key (id, id2)) with gc_grace_seconds=0 and compaction = {'class':'SizeTieredCompactionStrategy', 'only_purge_repaired_tombstones':true}");
for (int i = 0; i < 10; i++)
{
execute("delete from %s where id=? and id2=?", 1, i);
}
flush();
SSTableReader repairedSSTable = getCurrentColumnFamilyStore().getSSTables(SSTableSet.LIVE).iterator().next();
repair(getCurrentColumnFamilyStore(), repairedSSTable);
Thread.sleep(2000);
for (int i = 10; i < 20; i++)
{
execute("delete from %s where id=? and id2=?", 1, i);
}
flush();
Thread.sleep(1000);
getCurrentColumnFamilyStore().forceMajorCompaction();
verify();
verify2(1);
assertEquals(1, Iterables.size(getCurrentColumnFamilyStore().getSSTables(SSTableSet.LIVE)));
assertFalse(getCurrentColumnFamilyStore().getSSTables(SSTableSet.LIVE).iterator().next().isRepaired());
}
@Test
public void readTest() throws Throwable
{
createTable("create table %s (id int, id2 int, t text, t2 text, primary key (id, id2)) with gc_grace_seconds=0 and compaction = {'class':'SizeTieredCompactionStrategy', 'only_purge_repaired_tombstones':true}");
for (int i = 0; i < 10; i++)
{
execute("update %s set t2=null where id=? and id2=?", 123, i);
}
flush();
SSTableReader repairedSSTable = getCurrentColumnFamilyStore().getSSTables(SSTableSet.LIVE).iterator().next();
repair(getCurrentColumnFamilyStore(), repairedSSTable);
Thread.sleep(2000);
for (int i = 10; i < 20; i++)
{
execute("update %s set t2=null where id=? and id2=?", 123, i);
}
flush();
// allow gcgrace to properly expire:
Thread.sleep(1000);
// make sure we only see the unrepaired tombstones, the other ones are expired and can be purged
verify();
verify2(123);
}
@Test
public void readOnlyUnrepairedTest() throws Throwable
{
// make sure we keep all tombstones if we only have unrepaired data
createTable("create table %s (id int, id2 int, t text, t2 text, primary key (id, id2)) with gc_grace_seconds=0 and compaction = {'class':'SizeTieredCompactionStrategy', 'only_purge_repaired_tombstones':true}");
for (int i = 10; i < 20; i++)
{
execute("update %s set t2=null where id=? and id2=?", 123, i);
}
flush();
// allow gcgrace to properly expire:
Thread.sleep(1000);
verify();
verify2(123);
}
@Test
public void readTestRowTombstones() throws Throwable
{
createTable("create table %s (id int, id2 int, t text, t2 text, primary key (id, id2)) with gc_grace_seconds=0 and compaction = {'class':'SizeTieredCompactionStrategy', 'only_purge_repaired_tombstones':true}");
for (int i = 0; i < 10; i++)
{
execute("delete from %s where id=? and id2=?", 1, i);
}
flush();
SSTableReader repairedSSTable = getCurrentColumnFamilyStore().getSSTables(SSTableSet.LIVE).iterator().next();
repair(getCurrentColumnFamilyStore(), repairedSSTable);
Thread.sleep(2000);
for (int i = 10; i < 20; i++)
{
execute("delete from %s where id=? and id2=?", 1, i);
}
flush();
Thread.sleep(1000);
verify();
verify2(1);
}
@Test
public void readTestPartitionTombstones() throws Throwable
{
createTable("create table %s (id int, id2 int, t text, t2 text, primary key (id, id2)) with gc_grace_seconds=0 and compaction = {'class':'SizeTieredCompactionStrategy', 'only_purge_repaired_tombstones':true}");
for (int i = 0; i < 10; i++)
{
execute("delete from %s where id=?", i);
}
flush();
SSTableReader repairedSSTable = getCurrentColumnFamilyStore().getSSTables(SSTableSet.LIVE).iterator().next();
repair(getCurrentColumnFamilyStore(), repairedSSTable);
Thread.sleep(2000);
for (int i = 10; i < 20; i++)
{
execute("delete from %s where id=?", i);
}
flush();
Thread.sleep(1000);
ReadCommand cmd = Util.cmd(getCurrentColumnFamilyStore()).build();
int partitionsFound = 0;
try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); UnfilteredPartitionIterator iterator = cmd.executeLocally(orderGroup))
{
while (iterator.hasNext())
{
partitionsFound++;
UnfilteredRowIterator rowIter = iterator.next();
int val = ByteBufferUtil.toInt(rowIter.partitionKey().getKey());
assertTrue("val=" + val, val >= 10 && val < 20);
}
}
assertEquals(10, partitionsFound);
}
@Test
public void readTestOldUnrepaired() throws Throwable
{
createTable("create table %s (id int, id2 int, t text, t2 text, primary key (id, id2)) with gc_grace_seconds=0 and compaction = {'class':'SizeTieredCompactionStrategy', 'only_purge_repaired_tombstones':true}");
getCurrentColumnFamilyStore().disableAutoCompaction();
for (int i = 0; i < 10; i++)
{
execute("delete from %s where id=1 and id2=?", i);
}
flush();
SSTableReader oldSSTable = getCurrentColumnFamilyStore().getLiveSSTables().iterator().next();
Thread.sleep(2000);
for (int i = 10; i < 20; i++)
{
execute("delete from %s where id=1 and id2=?", i);
}
flush();
for (SSTableReader sstable : getCurrentColumnFamilyStore().getLiveSSTables())
if (sstable != oldSSTable)
repair(getCurrentColumnFamilyStore(), sstable);
Thread.sleep(2000);
for (int i = 20; i < 30; i++)
{
execute("delete from %s where id=1 and id2=?", i);
}
flush();
Thread.sleep(2000);
// we will keep all tombstones since the oldest tombstones are unrepaired:
verify(30, 0, 30);
verify2(1, 30, 0, 30);
}
private void verify()
{
verify(10, 10, 20);
}
private void verify(int expectedRows, int minVal, int maxVal)
{
ReadCommand cmd = Util.cmd(getCurrentColumnFamilyStore()).build();
int foundRows = 0;
try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); UnfilteredPartitionIterator iterator = cmd.executeLocally(orderGroup))
{
while (iterator.hasNext())
{
UnfilteredRowIterator rowIter = iterator.next();
if (!rowIter.partitionKey().equals(Util.dk(ByteBufferUtil.bytes(999)))) // partition key 999 is 'live' and used to avoid sstables from being dropped
{
while (rowIter.hasNext())
{
AbstractRow row = (AbstractRow) rowIter.next();
for (int i = 0; i < row.clustering().size(); i++)
{
foundRows++;
int val = ByteBufferUtil.toInt(row.clustering().get(i));
assertTrue("val=" + val, val >= minVal && val < maxVal);
}
}
}
}
}
assertEquals(expectedRows, foundRows);
}
private void verify2(int key)
{
verify2(key, 10, 10, 20);
}
private void verify2(int key, int expectedRows, int minVal, int maxVal)
{
ReadCommand cmd = Util.cmd(getCurrentColumnFamilyStore(), Util.dk(ByteBufferUtil.bytes(key))).build();
int foundRows = 0;
try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); UnfilteredPartitionIterator iterator = cmd.executeLocally(orderGroup))
{
while (iterator.hasNext())
{
UnfilteredRowIterator rowIter = iterator.next();
while (rowIter.hasNext())
{
AbstractRow row = (AbstractRow) rowIter.next();
for (int i = 0; i < row.clustering().size(); i++)
{
foundRows++;
int val = ByteBufferUtil.toInt(row.clustering().get(i));
assertTrue("val=" + val, val >= minVal && val < maxVal);
}
}
}
}
assertEquals(expectedRows, foundRows);
}
public static void repair(ColumnFamilyStore cfs, SSTableReader sstable) throws IOException
{
sstable.descriptor.getMetadataSerializer().mutateRepairedAt(sstable.descriptor, 1);
sstable.reloadSSTableMetadata();
cfs.getTracker().notifySSTableRepairedStatusChanged(Collections.singleton(sstable));
}
}