[SPARK-21213][SQL] check that partition columns in the partition spec…

… form a prefix of the partition columns defined in table schema

sql/core/src/main/scala/org/apache/spark/sql/execution/command/AnalyzePartitionCommand.scala

@@ -46,23 +46,37 @@ case class AnalyzePartitionCommand(
 
   private def getPartitionSpec(table: CatalogTable): Option[TablePartitionSpec] = {
     val partitionColumnNames = table.partitionColumnNames.toSet
-    val keys =
-      if (conf.caseSensitiveAnalysis) partitionSpec.keys
-      else partitionSpec.keys.map(_.toLowerCase)
-    val invalidColumnNames = keys.filterNot(partitionColumnNames.contains(_))
+    val partitionSpecWithCase =
+      if (conf.caseSensitiveAnalysis) partitionSpec
+      else partitionSpec.map { case (k, v) => (k.toLowerCase, v)}
+    val invalidColumnNames = partitionSpecWithCase.keys.filterNot(partitionColumnNames.contains(_))
     if (invalidColumnNames.nonEmpty) {
       val tableId = table.identifier
       throw new AnalysisException(s"Partition specification for table '${tableId.table}' " +
         s"in database '${tableId.database.get}' refers to unknown partition column(s): " +
         invalidColumnNames.mkString(","))
     }
 
-    val filteredSpec = partitionSpec.filter(_._2.isDefined)
+    // Report an error if partition columns in partition specification do not form
+    // a prefix of the list of partition columns defined in the table schema
+    val isSpecified =
+      table.partitionColumnNames.map(partitionSpecWithCase.getOrElse(_, None).isEmpty)
+    if (isSpecified.init.zip(isSpecified.tail).contains((true, false))) {
+      val tableId = table.identifier
+      val schemaColumns = table.partitionColumnNames.mkString(",")
+      val specColumns = partitionSpecWithCase.keys.mkString(",")
+      throw new AnalysisException("The list of partition columns with values " +
+        s"in partition specification for table '${tableId.table}' " +
+        s"in database '${tableId.database.get}' is not a prefix of the list of " +
+        "partition columns defined in the table schema. " +
+        s"Expected a prefix of [${schemaColumns}], but got [${specColumns}].")
+    }
+
+    val filteredSpec = partitionSpecWithCase.filter(_._2.isDefined).mapValues(_.get)
     if (filteredSpec.isEmpty) {
       None
     } else {
-      if (conf.caseSensitiveAnalysis) Some(filteredSpec.mapValues(_.get))
-      else Some(filteredSpec.map { case (key, value) => (key.toLowerCase, value.get) })
+      Some(filteredSpec)
     }
   }
 

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

@@ -440,6 +440,45 @@ class StatisticsSuite extends StatisticsCollectionTestBase with TestHiveSingleto
     }
   }
 
+  test("analyze partial partition specifications") {
+
+    val tableName = "analyzeTable_part"
+
+    def assertAnalysisException(partitionSpec: String): Unit = {
+      val message = intercept[AnalysisException] {
+        sql(s"ANALYZE TABLE $tableName $partitionSpec COMPUTE STATISTICS")
+      }.getMessage
+      assert(message.contains("The list of partition columns with values " +
+        s"in partition specification for table '${tableName.toLowerCase}' in database 'default' " +
+        "is not a prefix of the list of partition columns defined in the table schema"))
+    }
+
+    withTable(tableName) {
+      sql(
+        s"""
+           |CREATE TABLE $tableName (key STRING, value STRING)
+           |PARTITIONED BY (a STRING, b INT, c STRING)
+         """.stripMargin)
+
+      sql(s"INSERT INTO TABLE $tableName PARTITION (a='a1', b=10, c='c1') SELECT * FROM src")
+
+      sql(s"ANALYZE TABLE $tableName PARTITION (a='a1') COMPUTE STATISTICS")
+      sql(s"ANALYZE TABLE $tableName PARTITION (a='a1', b=10) COMPUTE STATISTICS")
+      sql(s"ANALYZE TABLE $tableName PARTITION (A='a1', b=10) COMPUTE STATISTICS")
+      sql(s"ANALYZE TABLE $tableName PARTITION (b=10, a='a1') COMPUTE STATISTICS")
+      sql(s"ANALYZE TABLE $tableName PARTITION (b=10, A='a1') COMPUTE STATISTICS")
+
+      assertAnalysisException("PARTITION (b=10)")
+      assertAnalysisException("PARTITION (a, b=10)")
+      assertAnalysisException("PARTITION (b=10, c='c1')")
+      assertAnalysisException("PARTITION (a, b=10, c='c1')")
+      assertAnalysisException("PARTITION (c='c1')")
+      assertAnalysisException("PARTITION (a, b, c='c1')")
+      assertAnalysisException("PARTITION (a='a1', c='c1')")
+      assertAnalysisException("PARTITION (a='a1', b, c='c1')")
+    }
+  }
+
   test("analyze non-existent partition") {
 
     def assertAnalysisException(analyzeCommand: String, errorMessage: String): Unit = {