rerun plan stability

Author: tanelk

.asf.yaml

@@ -13,7 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-# https://cwiki.apache.org/confluence/display/INFRA/.asf.yaml+features+for+git+repositories
+# https://cwiki.apache.org/confluence/display/INFRA/git+-+.asf.yaml+features
 ---
 github:
   description: "Apache Spark - A unified analytics engine for large-scale data processing"

R/pkg/R/functions.R

@@ -247,7 +247,7 @@ NULL
 #'            used to transform the data.  The first argument is the key, the second argument
 #'            is the value.
 #'          }
-#' @param zero a \code{Column} used as the initial value in \code{array_aggregate}
+#' @param initialValue a \code{Column} used as the initial value in \code{array_aggregate}
 #' @param merge a \code{function} a binary function \code{(Column, Column) -> Column}
 #'          used in \code{array_aggregate}to merge values (the second argument)
 #'          into accumulator (the first argument).
@@ -3666,11 +3666,11 @@ invoke_higher_order_function <- function(name, cols, funs) {
 #' @aliases array_aggregate array_aggregate,characterOrColumn,Column,function-method
 #' @note array_aggregate since 3.1.0
 setMethod("array_aggregate",
-          signature(x = "characterOrColumn", zero = "Column", merge = "function"),
-          function(x, zero, merge, finish = NULL) {
+          signature(x = "characterOrColumn", initialValue = "Column", merge = "function"),
+          function(x, initialValue, merge, finish = NULL) {
             invoke_higher_order_function(
               "ArrayAggregate",
-              cols = list(x, zero),
+              cols = list(x, initialValue),
               funs = if (is.null(finish)) {
                 list(merge)
               } else {

R/pkg/R/generics.R

@@ -780,7 +780,8 @@ setGeneric("approxCountDistinct", function(x, ...) { standardGeneric("approxCoun
 
 #' @rdname column_collection_functions
 #' @name NULL
-setGeneric("array_aggregate", function(x, zero, merge, ...) { standardGeneric("array_aggregate") })
+setGeneric("array_aggregate",
+           function(x, initialValue, merge, ...) { standardGeneric("array_aggregate") })
 
 #' @rdname column_collection_functions
 #' @name NULL

common/unsafe/src/main/java/org/apache/spark/unsafe/types/UTF8String.java

@@ -683,6 +683,17 @@ public UTF8String trimRight() {
     return copyUTF8String(0, e);
   }
 
+  /**
+   * Trims at most `numSpaces` space characters (ASCII 32) from the end of this string.
+   */
+  public UTF8String trimTrailingSpaces(int numSpaces) {
+    assert numSpaces > 0;
+    int endIdx = numBytes - 1;
+    int trimTo = numBytes - numSpaces;
+    while (endIdx >= trimTo && getByte(endIdx) == 0x20) endIdx--;
+    return copyUTF8String(0, endIdx);
+  }
+
   /**
    * Trims instances of the given trim string from the end of this string.
    *
@@ -1065,16 +1076,20 @@ public UTF8String replace(UTF8String search, UTF8String replace) {
     return buf.build();
   }
 
-  // TODO: Need to use `Code Point` here instead of Char in case the character longer than 2 bytes
-  public UTF8String translate(Map<Character, Character> dict) {
+  public UTF8String translate(Map<String, String> dict) {
     String srcStr = this.toString();
 
     StringBuilder sb = new StringBuilder();
-    for(int k = 0; k< srcStr.length(); k++) {
-      if (null == dict.get(srcStr.charAt(k))) {
-        sb.append(srcStr.charAt(k));
-      } else if ('\0' != dict.get(srcStr.charAt(k))){
-        sb.append(dict.get(srcStr.charAt(k)));
+    int charCount = 0;
+    for (int k = 0; k < srcStr.length(); k += charCount) {
+      int codePoint = srcStr.codePointAt(k);
+      charCount = Character.charCount(codePoint);
+      String subStr = srcStr.substring(k, k + charCount);
+      String translated = dict.get(subStr);
+      if (null == translated) {
+        sb.append(subStr);
+      } else if (!"\0".equals(translated)) {
+        sb.append(translated);
       }
     }
     return fromString(sb.toString());

common/unsafe/src/test/java/org/apache/spark/unsafe/types/UTF8StringSuite.java

@@ -465,27 +465,27 @@ public void translate() {
     assertEquals(
       fromString("1a2s3ae"),
       fromString("translate").translate(ImmutableMap.of(
-        'r', '1',
-        'n', '2',
-        'l', '3',
-        't', '\0'
+        "r", "1",
+        "n", "2",
+        "l", "3",
+        "t", "\0"
       )));
     assertEquals(
       fromString("translate"),
       fromString("translate").translate(new HashMap<>()));
     assertEquals(
       fromString("asae"),
       fromString("translate").translate(ImmutableMap.of(
-        'r', '\0',
-        'n', '\0',
-        'l', '\0',
-        't', '\0'
+        "r", "\0",
+        "n", "\0",
+        "l", "\0",
+        "t", "\0"
       )));
     assertEquals(
       fromString("aa世b"),
       fromString("花花世界").translate(ImmutableMap.of(
-        '花', 'a',
-        '界', 'b'
+        "花", "a",
+        "界", "b"
       )));
   }
 

core/pom.xml

@@ -35,18 +35,13 @@
   </properties>
 
   <dependencies>
-    <dependency>
-      <groupId>com.thoughtworks.paranamer</groupId>
-      <artifactId>paranamer</artifactId>
-    </dependency>
     <dependency>
       <groupId>org.apache.avro</groupId>
       <artifactId>avro</artifactId>
     </dependency>
     <dependency>
       <groupId>org.apache.avro</groupId>
       <artifactId>avro-mapred</artifactId>
-      <classifier>${avro.mapred.classifier}</classifier>
     </dependency>
     <dependency>
       <groupId>com.google.guava</groupId>
@@ -171,6 +166,10 @@
       <artifactId>jakarta.servlet-api</artifactId>
       <version>${jakartaservlet.version}</version>
     </dependency>
+    <dependency>
+      <groupId>commons-codec</groupId>
+      <artifactId>commons-codec</artifactId>
+    </dependency>
     <dependency>
       <groupId>org.apache.commons</groupId>
       <artifactId>commons-lang3</artifactId>

core/src/main/java/org/apache/spark/shuffle/api/WritableByteChannelWrapper.java

@@ -24,7 +24,6 @@
 
 /**
  * :: Private ::
- *
  * A thin wrapper around a {@link WritableByteChannel}.
  * <p>
  * This is primarily provided for the local disk shuffle implementation to provide a

core/src/main/java/org/apache/spark/shuffle/api/metadata/MapOutputCommitMessage.java

@@ -23,7 +23,6 @@
 
 /**
  * :: Private ::
- *
  * Represents the result of writing map outputs for a shuffle map task.
  * <p>
  * Partition lengths represents the length of each block written in the map task. This can

core/src/main/java/org/apache/spark/shuffle/api/metadata/MapOutputMetadata.java

@@ -21,7 +21,6 @@
 
 /**
  * :: Private ::
- *
  * An opaque metadata tag for registering the result of committing the output of a
  * shuffle map task.
  * <p>

core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeInMemorySorter.java

@@ -20,7 +20,7 @@
 import java.util.Comparator;
 import java.util.LinkedList;
 
-import org.apache.avro.reflect.Nullable;
+import javax.annotation.Nullable;
 
 import org.apache.spark.TaskContext;
 import org.apache.spark.memory.MemoryConsumer;