Revert "[SPARK-54106][SS] State store row checksum implementation"

This reverts commit ac717dd.

### What changes were proposed in this pull request?

Revert: ac717dd

### Why are the changes needed?

Cause tests to be slow in master

### Does this PR introduce _any_ user-facing change?

No

### How was this patch tested?

new Tests

### Was this patch authored or co-authored using generative AI tooling?

No

Closes #52827 from micheal-o/revert_row_check.

Authored-by: micheal-o <[email protected]>
Signed-off-by: Kousuke Saruta <[email protected]>

Author: micheal-o

common/utils/src/main/resources/error/error-conditions.json

@@ -5365,13 +5365,6 @@
     ],
     "sqlState" : "42K06"
   },
-  "STATE_STORE_ROW_CHECKSUM_VERIFICATION_FAILED" : {
-    "message" : [
-      "Row checksum verification failed for stateStore=<stateStoreId>. The row may be corrupted.",
-      "Expected checksum: <expectedChecksum>, Computed checksum: <computedChecksum>."
-    ],
-    "sqlState" : "XXKST"
-  },
   "STATE_STORE_STATE_SCHEMA_FILES_THRESHOLD_EXCEEDED" : {
     "message" : [
       "The number of state schema files <numStateSchemaFiles> exceeds the maximum number of state schema files for this query: <maxStateSchemaFiles>.",

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -2633,31 +2633,6 @@ object SQLConf {
       .checkValue(k => k >= 0, "Must be greater than or equal to 0")
       .createWithDefault(5)
 
-  val STATE_STORE_ROW_CHECKSUM_ENABLED =
-    buildConf("spark.sql.streaming.stateStore.rowChecksum.enabled")
-      .internal()
-      .doc("When true, checksum would be generated and verified for each state store row. " +
-        "This is used to detect row level corruption. " +
-        "Note: This configuration cannot be changed between query restarts " +
-        "from the same checkpoint location.")
-      .version("4.1.0")
-      .booleanConf
-      .createWithDefault(false)
-
-  val STATE_STORE_ROW_CHECKSUM_READ_VERIFICATION_RATIO =
-    buildConf("spark.sql.streaming.stateStore.rowChecksum.readVerificationRatio")
-      .internal()
-      .doc("When specified, Spark will do row checksum verification for every specified " +
-        "number of rows read from state store. The check is to ensure the row read from " +
-        "state store is not corrupt. Default is 0, which means no verification during read " +
-        "but we will still do verification when loading from checkpoint location." +
-        "Example, if you set to 1, it will do the check for every row read from the state store." +
-        "If set to 10, it will do the check for every 10th row read from the state store.")
-      .version("4.1.0")
-      .longConf
-      .checkValue(k => k >= 0, "Must be greater than or equal to 0")
-      .createWithDefault(if (Utils.isTesting) 1 else 0)
-
   val STATEFUL_SHUFFLE_PARTITIONS_INTERNAL =
     buildConf("spark.sql.streaming.internal.stateStore.partitions")
       .doc("WARN: This config is used internally and is not intended to be user-facing. This " +
@@ -6760,11 +6735,6 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
   def stateStoreCoordinatorMaxLaggingStoresToReport: Int =
     getConf(STATE_STORE_COORDINATOR_MAX_LAGGING_STORES_TO_REPORT)
 
-  def stateStoreRowChecksumEnabled: Boolean = getConf(STATE_STORE_ROW_CHECKSUM_ENABLED)
-
-  def stateStoreRowChecksumReadVerificationRatio: Long =
-    getConf(STATE_STORE_ROW_CHECKSUM_READ_VERIFICATION_RATIO)
-
   def checkpointLocation: Option[String] = getConf(CHECKPOINT_LOCATION)
 
   def checkpointFileChecksumEnabled: Boolean = getConf(STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED)

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/OffsetSeq.scala

@@ -113,8 +113,7 @@ object OffsetSeqMetadata extends Logging {
     FLATMAPGROUPSWITHSTATE_STATE_FORMAT_VERSION, STREAMING_AGGREGATION_STATE_FORMAT_VERSION,
     STREAMING_JOIN_STATE_FORMAT_VERSION, STATE_STORE_COMPRESSION_CODEC,
     STATE_STORE_ROCKSDB_FORMAT_VERSION, STATEFUL_OPERATOR_USE_STRICT_DISTRIBUTION,
-    PRUNE_FILTERS_CAN_PRUNE_STREAMING_SUBPLAN, STREAMING_STATE_STORE_ENCODING_FORMAT,
-    STATE_STORE_ROW_CHECKSUM_ENABLED
+    PRUNE_FILTERS_CAN_PRUNE_STREAMING_SUBPLAN, STREAMING_STATE_STORE_ENCODING_FORMAT
   )
 
   /**
@@ -160,8 +159,7 @@ object OffsetSeqMetadata extends Logging {
     STATE_STORE_COMPRESSION_CODEC.key -> CompressionCodec.LZ4,
     STATEFUL_OPERATOR_USE_STRICT_DISTRIBUTION.key -> "false",
     PRUNE_FILTERS_CAN_PRUNE_STREAMING_SUBPLAN.key -> "true",
-    STREAMING_STATE_STORE_ENCODING_FORMAT.key -> "unsaferow",
-    STATE_STORE_ROW_CHECKSUM_ENABLED.key -> "false"
+    STREAMING_STATE_STORE_ENCODING_FORMAT.key -> "unsaferow"
   )
 
   def readValue[T](metadataLog: OffsetSeqMetadata, confKey: ConfigEntry[T]): String = {

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

@@ -17,8 +17,6 @@
 
 package org.apache.spark.sql.execution.streaming.state
 
-import java.util.Map.Entry
-
 import scala.collection.mutable
 import scala.jdk.CollectionConverters._
 
@@ -29,13 +27,10 @@ import org.apache.spark.sql.types.{StructField, StructType}
 trait HDFSBackedStateStoreMap {
   def size(): Int
   def get(key: UnsafeRow): UnsafeRow
-  def put(key: UnsafeRow, value: UnsafeRowWrapper): UnsafeRowWrapper
+  def put(key: UnsafeRow, value: UnsafeRow): UnsafeRow
   def putAll(map: HDFSBackedStateStoreMap): Unit
-  def remove(key: UnsafeRow): UnsafeRowWrapper
+  def remove(key: UnsafeRow): UnsafeRow
   def iterator(): Iterator[UnsafeRowPair]
-  /** Returns entries in the underlying map and skips additional checks done by [[iterator]].
-   * [[iterator]] should be preferred over this. */
-  def entryIterator(): Iterator[Entry[UnsafeRow, UnsafeRowWrapper]]
   def prefixScan(prefixKey: UnsafeRow): Iterator[UnsafeRowPair]
 }
 
@@ -45,78 +40,50 @@ object HDFSBackedStateStoreMap {
   //   the map when the iterator was created
   // - Any updates to the map while iterating through the filtered iterator does not throw
   //   java.util.ConcurrentModificationException
-  type MapType = java.util.concurrent.ConcurrentHashMap[UnsafeRow, UnsafeRowWrapper]
+  type MapType = java.util.concurrent.ConcurrentHashMap[UnsafeRow, UnsafeRow]
 
-  def create(
-      keySchema: StructType,
-      numColsPrefixKey: Int,
-      readVerifier: Option[KeyValueIntegrityVerifier]): HDFSBackedStateStoreMap = {
+  def create(keySchema: StructType, numColsPrefixKey: Int): HDFSBackedStateStoreMap = {
     if (numColsPrefixKey > 0) {
-      new PrefixScannableHDFSBackedStateStoreMap(keySchema, numColsPrefixKey, readVerifier)
+      new PrefixScannableHDFSBackedStateStoreMap(keySchema, numColsPrefixKey)
     } else {
-      new NoPrefixHDFSBackedStateStoreMap(readVerifier)
-    }
-  }
-
-  /** Get the value row from the value wrapper and verify it */
-  def getAndVerifyValueRow(
-      key: UnsafeRow,
-      valueWrapper: UnsafeRowWrapper,
-      readVerifier: Option[KeyValueIntegrityVerifier]): UnsafeRow = {
-    Option(valueWrapper) match {
-      case Some(value) =>
-        readVerifier.foreach(_.verify(key, value))
-        value.unsafeRow()
-      case None => null
+      new NoPrefixHDFSBackedStateStoreMap()
     }
   }
 }
 
-class NoPrefixHDFSBackedStateStoreMap(private val readVerifier: Option[KeyValueIntegrityVerifier])
-    extends HDFSBackedStateStoreMap {
+class NoPrefixHDFSBackedStateStoreMap extends HDFSBackedStateStoreMap {
   private val map = new HDFSBackedStateStoreMap.MapType()
 
   override def size(): Int = map.size()
 
-  override def get(key: UnsafeRow): UnsafeRow = {
-    HDFSBackedStateStoreMap.getAndVerifyValueRow(key, map.get(key), readVerifier)
-  }
+  override def get(key: UnsafeRow): UnsafeRow = map.get(key)
 
-  override def put(key: UnsafeRow, value: UnsafeRowWrapper): UnsafeRowWrapper = map.put(key, value)
+  override def put(key: UnsafeRow, value: UnsafeRow): UnsafeRow = map.put(key, value)
 
   def putAll(other: HDFSBackedStateStoreMap): Unit = {
     other match {
       case o: NoPrefixHDFSBackedStateStoreMap => map.putAll(o.map)
-      case _ => other.entryIterator().foreach { pair => put(pair.getKey, pair.getValue) }
+      case _ => other.iterator().foreach { pair => put(pair.key, pair.value) }
     }
   }
 
-  override def remove(key: UnsafeRow): UnsafeRowWrapper = map.remove(key)
+  override def remove(key: UnsafeRow): UnsafeRow = map.remove(key)
 
   override def iterator(): Iterator[UnsafeRowPair] = {
     val unsafeRowPair = new UnsafeRowPair()
-    entryIterator().map { entry =>
-      val valueRow = HDFSBackedStateStoreMap
-        .getAndVerifyValueRow(entry.getKey, entry.getValue, readVerifier)
-      unsafeRowPair.withRows(entry.getKey, valueRow)
+    map.entrySet.asScala.iterator.map { entry =>
+      unsafeRowPair.withRows(entry.getKey, entry.getValue)
     }
   }
 
-  /** Returns entries in the underlying map and skips additional checks done by [[iterator]].
-   * [[iterator]] should be preferred over this. */
-  override def entryIterator(): Iterator[Entry[UnsafeRow, UnsafeRowWrapper]] = {
-    map.entrySet.asScala.iterator
-  }
-
   override def prefixScan(prefixKey: UnsafeRow): Iterator[UnsafeRowPair] = {
     throw SparkUnsupportedOperationException()
   }
 }
 
 class PrefixScannableHDFSBackedStateStoreMap(
     keySchema: StructType,
-    numColsPrefixKey: Int,
-    private val readVerifier: Option[KeyValueIntegrityVerifier]) extends HDFSBackedStateStoreMap {
+    numColsPrefixKey: Int) extends HDFSBackedStateStoreMap {
 
   private val map = new HDFSBackedStateStoreMap.MapType()
 
@@ -136,11 +103,9 @@ class PrefixScannableHDFSBackedStateStoreMap(
 
   override def size(): Int = map.size()
 
-  override def get(key: UnsafeRow): UnsafeRow = {
-    HDFSBackedStateStoreMap.getAndVerifyValueRow(key, map.get(key), readVerifier)
-  }
+  override def get(key: UnsafeRow): UnsafeRow = map.get(key)
 
-  override def put(key: UnsafeRow, value: UnsafeRowWrapper): UnsafeRowWrapper = {
+  override def put(key: UnsafeRow, value: UnsafeRow): UnsafeRow = {
     val ret = map.put(key, value)
 
     val prefixKey = prefixKeyProjection(key).copy()
@@ -171,11 +136,11 @@ class PrefixScannableHDFSBackedStateStoreMap(
           prefixKeyToKeysMap.put(prefixKey, newSet)
         }
 
-      case _ => other.entryIterator().foreach { pair => put(pair.getKey, pair.getValue) }
+      case _ => other.iterator().foreach { pair => put(pair.key, pair.value) }
     }
   }
 
-  override def remove(key: UnsafeRow): UnsafeRowWrapper = {
+  override def remove(key: UnsafeRow): UnsafeRow = {
     val ret = map.remove(key)
 
     if (ret != null) {
@@ -191,27 +156,15 @@ class PrefixScannableHDFSBackedStateStoreMap(
 
   override def iterator(): Iterator[UnsafeRowPair] = {
     val unsafeRowPair = new UnsafeRowPair()
-    entryIterator().map { entry =>
-      val valueRow = HDFSBackedStateStoreMap
-        .getAndVerifyValueRow(entry.getKey, entry.getValue, readVerifier)
-      unsafeRowPair.withRows(entry.getKey, valueRow)
+    map.entrySet.asScala.iterator.map { entry =>
+      unsafeRowPair.withRows(entry.getKey, entry.getValue)
     }
   }
 
-  /** Returns entries in the underlying map and skips additional checks done by [[iterator]].
-   * [[iterator]] should be preferred over this. */
-  override def entryIterator(): Iterator[Entry[UnsafeRow, UnsafeRowWrapper]] = {
-    map.entrySet.asScala.iterator
-  }
-
   override def prefixScan(prefixKey: UnsafeRow): Iterator[UnsafeRowPair] = {
     val unsafeRowPair = new UnsafeRowPair()
     prefixKeyToKeysMap.getOrDefault(prefixKey, mutable.Set.empty[UnsafeRow])
       .iterator
-      .map { keyRow =>
-        val valueRow = HDFSBackedStateStoreMap
-          .getAndVerifyValueRow(keyRow, map.get(keyRow), readVerifier)
-        unsafeRowPair.withRows(keyRow, valueRow)
-      }
+      .map { key => unsafeRowPair.withRows(key, map.get(key)) }
   }
 }