Add tests for supported compression types in RocksDB.

docs/streaming/apis-on-dataframes-and-datasets.md

@@ -1899,7 +1899,7 @@ Here are the configs regarding to RocksDB instance of the state store provider:
   </tr>
   <tr>
     <td>spark.sql.streaming.stateStore.rocksdb.compression</td>
-    <td>Compression type used in RocksDB. The string is converted RocksDB compression type through RocksDB Java API getCompressionType(). </td>
+    <td>Compression type used in RocksDB. The string is converted RocksDB compression type through RocksDB Java API getCompressionType(). Supported values: "bzip2", "lz4", "lz4hc", "snappy", "z", "zstd".</td>
     <td>lz4</td>
   </tr>
 </table>

sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala

@@ -1429,6 +1429,56 @@ class RocksDBSuite extends AlsoTestWithRocksDBFeatures with SharedSparkSession
     }
   }
 
+  test("RocksDB: all compression types") {
+    val remoteDir = Utils.createTempDir().toString
+    new File(remoteDir).delete()
+
+    // Empty string falls back to no compression.
+    var conf = RocksDBConf().copy(compression = "")
+    withDB(remoteDir, conf = conf) { db =>
+      assert(db.rocksDbOptions.compressionType() == CompressionType.NO_COMPRESSION)
+    }
+
+    // Unknown config falls back to no compression.
+    conf = RocksDBConf().copy(compression = "foobar")
+    withDB(remoteDir, conf = conf) { db =>
+      assert(db.rocksDbOptions.compressionType() == CompressionType.NO_COMPRESSION)
+    }
+
+    // All 6 possible compression types are supported.
+    conf = RocksDBConf().copy(compression = "snappy")
+    withDB(remoteDir, conf = conf) { db =>
+      assert(db.rocksDbOptions.compressionType() == CompressionType.SNAPPY_COMPRESSION)
+    }
+
+    conf = RocksDBConf().copy(compression = "z")
+    withDB(remoteDir, conf = conf) { db =>
+      assert(db.rocksDbOptions.compressionType() == CompressionType.ZLIB_COMPRESSION)
+    }
+
+    conf = RocksDBConf().copy(compression = "bzip2")
+    withDB(remoteDir, conf = conf) { db =>
+      assert(db.rocksDbOptions.compressionType() == CompressionType.BZLIB2_COMPRESSION)
+    }
+
+    conf = RocksDBConf().copy(compression = "lz4")
+    withDB(remoteDir, conf = conf) { db =>
+      assert(db.rocksDbOptions.compressionType() == CompressionType.LZ4_COMPRESSION)
+    }
+
+    conf = RocksDBConf().copy(compression = "lz4hc")
+    withDB(remoteDir, conf = conf) { db =>
+      assert(db.rocksDbOptions.compressionType() == CompressionType.LZ4HC_COMPRESSION)
+    }
+
+    conf = RocksDBConf().copy(compression = "zstd")
+    withDB(remoteDir, conf = conf) { db =>
+      assert(db.rocksDbOptions.compressionType() == CompressionType.ZSTD_COMPRESSION)
+    }
+
+    // "xpress" is not supported.
+  }
+
   testWithColumnFamilies(
     "RocksDB: test includesPrefix parameter during changelog replay",
     TestWithChangelogCheckpointingEnabled) { colFamiliesEnabled =>