[SPARK-40742][CORE][SQL] Fix Java compilation warnings related to generic type

### What changes were proposed in this pull request?
This pr aims to fix following Java compilation warnings related to generic type:

```
2022-10-08T01:43:33.6487078Z /home/runner/work/spark/spark/core/src/main/java/org/apache/spark/SparkThrowable.java:54: warning: [rawtypes] found raw type: HashMap
2022-10-08T01:43:33.6487456Z     return new HashMap();
2022-10-08T01:43:33.6487682Z                ^
2022-10-08T01:43:33.6487957Z   missing type arguments for generic class HashMap<K,V>
2022-10-08T01:43:33.6488617Z   where K,V are type-variables:
2022-10-08T01:43:33.6488911Z     K extends Object declared in class HashMap
2022-10-08T01:43:33.6489211Z     V extends Object declared in class HashMap

2022-10-08T01:50:21.5951932Z /home/runner/work/spark/spark/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SupportsAtomicPartitionManagement.java:55: warning: [rawtypes] found raw type: Map
2022-10-08T01:50:21.5999993Z       createPartitions(new InternalRow[]{ident}, new Map[]{properties});
2022-10-08T01:50:21.6000343Z                                                      ^
2022-10-08T01:50:21.6000642Z   missing type arguments for generic class Map<K,V>
2022-10-08T01:50:21.6001272Z   where K,V are type-variables:
2022-10-08T01:50:21.6001569Z     K extends Object declared in interface Map
2022-10-08T01:50:21.6002109Z     V extends Object declared in interface Map

2022-10-08T01:50:21.6006655Z /home/runner/work/spark/spark/sql/catalyst/src/main/java/org/apache/spark/sql/connector/util/V2ExpressionSQLBuilder.java:216: warning: [rawtypes] found raw type: Literal
2022-10-08T01:50:21.6007121Z   protected String visitLiteral(Literal literal) {
2022-10-08T01:50:21.6007395Z                                 ^
2022-10-08T01:50:21.6007673Z   missing type arguments for generic class Literal<T>
2022-10-08T01:50:21.6008032Z   where T is a type-variable:
2022-10-08T01:50:21.6008324Z     T extends Object declared in interface Literal

2022-10-08T01:50:21.6008785Z /home/runner/work/spark/spark/sql/catalyst/src/main/java/org/apache/spark/sql/util/NumericHistogram.java:56: warning: [rawtypes] found raw type: Comparable
2022-10-08T01:50:21.6009223Z   public static class Coord implements Comparable {
2022-10-08T01:50:21.6009503Z                                        ^
2022-10-08T01:50:21.6009791Z   missing type arguments for generic class Comparable<T>
2022-10-08T01:50:21.6010137Z   where T is a type-variable:
2022-10-08T01:50:21.6010433Z     T extends Object declared in interface Comparable
2022-10-08T01:50:21.6010976Z /home/runner/work/spark/spark/sql/catalyst/src/main/java/org/apache/spark/sql/util/NumericHistogram.java:191: warning: [unchecked] unchecked method invocation: method sort in class Collections is applied to given types
2022-10-08T01:50:21.6011474Z       Collections.sort(tmp_bins);
2022-10-08T01:50:21.6011714Z                       ^
2022-10-08T01:50:21.6012050Z   required: List<T>
2022-10-08T01:50:21.6012296Z   found: ArrayList<Coord>
2022-10-08T01:50:21.6012604Z   where T is a type-variable:
2022-10-08T01:50:21.6012926Z     T extends Comparable<? super T> declared in method <T>sort(List<T>)

2022-10-08T02:13:38.0769617Z /home/runner/work/spark/spark/sql/hive-thriftserver/src/main/java/org/apache/hive/service/cli/operation/OperationManager.java:85: warning: [rawtypes] found raw type: AbstractWriterAppender
2022-10-08T02:13:38.0770287Z     AbstractWriterAppender ap = new LogDivertAppender(this, OperationLog.getLoggingLevel(loggingMode));
2022-10-08T02:13:38.0770645Z     ^
2022-10-08T02:13:38.0770947Z   missing type arguments for generic class AbstractWriterAppender<M>
2022-10-08T02:13:38.0771330Z   where M is a type-variable:
2022-10-08T02:13:38.0771665Z     M extends WriterManager declared in class AbstractWriterAppender

2022-10-08T02:13:38.0774487Z /home/runner/work/spark/spark/sql/hive-thriftserver/src/main/java/org/apache/hive/service/cli/operation/LogDivertAppender.java:268: warning: [rawtypes] found raw type: Layout
2022-10-08T02:13:38.0774940Z         Layout l = ap.getLayout();
2022-10-08T02:13:38.0775173Z         ^
2022-10-08T02:13:38.0775441Z   missing type arguments for generic class Layout<T>
2022-10-08T02:13:38.0775849Z   where T is a type-variable:
2022-10-08T02:13:38.0776359Z     T extends Serializable declared in interface Layout

2022-10-08T02:19:55.0035795Z [WARNING] /home/runner/work/spark/spark/connector/avro/src/main/java/org/apache/spark/sql/avro/SparkAvroKeyOutputFormat.java:56:17:  [rawtypes] found raw type: SparkAvroKeyRecordWriter
2022-10-08T02:19:55.0037287Z [WARNING] /home/runner/work/spark/spark/connector/avro/src/main/java/org/apache/spark/sql/avro/SparkAvroKeyOutputFormat.java:56:13:  [unchecked] unchecked call to SparkAvroKeyRecordWriter(Schema,GenericData,CodecFactory,OutputStream,int,Map<String,String>) as a member of the raw type SparkAvroKeyRecordWriter
2022-10-08T02:19:55.0038442Z [WARNING] /home/runner/work/spark/spark/connector/avro/src/main/java/org/apache/spark/sql/avro/SparkAvroKeyOutputFormat.java:75:31:  [rawtypes] found raw type: DataFileWriter
2022-10-08T02:19:55.0039370Z [WARNING] /home/runner/work/spark/spark/connector/avro/src/main/java/org/apache/spark/sql/avro/SparkAvroKeyOutputFormat.java:75:27:  [unchecked] unchecked call to DataFileWriter(DatumWriter<D>) as a member of the raw type DataFileWriter

```

### Why are the changes needed?
Fix Java compilation warnings.

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

### How was this patch tested?
Pass GitHub Actions.

Closes apache#38198 from LuciferYang/fix-java-warn.

Lead-authored-by: yangjie01 <[email protected]>
Co-authored-by: YangJie <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>

Author: LuciferYang

connector/avro/src/main/java/org/apache/spark/sql/avro/SparkAvroKeyOutputFormat.java

@@ -25,6 +25,7 @@
 import org.apache.avro.file.CodecFactory;
 import org.apache.avro.file.DataFileWriter;
 import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericDatumWriter;
 import org.apache.avro.generic.GenericRecord;
 import org.apache.avro.mapred.AvroKey;
 import org.apache.avro.mapreduce.AvroKeyOutputFormat;
@@ -53,7 +54,7 @@ protected RecordWriter<AvroKey<GenericRecord>, NullWritable> create(
         CodecFactory compressionCodec,
         OutputStream outputStream,
         int syncInterval) throws IOException {
-      return new SparkAvroKeyRecordWriter(
+      return new SparkAvroKeyRecordWriter<>(
         writerSchema, dataModel, compressionCodec, outputStream, syncInterval, metadata);
     }
   }
@@ -72,7 +73,7 @@ class SparkAvroKeyRecordWriter<T> extends RecordWriter<AvroKey<T>, NullWritable>
       OutputStream outputStream,
       int syncInterval,
       Map<String, String> metadata) throws IOException {
-    this.mAvroFileWriter = new DataFileWriter(dataModel.createDatumWriter(writerSchema));
+    this.mAvroFileWriter = new DataFileWriter<>(new GenericDatumWriter<>(writerSchema, dataModel));
     for (Map.Entry<String, String> entry : metadata.entrySet()) {
       this.mAvroFileWriter.setMeta(entry.getKey(), entry.getValue());
     }

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

@@ -1075,7 +1075,7 @@ abstract class AvroSuite
           .save(s"$tempDir/${UUID.randomUUID()}")
       }.getMessage
       assert(message.contains("Caused by: java.lang.NullPointerException: "))
-      assert(message.contains("null in string in field Name"))
+      assert(message.contains("null value for (non-nullable) string at test_schema.Name"))
     }
   }
 

core/src/main/java/org/apache/spark/SparkThrowable.java

@@ -51,7 +51,7 @@ default boolean isInternalError() {
   }
 
   default Map<String, String> getMessageParameters() {
-    return new HashMap();
+    return new HashMap<>();
   }
 
   default QueryContext[] getQueryContext() { return new QueryContext[0]; }

sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SupportsAtomicPartitionManagement.java

@@ -45,6 +45,7 @@
 @Experimental
 public interface SupportsAtomicPartitionManagement extends SupportsPartitionManagement {
 
+  @SuppressWarnings("unchecked")
   @Override
   default void createPartition(
       InternalRow ident,

sql/catalyst/src/main/java/org/apache/spark/sql/connector/util/V2ExpressionSQLBuilder.java

@@ -47,7 +47,7 @@ public class V2ExpressionSQLBuilder {
 
   public String build(Expression expr) {
     if (expr instanceof Literal) {
-      return visitLiteral((Literal) expr);
+      return visitLiteral((Literal<?>) expr);
     } else if (expr instanceof NamedReference) {
       return visitNamedReference((NamedReference) expr);
     } else if (expr instanceof Cast) {
@@ -213,7 +213,7 @@ public String build(Expression expr) {
     }
   }
 
-  protected String visitLiteral(Literal literal) {
+  protected String visitLiteral(Literal<?> literal) {
     return literal.toString();
   }
 

sql/catalyst/src/main/java/org/apache/spark/sql/util/NumericHistogram.java

@@ -19,6 +19,7 @@
 
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.List;
 import java.util.Random;
 
 
@@ -53,20 +54,20 @@ public class NumericHistogram {
    *
    * @since 3.3.0
    */
-  public static class Coord implements Comparable {
+  public static class Coord implements Comparable<Coord> {
     public double x;
     public double y;
 
     @Override
-    public int compareTo(Object other) {
-      return Double.compare(x, ((Coord) other).x);
+    public int compareTo(Coord other) {
+      return Double.compare(x, other.x);
     }
   }
 
   // Class variables
   private int nbins;
   private int nusedbins;
-  private ArrayList<Coord> bins;
+  private List<Coord> bins;
   private Random prng;
 
   /**
@@ -146,7 +147,7 @@ public void addBin(double x, double y, int b) {
    */
   public void allocate(int num_bins) {
     nbins = num_bins;
-    bins = new ArrayList<Coord>();
+    bins = new ArrayList<>();
     nusedbins = 0;
   }
 
@@ -163,7 +164,7 @@ public void merge(NumericHistogram other) {
       // by deserializing the ArrayList of (x,y) pairs into an array of Coord objects
       nbins = other.nbins;
       nusedbins = other.nusedbins;
-      bins = new ArrayList<Coord>(nusedbins);
+      bins = new ArrayList<>(nusedbins);
       for (int i = 0; i < other.nusedbins; i += 1) {
         Coord bin = new Coord();
         bin.x = other.getBin(i).x;
@@ -174,7 +175,7 @@ public void merge(NumericHistogram other) {
       // The aggregation buffer already contains a partial histogram. Therefore, we need
       // to merge histograms using Algorithm #2 from the Ben-Haim and Tom-Tov paper.
 
-      ArrayList<Coord> tmp_bins = new ArrayList<Coord>(nusedbins + other.nusedbins);
+      List<Coord> tmp_bins = new ArrayList<>(nusedbins + other.nusedbins);
       // Copy all the histogram bins from us and 'other' into an overstuffed histogram
       for (int i = 0; i < nusedbins; i++) {
         Coord bin = new Coord();

sql/hive-thriftserver/src/main/java/org/apache/hive/service/cli/operation/LogDivertAppender.java

@@ -17,6 +17,7 @@
 
 package org.apache.hive.service.cli.operation;
 import java.io.CharArrayWriter;
+import java.io.Serializable;
 import java.util.Map;
 import java.util.regex.Pattern;
 
@@ -265,7 +266,7 @@ private static StringLayout initLayout(OperationLog.LoggingLevel loggingMode) {
     Map<String, Appender> appenders = root.getAppenders();
     for (Appender ap : appenders.values()) {
       if (ap.getClass().equals(ConsoleAppender.class)) {
-        Layout l = ap.getLayout();
+        Layout<? extends Serializable> l = ap.getLayout();
         if (l instanceof StringLayout) {
           layout = (StringLayout) l;
           break;

sql/hive-thriftserver/src/main/java/org/apache/hive/service/cli/operation/OperationManager.java

@@ -39,7 +39,7 @@
 import org.apache.hive.service.cli.session.HiveSession;
 import org.apache.hive.service.rpc.thrift.TRowSet;
 import org.apache.hive.service.rpc.thrift.TTableSchema;
-import org.apache.logging.log4j.core.appender.AbstractWriterAppender;
+import org.apache.logging.log4j.core.Appender;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -82,7 +82,7 @@ public synchronized void stop() {
 
   private void initOperationLogCapture(String loggingMode) {
     // Register another Appender (with the same layout) that talks to us.
-    AbstractWriterAppender ap = new LogDivertAppender(this, OperationLog.getLoggingLevel(loggingMode));
+    Appender ap = new LogDivertAppender(this, OperationLog.getLoggingLevel(loggingMode));
     ((org.apache.logging.log4j.core.Logger)org.apache.logging.log4j.LogManager.getRootLogger()).addAppender(ap);
     ap.start();
   }