Access SST full file checksum via RocksDB#getLiveFilesMetadata (#11770)

Summary:
**Description**

This PR passes along the native `LiveFileMetaData#file_checksum` field from the C++ class to the Java API as a copied byte array. If there is no file checksum generator factory set beforehand, then the array will empty. Please advise if you'd rather it be null - an empty array means one extra allocation, but it avoids possible null pointer exceptions.

> **Note**
> This functionality complements but does not supersede facebook/rocksdb#11736

It's outside the scope here to add support for Java based `FileChecksumGenFactory` implementations. As a workaround, users can already use the built-in one by creating their initial `DBOptions` via properties:

```java
final Properties props = new Properties();
props.put("file_checksum_gen_factory", "FileChecksumGenCrc32cFactory");

try (final DBOptions dbOptions = DBOptions.getDBOptionsFromProps(props);
     final ColumnFamilyOptions cfOptions = new ColumnFamilyOptions();
     final Options options = new Options(dbOptions, cfOptions).setCreateIfMissing(true)) {
// do stuff
}
```

I wanted to add a better test, but unfortunately there's no available CRC32C implementation available in Java 8 without adding a dependency or adding a JNI helper for RocksDB's own implementation (or bumping the minimum version for tests to Java 9). That said, I understand the test is rather poor, so happy to change it to whatever you'd like.

**Context**

To give some context, we replicate RocksDB checkpoints to other nodes. Part of this is verifying the integrity of each file during replication. With a large enough RocksDB, computing the checksum ourselves is prohibitively expensive. Since SST files comprise the bulk of the data, we'd much rather delegate this to RocksDB on file write, and read it back after to compare.

It's likely we will provide a follow up to read the file checksum list directly from the manifest without having to open the DB, but this was the easiest first step to get it working for us.

Pull Request resolved: facebook/rocksdb#11770

Reviewed By: hx235

Differential Revision: D52420729

Pulled By: ajkr

fbshipit-source-id: a873de35a48aaf315e125733091cd221a97b9073
(cherry picked from commit 5b073a7)

java/rocksjni/portal.h

@@ -7447,7 +7447,7 @@ class LiveFileMetaDataJni : public JavaClass {
 
     jmethodID mid = env->GetMethodID(
         jclazz, "<init>",
-        "([BILjava/lang/String;Ljava/lang/String;JJJ[B[BJZJJ)V");
+        "([BILjava/lang/String;Ljava/lang/String;JJJ[B[BJZJJ[B)V");
     if (mid == nullptr) {
       // exception thrown: NoSuchMethodException or OutOfMemoryError
       return nullptr;
@@ -7498,6 +7498,18 @@ class LiveFileMetaDataJni : public JavaClass {
       return nullptr;
     }
 
+    jbyteArray jfile_checksum = ROCKSDB_NAMESPACE::JniUtil::copyBytes(
+        env, live_file_meta_data->file_checksum);
+    if (env->ExceptionCheck()) {
+      // exception occurred creating java string
+      env->DeleteLocalRef(jcolumn_family_name);
+      env->DeleteLocalRef(jfile_name);
+      env->DeleteLocalRef(jpath);
+      env->DeleteLocalRef(jsmallest_key);
+      env->DeleteLocalRef(jlargest_key);
+      return nullptr;
+    }
+
     jobject jlive_file_meta_data = env->NewObject(
         jclazz, mid, jcolumn_family_name,
         static_cast<jint>(live_file_meta_data->level), jfile_name, jpath,
@@ -7508,14 +7520,15 @@ class LiveFileMetaDataJni : public JavaClass {
         static_cast<jlong>(live_file_meta_data->num_reads_sampled),
         static_cast<jboolean>(live_file_meta_data->being_compacted),
         static_cast<jlong>(live_file_meta_data->num_entries),
-        static_cast<jlong>(live_file_meta_data->num_deletions));
+        static_cast<jlong>(live_file_meta_data->num_deletions), jfile_checksum);
 
     if (env->ExceptionCheck()) {
       env->DeleteLocalRef(jcolumn_family_name);
       env->DeleteLocalRef(jfile_name);
       env->DeleteLocalRef(jpath);
       env->DeleteLocalRef(jsmallest_key);
       env->DeleteLocalRef(jlargest_key);
+      env->DeleteLocalRef(jfile_checksum);
       return nullptr;
     }
 
@@ -7525,6 +7538,7 @@ class LiveFileMetaDataJni : public JavaClass {
     env->DeleteLocalRef(jpath);
     env->DeleteLocalRef(jsmallest_key);
     env->DeleteLocalRef(jlargest_key);
+    env->DeleteLocalRef(jfile_checksum);
 
     return jlive_file_meta_data;
   }

java/src/main/java/org/rocksdb/LiveFileMetaData.java

@@ -16,22 +16,13 @@ public class LiveFileMetaData extends SstFileMetaData {
   /**
    * Called from JNI C++
    */
-  private LiveFileMetaData(
-      final byte[] columnFamilyName,
-      final int level,
-      final String fileName,
-      final String path,
-      final long size,
-      final long smallestSeqno,
-      final long largestSeqno,
-      final byte[] smallestKey,
-      final byte[] largestKey,
-      final long numReadsSampled,
-      final boolean beingCompacted,
-      final long numEntries,
-      final long numDeletions) {
-    super(fileName, path, size, smallestSeqno, largestSeqno, smallestKey,
-        largestKey, numReadsSampled, beingCompacted, numEntries, numDeletions);
+  private LiveFileMetaData(final byte[] columnFamilyName, final int level, final String fileName,
+      final String path, final long size, final long smallestSeqno, final long largestSeqno,
+      final byte[] smallestKey, final byte[] largestKey, final long numReadsSampled,
+      final boolean beingCompacted, final long numEntries, final long numDeletions,
+      final byte[] fileChecksum) {
+    super(fileName, path, size, smallestSeqno, largestSeqno, smallestKey, largestKey,
+        numReadsSampled, beingCompacted, numEntries, numDeletions, fileChecksum);
     this.columnFamilyName = columnFamilyName;
     this.level = level;
   }

java/src/main/java/org/rocksdb/SstFileMetaData.java

@@ -20,6 +20,7 @@ public class SstFileMetaData {
   private final boolean beingCompacted;
   private final long numEntries;
   private final long numDeletions;
+  private final byte[] fileChecksum;
 
   /**
    * Called from JNI C++
@@ -35,12 +36,13 @@ public class SstFileMetaData {
    * @param beingCompacted true if the file is being compacted, false otherwise
    * @param numEntries the number of entries
    * @param numDeletions the number of deletions
+   * @param fileChecksum the full file checksum (if enabled)
    */
   @SuppressWarnings("PMD.ArrayIsStoredDirectly")
   protected SstFileMetaData(final String fileName, final String path, final long size,
       final long smallestSeqno, final long largestSeqno, final byte[] smallestKey,
       final byte[] largestKey, final long numReadsSampled, final boolean beingCompacted,
-      final long numEntries, final long numDeletions) {
+      final long numEntries, final long numDeletions, final byte[] fileChecksum) {
     this.fileName = fileName;
     this.path = path;
     this.size = size;
@@ -52,6 +54,7 @@ protected SstFileMetaData(final String fileName, final String path, final long s
     this.beingCompacted = beingCompacted;
     this.numEntries = numEntries;
     this.numDeletions = numDeletions;
+    this.fileChecksum = fileChecksum;
   }
 
   /**
@@ -154,4 +157,14 @@ public long numEntries() {
   public long numDeletions() {
     return numDeletions;
   }
+
+  /**
+   * Get the full file checksum iff full file checksum is enabled.
+   *
+   * @return the file's checksum
+   */
+  @SuppressWarnings("PMD.MethodReturnsInternalArray")
+  public byte[] fileChecksum() {
+    return fileChecksum;
+  }
 }

java/src/test/java/org/rocksdb/RocksDBTest.java

@@ -1381,6 +1381,34 @@ public void getApproximateMemTableStatsSingleKey() throws RocksDBException {
     }
   }
 
+  @Test
+  public void getLiveFilesMetadataWithChecksum() throws RocksDBException {
+    final Properties props = new Properties();
+    final byte[] key1 = "key1".getBytes(UTF_8);
+    props.put("file_checksum_gen_factory", "FileChecksumGenCrc32cFactory");
+
+    try (final DBOptions dbOptions = DBOptions.getDBOptionsFromProps(props);
+         final ColumnFamilyOptions cfOptions = new ColumnFamilyOptions();
+         final Options options = new Options(dbOptions, cfOptions).setCreateIfMissing(true)) {
+      final String dbPath = dbFolder.getRoot().getAbsolutePath();
+
+      // disable WAL so we have a deterministic checksum
+      try (final RocksDB db = RocksDB.open(options, dbPath);
+           final WriteOptions writeOptions = new WriteOptions().setDisableWAL(true)) {
+        db.put(writeOptions, key1, key1);
+      }
+
+      try (final RocksDB db = RocksDB.open(options, dbPath)) {
+        final List<LiveFileMetaData> expectedFileMetadata = db.getLiveFilesMetaData();
+        assertThat(expectedFileMetadata).hasSize(1);
+        // ideally we could re-compute here, but CRC32C is a Java 9 feature, so we have no CRC32C
+        // implementation available here
+        final LiveFileMetaData sstFile = expectedFileMetadata.get(0);
+        assertThat(sstFile.fileChecksum()).isNotEmpty();
+      }
+    }
+  }
+
   @Ignore("TODO(AR) re-enable when ready!")
   @Test
   public void compactFiles() throws RocksDBException {