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

Merge branch 'cassandra-2.2' into cassandra-3.0

This commit is contained in:
Marcus Eriksson 2015-10-28 09:04:30 +01:00
parent d856d3d7bb 5a356a7063
commit 27ca4915e9
3 changed files with 82 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -12,6 +12,7 @@ Merged from 2.2:
* Expose phi values from failure detector via JMX and tweak debug
and trace logging (CASSANDRA-9526)
Merged from 2.1:
* Do STCS in DTCS windows (CASSANDRA-10276)
* Avoid repetition of JVM_OPTS in debian package (CASSANDRA-10251)
* Fix potential NPE from handling result of SIM.highestSelectivityIndex (CASSANDRA-10550)
* Fix paging issues with partitions containing only static columns data (CASSANDRA-10381)

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

@ -18,7 +18,6 @@
package org.apache.cassandra.db.compaction;
import java.util.*;
import java.util.concurrent.TimeUnit;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Predicate;
@ -44,6 +43,7 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy
protected volatile int estimatedRemainingTasks;
private final Set<SSTableReader> sstables = new HashSet<>();
private long lastExpiredCheck;
private final SizeTieredCompactionStrategyOptions stcsOptions;
public DateTieredCompactionStrategy(ColumnFamilyStore cfs, Map<String, String> options)
{
@ -58,6 +58,7 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy
else
logger.trace("Enabling tombstone compactions for DTCS");
this.stcsOptions = new SizeTieredCompactionStrategyOptions(options);
}
@Override
@ -143,7 +144,8 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy
cfs.getMinimumCompactionThreshold(),
cfs.getMaximumCompactionThreshold(),
now,
options.baseTime);
options.baseTime,
stcsOptions);
if (!mostInteresting.isEmpty())
return mostInteresting;
return null;
@ -332,7 +334,7 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy
for (List<SSTableReader> bucket : tasks)
{
if (bucket.size() >= cfs.getMinimumCompactionThreshold())
n += Math.ceil((double)bucket.size() / cfs.getMaximumCompactionThreshold());
n += getSTCSBuckets(bucket, stcsOptions).size();
}
estimatedRemainingTasks = n;
}
@ -345,7 +347,7 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy
* @return a bucket (list) of sstables to compact.
*/
@VisibleForTesting
static List<SSTableReader> newestBucket(List<List<SSTableReader>> buckets, int minThreshold, int maxThreshold, long now, long baseTime)
static List<SSTableReader> newestBucket(List<List<SSTableReader>> buckets, int minThreshold, int maxThreshold, long now, long baseTime, SizeTieredCompactionStrategyOptions stcsOptions)
{
// If the "incoming window" has at least minThreshold SSTables, choose that one.
// For any other bucket, at least 2 SSTables is enough.
@ -353,23 +355,31 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy
Target incomingWindow = getInitialTarget(now, baseTime);
for (List<SSTableReader> bucket : buckets)
{
if (bucket.size() >= minThreshold ||
(bucket.size() >= 2 && !incomingWindow.onTarget(bucket.get(0).getMinTimestamp())))
return trimToThreshold(bucket, maxThreshold);
boolean inFirstWindow = incomingWindow.onTarget(bucket.get(0).getMinTimestamp());
if (bucket.size() >= minThreshold || (bucket.size() >= 2 && !inFirstWindow))
{
List<SSTableReader> stcsSSTables = getSSTablesForSTCS(bucket, inFirstWindow ? minThreshold : 2, maxThreshold, stcsOptions);
if (!stcsSSTables.isEmpty())
return stcsSSTables;
}
}
return Collections.emptyList();
}
/**
* @param bucket list of sstables, ordered from newest to oldest by getMinTimestamp().
* @param maxThreshold maximum number of sstables in a single compaction task.
* @return A bucket trimmed to the <code>maxThreshold</code> newest sstables.
*/
@VisibleForTesting
static List<SSTableReader> trimToThreshold(List<SSTableReader> bucket, int maxThreshold)
private static List<SSTableReader> getSSTablesForSTCS(Collection<SSTableReader> sstables, int minThreshold, int maxThreshold, SizeTieredCompactionStrategyOptions stcsOptions)
{
// Trim the oldest sstables off the end to meet the maxThreshold
return bucket.subList(0, Math.min(bucket.size(), maxThreshold));
List<SSTableReader> s = SizeTieredCompactionStrategy.mostInterestingBucket(getSTCSBuckets(sstables, stcsOptions), minThreshold, maxThreshold);
logger.debug("Got sstables {} for STCS from {}", s, sstables);
return s;
}
private static List<List<SSTableReader>> getSTCSBuckets(Collection<SSTableReader> sstables, SizeTieredCompactionStrategyOptions stcsOptions)
{
List<Pair<SSTableReader,Long>> pairs = SizeTieredCompactionStrategy.createSSTableAndLengthPairs(AbstractCompactionStrategy.filterSuspectSSTables(sstables));
return SizeTieredCompactionStrategy.getBuckets(pairs,
stcsOptions.bucketHigh,
stcsOptions.bucketLow,
stcsOptions.minSSTableSize);
}
@Override
@ -380,7 +390,7 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy
if (modifier == null)
return null;
return Arrays.<AbstractCompactionTask>asList(new CompactionTask(cfs, modifier, gcBefore));
return Collections.<AbstractCompactionTask>singleton(new CompactionTask(cfs, modifier, gcBefore));
}
@Override
@ -431,6 +441,8 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy
uncheckedOptions.remove(CompactionParams.Option.MIN_THRESHOLD.toString());
uncheckedOptions.remove(CompactionParams.Option.MAX_THRESHOLD.toString());
uncheckedOptions = SizeTieredCompactionStrategyOptions.validateOptions(options, uncheckedOptions);
return uncheckedOptions;
}

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

@ -33,6 +33,7 @@ import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.RowUpdateBuilder;
import org.apache.cassandra.db.lifecycle.SSTableSet;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.schema.KeyspaceParams;
@ -40,7 +41,6 @@ import org.apache.cassandra.utils.Pair;
import static org.apache.cassandra.db.compaction.DateTieredCompactionStrategy.getBuckets;
import static org.apache.cassandra.db.compaction.DateTieredCompactionStrategy.newestBucket;
import static org.apache.cassandra.db.compaction.DateTieredCompactionStrategy.trimToThreshold;
import static org.apache.cassandra.db.compaction.DateTieredCompactionStrategy.filterOldSSTables;
import static org.apache.cassandra.db.compaction.DateTieredCompactionStrategy.validateOptions;
@ -55,8 +55,8 @@ public class DateTieredCompactionStrategyTest extends SchemaLoader
{
SchemaLoader.prepareServer();
SchemaLoader.createKeyspace(KEYSPACE1,
KeyspaceParams.simple(1),
SchemaLoader.standardCFMD(KEYSPACE1, CF_STANDARD1));
KeyspaceParams.simple(1),
SchemaLoader.standardCFMD(KEYSPACE1, CF_STANDARD1));
}
@Test
@ -137,10 +137,10 @@ public class DateTieredCompactionStrategyTest extends SchemaLoader
public void testGetBuckets()
{
List<Pair<String, Long>> pairs = Lists.newArrayList(
Pair.create("a", 199L),
Pair.create("b", 299L),
Pair.create("a", 1L),
Pair.create("b", 201L)
Pair.create("a", 199L),
Pair.create("b", 299L),
Pair.create("a", 1L),
Pair.create("b", 201L)
);
List<List<String>> buckets = getBuckets(pairs, 100L, 2, 200L);
assertEquals(2, buckets.size());
@ -223,27 +223,15 @@ public class DateTieredCompactionStrategyTest extends SchemaLoader
List<SSTableReader> sstrs = new ArrayList<>(cfs.getLiveSSTables());
List<SSTableReader> newBucket = newestBucket(Collections.singletonList(sstrs.subList(0, 2)), 4, 32, 9, 10);
List<SSTableReader> newBucket = newestBucket(Collections.singletonList(sstrs.subList(0, 2)), 4, 32, 9, 10, new SizeTieredCompactionStrategyOptions());
assertTrue("incoming bucket should not be accepted when it has below the min threshold SSTables", newBucket.isEmpty());
newBucket = newestBucket(Collections.singletonList(sstrs.subList(0, 2)), 4, 32, 10, 10);
newBucket = newestBucket(Collections.singletonList(sstrs.subList(0, 2)), 4, 32, 10, 10, new SizeTieredCompactionStrategyOptions());
assertFalse("non-incoming bucket should be accepted when it has at least 2 SSTables", newBucket.isEmpty());
assertEquals("an sstable with a single value should have equal min/max timestamps", sstrs.get(0).getMinTimestamp(), sstrs.get(0).getMaxTimestamp());
assertEquals("an sstable with a single value should have equal min/max timestamps", sstrs.get(1).getMinTimestamp(), sstrs.get(1).getMaxTimestamp());
assertEquals("an sstable with a single value should have equal min/max timestamps", sstrs.get(2).getMinTimestamp(), sstrs.get(2).getMaxTimestamp());
// if we have more than the max threshold, the oldest should be dropped
Collections.sort(sstrs, Collections.reverseOrder(new Comparator<SSTableReader>() {
public int compare(SSTableReader o1, SSTableReader o2) {
return Long.compare(o1.getMinTimestamp(), o2.getMinTimestamp()) ;
}
}));
List<SSTableReader> bucket = trimToThreshold(sstrs, 2);
assertEquals("one bucket should have been dropped", 2, bucket.size());
for (SSTableReader sstr : bucket)
assertFalse("the oldest sstable should be dropped", sstr.getMinTimestamp() == 0);
}
@Test
@ -334,4 +322,47 @@ public class DateTieredCompactionStrategyTest extends SchemaLoader
t.transaction.abort();
}
@Test
public void testSTCSBigWindow()
{
Keyspace keyspace = Keyspace.open(KEYSPACE1);
ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF_STANDARD1);
cfs.truncateBlocking();
cfs.disableAutoCompaction();
ByteBuffer bigValue = ByteBuffer.wrap(new byte[10000]);
ByteBuffer value = ByteBuffer.wrap(new byte[100]);
int numSSTables = 40;
// create big sstabels out of half:
long timestamp = System.currentTimeMillis();
for (int r = 0; r < numSSTables / 2; r++)
{
for (int i = 0; i < 10; i++)
{
DecoratedKey key = Util.dk(String.valueOf(r));
new RowUpdateBuilder(cfs.metadata, timestamp, key.getKey())
.clustering("column")
.add("val", bigValue).build().applyUnsafe();
}
cfs.forceBlockingFlush();
}
// and small ones:
for (int r = 0; r < numSSTables / 2; r++)
{
DecoratedKey key = Util.dk(String.valueOf(r));
new RowUpdateBuilder(cfs.metadata, timestamp, key.getKey())
.clustering("column")
.add("val", value).build().applyUnsafe();
cfs.forceBlockingFlush();
}
Map<String, String> options = new HashMap<>();
options.put(SizeTieredCompactionStrategyOptions.MIN_SSTABLE_SIZE_KEY, "1");
DateTieredCompactionStrategy dtcs = new DateTieredCompactionStrategy(cfs, options);
for (SSTableReader sstable : cfs.getSSTables(SSTableSet.CANONICAL))
dtcs.addSSTable(sstable);
AbstractCompactionTask task = dtcs.getNextBackgroundTask(0);
assertEquals(20, task.transaction.originals().size());
task.transaction.abort();
}
}