Merge branch 'master' into SPARK-29713

Author: yaooqinn

.github/workflows/master.yml

@@ -31,7 +31,7 @@ jobs:
       run: |
         export MAVEN_OPTS="-Xmx2g -XX:ReservedCodeCacheSize=1g -Dorg.slf4j.simpleLogger.defaultLogLevel=WARN"
         export MAVEN_CLI_OPTS="--no-transfer-progress"
-        ./build/mvn $MAVEN_CLI_OPTS -DskipTests -Pyarn -Pmesos -Pkubernetes -Phive -Phive-thriftserver -P${{ matrix.hadoop }} -Phadoop-cloud -Djava.version=${{ matrix.java }} package
+        ./build/mvn $MAVEN_CLI_OPTS -DskipTests -Pyarn -Pmesos -Pkubernetes -Phive -Phive-thriftserver -P${{ matrix.hadoop }} -Phadoop-cloud -Djava.version=${{ matrix.java }} install
 
 
   lint:

common/unsafe/src/main/java/org/apache/spark/sql/catalyst/util/DateTimeConstants.java

@@ -0,0 +1,57 @@
+/*
+ * 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.sql.catalyst.util;
+
+public class DateTimeConstants {
+
+  public static final int YEARS_PER_DECADE = 10;
+  public static final int YEARS_PER_CENTURY = 100;
+  public static final int YEARS_PER_MILLENNIUM = 1000;
+
+  public static final byte MONTHS_PER_QUARTER = 3;
+  public static final int MONTHS_PER_YEAR = 12;
+
+  public static final byte DAYS_PER_WEEK = 7;
+  public static final long DAYS_PER_MONTH = 30L;
+
+  public static final long HOURS_PER_DAY = 24L;
+
+  public static final long MINUTES_PER_HOUR = 60L;
+
+  public static final long SECONDS_PER_MINUTE = 60L;
+  public static final long SECONDS_PER_HOUR = MINUTES_PER_HOUR * SECONDS_PER_MINUTE;
+  public static final long SECONDS_PER_DAY = HOURS_PER_DAY * SECONDS_PER_HOUR;
+
+  public static final long MILLIS_PER_SECOND = 1000L;
+  public static final long MILLIS_PER_MINUTE = SECONDS_PER_MINUTE * MILLIS_PER_SECOND;
+  public static final long MILLIS_PER_HOUR = MINUTES_PER_HOUR * MILLIS_PER_MINUTE;
+  public static final long MILLIS_PER_DAY = HOURS_PER_DAY * MILLIS_PER_HOUR;
+
+  public static final long MICROS_PER_MILLIS = 1000L;
+  public static final long MICROS_PER_SECOND = MILLIS_PER_SECOND * MICROS_PER_MILLIS;
+  public static final long MICROS_PER_MINUTE = SECONDS_PER_MINUTE * MICROS_PER_SECOND;
+  public static final long MICROS_PER_HOUR = MINUTES_PER_HOUR * MICROS_PER_MINUTE;
+  public static final long MICROS_PER_DAY = HOURS_PER_DAY * MICROS_PER_HOUR;
+  public static final long MICROS_PER_MONTH = DAYS_PER_MONTH * MICROS_PER_DAY;
+  /* 365.25 days per year assumes leap year every four years */
+  public static final long MICROS_PER_YEAR = (36525L * MICROS_PER_DAY) / 100;
+
+  public static final long NANOS_PER_MICROS = 1000L;
+  public static final long NANOS_PER_MILLIS = MICROS_PER_MILLIS * NANOS_PER_MICROS;
+  public static final long NANOS_PER_SECOND = MILLIS_PER_SECOND * NANOS_PER_MILLIS;
+}

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

@@ -19,51 +19,27 @@
 
 import java.io.Serializable;
 import java.math.BigDecimal;
+import java.time.Duration;
+import java.time.Period;
+import java.time.temporal.ChronoUnit;
 import java.util.Objects;
 
+import static org.apache.spark.sql.catalyst.util.DateTimeConstants.*;
+
 /**
  * The internal representation of interval type.
  */
 public final class CalendarInterval implements Serializable {
-  public static final long MICROS_PER_MILLI = 1000L;
-  public static final long MICROS_PER_SECOND = MICROS_PER_MILLI * 1000;
-  public static final long MICROS_PER_MINUTE = MICROS_PER_SECOND * 60;
-  public static final long MICROS_PER_HOUR = MICROS_PER_MINUTE * 60;
-  public static final long MICROS_PER_DAY = MICROS_PER_HOUR * 24;
-  public static final long MICROS_PER_WEEK = MICROS_PER_DAY * 7;
-
   public final int months;
   public final int days;
   public final long microseconds;
 
-  public long milliseconds() {
-    return this.microseconds / MICROS_PER_MILLI;
-  }
-
   public CalendarInterval(int months, int days, long microseconds) {
     this.months = months;
     this.days = days;
     this.microseconds = microseconds;
   }
 
-  public CalendarInterval add(CalendarInterval that) {
-    int months = this.months + that.months;
-    int days = this.days + that.days;
-    long microseconds = this.microseconds + that.microseconds;
-    return new CalendarInterval(months, days, microseconds);
-  }
-
-  public CalendarInterval subtract(CalendarInterval that) {
-    int months = this.months - that.months;
-    int days = this.days - that.days;
-    long microseconds = this.microseconds - that.microseconds;
-    return new CalendarInterval(months, days, microseconds);
-  }
-
-  public CalendarInterval negate() {
-    return new CalendarInterval(-this.months, -this.days, -this.microseconds);
-  }
-
   @Override
   public boolean equals(Object o) {
     if (this == o) return true;
@@ -81,35 +57,53 @@ public int hashCode() {
 
   @Override
   public String toString() {
-    StringBuilder sb = new StringBuilder("interval");
+    if (months == 0 && days == 0 && microseconds == 0) {
+      return "0 seconds";
+    }
+
+    StringBuilder sb = new StringBuilder();
 
     if (months != 0) {
-      appendUnit(sb, months / 12, "year");
-      appendUnit(sb, months % 12, "month");
+      appendUnit(sb, months / 12, "years");
+      appendUnit(sb, months % 12, "months");
     }
 
-    appendUnit(sb, days, "day");
+    appendUnit(sb, days, "days");
 
     if (microseconds != 0) {
       long rest = microseconds;
-      appendUnit(sb, rest / MICROS_PER_HOUR, "hour");
+      appendUnit(sb, rest / MICROS_PER_HOUR, "hours");
       rest %= MICROS_PER_HOUR;
-      appendUnit(sb, rest / MICROS_PER_MINUTE, "minute");
+      appendUnit(sb, rest / MICROS_PER_MINUTE, "minutes");
       rest %= MICROS_PER_MINUTE;
       if (rest != 0) {
         String s = BigDecimal.valueOf(rest, 6).stripTrailingZeros().toPlainString();
-        sb.append(' ').append(s).append(" seconds");
+        sb.append(s).append(" seconds ");
       }
-    } else if (months == 0 && days == 0) {
-      sb.append(" 0 microseconds");
     }
 
+    sb.setLength(sb.length() - 1);
     return sb.toString();
   }
 
   private void appendUnit(StringBuilder sb, long value, String unit) {
     if (value != 0) {
-      sb.append(' ').append(value).append(' ').append(unit).append('s');
+      sb.append(value).append(' ').append(unit).append(' ');
     }
   }
+
+  /**
+   * Extracts the date part of the interval.
+   * @return an instance of {@code java.time.Period} based on the months and days fields
+   *         of the given interval, not null.
+   */
+  public Period extractAsPeriod() { return Period.of(0, months, days); }
+
+  /**
+   * Extracts the time part of the interval.
+   * @return an instance of {@code java.time.Duration} based on the microseconds field
+   *         of the given interval, not null.
+   * @throws ArithmeticException if a numeric overflow occurs
+   */
+  public Duration extractAsDuration() { return Duration.of(microseconds, ChronoUnit.MICROS); }
 }

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

@@ -370,7 +370,7 @@ private byte getByte(int i) {
     return Platform.getByte(base, offset + i);
   }
 
-  private boolean matchAt(final UTF8String s, int pos) {
+  public boolean matchAt(final UTF8String s, int pos) {
     if (s.numBytes + pos > numBytes || pos < 0) {
       return false;
     }

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

@@ -19,8 +19,11 @@
 
 import org.junit.Test;
 
+import java.time.Duration;
+import java.time.Period;
+
 import static org.junit.Assert.*;
-import static org.apache.spark.unsafe.types.CalendarInterval.*;
+import static org.apache.spark.sql.catalyst.util.DateTimeConstants.*;
 
 public class CalendarIntervalSuite {
 
@@ -48,50 +51,35 @@ public void toStringTest() {
     CalendarInterval i;
 
     i = new CalendarInterval(0, 0, 0);
-    assertEquals("interval 0 microseconds", i.toString());
+    assertEquals("0 seconds", i.toString());
 
     i = new CalendarInterval(34, 0, 0);
-    assertEquals("interval 2 years 10 months", i.toString());
+    assertEquals("2 years 10 months", i.toString());
 
     i = new CalendarInterval(-34, 0, 0);
-    assertEquals("interval -2 years -10 months", i.toString());
+    assertEquals("-2 years -10 months", i.toString());
 
     i = new CalendarInterval(0, 31, 0);
-    assertEquals("interval 31 days", i.toString());
+    assertEquals("31 days", i.toString());
 
     i = new CalendarInterval(0, -31, 0);
-    assertEquals("interval -31 days", i.toString());
+    assertEquals("-31 days", i.toString());
 
     i = new CalendarInterval(0, 0, 3 * MICROS_PER_HOUR + 13 * MICROS_PER_MINUTE + 123);
-    assertEquals("interval 3 hours 13 minutes 0.000123 seconds", i.toString());
+    assertEquals("3 hours 13 minutes 0.000123 seconds", i.toString());
 
     i = new CalendarInterval(0, 0, -3 * MICROS_PER_HOUR - 13 * MICROS_PER_MINUTE - 123);
-    assertEquals("interval -3 hours -13 minutes -0.000123 seconds", i.toString());
+    assertEquals("-3 hours -13 minutes -0.000123 seconds", i.toString());
 
     i = new CalendarInterval(34, 31, 3 * MICROS_PER_HOUR + 13 * MICROS_PER_MINUTE + 123);
-    assertEquals("interval 2 years 10 months 31 days 3 hours 13 minutes 0.000123 seconds",
+    assertEquals("2 years 10 months 31 days 3 hours 13 minutes 0.000123 seconds",
       i.toString());
   }
 
   @Test
-  public void addTest() {
-    CalendarInterval input1 = new CalendarInterval(3, 1, 1 * MICROS_PER_HOUR);
-    CalendarInterval input2 = new CalendarInterval(2, 4, 100 * MICROS_PER_HOUR);
-    assertEquals(new CalendarInterval(5, 5, 101 * MICROS_PER_HOUR), input1.add(input2));
-
-    input1 = new CalendarInterval(-10, -30, -81 * MICROS_PER_HOUR);
-    input2 = new CalendarInterval(75, 150, 200 * MICROS_PER_HOUR);
-    assertEquals(new CalendarInterval(65, 120, 119 * MICROS_PER_HOUR), input1.add(input2));
-  }
-
-  @Test
-  public void subtractTest() {
-    CalendarInterval input1 = new CalendarInterval(3, 1, 1 * MICROS_PER_HOUR);
-    CalendarInterval input2 = new CalendarInterval(2, 4, 100 * MICROS_PER_HOUR);
-    assertEquals(new CalendarInterval(1, -3, -99 * MICROS_PER_HOUR), input1.subtract(input2));
-
-    input1 = new CalendarInterval(-10, -30, -81 * MICROS_PER_HOUR);
-    input2 = new CalendarInterval(75, 150, 200 * MICROS_PER_HOUR);
-    assertEquals(new CalendarInterval(-85, -180, -281 * MICROS_PER_HOUR), input1.subtract(input2));
+  public void periodAndDurationTest() {
+    CalendarInterval interval = new CalendarInterval(120, -40, 123456);
+    assertEquals(Period.of(0, 120, -40), interval.extractAsPeriod());
+    assertEquals(Duration.ofNanos(123456000), interval.extractAsDuration());
   }
 }

core/src/main/java/org/apache/spark/api/plugin/DriverPlugin.java

@@ -0,0 +1,111 @@
+/*
+ * 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.api.plugin;
+
+import java.util.Collections;
+import java.util.Map;
+
+import org.apache.spark.SparkContext;
+import org.apache.spark.annotation.DeveloperApi;
+
+/**
+ * :: DeveloperApi ::
+ * Driver component of a {@link SparkPlugin}.
+ *
+ * @since 3.0.0
+ */
+@DeveloperApi
+public interface DriverPlugin {
+
+  /**
+   * Initialize the plugin.
+   * <p>
+   * This method is called early in the initialization of the Spark driver. Explicitly, it is
+   * called before the Spark driver's task scheduler is initialized. This means that a lot
+   * of other Spark subsystems may yet not have been initialized. This call also blocks driver
+   * initialization.
+   * <p>
+   * It's recommended that plugins be careful about what operations are performed in this call,
+   * preferrably performing expensive operations in a separate thread, or postponing them until
+   * the application has fully started.
+   *
+   * @param sc The SparkContext loading the plugin.
+   * @param pluginContext Additional plugin-specific about the Spark application where the plugin
+   *                      is running.
+   * @return A map that will be provided to the {@link ExecutorPlugin#init(PluginContext,Map)}
+   *         method.
+   */
+  default Map<String, String> init(SparkContext sc, PluginContext pluginContext) {
+    return Collections.emptyMap();
+  }
+
+  /**
+   * Register metrics published by the plugin with Spark's metrics system.
+   * <p>
+   * This method is called later in the initialization of the Spark application, after most
+   * subsystems are up and the application ID is known. If there are metrics registered in
+   * the registry ({@link PluginContext#metricRegistry()}), then a metrics source with the
+   * plugin name will be created.
+   * <p>
+   * Note that even though the metric registry is still accessible after this method is called,
+   * registering new metrics after this method is called may result in the metrics not being
+   * available.
+   *
+   * @param appId The application ID from the cluster manager.
+   * @param pluginContext Additional plugin-specific about the Spark application where the plugin
+   *                      is running.
+   */
+  default void registerMetrics(String appId, PluginContext pluginContext) {}
+
+  /**
+   * RPC message handler.
+   * <p>
+   * Plugins can use Spark's RPC system to send messages from executors to the driver (but not
+   * the other way around, currently). Messages sent by the executor component of the plugin will
+   * be delivered to this method, and the returned value will be sent back to the executor as
+   * the reply, if the executor has requested one.
+   * <p>
+   * Any exception thrown will be sent back to the executor as an error, in case it is expecting
+   * a reply. In case a reply is not expected, a log message will be written to the driver log.
+   * <p>
+   * The implementation of this handler should be thread-safe.
+   * <p>
+   * Note all plugins share RPC dispatch threads, and this method is called synchronously. So
+   * performing expensive operations in this handler may affect the operation of other active
+   * plugins. Internal Spark endpoints are not directly affected, though, since they use different
+   * threads.
+   * <p>
+   * Spark guarantees that the driver component will be ready to receive messages through this
+   * handler when executors are started.
+   *
+   * @param message The incoming message.
+   * @return Value to be returned to the caller. Ignored if the caller does not expect a reply.
+   */
+  default Object receive(Object message) throws Exception {
+    throw new UnsupportedOperationException();
+  }
+
+  /**
+   * Informs the plugin that the Spark application is shutting down.
+   * <p>
+   * This method is called during the driver shutdown phase. It is recommended that plugins
+   * not use any Spark functions (e.g. send RPC messages) during this call.
+   */
+  default void shutdown() {}
+
+}