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

Make sure LCS handles duplicate sstable added/removed notifications correctly. 

Patch by marcuse; reviewed by Blake Eggleston for CASSANDRA-14103
This commit is contained in:
Marcus Eriksson 2020-09-09 12:49:42 +02:00
parent 305e1e883b
commit f41ea9fb14
13 changed files with 1085 additions and 468 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.11.9
* Make sure LCS handles duplicate sstable added/removed notifications correctly (CASSANDRA-14103)
3.11.8
* Correctly interpret SASI's `max_compaction_flush_memory_in_mb` setting in megabytes not bytes (CASSANDRA-16071)

8
NEWS.txt
View File

@ -42,6 +42,14 @@ restore snapshots created with the previous major version using the
'sstableloader' tool. You can upgrade the file format of your snapshots
using the provided 'sstableupgrade' tool.
3.11.9
======
Upgrading
---------
- Custom compaction strategies must handle getting sstables added/removed notifications for
sstables already added/removed - see CASSANDRA-14103 for details. This has been a requirement
for correct operation since 3.11.0 due to an issue in CompactionStrategyManager.
3.11.7
======

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

@ -309,22 +309,36 @@ public abstract class AbstractCompactionStrategy
return getClass().getSimpleName();
}
/**
* Replaces sstables in the compaction strategy
*
* Note that implementations must be able to handle duplicate notifications here (that removed are already gone and
* added have already been added)
* */
public synchronized void replaceSSTables(Collection<SSTableReader> removed, Collection<SSTableReader> added)
{
for (SSTableReader remove : removed)
removeSSTable(remove);
for (SSTableReader add : added)
addSSTable(add);
addSSTables(added);
}
/**
* Adds sstable, note that implementations must handle duplicate notifications here (added already being in the compaction strategy)
*/
public abstract void addSSTable(SSTableReader added);
/**
* Adds sstables, note that implementations must handle duplicate notifications here (added already being in the compaction strategy)
*/
public synchronized void addSSTables(Iterable<SSTableReader> added)
{
for (SSTableReader sstable : added)
addSSTable(sstable);
}
/**
* Removes sstable from the strategy, implementations must be able to handle the sstable having already been removed.
*/
public abstract void removeSSTable(SSTableReader sstable);
public static class ScannerList implements AutoCloseable

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

@ -104,8 +104,10 @@ public class CompactionStrategyManager implements INotificationConsumer
we will use the new compaction parameters.
**/
private volatile CompactionParams schemaCompactionParams;
private boolean supportsEarlyOpen;
private int fanout;
private volatile boolean supportsEarlyOpen;
private volatile int fanout;
private volatile long maxSSTableSizeBytes;
private volatile String name;
public CompactionStrategyManager(ColumnFamilyStore cfs)
{
@ -217,6 +219,8 @@ public class CompactionStrategyManager implements INotificationConsumer
unrepaired.forEach(AbstractCompactionStrategy::startup);
supportsEarlyOpen = repaired.get(0).supportsEarlyOpen();
fanout = (repaired.get(0) instanceof LeveledCompactionStrategy) ? ((LeveledCompactionStrategy) repaired.get(0)).getLevelFanoutSize() : LeveledCompactionStrategy.DEFAULT_LEVEL_FANOUT_SIZE;
name = repaired.get(0).getName();
maxSSTableSizeBytes = repaired.get(0).getMaxSSTableBytes();
}
finally
{
@ -271,8 +275,7 @@ public class CompactionStrategyManager implements INotificationConsumer
* @param sstable
* @return
*/
@VisibleForTesting
protected int compactionStrategyIndexFor(SSTableReader sstable)
int compactionStrategyIndexFor(SSTableReader sstable)
{
// should not call maybeReload because it may be called from within lock
readLock.lock();
@ -340,18 +343,17 @@ public class CompactionStrategyManager implements INotificationConsumer
*/
//TODO improve this to reload after receiving a notification rather than trying to reload on every operation
@VisibleForTesting
protected boolean maybeReloadDiskBoundaries()
protected void maybeReloadDiskBoundaries()
{
if (!currentBoundaries.isOutOfDate())
return false;
return;
writeLock.lock();
try
{
if (!currentBoundaries.isOutOfDate())
return false;
return;
reload(params);
return true;
}
finally
{
@ -434,7 +436,7 @@ public class CompactionStrategyManager implements INotificationConsumer
{
if (repaired.get(0) instanceof LeveledCompactionStrategy && unrepaired.get(0) instanceof LeveledCompactionStrategy)
{
int[] res = new int[LeveledManifest.MAX_LEVEL_COUNT];
int[] res = new int[LeveledGenerations.MAX_LEVEL_COUNT];
for (AbstractCompactionStrategy strategy : repaired)
{
int[] repairedCountPerLevel = ((LeveledCompactionStrategy) strategy).getAllLevelSize();
@ -485,110 +487,115 @@ public class CompactionStrategyManager implements INotificationConsumer
}
}
/**
* Should only be called holding the readLock
*/
private void handleFlushNotification(Iterable<SSTableReader> added)
{
// If reloaded, SSTables will be placed in their correct locations
// so there is no need to process notification
if (maybeReloadDiskBoundaries())
return;
readLock.lock();
try
{
for (SSTableReader sstable : added)
compactionStrategyFor(sstable).addSSTable(sstable);
}
finally
{
readLock.unlock();
}
for (SSTableReader sstable : added)
compactionStrategyFor(sstable).addSSTable(sstable);
}
/**
* Should only be called holding the readLock
*/
private void handleListChangedNotification(Iterable<SSTableReader> added, Iterable<SSTableReader> removed)
{
// If reloaded, SSTables will be placed in their correct locations
// so there is no need to process notification
if (maybeReloadDiskBoundaries())
return;
// a bit of gymnastics to be able to replace sstables in compaction strategies
// we use this to know that a compaction finished and where to start the next compaction in LCS
int locationSize = partitionSSTablesByTokenRange? currentBoundaries.directories.size() : 1;
readLock.lock();
try
List<Set<SSTableReader>> repairedRemoved = new ArrayList<>(locationSize);
List<Set<SSTableReader>> repairedAdded = new ArrayList<>(locationSize);
List<Set<SSTableReader>> unrepairedRemoved = new ArrayList<>(locationSize);
List<Set<SSTableReader>> unrepairedAdded = new ArrayList<>(locationSize);
for (int i = 0; i < locationSize; i++)
{
// a bit of gymnastics to be able to replace sstables in compaction strategies
// we use this to know that a compaction finished and where to start the next compaction in LCS
int locationSize = partitionSSTablesByTokenRange? currentBoundaries.directories.size() : 1;
List<Set<SSTableReader>> repairedRemoved = new ArrayList<>(locationSize);
List<Set<SSTableReader>> repairedAdded = new ArrayList<>(locationSize);
List<Set<SSTableReader>> unrepairedRemoved = new ArrayList<>(locationSize);
List<Set<SSTableReader>> unrepairedAdded = new ArrayList<>(locationSize);
for (int i = 0; i < locationSize; i++)
{
repairedRemoved.add(new HashSet<>());
repairedAdded.add(new HashSet<>());
unrepairedRemoved.add(new HashSet<>());
unrepairedAdded.add(new HashSet<>());
}
for (SSTableReader sstable : removed)
{
int i = compactionStrategyIndexFor(sstable);
if (sstable.isRepaired())
repairedRemoved.get(i).add(sstable);
else
unrepairedRemoved.get(i).add(sstable);
}
for (SSTableReader sstable : added)
{
int i = compactionStrategyIndexFor(sstable);
if (sstable.isRepaired())
repairedAdded.get(i).add(sstable);
else
unrepairedAdded.get(i).add(sstable);
}
for (int i = 0; i < locationSize; i++)
{
if (!repairedRemoved.get(i).isEmpty())
repaired.get(i).replaceSSTables(repairedRemoved.get(i), repairedAdded.get(i));
else
repaired.get(i).addSSTables(repairedAdded.get(i));
if (!unrepairedRemoved.get(i).isEmpty())
unrepaired.get(i).replaceSSTables(unrepairedRemoved.get(i), unrepairedAdded.get(i));
else
unrepaired.get(i).addSSTables(unrepairedAdded.get(i));
}
repairedRemoved.add(new HashSet<>());
repairedAdded.add(new HashSet<>());
unrepairedRemoved.add(new HashSet<>());
unrepairedAdded.add(new HashSet<>());
}
finally
for (SSTableReader sstable : removed)
{
readLock.unlock();
int i = compactionStrategyIndexFor(sstable);
if (sstable.isRepaired())
repairedRemoved.get(i).add(sstable);
else
unrepairedRemoved.get(i).add(sstable);
}
for (SSTableReader sstable : added)
{
int i = compactionStrategyIndexFor(sstable);
if (sstable.isRepaired())
repairedAdded.get(i).add(sstable);
else
unrepairedAdded.get(i).add(sstable);
}
for (int i = 0; i < locationSize; i++)
{
if (!repairedRemoved.get(i).isEmpty())
repaired.get(i).replaceSSTables(repairedRemoved.get(i), repairedAdded.get(i));
else
repaired.get(i).addSSTables(repairedAdded.get(i));
if (!unrepairedRemoved.get(i).isEmpty())
unrepaired.get(i).replaceSSTables(unrepairedRemoved.get(i), unrepairedAdded.get(i));
else
unrepaired.get(i).addSSTables(unrepairedAdded.get(i));
}
}
private void handleRepairStatusChangedNotification(Iterable<SSTableReader> sstables)
{
// If reloaded, SSTables will be placed in their correct locations
// so there is no need to process notification
if (maybeReloadDiskBoundaries())
return;
// we need a write lock here since we move sstables from one strategy instance to another
for (SSTableReader sstable : sstables)
{
int index = compactionStrategyIndexFor(sstable);
if (sstable.isRepaired())
{
unrepaired.get(index).removeSSTable(sstable);
repaired.get(index).addSSTable(sstable);
}
else
{
repaired.get(index).removeSSTable(sstable);
unrepaired.get(index).addSSTable(sstable);
}
}
}
private void handleDeletingNotification(SSTableReader deleted)
{
compactionStrategyFor(deleted).removeSSTable(deleted);
}
public void handleNotification(INotification notification, Object sender)
{
// we might race with reload adding/removing the sstables, this means that compaction strategies
// must handle double notifications.
maybeReloadDiskBoundaries();
readLock.lock();
try
{
for (SSTableReader sstable : sstables)
if (notification instanceof SSTableAddedNotification)
{
int index = compactionStrategyIndexFor(sstable);
if (sstable.isRepaired())
{
unrepaired.get(index).removeSSTable(sstable);
repaired.get(index).addSSTable(sstable);
}
else
{
repaired.get(index).removeSSTable(sstable);
unrepaired.get(index).addSSTable(sstable);
}
handleFlushNotification(((SSTableAddedNotification) notification).added);
}
else if (notification instanceof SSTableListChangedNotification)
{
SSTableListChangedNotification listChangedNotification = (SSTableListChangedNotification) notification;
handleListChangedNotification(listChangedNotification.added, listChangedNotification.removed);
}
else if (notification instanceof SSTableRepairStatusChanged)
{
handleRepairStatusChangedNotification(((SSTableRepairStatusChanged) notification).sstables);
}
else if (notification instanceof SSTableDeletingNotification)
{
handleDeletingNotification(((SSTableDeletingNotification) notification).deleting);
}
}
finally
@ -597,44 +604,6 @@ public class CompactionStrategyManager implements INotificationConsumer
}
}
private void handleDeletingNotification(SSTableReader deleted)
{
// If reloaded, SSTables will be placed in their correct locations
// so there is no need to process notification
if (maybeReloadDiskBoundaries())
return;
readLock.lock();
try
{
compactionStrategyFor(deleted).removeSSTable(deleted);
}
finally
{
readLock.unlock();
}
}
public void handleNotification(INotification notification, Object sender)
{
if (notification instanceof SSTableAddedNotification)
{
handleFlushNotification(((SSTableAddedNotification) notification).added);
}
else if (notification instanceof SSTableListChangedNotification)
{
SSTableListChangedNotification listChangedNotification = (SSTableListChangedNotification) notification;
handleListChangedNotification(listChangedNotification.added, listChangedNotification.removed);
}
else if (notification instanceof SSTableRepairStatusChanged)
{
handleRepairStatusChangedNotification(((SSTableRepairStatusChanged) notification).sstables);
}
else if (notification instanceof SSTableDeletingNotification)
{
handleDeletingNotification(((SSTableDeletingNotification) notification).deleting);
}
}
public void enable()
{
writeLock.lock();
@ -750,15 +719,7 @@ public class CompactionStrategyManager implements INotificationConsumer
public long getMaxSSTableBytes()
{
readLock.lock();
try
{
return unrepaired.get(0).getMaxSSTableBytes();
}
finally
{
readLock.unlock();
}
return maxSSTableSizeBytes;
}
public AbstractCompactionTask getCompactionTask(LifecycleTransaction txn, int gcBefore, long maxSSTableBytes)
@ -925,16 +886,7 @@ public class CompactionStrategyManager implements INotificationConsumer
public String getName()
{
maybeReloadDiskBoundaries();
readLock.lock();
try
{
return unrepaired.get(0).getName();
}
finally
{
readLock.unlock();
}
return name;
}
public List<List<AbstractCompactionStrategy>> getStrategies()

30
src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java
View File

@ -161,7 +161,7 @@ public class LeveledCompactionStrategy extends AbstractCompactionStrategy
@SuppressWarnings("resource") // transaction is closed by AbstractCompactionTask::execute
public synchronized Collection<AbstractCompactionTask> getMaximalTask(int gcBefore, boolean splitOutput)
{
Iterable<SSTableReader> sstables = manifest.getAllSSTables();
Iterable<SSTableReader> sstables = manifest.getSSTables();
Iterable<SSTableReader> filteredSSTables = filterSuspectSSTables(sstables);
if (Iterables.isEmpty(sstables))
@ -339,10 +339,16 @@ public class LeveledCompactionStrategy extends AbstractCompactionStrategy
manifest.replace(removed, added);
}
@Override
public void addSSTables(Iterable<SSTableReader> sstables)
{
manifest.addSSTables(sstables);
}
@Override
public void addSSTable(SSTableReader added)
{
manifest.add(added);
manifest.addSSTables(Collections.singleton(added));
}
@Override
@ -493,21 +499,17 @@ public class LeveledCompactionStrategy extends AbstractCompactionStrategy
level:
for (int i = manifest.getLevelCount(); i >= 0; i--)
{
// sort sstables by droppable ratio in descending order
SortedSet<SSTableReader> sstables = manifest.getLevelSorted(i, new Comparator<SSTableReader>()
{
public int compare(SSTableReader o1, SSTableReader o2)
{
double r1 = o1.getEstimatedDroppableTombstoneRatio(gcBefore);
double r2 = o2.getEstimatedDroppableTombstoneRatio(gcBefore);
return -1 * Doubles.compare(r1, r2);
}
});
if (sstables.isEmpty())
if (manifest.getLevelSize(i) == 0)
continue;
// sort sstables by droppable ratio in descending order
List<SSTableReader> tombstoneSortedSSTables = manifest.getLevelSorted(i, (o1, o2) -> {
double r1 = o1.getEstimatedDroppableTombstoneRatio(gcBefore);
double r2 = o2.getEstimatedDroppableTombstoneRatio(gcBefore);
return -1 * Doubles.compare(r1, r2);
});
Set<SSTableReader> compacting = cfs.getTracker().getCompacting();
for (SSTableReader sstable : sstables)
for (SSTableReader sstable : tombstoneSortedSSTables)
{
if (sstable.getEstimatedDroppableTombstoneRatio(gcBefore) <= tombstoneThreshold)
continue level;

311
src/java/org/apache/cassandra/db/compaction/LeveledGenerations.java Normal file
View File

@ -0,0 +1,311 @@
/*
* 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.compaction;
import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.TimeUnit;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterators;
import com.google.common.collect.PeekingIterator;
import com.google.common.primitives.Ints;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.utils.FBUtilities;
/**
* Handles the leveled manifest generations
*
* Not thread safe, all access should be synchronized in LeveledManifest
*/
class LeveledGenerations
{
private static final Logger logger = LoggerFactory.getLogger(LeveledGenerations.class);
private final boolean strictLCSChecksTest = Boolean.getBoolean(Config.PROPERTY_PREFIX + "test.strict_lcs_checks");
// allocate enough generations for a PB of data, with a 1-MB sstable size. (Note that if maxSSTableSize is
// updated, we will still have sstables of the older, potentially smaller size. So don't make this
// dependent on maxSSTableSize.)
static final int MAX_LEVEL_COUNT = (int) Math.log10(1000 * 1000 * 1000);
private final Set<SSTableReader> l0 = new HashSet<>();
private static long lastOverlapCheck = System.nanoTime();
// note that since l0 is broken out, levels[0] represents L1:
private final TreeSet<SSTableReader> [] levels = new TreeSet[MAX_LEVEL_COUNT - 1];
private static final Comparator<SSTableReader> nonL0Comparator = (o1, o2) -> {
int cmp = SSTableReader.sstableComparator.compare(o1, o2);
if (cmp == 0)
cmp = Ints.compare(o1.descriptor.generation, o2.descriptor.generation);
return cmp;
};
LeveledGenerations()
{
for (int i = 0; i < MAX_LEVEL_COUNT - 1; i++)
levels[i] = new TreeSet<>(nonL0Comparator);
}
Set<SSTableReader> get(int level)
{
if (level > levelCount() - 1 || level < 0)
throw new ArrayIndexOutOfBoundsException("Invalid generation " + level + " - maximum is " + (levelCount() - 1));
if (level == 0)
return l0;
return levels[level - 1];
}
int levelCount()
{
return levels.length + 1;
}
/**
* Adds readers to the correct level
*
* If adding an sstable would cause an overlap in the level (if level > 1) we send it to L0. This can happen
* for example when moving sstables from unrepaired to repaired.
*
* If the sstable is already in the manifest we skip it.
*
* If the sstable exists in the manifest but has the wrong level, it is removed from the wrong level and added to the correct one
*
* todo: group sstables per level, add all if level is currently empty, improve startup speed
*/
void addAll(Iterable<SSTableReader> readers)
{
logDistribution();
for (SSTableReader sstable : readers)
{
assert sstable.getSSTableLevel() < levelCount() : "Invalid level " + sstable.getSSTableLevel() + " out of " + (levelCount() - 1);
int existingLevel = getLevelIfExists(sstable);
if (existingLevel != -1)
{
if (sstable.getSSTableLevel() != existingLevel)
{
logger.error("SSTable {} on the wrong level in the manifest - {} instead of {} as recorded in the sstable metadata, removing from level {}", sstable, existingLevel, sstable.getSSTableLevel(), existingLevel);
if (strictLCSChecksTest)
throw new AssertionError("SSTable not in matching level in manifest: "+sstable + ": "+existingLevel+" != " + sstable.getSSTableLevel());
get(existingLevel).remove(sstable);
}
else
{
logger.info("Manifest already contains {} in level {} - skipping", sstable, existingLevel);
continue;
}
}
if (sstable.getSSTableLevel() == 0)
{
l0.add(sstable);
continue;
}
TreeSet<SSTableReader> level = levels[sstable.getSSTableLevel() - 1];
/*
current level: |-----||----||----| |---||---|
new sstable: |--|
^ before
^ after
overlap if before.last >= newsstable.first or after.first <= newsstable.last
*/
SSTableReader after = level.ceiling(sstable);
SSTableReader before = level.floor(sstable);
if (before != null && before.last.compareTo(sstable.first) >= 0 ||
after != null && after.first.compareTo(sstable.last) <= 0)
{
if (strictLCSChecksTest) // we can only assert this in tests since this is normal when for example moving sstables from unrepaired to repaired
throw new AssertionError("Got unexpected overlap in level "+sstable.getSSTableLevel());
sendToL0(sstable);
}
else
{
level.add(sstable);
}
}
maybeVerifyLevels();
}
/**
* Sends sstable to L0 by mutating its level in the sstable metadata.
*
* SSTable should not exist in the manifest
*/
private void sendToL0(SSTableReader sstable)
{
try
{
sstable.descriptor.getMetadataSerializer().mutateLevel(sstable.descriptor, 0);
sstable.reloadSSTableMetadata();
}
catch (IOException e)
{
// Adding it to L0 and marking suspect is probably the best we can do here - it won't create overlap
// and we won't pick it for later compactions.
logger.error("Failed mutating sstable metadata for {} - adding it to L0 to avoid overlap. Marking suspect", sstable, e);
sstable.markSuspect();
}
l0.add(sstable);
}
/**
* Tries to find the sstable in the levels without using the sstable-recorded level
*
* Used to make sure we don't try to re-add an existing sstable
*/
private int getLevelIfExists(SSTableReader sstable)
{
for (int i = 0; i < levelCount(); i++)
{
if (get(i).contains(sstable))
return i;
}
return -1;
}
int remove(Collection<SSTableReader> readers)
{
int minLevel = Integer.MAX_VALUE;
for (SSTableReader sstable : readers)
{
int level = sstable.getSSTableLevel();
minLevel = Math.min(minLevel, level);
get(level).remove(sstable);
}
return minLevel;
}
int[] getAllLevelSize()
{
int[] counts = new int[levelCount()];
for (int i = 0; i < levelCount(); i++)
counts[i] = get(i).size();
return counts;
}
Set<SSTableReader> allSSTables()
{
ImmutableSet.Builder<SSTableReader> builder = ImmutableSet.builder();
builder.addAll(l0);
for (Set<SSTableReader> sstables : levels)
builder.addAll(sstables);
return builder.build();
}
/**
* given a level with sstables with first tokens [0, 10, 20, 30] and a lastCompactedSSTable with last = 15, we will
* return an Iterator over [20, 30, 0, 10].
*/
Iterator<SSTableReader> wrappingIterator(int lvl, SSTableReader lastCompactedSSTable)
{
assert lvl > 0; // only makes sense in L1+
TreeSet<SSTableReader> level = levels[lvl - 1];
if (level.isEmpty())
return Collections.emptyIterator();
if (lastCompactedSSTable == null)
return level.iterator();
PeekingIterator<SSTableReader> tail = Iterators.peekingIterator(level.tailSet(lastCompactedSSTable).iterator());
SSTableReader pivot = null;
// then we need to make sure that the first token of the pivot is greater than the last token of the lastCompactedSSTable
while (tail.hasNext())
{
SSTableReader potentialPivot = tail.peek();
if (potentialPivot.first.compareTo(lastCompactedSSTable.last) > 0)
{
pivot = potentialPivot;
break;
}
tail.next();
}
if (pivot == null)
return level.iterator();
return Iterators.concat(tail, level.headSet(pivot, false).iterator());
}
void logDistribution()
{
if (logger.isTraceEnabled())
{
for (int i = 0; i < levelCount(); i++)
{
Set<SSTableReader> level = get(i);
if (!level.isEmpty())
{
logger.trace("L{} contains {} SSTables ({}) in {}",
i,
level.size(),
FBUtilities.prettyPrintMemory(SSTableReader.getTotalBytes(level)),
this);
}
}
}
}
Set<SSTableReader>[] snapshot()
{
Set<SSTableReader> [] levelsCopy = new Set[levelCount()];
for (int i = 0; i < levelCount(); i++)
levelsCopy[i] = ImmutableSet.copyOf(get(i));
return levelsCopy;
}
/**
* do extra verification of the sstables in the generations
*
* only used during tests
*/
private void maybeVerifyLevels()
{
if (!strictLCSChecksTest || System.nanoTime() - lastOverlapCheck <= TimeUnit.NANOSECONDS.convert(5, TimeUnit.SECONDS))
return;
logger.info("LCS verifying levels");
lastOverlapCheck = System.nanoTime();
for (int i = 1; i < levelCount(); i++)
{
SSTableReader prev = null;
for (SSTableReader sstable : get(i))
{
// no overlap:
assert prev == null || prev.last.compareTo(sstable.first) < 0;
prev = sstable;
// make sure it does not exist in any other level:
for (int j = 0; j < levelCount(); j++)
{
if (i == j)
continue;
assert !get(j).contains(sstable);
}
}
}
}
}

342
src/java/org/apache/cassandra/db/compaction/LeveledManifest.java
View File

@ -17,19 +17,19 @@
*/
package org.apache.cassandra.db.compaction;
import java.io.IOException;
import java.util.*;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Predicate;
import com.google.common.base.Predicates;
import com.google.common.collect.ImmutableSortedSet;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import com.google.common.collect.Sets;
import com.google.common.primitives.Ints;
import org.apache.cassandra.db.PartitionPosition;
import org.apache.cassandra.io.sstable.Component;
import org.apache.cassandra.io.sstable.SSTable;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -40,9 +40,10 @@ import org.apache.cassandra.dht.Bounds;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
import static org.apache.cassandra.db.compaction.LeveledGenerations.MAX_LEVEL_COUNT;
public class LeveledManifest
{
private static final Logger logger = LoggerFactory.getLogger(LeveledManifest.class);
@ -59,14 +60,12 @@ public class LeveledManifest
* that level into lower level compactions
*/
private static final int NO_COMPACTION_LIMIT = 25;
// allocate enough generations for a PB of data, with a 1-MB sstable size. (Note that if maxSSTableSize is
// updated, we will still have sstables of the older, potentially smaller size. So don't make this
// dependent on maxSSTableSize.)
public static final int MAX_LEVEL_COUNT = (int) Math.log10(1000 * 1000 * 1000);
private final ColumnFamilyStore cfs;
@VisibleForTesting
protected final List<SSTableReader>[] generations;
private final PartitionPosition[] lastCompactedKeys;
private final LeveledGenerations generations;
private final SSTableReader[] lastCompactedSSTables;
private final long maxSSTableSizeInBytes;
private final SizeTieredCompactionStrategyOptions options;
private final int [] compactionCounter;
@ -79,13 +78,8 @@ public class LeveledManifest
this.options = options;
this.levelFanoutSize = fanoutSize;
generations = new List[MAX_LEVEL_COUNT];
lastCompactedKeys = new PartitionPosition[MAX_LEVEL_COUNT];
for (int i = 0; i < generations.length; i++)
{
generations[i] = new ArrayList<>();
lastCompactedKeys[i] = cfs.getPartitioner().getMinimumToken().minKeyBound();
}
lastCompactedSSTables = new SSTableReader[MAX_LEVEL_COUNT];
generations = new LeveledGenerations();
compactionCounter = new int[MAX_LEVEL_COUNT];
}
@ -99,14 +93,7 @@ public class LeveledManifest
LeveledManifest manifest = new LeveledManifest(cfs, maxSSTableSize, fanoutSize, options);
// ensure all SSTables are in the manifest
for (SSTableReader ssTableReader : sstables)
{
manifest.add(ssTableReader);
}
for (int i = 1; i < manifest.getAllLevelSize().length; i++)
{
manifest.repairOverlappingSSTables(i);
}
manifest.addSSTables(sstables);
manifest.calculateLastCompactedKeys();
return manifest;
}
@ -115,17 +102,18 @@ public class LeveledManifest
* If we want to start compaction in level n, find the newest (by modification time) file in level n+1
* and use its last token for last compacted key in level n;
*/
public void calculateLastCompactedKeys()
void calculateLastCompactedKeys()
{
for (int i = 0; i < generations.length - 1; i++)
for (int i = 0; i < generations.levelCount() - 1; i++)
{
Set<SSTableReader> level = generations.get(i + 1);
// this level is empty
if (generations[i + 1].isEmpty())
if (level.isEmpty())
continue;
SSTableReader sstableWithMaxModificationTime = null;
long maxModificationTime = Long.MIN_VALUE;
for (SSTableReader ssTableReader : generations[i + 1])
for (SSTableReader ssTableReader : level)
{
long modificationTime = ssTableReader.getCreationTimeFor(Component.DATA);
if (modificationTime >= maxModificationTime)
@ -135,80 +123,27 @@ public class LeveledManifest
}
}
lastCompactedKeys[i] = sstableWithMaxModificationTime.last;
lastCompactedSSTables[i] = sstableWithMaxModificationTime;
}
}
public synchronized void add(SSTableReader reader)
public synchronized void addSSTables(Iterable<SSTableReader> readers)
{
int level = reader.getSSTableLevel();
assert level < generations.length : "Invalid level " + level + " out of " + (generations.length - 1);
logDistribution();
if (canAddSSTable(reader))
{
// adding the sstable does not cause overlap in the level
logger.trace("Adding {} to L{}", reader, level);
generations[level].add(reader);
}
else
{
// this can happen if:
// * a compaction has promoted an overlapping sstable to the given level, or
// was also supposed to add an sstable at the given level.
// * we are moving sstables from unrepaired to repaired and the sstable
// would cause overlap
//
// The add(..):ed sstable will be sent to level 0
try
{
reader.descriptor.getMetadataSerializer().mutateLevel(reader.descriptor, 0);
reader.reloadSSTableMetadata();
}
catch (IOException e)
{
logger.error("Could not change sstable level - adding it at level 0 anyway, we will find it at restart.", e);
}
if (!contains(reader))
{
generations[0].add(reader);
}
else
{
// An SSTable being added multiple times to this manifest indicates a programming error, but we don't
// throw an AssertionError because this shouldn't break the compaction strategy. Instead we log it
// together with a RuntimeException so the stack is print for troubleshooting if this ever happens.
logger.warn("SSTable {} is already present on leveled manifest and should not be re-added.", reader, new RuntimeException());
}
}
}
private boolean contains(SSTableReader reader)
{
for (int i = 0; i < generations.length; i++)
{
if (generations[i].contains(reader))
return true;
}
return false;
generations.addAll(readers);
}
public synchronized void replace(Collection<SSTableReader> removed, Collection<SSTableReader> added)
{
assert !removed.isEmpty(); // use add() instead of promote when adding new sstables
logDistribution();
if (logger.isTraceEnabled())
{
generations.logDistribution();
logger.trace("Replacing [{}]", toString(removed));
}
// the level for the added sstables is the max of the removed ones,
// plus one if the removed were all on the same level
int minLevel = Integer.MAX_VALUE;
for (SSTableReader sstable : removed)
{
int thisLevel = remove(sstable);
minLevel = Math.min(minLevel, thisLevel);
}
int minLevel = generations.remove(removed);
// it's valid to do a remove w/o an add (e.g. on truncate)
if (added.isEmpty())
@ -217,76 +152,8 @@ public class LeveledManifest
if (logger.isTraceEnabled())
logger.trace("Adding [{}]", toString(added));
for (SSTableReader ssTableReader : added)
add(ssTableReader);
lastCompactedKeys[minLevel] = SSTableReader.sstableOrdering.max(added).last;
}
public synchronized void repairOverlappingSSTables(int level)
{
SSTableReader previous = null;
Collections.sort(generations[level], SSTableReader.sstableComparator);
List<SSTableReader> outOfOrderSSTables = new ArrayList<>();
for (SSTableReader current : generations[level])
{
if (previous != null && current.first.compareTo(previous.last) <= 0)
{
logger.warn("At level {}, {} [{}, {}] overlaps {} [{}, {}]. This could be caused by a bug in Cassandra 1.1.0 .. 1.1.3 or due to the fact that you have dropped sstables from another node into the data directory. " +
"Sending back to L0. If you didn't drop in sstables, and have not yet run scrub, you should do so since you may also have rows out-of-order within an sstable",
level, previous, previous.first, previous.last, current, current.first, current.last);
outOfOrderSSTables.add(current);
}
else
{
previous = current;
}
}
if (!outOfOrderSSTables.isEmpty())
{
for (SSTableReader sstable : outOfOrderSSTables)
sendBackToL0(sstable);
}
}
/**
* Checks if adding the sstable creates an overlap in the level
* @param sstable the sstable to add
* @return true if it is safe to add the sstable in the level.
*/
private boolean canAddSSTable(SSTableReader sstable)
{
int level = sstable.getSSTableLevel();
if (level == 0)
return true;
List<SSTableReader> copyLevel = new ArrayList<>(generations[level]);
copyLevel.add(sstable);
Collections.sort(copyLevel, SSTableReader.sstableComparator);
SSTableReader previous = null;
for (SSTableReader current : copyLevel)
{
if (previous != null && current.first.compareTo(previous.last) <= 0)
return false;
previous = current;
}
return true;
}
private synchronized void sendBackToL0(SSTableReader sstable)
{
remove(sstable);
try
{
sstable.descriptor.getMetadataSerializer().mutateLevel(sstable.descriptor, 0);
sstable.reloadSSTableMetadata();
add(sstable);
}
catch (IOException e)
{
throw new RuntimeException("Could not reload sstable meta data", e);
}
generations.addAll(added);
lastCompactedSSTables[minLevel] = SSTableReader.sstableOrdering.max(added);
}
private String toString(Collection<SSTableReader> sstables)
@ -329,7 +196,7 @@ public class LeveledManifest
// the streamed files can be placed in their original levels
if (StorageService.instance.isBootstrapMode())
{
List<SSTableReader> mostInteresting = getSSTablesForSTCS(getLevel(0));
List<SSTableReader> mostInteresting = getSSTablesForSTCS(generations.get(0));
if (!mostInteresting.isEmpty())
{
logger.info("Bootstrapping - doing STCS in L0");
@ -364,9 +231,9 @@ public class LeveledManifest
// This isn't a magic wand -- if you are consistently writing too fast for LCS to keep
// up, you're still screwed. But if instead you have intermittent bursts of activity,
// it can help a lot.
for (int i = generations.length - 1; i > 0; i--)
for (int i = generations.levelCount() - 1; i > 0; i--)
{
List<SSTableReader> sstables = getLevel(i);
Set<SSTableReader> sstables = generations.get(i);
if (sstables.isEmpty())
continue; // mostly this just avoids polluting the debug log with zero scores
// we want to calculate score excluding compacting ones
@ -400,7 +267,7 @@ public class LeveledManifest
}
// Higher levels are happy, time for a standard, non-STCS L0 compaction
if (getLevel(0).isEmpty())
if (generations.get(0).isEmpty())
return null;
Collection<SSTableReader> candidates = getCandidatesFor(0);
if (candidates.isEmpty())
@ -415,9 +282,9 @@ public class LeveledManifest
private CompactionCandidate getSTCSInL0CompactionCandidate()
{
if (!DatabaseDescriptor.getDisableSTCSInL0() && getLevel(0).size() > MAX_COMPACTING_L0)
if (!DatabaseDescriptor.getDisableSTCSInL0() && generations.get(0).size() > MAX_COMPACTING_L0)
{
List<SSTableReader> mostInteresting = getSSTablesForSTCS(getLevel(0));
List<SSTableReader> mostInteresting = getSSTablesForSTCS(generations.get(0));
if (!mostInteresting.isEmpty())
{
logger.debug("L0 is too far behind, performing size-tiering there first");
@ -454,7 +321,7 @@ public class LeveledManifest
{
Set<SSTableReader> withStarvedCandidate = new HashSet<>(candidates);
for (int i = generations.length - 1; i > 0; i--)
for (int i = generations.levelCount() - 1; i > 0; i--)
compactionCounter[i]++;
compactionCounter[targetLevel] = 0;
if (logger.isTraceEnabled())
@ -463,7 +330,7 @@ public class LeveledManifest
logger.trace("CompactionCounter: {}: {}", j, compactionCounter[j]);
}
for (int i = generations.length - 1; i > 0; i--)
for (int i = generations.levelCount() - 1; i > 0; i--)
{
if (getLevelSize(i) > 0)
{
@ -486,9 +353,9 @@ public class LeveledManifest
return candidates;
Set<SSTableReader> compacting = cfs.getTracker().getCompacting();
Range<PartitionPosition> boundaries = new Range<>(min, max);
for (SSTableReader sstable : getLevel(i))
for (SSTableReader sstable : generations.get(i))
{
Range<PartitionPosition> r = new Range<PartitionPosition>(sstable.first, sstable.last);
Range<PartitionPosition> r = new Range<>(sstable.first, sstable.last);
if (boundaries.contains(r) && !compacting.contains(sstable))
{
logger.info("Adding high-level (L{}) {} to candidates", sstable.getSSTableLevel(), sstable);
@ -506,35 +373,12 @@ public class LeveledManifest
public synchronized int getLevelSize(int i)
{
if (i >= generations.length)
throw new ArrayIndexOutOfBoundsException("Maximum valid generation is " + (generations.length - 1));
return getLevel(i).size();
return generations.get(i).size();
}
public synchronized int[] getAllLevelSize()
{
int[] counts = new int[generations.length];
for (int i = 0; i < counts.length; i++)
counts[i] = getLevel(i).size();
return counts;
}
private void logDistribution()
{
if (logger.isTraceEnabled())
{
for (int i = 0; i < generations.length; i++)
{
if (!getLevel(i).isEmpty())
{
logger.trace("L{} contains {} SSTables ({}) in {}",
i,
getLevel(i).size(),
FBUtilities.prettyPrintMemory(SSTableReader.getTotalBytes(getLevel(i))),
this);
}
}
}
return generations.getAllLevelSize();
}
@VisibleForTesting
@ -542,10 +386,15 @@ public class LeveledManifest
{
int level = reader.getSSTableLevel();
assert level >= 0 : reader + " not present in manifest: "+level;
generations[level].remove(reader);
generations.remove(Collections.singleton(reader));
return level;
}
public synchronized Set<SSTableReader> getSSTables()
{
return generations.allSSTables();
}
private static Set<SSTableReader> overlapping(Collection<SSTableReader> candidates, Iterable<SSTableReader> others)
{
assert !candidates.isEmpty();
@ -591,7 +440,7 @@ public class LeveledManifest
{
assert start.compareTo(end) <= 0;
Set<SSTableReader> overlapped = new HashSet<>();
Bounds<Token> promotedBounds = new Bounds<Token>(start, end);
Bounds<Token> promotedBounds = new Bounds<>(start, end);
for (Map.Entry<SSTableReader, Bounds<Token>> pair : sstables.entrySet())
{
@ -601,20 +450,12 @@ public class LeveledManifest
return overlapped;
}
private static final Predicate<SSTableReader> suspectP = new Predicate<SSTableReader>()
{
public boolean apply(SSTableReader candidate)
{
return candidate.isMarkedSuspect();
}
};
private static Map<SSTableReader, Bounds<Token>> genBounds(Iterable<SSTableReader> ssTableReaders)
{
Map<SSTableReader, Bounds<Token>> boundsMap = new HashMap<>();
for (SSTableReader sstable : ssTableReaders)
{
boundsMap.put(sstable, new Bounds<Token>(sstable.first.getToken(), sstable.last.getToken()));
boundsMap.put(sstable, new Bounds<>(sstable.first.getToken(), sstable.last.getToken()));
}
return boundsMap;
}
@ -626,14 +467,14 @@ public class LeveledManifest
*/
private Collection<SSTableReader> getCandidatesFor(int level)
{
assert !getLevel(level).isEmpty();
assert !generations.get(level).isEmpty();
logger.trace("Choosing candidates for L{}", level);
final Set<SSTableReader> compacting = cfs.getTracker().getCompacting();
if (level == 0)
{
Set<SSTableReader> compactingL0 = getCompacting(0);
Set<SSTableReader> compactingL0 = getCompactingL0();
PartitionPosition lastCompactingKey = null;
PartitionPosition firstCompactingKey = null;
@ -659,7 +500,7 @@ public class LeveledManifest
// basically screwed, since we expect all or most L0 sstables to overlap with each L1 sstable.
// So if an L1 sstable is suspect we can't do much besides try anyway and hope for the best.
Set<SSTableReader> candidates = new HashSet<>();
Map<SSTableReader, Bounds<Token>> remaining = genBounds(Iterables.filter(getLevel(0), Predicates.not(suspectP)));
Map<SSTableReader, Bounds<Token>> remaining = genBounds(Iterables.filter(generations.get(0), Predicates.not(SSTableReader::isMarkedSuspect)));
for (SSTableReader sstable : ageSortedSSTables(remaining.keySet()))
{
@ -672,7 +513,7 @@ public class LeveledManifest
for (SSTableReader newCandidate : overlappedL0)
{
if (firstCompactingKey == null || lastCompactingKey == null || overlapping(firstCompactingKey.getToken(), lastCompactingKey.getToken(), Arrays.asList(newCandidate)).size() == 0)
if (firstCompactingKey == null || lastCompactingKey == null || overlapping(firstCompactingKey.getToken(), lastCompactingKey.getToken(), Collections.singleton(newCandidate)).size() == 0)
candidates.add(newCandidate);
remaining.remove(newCandidate);
}
@ -691,7 +532,7 @@ public class LeveledManifest
// add sstables from L1 that overlap candidates
// if the overlapping ones are already busy in a compaction, leave it out.
// TODO try to find a set of L0 sstables that only overlaps with non-busy L1 sstables
Set<SSTableReader> l1overlapping = overlapping(candidates, getLevel(1));
Set<SSTableReader> l1overlapping = overlapping(candidates, generations.get(1));
if (Sets.intersection(l1overlapping, compacting).size() > 0)
return Collections.emptyList();
if (!overlapping(candidates, compactingL0).isEmpty())
@ -704,27 +545,16 @@ public class LeveledManifest
return candidates;
}
// for non-L0 compactions, pick up where we left off last time
Collections.sort(getLevel(level), SSTableReader.sstableComparator);
int start = 0; // handles case where the prior compaction touched the very last range
for (int i = 0; i < getLevel(level).size(); i++)
{
SSTableReader sstable = getLevel(level).get(i);
if (sstable.first.compareTo(lastCompactedKeys[level]) > 0)
{
start = i;
break;
}
}
// look for a non-suspect keyspace to compact with, starting with where we left off last time,
// and wrapping back to the beginning of the generation if necessary
Map<SSTableReader, Bounds<Token>> sstablesNextLevel = genBounds(getLevel(level + 1));
for (int i = 0; i < getLevel(level).size(); i++)
Map<SSTableReader, Bounds<Token>> sstablesNextLevel = genBounds(generations.get(level + 1));
Iterator<SSTableReader> levelIterator = generations.wrappingIterator(level, lastCompactedSSTables[level]);
while (levelIterator.hasNext())
{
SSTableReader sstable = getLevel(level).get((start + i) % getLevel(level).size());
SSTableReader sstable = levelIterator.next();
Set<SSTableReader> candidates = Sets.union(Collections.singleton(sstable), overlappingWithBounds(sstable, sstablesNextLevel));
if (Iterables.any(candidates, suspectP))
if (Iterables.any(candidates, SSTableReader::isMarkedSuspect))
continue;
if (Sets.intersection(candidates, compacting).isEmpty())
return candidates;
@ -734,10 +564,10 @@ public class LeveledManifest
return Collections.emptyList();
}
private Set<SSTableReader> getCompacting(int level)
private Set<SSTableReader> getCompactingL0()
{
Set<SSTableReader> sstables = new HashSet<>();
Set<SSTableReader> levelSSTables = new HashSet<>(getLevel(level));
Set<SSTableReader> levelSSTables = new HashSet<>(generations.get(0));
for (SSTableReader sstable : cfs.getTracker().getCompacting())
{
if (levelSSTables.contains(sstable))
@ -749,19 +579,14 @@ public class LeveledManifest
@VisibleForTesting
List<SSTableReader> ageSortedSSTables(Collection<SSTableReader> candidates)
{
List<SSTableReader> ageSortedCandidates = new ArrayList<>(candidates);
Collections.sort(ageSortedCandidates, SSTableReader.maxTimestampAscending);
return ageSortedCandidates;
List<SSTableReader> copy = new ArrayList<>(candidates);
copy.sort(SSTableReader.maxTimestampAscending);
return ImmutableList.copyOf(copy);
}
public synchronized Set<SSTableReader>[] getSStablesPerLevelSnapshot()
{
Set<SSTableReader>[] sstablesPerLevel = new Set[generations.length];
for (int i = 0; i < generations.length; i++)
{
sstablesPerLevel[i] = new HashSet<>(generations[i]);
}
return sstablesPerLevel;
return generations.snapshot();
}
@Override
@ -770,34 +595,24 @@ public class LeveledManifest
return "Manifest@" + hashCode();
}
public int getLevelCount()
public synchronized int getLevelCount()
{
for (int i = generations.length - 1; i >= 0; i--)
for (int i = generations.levelCount() - 1; i >= 0; i--)
{
if (getLevel(i).size() > 0)
if (generations.get(i).size() > 0)
return i;
}
return 0;
}
public synchronized SortedSet<SSTableReader> getLevelSorted(int level, Comparator<SSTableReader> comparator)
{
return ImmutableSortedSet.copyOf(comparator, getLevel(level));
}
public List<SSTableReader> getLevel(int i)
{
return generations[i];
}
public synchronized int getEstimatedTasks()
{
long tasks = 0;
long[] estimated = new long[generations.length];
long[] estimated = new long[generations.levelCount()];
for (int i = generations.length - 1; i >= 0; i--)
for (int i = generations.levelCount() - 1; i >= 0; i--)
{
List<SSTableReader> sstables = getLevel(i);
Set<SSTableReader> sstables = generations.get(i);
// If there is 1 byte over TBL - (MBL * 1.001), there is still a task left, so we need to round up.
estimated[i] = (long)Math.ceil((double)Math.max(0L, SSTableReader.getTotalBytes(sstables) - (long)(maxBytesForLevel(i, maxSSTableSizeInBytes) * 1.001)) / (double)maxSSTableSizeInBytes);
tasks += estimated[i];
@ -829,17 +644,18 @@ public class LeveledManifest
assert newLevel > 0;
}
return newLevel;
}
public Iterable<SSTableReader> getAllSSTables()
synchronized Set<SSTableReader> getLevel(int level)
{
Set<SSTableReader> sstables = new HashSet<>();
for (List<SSTableReader> generation : generations)
{
sstables.addAll(generation);
}
return sstables;
return ImmutableSet.copyOf(generations.get(level));
}
synchronized List<SSTableReader> getLevelSorted(int level, Comparator<SSTableReader> comparator)
{
List<SSTableReader> copy = new ArrayList<>(generations.get(level));
copy.sort(comparator);
return ImmutableList.copyOf(copy);
}
public static class CompactionCandidate

31
src/java/org/apache/cassandra/tools/StandaloneScrubber.java
View File

@ -245,30 +245,19 @@ public class StandaloneScrubber
if (strategyManager.getCompactionParams().klass().equals(LeveledCompactionStrategy.class))
{
int maxSizeInMB = (int)((cfs.getCompactionStrategyManager().getMaxSSTableBytes()) / (1024L * 1024L));
int fanOut = cfs.getCompactionStrategyManager().getLevelFanoutSize();
List<SSTableReader> repaired = new ArrayList<>();
List<SSTableReader> unrepaired = new ArrayList<>();
System.out.println("Checking leveled manifest");
Predicate<SSTableReader> repairedPredicate = new Predicate<SSTableReader>()
for (SSTableReader sstable : sstables)
{
@Override
public boolean apply(SSTableReader sstable)
{
return sstable.isRepaired();
}
};
List<SSTableReader> repaired = Lists.newArrayList(Iterables.filter(sstables, repairedPredicate));
List<SSTableReader> unRepaired = Lists.newArrayList(Iterables.filter(sstables, Predicates.not(repairedPredicate)));
LeveledManifest repairedManifest = LeveledManifest.create(cfs, maxSizeInMB, cfs.getLevelFanoutSize(), repaired);
for (int i = 1; i < repairedManifest.getLevelCount(); i++)
{
repairedManifest.repairOverlappingSSTables(i);
}
LeveledManifest unRepairedManifest = LeveledManifest.create(cfs, maxSizeInMB, cfs.getLevelFanoutSize(), unRepaired);
for (int i = 1; i < unRepairedManifest.getLevelCount(); i++)
{
unRepairedManifest.repairOverlappingSSTables(i);
if (sstable.isRepaired())
repaired.add(sstable);
else
unrepaired.add(sstable);
}
LeveledManifest.create(cfs, maxSizeInMB, fanOut, repaired);
LeveledManifest.create(cfs, maxSizeInMB, fanOut, unrepaired);
}
}

92
test/long/org/apache/cassandra/db/compaction/LongLeveledCompactionStrategyCQLTest.java Normal file
View File

@ -0,0 +1,92 @@
/*
* 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.compaction;
import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import com.google.common.util.concurrent.Uninterruptibles;
import org.junit.Test;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.Hex;
public class LongLeveledCompactionStrategyCQLTest extends CQLTester
{
@Test
public void stressTestCompactionStrategyManager() throws ExecutionException, InterruptedException
{
System.setProperty(Config.PROPERTY_PREFIX + "test.strict_lcs_checks", "true");
// flush/compact tons of sstables, invalidate token metadata in a loop to make CSM reload the strategies
createTable("create table %s (id int primary key, i text) with compaction = {'class':'LeveledCompactionStrategy', 'sstable_size_in_mb':1}");
ExecutorService es = Executors.newSingleThreadExecutor();
DatabaseDescriptor.setConcurrentCompactors(8);
AtomicBoolean stop = new AtomicBoolean(false);
long start = System.currentTimeMillis();
try
{
Random r = new Random();
Future<?> writes = es.submit(() -> {
byte[] b = new byte[1024];
while (!stop.get())
{
for (int i = 0 ; i < 100; i++)
{
try
{
r.nextBytes(b);
String s = Hex.bytesToHex(b);
execute("insert into %s (id, i) values (?,?)", r.nextInt(), s);
}
catch (Throwable throwable)
{
throw new RuntimeException(throwable);
}
}
getCurrentColumnFamilyStore().forceBlockingFlush();
Uninterruptibles.sleepUninterruptibly(r.nextInt(200), TimeUnit.MILLISECONDS);
}
});
while(System.currentTimeMillis() - start < TimeUnit.MILLISECONDS.convert(5, TimeUnit.MINUTES))
{
StorageService.instance.getTokenMetadata().invalidateCachedRings();
Uninterruptibles.sleepUninterruptibly(r.nextInt(1000), TimeUnit.MILLISECONDS);
}
stop.set(true);
writes.get();
}
finally
{
es.shutdown();
}
}
}

2
test/long/org/apache/cassandra/db/compaction/LongLeveledCompactionStrategyTest.java
View File

@ -128,7 +128,7 @@ public class LongLeveledCompactionStrategyTest
int levels = manifest.getLevelCount();
for (int level = 0; level < levels; level++)
{
List<SSTableReader> sstables = manifest.getLevel(level);
Set<SSTableReader> sstables = manifest.getLevel(level);
// score check
assert (double) SSTableReader.getTotalBytes(sstables) / manifest.maxBytesForLevel(level, 1 * 1024 * 1024) < 1.00;
// overlap check for levels greater than 0

27
test/unit/org/apache/cassandra/MockSchema.java
View File

@ -48,6 +48,8 @@ import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.utils.AlwaysPresentFilter;
import org.apache.cassandra.utils.ByteBufferUtil;
import static org.apache.cassandra.service.ActiveRepairService.UNREPAIRED_SSTABLE;
public class MockSchema
{
static
@ -83,6 +85,21 @@ public class MockSchema
}
public static SSTableReader sstable(int generation, int size, boolean keepRef, ColumnFamilyStore cfs)
{
return sstable(generation, size, keepRef, generation, generation, cfs);
}
public static SSTableReader sstableWithLevel(int generation, long firstToken, long lastToken, int level, ColumnFamilyStore cfs)
{
return sstable(generation, 0, false, firstToken, lastToken, level, cfs);
}
public static SSTableReader sstable(int generation, int size, boolean keepRef, long firstToken, long lastToken, ColumnFamilyStore cfs)
{
return sstable(generation, size, keepRef, firstToken, lastToken, 0, cfs);
}
public static SSTableReader sstable(int generation, int size, boolean keepRef, long firstToken, long lastToken, int level, ColumnFamilyStore cfs)
{
Descriptor descriptor = new Descriptor(cfs.getDirectories().getDirectoryForNewSSTables(),
cfs.keyspace.getName(),
@ -117,12 +134,14 @@ public class MockSchema
}
SerializationHeader header = SerializationHeader.make(cfs.metadata, Collections.emptyList());
StatsMetadata metadata = (StatsMetadata) new MetadataCollector(cfs.metadata.comparator)
.finalizeMetadata(cfs.metadata.partitioner.getClass().getCanonicalName(), 0.01f, -1, header)
.sstableLevel(level)
.finalizeMetadata(cfs.metadata.partitioner.getClass().getCanonicalName(), 0.01f, UNREPAIRED_SSTABLE, header)
.get(MetadataType.STATS);
SSTableReader reader = SSTableReader.internalOpen(descriptor, components, cfs.metadata,
RANDOM_ACCESS_READER_FACTORY.sharedCopy(), RANDOM_ACCESS_READER_FACTORY.sharedCopy(), indexSummary.sharedCopy(),
new AlwaysPresentFilter(), 1L, metadata, SSTableReader.OpenReason.NORMAL, header);
reader.first = reader.last = readerBounds(generation);
reader.first = readerBounds(firstToken);
reader.last = readerBounds(lastToken);
if (!keepRef)
reader.selfRef().release();
return reader;
@ -152,9 +171,9 @@ public class MockSchema
return metadata;
}
public static BufferDecoratedKey readerBounds(int generation)
public static BufferDecoratedKey readerBounds(long token)
{
return new BufferDecoratedKey(new Murmur3Partitioner.LongToken(generation), ByteBufferUtil.EMPTY_BYTE_BUFFER);
return new BufferDecoratedKey(new Murmur3Partitioner.LongToken(token), ByteBufferUtil.EMPTY_BYTE_BUFFER);
}
private static File temp(String id)

238
test/unit/org/apache/cassandra/db/compaction/LeveledCompactionStrategyTest.java
View File

@ -22,10 +22,14 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.UUID;
import java.util.stream.Collectors;
import junit.framework.Assert;
import org.junit.After;
@ -37,6 +41,7 @@ import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.MockSchema;
import org.apache.cassandra.OrderedJUnit4ClassRunner;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.Util;
@ -157,7 +162,6 @@ public class LeveledCompactionStrategyTest
assert groupLevel == tableLevel;
}
}
}
/*
@ -296,7 +300,7 @@ public class LeveledCompactionStrategyTest
strategy.manifest.remove(s);
s.descriptor.getMetadataSerializer().mutateLevel(s.descriptor, 6);
s.reloadSSTableMetadata();
strategy.manifest.add(s);
strategy.manifest.addSSTables(Collections.singleton(s));
}
// verify that all sstables in the changed set is level 6
for (SSTableReader s : cfs.getLiveSSTables())
@ -345,14 +349,12 @@ public class LeveledCompactionStrategyTest
for (SSTableReader sstable : cfs.getLiveSSTables())
assertFalse(sstable.isRepaired());
int sstableCount = 0;
for (List<SSTableReader> level : unrepaired.manifest.generations)
sstableCount += level.size();
int sstableCount = unrepaired.manifest.getSSTables().size();
// we only have unrepaired sstables:
assertEquals(sstableCount, cfs.getLiveSSTables().size());
SSTableReader sstable1 = unrepaired.manifest.generations[2].get(0);
SSTableReader sstable2 = unrepaired.manifest.generations[1].get(0);
SSTableReader sstable1 = unrepaired.manifest.getLevel(2).iterator().next();
SSTableReader sstable2 = unrepaired.manifest.getLevel(1).iterator().next();
sstable1.descriptor.getMetadataSerializer().mutateRepairedAt(sstable1.descriptor, System.currentTimeMillis());
sstable1.reloadSSTableMetadata();
@ -360,14 +362,12 @@ public class LeveledCompactionStrategyTest
manager.handleNotification(new SSTableRepairStatusChanged(Arrays.asList(sstable1)), this);
int repairedSSTableCount = 0;
for (List<SSTableReader> level : repaired.manifest.generations)
repairedSSTableCount += level.size();
int repairedSSTableCount = repaired.manifest.getSSTables().size();
assertEquals(1, repairedSSTableCount);
// make sure the repaired sstable ends up in the same level in the repaired manifest:
assertTrue(repaired.manifest.generations[2].contains(sstable1));
assertTrue(repaired.manifest.getLevel(2).contains(sstable1));
// and that it is gone from unrepaired
assertFalse(unrepaired.manifest.generations[2].contains(sstable1));
assertFalse(unrepaired.manifest.getLevel(2).contains(sstable1));
unrepaired.removeSSTable(sstable2);
manager.handleNotification(new SSTableAddedNotification(singleton(sstable2)), this);
@ -480,4 +480,218 @@ public class LeveledCompactionStrategyTest
lastMaxTimeStamp = sstable.getMaxTimestamp();
}
}
@Test
public void testAddingOverlapping()
{
ColumnFamilyStore cfs = MockSchema.newCFS();
LeveledManifest lm = new LeveledManifest(cfs, 10, 10, new SizeTieredCompactionStrategyOptions());
List<SSTableReader> currentLevel = new ArrayList<>();
int gen = 1;
currentLevel.add(MockSchema.sstableWithLevel(gen++, 10, 20, 1, cfs));
currentLevel.add(MockSchema.sstableWithLevel(gen++, 21, 30, 1, cfs));
currentLevel.add(MockSchema.sstableWithLevel(gen++, 51, 100, 1, cfs));
currentLevel.add(MockSchema.sstableWithLevel(gen++, 80, 120, 1, cfs));
currentLevel.add(MockSchema.sstableWithLevel(gen++, 90, 150, 1, cfs));
lm.addSSTables(currentLevel);
assertLevelsEqual(lm.getLevel(1), currentLevel.subList(0, 3));
assertLevelsEqual(lm.getLevel(0), currentLevel.subList(3, 5));
List<SSTableReader> newSSTables = new ArrayList<>();
// this sstable last token is the same as the first token of L1 above, should get sent to L0:
newSSTables.add(MockSchema.sstableWithLevel(gen++, 5, 10, 1, cfs));
lm.addSSTables(newSSTables);
assertLevelsEqual(lm.getLevel(1), currentLevel.subList(0, 3));
assertEquals(0, newSSTables.get(0).getSSTableLevel());
assertTrue(lm.getLevel(0).containsAll(newSSTables));
newSSTables.clear();
newSSTables.add(MockSchema.sstableWithLevel(gen++, 30, 40, 1, cfs));
lm.addSSTables(newSSTables);
assertLevelsEqual(lm.getLevel(1), currentLevel.subList(0, 3));
assertEquals(0, newSSTables.get(0).getSSTableLevel());
assertTrue(lm.getLevel(0).containsAll(newSSTables));
newSSTables.clear();
newSSTables.add(MockSchema.sstableWithLevel(gen++, 100, 140, 1, cfs));
lm.addSSTables(newSSTables);
assertLevelsEqual(lm.getLevel(1), currentLevel.subList(0, 3));
assertEquals(0, newSSTables.get(0).getSSTableLevel());
assertTrue(lm.getLevel(0).containsAll(newSSTables));
newSSTables.clear();
newSSTables.add(MockSchema.sstableWithLevel(gen++, 100, 140, 1, cfs));
newSSTables.add(MockSchema.sstableWithLevel(gen++, 120, 140, 1, cfs));
lm.addSSTables(newSSTables);
List<SSTableReader> newL1 = new ArrayList<>(currentLevel.subList(0, 3));
newL1.add(newSSTables.get(1));
assertLevelsEqual(lm.getLevel(1), newL1);
newSSTables.remove(1);
assertTrue(newSSTables.stream().allMatch(s -> s.getSSTableLevel() == 0));
assertTrue(lm.getLevel(0).containsAll(newSSTables));
}
@Test
public void singleTokenSSTableTest()
{
ColumnFamilyStore cfs = MockSchema.newCFS();
LeveledManifest lm = new LeveledManifest(cfs, 10, 10, new SizeTieredCompactionStrategyOptions());
List<SSTableReader> expectedL1 = new ArrayList<>();
int gen = 1;
// single sstable, single token (100)
expectedL1.add(MockSchema.sstableWithLevel(gen++, 100, 100, 1, cfs));
lm.addSSTables(expectedL1);
List<SSTableReader> expectedL0 = new ArrayList<>();
// should get moved to L0:
expectedL0.add(MockSchema.sstableWithLevel(gen++, 99, 101, 1, cfs));
expectedL0.add(MockSchema.sstableWithLevel(gen++, 100, 101, 1, cfs));
expectedL0.add(MockSchema.sstableWithLevel(gen++, 99, 100, 1, cfs));
expectedL0.add(MockSchema.sstableWithLevel(gen++, 100, 100, 1, cfs));
lm.addSSTables(expectedL0);
assertLevelsEqual(expectedL0, lm.getLevel(0));
assertTrue(expectedL0.stream().allMatch(s -> s.getSSTableLevel() == 0));
assertLevelsEqual(expectedL1, lm.getLevel(1));
assertTrue(expectedL1.stream().allMatch(s -> s.getSSTableLevel() == 1));
// should work:
expectedL1.add(MockSchema.sstableWithLevel(gen++, 98, 99, 1, cfs));
expectedL1.add(MockSchema.sstableWithLevel(gen++, 101, 101, 1, cfs));
lm.addSSTables(expectedL1.subList(1, expectedL1.size()));
assertLevelsEqual(expectedL1, lm.getLevel(1));
}
@Test
public void randomMultiLevelAddTest()
{
int iterations = 100;
int levelCount = 8;
ColumnFamilyStore cfs = MockSchema.newCFS();
LeveledManifest lm = new LeveledManifest(cfs, 10, 10, new SizeTieredCompactionStrategyOptions());
long seed = System.currentTimeMillis();
Random r = new Random(seed);
List<SSTableReader> newLevels = generateNewRandomLevels(cfs, 40, levelCount, 0, r);
int sstableCount = newLevels.size();
lm.addSSTables(newLevels);
int [] expectedLevelSizes = lm.getAllLevelSize();
for (int j = 0; j < iterations; j++)
{
newLevels = generateNewRandomLevels(cfs, 20, levelCount, sstableCount, r);
sstableCount += newLevels.size();
int[] canAdd = canAdd(lm, newLevels, levelCount);
for (int i = 0; i < levelCount; i++)
expectedLevelSizes[i] += canAdd[i];
lm.addSSTables(newLevels);
}
// and verify no levels overlap
int actualSSTableCount = 0;
for (int i = 0; i < levelCount; i++)
{
actualSSTableCount += lm.getLevelSize(i);
List<SSTableReader> level = new ArrayList<>(lm.getLevel(i));
int lvl = i;
assertTrue(level.stream().allMatch(s -> s.getSSTableLevel() == lvl));
if (i > 0)
{
level.sort(SSTableReader.sstableComparator);
SSTableReader prev = null;
for (SSTableReader sstable : level)
{
if (prev != null && sstable.first.compareTo(prev.last) <= 0)
{
String levelStr = level.stream().map(s -> String.format("[%s, %s]", s.first, s.last)).collect(Collectors.joining(", "));
String overlap = String.format("sstable [%s, %s] overlaps with [%s, %s] in level %d (%s) ", sstable.first, sstable.last, prev.first, prev.last, i, levelStr);
Assert.fail("[seed = "+seed+"] overlap in level "+lvl+": " + overlap);
}
prev = sstable;
}
}
}
assertEquals(sstableCount, actualSSTableCount);
for (int i = 0; i < levelCount; i++)
assertEquals("[seed = " + seed + "] wrong sstable count in level = " + i, expectedLevelSizes[i], lm.getLevel(i).size());
}
private static List<SSTableReader> generateNewRandomLevels(ColumnFamilyStore cfs, int maxSSTableCountPerLevel, int levelCount, int startGen, Random r)
{
List<SSTableReader> newLevels = new ArrayList<>();
for (int level = 0; level < levelCount; level++)
{
int numLevelSSTables = r.nextInt(maxSSTableCountPerLevel) + 1;
List<Integer> tokens = new ArrayList<>(numLevelSSTables * 2);
for (int i = 0; i < numLevelSSTables * 2; i++)
tokens.add(r.nextInt(4000));
Collections.sort(tokens);
for (int i = 0; i < tokens.size() - 1; i += 2)
{
SSTableReader sstable = MockSchema.sstableWithLevel(++startGen, tokens.get(i), tokens.get(i + 1), level, cfs);
newLevels.add(sstable);
}
}
return newLevels;
}
/**
* brute-force checks if the new sstables can be added to the correct level in manifest
*
* @return count of expected sstables to add to each level
*/
private static int[] canAdd(LeveledManifest lm, List<SSTableReader> newSSTables, int levelCount)
{
Map<Integer, Collection<SSTableReader>> sstableGroups = new HashMap<>();
newSSTables.forEach(s -> sstableGroups.computeIfAbsent(s.getSSTableLevel(), k -> new ArrayList<>()).add(s));
int[] canAdd = new int[levelCount];
for (Map.Entry<Integer, Collection<SSTableReader>> lvlGroup : sstableGroups.entrySet())
{
int level = lvlGroup.getKey();
if (level == 0)
{
canAdd[0] += lvlGroup.getValue().size();
continue;
}
List<SSTableReader> newLevel = new ArrayList<>(lm.getLevel(level));
for (SSTableReader sstable : lvlGroup.getValue())
{
newLevel.add(sstable);
newLevel.sort(SSTableReader.sstableComparator);
SSTableReader prev = null;
boolean kept = true;
for (SSTableReader sst : newLevel)
{
if (prev != null && prev.last.compareTo(sst.first) >= 0)
{
newLevel.remove(sstable);
kept = false;
break;
}
prev = sst;
}
if (kept)
canAdd[level] += 1;
else
canAdd[0] += 1;
}
}
return canAdd;
}
private static void assertLevelsEqual(Collection<SSTableReader> l1, Collection<SSTableReader> l2)
{
assertEquals(l1.size(), l2.size());
assertEquals(new HashSet<>(l1), new HashSet<>(l2));
}
}

199
test/unit/org/apache/cassandra/db/compaction/LeveledGenerationsTest.java Normal file
View File

@ -0,0 +1,199 @@
/*
* 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.compaction;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.MockSchema;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.db.BufferDecoratedKey;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.utils.ByteBufferUtil;
import static junit.framework.Assert.assertFalse;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;
public class LeveledGenerationsTest extends CQLTester
{
@BeforeClass
public static void setUp()
{
DatabaseDescriptor.daemonInitialization();
MockSchema.cleanup();
}
@Test
public void testWrappingIterable()
{
ColumnFamilyStore cfs = MockSchema.newCFS();
LeveledGenerations gens = new LeveledGenerations();
for (int i = 0; i < 10; i++)
{
SSTableReader sstable = MockSchema.sstable(i, 5, true, i, i, 2, cfs);
gens.addAll(Collections.singleton(sstable));
}
int gen = 10;
assertIter(gens.wrappingIterator(2, sst(++gen, cfs, 5, 5)),
6, 5, 10);
assertIter(gens.wrappingIterator(2, null),
0, 9, 10);
assertIter(gens.wrappingIterator(2, sst(++gen, cfs, -10, 0)),
1, 0, 10);
assertIter(gens.wrappingIterator(2, sst(++gen, cfs, 5, 9)),
0, 9, 10);
assertIter(gens.wrappingIterator(2, sst(++gen, cfs, 0, 1000)),
0, 9, 10);
gens.addAll(Collections.singleton(MockSchema.sstable(100, 5, true, 5, 10, 3, cfs)));
assertIter(gens.wrappingIterator(3, sst(++gen, cfs, -10, 0)),
5, 5, 1);
assertIter(gens.wrappingIterator(3, sst(++gen, cfs, 0, 100)),
5, 5, 1);
gens.addAll(Collections.singleton(MockSchema.sstable(200, 5, true, 5, 10, 4, cfs)));
gens.addAll(Collections.singleton(MockSchema.sstable(201, 5, true, 40, 50, 4, cfs)));
assertIter(gens.wrappingIterator(4, sst(++gen, cfs, 0, 0)),
5, 40, 2);
assertIter(gens.wrappingIterator(4, sst(++gen, cfs, 0, 5)),
40, 5, 2);
assertIter(gens.wrappingIterator(4, sst(++gen, cfs, 7, 8)),
40, 5, 2);
assertIter(gens.wrappingIterator(4, sst(++gen, cfs, 39, 39)),
40, 5, 2);
assertIter(gens.wrappingIterator(4, sst(++gen, cfs, 40, 40)),
5, 40, 2);
assertIter(gens.wrappingIterator(4, sst(++gen, cfs, 100, 1000)),
5, 40, 2);
}
@Test
public void testWrappingIterableWiderSSTables()
{
ColumnFamilyStore cfs = MockSchema.newCFS();
LeveledGenerations generations = new LeveledGenerations();
int gen = 0;
generations.addAll(Lists.newArrayList(
sst(++gen, cfs, 0, 50),
sst(++gen, cfs, 51, 100),
sst(++gen, cfs, 150, 200)));
assertIter(generations.wrappingIterator(2, sst(++gen, cfs, -100, -50)),
0, 150, 3);
assertIter(generations.wrappingIterator(2, sst(++gen, cfs, 0, 40)),
51, 0, 3);
assertIter(generations.wrappingIterator(2, sst(++gen, cfs, 0, 50)),
51, 0, 3);
assertIter(generations.wrappingIterator(2, sst(++gen, cfs, 0, 51)),
150, 51, 3);
assertIter(generations.wrappingIterator(2, sst(++gen, cfs, 100, 149)),
150, 51, 3);
assertIter(generations.wrappingIterator(2, sst(++gen, cfs, 100, 300)),
0, 150, 3);
}
@Test
public void testEmptyLevel()
{
ColumnFamilyStore cfs = MockSchema.newCFS();
LeveledGenerations generations = new LeveledGenerations();
assertFalse(generations.wrappingIterator(3, sst(0, cfs, 0, 10)).hasNext());
assertFalse(generations.wrappingIterator(3, null).hasNext());
}
@Test
public void testFillLevels()
{
LeveledGenerations generations = new LeveledGenerations();
ColumnFamilyStore cfs = MockSchema.newCFS();
for (int i = 0; i < LeveledGenerations.MAX_LEVEL_COUNT; i++)
generations.addAll(Collections.singleton(MockSchema.sstableWithLevel(i, i, i, i, cfs)));
for (int i = 0; i < generations.levelCount(); i++)
assertEquals(i, generations.get(i).iterator().next().getSSTableLevel());
assertEquals(9, generations.levelCount());
try
{
generations.get(9);
fail("don't have 9 generations");
}
catch (ArrayIndexOutOfBoundsException e)
{}
try
{
generations.get(-1);
fail("don't have -1 generations");
}
catch (ArrayIndexOutOfBoundsException e)
{}
}
private void assertIter(Iterator<SSTableReader> iter, long first, long last, int expectedCount)
{
List<SSTableReader> drained = Lists.newArrayList(iter);
assertEquals(expectedCount, drained.size());
assertEquals(dk(first).getToken(), first(drained).first.getToken());
assertEquals(dk(last).getToken(), last(drained).first.getToken()); // we sort by first token, so this is the first token of the last sstable in iter
}
private SSTableReader last(Iterable<SSTableReader> iter)
{
return Iterables.getLast(iter);
}
private SSTableReader first(Iterable<SSTableReader> iter)
{
SSTableReader first = Iterables.getFirst(iter, null);
if (first == null)
throw new RuntimeException();
return first;
}
private DecoratedKey dk(long x)
{
return new BufferDecoratedKey(new Murmur3Partitioner.LongToken(x), ByteBufferUtil.bytes(x));
}
private SSTableReader sst(int gen, ColumnFamilyStore cfs, long first, long last)
{
return MockSchema.sstable(gen, 5, true, first, last, 2, cfs);
}
private void print(SSTableReader sstable)
{
System.out.println(String.format("%d %s %s %d", sstable.descriptor.generation, sstable.first, sstable.last, sstable.getSSTableLevel()));
}
}