(spark) branch master updated: [SPARK-49461][SS][TESTS][FOLLOWUP] Add compatibility test for new commit log

[ka...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

kabhwan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/master by this push:
     new 290b4b31bae2 [SPARK-49461][SS][TESTS][FOLLOWUP] Add compatibility test for new commit log
290b4b31bae2 is described below

commit 290b4b31bae2e02b648d2c5ef61183f337b18f8f
Author: WweiL <z9...@gmail.com>
AuthorDate: Tue Dec 10 07:48:03 2024 +0900

    [SPARK-49461][SS][TESTS][FOLLOWUP] Add compatibility test for new commit log
    
    ### What changes were proposed in this pull request?
    
    Per comment from another PR: https://github.com/apache/spark/pull/48355#discussion_r1868853316
    Add a new compatibility test. This test verifies the following scenario:
    A Spark running under version 3.5 trying to read a commit log generated with the new changelog change but under V1.
    Then the commit log would look like:
    ```
    {"nextBatchWatermarkMs":1,"stateUniqueIds":{}}
    ```
    
    But in Spark 3.5 and before, there is no such `stateUniqueIds` field, so we need to make sure queries won't throw error in such cases.
    
    In the new test, I create a `CommitMetadataLegacy` that only has `nextBatchWatermarkMs` and no `stateUniqueIds`, to read from a commit log as above. This simulates the scenario stated above, and the test passed.
    
    ### Why are the changes needed?
    
    More testing
    
    ### Does this PR introduce _any_ user-facing change?
    
    No
    
    ### How was this patch tested?
    
    Test only addition
    
    ### Was this patch authored or co-authored using generative AI tooling?
    
    No
    
    Closes #49063 from WweiL/commitLog-followup.
    
    Authored-by: WweiL <z9...@gmail.com>
    Signed-off-by: Jungtaek Lim <ka...@gmail.com>
---
 .../spark/sql/streaming/CommitLogSuite.scala       | 62 +++++++++++++++++++++-
 1 file changed, 60 insertions(+), 2 deletions(-)

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/CommitLogSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/CommitLogSuite.scala
index 92bea82b3587..bdf4163c7206 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/CommitLogSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/CommitLogSuite.scala
@@ -17,11 +17,18 @@
 
 package org.apache.spark.sql.streaming
 
-import java.io.{ByteArrayInputStream, FileInputStream, FileOutputStream}
+import java.io.{ByteArrayInputStream, FileInputStream, FileOutputStream, InputStream, OutputStream}
+import java.nio.charset.StandardCharsets.UTF_8
 import java.nio.file.Path
 
+import scala.io.{Source => IOSource}
+
+import org.json4s.{Formats, NoTypeHints}
+import org.json4s.jackson.Serialization
+
 import org.apache.spark.SparkFunSuite
-import org.apache.spark.sql.execution.streaming.{CommitLog, CommitMetadata}
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.execution.streaming.{CommitLog, CommitMetadata, HDFSMetadataLog}
 import org.apache.spark.sql.test.SharedSparkSession
 
 class CommitLogSuite extends SparkFunSuite with SharedSparkSession {
@@ -108,6 +115,57 @@ class CommitLogSuite extends SparkFunSuite with SharedSparkSession {
     assert(commitMetadata.nextBatchWatermarkMs === 233)
     assert(commitMetadata.stateUniqueIds === Map.empty)
   }
+
+  // Test an old version of Spark can ser-de the new version of commit log,
+  // but running under V1 (i.e. no stateUniqueIds)
+  test("v1 Serde backward compatibility") {
+    // This is the json created by a V1 commit log
+    val commitLogV1WithStateUniqueId = """v1
+                        |{"nextBatchWatermarkMs":1,"stateUniqueIds":{}}""".stripMargin
+    val inputStream: ByteArrayInputStream =
+      new ByteArrayInputStream(commitLogV1WithStateUniqueId.getBytes("UTF-8"))
+    val commitMetadata: CommitMetadataLegacy = new CommitLogLegacy(
+      spark, testCommitLogV1FilePath.toString).deserialize(inputStream)
+    assert(commitMetadata.nextBatchWatermarkMs === 1)
+  }
+}
+
+// DO-NOT-MODIFY-THE-CODE-BELOW
+// Below are the legacy commit log code carbon copied from Spark branch-3.5, except
+// adding a "Legacy" to the class names.
+case class CommitMetadataLegacy(nextBatchWatermarkMs: Long = 0) {
+  def json: String = Serialization.write(this)(CommitMetadataLegacy.format)
+}
+
+object CommitMetadataLegacy {
+  implicit val format: Formats = Serialization.formats(NoTypeHints)
+
+  def apply(json: String): CommitMetadataLegacy = Serialization.read[CommitMetadataLegacy](json)
 }
 
+class CommitLogLegacy(sparkSession: SparkSession, path: String)
+  extends HDFSMetadataLog[CommitMetadataLegacy](sparkSession, path) {
+
+  private val VERSION = 1
+  private val EMPTY_JSON = "{}"
 
+  override def deserialize(in: InputStream): CommitMetadataLegacy = {
+    // called inside a try-finally where the underlying stream is closed in the caller
+    val lines = IOSource.fromInputStream(in, UTF_8.name()).getLines()
+    if (!lines.hasNext) {
+      throw new IllegalStateException("Incomplete log file in the offset commit log")
+    }
+    validateVersion(lines.next().trim, VERSION)
+    val metadataJson = if (lines.hasNext) lines.next() else EMPTY_JSON
+    CommitMetadataLegacy(metadataJson)
+  }
+
+  override def serialize(metadata: CommitMetadataLegacy, out: OutputStream): Unit = {
+    // called inside a try-finally where the underlying stream is closed in the caller
+    out.write(s"v${VERSION}".getBytes(UTF_8))
+    out.write('\n')
+
+    // write metadata
+    out.write(metadata.json.getBytes(UTF_8))
+  }
+}


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org