[SPARK-45501][CORE][SQL] Use pattern matching for type checking and c…

…onversion

### What changes were proposed in this pull request?
This pr change to use pattern matching for type checking and conversion instead of the explicit type casting statement in Java code.

The change refer to [JEP 394](https://openjdk.org/jeps/394), and this pr does not include parts of the `hive-thriftserver` module.

Example:

```java
  if (obj instanceof String) {
    String str = (String) obj;
    System.out.println(str);
  }
```
Can be replaced with
```java
  if (obj instanceof String str) {
    System.out.println(str);
  }
```

### Why are the changes needed?
Using `JEP 394: Pattern Matching for instanceof` can bring the following benefits:

1. **Code conciseness**: By eliminating explicit type conversion and redundant variable declarations, the code becomes more concise and easy to read.
2. **Improved safety**: In the past, explicit type conversion was required, and if accidentally converted to the wrong type, a `ClassCastException` would be thrown at runtime. Now, as type checking and type conversion occur in the same step, such errors are no longer possible.
3. **Better semantics**: Previously, instanceof and type casting were two independent steps, which could lead to unclear code intentions. Now, these two steps are merged into one, making the intentions of the code clearer.

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

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

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

Closes apache#43327 from LuciferYang/jep-394.

Authored-by: yangjie01 <[email protected]>
Signed-off-by: yangjie01 <[email protected]>

common/kvstore/src/test/java/org/apache/spark/util/kvstore/ArrayKeyIndexType.java

@@ -29,8 +29,7 @@ public class ArrayKeyIndexType {
 
   @Override
   public boolean equals(Object o) {
-    if (o instanceof ArrayKeyIndexType) {
-      ArrayKeyIndexType other = (ArrayKeyIndexType) o;
+    if (o instanceof ArrayKeyIndexType other) {
       return Arrays.equals(key, other.key) && Arrays.equals(id, other.id);
     }
     return false;

common/kvstore/src/test/java/org/apache/spark/util/kvstore/CustomType1.java

@@ -36,8 +36,7 @@ public class CustomType1 {
 
   @Override
   public boolean equals(Object o) {
-    if (o instanceof CustomType1) {
-      CustomType1 other = (CustomType1) o;
+    if (o instanceof CustomType1 other) {
       return id.equals(other.id) && name.equals(other.name);
     }
     return false;

common/kvstore/src/test/java/org/apache/spark/util/kvstore/CustomType2.java

@@ -30,8 +30,7 @@ public class CustomType2 {
 
   @Override
   public boolean equals(Object o) {
-    if (o instanceof CustomType2) {
-      CustomType2 other = (CustomType2) o;
+    if (o instanceof CustomType2 other) {
       return id.equals(other.id) && parentId.equals(other.parentId);
     }
     return false;

common/kvstore/src/test/java/org/apache/spark/util/kvstore/IntKeyType.java

@@ -31,8 +31,7 @@ public class IntKeyType {
 
   @Override
   public boolean equals(Object o) {
-    if (o instanceof IntKeyType) {
-      IntKeyType other = (IntKeyType) o;
+    if (o instanceof IntKeyType other) {
       return key == other.key && id.equals(other.id) && values.equals(other.values);
     }
     return false;

common/network-common/src/main/java/org/apache/spark/network/protocol/ChunkFetchFailure.java

@@ -62,8 +62,7 @@ public int hashCode() {
 
   @Override
   public boolean equals(Object other) {
-    if (other instanceof ChunkFetchFailure) {
-      ChunkFetchFailure o = (ChunkFetchFailure) other;
+    if (other instanceof ChunkFetchFailure o) {
       return streamChunkId.equals(o.streamChunkId) && errorString.equals(o.errorString);
     }
     return false;

common/network-common/src/main/java/org/apache/spark/network/protocol/ChunkFetchRequest.java

@@ -56,8 +56,7 @@ public int hashCode() {
 
   @Override
   public boolean equals(Object other) {
-    if (other instanceof ChunkFetchRequest) {
-      ChunkFetchRequest o = (ChunkFetchRequest) other;
+    if (other instanceof ChunkFetchRequest o) {
       return streamChunkId.equals(o.streamChunkId);
     }
     return false;

common/network-common/src/main/java/org/apache/spark/network/protocol/ChunkFetchSuccess.java

@@ -75,8 +75,7 @@ public int hashCode() {
 
   @Override
   public boolean equals(Object other) {
-    if (other instanceof ChunkFetchSuccess) {
-      ChunkFetchSuccess o = (ChunkFetchSuccess) other;
+    if (other instanceof ChunkFetchSuccess o) {
       return streamChunkId.equals(o.streamChunkId) && super.equals(o);
     }
     return false;

common/network-common/src/main/java/org/apache/spark/network/protocol/MergedBlockMetaRequest.java

@@ -84,8 +84,7 @@ public int hashCode() {
 
   @Override
   public boolean equals(Object other) {
-    if (other instanceof MergedBlockMetaRequest) {
-      MergedBlockMetaRequest o = (MergedBlockMetaRequest) other;
+    if (other instanceof MergedBlockMetaRequest o) {
       return requestId == o.requestId && shuffleId == o.shuffleId &&
         shuffleMergeId == o.shuffleMergeId && reduceId == o.reduceId &&
         Objects.equal(appId, o.appId);

common/network-common/src/main/java/org/apache/spark/network/protocol/MessageEncoder.java

@@ -59,8 +59,7 @@ public void encode(ChannelHandlerContext ctx, Message in, List<Object> out) thro
         isBodyInFrame = in.isBodyInFrame();
       } catch (Exception e) {
         in.body().release();
-        if (in instanceof AbstractResponseMessage) {
-          AbstractResponseMessage resp = (AbstractResponseMessage) in;
+        if (in instanceof AbstractResponseMessage resp) {
           // Re-encode this message as a failure response.
           String error = e.getMessage() != null ? e.getMessage() : "null";
           logger.error(String.format("Error processing %s for client %s",

common/network-common/src/main/java/org/apache/spark/network/protocol/OneWayMessage.java

@@ -66,8 +66,7 @@ public int hashCode() {
 
   @Override
   public boolean equals(Object other) {
-    if (other instanceof OneWayMessage) {
-      OneWayMessage o = (OneWayMessage) other;
+    if (other instanceof OneWayMessage o) {
       return super.equals(o);
     }
     return false;

common/network-common/src/main/java/org/apache/spark/network/protocol/RpcFailure.java

@@ -60,8 +60,7 @@ public int hashCode() {
 
   @Override
   public boolean equals(Object other) {
-    if (other instanceof RpcFailure) {
-      RpcFailure o = (RpcFailure) other;
+    if (other instanceof RpcFailure o) {
       return requestId == o.requestId && errorString.equals(o.errorString);
     }
     return false;

common/network-common/src/main/java/org/apache/spark/network/protocol/RpcRequest.java

@@ -72,8 +72,7 @@ public int hashCode() {
 
   @Override
   public boolean equals(Object other) {
-    if (other instanceof RpcRequest) {
-      RpcRequest o = (RpcRequest) other;
+    if (other instanceof RpcRequest o) {
       return requestId == o.requestId && super.equals(o);
     }
     return false;

common/network-common/src/main/java/org/apache/spark/network/protocol/RpcResponse.java

@@ -72,8 +72,7 @@ public int hashCode() {
 
   @Override
   public boolean equals(Object other) {
-    if (other instanceof RpcResponse) {
-      RpcResponse o = (RpcResponse) other;
+    if (other instanceof RpcResponse o) {
       return requestId == o.requestId && super.equals(o);
     }
     return false;

common/network-common/src/main/java/org/apache/spark/network/protocol/StreamChunkId.java

@@ -60,8 +60,7 @@ public int hashCode() {
 
   @Override
   public boolean equals(Object other) {
-    if (other instanceof StreamChunkId) {
-      StreamChunkId o = (StreamChunkId) other;
+    if (other instanceof StreamChunkId o) {
       return streamId == o.streamId && chunkIndex == o.chunkIndex;
     }
     return false;

common/network-common/src/main/java/org/apache/spark/network/protocol/StreamFailure.java

@@ -62,8 +62,7 @@ public int hashCode() {
 
   @Override
   public boolean equals(Object other) {
-    if (other instanceof StreamFailure) {
-      StreamFailure o = (StreamFailure) other;
+    if (other instanceof StreamFailure o) {
       return streamId.equals(o.streamId) && error.equals(o.error);
     }
     return false;

common/network-common/src/main/java/org/apache/spark/network/protocol/StreamRequest.java

@@ -61,8 +61,7 @@ public int hashCode() {
 
   @Override
   public boolean equals(Object other) {
-    if (other instanceof StreamRequest) {
-      StreamRequest o = (StreamRequest) other;
+    if (other instanceof StreamRequest o) {
       return streamId.equals(o.streamId);
     }
     return false;

common/network-common/src/main/java/org/apache/spark/network/protocol/StreamResponse.java

@@ -75,8 +75,7 @@ public int hashCode() {
 
   @Override
   public boolean equals(Object other) {
-    if (other instanceof StreamResponse) {
-      StreamResponse o = (StreamResponse) other;
+    if (other instanceof StreamResponse o) {
       return byteCount == o.byteCount && streamId.equals(o.streamId);
     }
     return false;

common/network-common/src/main/java/org/apache/spark/network/protocol/UploadStream.java

@@ -91,8 +91,7 @@ public int hashCode() {
 
   @Override
   public boolean equals(Object other) {
-    if (other instanceof UploadStream) {
-      UploadStream o = (UploadStream) other;
+    if (other instanceof UploadStream o) {
       return requestId == o.requestId && super.equals(o);
     }
     return false;

common/network-common/src/main/java/org/apache/spark/network/sasl/SparkSaslClient.java

@@ -127,17 +127,14 @@ private class ClientCallbackHandler implements CallbackHandler {
     public void handle(Callback[] callbacks) throws UnsupportedCallbackException {
 
       for (Callback callback : callbacks) {
-        if (callback instanceof NameCallback) {
+        if (callback instanceof NameCallback nc) {
           logger.trace("SASL client callback: setting username");
-          NameCallback nc = (NameCallback) callback;
           nc.setName(encodeIdentifier(secretKeyHolder.getSaslUser(secretKeyId)));
-        } else if (callback instanceof PasswordCallback) {
+        } else if (callback instanceof PasswordCallback pc) {
           logger.trace("SASL client callback: setting password");
-          PasswordCallback pc = (PasswordCallback) callback;
           pc.setPassword(encodePassword(secretKeyHolder.getSecretKey(secretKeyId)));
-        } else if (callback instanceof RealmCallback) {
+        } else if (callback instanceof RealmCallback rc) {
           logger.trace("SASL client callback: setting realm");
-          RealmCallback rc = (RealmCallback) callback;
           rc.setText(rc.getDefaultText());
         } else if (callback instanceof RealmChoiceCallback) {
           // ignore (?)

common/network-common/src/main/java/org/apache/spark/network/sasl/SparkSaslServer.java

@@ -156,20 +156,16 @@ private class DigestCallbackHandler implements CallbackHandler {
     @Override
     public void handle(Callback[] callbacks) throws UnsupportedCallbackException {
       for (Callback callback : callbacks) {
-        if (callback instanceof NameCallback) {
+        if (callback instanceof NameCallback nc) {
           logger.trace("SASL server callback: setting username");
-          NameCallback nc = (NameCallback) callback;
           nc.setName(encodeIdentifier(secretKeyHolder.getSaslUser(secretKeyId)));
-        } else if (callback instanceof PasswordCallback) {
+        } else if (callback instanceof PasswordCallback pc) {
           logger.trace("SASL server callback: setting password");
-          PasswordCallback pc = (PasswordCallback) callback;
           pc.setPassword(encodePassword(secretKeyHolder.getSecretKey(secretKeyId)));
-        } else if (callback instanceof RealmCallback) {
+        } else if (callback instanceof RealmCallback rc) {
           logger.trace("SASL server callback: setting realm");
-          RealmCallback rc = (RealmCallback) callback;
           rc.setText(rc.getDefaultText());
-        } else if (callback instanceof AuthorizeCallback) {
-          AuthorizeCallback ac = (AuthorizeCallback) callback;
+        } else if (callback instanceof AuthorizeCallback ac) {
           String authId = ac.getAuthenticationID();
           String authzId = ac.getAuthorizationID();
           ac.setAuthorized(authId.equals(authzId));

common/network-common/src/main/java/org/apache/spark/network/server/TransportChannelHandler.java

@@ -148,8 +148,7 @@ public void channelRead0(ChannelHandlerContext ctx, Message request) throws Exce
   /** Triggered based on events from an {@link io.netty.handler.timeout.IdleStateHandler}. */
   @Override
   public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
-    if (evt instanceof IdleStateEvent) {
-      IdleStateEvent e = (IdleStateEvent) evt;
+    if (evt instanceof IdleStateEvent e) {
       // See class comment for timeout semantics. In addition to ensuring we only timeout while
       // there are outstanding requests, we also do a secondary consistency check to ensure
       // there's no race between the idle timeout and incrementing the numOutstandingRequests

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

@@ -128,8 +128,7 @@ public StreamCallbackWithID receiveStream(
       ByteBuffer messageHeader,
       RpcResponseCallback callback) {
     BlockTransferMessage msgObj = BlockTransferMessage.Decoder.fromByteBuffer(messageHeader);
-    if (msgObj instanceof PushBlockStream) {
-      PushBlockStream message = (PushBlockStream) msgObj;
+    if (msgObj instanceof PushBlockStream message) {
       checkAuth(client, message.appId);
       return mergeManager.receiveBlockDataAsStream(message);
     } else {

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

@@ -131,8 +131,7 @@ protected void decode(ChannelHandlerContext channelHandlerContext,
         List<Object> list) throws Exception {
       delegate.decode(channelHandlerContext, byteBuf, list);
       Object msg = list.get(list.size() - 1);
-      if (msg instanceof RpcRequest) {
-        RpcRequest req = (RpcRequest) msg;
+      if (msg instanceof RpcRequest req) {
         ByteBuffer buffer = req.body().nioByteBuffer();
         byte type = Unpooled.wrappedBuffer(buffer).readByte();
         if (type == BlockTransferMessage.Type.FINALIZE_SHUFFLE_MERGE.id()) {
@@ -171,8 +170,7 @@ static class FinalizedHandler extends SimpleChannelInboundHandler<RpcRequestInte
 
     @Override
     public boolean acceptInboundMessage(Object msg) throws Exception {
-      if (msg instanceof RpcRequestInternal) {
-        RpcRequestInternal rpcRequestInternal = (RpcRequestInternal) msg;
+      if (msg instanceof RpcRequestInternal rpcRequestInternal) {
         return rpcRequestInternal.messageType == BlockTransferMessage.Type.FINALIZE_SHUFFLE_MERGE;
       }
       return false;

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

@@ -68,8 +68,7 @@ public String toString() {
 
   @Override
   public boolean equals(Object other) {
-    if (other instanceof BlockPushReturnCode) {
-      BlockPushReturnCode o = (BlockPushReturnCode) other;
+    if (other instanceof BlockPushReturnCode o) {
       return returnCode == o.returnCode && Objects.equals(failureBlockId, o.failureBlockId);
     }
     return false;

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

@@ -51,8 +51,7 @@ public String toString() {
 
   @Override
   public boolean equals(Object other) {
-    if (other instanceof BlocksRemoved) {
-      BlocksRemoved o = (BlocksRemoved) other;
+    if (other instanceof BlocksRemoved o) {
       return numRemovedBlocks == o.numRemovedBlocks;
     }
     return false;

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

@@ -69,8 +69,7 @@ public String toString() {
 
   @Override
   public boolean equals(Object other) {
-    if (other instanceof ExecutorShuffleInfo) {
-      ExecutorShuffleInfo o = (ExecutorShuffleInfo) other;
+    if (other instanceof ExecutorShuffleInfo o) {
       return Arrays.equals(localDirs, o.localDirs)
         && subDirsPerLocalDir == o.subDirsPerLocalDir
         && Objects.equals(shuffleManager, o.shuffleManager);

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

@@ -69,8 +69,7 @@ public String toString() {
 
   @Override
   public boolean equals(Object other) {
-    if (other instanceof FinalizeShuffleMerge) {
-      FinalizeShuffleMerge o = (FinalizeShuffleMerge) other;
+    if (other instanceof FinalizeShuffleMerge o) {
       return Objects.equal(appId, o.appId)
         && appAttemptId == o.appAttemptId
         && shuffleId == o.shuffleId

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

@@ -57,8 +57,7 @@ public String toString() {
 
   @Override
   public boolean equals(Object other) {
-    if (other instanceof GetLocalDirsForExecutors) {
-      GetLocalDirsForExecutors o = (GetLocalDirsForExecutors) other;
+    if (other instanceof GetLocalDirsForExecutors o) {
       return appId.equals(o.appId) && Arrays.equals(execIds, o.execIds);
     }
     return false;

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

@@ -73,8 +73,7 @@ public String toString() {
 
   @Override
   public boolean equals(Object other) {
-    if (other instanceof LocalDirsForExecutors) {
-      LocalDirsForExecutors o = (LocalDirsForExecutors) other;
+    if (other instanceof LocalDirsForExecutors o) {
       return Arrays.equals(execIds, o.execIds)
         && Arrays.equals(numLocalDirsByExec, o.numLocalDirsByExec)
         && Arrays.equals(allLocalDirs, o.allLocalDirs);

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

@@ -95,8 +95,7 @@ public String toString() {
 
   @Override
   public boolean equals(Object other) {
-    if (other instanceof MergeStatuses) {
-      MergeStatuses o = (MergeStatuses) other;
+    if (other instanceof MergeStatuses o) {
       return Objects.equal(shuffleId, o.shuffleId)
         && Objects.equal(shuffleMergeId, o.shuffleMergeId)
         && Arrays.equals(bitmaps, o.bitmaps)

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

@@ -60,8 +60,7 @@ public String toString() {
 
   @Override
   public boolean equals(Object other) {
-    if (other instanceof OpenBlocks) {
-      OpenBlocks o = (OpenBlocks) other;
+    if (other instanceof OpenBlocks o) {
       return Objects.equals(appId, o.appId)
         && Objects.equals(execId, o.execId)
         && Arrays.equals(blockIds, o.blockIds);

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

@@ -87,8 +87,7 @@ public String toString() {
 
   @Override
   public boolean equals(Object other) {
-    if (other instanceof PushBlockStream) {
-      PushBlockStream o = (PushBlockStream) other;
+    if (other instanceof PushBlockStream o) {
       return Objects.equal(appId, o.appId)
         && appAttemptId == o.appAttemptId
         && shuffleId == o.shuffleId

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

@@ -65,8 +65,7 @@ public String toString() {
 
   @Override
   public boolean equals(Object other) {
-    if (other instanceof RegisterExecutor) {
-      RegisterExecutor o = (RegisterExecutor) other;
+    if (other instanceof RegisterExecutor o) {
       return Objects.equals(appId, o.appId)
         && Objects.equals(execId, o.execId)
         && Objects.equals(executorInfo, o.executorInfo);

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

@@ -60,8 +60,7 @@ public String toString() {
 
   @Override
   public boolean equals(Object other) {
-    if (other instanceof RemoveBlocks) {
-      RemoveBlocks o = (RemoveBlocks) other;
+    if (other instanceof RemoveBlocks o) {
       return Objects.equals(appId, o.appId)
         && Objects.equals(execId, o.execId)
         && Arrays.equals(blockIds, o.blockIds);

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

@@ -69,8 +69,7 @@ public String toString() {
 
   @Override
   public boolean equals(Object other) {
-    if (other != null && other instanceof RemoveShuffleMerge) {
-      RemoveShuffleMerge o = (RemoveShuffleMerge) other;
+    if (other != null && other instanceof RemoveShuffleMerge o) {
       return Objects.equal(appId, o.appId)
         && appAttemptId == o.appAttemptId
         && shuffleId == o.shuffleId

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

@@ -57,8 +57,7 @@ public String toString() {
 
   @Override
   public boolean equals(Object other) {
-    if (other instanceof StreamHandle) {
-      StreamHandle o = (StreamHandle) other;
+    if (other instanceof StreamHandle o) {
       return Objects.equals(streamId, o.streamId)
         && Objects.equals(numChunks, o.numChunks);
     }