[SNAPPYDATA][MERGE-2.1] Missing patches and version changes

- updated optimized ByteArrayMethods.arrayEquals as per the code in Spark 2.1
  - adapt the word alignment code and optimize it a bit
  - in micro-benchmarks the new method is 30-60% faster than upstream version;
    at larger sizes it is 40-50% faster meaning its base word comparison loop itself is faster
- increase default locality time from 3s to 10s since the previous code to force
  executor-specific routing if it is alive has been removed
- added back cast removal optimization when types differ only in nullability
- add serialization and proper nanoTime handling from *CpuTime added in Spark 2.1.x;
  use DoubleAccumulator for these new fields like done for others to get more accurate results;
  also avoid the rare conditions where these cpu times could be negative
- cleanup handling of jobId and related new fields in Task with kryo serialization
- reverted change to AnalysisException with null check for plan since it is transient now
- reverted old Spark 2.0 code that was retained in InsertIntoTable and changed to Spark 2.1 code
- updated library versions and make them uniform as per upstream Spark for
  commons-lang3, metrics-core, py4j, breeze, univocity; also updated exclusions as
  per the changes to Spark side between 2.0.2 to 2.1.0
- added gradle build for the new mesos sub-project

assembly/build.gradle

@@ -31,6 +31,9 @@ dependencies {
   compile project(subprojectBase + 'snappy-spark-yarn_' + scalaBinaryVersion)
   compile project(subprojectBase + 'snappy-spark-mllib_' + scalaBinaryVersion)
   compile project(subprojectBase + 'snappy-spark-graphx_' + scalaBinaryVersion)
+  if (rootProject.hasProperty('mesos')) {
+    compile project(subprojectBase + 'snappy-spark-mesos_' + scalaBinaryVersion)
+  }
   if (rootProject.hasProperty('ganglia')) {
     compile project(subprojectBase + 'snappy-spark-ganglia-lgpl_' + scalaBinaryVersion)
   }

build.gradle

@@ -79,6 +79,8 @@ allprojects {
     seleniumVersion = '2.52.0'
     curatorVersion = '2.7.1'
     commonsCodecVersion = '1.10'
+    commonsLang3Version = '3.5'
+    commonsMath3Version = '3.4.1'
     avroVersion = '1.7.7'
     jsr305Version = '3.0.1'
     jlineVersion = '2.14.2'

common/network-common/build.gradle

@@ -24,7 +24,7 @@ dependencies {
   compile group: 'com.google.code.findbugs', name: 'jsr305', version: jsr305Version
   compile group: 'com.google.guava', name: 'guava', version: guavaVersion
   compile group: 'org.fusesource.leveldbjni', name: 'leveldbjni-all', version: '1.8'
-  compile group: 'com.fasterxml.jackson.core', name: 'jackson-databind', version: '2.6.5'
-  compile group: 'com.fasterxml.jackson.core', name: 'jackson-annotations', version: '2.8.5'
-  compile group: 'org.apache.commons', name: 'commons-lang3', version: '3.5'
+  compile group: 'com.fasterxml.jackson.core', name: 'jackson-databind', version: fasterXmlVersion
+  compile group: 'com.fasterxml.jackson.core', name: 'jackson-annotations', version: fasterXmlVersion
+  compile group: 'org.apache.commons', name: 'commons-lang3', version: commonsLang3Version
 }

common/network-shuffle/build.gradle

@@ -25,7 +25,7 @@ dependencies {
   compile group: 'com.fasterxml.jackson.core', name: 'jackson-databind', version: fasterXmlVersion
   compile group: 'com.fasterxml.jackson.core', name: 'jackson-annotations', version: fasterXmlVersion
   compile group: 'com.google.guava', name: 'guava', version: guavaVersion
-  compile group: 'io.dropwizard.metrics', name: 'metrics-core', version: '3.1.0'
+  compile group: 'io.dropwizard.metrics', name: 'metrics-core', version: metricsVersion
 
   testCompile project(path: subprojectBase + 'snappy-spark-network-common_' + scalaBinaryVersion, configuration: 'testOutput')
 }

common/unsafe/build.gradle

@@ -27,7 +27,7 @@ dependencies {
   compile group: 'com.google.code.findbugs', name: 'jsr305', version: jsr305Version
   compile group: 'com.google.guava', name: 'guava', version: guavaVersion
 
-  testCompile group: 'org.apache.commons', name: 'commons-lang3', version: '3.3.2'
+  testCompile group: 'org.apache.commons', name: 'commons-lang3', version: commonsLang3Version
 }
 
 // reset the srcDirs to allow javac compilation with specific args below

common/unsafe/src/main/java/org/apache/spark/unsafe/array/ByteArrayMethods.java

@@ -44,31 +44,58 @@ public static int roundNumberOfBytesToNearestWord(int numBytes) {
    * Optimized byte array equality check for byte arrays.
    * @return true if the arrays are equal, false otherwise
    */
-  public static boolean arrayEquals(
-      final Object leftBase, long leftOffset, final Object rightBase,
-      long rightOffset, final long length) {
-    long endOffset = leftOffset + length - 8;
-    while (leftOffset <= endOffset) {
-      if (Platform.getLong(leftBase, leftOffset) !=
-        Platform.getLong(rightBase, rightOffset)) {
-        return false;
+  public static boolean arrayEquals(final Object leftBase, long leftOffset,
+      final Object rightBase, long rightOffset, final long length) {
+    // try to align at least one side
+    if ((rightOffset & 0x7) != 0 && (leftOffset & 0x7) != 0) { // mod 8
+      final long endOffset = Math.min(((leftOffset + 7) >>> 3) << 3, leftOffset + length);
+      if (Platform.unaligned()) {
+        if (leftOffset <= (endOffset - 4)) {
+          if (Platform.getInt(leftBase, leftOffset) !=
+              Platform.getInt(rightBase, rightOffset)) {
+            return false;
+          }
+          leftOffset += 4;
+          rightOffset += 4;
+        }
+      }
+      while (leftOffset < endOffset) {
+        if (Platform.getByte(leftBase, leftOffset) !=
+            Platform.getByte(rightBase, rightOffset)) {
+          return false;
+        }
+        leftOffset++;
+        rightOffset++;
       }
-      leftOffset += 8;
-      rightOffset += 8;
     }
-    endOffset += 4;
-    while (leftOffset <= endOffset) {
-      if (Platform.getInt(leftBase, leftOffset) !=
-          Platform.getInt(rightBase, rightOffset)) {
-        return false;
+    long endOffset = leftOffset + length;
+    // for architectures that support unaligned accesses, chew it up 8 bytes at a time
+    if (Platform.unaligned() || (((leftOffset & 0x7) == 0) && ((rightOffset & 0x7) == 0))) {
+      endOffset -= 8;
+      while (leftOffset <= endOffset) {
+        if (Platform.getLong(leftBase, leftOffset) !=
+            Platform.getLong(rightBase, rightOffset)) {
+          return false;
+        }
+        leftOffset += 8;
+        rightOffset += 8;
+      }
+      endOffset += 4;
+      if (leftOffset <= endOffset) {
+        if (Platform.getInt(leftBase, leftOffset) !=
+            Platform.getInt(rightBase, rightOffset)) {
+          return false;
+        }
+        leftOffset += 4;
+        rightOffset += 4;
       }
-      leftOffset += 4;
-      rightOffset += 4;
+      endOffset += 4;
     }
-    endOffset += 4;
+    // this will finish off the unaligned comparisons, or do the entire aligned
+    // comparison whichever is needed.
     while (leftOffset < endOffset) {
       if (Platform.getByte(leftBase, leftOffset) !=
-        Platform.getByte(rightBase, rightOffset)) {
+          Platform.getByte(rightBase, rightOffset)) {
         return false;
       }
       leftOffset++;

core/build.gradle

@@ -99,10 +99,12 @@ dependencies {
   compile group: 'org.eclipse.jetty', name: 'jetty-client', version: jettyVersion
   compile group: 'org.eclipse.jetty', name: 'jetty-proxy', version: jettyVersion
   compile group: 'javax.servlet', name: 'javax.servlet-api', version: javaxServletVersion
-  compile group: 'org.apache.commons', name: 'commons-lang3', version: '3.3.2'
-  compile group: 'org.apache.commons', name: 'commons-math3', version: '3.4.1'
+  compile group: 'org.apache.commons', name: 'commons-lang3', version: commonsLang3Version
+  compile group: 'org.apache.commons', name: 'commons-math3', version: commonsMath3Version
   compile group: 'com.google.code.findbugs', name: 'jsr305', version: jsr305Version
-  compile group: 'org.apache.commons', name: 'commons-crypto', version: '1.0.0'
+  compile(group: 'org.apache.commons', name: 'commons-crypto', version: '1.0.0') {
+    exclude(group: 'net.java.dev.jna', module: 'jna')
+  }
   compile group: 'io.netty', name: 'netty', version: nettyVersion
   compile group: 'io.netty', name: 'netty-all', version: nettyAllVersion
   compile group: 'org.slf4j', name: 'jul-to-slf4j', version: slf4jVersion
@@ -137,7 +139,7 @@ dependencies {
   compile(group: 'net.razorvine', name: 'pyrolite', version: '4.13') {
     exclude(group: 'net.razorvine', module: 'serpent')
   }
-  compile group: 'net.sf.py4j', name: 'py4j', version: '0.10.1'
+  compile group: 'net.sf.py4j', name: 'py4j', version: '0.10.4'
 
   testCompile group: 'org.apache.avro', name: 'avro-ipc', version: avroVersion, classifier: 'tests'
   testCompile "org.apache.derby:derby:${derbyVersion}"

core/src/main/scala/org/apache/spark/executor/Executor.scala

@@ -325,13 +325,14 @@ private[spark] class Executor(
         // includes the Partition. Second, Task.run() deserializes the RDD and function to be run.
         task.metrics.setExecutorDeserializeTime(math.max(
           taskStart - deserializeStartTime + task.executorDeserializeTime, 0L) / 1000000.0)
-        task.metrics.setExecutorDeserializeCpuTime(
-          (taskStartCpu - deserializeStartCpuTime) + task.executorDeserializeCpuTime)
+        task.metrics.setExecutorDeserializeCpuTime(math.max(
+          taskStartCpu - deserializeStartCpuTime + task.executorDeserializeCpuTime, 0L) /
+            1000000.0)
         // We need to subtract Task.run()'s deserialization time to avoid double-counting
         task.metrics.setExecutorRunTime(math.max(
           taskFinish - taskStart - task.executorDeserializeTime, 0L) / 1000000.0)
-        task.metrics.setExecutorCpuTime(
-          (taskFinishCpu - taskStartCpu) - task.executorDeserializeCpuTime)
+        task.metrics.setExecutorCpuTime(math.max(
+          taskFinishCpu - taskStartCpu - task.executorDeserializeCpuTime, 0L) / 1000000.0)
         task.metrics.setJvmGCTime(computeTotalGcTime() - startGCTime)
         // Now resultSerializationTime is evaluated directly inside the
         // serialization write methods and added to final serialized bytes
@@ -399,6 +400,11 @@ private[spark] class Executor(
             if (task != null) {
               task.metrics.setExecutorRunTime(
                 math.max(System.nanoTime() - taskStart, 0L) / 1000000.0)
+              val taskEndCpu = if (threadMXBean.isCurrentThreadCpuTimeSupported) {
+                threadMXBean.getCurrentThreadCpuTime
+              } else 0L
+              task.metrics.setExecutorCpuTime(
+                math.max(taskEndCpu - taskStartCpu, 0L) / 1000000.0)
               task.metrics.setJvmGCTime(computeTotalGcTime() - startGCTime)
               task.collectAccumulatorUpdates(taskFailed = true)
             } else {

core/src/main/scala/org/apache/spark/executor/TaskMetrics.scala

@@ -48,9 +48,9 @@ import org.apache.spark.util._
 class TaskMetrics private[spark] () extends Serializable with KryoSerializable {
   // Each metric is internally represented as an accumulator
   private val _executorDeserializeTime = new DoubleAccumulator
-  private val _executorDeserializeCpuTime = new LongAccumulator
+  private val _executorDeserializeCpuTime = new DoubleAccumulator
   private val _executorRunTime = new DoubleAccumulator
-  private val _executorCpuTime = new LongAccumulator
+  private val _executorCpuTime = new DoubleAccumulator
   private val _resultSize = new LongAccumulator
   private val _jvmGCTime = new LongAccumulator
   private val _resultSerializationTime = new DoubleAccumulator
@@ -67,7 +67,7 @@ class TaskMetrics private[spark] () extends Serializable with KryoSerializable {
   /**
    * CPU Time taken on the executor to deserialize this task in nanoseconds.
    */
-  def executorDeserializeCpuTime: Long = _executorDeserializeCpuTime.sum
+  def executorDeserializeCpuTime: Long = _executorDeserializeCpuTime.sum.toLong
 
   /**
    * Time the executor spends actually running the task (including fetching shuffle data).
@@ -78,7 +78,7 @@ class TaskMetrics private[spark] () extends Serializable with KryoSerializable {
    * CPU Time the executor spends actually running the task
    * (including fetching shuffle data) in nanoseconds.
    */
-  def executorCpuTime: Long = _executorCpuTime.sum
+  def executorCpuTime: Long = _executorCpuTime.sum.toLong
 
   /**
    * The number of bytes this task transmitted back to the driver as the TaskResult.
@@ -125,10 +125,10 @@ class TaskMetrics private[spark] () extends Serializable with KryoSerializable {
   // Setters and increment-ers
   private[spark] def setExecutorDeserializeTime(v: Double): Unit =
     _executorDeserializeTime.setValue(v)
-  private[spark] def setExecutorDeserializeCpuTime(v: Long): Unit =
+  private[spark] def setExecutorDeserializeCpuTime(v: Double): Unit =
     _executorDeserializeCpuTime.setValue(v)
   private[spark] def setExecutorRunTime(v: Double): Unit = _executorRunTime.setValue(v)
-  private[spark] def setExecutorCpuTime(v: Long): Unit = _executorCpuTime.setValue(v)
+  private[spark] def setExecutorCpuTime(v: Double): Unit = _executorCpuTime.setValue(v)
   private[spark] def setResultSize(v: Long): Unit = _resultSize.setValue(v)
   private[spark] def setJvmGCTime(v: Long): Unit = _jvmGCTime.setValue(v)
   private[spark] def setResultSerializationTime(v: Double): Unit =

core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala

@@ -770,7 +770,8 @@ class DAGScheduler(
   // That should take care of at least part of the priority inversion problem with
   // cross-job dependencies.
   private def activeJobForStage(stage: Stage): Option[Int] = {
-    val jobsThatUseStage: Array[Int] = stage.jobIds.toArray.sorted
+    val jobsThatUseStage: Array[Int] = stage.jobIds.toArray
+    java.util.Arrays.sort(jobsThatUseStage)
     jobsThatUseStage.find(jobIdToActiveJob.contains)
   }
 
@@ -1031,7 +1032,7 @@ class DAGScheduler(
             val part = stage.rdd.partitions(id)
             new ShuffleMapTask(stage.id, stage.latestInfo.attemptId, taskData,
               taskBinary, part, locs, stage.latestInfo.taskMetrics, properties,
-              Option(jobId), Option(sc.applicationId), Option(sc.applicationId))
+              jobId, Option(sc.applicationId), Option(sc.applicationId))
           }
 
         case stage: ResultStage =>
@@ -1041,7 +1042,7 @@ class DAGScheduler(
             val locs = taskIdToLocations(id)
             new ResultTask(stage.id, stage.latestInfo.attemptId, taskData,
               taskBinary, part, locs, id, properties, stage.latestInfo.taskMetrics,
-              Option(jobId), Option(sc.applicationId), Option(sc.applicationId))
+              jobId, Option(sc.applicationId), Option(sc.applicationId))
           }
       }
     } catch {

core/src/main/scala/org/apache/spark/scheduler/ResultTask.scala

@@ -64,7 +64,7 @@ private[spark] class ResultTask[T, U](
     private var _outputId: Int,
     localProperties: Properties,
     metrics: TaskMetrics,
-    jobId: Option[Int] = None,
+    jobId: Int = -1,
     appId: Option[String] = None,
     appAttemptId: Option[String] = None)
   extends Task[U](stageId, stageAttemptId, partition.index, _taskData,
@@ -89,7 +89,7 @@ private[spark] class ResultTask[T, U](
       ByteBuffer.wrap(getTaskBytes), Thread.currentThread.getContextClassLoader)
     _executorDeserializeTime = math.max(System.nanoTime() - deserializeStartTime, 0L)
     _executorDeserializeCpuTime = if (threadMXBean.isCurrentThreadCpuTimeSupported) {
-      threadMXBean.getCurrentThreadCpuTime - deserializeStartCpuTime
+      math.max(threadMXBean.getCurrentThreadCpuTime - deserializeStartCpuTime, 0L)
     } else 0L
 
     func(context, rdd.iterator(partition, context))

core/src/main/scala/org/apache/spark/scheduler/ShuffleMapTask.scala

@@ -64,12 +64,12 @@ private[spark] class ShuffleMapTask(
     @transient private var locs: Seq[TaskLocation],
     metrics: TaskMetrics,
     localProperties: Properties,
-    jobId: Option[Int] = None,
+    jobId: Int = -1,
     appId: Option[String] = None,
     appAttemptId: Option[String] = None)
   extends Task[MapStatus](stageId, stageAttemptId, partition.index, _taskData,
     _taskBinary, metrics, localProperties, jobId, appId, appAttemptId)
-with KryoSerializable with Logging {
+  with KryoSerializable with Logging {
 
   /** A constructor used only in test suites. This does not require passing in an RDD. */
   def this(partitionId: Int) {
@@ -93,7 +93,7 @@ with KryoSerializable with Logging {
       ByteBuffer.wrap(getTaskBytes), Thread.currentThread.getContextClassLoader)
     _executorDeserializeTime = math.max(System.nanoTime() - deserializeStartTime, 0L)
     _executorDeserializeCpuTime = if (threadMXBean.isCurrentThreadCpuTimeSupported) {
-      threadMXBean.getCurrentThreadCpuTime - deserializeStartCpuTime
+      math.max(threadMXBean.getCurrentThreadCpuTime - deserializeStartCpuTime, 0L)
     } else 0L
 
     var writer: ShuffleWriter[Any, Any] = null

core/src/main/scala/org/apache/spark/scheduler/Task.scala

@@ -65,7 +65,7 @@ private[spark] abstract class Task[T](
     protected var taskBinary: Option[Broadcast[Array[Byte]]] = None,
     private var _metrics: TaskMetrics = TaskMetrics.registered,
     @transient var localProperties: Properties = new Properties,
-    private var _jobId: Option[Int] = None,
+    private var _jobId: Int = -1,
     private var _appId: Option[String] = None,
     private var _appAttemptId: Option[String] = None) extends Serializable {
 
@@ -75,13 +75,13 @@ private[spark] abstract class Task[T](
 
   final def partitionId: Int = _partitionId
 
-  final def metrics: TaskMetrics = _metrics
+  final def jobId: Int = _jobId
 
-  final def jobId: Int = _jobId.get
+  final def metrics: TaskMetrics = _metrics
 
-  final def appId: String = _appId.get
+  final def appId: String = if (_appId.isDefined) _appId.get else null
 
-  final def appAttemptId: String = _appAttemptId.get
+  final def appAttemptId: String = if (_appAttemptId.isDefined) _appAttemptId.get else null
 
   @transient private[spark] var taskDataBytes: Array[Byte] = _
 
@@ -118,8 +118,8 @@ private[spark] abstract class Task[T](
       kill(interruptThread = false)
     }
 
-    new CallerContext("TASK", _appId, _appAttemptId, _jobId, Option(stageId),
-      Option(stageAttemptId), Option(taskAttemptId), Option(attemptNumber))
+    new CallerContext("TASK", _appId, _appAttemptId, Some(jobId), Some(stageId),
+      Some(stageAttemptId), Some(taskAttemptId), Some(attemptNumber))
       .setCurrentContext()
 
     try {
@@ -231,8 +231,10 @@ private[spark] abstract class Task[T](
     output.writeInt(_stageId)
     output.writeVarInt(_stageAttemptId, true)
     output.writeVarInt(_partitionId, true)
+    output.writeVarInt(_jobId, true)
     output.writeLong(epoch)
     output.writeLong(_executorDeserializeTime)
+    output.writeLong(_executorDeserializeCpuTime)
     if ((taskData ne null) && taskData.uncompressedLen > 0) {
       // actual bytes will be shipped in TaskDescription
       output.writeBoolean(true)
@@ -241,17 +243,18 @@ private[spark] abstract class Task[T](
       kryo.writeClassAndObject(output, taskBinary.get)
     }
     _metrics.write(kryo, output)
-    output.writeInt(_jobId.get)
-    output.writeString(_appId.get)
-    output.writeString(_appAttemptId.get)
+    output.writeString(appId)
+    output.writeString(appAttemptId)
   }
 
   def readKryo(kryo: Kryo, input: Input): Unit = {
     _stageId = input.readInt()
     _stageAttemptId = input.readVarInt(true)
     _partitionId = input.readVarInt(true)
+    _jobId = input.readVarInt(true)
     epoch = input.readLong()
     _executorDeserializeTime = input.readLong()
+    _executorDeserializeCpuTime = input.readLong()
     // actual bytes are shipped in TaskDescription
     taskData = TaskData.EMPTY
     if (input.readBoolean()) {
@@ -262,9 +265,8 @@ private[spark] abstract class Task[T](
     }
     _metrics = new TaskMetrics
     _metrics.read(kryo, input)
-    _jobId = Some(input.readInt())
-    _appId = Some(input.readString())
-    _appAttemptId = Some(input.readString())
+    _appId = Option(input.readString())
+    _appAttemptId = Option(input.readString())
   }
 }
 

core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala

@@ -937,7 +937,7 @@ private[spark] class TaskSetManager(
   }
 
   private def getLocalityWait(level: TaskLocality.TaskLocality): Long = {
-    val defaultWait = conf.get("spark.locality.wait", "3s")
+    val defaultWait = conf.get("spark.locality.wait", "10s")
     val localityWaitKey = level match {
       case TaskLocality.PROCESS_LOCAL => "spark.locality.wait.process"
       case TaskLocality.NODE_LOCAL => "spark.locality.wait.node"

examples/build.gradle

@@ -26,7 +26,7 @@ dependencies {
   compile project(subprojectBase + 'snappy-spark-streaming-flume_' + scalaBinaryVersion)
   compile project(subprojectBase + 'snappy-spark-streaming-kafka-0.8_' + scalaBinaryVersion)
 
-  compile group: 'org.apache.commons', name: 'commons-math3', version: '3.4.1'
+  compile group: 'org.apache.commons', name: 'commons-math3', version: commonsMath3Version
   compile group: 'com.github.scopt', name: 'scopt_' + scalaBinaryVersion, version: '3.3.0'
   compile group: 'com.twitter', name: 'parquet-hadoop-bundle', version: hiveParquetVersion