put user defined repo before default repo

Author: windpiger

core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala

@@ -1031,10 +1031,8 @@ private[spark] object SparkSubmitUtils {
       val cr = new ChainResolver
       cr.setName("user-list")
 
-      // add current default resolver, if any
-      Option(ivySettings.getDefaultResolver).foreach(cr.add)
-
-      // add additional repositories, last resolution in chain takes precedence
+      // before default resolvers, add additional repositories,
+      // last resolution in chain takes precedence
       repositoryList.zipWithIndex.foreach { case (repo, i) =>
         val brr: IBiblioResolver = new IBiblioResolver
         brr.setM2compatible(true)
@@ -1047,6 +1045,9 @@ private[spark] object SparkSubmitUtils {
         // scalastyle:on println
       }
 
+      // add current default resolver, if any
+      Option(ivySettings.getDefaultResolver).foreach(cr.add)
+
       ivySettings.addResolver(cr)
       ivySettings.setDefaultResolver(cr.getName)
     }

core/src/test/scala/org/apache/spark/deploy/IvyTestUtils.scala

@@ -31,7 +31,7 @@ import org.apache.ivy.core.settings.IvySettings
 import org.apache.spark.TestUtils.{createCompiledClass, JavaSourceFromString}
 import org.apache.spark.deploy.SparkSubmitUtils.MavenCoordinate
 
-private[deploy] object IvyTestUtils {
+object IvyTestUtils {
 
   /**
    * Create the path for the jar and pom from the maven coordinate. Extension should be `jar`
@@ -355,7 +355,7 @@ private[deploy] object IvyTestUtils {
    * @param withPython Whether to pack python files inside the jar for extensive testing.
    * @return Root path of the repository. Will be `rootDir` if supplied.
    */
-  private[deploy] def withRepository(
+  def withRepository(
       artifact: MavenCoordinate,
       dependencies: Option[String],
       rootDir: Option[File],

core/src/test/scala/org/apache/spark/deploy/SparkSubmitUtilsSuite.scala

@@ -17,7 +17,7 @@
 
 package org.apache.spark.deploy
 
-import java.io.{File, OutputStream, PrintStream}
+import java.io.{File, FileInputStream, OutputStream, PrintStream}
 import java.nio.charset.StandardCharsets
 
 import scala.collection.mutable.ArrayBuffer
@@ -30,6 +30,7 @@ import org.scalatest.BeforeAndAfterAll
 
 import org.apache.spark.SparkFunSuite
 import org.apache.spark.deploy.SparkSubmitUtils.MavenCoordinate
+import org.apache.spark.unsafe.types.UTF8String
 import org.apache.spark.util.Utils
 
 class SparkSubmitUtilsSuite extends SparkFunSuite with BeforeAndAfterAll {
@@ -142,12 +143,13 @@ class SparkSubmitUtilsSuite extends SparkFunSuite with BeforeAndAfterAll {
   test("search for artifact at local repositories") {
     val main = new MavenCoordinate("my.great.lib", "mylib", "0.1")
     val dep = "my.great.dep:mydep:0.5"
-    // Local M2 repository
+
+    // Local M2 repositorya
     IvyTestUtils.withRepository(main, Some(dep), Some(SparkSubmitUtils.m2Path)) { repo =>
       val jarPath = SparkSubmitUtils.resolveMavenCoordinates(
         main.toString,
         SparkSubmitUtils.buildIvySettings(None, None),
-        isTest = true)
+        isTest = false)
       assert(jarPath.indexOf("mylib") >= 0, "should find artifact")
       assert(jarPath.indexOf("mydep") >= 0, "should find dependency")
     }
@@ -258,4 +260,53 @@ class SparkSubmitUtilsSuite extends SparkFunSuite with BeforeAndAfterAll {
       assert(jarPath.indexOf("mydep") >= 0, "should find dependency")
     }
   }
+  
+  test("search for artifact taking order from user defined repositories to default repositories") {
+    val main = new MavenCoordinate("a", "b", "0.1")
+
+    def isSameFile(left: String, right: String): Boolean = {
+      val leftInput: FileInputStream = new FileInputStream(left)
+      val leftMd5 = UTF8String.fromString(org.apache.commons.codec
+        .digest.DigestUtils.md5Hex(leftInput))
+
+      val rightInput: FileInputStream = new FileInputStream(left)
+      val rightMd5 = UTF8String.fromString(org.apache.commons.codec
+        .digest.DigestUtils.md5Hex(rightInput))
+
+      leftMd5 == rightMd5
+    }
+
+    var userDefinedRepo = Utils.createTempDir("my_m2")
+    try {
+      IvyTestUtils.withRepository(main, None, Some(userDefinedRepo)) { repo =>
+        IvyTestUtils.withRepository(main, None, Some(SparkSubmitUtils.m2Path)) {
+          defaultRepo =>
+            val jarPath = SparkSubmitUtils.resolveMavenCoordinates(
+              main.toString,
+              SparkSubmitUtils.buildIvySettings(Option(repo), None),
+              isTest = false)
+            assert(isSameFile(Seq(userDefinedRepo, main.groupId, main.artifactId, main.version,
+              "b-0.1.jar").mkString(File.separatorChar.toString), jarPath))
+            assert(jarPath.indexOf("b") >= 0, "should find artifact")
+
+        }
+      }
+
+      IvyTestUtils.withRepository(main, None, Some(SparkSubmitUtils.m2Path)) { defaultRepo =>
+        IvyTestUtils.withRepository(main, None, Some(userDefinedRepo)) {
+          repo =>
+            val jarPath = SparkSubmitUtils.resolveMavenCoordinates(
+              main.toString,
+              SparkSubmitUtils.buildIvySettings(Option(repo), None),
+              isTest = false)
+            assert(isSameFile(Seq(SparkSubmitUtils.m2Path.getCanonicalPath, main.groupId,
+              main.artifactId, main.version, "b-0.1.jar").mkString(File.separatorChar.toString),
+              jarPath))
+            assert(jarPath.indexOf("b") >= 0, "should find artifact")
+        }
+      }
+    } finally {
+      Utils.deleteRecursively(userDefinedRepo)
+    }
+  }
 }

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveUtils.scala

@@ -328,6 +328,7 @@ private[spark] object HiveUtils extends Logging {
         sparkConf = conf,
         hadoopConf = hadoopConf,
         config = configurations,
+        ivyPath = conf.getOption("spark.jars.ivy"),
         barrierPrefixes = hiveMetastoreBarrierPrefixes,
         sharedPrefixes = hiveMetastoreSharedPrefixes)
     } else {

sql/hive/src/main/scala/org/apache/spark/sql/hive/client/IsolatedClientLoader.scala

@@ -136,7 +136,7 @@ private[hive] object IsolatedClientLoader extends Logging {
 
   // A map from a given pair of HiveVersion and Hadoop version to jar files.
   // It is only used by forVersion.
-  private val resolvedVersions =
+  private[hive] val resolvedVersions =
     new scala.collection.mutable.HashMap[(HiveVersion, String), Seq[URL]]
 }
 

sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveUtilsSuite.scala

@@ -17,11 +17,15 @@
 
 package org.apache.spark.sql.hive
 
+import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.hive.conf.HiveConf.ConfVars
-
+import org.apache.spark.SparkConf
+import org.apache.spark.deploy.IvyTestUtils
+import org.apache.spark.deploy.SparkSubmitUtils.MavenCoordinate
 import org.apache.spark.sql.hive.test.TestHiveSingleton
 import org.apache.spark.sql.test.SQLTestUtils
 import org.apache.spark.sql.QueryTest
+import org.apache.spark.sql.hive.client.IsolatedClientLoader
 
 class HiveUtilsSuite extends QueryTest with SQLTestUtils with TestHiveSingleton {
 

sql/hive/src/test/scala/org/apache/spark/sql/hive/client/HiveClientSuite.scala

@@ -19,8 +19,9 @@ package org.apache.spark.sql.hive.client
 
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.hive.conf.HiveConf
-
 import org.apache.spark.SparkFunSuite
+import org.apache.spark.deploy.IvyTestUtils
+import org.apache.spark.deploy.SparkSubmitUtils.MavenCoordinate
 import org.apache.spark.sql.catalyst.catalog._
 import org.apache.spark.sql.catalyst.expressions.{AttributeReference, EqualTo, Literal}
 import org.apache.spark.sql.hive.HiveUtils