Don't use JValue in validateSourceVersion (#713)

* Don't use JValue in validateSourceVersion

* fix tests

---------

Co-authored-by: Sun Cao <sun.cao+data@databricks.com>

Author: littlegrasscao

client/src/main/scala/io/delta/sharing/spark/DeltaSharingErrors.scala

@@ -40,8 +40,8 @@ object DeltaSharingErrors {
   }
 
   def unsupportedTableReaderVersion(supportedVersion: Long, tableVersion: Long): Throwable = {
-    new IllegalStateException(s"The table reader version ${tableVersion} is larger than " +
-      s"supported reader version $supportedVersion. Please upgrade to a new release."
+    new IllegalStateException(s"The table reader version ${tableVersion} is not equal to " +
+      s"supported reader version $supportedVersion."
     )
   }
 

client/src/main/scala/io/delta/sharing/spark/DeltaSharingSourceOffset.scala

@@ -18,9 +18,7 @@ package io.delta.sharing.spark
 
 // scalastyle:off import.ordering.noEmptyLine
 import org.apache.spark.sql.connector.read.streaming.{Offset => OffsetV2}
-import org.apache.spark.sql.execution.streaming.{Offset, SerializedOffset}
-import org.json4s._
-import org.json4s.jackson.JsonMethods.parse
+import org.apache.spark.sql.execution.streaming.Offset
 
 import io.delta.sharing.client.util.JsonUtils
 
@@ -86,37 +84,19 @@ object DeltaSharingSourceOffset {
     offset match {
       case o: DeltaSharingSourceOffset => o
       case s =>
-        validateSourceVersion(s.json)
         val o = JsonUtils.fromJson[DeltaSharingSourceOffset](s.json)
+        validateSourceVersion(o)
         if (o.tableId != tableId) {
           throw DeltaSharingErrors.nonExistentDeltaSharingTable(o.tableId)
         }
         o
     }
   }
 
-  private def validateSourceVersion(json: String): Unit = {
-    val parsedJson = parse(json)
-    val versionOpt = jsonOption(parsedJson \ "sourceVersion").map {
-      case i: JInt => i.num.longValue
-      case other => throw DeltaSharingErrors.invalidSourceVersion(other.toString)
-    }
-    if (versionOpt.isEmpty) {
-      throw DeltaSharingErrors.cannotFindSourceVersionException(json)
-    }
-
-    val maxVersion = VERSION_1
-
-    if (versionOpt.get > maxVersion) {
-      throw DeltaSharingErrors.unsupportedTableReaderVersion(maxVersion, versionOpt.get)
-    }
-  }
-
-  /** Return an option that translates JNothing to None */
-  private def jsonOption(json: JValue): Option[JValue] = {
-    json match {
-      case JNothing => None
-      case value: JValue => Some(value)
+  private def validateSourceVersion(offset: DeltaSharingSourceOffset) = {
+    // Only version 1 is supported for now.
+    if (offset.sourceVersion != VERSION_1) {
+      throw DeltaSharingErrors.unsupportedTableReaderVersion(VERSION_1, offset.sourceVersion)
     }
   }
 

spark/src/test/scala/io/delta/sharing/spark/DeltaSharingSourceOffsetSuite.scala

@@ -18,14 +18,10 @@ package io.delta.sharing.spark
 
 import java.util.UUID
 
+import com.fasterxml.jackson.databind.exc.InvalidFormatException
 import org.apache.spark.sql.QueryTest
-import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.connector.read.streaming.ReadMaxFiles
 import org.apache.spark.sql.execution.streaming.SerializedOffset
-import org.apache.spark.sql.streaming.{DataStreamReader, StreamingQueryException, Trigger}
 import org.apache.spark.sql.test.SharedSparkSession
-import org.apache.spark.sql.types.{StringType, StructField, StructType, TimestampType}
-import org.scalatest.time.SpanSugar._
 
 class DeltaSharingSourceOffsetSuite extends QueryTest
   with SharedSparkSession with DeltaSharingIntegrationTest {
@@ -48,7 +44,7 @@ class DeltaSharingSourceOffsetSuite extends QueryTest
     val e = intercept[IllegalStateException] {
       DeltaSharingSourceOffset(UUID.randomUUID().toString, SerializedOffset(json))
     }
-    assert(e.getMessage.contains("Please upgrade to a new release"))
+    assert(e.getMessage.contains("is not equal to supported reader version"))
   }
 
   test("DeltaSharingSourceOffset sourceVersion - invalid value") {
@@ -61,10 +57,10 @@ class DeltaSharingSourceOffsetSuite extends QueryTest
         |  "isStartingVersion": true
         |}
       """.stripMargin
-    val e = intercept[IllegalStateException] {
+    val e = intercept[InvalidFormatException] {
       DeltaSharingSourceOffset(UUID.randomUUID().toString, SerializedOffset(json))
     }
-    for (msg <- Seq("foo", "invalid")) {
+    for (msg <- Seq("foo", "not a valid")) {
       assert(e.getMessage.contains(msg))
     }
   }
@@ -81,7 +77,7 @@ class DeltaSharingSourceOffsetSuite extends QueryTest
     val e = intercept[IllegalStateException] {
       DeltaSharingSourceOffset(UUID.randomUUID().toString, SerializedOffset(json))
     }
-    for (msg <- Seq("Cannot find", "sourceVersion")) {
+    for (msg <- Seq("The table reader version", "is not equal to")) {
       assert(e.getMessage.contains(msg))
     }
   }