cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From marc...@apache.org
Subject [3/6] cassandra git commit: Pre-create deletion log records to finish compactions quicker
Date Mon, 19 Feb 2018 15:56:25 GMT
Pre-create deletion log records to finish compactions quicker

Patch by marcuse; reviewed by Stefania Alborghetti for CASSANDRA-12763


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/d73f45ba
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/d73f45ba
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/d73f45ba

Branch: refs/heads/trunk
Commit: d73f45bad4cd6d8cf1cea7d9b35b76075dc277e1
Parents: f39e28d
Author: Marcus Eriksson <marcuse@apache.org>
Authored: Mon Dec 11 15:11:20 2017 +0100
Committer: Marcus Eriksson <marcuse@apache.org>
Committed: Mon Feb 19 07:52:20 2018 -0800

----------------------------------------------------------------------
 CHANGES.txt                                     |  1 +
 .../apache/cassandra/db/lifecycle/Helpers.java  | 19 ++-----
 .../apache/cassandra/db/lifecycle/LogFile.java  | 37 +++++++++++--
 .../cassandra/db/lifecycle/LogRecord.java       | 11 ++--
 .../cassandra/db/lifecycle/LogTransaction.java  | 33 ++++++------
 .../apache/cassandra/db/lifecycle/Tracker.java  |  2 +-
 .../cassandra/db/lifecycle/HelpersTest.java     | 56 +++++---------------
 7 files changed, 74 insertions(+), 85 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/d73f45ba/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 14a62a4..8cf665e 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 3.0.17
+ * Pre-create deletion log records to finish compactions quicker (CASSANDRA-12763)
 Merged from 2.1:
  * CVE-2017-5929 Security vulnerability in Logback warning in NEWS.txt (CASSANDRA-14183)
 Merged from 2.2:

http://git-wip-us.apache.org/repos/asf/cassandra/blob/d73f45ba/src/java/org/apache/cassandra/db/lifecycle/Helpers.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/lifecycle/Helpers.java b/src/java/org/apache/cassandra/db/lifecycle/Helpers.java
index b9adc4b..8e0d514 100644
--- a/src/java/org/apache/cassandra/db/lifecycle/Helpers.java
+++ b/src/java/org/apache/cassandra/db/lifecycle/Helpers.java
@@ -22,6 +22,7 @@ import java.util.*;
 import com.google.common.base.Predicate;
 import com.google.common.collect.*;
 
+import org.apache.cassandra.io.sstable.SSTable;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.utils.Throwables;
 
@@ -127,11 +128,12 @@ class Helpers
 
     static Throwable prepareForObsoletion(Iterable<SSTableReader> readers, LogTransaction
txnLogs, List<LogTransaction.Obsoletion> obsoletions, Throwable accumulate)
     {
+        Map<SSTable, LogRecord> logRecords = txnLogs.makeRemoveRecords(readers);
         for (SSTableReader reader : readers)
         {
             try
             {
-                obsoletions.add(new LogTransaction.Obsoletion(reader, txnLogs.obsoleted(reader)));
+                obsoletions.add(new LogTransaction.Obsoletion(reader, txnLogs.obsoleted(reader,
logRecords.get(reader))));
             }
             catch (Throwable t)
             {
@@ -141,21 +143,6 @@ 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())

http://git-wip-us.apache.org/repos/asf/cassandra/blob/d73f45ba/src/java/org/apache/cassandra/db/lifecycle/LogFile.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/lifecycle/LogFile.java b/src/java/org/apache/cassandra/db/lifecycle/LogFile.java
index be26163..8425a6d 100644
--- a/src/java/org/apache/cassandra/db/lifecycle/LogFile.java
+++ b/src/java/org/apache/cassandra/db/lifecycle/LogFile.java
@@ -281,18 +281,23 @@ final class LogFile implements AutoCloseable
 
     void add(Type type, SSTable table)
     {
-        if (!addRecord(makeRecord(type, table)))
+        add(makeRecord(type, table));
+    }
+
+    void add(LogRecord record)
+    {
+        if (!addRecord(record))
             throw new IllegalStateException();
     }
 
     public void addAll(Type type, Iterable<SSTableReader> toBulkAdd)
     {
-        for (LogRecord record : makeRecords(type, toBulkAdd))
+        for (LogRecord record : makeRecords(type, toBulkAdd).values())
             if (!addRecord(record))
                 throw new IllegalStateException();
     }
 
-    private Collection<LogRecord> makeRecords(Type type, Iterable<SSTableReader>
tables)
+    Map<SSTable, LogRecord> makeRecords(Type type, Iterable<SSTableReader> tables)
     {
         assert type == Type.ADD || type == Type.REMOVE;
 
@@ -313,6 +318,20 @@ final class LogFile implements AutoCloseable
         return LogRecord.make(type, table);
     }
 
+    /**
+     * this version of makeRecord takes an existing LogRecord and converts it to a
+     * record with the given type. This avoids listing the directory and if the
+     * LogRecord already exists, we have all components for the sstable
+     */
+    private LogRecord makeRecord(Type type, SSTable table, LogRecord record)
+    {
+        assert type == Type.ADD || type == Type.REMOVE;
+
+        File folder = table.descriptor.directory;
+        replicas.maybeCreateReplica(folder, getFileName(folder), records);
+        return record.asType(type);
+    }
+
     private boolean addRecord(LogRecord record)
     {
         if (records.contains(record))
@@ -334,7 +353,17 @@ final class LogFile implements AutoCloseable
 
     boolean contains(Type type, SSTable table)
     {
-        return records.contains(makeRecord(type, table));
+        return contains(makeRecord(type, table));
+    }
+
+    boolean contains(Type type, SSTable sstable, LogRecord record)
+    {
+        return contains(makeRecord(type, sstable, record));
+    }
+
+    private boolean contains(LogRecord record)
+    {
+        return records.contains(record);
     }
 
     void deleteFilesForRecordsOfType(Type type)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/d73f45ba/src/java/org/apache/cassandra/db/lifecycle/LogRecord.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/lifecycle/LogRecord.java b/src/java/org/apache/cassandra/db/lifecycle/LogRecord.java
index a322ea1..1dc17f6 100644
--- a/src/java/org/apache/cassandra/db/lifecycle/LogRecord.java
+++ b/src/java/org/apache/cassandra/db/lifecycle/LogRecord.java
@@ -158,7 +158,7 @@ final class LogRecord
         return make(type, getExistingFiles(absoluteTablePath), table.getAllFilePaths().size(),
absoluteTablePath);
     }
 
-    public static Collection<LogRecord> make(Type type, Iterable<SSTableReader>
tables)
+    public static Map<SSTable, 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<>();
@@ -167,13 +167,13 @@ final class LogRecord
 
         // 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());
+        Map<SSTable, LogRecord> records = new HashMap<>(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));
+            records.put(sstable, make(type, filesOnDisk, sstable.getAllFilePaths().size(),
baseFileName));
         }
         return records;
     }
@@ -415,4 +415,9 @@ final class LogRecord
         FBUtilities.updateChecksumInt(crc32, numFiles);
         return crc32.getValue() & (Long.MAX_VALUE);
     }
+
+    LogRecord asType(Type type)
+    {
+        return new LogRecord(type, absolutePath.orElse(null), updateTime, numFiles);
+    }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/d73f45ba/src/java/org/apache/cassandra/db/lifecycle/LogTransaction.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/lifecycle/LogTransaction.java b/src/java/org/apache/cassandra/db/lifecycle/LogTransaction.java
index 6599142..a10bcd2 100644
--- a/src/java/org/apache/cassandra/db/lifecycle/LogTransaction.java
+++ b/src/java/org/apache/cassandra/db/lifecycle/LogTransaction.java
@@ -143,19 +143,28 @@ class LogTransaction extends Transactional.AbstractTransactional implements
Tran
     }
 
     /**
+     * helper method for tests, creates the remove records per sstable
+     */
+    @VisibleForTesting
+    SSTableTidier obsoleted(SSTableReader sstable)
+    {
+        return obsoleted(sstable, LogRecord.make(Type.REMOVE, sstable));
+    }
+
+    /**
      * Schedule a reader for deletion as soon as it is fully unreferenced.
      */
-    SSTableTidier obsoleted(SSTableReader reader)
+    SSTableTidier obsoleted(SSTableReader reader, LogRecord logRecord)
     {
-        if (txnFile.contains(Type.ADD, reader))
+        if (txnFile.contains(Type.ADD, reader, logRecord))
         {
-            if (txnFile.contains(Type.REMOVE, reader))
+            if (txnFile.contains(Type.REMOVE, reader, logRecord))
                 throw new IllegalArgumentException();
 
             return new SSTableTidier(reader, true, this);
         }
 
-        txnFile.add(Type.REMOVE, reader);
+        txnFile.add(logRecord);
 
         if (tracker != null)
             tracker.notifyDeleting(reader);
@@ -163,22 +172,12 @@ class LogTransaction extends Transactional.AbstractTransactional implements
Tran
         return new SSTableTidier(reader, false, this);
     }
 
-    Map<SSTableReader, SSTableTidier> bulkObsoletion(Iterable<SSTableReader>
sstables)
+    Map<SSTable, LogRecord> makeRemoveRecords(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;
+        return txnFile.makeRecords(Type.REMOVE, sstables);
     }
 
+
     OperationType type()
     {
         return txnFile.type();

http://git-wip-us.apache.org/repos/asf/cassandra/blob/d73f45ba/src/java/org/apache/cassandra/db/lifecycle/Tracker.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/lifecycle/Tracker.java b/src/java/org/apache/cassandra/db/lifecycle/Tracker.java
index d281278..9feaa3e 100644
--- a/src/java/org/apache/cassandra/db/lifecycle/Tracker.java
+++ b/src/java/org/apache/cassandra/db/lifecycle/Tracker.java
@@ -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 = prepareForBulkObsoletion(removed, txnLogs, obsoletions, accumulate);
+            accumulate = prepareForObsoletion(removed, txnLogs, obsoletions, accumulate);
             try
             {
                 txnLogs.finish();

http://git-wip-us.apache.org/repos/asf/cassandra/blob/d73f45ba/test/unit/org/apache/cassandra/db/lifecycle/HelpersTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/db/lifecycle/HelpersTest.java b/test/unit/org/apache/cassandra/db/lifecycle/HelpersTest.java
index 1b8e265..1d9f8aa 100644
--- a/test/unit/org/apache/cassandra/db/lifecycle/HelpersTest.java
+++ b/test/unit/org/apache/cassandra/db/lifecycle/HelpersTest.java
@@ -18,21 +18,14 @@
 */
 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;
@@ -45,6 +38,7 @@ import org.apache.cassandra.io.sstable.format.SSTableReader;
 
 import static junit.framework.Assert.assertEquals;
 import static junit.framework.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
 
 public class HelpersTest
 {
@@ -165,23 +159,13 @@ 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(bulk ? Helpers.prepareForBulkObsoletion(readers, txnLogs, obsoletions,
null) : Helpers.prepareForObsoletion(readers, txnLogs, obsoletions, null));
+        Helpers.prepareForObsoletion(readers, txnLogs, obsoletions, null);
         assertNotNull(obsoletions);
         assertEquals(2, obsoletions.size());
 
@@ -200,37 +184,21 @@ public class HelpersTest
     }
 
     @Test
-    public void compareBulkAndNormalObsolete() throws IOException
+    public void testObsoletionPerformance()
     {
         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<SSTableReader> readers = new ArrayList<>();
 
-        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);
+        for (int i = 0; i < 10000; i++)
+        {
+            readers.add(MockSchema.sstable(i + 1, cfs));
+        }
+        long start = System.currentTimeMillis();
 
+        Helpers.prepareForObsoletion(readers.subList(0, 500), txnLogs, new ArrayList<>(),null
);
         txnLogs.finish();
-        txnLogs2.finish();
+        long time = System.currentTimeMillis() - start;
+        assertTrue(time < 20000);
     }
 }


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org


Mime
View raw message