[SPARK-32402][SQL] Implement ALTER TABLE in JDBC Table Catalog #29324

huaxingao · 2020-08-01T00:42:12Z

What changes were proposed in this pull request?

Implement ALTER TABLE in JDBC Table Catalog
The following ALTER TABLE are implemented:

ALTER TABLE table_name ADD COLUMNS ( column_name datatype [ , ... ] );
ALTER TABLE table_name RENAME COLUMN old_column_name TO new_column_name;
ALTER TABLE table_name DROP COLUMN column_name;
ALTER TABLE table_name ALTER COLUMN column_name TYPE new_type;
ALTER TABLE table_name ALTER COLUMN column_name SET NOT NULL;

I haven't checked ALTER TABLE syntax for all the databases yet. I will check. If there are different syntax, I will have a follow-up to override the dialect.

Seems most of the databases don't support updating comments and column position, so I didn't implement UpdateColumnComment and UpdateColumnPosition.

Why are the changes needed?

Complete the JDBCTableCatalog implementation

Does this PR introduce any user-facing change?

Yes
JDBCTableCatalog.alterTable

How was this patch tested?

add new tests

huaxingao · 2020-08-01T00:49:55Z

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

+      statement.setQueryTimeout(options.queryTimeout)
+      for (sql <- dialect.alterTable(tableName, changes)) {
+        statement.executeUpdate(sql)
+      }


I am debating if I should use statement.executeBatch. The code is simpler without using batch. Not sure if it is common for user to add lots of columns at one time.

What happens if one of the statements fails? Do we leave the table in partially modified state? Should we perform all the statements atomically?

maropu · 2020-08-01T00:59:05Z

Thanks for the work, @huaxingao . IMO we need to add exhaustive tests in integration test units for checking if all the patterns can work on our native supported databases (pg, oracle, db2, mariadb, and sqlserver).

huaxingao · 2020-08-01T01:06:14Z

@maropu Will check all the supported databases and have a follow-up to override the dialect if needed.

SparkQA · 2020-08-01T02:07:08Z

Test build #126906 has finished for PR 29324 at commit 9164379.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

huaxingao · 2020-08-01T02:13:55Z

retest this please

viirya · 2020-08-01T03:16:10Z

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

+            case _ =>
+              throw new IllegalArgumentException(s"Unsupported TableChange fieldNames" +
+                s" ${add.fieldNames}")


When the field names will not be supported?

fieldNames with 2 elements is not supported. It is for nested columns, e.g.

val pointStruct = new StructType().add("x", DoubleType).add("y", DoubleType) val tableSchema = schema.add("point", pointStruct) TableChange.addColumn(Array("point", "z"), DoubleType))

I don't think add nested column is supported by any of these commonly used databases (Oracle, DB2, MS SQL Server, Postgres, etc)

SparkQA · 2020-08-01T04:03:43Z

Test build #126909 has finished for PR 29324 at commit 9164379.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

huaxingao · 2020-08-01T05:46:59Z

retest this please

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

MaxGekk · 2020-08-01T09:04:15Z

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

+    val statement = conn.createStatement
+    try {
+      statement.setQueryTimeout(options.queryTimeout)


Other methods have similar code:

val statement = ... try { statement.setQueryTimeout(options.queryTimeout) statement.execute ... } finally { statement.close() }

Could you put it to a private method.

MaxGekk · 2020-08-01T09:10:06Z

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

+      statement.setQueryTimeout(options.queryTimeout)
+      for (sql <- dialect.alterTable(tableName, changes)) {
+        statement.executeUpdate(sql)
+      }


What happens if one of the statements fails? Do we leave the table in partially modified state? Should we perform all the statements atomically?

MaxGekk · 2020-08-01T09:13:59Z

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


 import java.sql.{Connection, Date, Timestamp}

+import scala.collection.mutable


nit: import scala.collection.mutable.ArrayBuilder

MaxGekk · 2020-08-01T09:16:51Z

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

+   *
+   * @param tableName The name of the table to be altered.
+   * @param changes Changes to apply to the table.
+   * @return The SQL statement to use for altering the table.


It returns multiple SQL statements, correct? And mapping of input changes to SQL statement isn't 1<->1. Could correct this.

Yes, it returns multiple SQL statements. I think the mapping of input changes to SQL statement is 1 to 1.

MaxGekk · 2020-08-01T09:26:13Z

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

+          rename.fieldNames match {
+            case Array(name) =>
+              updateClause += s"ALTER TABLE $tableName RENAME COLUMN $name TO ${rename.newName}"
+            case _ =>


It is not clear for me when it happens. Could you write a test for the case.

MaxGekk · 2020-08-01T09:27:21Z

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

+              throw new IllegalArgumentException(s"Unsupported TableChange fieldNames" +
+                s" ${delete.fieldNames}")
+          }
+        case _ => throw new IllegalArgumentException(s"JDBC alterTable has Unsupported" +


I think throw new NotImplementedError fits better for the case.

Could create sub-tasks for other changes like UpdateColumnNullability and add TODO here if you are not going to implement them in this PR.

I added UpdateColumnType and UpdateColumnNullability. I didn't add UpdateColumnComment and UpdateColumnPosition. Seems most of the databases don't support these two, so I didn't implement these two.

I added UpdateColumnNullability, but didn't add a test for it yet, because the test failed at the check here

case update: UpdateColumnNullability => val field = findField("update", update.fieldNames) val fieldName = update.fieldNames.quoted if (!update.nullable && field.nullable) { alter.failAnalysis( s"Cannot change nullable column to non-nullable: $fieldName") }

I am confused. I thought the whole point of UpdateColumnNullability is to change nullable column to non-nullable?

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

MaxGekk · 2020-08-01T09:33:52Z

...rc/test/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCTableCatalogSuite.scala

+      withConnection { conn =>
+        conn.prepareStatement("""CREATE TABLE "test"."alt_table" (id INTEGER)""").executeUpdate()
+      }


Could you use JDBC Table Catalog functionality:

sql("CREATE TABLE h2.test.alt_table ...")

MaxGekk · 2020-08-01T09:36:01Z

...rc/test/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCTableCatalogSuite.scala

+      assert(sql("DESCRIBE TABLE h2.test.alt_table").select("col_name").take(3) ===
+        Seq(Row("ID"), Row("C1"), Row("C2")))


Could add a helper function which will check column existence in a table.

SparkQA · 2020-08-01T10:03:25Z

Test build #126911 has finished for PR 29324 at commit 9164379.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

…rces/jdbc/JdbcUtils.scala Co-authored-by: Maxim Gekk <[email protected]>

….scala Co-authored-by: Maxim Gekk <[email protected]>

SparkQA · 2020-08-01T22:12:53Z

Test build #126923 has finished for PR 29324 at commit ce00d06.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

huaxingao · 2020-08-02T02:28:20Z

@MaxGekk Thank you very much for your review! I addressed all the comments. Could you please take one more look? Thanks!

SparkQA · 2020-08-02T05:47:04Z

Test build #126927 has finished for PR 29324 at commit 3672250.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

huaxingao · 2020-08-02T05:54:23Z

retest this please

SparkQA · 2020-08-02T07:05:01Z

Test build #126930 has finished for PR 29324 at commit 3672250.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

huaxingao · 2020-08-02T16:34:25Z

retest this please

SparkQA · 2020-08-02T20:23:33Z

Test build #126941 has finished for PR 29324 at commit 3672250.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

huaxingao · 2020-08-02T21:07:57Z

retest this please

SparkQA · 2020-08-03T01:26:15Z

Test build #126946 has finished for PR 29324 at commit 3672250.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

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

...rc/test/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCTableCatalogSuite.scala

….scala Co-authored-by: Maxim Gekk <[email protected]>

…rces/v2/jdbc/JDBCTableCatalogSuite.scala Co-authored-by: Maxim Gekk <[email protected]>

cloud-fan · 2020-08-04T03:55:58Z

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

+    } else {
+      val metadata = conn.getMetaData
+      if (!metadata.supportsTransactions) {
+        throw new SQLFeatureNotSupportedException(s"${this.getClass.getName}.alterTable doesn't" +


this.getClass.getName is JdbcUtils, I don't think it's useful in the error message. How about

The target JDBC server does not support transaction and can only support ALTER TABLE with a single action.

cloud-fan · 2020-08-04T03:57:10Z

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

+      change match {
+        case add: AddColumn =>
+          add.fieldNames match {
+            case Array(name) =>


we should have a better error message if the field name has more than one parts.

Seems to me fieldName always has only one element.

If we do believe it always has one element, maybe add an assert?

People can alter a nested field, that why the type of fieldNames is Array[String]. e.g. ALTER TABLE t RENAME COLUMN a.b TO c

Does this look OK?

case _ => throw new SQLFeatureNotSupportedException("Nested column is not supported.")

add.fieldNames match { case Array(name) =>

We will fail at this pattern match.

One way is

case add: AddColumn if add.fieldNames.length == 1 =>

@cloud-fan I changed code to what you suggested.
We don't support nested column in JDBC yet. In JdbcUtils.getCatalystType, we map java.sql.Types.STRUCT => StringType. In getCommonJDBCType, we don't have a match case for StructType and the default is None. Seems to me currently it is not possible to reach the code path of multiple parts fieldNames in JdbcDialects.alterTable, so I will not have a negative test case for this code path for now.

cloud-fan · 2020-08-04T03:59:30Z

...rc/test/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCTableCatalogSuite.scala

+      assert(t.schema === expectedSchema)
+      sql("ALTER TABLE h2.test.alt_table ADD COLUMNS (C3 DOUBLE)")
+      t = spark.table("h2.test.alt_table")
+      expectedSchema = new StructType()


nit: expectedSchema = expectedSchema.add("C3", DoubleType)

cloud-fan

LGTM except some minor comments.

viirya · 2020-08-04T04:13:36Z

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

+            throw e
+        } finally {
+          statement.close()
+          conn.setAutoCommit(true)


Should we record auto commit status and restore back?

I guess we don't need to record the original auto commit status. The default value of autocommit is true.

viirya · 2020-08-04T04:22:30Z

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

+          update.fieldNames match {
+            case Array(name) =>
+              val nullable = if (update.nullable()) "NULL" else "NOT NULL"
+              updateClause += s"ALTER TABLE $tableName ALTER COLUMN $name SET " + nullable


Since you already use s"", just put nullable into it too.

viirya · 2020-08-04T04:24:53Z

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

+          s" support multiple alter table changes simultaneously for database server" +
+            s" that doesn't have transaction support.")


No need s"" here.

viirya · 2020-08-04T04:26:56Z

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

+          case e: SQLException =>
+            if (conn != null) conn.rollback()
+            throw e


Do we only need to rollback when SQLException? Isn't safer to rollback for all exception?

SparkQA · 2020-08-04T05:01:32Z

Test build #127014 has finished for PR 29324 at commit 114b7a3.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds no public classes.

huaxingao · 2020-08-04T05:53:12Z

Thanks for reviewing. I have addressed the comments.

cloud-fan · 2020-08-04T06:00:45Z

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

+    } else {
+      val metadata = conn.getMetaData
+      if (!metadata.supportsTransactions) {
+        throw new SQLFeatureNotSupportedException("The target JDBC server does not support" +


nit: put a space at the end ... not support " +

SparkQA · 2020-08-04T07:05:01Z

Test build #127029 has finished for PR 29324 at commit db266fa.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-08-04T08:54:35Z

Test build #127041 has finished for PR 29324 at commit a7eccc1.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

huaxingao · 2020-08-04T16:16:04Z

retest this please

SparkQA · 2020-08-04T20:57:18Z

Test build #127059 has finished for PR 29324 at commit a7eccc1.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

HyukjinKwon · 2020-08-05T06:05:37Z

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

  }

-  private def getJdbcType(dt: DataType, dialect: JdbcDialect): JdbcType = {
+  private[sql] def getJdbcType(dt: DataType, dialect: JdbcDialect): JdbcType = {


It's okay to remove private[sql] because execution is already in the private package (see also SPARK-16964)

viirya · 2020-08-05T06:28:58Z

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

+
+  /**
+   * Alter an existing table.
+   * TODO (SPARK-32523): Override this method in the dialects that have different syntax.


Because you will override this method in other places, not here. Remember to remove this later. :)

viirya · 2020-08-05T06:32:38Z

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

+          update.fieldNames match {
+            case Array(name) =>
+              val dataType = JdbcUtils.getJdbcType(update.newDataType(), this)


nit: We know fieldNames must be one element now. We don't need match and can just access fieldNames(0).

SparkQA · 2020-08-05T07:05:01Z

Test build #127082 has finished for PR 29324 at commit dfc7387.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-08-05T11:25:12Z

Test build #127087 has finished for PR 29324 at commit ef55b5e.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2020-08-05T11:56:32Z

thanks, merging to master!

huaxingao · 2020-08-05T15:35:37Z

Thanks everyone!

huaxingao added 2 commits July 31, 2020 16:23

[SPARK-32402][SQL] Implement ALTER TABLE in JDBC Table Catalog

1f38d8a

nit

9164379

probot-autolabeler bot added the SQL label Aug 1, 2020

huaxingao commented Aug 1, 2020

View reviewed changes

viirya reviewed Aug 1, 2020

View reviewed changes

MaxGekk reviewed Aug 1, 2020

View reviewed changes

huaxingao and others added 2 commits August 1, 2020 14:34

Update sql/core/src/main/scala/org/apache/spark/sql/execution/datasou…

dc75eee

…rces/jdbc/JdbcUtils.scala Co-authored-by: Maxim Gekk <[email protected]>

Update sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects…

ce00d06

….scala Co-authored-by: Maxim Gekk <[email protected]>

address comments

3672250

MaxGekk reviewed Aug 3, 2020

View reviewed changes

huaxingao and others added 4 commits August 3, 2020 00:33

Update sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects…

b868b1d

….scala Co-authored-by: Maxim Gekk <[email protected]>

Update sql/core/src/test/scala/org/apache/spark/sql/execution/datasou…

6c32593

…rces/v2/jdbc/JDBCTableCatalogSuite.scala Co-authored-by: Maxim Gekk <[email protected]>

Update sql/core/src/test/scala/org/apache/spark/sql/execution/datasou…

8170ec3

…rces/v2/jdbc/JDBCTableCatalogSuite.scala Co-authored-by: Maxim Gekk <[email protected]>

Update sql/core/src/test/scala/org/apache/spark/sql/execution/datasou…

81f34d9

…rces/v2/jdbc/JDBCTableCatalogSuite.scala Co-authored-by: Maxim Gekk <[email protected]>

cloud-fan reviewed Aug 4, 2020

View reviewed changes

cloud-fan approved these changes Aug 4, 2020

View reviewed changes

viirya reviewed Aug 4, 2020

View reviewed changes

viirya approved these changes Aug 4, 2020

View reviewed changes

address comments

db266fa

cloud-fan reviewed Aug 4, 2020

View reviewed changes

fix

a7eccc1

HyukjinKwon reviewed Aug 5, 2020

View reviewed changes

remove private[sql]

dfc7387

viirya reviewed Aug 5, 2020

View reviewed changes

remove unnecessary pattern matching

ef55b5e

cloud-fan approved these changes Aug 5, 2020

View reviewed changes

cloud-fan closed this in b14a1e2 Aug 5, 2020

huaxingao deleted the alter_table branch August 5, 2020 15:35


		import java.sql.{Connection, Date, Timestamp}

		import scala.collection.mutable

		assert(sql("DESCRIBE TABLE h2.test.alt_table").select("col_name").take(3) ===
		Seq(Row("ID"), Row("C1"), Row("C2")))

		s" support multiple alter table changes simultaneously for database server" +
		s" that doesn't have transaction support.")

[SPARK-32402][SQL] Implement ALTER TABLE in JDBC Table Catalog #29324

[SPARK-32402][SQL] Implement ALTER TABLE in JDBC Table Catalog #29324

Uh oh!

Conversation

huaxingao commented Aug 1, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

maropu commented Aug 1, 2020

Uh oh!

huaxingao commented Aug 1, 2020

Uh oh!

SparkQA commented Aug 1, 2020

Uh oh!

huaxingao commented Aug 1, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Aug 1, 2020

Uh oh!

huaxingao commented Aug 1, 2020

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Aug 1, 2020

Uh oh!

SparkQA commented Aug 1, 2020

Uh oh!

huaxingao commented Aug 2, 2020

Uh oh!

SparkQA commented Aug 2, 2020

Uh oh!

huaxingao commented Aug 2, 2020

Uh oh!

SparkQA commented Aug 2, 2020

Uh oh!

huaxingao commented Aug 2, 2020

Uh oh!

SparkQA commented Aug 2, 2020

Uh oh!

huaxingao commented Aug 2, 2020

Uh oh!

SparkQA commented Aug 3, 2020

Uh oh!

Uh oh!

Uh oh!

Uh oh!

huaxingao commented Aug 1, 2020 •

edited

Loading

cloud-fan Aug 4, 2020 •

edited

Loading