review comments from Micheal

Author: HeartSaVioR

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

@@ -1672,7 +1672,7 @@ class RocksDB(
    * @param cfName   The column family name.
    * @return An iterator of ByteArrayPairs in the given range.
    */
-  def scan(
+  def rangeScan(
       startKey: Option[Array[Byte]],
       endKey: Option[Array[Byte]],
       cfName: String = StateStore.DEFAULT_COL_FAMILY_NAME): NextIterator[ByteArrayPair] = {

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala

@@ -564,7 +564,7 @@ private[sql] class RocksDBStateStoreProvider
       val encodedEndKey = endKey.map(kvEncoder._1.encodeKey)
 
       val rowPair = new UnsafeRowPair()
-      val rocksDbIter = rocksDB.scan(encodedStartKey, encodedEndKey, colFamilyName)
+      val rocksDbIter = rocksDB.rangeScan(encodedStartKey, encodedEndKey, colFamilyName)
       val iter = rocksDbIter.map { kv =>
         rowPair.withRows(kvEncoder._1.decodeKey(kv.key),
           kvEncoder._2.decodeValue(kv.value))
@@ -591,7 +591,7 @@ private[sql] class RocksDBStateStoreProvider
 
       val encodedStartKey = startKey.map(kvEncoder._1.encodeKey)
       val encodedEndKey = endKey.map(kvEncoder._1.encodeKey)
-      val rocksDbIter = rocksDB.scan(encodedStartKey, encodedEndKey, colFamilyName)
+      val rocksDbIter = rocksDB.rangeScan(encodedStartKey, encodedEndKey, colFamilyName)
 
       val rowPair = new UnsafeRowPair()
       val iter = rocksDbIter.flatMap { kv =>

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

@@ -1634,7 +1634,7 @@ class RocksDBStateStoreSuite extends StateStoreSuiteBase[RocksDBStateStoreProvid
     -32L, -64L, -256L, 64L, 32L, 1024L, 4096L, 0L)
 
   testWithColumnFamiliesAndEncodingTypes("rocksdb range scan - rangeScan",
-    TestWithBothChangelogCheckpointingEnabledAndDisabled) { colFamiliesEnabled =>
+    TestWithChangelogCheckpointingDisabled) { colFamiliesEnabled =>
 
     tryWithProviderResource(newStoreProvider(keySchemaWithRangeScan,
       RangeKeyScanStateEncoderSpec(keySchemaWithRangeScan, Seq(0)),
@@ -1704,6 +1704,12 @@ class RocksDBStateStoreSuite extends StateStoreSuiteBase[RocksDBStateStoreProvid
         val negResults = negIter.map(_.key.getLong(0)).toList
         negIter.close()
         assert(negResults === diverseTimestamps.filter(ts => ts >= -300 && ts < 0).sorted)
+
+        // Both None: scan entire column family
+        val allIter = store.rangeScan(None, None, cfName)
+        val allResults = allIter.map(_.key.getLong(0)).toList
+        allIter.close()
+        assert(allResults === diverseTimestamps.sorted)
       } finally {
         if (!store.hasCommitted) store.abort()
       }
@@ -1712,7 +1718,7 @@ class RocksDBStateStoreSuite extends StateStoreSuiteBase[RocksDBStateStoreProvid
 
   testWithColumnFamiliesAndEncodingTypes(
     "rocksdb range scan - scan with multiple key2 values within same key1 range",
-    TestWithBothChangelogCheckpointingEnabledAndDisabled) { colFamiliesEnabled =>
+    TestWithChangelogCheckpointingDisabled) { colFamiliesEnabled =>
 
     tryWithProviderResource(newStoreProvider(keySchemaWithRangeScan,
       RangeKeyScanStateEncoderSpec(keySchemaWithRangeScan, Seq(0)),
@@ -1752,7 +1758,7 @@ class RocksDBStateStoreSuite extends StateStoreSuiteBase[RocksDBStateStoreProvid
 
   testWithColumnFamiliesAndEncodingTypes(
     "rocksdb range scan - rangeScanWithMultiValues",
-    TestWithBothChangelogCheckpointingEnabledAndDisabled) { colFamiliesEnabled =>
+    TestWithChangelogCheckpointingDisabled) { colFamiliesEnabled =>
 
     if (colFamiliesEnabled) {
       tryWithProviderResource(newStoreProvider(
@@ -1833,6 +1839,12 @@ class RocksDBStateStoreSuite extends StateStoreSuiteBase[RocksDBStateStoreProvid
           negIter.close()
           assert(negResults.distinct === diverseTimestamps
             .filter(ts => ts >= -300 && ts < 0).sorted)
+
+          // Both None: scan entire column family
+          val allIter = store.rangeScanWithMultiValues(None, None, cfName)
+          val allResults = allIter.map(_.key.getLong(0)).toList
+          allIter.close()
+          assert(allResults.distinct === diverseTimestamps.sorted)
         } finally {
           if (!store.hasCommitted) store.abort()
         }