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

Improve TRUNCATE performance 

Patch by marcuse; reviewed by Stefania Alborghetti for CASSANDRA-13909
This commit is contained in:
Marcus Eriksson 2017-09-25 14:44:37 +02:00
parent 15cee48bba
commit b32a9e6452
7 changed files with 179 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
3.0.15
* Improve TRUNCATE performance (CASSANDRA-13909)
* Implement short read protection on partition boundaries (CASSANDRA-13595)
* Fix ISE thrown by UPI.Serializer.hasNext() for some SELECT queries (CASSANDRA-13911)
* Filter header only commit logs before recovery (CASSANDRA-13918)

15
src/java/org/apache/cassandra/db/lifecycle/Helpers.java
View File

@ -141,6 +141,21 @@ class Helpers
return accumulate;
}
static Throwable prepareForBulkObsoletion(Iterable<SSTableReader> readers, LogTransaction txnLogs, List<LogTransaction.Obsoletion> obsoletions, Throwable accumulate)
{
try
{
for (Map.Entry<SSTableReader, LogTransaction.SSTableTidier> entry : txnLogs.bulkObsoletion(readers).entrySet())
obsoletions.add(new LogTransaction.Obsoletion(entry.getKey(), entry.getValue()));
}
catch (Throwable t)
{
accumulate = Throwables.merge(accumulate, t);
}
return accumulate;
}
static Throwable abortObsoletion(List<LogTransaction.Obsoletion> obsoletions, Throwable accumulate)
{
if (obsoletions == null || obsoletions.isEmpty())

25
src/java/org/apache/cassandra/db/lifecycle/LogFile.java
View File

@ -37,6 +37,7 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.db.compaction.OperationType;
import org.apache.cassandra.db.lifecycle.LogRecord.Type;
import org.apache.cassandra.io.sstable.SSTable;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.sstable.format.big.BigFormat;
import org.apache.cassandra.utils.Throwables;
@ -284,6 +285,25 @@ final class LogFile implements AutoCloseable
throw new IllegalStateException();
}
public void addAll(Type type, Iterable<SSTableReader> toBulkAdd)
{
for (LogRecord record : makeRecords(type, toBulkAdd))
if (!addRecord(record))
throw new IllegalStateException();
}
private Collection<LogRecord> makeRecords(Type type, Iterable<SSTableReader> tables)
{
assert type == Type.ADD || type == Type.REMOVE;
for (SSTableReader sstable : tables)
{
File folder = sstable.descriptor.directory;
replicas.maybeCreateReplica(folder, getFileName(folder), records);
}
return LogRecord.make(type, tables);
}
private LogRecord makeRecord(Type type, SSTable table)
{
assert type == Type.ADD || type == Type.REMOVE;
@ -414,4 +434,9 @@ final class LogFile implements AutoCloseable
{
return records;
}
public boolean isEmpty()
{
return records.isEmpty();
}
}

65
src/java/org/apache/cassandra/db/lifecycle/LogRecord.java
View File

@ -21,6 +21,7 @@
package org.apache.cassandra.db.lifecycle;
import java.io.File;
import java.io.FilenameFilter;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.*;
@ -30,7 +31,9 @@ import java.util.stream.Collectors;
import java.util.zip.CRC32;
import org.apache.cassandra.io.sstable.Component;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.sstable.SSTable;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.utils.FBUtilities;
@ -151,10 +154,35 @@ final class LogRecord
// there is no separator after the generation number, and this would cause files of sstables with
// a higher generation number that starts with the same number, to be incorrectly classified as files
// of this record sstable
String absoluteTablePath = FileUtils.getCanonicalPath(table.descriptor.baseFilename() + Component.separator);
String absoluteTablePath = absolutePath(table.descriptor.baseFilename());
return make(type, getExistingFiles(absoluteTablePath), table.getAllFilePaths().size(), absoluteTablePath);
}
public static Collection<LogRecord> make(Type type, Iterable<SSTableReader> tables)
{
// contains a mapping from sstable absolute path (everything up until the 'Data'/'Index'/etc part of the filename) to the sstable
Map<String, SSTable> absolutePaths = new HashMap<>();
for (SSTableReader table : tables)
absolutePaths.put(absolutePath(table.descriptor.baseFilename()), table);
// maps sstable base file name to the actual files on disk
Map<String, List<File>> existingFiles = getExistingFiles(absolutePaths.keySet());
List<LogRecord> records = new ArrayList<>(existingFiles.size());
for (Map.Entry<String, List<File>> entry : existingFiles.entrySet())
{
List<File> filesOnDisk = entry.getValue();
String baseFileName = entry.getKey();
SSTable sstable = absolutePaths.get(baseFileName);
records.add(make(type, filesOnDisk, sstable.getAllFilePaths().size(), baseFileName));
}
return records;
}
private static String absolutePath(String baseFilename)
{
return FileUtils.getCanonicalPath(baseFilename + Component.separator);
}
public LogRecord withExistingFiles()
{
return make(type, getExistingFiles(), 0, absolutePath.get());
@ -275,6 +303,41 @@ final class LogRecord
return files == null ? Collections.emptyList() : Arrays.asList(files);
}
/**
* absoluteFilePaths contains full file parts up to the component name
*
* this method finds all files on disk beginning with any of the paths in absoluteFilePaths
* @return a map from absoluteFilePath to actual file on disk.
*/
public static Map<String, List<File>> getExistingFiles(Set<String> absoluteFilePaths)
{
Set<File> uniqueDirectories = absoluteFilePaths.stream().map(path -> Paths.get(path).getParent().toFile()).collect(Collectors.toSet());
Map<String, List<File>> fileMap = new HashMap<>();
FilenameFilter ff = (dir, name) -> {
Descriptor descriptor = null;
try
{
descriptor = Descriptor.fromFilename(dir, name).left;
}
catch (Throwable t)
{// ignored - if we can't parse the filename, just skip the file
}
String absolutePath = descriptor != null ? absolutePath(descriptor.baseFilename()) : null;
if (absolutePath != null && absoluteFilePaths.contains(absolutePath))
fileMap.computeIfAbsent(absolutePath, k -> new ArrayList<>()).add(new File(dir, name));
return false;
};
// populate the file map:
for (File f : uniqueDirectories)
f.listFiles(ff);
return fileMap;
}
public boolean isFinal()
{
return type.isFinal();

16
src/java/org/apache/cassandra/db/lifecycle/LogTransaction.java
View File

@ -163,6 +163,22 @@ class LogTransaction extends Transactional.AbstractTransactional implements Tran
return new SSTableTidier(reader, false, this);
}
Map<SSTableReader, SSTableTidier> bulkObsoletion(Iterable<SSTableReader> sstables)
{
if (!txnFile.isEmpty())
throw new IllegalStateException("Bad state when doing bulk obsoletions");
txnFile.addAll(Type.REMOVE, sstables);
Map<SSTableReader, SSTableTidier> tidiers = new HashMap<>();
for (SSTableReader sstable : sstables)
{
if (tracker != null)
tracker.notifyDeleting(sstable);
tidiers.put(sstable, new SSTableTidier(sstable, false, this));
}
return tidiers;
}
OperationType type()
{
return txnFile.type();

2
src/java/org/apache/cassandra/db/lifecycle/Tracker.java
View File

@ -245,7 +245,7 @@ public class Tracker
// It is important that any method accepting/returning a Throwable never throws an exception, and does its best
// to complete the instructions given to it
List<LogTransaction.Obsoletion> obsoletions = new ArrayList<>();
accumulate = prepareForObsoletion(removed, txnLogs, obsoletions, accumulate);
accumulate = prepareForBulkObsoletion(removed, txnLogs, obsoletions, accumulate);
try
{
txnLogs.finish();

58
test/unit/org/apache/cassandra/db/lifecycle/HelpersTest.java
View File

@ -18,14 +18,21 @@
*/
package org.apache.cassandra.db.lifecycle;
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import org.junit.BeforeClass;
import org.junit.Test;
@ -157,13 +164,24 @@ public class HelpersTest
@Test
public void testMarkObsolete()
{
testMarkObsoleteHelper(false);
}
@Test
public void testBulkMarkObsolete()
{
testMarkObsoleteHelper(true);
}
public void testMarkObsoleteHelper(boolean bulk)
{
ColumnFamilyStore cfs = MockSchema.newCFS();
LogTransaction txnLogs = new LogTransaction(OperationType.UNKNOWN);
Iterable<SSTableReader> readers = Lists.newArrayList(MockSchema.sstable(1, cfs), MockSchema.sstable(2, cfs));
Iterable<SSTableReader> readersToKeep = Lists.newArrayList(MockSchema.sstable(3, cfs), MockSchema.sstable(4, cfs));
List<LogTransaction.Obsoletion> obsoletions = new ArrayList<>();
Assert.assertNull(Helpers.prepareForObsoletion(readers, txnLogs, obsoletions, null));
Assert.assertNull(bulk ? Helpers.prepareForBulkObsoletion(readers, txnLogs, obsoletions, null) : Helpers.prepareForObsoletion(readers, txnLogs, obsoletions, null));
assertNotNull(obsoletions);
assertEquals(2, obsoletions.size());
@ -172,9 +190,47 @@ public class HelpersTest
for (SSTableReader reader : readers)
Assert.assertTrue(reader.isMarkedCompacted());
for (SSTableReader reader : readersToKeep)
Assert.assertFalse(reader.isMarkedCompacted());
accumulate = Helpers.markObsolete(obsoletions, null);
assertNotNull(accumulate);
txnLogs.finish();
}
@Test
public void compareBulkAndNormalObsolete() throws IOException
{
ColumnFamilyStore cfs = MockSchema.newCFS();
LogTransaction txnLogs = new LogTransaction(OperationType.UNKNOWN);
LogTransaction txnLogs2 = new LogTransaction(OperationType.UNKNOWN);
Collection<SSTableReader> readers = Lists.newArrayList(MockSchema.sstable(1, cfs), MockSchema.sstable(2, cfs));
// add a few readers that should not be removed:
Lists.newArrayList(MockSchema.sstable(3, cfs), MockSchema.sstable(4, cfs));
List<LogTransaction.Obsoletion> normalObsoletions = new ArrayList<>();
List<LogTransaction.Obsoletion> bulkObsoletions = new ArrayList<>();
Assert.assertNull(Helpers.prepareForBulkObsoletion(readers, txnLogs, normalObsoletions, null));
Assert.assertNull(Helpers.prepareForObsoletion(readers, txnLogs2, bulkObsoletions, null));
assertEquals(Sets.newHashSet(readers), normalObsoletions.stream().map(obs -> obs.reader).collect(Collectors.toSet()));
assertEquals(Sets.newHashSet(readers), bulkObsoletions.stream().map(obs -> obs.reader).collect(Collectors.toSet()));
Set<String> normalLogRecords = new HashSet<>();
Set<String> bulkLogRecords = new HashSet<>();
for (File f : txnLogs.logFiles())
Files.lines(f.toPath()).forEach(bulkLogRecords::add);
for (File f : txnLogs2.logFiles())
Files.lines(f.toPath()).forEach(normalLogRecords::add);
Assert.assertEquals(readers.size(), normalLogRecords.size());
Assert.assertEquals(bulkLogRecords, normalLogRecords);
txnLogs.finish();
txnLogs2.finish();
}
}