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

Make LCS split compaction results over many directories 

Patch by marcuse; reviewed by yukim for CASSANDRA-8329
This commit is contained in:
Marcus Eriksson 2014-11-18 11:01:17 +01:00
parent 025b4060cb
commit 2ce1ad8e6f
4 changed files with 70 additions and 64 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.12:
* Make LCS split compaction results over all data directories (CASSANDRA-8329)
* Fix some failing queries that use multi-column relations
on COMPACT STORAGE tables (CASSANDRA-8264)
* Fix InvalidRequestException with ORDER BY (CASSANDRA-8286)

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

@ -336,13 +336,23 @@ public class Memtable
return estimatedSize;
}
protected void runWith(File sstableDirectory) throws Exception
protected void runMayThrow() throws Exception
{
long writeSize = getExpectedWriteSize();
Directories.DataDirectory dataDirectory = getWriteDirectory(writeSize);
File sstableDirectory = cfs.directories.getLocationForDisk(dataDirectory);
assert sstableDirectory != null : "Flush task is not bound to any disk";
SSTableReader sstable = writeSortedContents(context, sstableDirectory);
cfs.replaceFlushed(Memtable.this, sstable);
latch.countDown();
try
{
SSTableReader sstable = writeSortedContents(context, sstableDirectory);
cfs.replaceFlushed(Memtable.this, sstable);
latch.countDown();
}
finally
{
if (dataDirectory != null)
returnWriteDirectory(dataDirectory, writeSize);
}
}
protected Directories getDirectories()

76
src/java/org/apache/cassandra/db/compaction/CompactionTask.java
View File

@ -87,11 +87,11 @@ public class CompactionTask extends AbstractCompactionTask
* which are properly serialized.
* Caller is in charge of marking/unmarking the sstables as compacting.
*/
protected void runWith(File sstableDirectory) throws Exception
protected void runMayThrow() throws Exception
{
// The collection of sstables passed may be empty (but not null); even if
// it is not empty, it may compact down to nothing if all rows are deleted.
assert sstables != null && sstableDirectory != null;
assert sstables != null;
// Note that the current compaction strategy, is not necessarily the one this task was created under.
// This should be harmless; see comments to CFS.maybeReloadCompactionStrategy.
@ -149,45 +149,60 @@ public class CompactionTask extends AbstractCompactionTask
return;
}
SSTableWriter writer = createCompactionWriter(sstableDirectory, keysPerSSTable);
long writeSize = getExpectedWriteSize() / estimatedSSTables;
Directories.DataDirectory dataDirectory = getWriteDirectory(writeSize);
SSTableWriter writer = createCompactionWriter(cfs.directories.getLocationForDisk(dataDirectory), keysPerSSTable);
writers.add(writer);
while (iter.hasNext())
try
{
if (ci.isStopRequested())
throw new CompactionInterruptedException(ci.getCompactionInfo());
AbstractCompactedRow row = iter.next();
RowIndexEntry indexEntry = writer.append(row);
if (indexEntry == null)
while (iter.hasNext())
{
controller.invalidateCachedRow(row.key);
row.close();
continue;
}
if (ci.isStopRequested())
throw new CompactionInterruptedException(ci.getCompactionInfo());
totalkeysWritten++;
if (DatabaseDescriptor.getPreheatKeyCache())
{
for (SSTableReader sstable : actuallyCompact)
AbstractCompactedRow row = iter.next();
RowIndexEntry indexEntry = writer.append(row);
if (indexEntry == null)
{
if (sstable.getCachedPosition(row.key, false) != null)
controller.invalidateCachedRow(row.key);
row.close();
continue;
}
totalkeysWritten++;
if (DatabaseDescriptor.getPreheatKeyCache())
{
for (SSTableReader sstable : actuallyCompact)
{
cachedKeys.put(row.key, indexEntry);
break;
if (sstable.getCachedPosition(row.key, false) != null)
{
cachedKeys.put(row.key, indexEntry);
break;
}
}
}
}
if (newSSTableSegmentThresholdReached(writer))
{
// tmp = false because later we want to query it with descriptor from SSTableReader
cachedKeyMap.put(writer.descriptor.asTemporary(false), cachedKeys);
writer = createCompactionWriter(sstableDirectory, keysPerSSTable);
writers.add(writer);
cachedKeys = new HashMap<DecoratedKey, RowIndexEntry>();
if (newSSTableSegmentThresholdReached(writer))
{
// tmp = false because later we want to query it with descriptor from SSTableReader
cachedKeyMap.put(writer.descriptor.asTemporary(false), cachedKeys);
returnWriteDirectory(dataDirectory, writeSize);
// make sure we don't try to call returnWriteDirectory in finally {..} if we throw exception in getWriteDirectory() below:
dataDirectory = null;
writeSize = getExpectedWriteSize() / estimatedSSTables;
dataDirectory = getWriteDirectory(writeSize);
writer = createCompactionWriter(cfs.directories.getLocationForDisk(dataDirectory), keysPerSSTable);
writers.add(writer);
cachedKeys = new HashMap<DecoratedKey, RowIndexEntry>();
}
}
}
finally
{
if (dataDirectory != null)
returnWriteDirectory(dataDirectory, writeSize);
}
if (writer.getFilePointer() > 0)
{
@ -291,6 +306,7 @@ public class CompactionTask extends AbstractCompactionTask
private SSTableWriter createCompactionWriter(File sstableDirectory, long keysPerSSTable)
{
assert sstableDirectory != null;
return new SSTableWriter(cfs.getTempSSTablePath(sstableDirectory),
keysPerSSTable,
cfs.metadata,

37
src/java/org/apache/cassandra/io/util/DiskAwareRunnable.java
View File

@ -17,23 +17,16 @@
*/
package org.apache.cassandra.io.util;
import java.io.File;
import org.apache.cassandra.db.Directories;
import org.apache.cassandra.utils.WrappedRunnable;
public abstract class DiskAwareRunnable extends WrappedRunnable
{
/**
* Run this task after selecting the optimal disk for it
*/
protected void runMayThrow() throws Exception
protected Directories.DataDirectory getWriteDirectory(long writeSize)
{
long writeSize;
Directories.DataDirectory directory;
while (true)
{
writeSize = getExpectedWriteSize();
directory = getDirectories().getWriteableLocation();
if (directory != null || !reduceScopeForLimitedSpace())
break;
@ -43,15 +36,13 @@ public abstract class DiskAwareRunnable extends WrappedRunnable
directory.currentTasks.incrementAndGet();
directory.estimatedWorkingSize.addAndGet(writeSize);
try
{
runWith(getDirectories().getLocationForDisk(directory));
}
finally
{
directory.estimatedWorkingSize.addAndGet(-1 * writeSize);
directory.currentTasks.decrementAndGet();
}
return directory;
}
protected void returnWriteDirectory(Directories.DataDirectory directory, long writeSize)
{
directory.estimatedWorkingSize.addAndGet(-1 * writeSize);
directory.currentTasks.decrementAndGet();
}
/**
@ -60,18 +51,6 @@ public abstract class DiskAwareRunnable extends WrappedRunnable
*/
protected abstract Directories getDirectories();
/**
* Executes this task on given {@code sstableDirectory}.
* @param sstableDirectory sstable directory to work on
*/
protected abstract void runWith(File sstableDirectory) throws Exception;
/**
* Get expected write size to determine which disk to use for this task.
* @return expected size in bytes this task will write to disk.
*/
public abstract long getExpectedWriteSize();
/**
* Called if no disk is available with free space for the full write size.
* @return true if the scope of the task was successfully reduced.