Checksum sstable metadata

Patch by marcuse; reviewed by Jason Brown for CASSANDRA-13321

CHANGES.txt

@@ -1,4 +1,5 @@
 4.0
+ * Checksum sstable metadata (CASSANDRA-13321)
  * Expose recent histograms in JmxHistograms (CASSANDRA-13642)
  * Fix buffer length comparison when decompressing in netty-based streaming (CASSANDRA-13899)
  * Properly close StreamCompressionInputStream to release any ByteBuf (CASSANDRA-13906)

src/java/org/apache/cassandra/io/sstable/format/Version.java

@@ -55,6 +55,8 @@ public abstract class Version
 
     public abstract boolean hasPendingRepair();
 
+    public abstract boolean hasMetadataChecksum();
+
     public String getVersion()
     {
         return version;

src/java/org/apache/cassandra/io/sstable/format/big/BigFormat.java

@@ -120,7 +120,7 @@ public class BigFormat implements SSTableFormat
         // mb (3.0.7, 3.7): commit log lower bound included
         // mc (3.0.8, 3.9): commit log intervals included
 
-        // na (4.0.0): uncompressed chunks, pending repair session
+        // na (4.0.0): uncompressed chunks, pending repair session, checksummed sstable metadata file
         //
         // NOTE: when adding a new version, please add that to LegacySSTableTest, too.
 
@@ -130,6 +130,7 @@ public class BigFormat implements SSTableFormat
         private final boolean hasCommitLogIntervals;
         public final boolean hasMaxCompressedLength;
         private final boolean hasPendingRepair;
+        private final boolean hasMetadataChecksum;
 
         BigVersion(String version)
         {
@@ -142,6 +143,7 @@ public class BigFormat implements SSTableFormat
             hasCommitLogIntervals = version.compareTo("mc") >= 0;
             hasMaxCompressedLength = version.compareTo("na") >= 0;
             hasPendingRepair = version.compareTo("na") >= 0;
+            hasMetadataChecksum = version.compareTo("na") >= 0;
         }
 
         @Override
@@ -174,6 +176,11 @@ public class BigFormat implements SSTableFormat
         }
 
         @Override
+        public boolean hasMetadataChecksum()
+        {
+            return hasMetadataChecksum;
+        }
+
         public boolean isCompatible()
         {
             return version.compareTo(earliest_supported_version) >= 0 && version.charAt(0) <= current_version.charAt(0);

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

@@ -21,12 +21,18 @@ import java.io.*;
 import java.util.*;
 
 import com.google.common.collect.Lists;
+import com.google.common.hash.HashFunction;
+import com.google.common.hash.Hasher;
+import com.google.common.hash.Hashing;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.apache.cassandra.io.sstable.Component;
+import org.apache.cassandra.io.sstable.CorruptSSTableException;
 import org.apache.cassandra.io.sstable.Descriptor;
 import org.apache.cassandra.io.sstable.format.Version;
+import org.apache.cassandra.io.util.DataInputBuffer;
+import org.apache.cassandra.io.util.DataOutputBuffer;
 import org.apache.cassandra.io.util.DataOutputPlus;
 import org.apache.cassandra.io.util.DataOutputStreamPlus;
 import org.apache.cassandra.io.util.FileDataInput;
@@ -39,7 +45,7 @@ import org.apache.cassandra.utils.FBUtilities;
  * Metadata serializer for SSTables {@code version >= 'k'}.
  *
  * <pre>
- * File format := | number of components (4 bytes) | toc | component1 | component2 | ... |
+ * File format := | number of components (4 bytes) | toc | component1 | c1 hash | component2 | c2 hash | ... |
  * toc         := | component type (4 bytes) | position of component |
  * </pre>
  *
@@ -48,9 +54,11 @@ import org.apache.cassandra.utils.FBUtilities;
 public class MetadataSerializer implements IMetadataSerializer
 {
     private static final Logger logger = LoggerFactory.getLogger(MetadataSerializer.class);
+    private static final HashFunction hashFunction = Hashing.md5();
 
     public void serialize(Map<MetadataType, MetadataComponent> components, DataOutputPlus out, Version version) throws IOException
     {
+        boolean checksum = version.hasMetadataChecksum();
         // sort components by type
         List<MetadataComponent> sortedComponents = Lists.newArrayList(components.values());
         Collections.sort(sortedComponents);
@@ -59,6 +67,7 @@ public class MetadataSerializer implements IMetadataSerializer
         out.writeInt(components.size());
         // build and write toc
         int lastPosition = 4 + (8 * sortedComponents.size());
+        Map<MetadataType, Integer> sizes = new EnumMap<>(MetadataType.class);
         for (MetadataComponent component : sortedComponents)
         {
             MetadataType type = component.getType();
@@ -66,12 +75,22 @@ public class MetadataSerializer implements IMetadataSerializer
             out.writeInt(type.ordinal());
             // serialize position
             out.writeInt(lastPosition);
-            lastPosition += type.serializer.serializedSize(version, component);
+            int size = type.serializer.serializedSize(version, component);
+            lastPosition += size + (checksum ? 8 : 0); // checksum is long
+            sizes.put(type, size);
         }
         // serialize components
         for (MetadataComponent component : sortedComponents)
         {
-            component.getType().serializer.serialize(version, component, out);
+            byte[] bytes;
+            try (DataOutputBuffer dob = new DataOutputBuffer(sizes.get(component.getType())))
+            {
+                component.getType().serializer.serialize(version, component, dob);
+                bytes = dob.getData();
+            }
+            out.write(bytes);
+            if (checksum)
+                out.writeLong(hashFunction.hashBytes(bytes).asLong());
         }
     }
 
@@ -103,24 +122,59 @@ public class MetadataSerializer implements IMetadataSerializer
 
     public Map<MetadataType, MetadataComponent> deserialize(Descriptor descriptor, FileDataInput in, EnumSet<MetadataType> types) throws IOException
     {
+        int totalSize = (int) in.bytesRemaining();
         Map<MetadataType, MetadataComponent> components = new EnumMap<>(MetadataType.class);
         // read number of components
         int numComponents = in.readInt();
         // read toc
         Map<MetadataType, Integer> toc = new EnumMap<>(MetadataType.class);
         MetadataType[] values = MetadataType.values();
+        Map<MetadataType, Integer> lengths = new EnumMap<>(MetadataType.class);
+        int start = 0;
+        MetadataType lastType = null;
         for (int i = 0; i < numComponents; i++)
         {
-            toc.put(values[in.readInt()], in.readInt());
+            int metadataTypeId = in.readInt();
+            int position = in.readInt();
+
+            toc.put(values[metadataTypeId], position);
+            if (lastType != null)
+                lengths.put(lastType, position - start);
+            start = position;
+            lastType = values[metadataTypeId];
         }
+        lengths.put(lastType, totalSize - start);
         for (MetadataType type : types)
         {
             Integer offset = toc.get(type);
             if (offset != null)
             {
                 in.seek(offset);
-                MetadataComponent component = type.serializer.deserialize(descriptor.version, in);
-                components.put(type, component);
+
+                if (descriptor.version.hasMetadataChecksum())
+                {
+                    int size = lengths.get(type) - 8; // 8 bytes checksum
+                    byte[] bytes = new byte[size];
+                    in.readFully(bytes);
+                    MetadataComponent component;
+                    try (DataInputBuffer dib = new DataInputBuffer(bytes))
+                    {
+                        component = type.serializer.deserialize(descriptor.version, dib);
+                    }
+                    long writtenChecksum = in.readLong();
+                    if (writtenChecksum != hashFunction.hashBytes(bytes).asLong())
+                    {
+                        String filename = descriptor.filenameFor(Component.STATS);
+                        throw new CorruptSSTableException(new IOException("Checksums do not match for " + filename), filename);
+                    }
+                    components.put(type, component);
+                }
+                else
+                {
+                    MetadataComponent component = type.serializer.deserialize(descriptor.version, in);
+                    components.put(type, component);
+                }
+
             }
         }
         return components;

test/data/legacy-sstables/na/legacy_tables/legacy_na_clust/na-1-big-Digest.crc32

@@ -1 +1 @@
-1337396261
+829376603

test/data/legacy-sstables/na/legacy_tables/legacy_na_clust/na-1-big-TOC.txt

@@ -1,8 +1,8 @@
-Summary.db
-Filter.db
 Index.db
-CompressionInfo.db
 Data.db
-TOC.txt
+CompressionInfo.db
 Statistics.db
+Summary.db
+TOC.txt
 Digest.crc32
+Filter.db

test/data/legacy-sstables/na/legacy_tables/legacy_na_clust_compact/na-1-big-Digest.crc32

@@ -1 +1 @@
-2454456604
+818525950

test/data/legacy-sstables/na/legacy_tables/legacy_na_clust_compact/na-1-big-TOC.txt

@@ -1,8 +1,8 @@
-Summary.db
-Filter.db
 Index.db
-CompressionInfo.db
 Data.db
-TOC.txt
+CompressionInfo.db
 Statistics.db
+Summary.db
+TOC.txt
 Digest.crc32
+Filter.db

test/data/legacy-sstables/na/legacy_tables/legacy_na_clust_counter/na-1-big-Digest.crc32

@@ -1 +1 @@
-41575897
+1278737554

test/data/legacy-sstables/na/legacy_tables/legacy_na_clust_counter/na-1-big-TOC.txt

@@ -1,8 +1,8 @@
-Summary.db
-Filter.db
 Index.db
-CompressionInfo.db
 Data.db
-TOC.txt
+CompressionInfo.db
 Statistics.db
+Summary.db
+TOC.txt
 Digest.crc32
+Filter.db

test/data/legacy-sstables/na/legacy_tables/legacy_na_clust_counter_compact/na-1-big-Digest.crc32

@@ -1 +1 @@
-4050499673
+2193718788

test/data/legacy-sstables/na/legacy_tables/legacy_na_clust_counter_compact/na-1-big-TOC.txt

@@ -1,8 +1,8 @@
-Summary.db
-Filter.db
 Index.db
-CompressionInfo.db
 Data.db
-TOC.txt
+CompressionInfo.db
 Statistics.db
+Summary.db
+TOC.txt
 Digest.crc32
+Filter.db

test/data/legacy-sstables/na/legacy_tables/legacy_na_simple/na-1-big-Digest.crc32

@@ -1 +1 @@
-243065670
+3074491452

test/data/legacy-sstables/na/legacy_tables/legacy_na_simple/na-1-big-TOC.txt

@@ -1,8 +1,8 @@
-Summary.db
-Filter.db
 Index.db
-CompressionInfo.db
 Data.db
-TOC.txt
+CompressionInfo.db
 Statistics.db
+Summary.db
+TOC.txt
 Digest.crc32
+Filter.db

test/data/legacy-sstables/na/legacy_tables/legacy_na_simple_compact/na-1-big-Digest.crc32

@@ -1 +1 @@
-576034138
+1825853914

test/data/legacy-sstables/na/legacy_tables/legacy_na_simple_compact/na-1-big-TOC.txt

@@ -1,8 +1,8 @@
-Summary.db
-Filter.db
 Index.db
-CompressionInfo.db
 Data.db
-TOC.txt
+CompressionInfo.db
 Statistics.db
+Summary.db
+TOC.txt
 Digest.crc32
+Filter.db

test/data/legacy-sstables/na/legacy_tables/legacy_na_simple_counter/na-1-big-Digest.crc32

@@ -1 +1 @@
-381306786
+3361465012

test/data/legacy-sstables/na/legacy_tables/legacy_na_simple_counter/na-1-big-TOC.txt

@@ -1,8 +1,8 @@
-Summary.db
-Filter.db
 Index.db
-CompressionInfo.db
 Data.db
-TOC.txt
+CompressionInfo.db
 Statistics.db
+Summary.db
+TOC.txt
 Digest.crc32
+Filter.db

test/data/legacy-sstables/na/legacy_tables/legacy_na_simple_counter_compact/na-1-big-Digest.crc32

@@ -1 +1 @@
-2161889354
+2410392183

test/data/legacy-sstables/na/legacy_tables/legacy_na_simple_counter_compact/na-1-big-TOC.txt

@@ -1,8 +1,8 @@
-Summary.db
-Filter.db
 Index.db
-CompressionInfo.db
 Data.db
-TOC.txt
+CompressionInfo.db
 Statistics.db
+Summary.db
+TOC.txt
 Digest.crc32
+Filter.db

test/unit/org/apache/cassandra/io/sstable/LegacySSTableTest.java

@@ -24,6 +24,7 @@ import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Random;
+import java.util.UUID;
 
 import org.junit.After;
 import org.junit.Assert;
@@ -146,6 +147,28 @@ public class LegacySSTableTest
         doTestLegacyCqlTables();
     }
 
+    @Test
+    public void testMutateMetadata() throws Exception
+    {
+        // we need to make sure we write old version metadata in the format for that version
+        for (String legacyVersion : legacyVersions)
+        {
+            logger.info("Loading legacy version: {}", legacyVersion);
+            truncateLegacyTables(legacyVersion);
+            loadLegacyTables(legacyVersion);
+            CacheService.instance.invalidateKeyCache();
+
+            for (ColumnFamilyStore cfs : Keyspace.open("legacy_tables").getColumnFamilyStores())
+            {
+                for (SSTableReader sstable : cfs.getLiveSSTables())
+                {
+                    sstable.descriptor.getMetadataSerializer().mutateRepaired(sstable.descriptor, 1234, UUID.randomUUID());
+                    sstable.reloadSSTableMetadata();
+                }
+            }
+        }
+    }
+
     private void doTestLegacyCqlTables() throws Exception
     {
         for (String legacyVersion : legacyVersions)