Merge remote-tracking branch 'oss-spark/master' into to_avro_improve_NPE

Author: fanyue-xia

LICENSE-binary

@@ -268,6 +268,8 @@ io.fabric8:kubernetes-model-storageclass
 io.fabric8:zjsonpatch
 io.github.java-diff-utils:java-diff-utils
 io.jsonwebtoken:jjwt-api
+io.jsonwebtoken:jjwt-impl
+io.jsonwebtoken:jjwt-jackson
 io.netty:netty-all
 io.netty:netty-buffer
 io.netty:netty-codec

assembly/pom.xml

@@ -345,11 +345,10 @@
       </properties>
     </profile>
 
-    <!-- Pull in jjwt-impl and jjwt-jackson jars -->
     <profile>
-      <id>jjwt</id>
+      <id>jjwt-provided</id>
       <properties>
-        <jjwt.deps.scope>compile</jjwt.deps.scope>
+        <jjwt.deps.scope>provided</jjwt.deps.scope>
       </properties>
     </profile>
 

common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/DownloadFileWritableChannel.java

@@ -19,12 +19,13 @@
 
 import org.apache.spark.network.buffer.ManagedBuffer;
 
+import java.io.IOException;
 import java.nio.channels.WritableByteChannel;
 
 /**
  * A channel for writing data which is fetched to disk, which allows access to the written data only
  * after the writer has been closed.  Used with DownloadFile and DownloadFileManager.
  */
 public interface DownloadFileWritableChannel extends WritableByteChannel {
-  ManagedBuffer closeAndRead();
+  ManagedBuffer closeAndRead() throws IOException;
 }

common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/SimpleDownloadFile.java

@@ -69,7 +69,8 @@ private class SimpleDownloadWritableChannel implements DownloadFileWritableChann
     }
 
     @Override
-    public ManagedBuffer closeAndRead() {
+    public ManagedBuffer closeAndRead() throws IOException {
+      channel.close();
       return new FileSegmentManagedBuffer(transportConf, file, 0, file.length());
     }
 

common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/SimpleDownloadFileSuite.java

@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.shuffle;
+
+import org.apache.spark.network.util.MapConfigProvider;
+import org.apache.spark.network.util.TransportConf;
+import org.junit.jupiter.api.Test;
+
+import java.io.File;
+import java.io.IOException;
+
+import org.junit.jupiter.api.Assertions;
+
+public class SimpleDownloadFileSuite {
+  @Test
+  public void testChannelIsClosedAfterCloseAndRead() throws IOException {
+    File tempFile = File.createTempFile("testChannelIsClosed", ".tmp");
+    tempFile.deleteOnExit();
+    TransportConf conf = new TransportConf("test", MapConfigProvider.EMPTY);
+
+    DownloadFile downloadFile = null;
+    try {
+      downloadFile = new SimpleDownloadFile(tempFile, conf);
+      DownloadFileWritableChannel channel = downloadFile.openForWriting();
+      channel.closeAndRead();
+      Assertions.assertFalse(channel.isOpen(), "Channel should be closed after closeAndRead.");
+    } finally {
+      if (downloadFile != null) {
+        downloadFile.delete();
+      }
+    }
+  }
+}

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

@@ -82,6 +82,12 @@
     ],
     "sqlState" : "22003"
   },
+  "ARTIFACT_ALREADY_EXISTS" : {
+    "message" : [
+      "The artifact <normalizedRemoteRelativePath> already exists. Please choose a different name for the new artifact because it cannot be overwritten."
+    ],
+    "sqlState" : "42713"
+  },
   "ASSIGNMENT_ARITY_MISMATCH" : {
     "message" : [
       "The number of columns or variables assigned or aliased: <numTarget> does not match the number of source expressions: <numExpr>."
@@ -1702,6 +1708,12 @@
     ],
     "sqlState" : "42822"
   },
+  "HINT_UNSUPPORTED_FOR_JDBC_DIALECT" : {
+    "message" : [
+      "The option `hint` is not supported for <jdbcDialect> in JDBC data source. Supported dialects are `MySQLDialect`, `OracleDialect` and `DatabricksDialect`."
+    ],
+    "sqlState" : "42822"
+  },
   "HLL_INVALID_INPUT_SKETCH_BUFFER" : {
     "message" : [
       "Invalid call to <function>; only valid HLL sketch buffers are supported as inputs (such as those produced by the `hll_sketch_agg` function)."
@@ -2744,7 +2756,7 @@
   },
   "INVALID_INTERVAL_WITH_MICROSECONDS_ADDITION" : {
     "message" : [
-      "Cannot add an interval to a date because its microseconds part is not 0. If necessary set <ansiConfig> to \"false\" to bypass this error."
+      "Cannot add an interval to a date because its microseconds part is not 0. To resolve this, cast the input date to a timestamp, which supports the addition of intervals with non-zero microseconds."
     ],
     "sqlState" : "22006"
   },
@@ -3136,6 +3148,24 @@
     ],
     "sqlState" : "42836"
   },
+  "INVALID_RECURSIVE_REFERENCE" : {
+    "message" : [
+      "Invalid recursive reference found inside WITH RECURSIVE clause."
+    ],
+    "subClass" : {
+      "NUMBER" : {
+        "message" : [
+          "Multiple self-references to one recursive CTE are not allowed."
+        ]
+      },
+      "PLACE" : {
+        "message" : [
+          "Recursive references cannot be used on the right side of left outer/semi/anti joins, on the left side of right outer joins, in full outer joins, in aggregates, and in subquery expressions."
+        ]
+      }
+    },
+    "sqlState" : "42836"
+  },
   "INVALID_REGEXP_REPLACE" : {
     "message" : [
       "Could not perform regexp_replace for source = \"<source>\", pattern = \"<pattern>\", replacement = \"<replacement>\" and position = <position>."
@@ -3969,6 +3999,12 @@
     ],
     "sqlState" : "0A000"
   },
+  "NOT_SUPPORTED_CHANGE_SAME_COLUMN" : {
+    "message" : [
+      "ALTER TABLE ALTER/CHANGE COLUMN is not supported for changing <table>'s column <fieldName> including its nested fields multiple times in the same command."
+    ],
+    "sqlState" : "0A000"
+  },
   "NOT_SUPPORTED_COMMAND_FOR_V2_TABLE" : {
     "message" : [
       "<cmd> is not supported for v2 tables."
@@ -5254,6 +5290,11 @@
           "Resilient Distributed Datasets (RDDs)."
         ]
       },
+      "REGISTER_UDAF" : {
+        "message" : [
+          "Registering User Defined Aggregate Functions (UDAFs)."
+        ]
+      },
       "SESSION_BASE_RELATION_TO_DATAFRAME" : {
         "message" : [
           "Invoking SparkSession 'baseRelationToDataFrame'. This is server side developer API"

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

@@ -25,7 +25,7 @@ import java.util.UUID
 
 import scala.jdk.CollectionConverters._
 
-import org.apache.avro.{Schema, SchemaBuilder}
+import org.apache.avro.{AvroTypeException, Schema, SchemaBuilder, SchemaFormatter}
 import org.apache.avro.Schema.{Field, Type}
 import org.apache.avro.Schema.Type._
 import org.apache.avro.file.{DataFileReader, DataFileWriter}
@@ -86,7 +86,7 @@ abstract class AvroSuite
   }
 
   def getAvroSchemaStringFromFiles(filePath: String): String = {
-    new DataFileReader({
+    val schema = new DataFileReader({
       val file = new File(filePath)
       if (file.isFile) {
         file
@@ -96,7 +96,8 @@ abstract class AvroSuite
           .filter(_.getName.endsWith("avro"))
           .head
       }
-    }, new GenericDatumReader[Any]()).getSchema.toString(false)
+    }, new GenericDatumReader[Any]()).getSchema
+    SchemaFormatter.format(AvroUtils.JSON_INLINE_FORMAT, schema)
   }
 
   private def getRootCause(ex: Throwable): Throwable = {

connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/application/ConnectRepl.scala

@@ -31,8 +31,8 @@ import ammonite.util.Util.newLine
 
 import org.apache.spark.SparkBuildInfo.spark_version
 import org.apache.spark.annotation.DeveloperApi
-import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.SparkSession.withLocalConnectServer
+import org.apache.spark.sql.connect.SparkSession
+import org.apache.spark.sql.connect.SparkSession.withLocalConnectServer
 import org.apache.spark.sql.connect.client.{SparkConnectClient, SparkConnectClientParser}
 
 /**

connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/catalog/Catalog.scala

@@ -28,7 +28,7 @@
 import static org.apache.spark.sql.functions.*;
 import static org.apache.spark.sql.RowFactory.create;
 import org.apache.spark.api.java.function.MapFunction;
-import org.apache.spark.sql.test.SparkConnectServerUtils;
+import org.apache.spark.sql.connect.test.SparkConnectServerUtils;
 import org.apache.spark.sql.types.StructType;
 
 /**