Fix resurrection by hanging ref and deletion by invalidated ancestry

src/java/org/apache/cassandra/db/ColumnFamilyStore.java

@@ -49,6 +49,7 @@
 import org.apache.cassandra.db.lifecycle.Tracker;
 import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
 import org.apache.cassandra.io.FSWriteError;
+import org.apache.cassandra.io.sstable.metadata.MetadataComponent;
 import org.json.simple.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -699,8 +700,15 @@ public static void removeUnfinishedCompactionLeftovers(CFMetaData metadata, Map<
             Set<Integer> ancestors;
             try
             {
-                CompactionMetadata compactionMetadata = (CompactionMetadata) desc.getMetadataSerializer().deserialize(desc, MetadataType.COMPACTION);
-                ancestors = compactionMetadata.ancestors;
+                Map<MetadataType, MetadataComponent> compactionMetadata = desc.getMetadataSerializer().deserialize(desc, EnumSet.of(MetadataType.COMPACTION, MetadataType.VALID_ANCESTORS));
+                if (compactionMetadata.get(MetadataType.VALID_ANCESTORS) != null)
+                {
+                    ancestors = ((CompactionMetadata) compactionMetadata.get(MetadataType.COMPACTION)).ancestors;
+                }
+                else
+                {
+                    ancestors = Collections.emptySet();
+                }
             }
             catch (IOException e)
             {
@@ -735,8 +743,15 @@ public static void removeUnfinishedCompactionLeftovers(CFMetaData metadata, Map<
             if (completedAncestors.contains(desc.generation))
             {
                 // if any of the ancestors were participating in a compaction, finish that compaction
-                logger.info("Going to delete leftover compaction ancestor {}", desc);
-                SSTable.delete(desc, sstableFiles.getValue());
+                if (Boolean.getBoolean("palantir_cassandra.dry_run_ancestor_deletion"))
+                {
+                    logger.info("Would have deleted leftover compaction ancestor {} if palantir_cassandra.dry_run_ancestor_deletion was false", desc);
+                }
+                else
+                {
+                    logger.info("Going to delete leftover compaction ancestor {}", desc);
+                    SSTable.delete(desc, sstableFiles.getValue());
+                }
                 UUID compactionTaskID = unfinishedCompactions.get(desc.generation);
                 if (compactionTaskID != null)
                     SystemKeyspace.finishCompaction(unfinishedCompactions.get(desc.generation));
@@ -833,6 +848,7 @@ public synchronized int loadNewSSTablesWithCount(boolean assumeCfIsEmpty)
         for (Map.Entry<Descriptor, Set<Component>> entry : lister.list().entrySet())
         {
             Descriptor descriptor = entry.getKey();
+            Set<Component> components = entry.getValue();
 
             if (currentDescriptors.contains(descriptor))
                 continue; // old (initialized) SSTable found, skipping
@@ -856,28 +872,39 @@ public synchronized int loadNewSSTablesWithCount(boolean assumeCfIsEmpty)
                 continue;
             }
 
-            // Increment the generation until we find a filename that doesn't exist. This is needed because the new
-            // SSTables that are being loaded might already use these generation numbers.
             Descriptor newDescriptor;
-            do
+            if (assumeCfIsEmpty)
             {
-                newDescriptor = new Descriptor(descriptor.version,
-                                               descriptor.directory,
-                                               descriptor.ksname,
-                                               descriptor.cfname,
-                                               fileIndexGenerator.incrementAndGet(),
-                                               Descriptor.Type.FINAL,
-                                               descriptor.formatType);
+                newDescriptor = descriptor;
             }
-            while (new File(newDescriptor.filenameFor(Component.DATA)).exists());
+            else
+            {
+                // Increment the generation until we find a filename that doesn't exist. This is needed because the new
+                // SSTables that are being loaded might already use these generation numbers.
+                do
+                {
+                    newDescriptor = new Descriptor(descriptor.version,
+                                                   descriptor.directory,
+                                                   descriptor.ksname,
+                                                   descriptor.cfname,
+                                                   fileIndexGenerator.incrementAndGet(),
+                                                   Descriptor.Type.FINAL,
+                                                   descriptor.formatType);
+                }
+                while (new File(newDescriptor.filenameFor(Component.DATA)).exists());
 
-            logger.info("Renaming new SSTable {} to {}", descriptor, newDescriptor);
-            SSTableWriter.rename(descriptor, newDescriptor, entry.getValue());
+                logger.info("Removing Statistics.db for new SSTable {} to clear old ancestor metadata", descriptor);
+                FileUtils.delete(new File(descriptor.filenameFor(Component.STATS)));
+                components = Sets.difference(components, ImmutableSet.of(Component.STATS));
+
+                logger.info("Renaming new SSTable {} to {}", descriptor, newDescriptor);
+                SSTableWriter.rename(descriptor, newDescriptor, components);
+            }
 
             SSTableReader reader;
             try
             {
-                reader = SSTableReader.open(newDescriptor, entry.getValue(), metadata, partitioner);
+                reader = SSTableReader.open(newDescriptor, components, metadata, partitioner);
             }
             catch (IOException e)
             {

src/java/org/apache/cassandra/io/sstable/metadata/MetadataCollector.java

@@ -295,6 +295,7 @@ public Map<MetadataType, MetadataComponent> finalizeMetadata(String partitioner,
                                                              hasLegacyCounterShards,
                                                              repairedAt));
         components.put(MetadataType.COMPACTION, new CompactionMetadata(ancestors, cardinality));
+        components.put(MetadataType.VALID_ANCESTORS, ValidAncestorsMetadata.instance);
         return components;
     }
 }

src/java/org/apache/cassandra/io/sstable/metadata/MetadataSerializer.java

@@ -122,8 +122,8 @@ public Map<MetadataType, MetadataComponent> deserialize(Descriptor descriptor, F
             {
                 in.seek(offset);
                 component = type.serializer.deserialize(descriptor.version, in);
+                components.put(type, component);
             }
-            components.put(type, component);
         }
         return components;
     }

src/java/org/apache/cassandra/io/sstable/metadata/MetadataType.java

@@ -27,7 +27,9 @@ public enum MetadataType
     /** Metadata only used at compaction */
     COMPACTION(CompactionMetadata.serializer),
     /** Metadata always keep in memory */
-    STATS(StatsMetadata.serializer);
+    STATS(StatsMetadata.serializer),
+    /** Meta-metadata about whether the ancestors metadata is valid **/
+    VALID_ANCESTORS(ValidAncestorsMetadata.serializer);
 
     public final IMetadataComponentSerializer<MetadataComponent> serializer;
 

src/java/org/apache/cassandra/io/sstable/metadata/ValidAncestorsMetadata.java

@@ -0,0 +1,58 @@
+/*
+ * 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.io.sstable.metadata;
+
+import java.io.DataInput;
+
+import org.apache.cassandra.io.sstable.format.Version;
+import org.apache.cassandra.io.util.DataOutputPlus;
+
+/**
+ * Marker metadata component indicating this SSTable has valid compaction ancestor information.
+ */
+public class ValidAncestorsMetadata extends MetadataComponent
+{
+    public static final IMetadataComponentSerializer serializer = new ValidAncestorsMetadataSerializer();
+    public static final ValidAncestorsMetadata instance = new ValidAncestorsMetadata();
+
+    private ValidAncestorsMetadata() {}
+
+    public MetadataType getType()
+    {
+        return MetadataType.VALID_ANCESTORS;
+    }
+
+    public static class ValidAncestorsMetadataSerializer implements IMetadataComponentSerializer<ValidAncestorsMetadata>
+    {
+        @Override
+        public int serializedSize(ValidAncestorsMetadata component, Version version)
+        {
+            return 0;
+        }
+
+        @Override
+        public void serialize(ValidAncestorsMetadata component, Version version, DataOutputPlus out) {}
+
+        @Override
+        public ValidAncestorsMetadata deserialize(Version version, DataInput in)
+        {
+            return instance;
+        }
+    }
+}

src/java/org/apache/cassandra/tools/SSTableMetadataViewer.java

@@ -54,6 +54,7 @@ public static void main(String[] args) throws IOException
                 ValidationMetadata validation = (ValidationMetadata) metadata.get(MetadataType.VALIDATION);
                 StatsMetadata stats = (StatsMetadata) metadata.get(MetadataType.STATS);
                 CompactionMetadata compaction = (CompactionMetadata) metadata.get(MetadataType.COMPACTION);
+                ValidAncestorsMetadata validAncestors = (ValidAncestorsMetadata) metadata.get(MetadataType.VALID_ANCESTORS);
 
                 out.printf("SSTable: %s%n", descriptor);
                 if (validation != null)
@@ -81,7 +82,15 @@ public static void main(String[] args) throws IOException
                 }
                 if (compaction != null)
                 {
-                    out.printf("Ancestors: %s%n", compaction.ancestors.toString());
+                    out.printf("Ancestors: %s", compaction.ancestors.toString());
+                    if (validAncestors != null)
+                    {
+                        out.println(" (considered valid)");
+                    }
+                    else
+                    {
+                        out.println(" (considered invalid)");
+                    }
                     out.printf("Estimated cardinality: %s%n", compaction.cardinalityEstimator.cardinality());
 
                 }

test/unit/org/apache/cassandra/io/sstable/format/SSTableReaderTest.java

@@ -196,26 +196,6 @@ public void testSpannedIndexPositions() throws IOException
         }
     }
 
-    @Test
-    public void testPersistentStatistics()
-    {
-
-        Keyspace keyspace = Keyspace.open(KEYSPACE1);
-        ColumnFamilyStore store = keyspace.getColumnFamilyStore("Standard1");
-
-        for (int j = 0; j < 100; j += 2)
-        {
-            ByteBuffer key = ByteBufferUtil.bytes(String.valueOf(j));
-            Mutation rm = new Mutation(KEYSPACE1, key);
-            rm.add("Standard1", cellname("0"), ByteBufferUtil.EMPTY_BYTE_BUFFER, j);
-            rm.applyUnsafe();
-        }
-        store.forceBlockingFlush();
-
-        clearAndLoad(store);
-        assert store.metric.maxRowSize.getValue() != 0;
-    }
-
     private void clearAndLoad(ColumnFamilyStore cfs)
     {
         cfs.clearUnsafe();