[SPARK-49146][SS] Move assertion errors related to watermark missing …

…in append mode streaming queries to error framework

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

Move assertion errors related to watermark missing in append mode streaming queries to error framework.

### Why are the changes needed?

This is a followup of #43370 to make sure we throw classified user errors rather than assertion errors for better monitoring.

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

No

### How was this patch tested?

Existing unit tests.

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

No

Closes #47656 from bogao007/watermark-error.

Authored-by: bogao007 <bo.gao@databricks.com>
Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com>

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

@@ -5004,6 +5004,12 @@
     },
     "sqlState" : "0A000"
   },
+  "UNSUPPORTED_STREAMING_OPERATOR_WITHOUT_WATERMARK" : {
+    "message" : [
+      "<outputMode> output mode not supported for <statefulOperator> on streaming DataFrames/DataSets without watermark."
+    ],
+    "sqlState" : "0A000"
+  },
   "UNSUPPORTED_SUBQUERY_EXPRESSION_CATEGORY" : {
     "message" : [
       "Unsupported subquery expression:"

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

@@ -2838,6 +2838,17 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase with ExecutionE
         "invalidValue" -> s"'$invalidValue'"))
   }
 
+  def unsupportedStreamingOperatorWithoutWatermark(
+      outputMode: String,
+      statefulOperator: String): AnalysisException = {
+    new AnalysisException(
+      errorClass = "UNSUPPORTED_STREAMING_OPERATOR_WITHOUT_WATERMARK",
+      messageParameters = Map(
+        "outputMode" -> outputMode,
+        "statefulOperator" -> statefulOperator)
+    )
+  }
+
   def conflictingPartitionColumnNamesError(
       distinctPartColLists: Seq[String],
       suspiciousPaths: Seq[Path]): SparkRuntimeException = {

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

@@ -602,11 +602,11 @@ case class StateStoreSaveExec(
           // Update and output only rows being evicted from the StateStore
           // Assumption: watermark predicates must be non-empty if append mode is allowed
           case Some(Append) =>
-            assert(watermarkPredicateForDataForLateEvents.isDefined,
-              "Watermark needs to be defined for streaming aggregation query in append mode")
-
-            assert(watermarkPredicateForKeysForEviction.isDefined,
-              "Watermark needs to be defined for streaming aggregation query in append mode")
+            if (watermarkPredicateForDataForLateEvents.isEmpty ||
+              watermarkPredicateForKeysForEviction.isEmpty) {
+              throw QueryExecutionErrors.unsupportedStreamingOperatorWithoutWatermark(
+                "Append", "aggregations")
+            }
 
             allUpdatesTimeMs += timeTakenMs {
               val filteredIter = applyRemovingRowsOlderThanWatermark(iter,
@@ -903,8 +903,10 @@ case class SessionWindowStateStoreSaveExec(
         // Update and output only rows being evicted from the StateStore
         // Assumption: watermark predicates must be non-empty if append mode is allowed
         case Some(Append) =>
-          assert(watermarkPredicateForDataForEviction.isDefined,
-              "Watermark needs to be defined for session window query in append mode")
+          if (watermarkPredicateForDataForEviction.isEmpty) {
+            throw QueryExecutionErrors.unsupportedStreamingOperatorWithoutWatermark(
+              "Append", "session window aggregations")
+          }
 
           allUpdatesTimeMs += timeTakenMs {
             putToStore(iter, store)