Merge remote-tracking branch 'upstream' into vector-avg-sum-functions

Author: zhidongqu-db

.github/workflows/build_and_test.yml

@@ -235,7 +235,7 @@ jobs:
     needs: precondition
     if: fromJson(needs.precondition.outputs.required).build == 'true'
     runs-on: ubuntu-latest
-    timeout-minutes: 120
+    timeout-minutes: 150
     strategy:
       fail-fast: false
       matrix:

common/utils-java/src/main/java/org/apache/spark/internal/LogKeys.java

@@ -316,6 +316,7 @@ public enum LogKeys implements LogKey {
   JAVA_VERSION,
   JAVA_VM_NAME,
   JOB_ID,
+  JOB_IDS,
   JOIN_CONDITION,
   JOIN_CONDITION_SUB_EXPR,
   JOIN_TYPE,

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

@@ -1043,6 +1043,48 @@
     },
     "sqlState" : "21S01"
   },
+  "CURSOR_ALREADY_EXISTS" : {
+    "message" : [
+      "Cannot declare cursor <cursorName> because it already exists in the current scope."
+    ],
+    "sqlState" : "42723"
+  },
+  "CURSOR_ALREADY_OPEN" : {
+    "message" : [
+      "Cannot open cursor <cursorName> because it is already open."
+    ],
+    "sqlState" : "24502"
+  },
+  "CURSOR_NOT_FOUND" : {
+    "message" : [
+      "Cursor <cursorName> not found in the current scope."
+    ],
+    "sqlState" : "42883"
+  },
+  "CURSOR_NOT_OPEN" : {
+    "message" : [
+      "Cannot fetch from or close cursor <cursorName> because it is not open."
+    ],
+    "sqlState" : "24501"
+  },
+  "CURSOR_NO_MORE_ROWS" : {
+    "message" : [
+      "No more rows available to fetch from cursor <cursorName>."
+    ],
+    "sqlState" : "02000"
+  },
+  "CURSOR_OUTSIDE_SCRIPT" : {
+    "message" : [
+      "Cursor operations can only be used within SQL scripts."
+    ],
+    "sqlState" : "0A000"
+  },
+  "CURSOR_REFERENCE_INVALID_QUALIFIER" : {
+    "message" : [
+      "Cursor reference <cursorName> is invalid. Cursor references can only have at most one qualifier (e.g., label.cursor)."
+    ],
+    "sqlState" : "42601"
+  },
   "CYCLIC_FUNCTION_REFERENCE" : {
     "message" : [
       "Cyclic function reference detected: <path>."
@@ -2796,6 +2838,12 @@
     },
     "sqlState" : "HY109"
   },
+  "INVALID_CURSOR_DECLARATION" : {
+    "message" : [
+      "Cursors must be declared after variable/condition declarations, and before handlers and other statements."
+    ],
+    "sqlState" : "42601"
+  },
   "INVALID_DATETIME_PATTERN" : {
     "message" : [
       "Unrecognized datetime pattern: <pattern>."
@@ -3155,7 +3203,7 @@
       },
       "WRONG_PLACE_OF_DECLARATION" : {
         "message" : [
-          "Handlers must be declared after variable/condition declaration, and before other statements."
+          "Handlers must be declared after variable/condition/cursor declarations, and before other statements."
         ]
       }
     },
@@ -7052,6 +7100,11 @@
           "<variableName> is a VARIABLE and cannot be updated using the SET statement. Use SET VARIABLE <variableName> = ... instead."
         ]
       },
+      "SQL_CURSOR" : {
+        "message" : [
+          "SQL cursor operations (DECLARE CURSOR, OPEN, FETCH, CLOSE) are not supported."
+        ]
+      },
       "SQL_SCRIPTING" : {
         "message" : [
           "SQL Scripting is under development and not all features are supported. SQL Scripting enables users to write procedural SQL including control flow and error handling. To enable existing features set <sqlScriptingEnabled> to `true`."

connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala

@@ -3482,4 +3482,29 @@ class AvroV2Suite extends AvroSuite with ExplainSuiteHelper {
       checkAnswer(readDf, df)
     }
   }
+
+  test("Geospatial types are not supported in Avro") {
+    withTempDir { dir =>
+      // Temporary directory for writing the test data.
+      val tempDir = new File(dir, "files").getCanonicalPath
+      // Test data: WKB representation of POINT(1 2).
+      val wkb = "0101000000000000000000F03F0000000000000040"
+      // Test GEOMETRY and GEOGRAPHY data types.
+      val geoTestCases = Seq(
+        (s"ST_GeomFromWKB(X'$wkb')", "\"GEOMETRY(0)\""),
+        (s"ST_GeogFromWKB(X'$wkb')", "\"GEOGRAPHY(4326)\"")
+      )
+      geoTestCases.foreach { case (expr, expectedType) =>
+        checkError(
+          exception = intercept[AnalysisException] {
+            sql(s"select $expr as g").write.format("avro").mode("overwrite").save(tempDir)
+          },
+          condition = "UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE",
+          parameters = Map(
+            "columnName" -> "`g`",
+            "columnType" -> expectedType,
+            "format" -> "Avro"))
+      }
+    }
+  }
 }

core/src/main/scala/org/apache/spark/Dependency.scala

@@ -90,7 +90,8 @@ class ShuffleDependency[K: ClassTag, V: ClassTag, C: ClassTag](
     val mapSideCombine: Boolean = false,
     val shuffleWriterProcessor: ShuffleWriteProcessor = new ShuffleWriteProcessor,
     val rowBasedChecksums: Array[RowBasedChecksum] = ShuffleDependency.EMPTY_ROW_BASED_CHECKSUMS,
-    private val _checksumMismatchFullRetryEnabled: Boolean = false)
+    private val _checksumMismatchFullRetryEnabled: Boolean = false,
+    val checksumMismatchQueryLevelRollbackEnabled: Boolean = false)
   extends Dependency[Product2[K, V]] with Logging {
 
   def this(

core/src/main/scala/org/apache/spark/SparkContext.scala

@@ -3152,6 +3152,7 @@ object SparkContext extends Logging {
   private[spark] val SPARK_SCHEDULER_POOL = "spark.scheduler.pool"
   private[spark] val RDD_SCOPE_KEY = "spark.rdd.scope"
   private[spark] val RDD_SCOPE_NO_OVERRIDE_KEY = "spark.rdd.scope.noOverride"
+  private[spark] val SQL_EXECUTION_ID_KEY = "spark.sql.execution.id"
 
   /**
    * Executor id for the driver.  In earlier versions of Spark, this was `<driver>`, but this was