[Spark] Check for null arguments in CDC read (delta-io#4489)

#### Which Delta project/connector is this regarding?

- [x] Spark
- [ ] Standalone
- [ ] Flink
- [ ] Kernel
- [ ] Other (fill in here)

## Description

CDC read query `table_changes` does not support `null` value as the
start or end parameter. However, passing `null` argument via the SQL API
results in NullPointerException. This PR adds a check for `null`
argument and throws a DeltaIllegalArgumentException with the new error
class `DELTA_CDC_READ_NULL_RANGE_BOUNDARY` for the SQL API.

Similarly, providing `null` version for Scala API resulted in
NumberFormatException. Added a new error class `DELTA_VERSION_INVALID`
for this purpose (similar to existing `DELTA_TIMESTAMP_INVALID` for null
timestamps).

## How was this patch tested?

New unit tests are added in
[DeltaCDCScalaSuite](spark/src/test/scala/org/apache/spark/sql/delta/DeltaCDCSuite.scala)
and
[DeltaCDCSQLSuite](spark/src/test/scala/org/apache/spark/sql/delta/DeltaCDCSQLSuite.scala)
to test for cases when either the start or the end parameter is `null`.
Tests expect a proper DeltaIllegalArgumentException (SQL) or
DeltaAnalysisException (Scala) to be thrown, which was not the case
before.

Other unit tests are added to
[DeltaErrorsSuite](spark/src/test/scala/org/apache/spark/sql/delta/DeltaErrorsSuite.scala)
to test if the error messages are correctly reported for the new error
classes.

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

No

Author: ZiyaZa

spark/src/main/resources/error/delta-error-classes.json

@@ -359,6 +359,12 @@
     ],
     "sqlState" : "0AKDC"
   },
+  "DELTA_CDC_READ_NULL_RANGE_BOUNDARY" : {
+    "message" : [
+      "CDC read start/end parameters cannot be null. Please provide a valid version or timestamp."
+    ],
+    "sqlState" : "22004"
+  },
   "DELTA_CHANGE_DATA_FEED_INCOMPATIBLE_DATA_SCHEMA" : {
     "message" : [
       "Retrieving table changes between version <start> and <end> failed because of an incompatible data schema.",
@@ -3031,6 +3037,12 @@
     ],
     "sqlState" : "KD00C"
   },
+  "DELTA_VERSION_INVALID" : {
+    "message" : [
+      "The provided version (<version>) is not a valid version."
+    ],
+    "sqlState" : "42815"
+  },
   "DELTA_VIOLATE_CONSTRAINT_WITH_VALUES" : {
     "message" : [
       "CHECK constraint <constraintName> <expression> violated by row with values:",

spark/src/main/scala/org/apache/spark/sql/delta/DeltaErrors.scala

@@ -498,6 +498,11 @@ trait DeltaErrorsBase
       pos = 0)
   }
 
+  /** Throwable used when a null 'start' or 'end' is provided in CDC reads. */
+  def nullRangeBoundaryInCDCRead(): Throwable = {
+    new DeltaIllegalArgumentException(errorClass = "DELTA_CDC_READ_NULL_RANGE_BOUNDARY")
+  }
+
   /**
    * Throwable used for invalid CDC 'start' and 'end' options, where end < start
    */
@@ -1508,6 +1513,13 @@ trait DeltaErrorsBase
     )
   }
 
+  def versionInvalid(version: String): Throwable = {
+    new DeltaAnalysisException(
+      errorClass = "DELTA_VERSION_INVALID",
+      messageParameters = Array(s"$version")
+    )
+  }
+
   case class TemporallyUnstableInputException(
       userTimestamp: java.sql.Timestamp,
       commitTs: java.sql.Timestamp,

spark/src/main/scala/org/apache/spark/sql/delta/DeltaTableValueFunctions.scala

@@ -29,7 +29,7 @@ import org.apache.spark.sql.delta.sources.DeltaDataSource
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.catalyst.FunctionIdentifier
 import org.apache.spark.sql.catalyst.analysis.{FunctionRegistryBase, NamedRelation, TableFunctionRegistry, UnresolvedLeafNode, UnresolvedRelation}
-import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression, ExpressionInfo, StringLiteral}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression, ExpressionInfo, Literal, StringLiteral}
 import org.apache.spark.sql.catalyst.plans.logical.{LeafNode, LogicalPlan, UnaryNode}
 import org.apache.spark.sql.connector.catalog.V1Table
 import org.apache.spark.sql.execution.datasources.LogicalRelation
@@ -110,6 +110,9 @@ trait CDCStatementBase extends DeltaTableValueFunction {
 
   protected def getOptions: CaseInsensitiveStringMap = {
     def toDeltaOption(keyPrefix: String, value: Expression): (String, String) = {
+      if (value == Literal(null)) {
+        throw DeltaErrors.nullRangeBoundaryInCDCRead()
+      }
       val evaluated = DeltaTableValueFunctionsShims.evaluateTimeOption(value)
       value.dataType match {
         // We dont need to explicitly handle ShortType as it is parsed as IntegerType.

spark/src/main/scala/org/apache/spark/sql/delta/commands/cdc/CDCReader.scala

@@ -241,7 +241,12 @@ trait CDCReaderImpl extends DeltaLogging {
       versionKey: String,
       timestampKey: String): Option[ResolvedCDFVersion] = {
     if (options.containsKey(versionKey)) {
-      Some(ResolvedCDFVersion(options.get(versionKey).toLong, timestamp = None))
+      val version = options.get(versionKey)
+      try {
+        Some(ResolvedCDFVersion(version.toLong, timestamp = None))
+      } catch {
+        case _: NumberFormatException => throw DeltaErrors.versionInvalid(version)
+      }
     } else if (options.containsKey(timestampKey)) {
       val ts = options.get(timestampKey)
       val spec = DeltaTimeTravelSpec(Some(Literal(ts)), None, Some("cdcReader"))

spark/src/test/scala/org/apache/spark/sql/delta/DeltaCDCSQLSuite.scala

@@ -101,6 +101,32 @@ class DeltaCDCSQLSuite extends DeltaCDCSuiteBase with DeltaColumnMappingTestUtil
     }
   }
 
+  private def testNullRangeBoundary(start: Boundary, end: Boundary): Unit = {
+    test(s"range boundary cannot be null - start=$start end=$end") {
+      val tblName = "tbl"
+      withTable(tblName) {
+        createTblWithThreeVersions(tblName = Some(tblName))
+
+        checkError(intercept[DeltaIllegalArgumentException] {
+          cdcRead(new TableName(tblName), start, end)
+        }, "DELTA_CDC_READ_NULL_RANGE_BOUNDARY")
+      }
+    }
+  }
+
+  for (end <- Seq(
+    Unbounded,
+    EndingVersion("null"),
+    EndingVersion("0"),
+    EndingTimestamp(dateFormat.format(new Date(1)))
+  )) {
+    testNullRangeBoundary(StartingVersion("null"), end)
+  }
+
+  for (start <- Seq(StartingVersion("0"), StartingTimestamp(dateFormat.format(new Date(1))))) {
+    testNullRangeBoundary(start, EndingVersion("null"))
+  }
+
   test("select individual column should push down filters") {
     val tblName = "tbl"
     withTable(tblName) {

spark/src/test/scala/org/apache/spark/sql/delta/DeltaCDCSuite.scala

@@ -958,6 +958,52 @@ class DeltaCDCScalaSuite extends DeltaCDCSuiteBase {
     }
   }
 
+  private def testNullRangeBoundary(start: Boundary, end: Boundary): Unit = {
+    test(s"range boundary cannot be null - start=$start end=$end") {
+      val tblName = "tbl"
+      withTable(tblName) {
+        createTblWithThreeVersions(tblName = Some(tblName))
+
+        val expectedError = (start, end) match {
+          case (StartingVersion(null), _) => "DELTA_VERSION_INVALID"
+          case (StartingTimestamp(null), _) => "DELTA_TIMESTAMP_INVALID"
+          case (_, EndingVersion(null)) => "DELTA_VERSION_INVALID"
+          case (_, EndingTimestamp(null)) => "DELTA_TIMESTAMP_INVALID"
+        }
+        val expectedErrorParameters = expectedError match {
+          case "DELTA_VERSION_INVALID" => Map("version" -> "null")
+          case "DELTA_TIMESTAMP_INVALID" => Map("expr" -> "NULL")
+        }
+
+        checkError(
+          intercept[DeltaAnalysisException] {
+            cdcRead(new TableName(tblName), start, end)
+          },
+          expectedError,
+          parameters = expectedErrorParameters)
+      }
+    }
+  }
+
+  for {
+    start <- Seq(StartingVersion("0"), StartingTimestamp(dateFormat.format(new Date(1))))
+    end <- Seq(EndingVersion(null), EndingTimestamp(null))
+  } {
+    testNullRangeBoundary(start, end)
+  }
+
+  for {
+    start <- Seq(StartingVersion(null), StartingTimestamp(null))
+    end <- Seq(
+      Unbounded,
+      EndingVersion(null),
+      EndingTimestamp(null),
+      EndingVersion("0"),
+      EndingTimestamp(dateFormat.format(new Date(1))))
+  } {
+    testNullRangeBoundary(start, end)
+  }
+
   test("filters should be pushed down") {
     val tblName = "tbl"
     withTable(tblName) {

spark/src/test/scala/org/apache/spark/sql/delta/DeltaErrorsSuite.scala

@@ -1521,6 +1521,13 @@ trait DeltaErrorsSuiteBase
       }
       checkError(e, "DELTA_TIMESTAMP_INVALID", "42816", Map("expr" -> expr.sql))
     }
+    {
+      val version = "null"
+      val e = intercept[DeltaAnalysisException] {
+        throw DeltaErrors.versionInvalid(version)
+      }
+      checkError(e, "DELTA_VERSION_INVALID", "42815", Map("version" -> version))
+    }
     {
       val e = intercept[DeltaAnalysisException] {
         throw DeltaErrors.notADeltaSourceException("sample")
@@ -2149,6 +2156,12 @@ trait DeltaErrorsSuiteBase
       }
       checkError(e, "DELTA_UNRECOGNIZED_COLUMN_CHANGE", "42601", Map("otherClass" -> "change1"))
     }
+    {
+      val e = intercept[DeltaIllegalArgumentException] {
+        throw DeltaErrors.nullRangeBoundaryInCDCRead()
+      }
+      checkError(e, "DELTA_CDC_READ_NULL_RANGE_BOUNDARY", "22004", Map.empty[String, String])
+    }
     {
       val e = intercept[DeltaIllegalArgumentException] {
         throw DeltaErrors.endBeforeStartVersionInCDC(2, 1)