Skip to content

[SPARK-17732][SQL] ALTER TABLE DROP PARTITION should support comparators #15704

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
dongjoon-hyun wants to merge 14 commits into from
Closed

[SPARK-17732][SQL] ALTER TABLE DROP PARTITION should support comparators #15704

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
e97bde9
[SPARK-17732][SQL] ALTER TABLE DROP PARTITION should support comparators
dongjoon-hyun Oct 31, 2016
ea33aba
Fix HiveDDLSuite and move testcases into it.
dongjoon-hyun Nov 1, 2016
475e5dc
Address comments.
dongjoon-hyun Nov 1, 2016
1556819
Fix indentation.
dongjoon-hyun Nov 1, 2016
17b7fa4
Simplify codes and add error handling and revise error messages.
dongjoon-hyun Nov 2, 2016
f006316
Fix in order to handle multiple partition correctly.
dongjoon-hyun Nov 3, 2016
ac44a0a
Address comments.
dongjoon-hyun Nov 3, 2016
9f04cb9
Use expression in grammar, rename function, and use `match`.
dongjoon-hyun Nov 10, 2016
e157b66
Add a corner case about multi partition specs.
dongjoon-hyun Nov 11, 2016
580a5e7
Prevent invalid range spec.
dongjoon-hyun Nov 11, 2016
2ff93fe
Rebase and fix DataSourceStrategy.
dongjoon-hyun Nov 11, 2016
f3f0ad5
Fix comments.
dongjoon-hyun Nov 12, 2016
ae1d7df
Minimize the underlying calls.
dongjoon-hyun Nov 12, 2016
fab5682
Remove `partitionVal`, use `flatMap/distinct`, and fixes other things.
dongjoon-hyun Nov 14, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 1 addition & 5 deletions sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -239,11 +239,7 @@ partitionSpecLocation
;

partitionSpec
: PARTITION '(' partitionVal (',' partitionVal)* ')'
;

partitionVal
: identifier (EQ constant)?
: PARTITION '(' expression (',' expression)* ')'
;

describeFuncName
Expand Down
30 changes: 26 additions & 4 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -194,10 +194,15 @@ class AstBuilder extends SqlBaseBaseVisitor[AnyRef] with Logging {
*/
override def visitPartitionSpec(
ctx: PartitionSpecContext): Map[String, Option[String]] = withOrigin(ctx) {
val parts = ctx.partitionVal.asScala.map { pVal =>
val name = pVal.identifier.getText
val value = Option(pVal.constant).map(visitStringConstant)
name -> value
val parts = ctx.expression.asScala.map { pVal =>
expression(pVal) match {
case UnresolvedAttribute(name :: Nil) =>
name -> None
case cmp @ EqualTo(UnresolvedAttribute(name :: Nil), constant: Literal) =>
name -> Option(constant.toString)
case _ =>
throw new ParseException("Invalid partition filter specification", ctx)
}
}
// Before calling `toMap`, we check duplicated keys to avoid silently ignore partition values
// in partition spec like PARTITION(a='1', b='2', a='3'). The real semantical check for
Expand All @@ -206,6 +211,23 @@ class AstBuilder extends SqlBaseBaseVisitor[AnyRef] with Logging {
parts.toMap
}

/**
* Create a partition filter specification.
*/
def visitPartitionFilterSpec(ctx: PartitionSpecContext): Expression = withOrigin(ctx) {
val parts = ctx.expression.asScala.map { pVal =>
expression(pVal) match {
case EqualNullSafe(_, _) =>
throw new ParseException("'<=>' operator is not allowed in partition specification.", ctx)
case cmp @ BinaryComparison(UnresolvedAttribute(name :: Nil), constant: Literal) =>
cmp.withNewChildren(Seq(AttributeReference(name, StringType)(), constant))
case _ =>
throw new ParseException("Invalid partition filter specification", ctx)
}
}
parts.reduceLeft(And)
}

/**
* Create a partition specification map without optional values.
*/
Expand Down
2 changes: 1 addition & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -813,7 +813,7 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
}
AlterTableDropPartitionCommand(
visitTableIdentifier(ctx.tableIdentifier),
ctx.partitionSpec.asScala.map(visitNonOptionalPartitionSpec),
ctx.partitionSpec.asScala.map(visitPartitionFilterSpec),
ctx.EXISTS != null,
ctx.PURGE != null)
}
Expand Down
51 changes: 40 additions & 11 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,8 @@ import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.analysis.Resolver
import org.apache.spark.sql.catalyst.catalog._
import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference}
import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, BinaryComparison}
import org.apache.spark.sql.catalyst.expressions.{EqualTo, Expression, PredicateHelper}
import org.apache.spark.sql.execution.datasources.{CaseInsensitiveMap, PartitioningUtils}
import org.apache.spark.sql.types._
import org.apache.spark.util.SerializableConfiguration
Expand Down Expand Up @@ -418,27 +419,55 @@ case class AlterTableRenamePartitionCommand(
*/
case class AlterTableDropPartitionCommand(
tableName: TableIdentifier,
specs: Seq[TablePartitionSpec],
specs: Seq[Expression],
ifExists: Boolean,
purge: Boolean)
extends RunnableCommand {
extends RunnableCommand with PredicateHelper {

private def isRangeComparison(expr: Expression): Boolean = {
expr.find(e => e.isInstanceOf[BinaryComparison] && !e.isInstanceOf[EqualTo]).isDefined
}

override def run(sparkSession: SparkSession): Seq[Row] = {
val catalog = sparkSession.sessionState.catalog
val table = catalog.getTableMetadata(tableName)
val resolver = sparkSession.sessionState.conf.resolver
DDLUtils.verifyAlterTableType(catalog, table, isView = false)
DDLUtils.verifyPartitionProviderIsHive(sparkSession, table, "ALTER TABLE DROP PARTITION")

val normalizedSpecs = specs.map { spec =>
PartitioningUtils.normalizePartitionSpec(
spec,
table.partitionColumnNames,
table.identifier.quotedString,
sparkSession.sessionState.conf.resolver)
specs.foreach { expr =>
Copy link
Contributor

@hvanhovell hvanhovell Nov 3, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

rant:

this code here to me is evidence that we should actually analyze these commands. The analyzer should throw out invalid commands, instead of doing it all during run().

expr.references.foreach { attr =>
if (!table.partitionColumnNames.exists(resolver(_, attr.name))) {
throw new AnalysisException(s"${attr.name} is not a valid partition column " +
s"in table ${table.identifier.quotedString}.")
}
}
}

catalog.dropPartitions(
table.identifier, normalizedSpecs, ignoreIfNotExists = ifExists, purge = purge)
if (specs.exists(isRangeComparison)) {
val partitionSet = specs.flatMap { spec =>
val partitions = catalog.listPartitionsByFilter(table.identifier, Seq(spec)).map(_.spec)
if (partitions.isEmpty && !ifExists) {
throw new AnalysisException(s"There is no partition for ${spec.sql}")
}
partitions
}.distinct
catalog.dropPartitions(
table.identifier, partitionSet, ignoreIfNotExists = ifExists, purge = purge)
} else {
val normalizedSpecs = specs.map { expr =>
val spec = splitConjunctivePredicates(expr).map {
case BinaryComparison(AttributeReference(name, _, _, _), right) => name -> right.toString
}.toMap
PartitioningUtils.normalizePartitionSpec(
spec,
table.partitionColumnNames,
table.identifier.quotedString,
resolver)
}
catalog.dropPartitions(
table.identifier, normalizedSpecs, ignoreIfNotExists = ifExists, purge = purge)
}
Seq.empty[Row]
}

Expand Down
8 changes: 7 additions & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -215,8 +215,14 @@ case class DataSourceAnalysis(conf: CatalystConf) extends Rule[LogicalPlan] {
if (overwrite.enabled) {
val deletedPartitions = initialMatchingPartitions.toSet -- updatedPartitions
if (deletedPartitions.nonEmpty) {
import org.apache.spark.sql.catalyst.expressions._
val expressions = deletedPartitions.map { specs =>
specs.map { case (key, value) =>
EqualTo(AttributeReference(key, StringType)(), Literal.create(value, StringType))
}.reduceLeft(And)
}.toSeq
AlterTableDropPartitionCommand(
l.catalogTable.get.identifier, deletedPartitions.toSeq,
l.catalogTable.get.identifier, expressions,
ifExists = true, purge = true).run(t.sparkSession)
}
}
Expand Down
9 changes: 7 additions & 2 deletions sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLCommandSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ import scala.reflect.{classTag, ClassTag}

import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.catalog._
import org.apache.spark.sql.catalyst.expressions.{And, AttributeReference, EqualTo, Literal}
import org.apache.spark.sql.catalyst.parser.ParseException
import org.apache.spark.sql.catalyst.plans.PlanTest
import org.apache.spark.sql.catalyst.plans.logical.Project
Expand Down Expand Up @@ -612,8 +613,12 @@ class DDLCommandSuite extends PlanTest {
val expected1_table = AlterTableDropPartitionCommand(
tableIdent,
Seq(
Map("dt" -> "2008-08-08", "country" -> "us"),
Map("dt" -> "2009-09-09", "country" -> "uk")),
And(
EqualTo(AttributeReference("dt", StringType)(), Literal.create("2008-08-08", StringType)),
EqualTo(AttributeReference("country", StringType)(), Literal.create("us", StringType))),
And(
EqualTo(AttributeReference("dt", StringType)(), Literal.create("2009-09-09", StringType)),
EqualTo(AttributeReference("country", StringType)(), Literal.create("uk", StringType)))),
ifExists = true,
purge = false)
val expected2_table = expected1_table.copy(ifExists = false)
Expand Down
103 changes: 103 additions & 0 deletions sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ import org.apache.spark.sql.{AnalysisException, QueryTest, Row, SaveMode}
import org.apache.spark.sql.catalyst.analysis.{NoSuchPartitionException, TableAlreadyExistsException}
import org.apache.spark.sql.catalyst.catalog.{CatalogDatabase, CatalogTable, CatalogTableType}
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.parser.ParseException
import org.apache.spark.sql.execution.command.DDLUtils
import org.apache.spark.sql.hive.HiveExternalCatalog
import org.apache.spark.sql.hive.test.TestHiveSingleton
Expand Down Expand Up @@ -225,6 +226,108 @@ class HiveDDLSuite
}
}

test("SPARK-17732: Drop partitions by filter") {
withTable("sales") {
sql("CREATE TABLE sales(id INT) PARTITIONED BY (country STRING, quarter STRING)")

for (country <- Seq("US", "CA", "KR")) {
for (quarter <- 1 to 4) {
sql(s"ALTER TABLE sales ADD PARTITION (country = '$country', quarter = '$quarter')")
}
}

sql("ALTER TABLE sales DROP PARTITION (country < 'KR', quarter > '2')")
checkAnswer(sql("SHOW PARTITIONS sales"),
Row("country=CA/quarter=1") ::
Row("country=CA/quarter=2") ::
Row("country=KR/quarter=1") ::
Row("country=KR/quarter=2") ::
Row("country=KR/quarter=3") ::
Row("country=KR/quarter=4") ::
Row("country=US/quarter=1") ::
Row("country=US/quarter=2") ::
Row("country=US/quarter=3") ::
Row("country=US/quarter=4") :: Nil)

sql("ALTER TABLE sales DROP PARTITION (country < 'KR'), PARTITION (quarter <= '1')")
checkAnswer(sql("SHOW PARTITIONS sales"),
Row("country=KR/quarter=2") ::
Row("country=KR/quarter=3") ::
Row("country=KR/quarter=4") ::
Row("country=US/quarter=2") ::
Row("country=US/quarter=3") ::
Row("country=US/quarter=4") :: Nil)

sql("ALTER TABLE sales DROP PARTITION (country='KR', quarter='4')")
sql("ALTER TABLE sales DROP PARTITION (country='US', quarter='3')")
Copy link
Member

@viirya viirya Nov 2, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let's add a test for dropping multiple partition specs?

Copy link
Member Author

@dongjoon-hyun dongjoon-hyun Nov 2, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Line 251 ?

Copy link
Member

@viirya viirya Nov 3, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I mean something like PARTITON (quarter <= '2'), PARTITION (quarter >= '4').

Copy link
Member Author

@dongjoon-hyun dongjoon-hyun Nov 3, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh, I missed that case. Sure! Thank you again.

Copy link
Member

@viirya viirya Nov 11, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We should add a test case like PARTITON (quarter <= '4'), PARTITION (quarter <= '2') to see what will happen since after the first partition spec is removed the second one may be failed.

Copy link
Member Author

@dongjoon-hyun dongjoon-hyun Nov 11, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sure!

Copy link
Member Author

@dongjoon-hyun dongjoon-hyun Nov 11, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

To add that, we should make another testcases because the remaining partitions are not enough to test that.

Copy link
Member Author

@dongjoon-hyun dongjoon-hyun Nov 11, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I added the case by updating the existing testcases.

checkAnswer(sql("SHOW PARTITIONS sales"),
Row("country=KR/quarter=2") ::
Row("country=KR/quarter=3") ::
Row("country=US/quarter=2") ::
Row("country=US/quarter=4") :: Nil)

sql("ALTER TABLE sales DROP PARTITION (quarter <= 2), PARTITION (quarter >= '4')")
checkAnswer(sql("SHOW PARTITIONS sales"),
Row("country=KR/quarter=3") :: Nil)

// According to the declarative partition spec definitions, this drops the union of target
// partitions without exceptions. Hive raises exceptions because it handles them sequentially.
sql("ALTER TABLE sales DROP PARTITION (quarter <= 4), PARTITION (quarter <= '3')")
checkAnswer(sql("SHOW PARTITIONS sales"), Nil)
}
}

test("SPARK-17732: Error handling for drop partitions by filter") {
withTable("sales") {
sql("CREATE TABLE sales(id INT) PARTITIONED BY (country STRING, quarter STRING)")

val m = intercept[AnalysisException] {
sql("ALTER TABLE sales DROP PARTITION (unknown = 'KR')")
}.getMessage
assert(m.contains("unknown is not a valid partition column in table"))

val m2 = intercept[AnalysisException] {
sql("ALTER TABLE sales DROP PARTITION (unknown < 'KR')")
}.getMessage
assert(m2.contains("unknown is not a valid partition column in table"))

val m3 = intercept[AnalysisException] {
sql("ALTER TABLE sales DROP PARTITION (unknown <=> 'KR')")
}.getMessage
assert(m3.contains("'<=>' operator is not allowed in partition specification"))
Copy link
Member

@gatorsmile gatorsmile Nov 3, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

One more negative case? How about unknown <=> upper('KR')?

Copy link
Member Author

@dongjoon-hyun dongjoon-hyun Nov 3, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The current master behavior looks the following. I added new case with the same behavior.

scala> sql("ALTER TABLE sales DROP PARTITION (unknown = upper('KR'))").show
org.apache.spark.sql.catalyst.parser.ParseException:
extraneous input '(' expecting STRING(line 1, pos 49)


val m4 = intercept[ParseException] {
sql("ALTER TABLE sales DROP PARTITION (unknown <=> upper('KR'))")
}.getMessage
assert(m4.contains("'<=>' operator is not allowed in partition specification"))

val m5 = intercept[ParseException] {
sql("ALTER TABLE sales DROP PARTITION (country < 'KR', quarter)")
}.getMessage
assert(m5.contains("Invalid partition filter specification"))

sql(s"ALTER TABLE sales ADD PARTITION (country = 'KR', quarter = '3')")
val m6 = intercept[AnalysisException] {
sql("ALTER TABLE sales DROP PARTITION (quarter <= '4'), PARTITION (quarter <= '2')")
}.getMessage
// The query is not executed because `PARTITION (quarter <= '2')` is invalid.
checkAnswer(sql("SHOW PARTITIONS sales"),
Row("country=KR/quarter=3") :: Nil)
assert(m6.contains("There is no partition for (`quarter` <= '2')"))
}
}

test("SPARK-17732: Partition filter is not allowed in ADD PARTITION") {
withTable("sales") {
sql("CREATE TABLE sales(id INT) PARTITIONED BY (country STRING, quarter STRING)")

val m = intercept[ParseException] {
sql("ALTER TABLE sales ADD PARTITION (country = 'US', quarter < '1')")
}.getMessage()
assert(m.contains("Invalid partition filter specification"))
}
}

test("drop views") {
withTable("tab1") {
val tabName = "tab1"
Expand Down