fixes

Author: Patrick Woody

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

@@ -359,7 +359,8 @@ case class DataSource(
         val globbedPaths =
           checkAndGlobPathIfNecessary(checkEmptyGlobPath = true, checkFilesExist = checkFilesExist)
         val useCatalogFileIndex = sparkSession.sqlContext.conf.manageFilesourcePartitions &&
-          catalogTable.isDefined
+          catalogTable.isDefined && catalogTable.get.tracksPartitionsInCatalog &&
+          catalogTable.get.dataSchema.nonEmpty
         val (fileCatalog, dataSchema, partitionSchema) = if (useCatalogFileIndex) {
           val defaultTableSize = sparkSession.sessionState.conf.defaultSizeInBytes
           val index = fileIndexFactory.create(

sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala

@@ -498,7 +498,7 @@ abstract class DDLSuite extends QueryTest with SQLTestUtils {
 
   test("Create partitioned data source table with user specified schema") {
     import testImplicits._
-    val df = sparkContext.parallelize(1 to 10).map(i => (i, i.toString)).toDF("num", "str")
+    val df = sparkContext.parallelize(1 to 10).map(i => (i.toString, i)).toDF("str", "num")
 
     // Case 1: with partitioning columns but no schema: Option("num")
     // Case 2: without schema and partitioning columns: None
@@ -508,7 +508,7 @@ abstract class DDLSuite extends QueryTest with SQLTestUtils {
           .save(pathToPartitionedTable.getCanonicalPath)
         checkSchemaInCreatedDataSourceTable(
           pathToPartitionedTable,
-          userSpecifiedSchema = Option("num int, str string"),
+          userSpecifiedSchema = Option("str string, num int"),
           userSpecifiedPartitionCols = partitionCols,
           expectedSchema = new StructType().add("str", StringType).add("num", IntegerType),
           expectedPartitionCols = partitionCols.map(Seq(_)).getOrElse(Seq.empty[String]))