[SPARK-29869][SQL] improve error message in HiveMetastoreCatalog#convertToLogicalRelation

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

@@ -330,7 +330,9 @@ case class DataSource(
    *                        is considered as a non-streaming file based data source. Since we know
    *                        that files already exist, we don't need to check them again.
    */
-  def resolveRelation(checkFilesExist: Boolean = true): BaseRelation = {
+  def resolveRelation(
+      checkFilesExist: Boolean = true,
+      needPartitionInferring: Boolean = true): BaseRelation = {
     val relation = (providingInstance(), userSpecifiedSchema) match {
       // TODO: Throw when too much is given.
       case (dataSource: SchemaRelationProvider, Some(schema)) =>
@@ -391,7 +393,7 @@ case class DataSource(
         } else {
           val globbedPaths = checkAndGlobPathIfNecessary(
             checkEmptyGlobPath = true, checkFilesExist = checkFilesExist)
-          val index = createInMemoryFileIndex(globbedPaths)
+          val index = createInMemoryFileIndex(globbedPaths, needPartitionInferring)
           val (resultDataSchema, resultPartitionSchema) =
             getOrInferFileFormatSchema(format, () => index)
           (index, resultDataSchema, resultPartitionSchema)
@@ -427,7 +429,6 @@ case class DataSource(
           "in the data schema",
           equality)
     }
-
     relation
   }
 
@@ -551,10 +552,12 @@ case class DataSource(
   }
 
   /** Returns an [[InMemoryFileIndex]] that can be used to get partition schema and file list. */
-  private def createInMemoryFileIndex(globbedPaths: Seq[Path]): InMemoryFileIndex = {
+  private def createInMemoryFileIndex(
+      globbedPaths: Seq[Path],
+      needPartitionInferring: Boolean = true): InMemoryFileIndex = {
     val fileStatusCache = FileStatusCache.getOrCreate(sparkSession)
-    new InMemoryFileIndex(
-      sparkSession, globbedPaths, options, userSpecifiedSchema, fileStatusCache)
+    new InMemoryFileIndex(sparkSession, globbedPaths, options, userSpecifiedSchema,
+      fileStatusCache, needPartitionInferring)
   }
 
   /**

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

@@ -50,7 +50,8 @@ class InMemoryFileIndex(
     rootPathsSpecified: Seq[Path],
     parameters: Map[String, String],
     userSpecifiedSchema: Option[StructType],
-    fileStatusCache: FileStatusCache = NoopCache)
+    fileStatusCache: FileStatusCache = NoopCache,
+    needPartitionInferring: Boolean = true)
   extends PartitioningAwareFileIndex(
     sparkSession, parameters, userSpecifiedSchema, fileStatusCache) {
 
@@ -69,7 +70,11 @@ class InMemoryFileIndex(
 
   override def partitionSpec(): PartitionSpec = {
     if (cachedPartitionSpec == null) {
-      cachedPartitionSpec = inferPartitioning()
+      cachedPartitionSpec = if (needPartitionInferring) {
+        inferPartitioning()
+      } else {
+        PartitionSpec.emptySpec
+      }
     }
     logTrace(s"Partition spec: $cachedPartitionSpec")
     cachedPartitionSpec

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala

@@ -245,20 +245,29 @@ private[hive] class HiveMetastoreCatalog(sparkSession: SparkSession) extends Log
                 userSpecifiedSchema = Option(updatedTable.dataSchema),
                 bucketSpec = None,
                 options = options,
-                className = fileType).resolveRelation(),
+                className = fileType).resolveRelation(needPartitionInferring = false),
               table = updatedTable)
 
           catalogProxy.cacheTable(tableIdentifier, created)
           created
         }
-
         logicalRelation
       })
     }
     // The inferred schema may have different field names as the table schema, we should respect
     // it, but also respect the exprId in table relation output.
-    assert(result.output.length == relation.output.length &&
-      result.output.zip(relation.output).forall { case (a1, a2) => a1.dataType == a2.dataType })
+    assert(result.output.length == relation.output.length,
+      s"Target table has ${result.output.length} columns, " +
+        s"but source table has ${relation.output.length} columns. " +
+        s"It may need to recreate the table ${relation.tableMeta.identifier} " +
+        s"or set ${HiveUtils.CONVERT_METASTORE_PARQUET.key} to false to workaround.")
+    result.output.zip(relation.output).foreach { case (a1, a2) =>
+      assert(a1.dataType.sameType(a2.dataType),
+        s"Data type of column ${a1.name} in target table is ${a1.dataType.typeName}, " +
+          s"but column ${a2.name} in source table is ${a2.dataType.typeName}. " +
+          s"It may need to recreate the table ${relation.tableMeta.identifier} " +
+          s"or set ${HiveUtils.CONVERT_METASTORE_PARQUET.key} to false to workaround.")
+    }
     val newOutput = result.output.zip(relation.output).map {
       case (a1, a2) => a1.withExprId(a2.exprId)
     }

sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveMetastoreCatalogSuite.scala

@@ -17,6 +17,9 @@
 
 package org.apache.spark.sql.hive
 
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+
 import org.apache.spark.sql.{QueryTest, Row, SaveMode}
 import org.apache.spark.sql.catalyst.{AliasIdentifier, TableIdentifier}
 import org.apache.spark.sql.catalyst.catalog.CatalogTableType
@@ -358,4 +361,23 @@ class DataSourceWithHiveMetastoreCatalogSuite
         Seq(table("src").count().toString))
     }
   }
+
+  test("SPARK-29869: fix HiveMetastoreCatalog#convertToLogicalRelation throws AssertionError") {
+    withTempPath(dir => {
+      val baseDir = s"${dir.getCanonicalFile.toURI.toString}/test"
+      val partitionLikeDir = s"${dir.getCanonicalFile.toURI.toString}/test/dt=20191113"
+      spark.range(3).selectExpr("id").write.parquet(partitionLikeDir)
+      withTable("non_partition_table") {
+        withSQLConf(HiveUtils.CONVERT_METASTORE_PARQUET.key -> "true") {
+          spark.sql(
+            s"""
+               |CREATE TABLE non_partition_table (id bigint)
+               |STORED AS PARQUET LOCATION '$baseDir'
+               |""".stripMargin)
+          assert(spark.sql("select * from non_partition_table").collect() ===
+            Array(Row(0), Row(1), Row(2)))
+        }
+      }
+    })
+  }
 }