started working on service loader

Author: Joseph Batchik

sql/core/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceProvider

@@ -0,0 +1,3 @@
+org.apache.spark.sql.json.DefaultSource
+org.apache.spark.sql.parquet.DefaultSource
+org.apache.spark.sql.jdbc.DefaultSource

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ddl.scala

@@ -17,6 +17,8 @@
 
 package org.apache.spark.sql.execution.datasources
 
+import java.util.ServiceLoader
+
 import scala.language.{existentials, implicitConversions}
 import scala.util.matching.Regex
 
@@ -190,23 +192,24 @@ private[sql] class DDLParser(
     }
 }
 
-private[sql] object ResolvedDataSource {
-
-  private val builtinSources = Map(
-    "jdbc" -> "org.apache.spark.sql.jdbc.DefaultSource",
-    "json" -> "org.apache.spark.sql.json.DefaultSource",
-    "parquet" -> "org.apache.spark.sql.parquet.DefaultSource",
-    "orc" -> "org.apache.spark.sql.hive.orc.DefaultSource"
-  )
+private[sql] object ResolvedDataSource extends Logging {
 
   /** Given a provider name, look up the data source class definition. */
   def lookupDataSource(provider: String): Class[_] = {
     val loader = Utils.getContextOrSparkClassLoader
 
-    if (builtinSources.contains(provider)) {
-      return loader.loadClass(builtinSources(provider))
+    val sl = ServiceLoader.load(classOf[DataSourceProvider], loader)
+
+    val itr = sl.iterator()
+    while (itr.hasNext) {
+      val service = itr.next()
+      if (service.format() == provider) {
+        return service.getClass
+      }
     }
 
+    logInfo(s"could not find registered data source for $provider")
+
     try {
       loader.loadClass(provider)
     } catch {

sql/core/src/main/scala/org/apache/spark/sql/jdbc/JDBCRelation.scala

@@ -78,6 +78,9 @@ private[sql] object JDBCRelation {
 }
 
 private[sql] class DefaultSource extends RelationProvider {
+
+  override def format(): String = "jdbc"
+
   /** Returns a new base relation with the given parameters. */
   override def createRelation(
       sqlContext: SQLContext,

sql/core/src/main/scala/org/apache/spark/sql/json/JSONRelation.scala

@@ -34,6 +34,8 @@ private[sql] class DefaultSource
   with SchemaRelationProvider
   with CreatableRelationProvider {
 
+  override def format(): String = "json"
+
   private def checkPath(parameters: Map[String, String]): String = {
     parameters.getOrElse("path", sys.error("'path' must be specified for json data."))
   }

sql/core/src/main/scala/org/apache/spark/sql/parquet/ParquetRelation.scala

@@ -51,6 +51,9 @@ import org.apache.spark.util.{SerializableConfiguration, Utils}
 
 
 private[sql] class DefaultSource extends HadoopFsRelationProvider {
+
+  override def format(): String = "parquet"
+
   override def createRelation(
       sqlContext: SQLContext,
       paths: Array[String],

sql/core/src/main/scala/org/apache/spark/sql/sources/interfaces.scala

@@ -37,6 +37,27 @@ import org.apache.spark.sql.types.StructType
 import org.apache.spark.sql._
 import org.apache.spark.util.SerializableConfiguration
 
+/**
+ * ::DeveloperApi::
+ * Base trait for all relation providers. All relation providers need to provide a string
+ * representing the format they load. ex: parquet.DefaultSource.format = "parquet".
+ * This allows users to specify that string as the format to read / write instead of providing
+ * the entire class.
+ *
+ * A new instance of this class with be instantiated each time a DDL call is made.
+ */
+@DeveloperApi
+trait DataSourceProvider {
+
+  /**
+   * The string that represents the format that this data source provider uses. By default, it is
+   * the name of the class, ex: "org.apache.spark.sql.parquet.DefaultSource". This should be
+   * overridden by children to provide a nice alias for the data source,
+   * ex: override def format(): String = "parquet"
+   */
+  def format(): String = getClass.getName
+}
+
 /**
  * ::DeveloperApi::
  * Implemented by objects that produce relations for a specific kind of data source.  When
@@ -53,7 +74,7 @@ import org.apache.spark.util.SerializableConfiguration
  * @since 1.3.0
  */
 @DeveloperApi
-trait RelationProvider {
+trait RelationProvider extends DataSourceProvider {
   /**
    * Returns a new base relation with the given parameters.
    * Note: the parameters' keywords are case insensitive and this insensitivity is enforced
@@ -84,7 +105,7 @@ trait RelationProvider {
  * @since 1.3.0
  */
 @DeveloperApi
-trait SchemaRelationProvider {
+trait SchemaRelationProvider extends DataSourceProvider {
   /**
    * Returns a new base relation with the given parameters and user defined schema.
    * Note: the parameters' keywords are case insensitive and this insensitivity is enforced
@@ -120,7 +141,7 @@ trait SchemaRelationProvider {
  * @since 1.4.0
  */
 @Experimental
-trait HadoopFsRelationProvider {
+trait HadoopFsRelationProvider extends DataSourceProvider {
   /**
    * Returns a new base relation with the given parameters, a user defined schema, and a list of
    * partition columns. Note: the parameters' keywords are case insensitive and this insensitivity
@@ -140,7 +161,7 @@ trait HadoopFsRelationProvider {
  * @since 1.3.0
  */
 @DeveloperApi
-trait CreatableRelationProvider {
+trait CreatableRelationProvider extends DataSourceProvider {
   /**
     * Creates a relation with the given parameters based on the contents of the given
     * DataFrame. The mode specifies the expected behavior of createRelation when

sql/core/src/test/scala/org/apache/spark/sql/sources/FilteredScanSuite.scala

@@ -25,6 +25,9 @@ import org.apache.spark.sql.types._
 
 
 class FilteredScanSource extends RelationProvider {
+
+  override def format(): String = "test format"
+
   override def createRelation(
       sqlContext: SQLContext,
       parameters: Map[String, String]): BaseRelation = {

sql/core/src/test/scala/org/apache/spark/sql/sources/PrunedScanSuite.scala

@@ -24,6 +24,9 @@ import org.apache.spark.sql._
 import org.apache.spark.sql.types._
 
 class PrunedScanSource extends RelationProvider {
+
+  override def format(): String = "test format"
+
   override def createRelation(
       sqlContext: SQLContext,
       parameters: Map[String, String]): BaseRelation = {

sql/core/src/test/scala/org/apache/spark/sql/sources/TableScanSuite.scala

@@ -29,6 +29,9 @@ import org.apache.spark.unsafe.types.UTF8String
 class DefaultSource extends SimpleScanSource
 
 class SimpleScanSource extends RelationProvider {
+
+  override def format(): String = "test format"
+
   override def createRelation(
       sqlContext: SQLContext,
       parameters: Map[String, String]): BaseRelation = {
@@ -46,6 +49,9 @@ case class SimpleScan(from: Int, to: Int)(@transient val sqlContext: SQLContext)
 }
 
 class AllDataTypesScanSource extends SchemaRelationProvider {
+
+  override def format(): String = "test format"
+
   override def createRelation(
       sqlContext: SQLContext,
       parameters: Map[String, String],

sql/hive/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceProvider

@@ -0,0 +1 @@
+org.apache.spark.sql.hive.orc.DefaultSource