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

More aggressive check for expired sstables in DTCS 

Patch by Björn Hegerfors; reviewed by marcuse for CASSANDRA-8359
This commit is contained in:
Björn Hegerfors 2015-04-01 12:26:00 +02:00 committed by Marcus Eriksson
parent 74bfa773a2
commit 3d0c4e78c6
3 changed files with 67 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
2.0.14:
* More aggressive check for expired sstables in DTCS (CASSANDRA-8359)
* Don't set clientMode to true when bulk-loading sstables to avoid
a NullPointerException (CASSANDRA-8238)
* Fix ignored index_interval change in ALTER TABLE statements (CASSANDRA-7976)

31
src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java
View File

@ -53,7 +53,7 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy
while (true)
{
List<SSTableReader> latestBucket = getNextBackgroundSStables(gcBefore);
List<SSTableReader> latestBucket = getNextBackgroundSSTables(gcBefore);
if (latestBucket.isEmpty())
return null;
@ -68,24 +68,39 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy
* @param gcBefore
* @return
*/
private List<SSTableReader> getNextBackgroundSStables(final int gcBefore)
private List<SSTableReader> getNextBackgroundSSTables(final int gcBefore)
{
if (!isEnabled() || cfs.getSSTables().isEmpty())
return Collections.emptyList();
Set<SSTableReader> uncompacting = cfs.getUncompactingSSTables();
// Find fully expired SSTables. Those will be included no matter what.
Set<SSTableReader> expired = CompactionController.getFullyExpiredSSTables(cfs, uncompacting, cfs.getOverlappingSSTables(uncompacting), gcBefore);
Set<SSTableReader> candidates = Sets.newHashSet(filterSuspectSSTables(uncompacting));
List<SSTableReader> compactionCandidates = new ArrayList<>(getNextNonExpiredSSTables(Sets.difference(candidates, expired), gcBefore));
if (!expired.isEmpty())
{
logger.debug("Including expired sstables: {}", expired);
compactionCandidates.addAll(expired);
}
return compactionCandidates;
}
private List<SSTableReader> getNextNonExpiredSSTables(Iterable<SSTableReader> nonExpiringSSTables, final int gcBefore)
{
int base = cfs.getMinimumCompactionThreshold();
long now = getNow();
Iterable<SSTableReader> candidates = filterSuspectSSTables(cfs.getUncompactingSSTables());
List<SSTableReader> mostInteresting = getCompactionCandidates(candidates, now, base);
List<SSTableReader> mostInteresting = getCompactionCandidates(nonExpiringSSTables, now, base);
if (mostInteresting != null)
return mostInteresting;
// if there is no sstable to compact in standard way, try compacting single sstable whose droppable tombstone
// ratio is greater than threshold.
List<SSTableReader> sstablesWithTombstones = Lists.newArrayList();
for (SSTableReader sstable : candidates)
for (SSTableReader sstable : nonExpiringSSTables)
{
if (worthDroppingTombstones(sstable, gcBefore))
sstablesWithTombstones.add(sstable);
@ -106,8 +121,8 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy
List<SSTableReader> mostInteresting = newestBucket(buckets,
cfs.getMinimumCompactionThreshold(),
cfs.getMaximumCompactionThreshold(),
options.baseTime,
now);
now,
options.baseTime);
if (!mostInteresting.isEmpty())
return mostInteresting;
return null;

43
test/unit/org/apache/cassandra/db/compaction/DateTieredCompactionStrategyTest.java
View File

@ -273,4 +273,47 @@ public class DateTieredCompactionStrategyTest extends SchemaLoader
filtered = filterOldSSTables(sstrs, 1, 4);
assertEquals("no sstables should remain when all are too old", 0, Iterables.size(filtered));
}
@Test
public void testDropExpiredSSTables() throws InterruptedException
{
Keyspace keyspace = Keyspace.open(KEYSPACE1);
ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF_STANDARD1);
cfs.truncateBlocking();
cfs.disableAutoCompaction();
ByteBuffer value = ByteBuffer.wrap(new byte[100]);
// create 2 sstables
DecoratedKey key = Util.dk(String.valueOf("expired"));
RowMutation rm = new RowMutation(KEYSPACE1, key.key);
rm.add(CF_STANDARD1, ByteBufferUtil.bytes("column"), value, System.currentTimeMillis(), 5);
rm.apply();
cfs.forceBlockingFlush();
SSTableReader expiredSSTable = cfs.getSSTables().iterator().next();
Thread.sleep(10);
key = Util.dk(String.valueOf("nonexpired"));
rm = new RowMutation(KEYSPACE1, key.key);
rm.add(CF_STANDARD1, ByteBufferUtil.bytes("column"), value, System.currentTimeMillis());
rm.apply();
cfs.forceBlockingFlush();
assertEquals(cfs.getSSTables().size(), 2);
Map<String, String> options = new HashMap<>();
options.put(DateTieredCompactionStrategyOptions.BASE_TIME_KEY, "30");
options.put(DateTieredCompactionStrategyOptions.TIMESTAMP_RESOLUTION_KEY, "MILLISECONDS");
options.put(DateTieredCompactionStrategyOptions.MAX_SSTABLE_AGE_KEY, Double.toString((1d / (24 * 60 * 60))));
DateTieredCompactionStrategy dtcs = new DateTieredCompactionStrategy(cfs, options);
dtcs.startup();
assertNull(dtcs.getNextBackgroundTask((int) (System.currentTimeMillis() / 1000)));
Thread.sleep(7000);
AbstractCompactionTask t = dtcs.getNextBackgroundTask((int) (System.currentTimeMillis()/1000));
assertNotNull(t);
assertEquals(1, Iterables.size(t.sstables));
SSTableReader sstable = t.sstables.iterator().next();
assertEquals(sstable, expiredSSTable);
}
}