New Bloomfilter format without changing the byte ordering

patch by Jay Zhuang; reviewed by jasobrown for CASSANDRA-9067

CHANGES.txt

@@ -1,4 +1,5 @@
 4.0
+ * BloomFilter serialization format should not change byte ordering (CASSANDRA-9067)
  * Remove unused on-heap BloomFilter implementation (CASSANDRA-14152)
  * Delete temp test files on exit (CASSANDRA-14153)
  * Make PartitionUpdate and Mutation immutable (CASSANDRA-13867)

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

@@ -75,6 +75,7 @@ import org.apache.cassandra.utils.*;
 import org.apache.cassandra.utils.concurrent.OpOrder;
 import org.apache.cassandra.utils.concurrent.Ref;
 import org.apache.cassandra.utils.concurrent.SelfRefCounted;
+import org.apache.cassandra.utils.BloomFilterSerializer;
 
 import static org.apache.cassandra.db.Directories.SECONDARY_INDEX_NAME_SEPARATOR;
 
@@ -728,7 +729,7 @@ public abstract class SSTableReader extends SSTable implements SelfRefCounted<SS
         {
             // bf is enabled and fp chance matches the currently configured value.
             load(false, true);
-            loadBloomFilter();
+            loadBloomFilter(descriptor.version.hasOldBfFormat());
         }
     }
 
@@ -736,12 +737,13 @@ public abstract class SSTableReader extends SSTable implements SelfRefCounted<SS
      * Load bloom filter from Filter.db file.
      *
      * @throws IOException
+     * @param oldBfFormat
      */
-    private void loadBloomFilter() throws IOException
+    private void loadBloomFilter(boolean oldBfFormat) throws IOException
     {
         try (DataInputStream stream = new DataInputStream(new BufferedInputStream(Files.newInputStream(Paths.get(descriptor.filenameFor(Component.FILTER))))))
         {
-            bf = FilterFactory.deserialize(stream);
+            bf = BloomFilterSerializer.deserialize(stream, oldBfFormat);
         }
     }
 

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

@@ -57,6 +57,13 @@ public abstract class Version
 
     public abstract boolean hasMetadataChecksum();
 
+    /**
+     * The old bloomfilter format serializes the data as BIG_ENDIAN long's, the new one uses the
+     * same format as in memory (serializes as bytes).
+     * @return True if the bloomfilter file is old serialization format
+     */
+    public abstract boolean hasOldBfFormat();
+
     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, checksummed sstable metadata file
+        // na (4.0.0): uncompressed chunks, pending repair session, checksummed sstable metadata file, new Bloomfilter format
         //
         // NOTE: when adding a new version, please add that to LegacySSTableTest, too.
 
@@ -131,6 +131,11 @@ public class BigFormat implements SSTableFormat
         public final boolean hasMaxCompressedLength;
         private final boolean hasPendingRepair;
         private final boolean hasMetadataChecksum;
+        /**
+         * CASSANDRA-9067: 4.0 bloom filter representation changed (two longs just swapped)
+         * have no 'static' bits caused by using the same upper bits for both bloom filter and token distribution.
+         */
+        private final boolean hasOldBfFormat;
 
         BigVersion(String version)
         {
@@ -144,6 +149,7 @@ public class BigFormat implements SSTableFormat
             hasMaxCompressedLength = version.compareTo("na") >= 0;
             hasPendingRepair = version.compareTo("na") >= 0;
             hasMetadataChecksum = version.compareTo("na") >= 0;
+            hasOldBfFormat = version.compareTo("na") < 0;
         }
 
         @Override
@@ -197,5 +203,11 @@ public class BigFormat implements SSTableFormat
         {
             return hasMaxCompressedLength;
         }
+
+        @Override
+        public boolean hasOldBfFormat()
+        {
+            return hasOldBfFormat;
+        }
     }
 }

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

@@ -484,7 +484,7 @@ public class BigTableWriter extends SSTableWriter
                      DataOutputStreamPlus stream = new BufferedDataOutputStreamPlus(fos))
                 {
                     // bloom filter
-                    FilterFactory.serialize(bf, stream);
+                    BloomFilterSerializer.serialize((BloomFilter) bf, stream);
                     stream.flush();
                     SyncUtil.sync(fos);
                 }

src/java/org/apache/cassandra/utils/BloomFilterSerializer.java

@@ -17,7 +17,7 @@
  */
 package org.apache.cassandra.utils;
 
-import java.io.DataInput;
+import java.io.DataInputStream;
 import java.io.IOException;
 
 import org.apache.cassandra.db.TypeSizes;
@@ -25,7 +25,7 @@ import org.apache.cassandra.io.util.DataOutputPlus;
 import org.apache.cassandra.utils.obs.IBitSet;
 import org.apache.cassandra.utils.obs.OffHeapBitSet;
 
-final class BloomFilterSerializer
+public final class BloomFilterSerializer
 {
     private BloomFilterSerializer()
     {
@@ -38,10 +38,10 @@ final class BloomFilterSerializer
     }
 
     @SuppressWarnings("resource")
-    public static BloomFilter deserialize(DataInput in) throws IOException
+    public static BloomFilter deserialize(DataInputStream in, boolean oldBfFormat) throws IOException
     {
         int hashes = in.readInt();
-        IBitSet bs = OffHeapBitSet.deserialize(in);
+        IBitSet bs = OffHeapBitSet.deserialize(in, oldBfFormat);
 
         return new BloomFilter(hashes, bs);
     }

src/java/org/apache/cassandra/utils/FilterFactory.java

@@ -17,13 +17,9 @@
  */
 package org.apache.cassandra.utils;
 
-import java.io.DataInput;
-import java.io.IOException;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import org.apache.cassandra.io.util.DataOutputPlus;
 import org.apache.cassandra.utils.obs.IBitSet;
 import org.apache.cassandra.utils.obs.OffHeapBitSet;
 
@@ -34,16 +30,6 @@ public class FilterFactory
     private static final Logger logger = LoggerFactory.getLogger(FilterFactory.class);
     private static final long BITSET_EXCESS = 20;
 
-    public static void serialize(IFilter bf, DataOutputPlus output) throws IOException
-    {
-        BloomFilterSerializer.serialize((BloomFilter) bf, output);
-    }
-
-    public static IFilter deserialize(DataInput input) throws IOException
-    {
-        return BloomFilterSerializer.deserialize(input);
-    }
-
     /**
      * @return A BloomFilter with the lowest practical false positive
      *         probability for the given number of elements.

src/java/org/apache/cassandra/utils/obs/IBitSet.java

@@ -18,9 +18,9 @@
 package org.apache.cassandra.utils.obs;
 
 import java.io.Closeable;
-import java.io.DataOutput;
 import java.io.IOException;
 
+import org.apache.cassandra.io.util.DataOutputPlus;
 import org.apache.cassandra.utils.concurrent.Ref;
 
 public interface IBitSet extends Closeable
@@ -44,7 +44,7 @@ public interface IBitSet extends Closeable
      */
     public void clear(long index);
 
-    public void serialize(DataOutput out) throws IOException;
+    public void serialize(DataOutputPlus out) throws IOException;
 
     public long serializedSize();
 

src/java/org/apache/cassandra/utils/obs/OffHeapBitSet.java

@@ -18,11 +18,17 @@
 package org.apache.cassandra.utils.obs;
 
 import java.io.DataInput;
+import java.io.DataInputStream;
 import java.io.DataOutput;
 import java.io.IOException;
 
+import com.google.common.annotations.VisibleForTesting;
+
 import org.apache.cassandra.db.TypeSizes;
+import org.apache.cassandra.io.util.DataOutputPlus;
 import org.apache.cassandra.io.util.Memory;
+import org.apache.cassandra.io.util.MemoryOutputStream;
+import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.concurrent.Ref;
 
 /**
@@ -109,19 +115,26 @@ public class OffHeapBitSet implements IBitSet
         bytes.setMemory(0, bytes.size(), (byte) 0);
     }
 
-    public void serialize(DataOutput out) throws IOException
+    public void serialize(DataOutputPlus out) throws IOException
     {
         out.writeInt((int) (bytes.size() / 8));
-        for (long i = 0; i < bytes.size();)
+        out.write(bytes, 0, bytes.size());
+    }
+
+    @VisibleForTesting
+    public void serializeOldBfFormat(DataOutputPlus out) throws IOException
+    {
+        out.writeInt((int) (bytes.size() / 8));
+        for (long i = 0; i < bytes.size(); )
         {
             long value = ((bytes.getByte(i++) & 0xff) << 0)
-                       + ((bytes.getByte(i++) & 0xff) << 8)
-                       + ((bytes.getByte(i++) & 0xff) << 16)
-                       + ((long) (bytes.getByte(i++) & 0xff) << 24)
-                       + ((long) (bytes.getByte(i++) & 0xff) << 32)
-                       + ((long) (bytes.getByte(i++) & 0xff) << 40)
-                       + ((long) (bytes.getByte(i++) & 0xff) << 48)
-                       + ((long) bytes.getByte(i++) << 56);
+                         + ((bytes.getByte(i++) & 0xff) << 8)
+                         + ((bytes.getByte(i++) & 0xff) << 16)
+                         + ((long) (bytes.getByte(i++) & 0xff) << 24)
+                         + ((long) (bytes.getByte(i++) & 0xff) << 32)
+                         + ((long) (bytes.getByte(i++) & 0xff) << 40)
+                         + ((long) (bytes.getByte(i++) & 0xff) << 48)
+                         + ((long) bytes.getByte(i++) << 56);
             out.writeLong(value);
         }
     }
@@ -132,21 +145,28 @@ public class OffHeapBitSet implements IBitSet
     }
 
     @SuppressWarnings("resource")
-    public static OffHeapBitSet deserialize(DataInput in) throws IOException
+    public static OffHeapBitSet deserialize(DataInputStream in, boolean oldBfFormat) throws IOException
     {
         long byteCount = in.readInt() * 8L;
         Memory memory = Memory.allocate(byteCount);
-        for (long i = 0; i < byteCount;)
+        if (oldBfFormat)
         {
-            long v = in.readLong();
-            memory.setByte(i++, (byte) (v >>> 0));
-            memory.setByte(i++, (byte) (v >>> 8));
-            memory.setByte(i++, (byte) (v >>> 16));
-            memory.setByte(i++, (byte) (v >>> 24));
-            memory.setByte(i++, (byte) (v >>> 32));
-            memory.setByte(i++, (byte) (v >>> 40));
-            memory.setByte(i++, (byte) (v >>> 48));
-            memory.setByte(i++, (byte) (v >>> 56));
+            for (long i = 0; i < byteCount; )
+            {
+                long v = in.readLong();
+                memory.setByte(i++, (byte) (v >>> 0));
+                memory.setByte(i++, (byte) (v >>> 8));
+                memory.setByte(i++, (byte) (v >>> 16));
+                memory.setByte(i++, (byte) (v >>> 24));
+                memory.setByte(i++, (byte) (v >>> 32));
+                memory.setByte(i++, (byte) (v >>> 40));
+                memory.setByte(i++, (byte) (v >>> 48));
+                memory.setByte(i++, (byte) (v >>> 56));
+            }
+        }
+        else
+        {
+            FBUtilities.copy(in, new MemoryOutputStream(memory), byteCount);
         }
         return new OffHeapBitSet(memory);
     }

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

@@ -1 +1 @@
-4285275084
+4004129384

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

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

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

@@ -1 +1 @@
-1093063834
+4072239034

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

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

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

@@ -1 +1 @@
-2354437953
+3772296151

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

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

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

@@ -1 +1 @@
-2023874595
+4035692752

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

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

test/microbench/org/apache/cassandra/test/microbench/BloomFilterSerializerBench.java

@@ -0,0 +1,99 @@
+/*
+ * 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.test.microbench;
+
+import java.io.DataInputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.cassandra.db.BufferDecoratedKey;
+import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.io.util.BufferedDataOutputStreamPlus;
+import org.apache.cassandra.io.util.DataOutputStreamPlus;
+import org.apache.cassandra.io.util.FileUtils;
+import org.apache.cassandra.utils.BloomFilter;
+import org.apache.cassandra.utils.BloomFilterSerializer;
+import org.apache.cassandra.utils.FilterFactory;
+import org.apache.cassandra.utils.IFilter;
+import org.apache.cassandra.utils.SerializationsTest;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+@Warmup(iterations = 1, time = 1, timeUnit = TimeUnit.SECONDS)
+@Measurement(iterations = 2, time = 4, timeUnit = TimeUnit.SECONDS)
+@Fork(value = 2)
+@State(Scope.Benchmark)
+public class BloomFilterSerializerBench
+{
+
+    @Param({"1", "10", "100", "1024"})
+    private long numElemsInK;
+
+    @Param({"true", "false"})
+    public boolean oldBfFormat;
+
+    static final IFilter.FilterKey wrap(ByteBuffer buf)
+    {
+        return new BufferDecoratedKey(new Murmur3Partitioner.LongToken(0L), buf);
+    }
+
+    private ByteBuffer testVal = ByteBuffer.wrap(new byte[] { 0, 1});
+
+    @Benchmark
+    public void serializationTest() throws IOException
+    {
+        File file = FileUtils.createTempFile("bloomFilterTest-", ".dat");
+        try
+        {
+            BloomFilter filter = (BloomFilter) FilterFactory.getFilter(numElemsInK * 1024, 0.01d);
+            filter.add(wrap(testVal));
+            DataOutputStreamPlus out = new BufferedDataOutputStreamPlus(new FileOutputStream(file));
+            if (oldBfFormat)
+                SerializationsTest.serializeOldBfFormat(filter, out);
+            else
+                BloomFilterSerializer.serialize(filter, out);
+            out.close();
+            filter.close();
+
+            DataInputStream in = new DataInputStream(new FileInputStream(file));
+            BloomFilter filter2 = BloomFilterSerializer.deserialize(in, oldBfFormat);
+            FileUtils.closeQuietly(in);
+            filter2.close();
+        }
+        finally
+        {
+            file.delete();
+        }
+    }
+
+}

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

@@ -94,7 +94,7 @@ public class LegacySSTableTest
     {
         String scp = System.getProperty(LEGACY_SSTABLE_PROP);
         Assert.assertNotNull("System property " + LEGACY_SSTABLE_ROOT + " not set", scp);
-        
+
         LEGACY_SSTABLE_ROOT = new File(scp).getAbsoluteFile();
         Assert.assertTrue("System property " + LEGACY_SSTABLE_ROOT + " does not specify a directory", LEGACY_SSTABLE_ROOT.isDirectory());
 
@@ -270,7 +270,6 @@ public class LegacySSTableTest
                 logger.debug("for pk={} ck={}", pk, ck);
 
                 String pkValue = Integer.toString(pk);
-                UntypedResultSet rs;
                 if (ck == 0)
                 {
                     readSimpleTable(legacyVersion, pkValue);

test/unit/org/apache/cassandra/utils/BloomFilterTest.java

@@ -43,19 +43,23 @@ public class BloomFilterTest
 {
     public IFilter bfInvHashes;
 
-    public BloomFilterTest()
-    {
 
-    }
 
-    public static IFilter testSerialize(IFilter f) throws IOException
+    public static IFilter testSerialize(IFilter f, boolean oldBfFormat) throws IOException
     {
         f.add(FilterTestHelper.bytes("a"));
         DataOutputBuffer out = new DataOutputBuffer();
-        FilterFactory.serialize(f, out);
+        if (oldBfFormat)
+        {
+            SerializationsTest.serializeOldBfFormat((BloomFilter) f, out);
+        }
+        else
+        {
+            BloomFilterSerializer.serialize((BloomFilter) f, out);
+        }
 
         ByteArrayInputStream in = new ByteArrayInputStream(out.getData(), 0, out.getLength());
-        IFilter f2 = FilterFactory.deserialize(new DataInputStream(in));
+        IFilter f2 = BloomFilterSerializer.deserialize(new DataInputStream(in), oldBfFormat);
 
         assert f2.isPresent(FilterTestHelper.bytes("a"));
         assert !f2.isPresent(FilterTestHelper.bytes("b"));
@@ -132,7 +136,8 @@ public class BloomFilterTest
     @Test
     public void testSerialize() throws IOException
     {
-        BloomFilterTest.testSerialize(bfInvHashes).close();
+        BloomFilterTest.testSerialize(bfInvHashes, true).close();
+        BloomFilterTest.testSerialize(bfInvHashes, false).close();
     }
 
     @Test
@@ -206,13 +211,12 @@ public class BloomFilterTest
         BloomFilter filter = (BloomFilter) FilterFactory.getFilter(((long) Integer.MAX_VALUE / 8) + 1, 0.01d);
         filter.add(FilterTestHelper.wrap(test));
         DataOutputStreamPlus out = new BufferedDataOutputStreamPlus(new FileOutputStream(file));
-        FilterFactory.serialize(filter, out);
-        filter.bitset.serialize(out);
+        BloomFilterSerializer.serialize(filter, out);
         out.close();
         filter.close();
 
         DataInputStream in = new DataInputStream(new FileInputStream(file));
-        BloomFilter filter2 = (BloomFilter) FilterFactory.deserialize(in);
+        BloomFilter filter2 = BloomFilterSerializer.deserialize(in, false);
         Assert.assertTrue(filter2.isPresent(FilterTestHelper.wrap(test)));
         FileUtils.closeQuietly(in);
         filter2.close();

test/unit/org/apache/cassandra/utils/SerializationsTest.java

@@ -31,30 +31,42 @@ import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.DecoratedKey;
 import org.apache.cassandra.db.marshal.Int32Type;
 import org.apache.cassandra.io.util.DataInputPlus.DataInputStreamPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
 import org.apache.cassandra.io.util.DataOutputStreamPlus;
-import org.apache.cassandra.dht.IPartitioner;
 import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.utils.obs.OffHeapBitSet;
 
 import java.io.File;
 import java.io.FileInputStream;
 
 public class SerializationsTest extends AbstractSerializationsTester
 {
+    // Helper function to serialize old Bloomfilter format, should be removed once the old format is not supported
+    public static void serializeOldBfFormat(BloomFilter bf, DataOutputPlus out) throws IOException
+    {
+        out.writeInt(bf.hashCount);
+        Assert.assertTrue(bf.bitset instanceof OffHeapBitSet);
+        ((OffHeapBitSet) bf.bitset).serializeOldBfFormat(out);
+    }
+
     @BeforeClass
     public static void initDD()
     {
         DatabaseDescriptor.daemonInitialization();
     }
 
-    private static void testBloomFilterWrite1000() throws IOException
+    private static void testBloomFilterWrite1000(boolean oldBfFormat) throws IOException
     {
         try (IFilter bf = FilterFactory.getFilter(1000000, 0.0001))
         {
             for (int i = 0; i < 1000; i++)
                 bf.add(Util.dk(Int32Type.instance.decompose(i)));
-            try (DataOutputStreamPlus out = getOutput("3.0", "utils.BloomFilter1000.bin"))
+            try (DataOutputStreamPlus out = getOutput(oldBfFormat ? "3.0" : "4.0", "utils.BloomFilter1000.bin"))
             {
-                FilterFactory.serialize(bf, out);
+                if (oldBfFormat)
+                    serializeOldBfFormat((BloomFilter) bf, out);
+                else
+                    BloomFilterSerializer.serialize((BloomFilter) bf, out);
             }
         }
     }
@@ -63,10 +75,29 @@ public class SerializationsTest extends AbstractSerializationsTester
     public void testBloomFilterRead1000() throws IOException
     {
         if (EXECUTE_WRITES)
-            testBloomFilterWrite1000();
+        {
+            testBloomFilterWrite1000(false);
+            testBloomFilterWrite1000(true);
+        }
+
+        try (DataInputStream in = getInput("4.0", "utils.BloomFilter1000.bin");
+             IFilter filter = BloomFilterSerializer.deserialize(in, false))
+        {
+            boolean present;
+            for (int i = 0 ; i < 1000 ; i++)
+            {
+                present = filter.isPresent(Util.dk(Int32Type.instance.decompose(i)));
+                Assert.assertTrue(present);
+            }
+            for (int i = 1000 ; i < 2000 ; i++)
+            {
+                present = filter.isPresent(Util.dk(Int32Type.instance.decompose(i)));
+                Assert.assertFalse(present);
+            }
+        }
 
         try (DataInputStream in = getInput("3.0", "utils.BloomFilter1000.bin");
-             IFilter filter = FilterFactory.deserialize(in))
+             IFilter filter = BloomFilterSerializer.deserialize(in, true))
         {
             boolean present;
             for (int i = 0 ; i < 1000 ; i++)
@@ -85,15 +116,15 @@ public class SerializationsTest extends AbstractSerializationsTester
     @Test
     public void testBloomFilterTable() throws Exception
     {
-        testBloomFilterTable("test/data/bloom-filter/la/foo/la-1-big-Filter.db");
+        testBloomFilterTable("test/data/bloom-filter/la/foo/la-1-big-Filter.db", true);
     }
 
-    private static void testBloomFilterTable(String file) throws Exception
+    private static void testBloomFilterTable(String file, boolean oldBfFormat) throws Exception
     {
         Murmur3Partitioner partitioner = new Murmur3Partitioner();
 
         try (DataInputStream in = new DataInputStream(new FileInputStream(new File(file)));
-             IFilter filter = FilterFactory.deserialize(in))
+             IFilter filter = BloomFilterSerializer.deserialize(in, oldBfFormat))
         {
             for (int i = 1; i <= 10; i++)
             {

test/unit/org/apache/cassandra/utils/obs/OffHeapBitSetTest.java

@@ -25,11 +25,9 @@ import java.util.List;
 import java.util.Random;
 
 import com.google.common.collect.Lists;
+import org.apache.cassandra.io.util.DataOutputBuffer;
 import org.junit.Assert;
 import org.junit.Test;
-import org.junit.rules.ExpectedException;
-
-import org.apache.cassandra.io.util.DataOutputBuffer;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
@@ -46,8 +44,7 @@ public class OffHeapBitSetTest
             Assert.assertEquals(bs.get(i), newbs.get(i));
     }
 
-    @Test
-    public void testOffHeapSerialization() throws IOException
+    private void testOffHeapSerialization(boolean oldBfFormat) throws IOException
     {
         try (OffHeapBitSet bs = new OffHeapBitSet(100000))
         {
@@ -56,16 +53,26 @@ public class OffHeapBitSetTest
                     bs.set(i);
 
             DataOutputBuffer out = new DataOutputBuffer();
-            bs.serialize(out);
+            if (oldBfFormat)
+                bs.serializeOldBfFormat(out);
+            else
+                bs.serialize(out);
 
             DataInputStream in = new DataInputStream(new ByteArrayInputStream(out.getData()));
-            try (OffHeapBitSet newbs = OffHeapBitSet.deserialize(in))
+            try (OffHeapBitSet newbs = OffHeapBitSet.deserialize(in, oldBfFormat))
             {
                 compare(bs, newbs);
             }
         }
     }
 
+    @Test
+    public void testSerialization() throws IOException
+    {
+        testOffHeapSerialization(true);
+        testOffHeapSerialization(false);
+    }
+
     @Test
     public void testBitSetGetClear()
     {