Upgrade rockdb to 5.10.3; Use DeleteFilesInRange to accelerate cleanu…

…p process

Summary: as titled

Test Plan: unit test and test in production

Reviewers: dikang, sdev

Differential Revision: https://phabricator.intern.facebook.com/D7134922

build.xml

@@ -479,7 +479,7 @@
         <dependency groupId="org.openjdk.jmh" artifactId="jmh-generator-annprocess"/>
         <dependency groupId="net.ju-n.compile-command-annotations" artifactId="compile-command-annotations"/>
         <dependency groupId="org.apache.ant" artifactId="ant-junit" version="1.9.4" />
-        <dependency groupId="org.rocksdb" artifactId="rocksdbjni" version="5.8.0" scope="system" systemPath="${basedir}/lib/rocksdbjni-5.8.0.jar"/>
+        <dependency groupId="org.rocksdb" artifactId="rocksdbjni" version="5.10.3" scope="system" systemPath="${basedir}/lib/rocksdbjni-5.10.3.jar"/>
       </artifact:pom>
       <!-- this build-deps-pom-sources "artifact" is the same as build-deps-pom but only with those
            artifacts that have "-source.jar" files -->

src/java/org/apache/cassandra/rocksdb/RocksDBCF.java

@@ -325,7 +325,10 @@ public void merge(DecoratedKey partitionKey, byte[] key, byte[] value) throws Ro
     public void deleteRange(byte[] start, byte[] end) throws RocksDBException
     {
         for (RocksDB rocksDB : rocksDBLists)
+        {
+            rocksDB.deleteFilesInRange(start, end);
             rocksDB.deleteRange(start, end);
+        }
     }
 
     public void compactRange() throws RocksDBException

test/unit/org/apache/cassandra/rocksdb/RocksDBCFTest.java

@@ -70,6 +70,33 @@ private byte[] encodeValue(ColumnFamilyStore cfs, String value)
         return RowValueEncoder.encode(cfs.metadata, builder.build());
     }
 
+    @Test
+    public void testDeleteRange() throws RocksDBException
+    {
+        createTable("CREATE TABLE %s (p text, c text, v text, PRIMARY KEY (p, c))");
+        ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
+
+        RocksDBCF rocksDBCF = RocksDBEngine.getRocksDBCF(cfs.metadata.cfId);
+
+        byte[] a = "a".getBytes();
+        byte[] b = "b".getBytes();
+        byte[] c = "c".getBytes();
+        byte[] d = "d".getBytes();
+        byte[] value = encodeValue(cfs, "test_value");
+
+        rocksDBCF.merge(dk, a, value);
+        rocksDBCF.merge(dk, b, value);
+        rocksDBCF.merge(dk, c, value);
+        rocksDBCF.merge(dk, d, value);
+
+        rocksDBCF.deleteRange(b, d);
+        rocksDBCF.compactRange();
+        assertArrayEquals(value, rocksDBCF.get(dk, a));
+        assertNull(rocksDBCF.get(dk, b));
+        assertNull(rocksDBCF.get(dk, c));
+        assertArrayEquals(value, rocksDBCF.get(dk, d));
+    }
+
     @Test
     public void testTruncate() throws RocksDBException
     {