cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From yu...@apache.org
Subject [2/3] git commit: revert CASSANDRA-5151 from 1.2
Date Thu, 14 Feb 2013 17:28:06 GMT
revert CASSANDRA-5151 from 1.2


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

Branch: refs/heads/trunk
Commit: 7ff2805c7e6a362eb451e6b8dfaec59642fee0f2
Parents: 694e3a7
Author: Yuki Morishita <yukim@apache.org>
Authored: Thu Feb 14 11:13:11 2013 -0600
Committer: Yuki Morishita <yukim@apache.org>
Committed: Thu Feb 14 11:24:09 2013 -0600

----------------------------------------------------------------------
 CHANGES.txt                                        |    1 -
 .../org/apache/cassandra/config/CFMetaData.java    |    7 -
 .../org/apache/cassandra/config/KSMetaData.java    |    1 -
 .../org/apache/cassandra/db/ColumnFamilyStore.java |   95 ++++----------
 src/java/org/apache/cassandra/db/SystemTable.java  |   69 -----------
 .../cassandra/db/compaction/CompactionTask.java    |    7 -
 .../apache/cassandra/service/CassandraDaemon.java  |    9 --
 .../apache/cassandra/db/ColumnFamilyStoreTest.java |   55 ---------
 .../cassandra/db/compaction/CompactionsTest.java   |   34 -----
 9 files changed, 27 insertions(+), 251 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/7ff2805c/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 9281b5e..aa25b63 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,6 +1,5 @@
 1.2.2
  * avoid no-op caching of byte[] on commitlog append (CASSANDRA-5199)
- * more robust solution to incomplete compactions + counters (CASSANDRA-5151)
  * fix symlinks under data dir not working (CASSANDRA-5185)
  * fix bug in compact storage metadata handling (CASSANDRA-5189)
  * Validate login for USE queries (CASSANDRA-5207)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/7ff2805c/src/java/org/apache/cassandra/config/CFMetaData.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/config/CFMetaData.java b/src/java/org/apache/cassandra/config/CFMetaData.java
index 9c76c9b..73b7b6b 100644
--- a/src/java/org/apache/cassandra/config/CFMetaData.java
+++ b/src/java/org/apache/cassandra/config/CFMetaData.java
@@ -226,13 +226,6 @@ public final class CFMetaData
                                                               + "requested_at timestamp"
                                                               + ") WITH COMMENT='ranges requested
for transfer here'");
 
-    public static final CFMetaData CompactionLogCF = compile(18, "CREATE TABLE " + SystemTable.COMPACTION_LOG
+ " ("
-                                                                 + "id uuid PRIMARY KEY,"
-                                                                 + "keyspace_name text,"
-                                                                 + "columnfamily_name text,"
-                                                                 + "inputs set<int>"
-                                                                 + ") WITH COMMENT='unfinished
compactions'");
-
     public enum Caching
     {
         ALL, KEYS_ONLY, ROWS_ONLY, NONE;

http://git-wip-us.apache.org/repos/asf/cassandra/blob/7ff2805c/src/java/org/apache/cassandra/config/KSMetaData.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/config/KSMetaData.java b/src/java/org/apache/cassandra/config/KSMetaData.java
index b0764cc..9522aa0 100644
--- a/src/java/org/apache/cassandra/config/KSMetaData.java
+++ b/src/java/org/apache/cassandra/config/KSMetaData.java
@@ -89,7 +89,6 @@ public final class KSMetaData
                                                 CFMetaData.SchemaKeyspacesCf,
                                                 CFMetaData.SchemaColumnFamiliesCf,
                                                 CFMetaData.SchemaColumnsCf,
-                                                CFMetaData.CompactionLogCF,
                                                 CFMetaData.OldStatusCf,
                                                 CFMetaData.OldHintsCf,
                                                 CFMetaData.OldMigrationsCf,

http://git-wip-us.apache.org/repos/asf/cassandra/blob/7ff2805c/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index 0769d5c..6043be5 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -61,7 +61,6 @@ import org.apache.cassandra.db.index.SecondaryIndexManager;
 import org.apache.cassandra.db.marshal.AbstractType;
 import org.apache.cassandra.dht.*;
 import org.apache.cassandra.exceptions.ConfigurationException;
-import org.apache.cassandra.io.FSReadError;
 import org.apache.cassandra.io.compress.CompressionParameters;
 import org.apache.cassandra.io.sstable.*;
 import org.apache.cassandra.io.sstable.Descriptor;
@@ -242,7 +241,33 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
         {
             Directories.SSTableLister sstableFiles = directories.sstableLister().skipTemporary(true);
             Collection<SSTableReader> sstables = SSTableReader.batchOpen(sstableFiles.list().entrySet(),
metadata, this.partitioner);
-            data.addInitialSSTables(sstables);
+            if (metadata.getDefaultValidator().isCommutative())
+            {
+                // Filter non-compacted sstables, remove compacted ones
+                Set<Integer> compactedSSTables = new HashSet<Integer>();
+                for (SSTableReader sstable : sstables)
+                    compactedSSTables.addAll(sstable.getAncestors());
+
+                Set<SSTableReader> liveSSTables = new HashSet<SSTableReader>();
+                for (SSTableReader sstable : sstables)
+                {
+                    if (compactedSSTables.contains(sstable.descriptor.generation))
+                    {
+                        logger.info("{} is already compacted and will be removed.", sstable);
+                        sstable.markCompacted(); // we need to mark as compacted to be deleted
+                        sstable.releaseReference(); // this amount to deleting the sstable
+                    }
+                    else
+                    {
+                        liveSSTables.add(sstable);
+                    }
+                }
+                data.addInitialSSTables(liveSSTables);
+            }
+            else
+            {
+                data.addInitialSSTables(sstables);
+            }
         }
 
         if (caching == Caching.ALL || caching == Caching.KEYS_ONLY)
@@ -419,72 +444,6 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
         }
     }
 
-    /**
-     * Replacing compacted sstables is atomic as far as observers of DataTracker are concerned,
but not on the
-     * filesystem: first the new sstables are renamed to "live" status (i.e., the tmp marker
is removed), then
-     * their ancestors are removed.
-     *
-     * If an unclean shutdown happens at the right time, we can thus end up with both the
new ones and their
-     * ancestors "live" in the system.  This is harmless for normal data, but for counters
it can cause overcounts.
-     *
-     * To prevent this, we record sstables being compacted in the system keyspace.  If we
find unfinished
-     * compactions, we remove the new ones (since those may be incomplete -- under LCS, we
may create multiple
-     * sstables from any given ancestor).
-     */
-    public static void removeUnfinishedCompactionLeftovers(String keyspace, String columnfamily,
Set<Integer> unfinishedGenerations)
-    {
-        Directories directories = Directories.create(keyspace, columnfamily);
-
-        // sanity-check unfinishedGenerations
-        Set<Integer> allGenerations = new HashSet<Integer>();
-        for (Descriptor desc : directories.sstableLister().list().keySet())
-            allGenerations.add(desc.generation);
-        if (!allGenerations.containsAll(unfinishedGenerations))
-        {
-            throw new IllegalStateException("Unfinished compactions reference missing sstables."
-                                            + " This should never happen since compactions
are marked finished before we start removing the old sstables.");
-        }
-
-        // remove new sstables from compactions that didn't complete, and compute
-        // set of ancestors that shouldn't exist anymore
-        Set<Integer> completedAncestors = new HashSet<Integer>();
-        for (Map.Entry<Descriptor, Set<Component>> sstableFiles : directories.sstableLister().list().entrySet())
-        {
-            Descriptor desc = sstableFiles.getKey();
-            Set<Component> components = sstableFiles.getValue();
-
-            SSTableMetadata meta;
-            try
-            {
-                meta = SSTableMetadata.serializer.deserialize(desc);
-            }
-            catch (IOException e)
-            {
-                throw new FSReadError(e, desc.filenameFor(Component.STATS));
-            }
-
-            Set<Integer> ancestors = meta.ancestors;
-            if (!ancestors.isEmpty() && unfinishedGenerations.containsAll(ancestors))
-            {
-                SSTable.delete(desc, components);
-            }
-            else
-            {
-                completedAncestors.addAll(ancestors);
-            }
-        }
-
-        // remove old sstables from compactions that did complete
-        for (Map.Entry<Descriptor, Set<Component>> sstableFiles : directories.sstableLister().list().entrySet())
-        {
-            Descriptor desc = sstableFiles.getKey();
-            Set<Component> components = sstableFiles.getValue();
-
-            if (completedAncestors.contains(desc.generation))
-                SSTable.delete(desc, components);
-        }
-    }
-
     // must be called after all sstables are loaded since row cache merges all row versions
     public void initRowCache()
     {

http://git-wip-us.apache.org/repos/asf/cassandra/blob/7ff2805c/src/java/org/apache/cassandra/db/SystemTable.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/SystemTable.java b/src/java/org/apache/cassandra/db/SystemTable.java
index adf08cc..0422a21 100644
--- a/src/java/org/apache/cassandra/db/SystemTable.java
+++ b/src/java/org/apache/cassandra/db/SystemTable.java
@@ -73,7 +73,6 @@ public class SystemTable
     public static final String SCHEMA_KEYSPACES_CF = "schema_keyspaces";
     public static final String SCHEMA_COLUMNFAMILIES_CF = "schema_columnfamilies";
     public static final String SCHEMA_COLUMNS_CF = "schema_columns";
-    public static final String COMPACTION_LOG = "compactions_in_progress";
 
     @Deprecated
     public static final String OLD_STATUS_CF = "LocationInfo";
@@ -184,74 +183,6 @@ public class SystemTable
         }
     }
 
-    /**
-     * Write compaction log, except columfamilies under system keyspace.
-     *
-     * @param cfs
-     * @param toCompact sstables to compact
-     * @return compaction task id or null if cfs is under system keyspace
-     */
-    public static UUID startCompaction(ColumnFamilyStore cfs, Iterable<SSTableReader>
toCompact)
-    {
-        if (Table.SYSTEM_KS.equals(cfs.table.name))
-            return null;
-
-        UUID compactionId = UUIDGen.getTimeUUID();
-        String req = "INSERT INTO system.%s (id, keyspace_name, columnfamily_name, inputs)
VALUES (%s, '%s', '%s', {%s})";
-        Iterable<Integer> generations = Iterables.transform(toCompact, new Function<SSTableReader,
Integer>()
-        {
-            public Integer apply(SSTableReader sstable)
-            {
-                return sstable.descriptor.generation;
-            }
-        });
-        processInternal(String.format(req, COMPACTION_LOG, compactionId, cfs.table.name,
cfs.columnFamily, StringUtils.join(Sets.newHashSet(generations), ',')));
-        forceBlockingFlush(COMPACTION_LOG);
-        return compactionId;
-    }
-
-    public static void finishCompaction(UUID taskId)
-    {
-        assert taskId != null;
-
-        String req = "DELETE FROM system.%s WHERE id = %s";
-        processInternal(String.format(req, COMPACTION_LOG, taskId));
-        forceBlockingFlush(COMPACTION_LOG);
-    }
-
-    /**
-     * @return unfinished compactions, grouped by keyspace/columnfamily pair.
-     */
-    public static SetMultimap<Pair<String, String>, Integer> getUnfinishedCompactions()
-    {
-        String req = "SELECT * FROM system.%s";
-        UntypedResultSet resultSet = processInternal(String.format(req, COMPACTION_LOG));
-
-        SetMultimap<Pair<String, String>, Integer> unfinishedCompactions = HashMultimap.create();
-        for (UntypedResultSet.Row row : resultSet)
-        {
-            String keyspace = row.getString("keyspace_name");
-            String columnfamily = row.getString("columnfamily_name");
-            Set<Integer> inputs = row.getSet("inputs", Int32Type.instance);
-
-            unfinishedCompactions.putAll(Pair.create(keyspace, columnfamily), inputs);
-        }
-        return unfinishedCompactions;
-    }
-
-    public static void discardCompactionsInProgress()
-    {
-        ColumnFamilyStore compactionLog = Table.open(Table.SYSTEM_KS).getColumnFamilyStore(COMPACTION_LOG);
-        try
-        {
-            compactionLog.truncate().get();
-        }
-        catch (Exception e)
-        {
-            throw new RuntimeException(e);
-        }
-    }
-
     public static void saveTruncationPosition(ColumnFamilyStore cfs, ReplayPosition position)
     {
         String req = "UPDATE system.%s SET truncated_at = truncated_at + %s WHERE key = '%s'";

http://git-wip-us.apache.org/repos/asf/cassandra/blob/7ff2805c/src/java/org/apache/cassandra/db/compaction/CompactionTask.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java
index 1b1599b..3893f1c 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java
@@ -103,8 +103,6 @@ public class CompactionTask extends AbstractCompactionTask
         for (SSTableReader sstable : toCompact)
             assert sstable.descriptor.cfname.equals(cfs.columnFamily);
 
-        UUID taskId = SystemTable.startCompaction(cfs, toCompact);
-
         CompactionController controller = new CompactionController(cfs, toCompact, gcBefore);
         // new sstables from flush can be added during a compaction, but only the compaction
can remove them,
         // so in our single-threaded compaction world this is a valid way of determining
if we're compacting
@@ -230,11 +228,6 @@ public class CompactionTask extends AbstractCompactionTask
                 throw new RuntimeException(e);
             }
 
-            // point of no return -- the new sstables are live on disk; next we'll start
deleting the old ones
-            // (in replaceCompactedSSTables)
-            if (taskId != null)
-                SystemTable.finishCompaction(taskId);
-
             if (collector != null)
                 collector.finishCompaction(ci);
         }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/7ff2805c/src/java/org/apache/cassandra/service/CassandraDaemon.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java
index 0a7b957..7d78196 100644
--- a/src/java/org/apache/cassandra/service/CassandraDaemon.java
+++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java
@@ -27,7 +27,6 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import com.google.common.collect.Iterables;
-import com.google.common.collect.SetMultimap;
 import org.apache.log4j.PropertyConfigurator;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -44,7 +43,6 @@ import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.thrift.ThriftServer;
 import org.apache.cassandra.utils.CLibrary;
 import org.apache.cassandra.utils.Mx4jTool;
-import org.apache.cassandra.utils.Pair;
 
 /**
  * The <code>CassandraDaemon</code> is an abstraction for a Cassandra daemon
@@ -202,13 +200,6 @@ public class CassandraDaemon
                 ColumnFamilyStore.scrubDataDirectories(table, cfm.cfName);
             }
         }
-        // clean up compaction leftovers
-        SetMultimap<Pair<String, String>, Integer> unfinishedCompactions = SystemTable.getUnfinishedCompactions();
-        for (Pair<String, String> kscf : unfinishedCompactions.keySet())
-        {
-            ColumnFamilyStore.removeUnfinishedCompactionLeftovers(kscf.left, kscf.right,
unfinishedCompactions.get(kscf));
-        }
-        SystemTable.discardCompactionsInProgress();
 
         // initialize keyspaces
         for (String table : Schema.instance.getTables())

http://git-wip-us.apache.org/repos/asf/cassandra/blob/7ff2805c/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java b/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java
index 8048fcb..8318f0d 100644
--- a/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java
+++ b/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java
@@ -66,7 +66,6 @@ import org.apache.cassandra.io.sstable.*;
 import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.thrift.*;
 import org.apache.cassandra.utils.ByteBufferUtil;
-import org.apache.cassandra.utils.UUIDGen;
 import org.apache.cassandra.utils.WrappedRunnable;
 
 public class ColumnFamilyStoreTest extends SchemaLoader
@@ -1241,60 +1240,6 @@ public class ColumnFamilyStoreTest extends SchemaLoader
         testMultiRangeSlicesBehavior(prepareMultiRangeSlicesTest(10, false));
     }
 
-    @Test
-    public void testRemoveUnifinishedCompactionLeftovers() throws Throwable
-    {
-        String ks = "Keyspace1";
-        String cf = "Standard3"; // should be empty
-
-        CFMetaData cfmeta = Schema.instance.getCFMetaData(ks, cf);
-        Directories dir = Directories.create(ks, cf);
-        ByteBuffer key = bytes("key");
-
-        // 1st sstable
-        SSTableSimpleWriter writer = new SSTableSimpleWriter(dir.getDirectoryForNewSSTables(100),
-                                                             cfmeta, StorageService.getPartitioner());
-        writer.newRow(key);
-        writer.addColumn(bytes("col"), bytes("val"), 1);
-        writer.close();
-
-        Map<Descriptor, Set<Component>> sstables = dir.sstableLister().list();
-        assert sstables.size() == 1;
-
-        Map.Entry<Descriptor, Set<Component>> sstableToOpen = sstables.entrySet().iterator().next();
-        final SSTableReader sstable1 = SSTableReader.open(sstableToOpen.getKey());
-
-        // simulate incomplete compaction
-        writer = new SSTableSimpleWriter(dir.getDirectoryForNewSSTables(100),
-                                                             cfmeta, StorageService.getPartitioner())
-        {
-            protected SSTableWriter getWriter()
-            {
-                SSTableMetadata.Collector collector = SSTableMetadata.createCollector();
-                collector.addAncestor(sstable1.descriptor.generation); // add ancestor from
previously written sstable
-                return new SSTableWriter(makeFilename(directory, metadata.ksName, metadata.cfName),
-                                         0,
-                                         metadata,
-                                         StorageService.getPartitioner(),
-                                         collector);
-            }
-        };
-        writer.newRow(key);
-        writer.addColumn(bytes("col"), bytes("val"), 1);
-        writer.close();
-
-        // should have 2 sstables now
-        sstables = dir.sstableLister().list();
-        assert sstables.size() == 2;
-
-        ColumnFamilyStore.removeUnfinishedCompactionLeftovers(ks, cf, Sets.newHashSet(sstable1.descriptor.generation));
-
-        // 2nd sstable should be removed (only 1st sstable exists in set of size 1)
-        sstables = dir.sstableLister().list();
-        assert sstables.size() == 1;
-        assert sstables.containsKey(sstable1.descriptor);
-    }
-
     private ColumnFamilyStore prepareMultiRangeSlicesTest(int valueSize, boolean flush) throws
Throwable
     {
         String tableName = "Keyspace1";

http://git-wip-us.apache.org/repos/asf/cassandra/blob/7ff2805c/test/unit/org/apache/cassandra/db/compaction/CompactionsTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionsTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionsTest.java
index b41bf19..a09451e 100644
--- a/test/unit/org/apache/cassandra/db/compaction/CompactionsTest.java
+++ b/test/unit/org/apache/cassandra/db/compaction/CompactionsTest.java
@@ -24,10 +24,6 @@ import java.util.*;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 
-import com.google.common.base.Function;
-import com.google.common.collect.Iterables;
-import com.google.common.collect.SetMultimap;
-import com.google.common.collect.Sets;
 import org.junit.Test;
 
 import org.apache.cassandra.SchemaLoader;
@@ -44,7 +40,6 @@ import org.apache.cassandra.io.sstable.SSTableScanner;
 import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.FBUtilities;
-import org.apache.cassandra.utils.Pair;
 
 import static junit.framework.Assert.*;
 
@@ -280,35 +275,6 @@ public class CompactionsTest extends SchemaLoader
         assert sstables.iterator().next().descriptor.generation == prevGeneration + 1;
     }
 
-    @Test
-    public void testCompactionLog() throws Exception
-    {
-        SystemTable.discardCompactionsInProgress();
-
-        String cf = "Standard4";
-        ColumnFamilyStore cfs = Table.open(TABLE1).getColumnFamilyStore(cf);
-        insertData(TABLE1, cf, 0, 1);
-        cfs.forceBlockingFlush();
-
-        Collection<SSTableReader> sstables = cfs.getSSTables();
-        assert !sstables.isEmpty();
-        Set<Integer> generations = Sets.newHashSet(Iterables.transform(sstables, new
Function<SSTableReader, Integer>()
-        {
-            public Integer apply(SSTableReader sstable)
-            {
-                return sstable.descriptor.generation;
-            }
-        }));
-        UUID taskId = SystemTable.startCompaction(cfs, sstables);
-        SetMultimap<Pair<String, String>, Integer> compactionLogs = SystemTable.getUnfinishedCompactions();
-        Set<Integer> unfinishedCompactions = compactionLogs.get(Pair.create(TABLE1,
cf));
-        assert unfinishedCompactions.containsAll(generations);
-
-        SystemTable.finishCompaction(taskId);
-        compactionLogs = SystemTable.getUnfinishedCompactions();
-        assert !compactionLogs.containsKey(Pair.create(TABLE1, cf));
-    }
-
     private void testDontPurgeAccidentaly(String k, String cfname, boolean forceDeserialize)
throws IOException, ExecutionException, InterruptedException
     {
         // This test catches the regression of CASSANDRA-2786


Mime
View raw message