Spark: SQL extention to update partition field atomically #2365

jackye1995 · 2021-03-24T03:43:47Z

I received some feedback from users about the current Spark SQL extension not able to directly update partition field. Currently it has to first drop and then add the new field, which (1) is not straight-forward for the common use case that updates the granularity of timestamp or bucket transform, (2) creates a time period between 2 commits that is not locked and might cause writer to write data with a wrong partition spec.

This PR introduces the syntax of ALTER TABLE table CHANGE PARTITION FIELD transform TO transform that drops the old transform and adds the new transform in a single commit to solve the issue above.

There is no similar syntax as reference in other systems, Delta lake took the route of directly adding or dropping the entire partition spec so I could not use that as a basis. I chose the current syntax based on the following reasons:

keyword CHANGE is chosen based on the Hive syntax of CHANGE COLUMN col ..., I think we might be able to reuse this keyword in the future for column DDL extensions.
keyword TO is chosen to be consistent with a similar syntax for RENAME col TO col

rdblue · 2021-03-24T19:33:29Z

I considered adding similar syntax when I added the ADD and DROP DDL. I think the problem with this is that it implies that the original field is changed, rather than dropped and added. In format v2, we can sort of do that by replacing the field and using the previous field's name, but even then the values from the old field will no longer appear. In v1, a drop actually replaces the field with a void or alwaysNull transform and adds a new field to the end of the spec. It seems to me like that would be confusing behavior for CHANGE.

One option to fix this is to use REPLACE instead, which I think doesn't imply that the field itself is still there, but updated or changed. But then the syntax is further away from column DDL so I'm not sure if that's a good idea. What do you think, @jackye1995, @aokolnychyi, @yyanyy?

jackye1995 · 2021-03-24T21:22:51Z

@rdblue yes I agree CHANGE might cause some wrong interpretation, but I don't think it is a big issue for end users, especially people who are only interacting through SQL. If they see the alwaysNull partition field I think people would understand it means that partition field was dropped in favor of the current one.

I am good with REPLACE, it is also a part of HiveQL, and Athena also supports it for column updates.

rdblue · 2021-03-25T18:24:38Z

...sions/src/main/antlr/org.apache.spark.sql.catalyst.parser.extensions/IcebergSqlExtensions.g4

    : CALL multipartIdentifier '(' (callArgument (',' callArgument)*)? ')'                  #call
    | ALTER TABLE multipartIdentifier ADD PARTITION FIELD transform (AS name=identifier)?   #addPartitionField
    | ALTER TABLE multipartIdentifier DROP PARTITION FIELD transform                        #dropPartitionField
+    | ALTER TABLE multipartIdentifier REPLACE PARTITION FIELD transform TO transform (AS name=identifier)? #replacePartitionField


REPLACE ... TO doesn't make sense. What about REPLACE ... WITH instead?

Also, why support AS? Shouldn't the new partition field use the same name as the old partition field?

I would also expect using a partition field name to work. So if I used bucket(16, id) AS shard to create the partition, then I should also be able to use REPLACE shard WITH bucket(32, id).

Looks like the implementation supports looking up the field by name, and AS can support rename, like REPLACE ts_day WITH hour(ts) AS ts_hour.

I think there are 2 use cases that have contradicting behaviors:

ADD PARTITION FIELD bucket(id, 16) AS shard, then REPLACE PARTITION FIELD shard WITH bucket(id, 32)

ADD PARTITION FIELD days(ts) AS days_col, then REPLACE PARTITION FIELD days_col WITH hours(ts)

For case 1, we do want the bucket(id, 32) to also be called shard, but we don't really want to call the hours(ts) partition as days_col.

So here are a couple of observations for REPLACE transformFrom WITH transformTo:

if transformFrom is an expression, the default partition field has very specific meanings such as ts_days, id_bucket_16, and the replaced partition field transformTo should not inherit that name

if there is a custom name for the transformFrom partition field, the behavior really depends. The 2 examples above shows this contradicting expectations.

So I think the safest approach is to not infer the behavior for the custom partition name. If the caller wants to use the same name, just use the AS clause to specify it again, such as REPLACE PARTITION FIELD shard WITH bucket(id, 32) AS shard.

What do you think?

Currently we strictly disallow rename -> delete renamed field -> add field with the same name in a single commit for use case 1, but I think that sounds like something we can support in BaseUpdatePartitionSpec, I am thinking about any consequence of it, will update later.

I think you make a good argument that names should not be reused automatically. If the caller wants to use the same name, then it isn't hard to put the name twice: REPLACE shard WITH bucket(32, id) AS shard. Then all we need to support is drop and add with the same name. I think that's simpler to implement and works well.

rdblue · 2021-03-25T18:25:32Z

...sions/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/ReplacePartitionField.scala

+import org.apache.spark.sql.connector.expressions.Transform
+
+case class ReplacePartitionField(
+  table: Seq[String],


Don't we normally use 2 indents for arguments in Scala?

Yeah sorry my IDE was configured wrongly for indentation, let me fix it. We should probably add checkstyle rules for this

rdblue · 2021-03-25T18:29:28Z

...src/main/scala/org/apache/spark/sql/execution/datasources/v2/ReplacePartitionFieldExec.scala

+      case iceberg: SparkTable =>
+        val schema = iceberg.table.schema
+        transformFrom match {
+          case IdentityTransform(FieldReference(parts)) if parts.size == 1 && schema.findField(parts.head) == null =>


Does it work to match on FieldReference(Seq(name)) instead of checking parts?

The current approach is used to be consistent with https://github.com/apache/iceberg/blob/master/spark3-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DropPartitionFieldExec.scala#L45, is there any consideration to deviate from that logic?

I just thought it might be shorter. Not a problem to have it this way.

rdblue · 2021-03-25T18:32:04Z

...src/main/scala/org/apache/spark/sql/execution/datasources/v2/ReplacePartitionFieldExec.scala

+          case _ =>
+            iceberg.table.updateSpec()
+              .addField(name.orNull, Spark3Util.toIcebergTerm(transformTo))
+              .removeField(Spark3Util.toIcebergTerm(transformFrom))


Wouldn't adding and then removing cause a duplicate in the intermediate state? The result of multiple API calls should be the same as the result of multiple commits with a single call. So I think it is safer to remove the term and then add it after.

Yes you are correct, I added a test for it.

rdblue · 2021-03-25T18:32:32Z

...ensions/src/test/java/org/apache/iceberg/spark/extensions/TestAlterTablePartitionFields.java

+        .build();
+    Assert.assertEquals("Should have new spec field", expected, table.spec());
+
+    sql("ALTER TABLE %s REPLACE PARTITION FIELD days(ts) TO hours(ts) AS hour_col", tableName);


Good to see a test for the AS use case.

...ensions/src/test/java/org/apache/iceberg/spark/extensions/TestAlterTablePartitionFields.java

rdblue · 2021-03-27T00:13:32Z

...sions/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/ReplacePartitionField.scala

+
+  override def simpleString(maxFields: Int): String = {
+    s"ReplacePartitionField ${table.quoted} ${transformFrom.describe} " +
+        s"to ${name.map(n => s"$n=").getOrElse("")}${transformTo.describe}"


Should this also be "with" instead of "to"?

rdblue · 2021-03-27T00:27:19Z

This looks ready to go to me. I think there's a minor typo in the simpleString method to fix, but otherwise this is good to go. Thanks @jackye1995!

jackye1995 · 2021-03-29T17:17:34Z

@rdblue thanks for the review, I updated the simple string.

rdblue · 2021-03-29T20:52:38Z

Merged. Thanks @jackye1995!

…2365)

* Add 0.12.0 release notes pt 2 * Add more blurbs and fix formatting. - Add blurbs for #2565, #2583, and #2547. - Make formatting consistent. * Add blurb for #2613 Hive Vectorized Reader * Reword blurbs for #2565 and #2365 * More changes based on review comments * More updates to the 0.12.0 release notes * Add blurb for #2232 fix parquet row group filters * Add blurb for #2308

Spark: SQL extention to update partition field

4e1adab

github-actions bot added the spark label Mar 24, 2021

fix checkstyle

a42497d

use REPLACE instead of CHANGE

d4d07eb

rdblue reviewed Mar 25, 2021

View reviewed changes

...ensions/src/test/java/org/apache/iceberg/spark/extensions/TestAlterTablePartitionFields.java Show resolved Hide resolved

Jack Ye added 2 commits March 25, 2021 11:38

fix based on comments

0114cfc

update based on feedback

0109d78

rdblue reviewed Mar 27, 2021

View reviewed changes

rdblue approved these changes Mar 27, 2021

View reviewed changes

fix simpleString

cc28294

rdblue merged commit e698400 into apache:master Mar 29, 2021

coolderli pushed a commit to coolderli/iceberg that referenced this pull request Apr 26, 2021

Spark: Add REPLACE PARTITION FIELD command to DDL extensions (apache#…

a0d215a

…2365)

jackye1995 mentioned this pull request May 7, 2021

Spark: add spark extension for identifier fields #2560

Merged

jackye1995 mentioned this pull request Aug 17, 2021

Add release notes for 0.12.0 #2973

Merged

cwsteinbach added a commit to cwsteinbach/apache-iceberg that referenced this pull request Aug 17, 2021

Reword blurbs for apache#2565 and apache#2365

6c798f5

Spark: SQL extention to update partition field atomically #2365

Spark: SQL extention to update partition field atomically #2365

Uh oh!

Conversation

jackye1995 commented Mar 24, 2021

Uh oh!

rdblue commented Mar 24, 2021

Uh oh!

jackye1995 commented Mar 24, 2021

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!

rdblue Mar 27, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

rdblue commented Mar 27, 2021

Uh oh!

jackye1995 commented Mar 29, 2021

Uh oh!

rdblue commented Mar 29, 2021

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

rdblue Mar 27, 2021 •

edited

Loading