[SPARK-48888][SS] Remove snapshot creation based on changelog ops size

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

@@ -573,8 +573,7 @@ class RocksDB(
     if (enableChangelogCheckpointing) {
       assert(changelogWriter.isDefined)
       val newVersion = loadedVersion + 1
-      newVersion - lastSnapshotVersion >= conf.minDeltasForSnapshot ||
-        changelogWriter.get.size > 10000
+      newVersion - lastSnapshotVersion >= conf.minDeltasForSnapshot
     } else true
   }
 

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

@@ -96,7 +96,6 @@ abstract class StateStoreChangelogWriter(
   protected var backingFileStream: CancellableFSDataOutputStream =
     fm.createAtomic(file, overwriteIfPossible = true)
   protected var compressedStream: DataOutputStream = compressStream(backingFileStream)
-  var size = 0
 
   def put(key: Array[Byte], value: Array[Byte]): Unit
 
@@ -147,7 +146,6 @@ class StateStoreChangelogWriterV1(
     compressedStream.write(key)
     compressedStream.writeInt(value.size)
     compressedStream.write(value)
-    size += 1
   }
 
   override def delete(key: Array[Byte]): Unit = {
@@ -156,7 +154,6 @@ class StateStoreChangelogWriterV1(
     compressedStream.write(key)
     // -1 in the value field means record deletion.
     compressedStream.writeInt(-1)
-    size += 1
   }
 
   override def merge(key: Array[Byte], value: Array[Byte]): Unit = {
@@ -208,7 +205,6 @@ class StateStoreChangelogWriterV2(
     compressedStream.write(key)
     // -1 in the value field means record deletion.
     compressedStream.writeInt(-1)
-    size += 1
   }
 
   override def merge(key: Array[Byte], value: Array[Byte]): Unit = {
@@ -225,7 +221,6 @@ class StateStoreChangelogWriterV2(
     compressedStream.write(key)
     compressedStream.writeInt(value.size)
     compressedStream.write(value)
-    size += 1
   }
 
   def commit(): Unit = {

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

@@ -291,25 +291,26 @@ class RocksDBSuite extends AlsoTestWithChangelogCheckpointingEnabled with Shared
       db.doMaintenance()
       assert(snapshotVersionsPresent(remoteDir) === Seq(3))
       db.load(3)
-      for (i <- 1 to 10001) {
-        db.put(i.toString, i.toString)
-      }
-      db.commit()
-      db.doMaintenance()
-      // Snapshot should be created this time because the size of the change log > 1000
-      assert(snapshotVersionsPresent(remoteDir) === Seq(3, 4))
-      for (version <- 4 to 7) {
+
+      for (version <- 3 to 7) {
         db.load(version)
         db.commit()
         db.doMaintenance()
       }
-      assert(snapshotVersionsPresent(remoteDir) === Seq(3, 4, 7))
-      for (version <- 8 to 20) {
+      assert(snapshotVersionsPresent(remoteDir) === Seq(3, 6))
+      for (version <- 8 to 17) {
         db.load(version)
         db.commit()
       }
       db.doMaintenance()
-      assert(snapshotVersionsPresent(remoteDir) === Seq(3, 4, 7, 19))
+      assert(snapshotVersionsPresent(remoteDir) === Seq(3, 6, 18))
+    }
+
+    withDB(remoteDir, conf = conf) { db =>
+      db.load(18)
+      db.commit()
+      db.doMaintenance()
+      assert(snapshotVersionsPresent(remoteDir) === Seq(3, 6, 18, 19))
     }
   }