address comments

Author: scwf

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

@@ -209,12 +209,15 @@ private[sql] case class CreateTableUsing(
             sys.error(s"Failed to load class for data source: $provider")
         }
     }
-    val dataSource =
-      clazz.newInstance().asInstanceOf[org.apache.spark.sql.sources.SchemaRelationProvider]
     val relation = if(tableCols.isEmpty) {
-      dataSource.createRelation(
-        sqlContext, new CaseInsensitiveMap(options))
+      val dataSource =
+        clazz.newInstance().asInstanceOf[org.apache.spark.sql.sources.RelationProvider]
+
+      dataSource.createRelation(sqlContext, new CaseInsensitiveMap(options))
     } else {
+      val dataSource =
+        clazz.newInstance().asInstanceOf[org.apache.spark.sql.sources.SchemaRelationProvider]
+
       dataSource.createRelation(
         sqlContext, new CaseInsensitiveMap(options), Some(StructType(tableCols)))
     }

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

@@ -21,11 +21,10 @@ import scala.language.existentials
 
 import org.apache.spark.sql._
 
-class FilteredScanSource extends SchemaRelationProvider {
+class FilteredScanSource extends RelationProvider {
   override def createRelation(
       sqlContext: SQLContext,
-      parameters: Map[String, String],
-      schema: Option[StructType] = None): BaseRelation = {
+      parameters: Map[String, String]): BaseRelation = {
     SimpleFilteredScan(parameters("from").toInt, parameters("to").toInt)(sqlContext)
   }
 }

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

@@ -19,11 +19,10 @@ package org.apache.spark.sql.sources
 
 import org.apache.spark.sql._
 
-class PrunedScanSource extends SchemaRelationProvider {
+class PrunedScanSource extends RelationProvider {
   override def createRelation(
       sqlContext: SQLContext,
-      parameters: Map[String, String],
-      schema: Option[StructType] = None): BaseRelation = {
+      parameters: Map[String, String]): BaseRelation = {
     SimplePrunedScan(parameters("from").toInt, parameters("to").toInt)(sqlContext)
   }
 }

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

@@ -21,11 +21,10 @@ import org.apache.spark.sql._
 
 class DefaultSource extends SimpleScanSource
 
-class SimpleScanSource extends SchemaRelationProvider {
+class SimpleScanSource extends RelationProvider {
   override def createRelation(
       sqlContext: SQLContext,
-      parameters: Map[String, String],
-      schema: Option[StructType] = None): BaseRelation = {
+      parameters: Map[String, String]): BaseRelation = {
     SimpleScan(parameters("from").toInt, parameters("TO").toInt)(sqlContext)
   }
 }