update

wayneguow · Aug 14, 2024 · da67f7c · da67f7c
1 parent 3a48bfe
commit da67f7c
Show file tree

Hide file tree

Showing 11 changed files with 35 additions and 30 deletions.
diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json
@@ -1059,12 +1059,6 @@
     ],
     "sqlState" : "42K03"
   },
-  "DATA_SOURCE_UNSUPPORTED_STREAMING_OUTPUT_MODE" : {
-    "message" : [
-      "Data source <className> does not support streaming output mode <outputMode>."
-    ],
-    "sqlState" : "42KDE"
-  },
   "DATETIME_OVERFLOW" : {
     "message" : [
       "Datetime operation overflow: <operation>."
@@ -2946,12 +2940,6 @@
     ],
     "sqlState" : "42601"
   },
-  "INVALID_STREAMING_OUTPUT_MODE" : {
-    "message" : [
-      "Invalid streaming output mode: <outputMode>. Accepted output modes are 'Append', 'Complete', 'Update'."
-    ],
-    "sqlState" : "42KDE"
-  },
   "INVALID_SUBQUERY_EXPRESSION" : {
     "message" : [
       "Invalid subquery:"
@@ -4156,6 +4144,29 @@
     ],
     "sqlState" : "42601"
   },
+  "STREAMING_OUTPUT_MODE" : {
+    "message" : [
+      "Invalid streaming output mode: <outputMode>."
+    ],
+    "subClass" : {
+      "INVALID" : {
+        "message" : [
+          "Accepted output modes are 'Append', 'Complete', 'Update'."
+        ]
+      },
+      "UNSUPPORTED_DATASOURCE" : {
+        "message" : [
+          "This output mode is not supported in Data Source <className>."
+        ]
+      },
+      "UNSUPPORTED_OPERATION" : {
+        "message" : [
+          "This output mode is not supported for <operation> on streaming DataFrames/DataSets."
+        ]
+      }
+    },
+    "sqlState" : "42KDE"
+  },
   "STREAMING_PYTHON_RUNNER_INITIALIZATION_FAILURE" : {
     "message" : [
       "Streaming Runner initialization failed, returned <resFromPython>. Cause: <msg>"
@@ -4986,12 +4997,6 @@
     },
     "sqlState" : "42K0E"
   },
-  "UNSUPPORTED_OUTPUT_MODE_FOR_STREAMING_OPERATION" : {
-    "message" : [
-      "The output mode <outputMode> is not supported for <operation> on streaming DataFrames/DataSets."
-    ],
-    "sqlState" : "42KDE"
-  },
   "UNSUPPORTED_OVERWRITE" : {
     "message" : [
       "Can't overwrite the target that is also being read from."

diff --git a/sql/api/src/main/scala/org/apache/spark/sql/catalyst/streaming/InternalOutputModes.scala b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/streaming/InternalOutputModes.scala
@@ -58,7 +58,7 @@ private[sql] object InternalOutputModes {
       case "update" =>
         OutputMode.Update
       case _ => throw new SparkIllegalArgumentException(
-          errorClass = "INVALID_STREAMING_OUTPUT_MODE",
+          errorClass = "STREAMING_OUTPUT_MODE.INVALID",
           messageParameters = Map("outputMode" -> outputMode))
     }
   }

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
@@ -1640,7 +1640,7 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
   def dataSourceOutputModeUnsupportedError(
       className: String, outputMode: OutputMode): AnalysisException = {
     new AnalysisException(
-      errorClass = "DATA_SOURCE_UNSUPPORTED_STREAMING_OUTPUT_MODE",
+      errorClass = "STREAMING_OUTPUT_MODE.UNSUPPORTED_DATASOURCE",
       messageParameters = Map(
         "className" -> className,
         "outputMode" -> outputMode.toString.toLowerCase(Locale.ROOT)))
@@ -1649,7 +1649,7 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
   def unsupportedOutputModeForStreamingOperationError(
       outputMode: OutputMode, operation: String): AnalysisException = {
     new AnalysisException(
-      errorClass = "UNSUPPORTED_OUTPUT_MODE_FOR_STREAMING_OPERATION",
+      errorClass = "STREAMING_OUTPUT_MODE.UNSUPPORTED_OPERATION",
       messageParameters = Map(
         "outputMode" -> outputMode.toString().toLowerCase(Locale.ROOT),
         "operation" -> operation))

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
@@ -1753,7 +1753,7 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase with ExecutionE
       outputMode: OutputMode,
       operation: String): SparkUnsupportedOperationException = {
     new SparkUnsupportedOperationException(
-      errorClass = "UNSUPPORTED_OUTPUT_MODE_FOR_STREAMING_OPERATION",
+      errorClass = "STREAMING_OUTPUT_MODE.UNSUPPORTED_OPERATION",
       messageParameters = Map(
         "outputMode" -> outputMode.toString().toLowerCase(Locale.ROOT),
         "operation" -> operation))

diff --git a/...st/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala b/...st/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala
@@ -1052,7 +1052,7 @@ class UnsupportedOperationsSuite extends SparkFunSuite with SQLHelper {
         exception = intercept[AnalysisException] {
           UnsupportedOperationChecker.checkForStreaming(wrapInStreaming(plan), outputMode)
         },
-        errorClass = "UNSUPPORTED_OUTPUT_MODE_FOR_STREAMING_OPERATION",
+        errorClass = "STREAMING_OUTPUT_MODE.UNSUPPORTED_OPERATION",
         sqlState = "42KDE",
         parameters = Map(
           "outputMode" -> outputMode.toString.toLowerCase(Locale.ROOT),

diff --git a/...yst/src/test/scala/org/apache/spark/sql/catalyst/streaming/InternalOutputModesSuite.scala b/...yst/src/test/scala/org/apache/spark/sql/catalyst/streaming/InternalOutputModesSuite.scala
@@ -41,7 +41,7 @@ class InternalOutputModesSuite extends SparkFunSuite {
       exception = intercept[SparkIllegalArgumentException] {
         InternalOutputModes(outputMode)
       },
-      errorClass = "INVALID_STREAMING_OUTPUT_MODE",
+      errorClass = "STREAMING_OUTPUT_MODE.INVALID",
       parameters = Map("outputMode" -> outputMode))
   }
 }
diff --git a/...src/test/scala/org/apache/spark/sql/execution/python/PythonStreamingDataSourceSuite.scala b/...src/test/scala/org/apache/spark/sql/execution/python/PythonStreamingDataSourceSuite.scala
@@ -1035,7 +1035,7 @@ class PythonStreamingDataSourceWriteSuite extends PythonDataSourceSuiteBase {
         exception = intercept[AnalysisException] {
           runQuery("complete")
         },
-        errorClass = "UNSUPPORTED_OUTPUT_MODE_FOR_STREAMING_OPERATION",
+        errorClass = "STREAMING_OUTPUT_MODE.UNSUPPORTED_OPERATION",
         sqlState = "42KDE",
         parameters = Map(
           "outputMode" -> "complete",

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/FileStreamSinkSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/FileStreamSinkSuite.scala
@@ -280,7 +280,7 @@ abstract class FileStreamSinkSuite extends StreamTest {
           exception = intercept[AnalysisException] {
             df.writeStream.format("parquet").outputMode(mode).start(dir.getCanonicalPath)
           },
-          errorClass = "DATA_SOURCE_UNSUPPORTED_STREAMING_OUTPUT_MODE",
+          errorClass = "STREAMING_OUTPUT_MODE.UNSUPPORTED_DATASOURCE",
           sqlState = "42KDE",
           parameters = Map("className" -> "parquet", "outputMode" -> mode))
       }

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQuerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQuerySuite.scala
@@ -1458,7 +1458,7 @@ class StreamingQuerySuite extends StreamTest with BeforeAndAfter with Logging wi
               .option("checkpointLocation", checkpointDir.getCanonicalPath)
               .start(outputDir.getCanonicalPath)
           },
-          errorClass = "UNSUPPORTED_OUTPUT_MODE_FOR_STREAMING_OPERATION",
+          errorClass = "STREAMING_OUTPUT_MODE.UNSUPPORTED_OPERATION",
           sqlState = "42KDE",
           parameters = Map(
             "outputMode" -> "append",

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingSessionWindowSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingSessionWindowSuite.scala
@@ -607,7 +607,7 @@ class StreamingSessionWindowSuite extends StreamTest
           CheckAnswer() // this is just to trigger the exception
         )
       },
-      errorClass = "UNSUPPORTED_OUTPUT_MODE_FOR_STREAMING_OPERATION",
+      errorClass = "STREAMING_OUTPUT_MODE.UNSUPPORTED_OPERATION",
       sqlState = "42KDE",
       parameters = Map(
         "outputMode" -> OutputMode.Update().toString.toLowerCase(Locale.ROOT),
@@ -625,7 +625,7 @@ class StreamingSessionWindowSuite extends StreamTest
           CheckAnswer() // this is just to trigger the exception
         )
       },
-      errorClass = "UNSUPPORTED_OUTPUT_MODE_FOR_STREAMING_OPERATION",
+      errorClass = "STREAMING_OUTPUT_MODE.UNSUPPORTED_OPERATION",
       sqlState = "42KDE",
       parameters = Map(
         "outputMode" -> OutputMode.Update().toString.toLowerCase(Locale.ROOT),

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithStateChainingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithStateChainingSuite.scala
@@ -191,7 +191,7 @@ class TransformWithStateChainingSuite extends StreamTest {
             StartStream()
           )
         },
-        errorClass = "UNSUPPORTED_OUTPUT_MODE_FOR_STREAMING_OPERATION",
+        errorClass = "STREAMING_OUTPUT_MODE.UNSUPPORTED_OPERATION",
         sqlState = "42KDE",
         parameters = Map(
           "outputMode" -> "append",