added test cases for data source loader

Author: Joseph Batchik

.rat-excludes

@@ -93,4 +93,4 @@ INDEX
 .lintr
 gen-java.*
 .*avpr
-org.apache.spark.sql.sources.DataSourceProvider
+services/*

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

@@ -1,3 +1,3 @@
+org.apache.spark.sql.jdbc.DefaultSource
 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

@@ -202,7 +202,7 @@ private[sql] object ResolvedDataSource extends Logging {
     val sl = ServiceLoader.load(classOf[DataSourceProvider], loader)
 
     sl.iterator().filter(_.format() == provider).toList match {
-      case Nil => logInfo(s"provider: $provider is not registered in the service loader")
+      case Nil => logDebug(s"provider: $provider is not registered in the service loader")
       case head :: Nil => return head.getClass
       case sources => sys.error(s"Multiple sources found for $provider, " +
         s"(${sources.map(_.getClass.getName).mkString(", ")}, " +

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

@@ -0,0 +1,3 @@
+org.apache.spark.sql.sources.FakeSourceOne
+org.apache.spark.sql.sources.FakeSourceTwo
+org.apache.spark.sql.sources.FakeSourceThree

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

@@ -0,0 +1,75 @@
+/*
+* 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.sources
+
+import org.apache.spark.sql.SQLContext
+import org.apache.spark.sql.types.{StringType, StructField, StructType}
+
+class FakeSourceOne extends RelationProvider {
+
+  override def format() = "Fluet da Bomb"
+
+  override def createRelation(cont: SQLContext, param: Map[String, String]) = new BaseRelation {
+    override def sqlContext: SQLContext = cont
+
+    override def schema: StructType =
+      StructType(Seq(StructField("stringType", StringType, nullable = false)))
+  }
+}
+
+class FakeSourceTwo extends RelationProvider {
+
+  override def format() = "Fluet da Bomb"
+
+  override def createRelation(cont: SQLContext, param: Map[String, String]) = new BaseRelation {
+    override def sqlContext: SQLContext = cont
+
+    override def schema: StructType =
+      StructType(Seq(StructField("stringType", StringType, nullable = false)))
+  }
+}
+
+class FakeSourceThree extends RelationProvider {
+  override def format() = "gathering quorum"
+
+  override def createRelation(cont: SQLContext, param: Map[String, String]) = new BaseRelation {
+    override def sqlContext: SQLContext = cont
+
+    override def schema: StructType =
+      StructType(Seq(StructField("stringType", StringType, nullable = false)))
+  }
+}
+// please note that the META-INF/services had to be modified for the test directory for this to work
+class DDLSourceLoadSuite extends DataSourceTest {
+
+  test("data sources with the same name") {
+    intercept[RuntimeException] {
+      caseInsensitiveContext.read.format("Fluet da Bomb").load()
+    }
+  }
+
+  test("load data source from format alias") {
+    caseInsensitiveContext.read.format("gathering quorum").load().schema ==
+      StructType(Seq(StructField("stringType", StringType, nullable = false)))
+  }
+
+  test("specify full classname with duplicate formats") {
+    caseInsensitiveContext.read.format("org.apache.spark.sql.sources.FakeSourceOne")
+      .load().schema == StructType(Seq(StructField("stringType", StringType, nullable = false)))
+  }
+}