ODIn/spark-instrumented-optimizer
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

[SPARK-29695][SQL] ALTER TABLE (SerDe properties) should look up catalog/table like v2 commands

Browse source 
### What changes were proposed in this pull request?
Add AlterTableSerDePropertiesStatement and make ALTER TABLE ... SET SERDE/SERDEPROPERTIES go through the same catalog/table resolution framework of v2 commands.

### Why are the changes needed?
It's important to make all the commands have the same table resolution behavior, to avoid confusing end-users. e.g.
```
USE my_catalog
DESC t // success and describe the table t from my_catalog
ALTER TABLE t SET SERDE 'org.apache.class' // report table not found as there is no table t in the session catalog
```

### Does this PR introduce any user-facing change?
Yes. When running ALTER TABLE ... SET SERDE/SERDEPROPERTIES, Spark fails the command if the current catalog is set to a v2 catalog, or the table name specified a v2 catalog.

### How was this patch tested?
Unit tests.

Closes #26374 from huaxingao/spark_29695.

Authored-by: Huaxin Gao <huaxing@us.ibm.com>
Signed-off-by: Dongjoon Hyun <dhyun@apple.com>
This commit is contained in:
Huaxin Gao 2019-11-04 21:42:39 -08:00 committed by Dongjoon Hyun
parent 66619b84d8
commit 02eecfec99
8 changed files with 134 additions and 75 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4
View file

@ -154,9 +154,9 @@ statement
| ALTER TABLE tableIdentifier partitionSpec?
CHANGE COLUMN?
colName=errorCapturingIdentifier colType colPosition? #changeColumn
| ALTER TABLE tableIdentifier (partitionSpec)?
| ALTER TABLE multipartIdentifier (partitionSpec)?
SET SERDE STRING (WITH SERDEPROPERTIES tablePropertyList)? #setTableSerDe
| ALTER TABLE tableIdentifier (partitionSpec)?
| ALTER TABLE multipartIdentifier (partitionSpec)?
SET SERDEPROPERTIES tablePropertyList #setTableSerDe
| ALTER (TABLE | VIEW) multipartIdentifier ADD (IF NOT EXISTS)?
partitionSpecLocation+ #addTablePartition

19
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
View file

@ -3048,4 +3048,23 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
purge = ctx.PURGE != null,
retainData = false)
}
/**
* Create an [[AlterTableSerDePropertiesStatement]]
*
* For example:
* {{{
* ALTER TABLE multi_part_name [PARTITION spec] SET SERDE serde_name
* [WITH SERDEPROPERTIES props];
* ALTER TABLE multi_part_name [PARTITION spec] SET SERDEPROPERTIES serde_properties;
* }}}
*/
override def visitSetTableSerDe(ctx: SetTableSerDeContext): LogicalPlan = withOrigin(ctx) {
AlterTableSerDePropertiesStatement(
visitMultipartIdentifier(ctx.multipartIdentifier),
Option(ctx.STRING).map(string),
Option(ctx.tablePropertyList).map(visitPropertyKeyValues),
// TODO a partition spec is allowed to have optional values. This is currently violated.
Option(ctx.partitionSpec).map(visitNonOptionalPartitionSpec))
}
}

9
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statements.scala
View file

@ -214,6 +214,15 @@ case class AlterTableDropPartitionStatement(
purge: Boolean,
retainData: Boolean) extends ParsedStatement
/**
* ALTER TABLE ... SERDEPROPERTIES command, as parsed from SQL
*/
case class AlterTableSerDePropertiesStatement(
tableName: Seq[String],
serdeClassName: Option[String],
serdeProperties: Option[Map[String, String]],
partitionSpec: Option[TablePartitionSpec]) extends ParsedStatement
/**
* ALTER VIEW ... SET TBLPROPERTIES command, as parsed from SQL.
*/

84
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala
View file

@ -1320,6 +1320,90 @@ class DDLParserSuite extends AnalysisTest {
ShowCurrentNamespaceStatement())
}
test("alter table: SerDe properties") {
val sql1 = "ALTER TABLE table_name SET SERDE 'org.apache.class'"
val parsed1 = parsePlan(sql1)
val expected1 = AlterTableSerDePropertiesStatement(
Seq("table_name"), Some("org.apache.class"), None, None)
comparePlans(parsed1, expected1)
val sql2 =
"""
|ALTER TABLE table_name SET SERDE 'org.apache.class'
|WITH SERDEPROPERTIES ('columns'='foo,bar', 'field.delim' = ',')
""".stripMargin
val parsed2 = parsePlan(sql2)
val expected2 = AlterTableSerDePropertiesStatement(
Seq("table_name"),
Some("org.apache.class"),
Some(Map("columns" -> "foo,bar", "field.delim" -> ",")),
None)
comparePlans(parsed2, expected2)
val sql3 =
"""
|ALTER TABLE table_name
|SET SERDEPROPERTIES ('columns'='foo,bar', 'field.delim' = ',')
""".stripMargin
val parsed3 = parsePlan(sql3)
val expected3 = AlterTableSerDePropertiesStatement(
Seq("table_name"), None, Some(Map("columns" -> "foo,bar", "field.delim" -> ",")), None)
comparePlans(parsed3, expected3)
val sql4 =
"""
|ALTER TABLE table_name PARTITION (test=1, dt='2008-08-08', country='us')
|SET SERDE 'org.apache.class'
|WITH SERDEPROPERTIES ('columns'='foo,bar', 'field.delim' = ',')
""".stripMargin
val parsed4 = parsePlan(sql4)
val expected4 = AlterTableSerDePropertiesStatement(
Seq("table_name"),
Some("org.apache.class"),
Some(Map("columns" -> "foo,bar", "field.delim" -> ",")),
Some(Map("test" -> "1", "dt" -> "2008-08-08", "country" -> "us")))
comparePlans(parsed4, expected4)
val sql5 =
"""
|ALTER TABLE table_name PARTITION (test=1, dt='2008-08-08', country='us')
|SET SERDEPROPERTIES ('columns'='foo,bar', 'field.delim' = ',')
""".stripMargin
val parsed5 = parsePlan(sql5)
val expected5 = AlterTableSerDePropertiesStatement(
Seq("table_name"),
None,
Some(Map("columns" -> "foo,bar", "field.delim" -> ",")),
Some(Map("test" -> "1", "dt" -> "2008-08-08", "country" -> "us")))
comparePlans(parsed5, expected5)
val sql6 =
"""
|ALTER TABLE a.b.c SET SERDE 'org.apache.class'
|WITH SERDEPROPERTIES ('columns'='foo,bar', 'field.delim' = ',')
""".stripMargin
val parsed6 = parsePlan(sql6)
val expected6 = AlterTableSerDePropertiesStatement(
Seq("a", "b", "c"),
Some("org.apache.class"),
Some(Map("columns" -> "foo,bar", "field.delim" -> ",")),
None)
comparePlans(parsed6, expected6)
val sql7 =
"""
|ALTER TABLE a.b.c PARTITION (test=1, dt='2008-08-08', country='us')
|SET SERDEPROPERTIES ('columns'='foo,bar', 'field.delim' = ',')
""".stripMargin
val parsed7 = parsePlan(sql7)
val expected7 = AlterTableSerDePropertiesStatement(
Seq("a", "b", "c"),
None,
Some(Map("columns" -> "foo,bar", "field.delim" -> ",")),
Some(Map("test" -> "1", "dt" -> "2008-08-08", "country" -> "us")))
comparePlans(parsed7, expected7)
}
private case class TableSpec(
name: Seq[String],
schema: Option[StructType],

9
sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala
View file

@ -412,6 +412,15 @@ class ResolveSessionCatalog(
ifExists,
purge,
retainData)
case AlterTableSerDePropertiesStatement(
tableName, serdeClassName, serdeProperties, partitionSpec) =>
val v1TableName = parseV1Table(tableName, "ALTER TABLE SerDe Properties")
AlterTableSerDePropertiesCommand(
v1TableName.asTableIdentifier,
serdeClassName,
serdeProperties,
partitionSpec)
}
private def parseV1Table(tableName: Seq[String], sql: String): Seq[String] = {

18
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
View file

@ -408,24 +408,6 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder(conf) {
ctx.VIEW != null)
}
/**
* Create an [[AlterTableSerDePropertiesCommand]] command.
*
* For example:
* {{{
* ALTER TABLE table [PARTITION spec] SET SERDE serde_name [WITH SERDEPROPERTIES props];
* ALTER TABLE table [PARTITION spec] SET SERDEPROPERTIES serde_properties;
* }}}
*/
override def visitSetTableSerDe(ctx: SetTableSerDeContext): LogicalPlan = withOrigin(ctx) {
AlterTableSerDePropertiesCommand(
visitTableIdentifier(ctx.tableIdentifier),
Option(ctx.STRING).map(string),
Option(ctx.tablePropertyList).map(visitPropertyKeyValues),
// TODO a partition spec is allowed to have optional values. This is currently violated.
Option(ctx.partitionSpec).map(visitNonOptionalPartitionSpec))
}
/**
* Create a [[AlterTableChangeColumnCommand]] command.
*

11
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
View file

@ -1441,6 +1441,17 @@ class DataSourceV2SQLSuite
}
}
test("ALTER TABLE SerDe properties") {
val t = "testcat.ns1.ns2.tbl"
withTable(t) {
spark.sql(s"CREATE TABLE $t (id bigint, data string) USING foo PARTITIONED BY (id)")
val e = intercept[AnalysisException] {
sql(s"ALTER TABLE $t SET SERDEPROPERTIES ('columns'='foo,bar', 'field.delim' = ',')")
}
assert(e.message.contains("ALTER TABLE SerDe Properties is only supported with v1 tables"))
}
}
private def testV1Command(sqlCommand: String, sqlParams: String): Unit = {
val e = intercept[AnalysisException] {
sql(s"$sqlCommand $sqlParams")

55
sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLParserSuite.scala
View file

@ -458,61 +458,6 @@ class DDLParserSuite extends AnalysisTest with SharedSparkSession {
containsThesePhrases = Seq("key_with_value"))
}
test("alter table: SerDe properties") {
val sql1 = "ALTER TABLE table_name SET SERDE 'org.apache.class'"
val sql2 =
"""
|ALTER TABLE table_name SET SERDE 'org.apache.class'
|WITH SERDEPROPERTIES ('columns'='foo,bar', 'field.delim' = ',')
""".stripMargin
val sql3 =
"""
|ALTER TABLE table_name SET SERDEPROPERTIES ('columns'='foo,bar',
|'field.delim' = ',')
""".stripMargin
val sql4 =
"""
|ALTER TABLE table_name PARTITION (test=1, dt='2008-08-08',
|country='us') SET SERDE 'org.apache.class' WITH SERDEPROPERTIES ('columns'='foo,bar',
|'field.delim' = ',')
""".stripMargin
val sql5 =
"""
|ALTER TABLE table_name PARTITION (test=1, dt='2008-08-08',
|country='us') SET SERDEPROPERTIES ('columns'='foo,bar', 'field.delim' = ',')
""".stripMargin
val parsed1 = parser.parsePlan(sql1)
val parsed2 = parser.parsePlan(sql2)
val parsed3 = parser.parsePlan(sql3)
val parsed4 = parser.parsePlan(sql4)
val parsed5 = parser.parsePlan(sql5)
val tableIdent = TableIdentifier("table_name", None)
val expected1 = AlterTableSerDePropertiesCommand(
tableIdent, Some("org.apache.class"), None, None)
val expected2 = AlterTableSerDePropertiesCommand(
tableIdent,
Some("org.apache.class"),
Some(Map("columns" -> "foo,bar", "field.delim" -> ",")),
None)
val expected3 = AlterTableSerDePropertiesCommand(
tableIdent, None, Some(Map("columns" -> "foo,bar", "field.delim" -> ",")), None)
val expected4 = AlterTableSerDePropertiesCommand(
tableIdent,
Some("org.apache.class"),
Some(Map("columns" -> "foo,bar", "field.delim" -> ",")),
Some(Map("test" -> "1", "dt" -> "2008-08-08", "country" -> "us")))
val expected5 = AlterTableSerDePropertiesCommand(
tableIdent,
None,
Some(Map("columns" -> "foo,bar", "field.delim" -> ",")),
Some(Map("test" -> "1", "dt" -> "2008-08-08", "country" -> "us")))
comparePlans(parsed1, expected1)
comparePlans(parsed2, expected2)
comparePlans(parsed3, expected3)
comparePlans(parsed4, expected4)
comparePlans(parsed5, expected5)
}
test("alter table - SerDe property values must be set") {
assertUnsupported(
sql = "ALTER TABLE my_tab SET SERDE 'serde' " +