Merge remote-tracking branch 'origin/master' into arrowudf

Author: asl3

common/network-common/src/main/java/org/apache/spark/network/client/TransportResponseHandler.java

@@ -26,8 +26,6 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import io.netty.channel.Channel;
-import org.apache.commons.lang3.tuple.ImmutablePair;
-import org.apache.commons.lang3.tuple.Pair;
 
 import org.apache.spark.internal.SparkLogger;
 import org.apache.spark.internal.SparkLoggerFactory;
@@ -45,6 +43,7 @@
 import org.apache.spark.network.server.MessageHandler;
 import static org.apache.spark.network.util.NettyUtils.getRemoteAddress;
 import org.apache.spark.network.util.TransportFrameDecoder;
+import org.apache.spark.util.Pair;
 
 /**
  * Handler that processes server responses, in response to requests issued from a
@@ -96,7 +95,7 @@ public void removeRpcRequest(long requestId) {
 
   public void addStreamCallback(String streamId, StreamCallback callback) {
     updateTimeOfLastRequest();
-    streamCallbacks.offer(ImmutablePair.of(streamId, callback));
+    streamCallbacks.offer(Pair.of(streamId, callback));
   }
 
   @VisibleForTesting
@@ -125,7 +124,7 @@ private void failOutstandingRequests(Throwable cause) {
     }
     for (Pair<String, StreamCallback> entry : streamCallbacks) {
       try {
-        entry.getValue().onFailure(entry.getKey(), cause);
+        entry.getRight().onFailure(entry.getLeft(), cause);
       } catch (Exception e) {
         logger.warn("StreamCallback.onFailure throws exception", e);
       }
@@ -236,7 +235,7 @@ public void handle(ResponseMessage message) throws Exception {
     } else if (message instanceof StreamResponse resp) {
       Pair<String, StreamCallback> entry = streamCallbacks.poll();
       if (entry != null) {
-        StreamCallback callback = entry.getValue();
+        StreamCallback callback = entry.getRight();
         if (resp.byteCount > 0) {
           StreamInterceptor<ResponseMessage> interceptor = new StreamInterceptor<>(
             this, resp.streamId, resp.byteCount, callback);
@@ -262,7 +261,7 @@ public void handle(ResponseMessage message) throws Exception {
     } else if (message instanceof StreamFailure resp) {
       Pair<String, StreamCallback> entry = streamCallbacks.poll();
       if (entry != null) {
-        StreamCallback callback = entry.getValue();
+        StreamCallback callback = entry.getRight();
         try {
           callback.onFailure(resp.streamId, new RuntimeException(resp.error));
         } catch (IOException ioe) {

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

@@ -26,13 +26,12 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import io.netty.channel.Channel;
-import org.apache.commons.lang3.tuple.ImmutablePair;
-import org.apache.commons.lang3.tuple.Pair;
 
 import org.apache.spark.internal.SparkLogger;
 import org.apache.spark.internal.SparkLoggerFactory;
 import org.apache.spark.network.buffer.ManagedBuffer;
 import org.apache.spark.network.client.TransportClient;
+import org.apache.spark.util.Pair;
 
 /**
  * StreamManager which allows registration of an Iterator<ManagedBuffer>, which are
@@ -127,7 +126,7 @@ public static Pair<Long, Integer> parseStreamChunkId(String streamChunkId) {
       "Stream id and chunk index should be specified.";
     long streamId = Long.valueOf(array[0]);
     int chunkIndex = Integer.valueOf(array[1]);
-    return ImmutablePair.of(streamId, chunkIndex);
+    return Pair.of(streamId, chunkIndex);
   }
 
   @Override

common/network-common/src/test/java/org/apache/spark/network/ChunkFetchRequestHandlerSuite.java

@@ -27,15 +27,14 @@
 
 import static org.mockito.Mockito.*;
 
-import org.apache.commons.lang3.tuple.ImmutablePair;
-import org.apache.commons.lang3.tuple.Pair;
 import org.apache.spark.network.buffer.ManagedBuffer;
 import org.apache.spark.network.client.TransportClient;
 import org.apache.spark.network.protocol.*;
 import org.apache.spark.network.server.ChunkFetchRequestHandler;
 import org.apache.spark.network.server.NoOpRpcHandler;
 import org.apache.spark.network.server.OneForOneStreamManager;
 import org.apache.spark.network.server.RpcHandler;
+import org.apache.spark.util.Pair;
 
 public class ChunkFetchRequestHandlerSuite {
 
@@ -54,7 +53,7 @@ public void handleChunkFetchRequest() throws Exception {
       .thenAnswer(invocationOnMock0 -> {
         Object response = invocationOnMock0.getArguments()[0];
         ExtendedChannelPromise channelFuture = new ExtendedChannelPromise(channel);
-        responseAndPromisePairs.add(ImmutablePair.of(response, channelFuture));
+        responseAndPromisePairs.add(Pair.of(response, channelFuture));
         return channelFuture;
       });
 

common/network-common/src/test/java/org/apache/spark/network/RpcIntegrationSuite.java

@@ -26,8 +26,6 @@
 
 import com.google.common.collect.Sets;
 import com.google.common.io.Files;
-import org.apache.commons.lang3.tuple.ImmutablePair;
-import org.apache.commons.lang3.tuple.Pair;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
@@ -41,6 +39,7 @@
 import org.apache.spark.network.util.JavaUtils;
 import org.apache.spark.network.util.MapConfigProvider;
 import org.apache.spark.network.util.TransportConf;
+import org.apache.spark.util.Pair;
 
 public class RpcIntegrationSuite {
   static TransportConf conf;
@@ -408,7 +407,7 @@ private Pair<Set<String>, Set<String>> checkErrorsContain(
         notFound.add(contain);
       }
     }
-    return new ImmutablePair<>(remainingErrors, notFound);
+    return new Pair<>(remainingErrors, notFound);
   }
 
   private static class VerifyingStreamCallback implements StreamCallbackWithID {

common/network-common/src/test/java/org/apache/spark/network/TransportRequestHandlerSuite.java

@@ -28,8 +28,6 @@
 import static org.junit.jupiter.api.Assertions.*;
 import static org.mockito.Mockito.*;
 
-import org.apache.commons.lang3.tuple.ImmutablePair;
-import org.apache.commons.lang3.tuple.Pair;
 import org.apache.spark.network.buffer.ManagedBuffer;
 import org.apache.spark.network.client.RpcResponseCallback;
 import org.apache.spark.network.client.TransportClient;
@@ -39,6 +37,7 @@
 import org.apache.spark.network.server.RpcHandler;
 import org.apache.spark.network.server.StreamManager;
 import org.apache.spark.network.server.TransportRequestHandler;
+import org.apache.spark.util.Pair;
 
 public class TransportRequestHandlerSuite {
 
@@ -53,7 +52,7 @@ public void handleStreamRequest() throws Exception {
       .thenAnswer(invocationOnMock0 -> {
         Object response = invocationOnMock0.getArguments()[0];
         ExtendedChannelPromise channelFuture = new ExtendedChannelPromise(channel);
-        responseAndPromisePairs.add(ImmutablePair.of(response, channelFuture));
+        responseAndPromisePairs.add(Pair.of(response, channelFuture));
         return channelFuture;
       });
 
@@ -145,7 +144,7 @@ public MergedBlockMetaReqHandler getMergedBlockMetaReqHandler() {
     when(channel.writeAndFlush(any())).thenAnswer(invocationOnMock0 -> {
       Object response = invocationOnMock0.getArguments()[0];
       ExtendedChannelPromise channelFuture = new ExtendedChannelPromise(channel);
-      responseAndPromisePairs.add(ImmutablePair.of(response, channelFuture));
+      responseAndPromisePairs.add(Pair.of(response, channelFuture));
       return channelFuture;
     });
 

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

@@ -28,7 +28,6 @@
 
 import org.apache.commons.lang3.builder.ToStringBuilder;
 import org.apache.commons.lang3.builder.ToStringStyle;
-import org.apache.commons.lang3.tuple.Pair;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -56,6 +55,7 @@
 import org.apache.spark.network.util.JavaUtils;
 import org.apache.spark.network.util.NettyUtils;
 import org.apache.spark.network.util.TransportConf;
+import org.apache.spark.util.Pair;
 
 /**
  * Manages converting shuffle BlockIds into physical segments of local files, from a process outside
@@ -400,7 +400,7 @@ public Map<String, String[]> getLocalDirs(String appId, Set<String> execIds) {
         }
         return Pair.of(exec, info.localDirs);
       })
-      .collect(Collectors.toMap(Pair::getKey, Pair::getValue));
+      .collect(Collectors.toMap(Pair::getLeft, Pair::getRight));
   }
 
   /**

common/utils/src/main/java/org/apache/spark/util/Pair.java

@@ -0,0 +1,29 @@
+/*
+ * 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.util;
+
+/**
+ * An immutable pair of values. Note that the fields are intentionally designed to be `getLeft` and
+ * `getRight` instead of `left` and `right` in order to mitigate the migration burden
+ * from `org.apache.commons.lang3.tuple.Pair`.
+ */
+public record Pair<L, R>(L getLeft, R getRight) {
+  public static <L, R> Pair<L, R> of(L left, R right) {
+    return new Pair<>(left, right);
+  }
+}

common/utils/src/main/scala/org/apache/spark/util/SparkStringUtils.scala

@@ -16,11 +16,89 @@
  */
 package org.apache.spark.util
 
+import java.util.HexFormat
+import java.util.concurrent.atomic.AtomicBoolean
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.util.ArrayImplicits._
+
 private[spark] trait SparkStringUtils {
+  private final lazy val SPACE_DELIMITED_UPPERCASE_HEX =
+    HexFormat.of().withDelimiter(" ").withUpperCase()
+
+  /**
+   * Returns a pretty string of the byte array which prints each byte as a hex digit and add
+   * spaces between them. For example, [1A C0].
+   */
+  def getHexString(bytes: Array[Byte]): String = {
+    s"[${SPACE_DELIMITED_UPPERCASE_HEX.formatHex(bytes)}]"
+  }
+
+  def sideBySide(left: String, right: String): Seq[String] = {
+    sideBySide(left.split("\n").toImmutableArraySeq, right.split("\n").toImmutableArraySeq)
+  }
+
+  def sideBySide(left: Seq[String], right: Seq[String]): Seq[String] = {
+    val maxLeftSize = left.map(_.length).max
+    val leftPadded = left ++ Seq.fill(math.max(right.size - left.size, 0))("")
+    val rightPadded = right ++ Seq.fill(math.max(left.size - right.size, 0))("")
+
+    leftPadded.zip(rightPadded).map { case (l, r) =>
+      (if (l == r) " " else "!") + l + (" " * ((maxLeftSize - l.length) + 3)) + r
+    }
+  }
+
   def stringToSeq(str: String): Seq[String] = {
     import org.apache.spark.util.ArrayImplicits._
     str.split(",").map(_.trim()).filter(_.nonEmpty).toImmutableArraySeq
   }
 }
 
-private[spark] object SparkStringUtils extends SparkStringUtils
+private[spark] object SparkStringUtils extends SparkStringUtils with Logging {
+
+  /** Whether we have warned about plan string truncation yet. */
+  private val truncationWarningPrinted = new AtomicBoolean(false)
+
+  /**
+   * Format a sequence with semantics similar to calling .mkString(). Any elements beyond
+   * `maxFields` will be dropped and replaced by a "... N more fields" placeholder.
+   *
+   * @return
+   *   the trimmed and formatted string.
+   */
+  def truncatedString[T](
+      seq: Seq[T],
+      start: String,
+      sep: String,
+      end: String,
+      maxFields: Int,
+      customToString: Option[T => String] = None): String = {
+    if (seq.length > maxFields) {
+      if (truncationWarningPrinted.compareAndSet(false, true)) {
+        logWarning(
+          "Truncated the string representation of a plan since it was too large. This " +
+            s"behavior can be adjusted by setting 'spark.sql.debug.maxToStringFields'.")
+      }
+      val numFields = math.max(0, maxFields)
+      val restNum = seq.length - numFields
+      val ending = (if (numFields == 0) "" else sep) +
+        (if (restNum == 0) "" else s"... $restNum more fields") + end
+      if (customToString.isDefined) {
+        seq.take(numFields).map(customToString.get).mkString(start, sep, ending)
+      } else {
+        seq.take(numFields).mkString(start, sep, ending)
+      }
+    } else {
+      if (customToString.isDefined) {
+        seq.map(customToString.get).mkString(start, sep, end)
+      } else {
+        seq.mkString(start, sep, end)
+      }
+    }
+  }
+
+  /** Shorthand for calling truncatedString() without start or end strings. */
+  def truncatedString[T](seq: Seq[T], sep: String, maxFields: Int): String = {
+    truncatedString(seq, "", sep, "", maxFields)
+  }
+}

common/utils/src/test/java/org/apache/spark/util/SparkLoggerSuiteBase.java

@@ -22,7 +22,6 @@
 import java.nio.file.Files;
 import java.util.List;
 
-import org.apache.commons.lang3.tuple.Pair;
 import org.apache.logging.log4j.Level;
 import org.junit.jupiter.api.Test;
 

dev/checkstyle.xml

@@ -183,6 +183,7 @@
         <module name="IllegalImport">
             <property name="illegalPkgs" value="org.apache.log4j" />
             <property name="illegalPkgs" value="org.apache.commons.lang" />
+            <property name="illegalPkgs" value="org.apache.commons.lang3.tuple" />
             <property name="illegalClasses" value="org.apache.commons.lang3.JavaVersion" />
         </module>
         <module name="RegexpSinglelineJava">