[SPARK-51972][SS] Introduce State Store file integrity verification using checksum

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

Introducing file integrity verification for state store files by also generating and uploading checksum file. This allows us to verify the file checksum during read. Introduced a new spark conf to enable/disable this (enabled by default). This can be enabled then disabled and vice versa, on the same checkpoint location, giving users the flexibility to turn on/off as needed.

This implementation is completely backward compatible, it can be enabled on an existing query/checkpoint, and can be disabled later on without breaking the query/checkpoint.

It is currently used for the following state files: delta, snapshot, changelog, zip. We can later enable this for other checkpoint files too e.g. operator metadata files, commit/offset log etc.

### Why are the changes needed?

Integrity verification

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

No

### How was this patch tested?

New tests added

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

No

Closes apache#52787 from micheal-o/file_checksum.

Authored-by: micheal-o <[email protected]>
Signed-off-by: Anish Shrigondekar <[email protected]>

Author: micheal-o

common/utils-java/src/main/java/org/apache/spark/internal/LogKeys.java

@@ -93,6 +93,7 @@ public enum LogKeys implements LogKey {
   CHECKPOINT_PATH,
   CHECKPOINT_ROOT,
   CHECKPOINT_TIME,
+  CHECKSUM,
   CHOSEN_WATERMARK,
   CLASSIFIER,
   CLASS_LOADER,

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

@@ -626,6 +626,13 @@
     ],
     "sqlState" : "42P08"
   },
+  "CHECKPOINT_FILE_CHECKSUM_VERIFICATION_FAILED" : {
+    "message" : [
+      "Checksum verification failed, the file may be corrupted. File: <fileName>",
+      "Expected (file size: <expectedSize>, checksum: <expectedChecksum>), Computed (file size: <computedSize>, checksum: <computedChecksum>)."
+    ],
+    "sqlState" : "XX000"
+  },
   "CHECKPOINT_RDD_BLOCK_ID_NOT_FOUND" : {
     "message" : [
       "Checkpoint block <rddBlockId> not found!",

sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala

@@ -2665,6 +2665,23 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase with ExecutionE
       cause = null)
   }
 
+  def checkpointFileChecksumVerificationFailed(
+      file: Path,
+      expectedSize: Long,
+      expectedChecksum: Int,
+      computedSize: Long,
+      computedChecksum: Int): Throwable = {
+    new SparkException(
+      errorClass = "CHECKPOINT_FILE_CHECKSUM_VERIFICATION_FAILED",
+      messageParameters = Map(
+        "fileName" -> file.toString,
+        "expectedSize" -> expectedSize.toString,
+        "expectedChecksum" -> expectedChecksum.toString,
+        "computedSize" -> computedSize.toString,
+        "computedChecksum" -> computedChecksum.toString),
+      cause = null)
+  }
+
   def cannotReadCheckpoint(expectedVersion: String, actualVersion: String): Throwable = {
     new SparkException (
       errorClass = "CANNOT_LOAD_STATE_STORE.CANNOT_READ_CHECKPOINT",

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

@@ -3430,6 +3430,15 @@ object SQLConf {
       .booleanConf
       .createWithDefault(true)
 
+  val STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED =
+    buildConf("spark.sql.streaming.checkpoint.fileChecksum.enabled")
+      .internal()
+      .doc("When true, checksum would be generated and verified for checkpoint files. " +
+        "This is used to detect file corruption.")
+      .version("4.1.0")
+      .booleanConf
+      .createWithDefault(true)
+
   val PARALLEL_FILE_LISTING_IN_STATS_COMPUTATION =
     buildConf("spark.sql.statistics.parallelFileListingInStatsComputation.enabled")
       .internal()
@@ -6728,6 +6737,8 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
 
   def checkpointLocation: Option[String] = getConf(CHECKPOINT_LOCATION)
 
+  def checkpointFileChecksumEnabled: Boolean = getConf(STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED)
+
   def isUnsupportedOperationCheckEnabled: Boolean = getConf(UNSUPPORTED_OPERATION_CHECK_ENABLED)
 
   def useDeprecatedKafkaOffsetFetching: Boolean = getConf(USE_DEPRECATED_KAFKA_OFFSET_FETCHING)

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

@@ -93,6 +93,8 @@ trait CheckpointFileManager {
    * checkpoint path.
    */
   def createCheckpointDirectory(): Path
+
+  def close(): Unit = {}
 }
 
 object CheckpointFileManager extends Logging {