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

[SPARK-32402][SQL][FOLLOW-UP] Use quoted column name for JDBCTableCatalog.alterTable

Browse source 
### What changes were proposed in this pull request?
I currently have unquoted column names in alter table, e.g. ```ALTER TABLE "test"."alt_table" DROP COLUMN c1```
should change to quoted column name ```ALTER TABLE "test"."alt_table" DROP COLUMN "c1"```

### Why are the changes needed?
We should always use quoted identifiers in JDBC SQLs, e.g. ```CREATE TABLE "test"."abc" ("col" INTEGER )  ``` or ```INSERT INTO "test"."abc" ("col") VALUES (?)```. Using unquoted column name in alterTable causes problems, for example:
```
sql("CREATE TABLE h2.test.alt_table (c1 INTEGER, c2 INTEGER) USING _")
sql("ALTER TABLE h2.test.alt_table DROP COLUMN c1")

org.apache.spark.sql.AnalysisException: Failed table altering: test.alt_table;
......

Caused by: org.h2.jdbc.JdbcSQLException: Column "C1" not found; SQL statement:
ALTER TABLE "test"."alt_table" DROP COLUMN c1 [42122-195]

```

### Does this PR introduce _any_ user-facing change?
No

### How was this patch tested?
Existing tests

Closes #30041 from huaxingao/alter_table_followup.

Authored-by: Huaxin Gao <huaxing@us.ibm.com>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>
This commit is contained in:
Huaxin Gao 2020-10-15 15:33:23 +00:00 committed by Wenchen Fan
parent 513b6f5af2
commit 31f7097ce0
4 changed files with 39 additions and 28 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
sql/core/src/main/scala/org/apache/spark/sql/jdbc/DB2Dialect.scala
View file

@ -66,7 +66,8 @@ private object DB2Dialect extends JdbcDialect {
tableName: String,
columnName: String,
newDataType: String): String =
s"ALTER TABLE $tableName ALTER COLUMN $columnName SET DATA TYPE $newDataType"
s"ALTER TABLE $tableName ALTER COLUMN ${quoteIdentifier(columnName)}" +
s" SET DATA TYPE $newDataType"
// scalastyle:off line.size.limit
// See https://www.ibm.com/support/knowledgecenter/en/SSEPGG_11.5.0/com.ibm.db2.luw.sql.ref.doc/doc/r0000888.html
@ -76,6 +77,6 @@ private object DB2Dialect extends JdbcDialect {
columnName: String,
isNullable: Boolean): String = {
val nullable = if (isNullable) "DROP NOT NULL" else "SET NOT NULL"
s"ALTER TABLE $tableName ALTER COLUMN $columnName $nullable"
s"ALTER TABLE $tableName ALTER COLUMN ${quoteIdentifier(columnName)} $nullable"
}
}

25
sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala
View file

@ -200,7 +200,6 @@ abstract class JdbcDialect extends Serializable {
/**
* Alter an existing table.
* TODO (SPARK-32523): Override this method in the dialects that have different syntax.
*
* @param tableName The name of the table to be altered.
* @param changes Changes to apply to the table.
@ -216,10 +215,10 @@ abstract class JdbcDialect extends Serializable {
updateClause += getAddColumnQuery(tableName, name(0), dataType)
case rename: RenameColumn if rename.fieldNames.length == 1 =>
val name = rename.fieldNames
updateClause += s"ALTER TABLE $tableName RENAME COLUMN ${name(0)} TO ${rename.newName}"
updateClause += getRenameColumnQuery(tableName, name(0), rename.newName)
case delete: DeleteColumn if delete.fieldNames.length == 1 =>
val name = delete.fieldNames
updateClause += s"ALTER TABLE $tableName DROP COLUMN ${name(0)}"
updateClause += getDeleteColumnQuery(tableName, name(0))
case updateColumnType: UpdateColumnType if updateColumnType.fieldNames.length == 1 =>
val name = updateColumnType.fieldNames
val dataType = JdbcUtils.getJdbcType(updateColumnType.newDataType(), this)
@ -227,7 +226,6 @@ abstract class JdbcDialect extends Serializable {
updateClause += getUpdateColumnTypeQuery(tableName, name(0), dataType)
case updateNull: UpdateColumnNullability if updateNull.fieldNames.length == 1 =>
val name = updateNull.fieldNames
val nullable = if (updateNull.nullable()) "NULL" else "NOT NULL"
updateClause += getUpdateColumnNullabilityQuery(tableName, name(0), updateNull.nullable())
case _ =>
throw new SQLFeatureNotSupportedException(s"Unsupported TableChange $change")
@ -236,23 +234,28 @@ abstract class JdbcDialect extends Serializable {
updateClause.result()
}
def getAddColumnQuery(tableName: String, columnName: String, dataType: String): String = {
s"ALTER TABLE $tableName ADD COLUMN $columnName $dataType"
}
def getAddColumnQuery(tableName: String, columnName: String, dataType: String): String =
s"ALTER TABLE $tableName ADD COLUMN ${quoteIdentifier(columnName)} $dataType"
def getRenameColumnQuery(tableName: String, columnName: String, newName: String): String =
s"ALTER TABLE $tableName RENAME COLUMN ${quoteIdentifier(columnName)} TO" +
s" ${quoteIdentifier(newName)}"
def getDeleteColumnQuery(tableName: String, columnName: String): String =
s"ALTER TABLE $tableName DROP COLUMN ${quoteIdentifier(columnName)}"
def getUpdateColumnTypeQuery(
tableName: String,
columnName: String,
newDataType: String): String = {
s"ALTER TABLE $tableName ALTER COLUMN $columnName $newDataType"
}
newDataType: String): String =
s"ALTER TABLE $tableName ALTER COLUMN ${quoteIdentifier(columnName)} $newDataType"
def getUpdateColumnNullabilityQuery(
tableName: String,
columnName: String,
isNullable: Boolean): String = {
val nullable = if (isNullable) "NULL" else "NOT NULL"
s"ALTER TABLE $tableName ALTER COLUMN $columnName SET $nullable"
s"ALTER TABLE $tableName ALTER COLUMN ${quoteIdentifier(columnName)} SET $nullable"
}
/**

11
sql/core/src/main/scala/org/apache/spark/sql/jdbc/OracleDialect.scala
View file

@ -120,21 +120,24 @@ private case object OracleDialect extends JdbcDialect {
}
// see https://docs.oracle.com/cd/B28359_01/server.111/b28286/statements_3001.htm#SQLRF01001
override def getAddColumnQuery(tableName: String, columnName: String, dataType: String): String =
s"ALTER TABLE $tableName ADD $columnName $dataType"
override def getAddColumnQuery(
tableName: String,
columnName: String,
dataType: String): String =
s"ALTER TABLE $tableName ADD ${quoteIdentifier(columnName)} $dataType"
// see https://docs.oracle.com/cd/B28359_01/server.111/b28286/statements_3001.htm#SQLRF01001
override def getUpdateColumnTypeQuery(
tableName: String,
columnName: String,
newDataType: String): String =
s"ALTER TABLE $tableName MODIFY $columnName $newDataType"
s"ALTER TABLE $tableName MODIFY ${quoteIdentifier(columnName)} $newDataType"
override def getUpdateColumnNullabilityQuery(
tableName: String,
columnName: String,
isNullable: Boolean): String = {
val nullable = if (isNullable) "NULL" else "NOT NULL"
s"ALTER TABLE $tableName MODIFY $columnName $nullable"
s"ALTER TABLE $tableName MODIFY ${quoteIdentifier(columnName)} $nullable"
}
}

26
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCTableCatalogSuite.scala
View file

@ -178,15 +178,15 @@ class JDBCTableCatalogSuite extends QueryTest with SharedSparkSession {
.add("C1", IntegerType)
.add("C2", StringType)
assert(t.schema === expectedSchema)
sql("ALTER TABLE h2.test.alt_table ADD COLUMNS (C3 DOUBLE)")
sql("ALTER TABLE h2.test.alt_table ADD COLUMNS (c3 DOUBLE)")
t = spark.table("h2.test.alt_table")
expectedSchema = expectedSchema.add("C3", DoubleType)
expectedSchema = expectedSchema.add("c3", DoubleType)
assert(t.schema === expectedSchema)
// Add already existing column
val msg = intercept[AnalysisException] {
sql("ALTER TABLE h2.test.alt_table ADD COLUMNS (C3 DOUBLE)")
sql("ALTER TABLE h2.test.alt_table ADD COLUMNS (c3 DOUBLE)")
}.getMessage
assert(msg.contains("Cannot add column, because C3 already exists"))
assert(msg.contains("Cannot add column, because c3 already exists"))
}
// Add a column to not existing table and namespace
Seq("h2.test.not_existing_table", "h2.bad_test.not_existing_table").foreach { table =>
@ -199,8 +199,8 @@ class JDBCTableCatalogSuite extends QueryTest with SharedSparkSession {
test("alter table ... rename column") {
withTable("h2.test.alt_table") {
sql("CREATE TABLE h2.test.alt_table (ID INTEGER, C0 INTEGER) USING _")
sql("ALTER TABLE h2.test.alt_table RENAME COLUMN ID TO C")
sql("CREATE TABLE h2.test.alt_table (id INTEGER, C0 INTEGER) USING _")
sql("ALTER TABLE h2.test.alt_table RENAME COLUMN id TO C")
val t = spark.table("h2.test.alt_table")
val expectedSchema = new StructType()
.add("C", IntegerType)
@ -223,8 +223,9 @@ class JDBCTableCatalogSuite extends QueryTest with SharedSparkSession {
test("alter table ... drop column") {
withTable("h2.test.alt_table") {
sql("CREATE TABLE h2.test.alt_table (C1 INTEGER, C2 INTEGER) USING _")
sql("CREATE TABLE h2.test.alt_table (C1 INTEGER, C2 INTEGER, c3 INTEGER) USING _")
sql("ALTER TABLE h2.test.alt_table DROP COLUMN C1")
sql("ALTER TABLE h2.test.alt_table DROP COLUMN c3")
val t = spark.table("h2.test.alt_table")
val expectedSchema = new StructType().add("C2", IntegerType)
assert(t.schema === expectedSchema)
@ -245,10 +246,11 @@ class JDBCTableCatalogSuite extends QueryTest with SharedSparkSession {
test("alter table ... update column type") {
withTable("h2.test.alt_table") {
sql("CREATE TABLE h2.test.alt_table (ID INTEGER) USING _")
sql("CREATE TABLE h2.test.alt_table (ID INTEGER, deptno INTEGER) USING _")
sql("ALTER TABLE h2.test.alt_table ALTER COLUMN id TYPE DOUBLE")
sql("ALTER TABLE h2.test.alt_table ALTER COLUMN deptno TYPE DOUBLE")
val t = spark.table("h2.test.alt_table")
val expectedSchema = new StructType().add("ID", DoubleType)
val expectedSchema = new StructType().add("ID", DoubleType).add("deptno", DoubleType)
assert(t.schema === expectedSchema)
// Update not existing column
val msg1 = intercept[AnalysisException] {
@ -272,10 +274,12 @@ class JDBCTableCatalogSuite extends QueryTest with SharedSparkSession {
test("alter table ... update column nullability") {
withTable("h2.test.alt_table") {
sql("CREATE TABLE h2.test.alt_table (ID INTEGER NOT NULL) USING _")
sql("CREATE TABLE h2.test.alt_table (ID INTEGER NOT NULL, deptno INTEGER NOT NULL) USING _")
sql("ALTER TABLE h2.test.alt_table ALTER COLUMN ID DROP NOT NULL")
sql("ALTER TABLE h2.test.alt_table ALTER COLUMN deptno DROP NOT NULL")
val t = spark.table("h2.test.alt_table")
val expectedSchema = new StructType().add("ID", IntegerType, nullable = true)
val expectedSchema = new StructType()
.add("ID", IntegerType, nullable = true).add("deptno", IntegerType, nullable = true)
assert(t.schema === expectedSchema)
// Update nullability of not existing column
val msg = intercept[AnalysisException] {