changed the new trait to be used as a mixin for data source to register themselves

Author: Joseph Batchik

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

@@ -197,7 +197,7 @@ private[sql] class DDLParser(
 private[sql] object ResolvedDataSource extends Logging {
 
   private lazy val loader = Utils.getContextOrSparkClassLoader
-  private lazy val serviceLoader = ServiceLoader.load(classOf[DataSourceProvider], loader)
+  private lazy val serviceLoader = ServiceLoader.load(classOf[DataSourceRegister], loader)
 
   /** Tries to load the particular class */
   private def tryLoad(provider: String): Option[Class[_]] = try {
@@ -213,9 +213,8 @@ private[sql] object ResolvedDataSource extends Logging {
   /** Given a provider name, look up the data source class definition. */
   def lookupDataSource(provider: String): Class[_] = {
     serviceLoader.iterator().filter(_.format() == provider).toList match {
-      case Nil => tryLoad(provider).orElse(tryLoad(s"$provider.DefaultSource")).getOrElse {
-        sys.error(s"Failed to load class for data source: $provider")
-      }
+      case Nil => tryLoad(provider).orElse(tryLoad(s"$provider.DefaultSource"))
+        .getOrElse(sys.error(s"Failed to load class for data source: $provider"))
       case head :: Nil => head.getClass
       case sources => sys.error(s"Multiple sources found for $provider, " +
         s"(${sources.map(_.getClass.getName).mkString(", ")}), " +

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

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

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

@@ -32,9 +32,9 @@ import org.apache.spark.sql.{DataFrame, Row, SQLContext, SaveMode}
 private[sql] class DefaultSource
   extends RelationProvider
   with SchemaRelationProvider
-  with CreatableRelationProvider {
+  with CreatableRelationProvider with DataSourceRegister {
 
-  override def format(): String = "json"
+  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/json/JSONRelation.scala

@@ -50,9 +50,9 @@ import org.apache.spark.sql.types.{DataType, StructType}
 import org.apache.spark.util.{SerializableConfiguration, Utils}
 
 
-private[sql] class DefaultSource extends HadoopFsRelationProvider {
+private[sql] class DefaultSource extends HadoopFsRelationProvider with DataSourceRegister {
 
-  override def format(): String = "parquet"
+  def format(): String = "parquet"
 
   override def createRelation(
       sqlContext: SQLContext,

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

@@ -39,23 +39,23 @@ 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.
+ * Data sources should implement this trait so that they can register an alias to their data source.
+ * This allows users to give the data source alias as the format type over the fully qualified
+ * class name.
+ *
+ * ex: parquet.DefaultSource.format = "parquet".
  *
  * A new instance of this class with be instantiated each time a DDL call is made.
  */
 @DeveloperApi
-trait DataSourceProvider {
+trait DataSourceRegister {
 
   /**
-   * 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
+   * The string that represents the format that this data source provider uses. This is
    * overridden by children to provide a nice alias for the data source,
    * ex: override def format(): String = "parquet"
    */
-  def format(): String = getClass.getName
+  def format(): String
 }
 
 /**
@@ -74,7 +74,7 @@ trait DataSourceProvider {
  * @since 1.3.0
  */
 @DeveloperApi
-trait RelationProvider extends DataSourceProvider {
+trait RelationProvider {
   /**
    * Returns a new base relation with the given parameters.
    * Note: the parameters' keywords are case insensitive and this insensitivity is enforced
@@ -105,7 +105,7 @@ trait RelationProvider extends DataSourceProvider {
  * @since 1.3.0
  */
 @DeveloperApi
-trait SchemaRelationProvider extends DataSourceProvider {
+trait SchemaRelationProvider {
   /**
    * 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
@@ -141,7 +141,7 @@ trait SchemaRelationProvider extends DataSourceProvider {
  * @since 1.4.0
  */
 @Experimental
-trait HadoopFsRelationProvider extends DataSourceProvider {
+trait HadoopFsRelationProvider {
   /**
    * 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
@@ -161,7 +161,7 @@ trait HadoopFsRelationProvider extends DataSourceProvider {
  * @since 1.3.0
  */
 @DeveloperApi
-trait CreatableRelationProvider extends DataSourceProvider {
+trait CreatableRelationProvider {
   /**
     * 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/main/scala/org/apache/spark/sql/sources/interfaces.scala

@@ -20,9 +20,9 @@ package org.apache.spark.sql.sources
 import org.apache.spark.sql.SQLContext
 import org.apache.spark.sql.types.{StringType, StructField, StructType}
 
-class FakeSourceOne extends RelationProvider {
+class FakeSourceOne extends RelationProvider with DataSourceRegister {
 
-  override def format(): String = "Fluet da Bomb"
+  def format(): String = "Fluet da Bomb"
 
   override def createRelation(cont: SQLContext, param: Map[String, String]): BaseRelation =
     new BaseRelation {
@@ -33,9 +33,9 @@ class FakeSourceOne extends RelationProvider {
     }
 }
 
-class FakeSourceTwo extends RelationProvider {
+class FakeSourceTwo extends RelationProvider  with DataSourceRegister {
 
-  override def format(): String = "Fluet da Bomb"
+  def format(): String = "Fluet da Bomb"
 
   override def createRelation(cont: SQLContext, param: Map[String, String]): BaseRelation =
     new BaseRelation {
@@ -46,9 +46,9 @@ class FakeSourceTwo extends RelationProvider {
     }
 }
 
-class FakeSourceThree extends RelationProvider {
+class FakeSourceThree extends RelationProvider with DataSourceRegister {
 
-  override def format(): String = "gathering quorum"
+  def format(): String = "gathering quorum"
 
   override def createRelation(cont: SQLContext, param: Map[String, String]): BaseRelation =
     new BaseRelation {

sql/core/src/test/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceProvider renamed to sql/core/src/test/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister

@@ -26,8 +26,6 @@ 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/DDLSourceLoadSuite.scala

@@ -25,8 +25,6 @@ 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 = {