You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by dw...@apache.org on 2023/07/20 15:29:33 UTC
[iceberg] branch master updated: Spark 3.4: Create non-existing Tag/Branch when using CREATE OR REPLACE (#8086)
This is an automated email from the ASF dual-hosted git repository.
dweeks pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/master by this push:
new 9204b2dd17 Spark 3.4: Create non-existing Tag/Branch when using CREATE OR REPLACE (#8086)
9204b2dd17 is described below
commit 9204b2dd17809380744870f57016ca053f6fd312
Author: Eduard Tudenhoefner <et...@gmail.com>
AuthorDate: Thu Jul 20 17:29:27 2023 +0200
Spark 3.4: Create non-existing Tag/Branch when using CREATE OR REPLACE (#8086)
Currently, executing `ALTER TABLE x CREATE OR REPLACE TAG xyz` will fail
with `Tag does not exist: xyz`.
As a user I'd expect this to create the tag due to the `CREATE OR
REPLACE` usage. The same issue happens with branches.
---
.../IcebergSqlExtensionsAstBuilder.scala | 4 ++
.../plans/logical/CreateOrReplaceBranch.scala | 1 +
.../plans/logical/CreateOrReplaceTag.scala | 1 +
.../datasources/v2/CreateOrReplaceBranchExec.scala | 14 +++--
.../datasources/v2/CreateOrReplaceTagExec.scala | 14 +++--
.../v2/ExtendedDataSourceV2Strategy.scala | 9 +--
.../iceberg/spark/extensions/TestBranchDDL.java | 67 ++++++++++++++++++++++
.../iceberg/spark/extensions/TestTagDDL.java | 14 +++++
8 files changed, 110 insertions(+), 14 deletions(-)
diff --git a/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/IcebergSqlExtensionsAstBuilder.scala b/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/IcebergSqlExtensionsAstBuilder.scala
index f758cb08fd..2e438de2b8 100644
--- a/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/IcebergSqlExtensionsAstBuilder.scala
+++ b/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/IcebergSqlExtensionsAstBuilder.scala
@@ -116,6 +116,7 @@ class IcebergSqlExtensionsAstBuilder(delegate: ParserInterface) extends IcebergS
val branchRetention = branchOptionsContext.flatMap(branchOptions => Option(branchOptions.refRetain()))
val branchRefAgeMs = branchRetention.map(retain =>
TimeUnit.valueOf(retain.timeUnit().getText.toUpperCase(Locale.ENGLISH)).toMillis(retain.number().getText.toLong))
+ val create = createOrReplaceBranchClause.CREATE() != null
val replace = ctx.createReplaceBranchClause().REPLACE() != null
val ifNotExists = createOrReplaceBranchClause.EXISTS() != null
@@ -130,6 +131,7 @@ class IcebergSqlExtensionsAstBuilder(delegate: ParserInterface) extends IcebergS
typedVisit[Seq[String]](ctx.multipartIdentifier),
branchName.getText,
branchOptions,
+ create,
replace,
ifNotExists)
}
@@ -153,12 +155,14 @@ class IcebergSqlExtensionsAstBuilder(delegate: ParserInterface) extends IcebergS
tagRefAgeMs
)
+ val create = createTagClause.CREATE() != null
val replace = createTagClause.REPLACE() != null
val ifNotExists = createTagClause.EXISTS() != null
CreateOrReplaceTag(typedVisit[Seq[String]](ctx.multipartIdentifier),
tagName,
tagOptions,
+ create,
replace,
ifNotExists)
}
diff --git a/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/CreateOrReplaceBranch.scala b/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/CreateOrReplaceBranch.scala
index 2a22484499..b7981a3c7a 100644
--- a/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/CreateOrReplaceBranch.scala
+++ b/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/CreateOrReplaceBranch.scala
@@ -25,6 +25,7 @@ case class CreateOrReplaceBranch(
table: Seq[String],
branch: String,
branchOptions: BranchOptions,
+ create: Boolean,
replace: Boolean,
ifNotExists: Boolean) extends LeafCommand {
diff --git a/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/CreateOrReplaceTag.scala b/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/CreateOrReplaceTag.scala
index e48f7d8ed0..6e7db84a90 100644
--- a/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/CreateOrReplaceTag.scala
+++ b/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/CreateOrReplaceTag.scala
@@ -25,6 +25,7 @@ case class CreateOrReplaceTag(
table: Seq[String],
tag: String,
tagOptions: TagOptions,
+ create: Boolean,
replace: Boolean,
ifNotExists: Boolean) extends LeafCommand {
diff --git a/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/CreateOrReplaceBranchExec.scala b/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/CreateOrReplaceBranchExec.scala
index 6457875b15..d4328d4b92 100644
--- a/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/CreateOrReplaceBranchExec.scala
+++ b/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/CreateOrReplaceBranchExec.scala
@@ -32,6 +32,7 @@ case class CreateOrReplaceBranchExec(
ident: Identifier,
branch: String,
branchOptions: BranchOptions,
+ create: Boolean,
replace: Boolean,
ifNotExists: Boolean) extends LeafV2CommandExec {
@@ -51,15 +52,18 @@ case class CreateOrReplaceBranchExec(
"Cannot complete create or replace branch operation on %s, main has no snapshot", ident)
val manageSnapshots = iceberg.table().manageSnapshots()
- if (!replace) {
- val ref = iceberg.table().refs().get(branch)
- if (ref != null && ifNotExists) {
+ val refExists = null != iceberg.table().refs().get(branch)
+
+ if (create && replace && !refExists) {
+ manageSnapshots.createBranch(branch, snapshotId)
+ } else if (replace) {
+ manageSnapshots.replaceBranch(branch, snapshotId)
+ } else {
+ if (refExists && ifNotExists) {
return Nil
}
manageSnapshots.createBranch(branch, snapshotId)
- } else {
- manageSnapshots.replaceBranch(branch, snapshotId)
}
if (branchOptions.numSnapshots.nonEmpty) {
diff --git a/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/CreateOrReplaceTagExec.scala b/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/CreateOrReplaceTagExec.scala
index 7ca193d1b1..372cd75486 100644
--- a/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/CreateOrReplaceTagExec.scala
+++ b/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/CreateOrReplaceTagExec.scala
@@ -31,6 +31,7 @@ case class CreateOrReplaceTagExec(
ident: Identifier,
tag: String,
tagOptions: TagOptions,
+ create: Boolean,
replace: Boolean,
ifNotExists: Boolean) extends LeafV2CommandExec {
@@ -50,15 +51,18 @@ case class CreateOrReplaceTagExec(
"Cannot complete create or replace tag operation on %s, main has no snapshot", ident)
val manageSnapshot = iceberg.table.manageSnapshots()
- if (!replace) {
- val ref = iceberg.table().refs().get(tag)
- if (ref != null && ifNotExists) {
+ val refExists = null != iceberg.table().refs().get(tag)
+
+ if (create && replace && !refExists) {
+ manageSnapshot.createTag(tag, snapshotId)
+ } else if (replace) {
+ manageSnapshot.replaceTag(tag, snapshotId)
+ } else {
+ if (refExists && ifNotExists) {
return Nil
}
manageSnapshot.createTag(tag, snapshotId)
- } else {
- manageSnapshot.replaceTag(tag, snapshotId)
}
if (tagOptions.snapshotRefRetain.nonEmpty) {
diff --git a/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ExtendedDataSourceV2Strategy.scala b/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ExtendedDataSourceV2Strategy.scala
index 2f67508c1f..3af5ddd131 100644
--- a/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ExtendedDataSourceV2Strategy.scala
+++ b/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ExtendedDataSourceV2Strategy.scala
@@ -63,11 +63,12 @@ case class ExtendedDataSourceV2Strategy(spark: SparkSession) extends Strategy wi
AddPartitionFieldExec(catalog, ident, transform, name) :: Nil
case CreateOrReplaceBranch(
- IcebergCatalogAndIdentifier(catalog, ident), branch, branchOptions, replace, ifNotExists) =>
- CreateOrReplaceBranchExec(catalog, ident, branch, branchOptions, replace, ifNotExists) :: Nil
+ IcebergCatalogAndIdentifier(catalog, ident), branch, branchOptions, create, replace, ifNotExists) =>
+ CreateOrReplaceBranchExec(catalog, ident, branch, branchOptions, create, replace, ifNotExists) :: Nil
- case CreateOrReplaceTag(IcebergCatalogAndIdentifier(catalog, ident), tag, tagOptions, replace, ifNotExists) =>
- CreateOrReplaceTagExec(catalog, ident, tag, tagOptions, replace, ifNotExists) :: Nil
+ case CreateOrReplaceTag(
+ IcebergCatalogAndIdentifier(catalog, ident), tag, tagOptions, create, replace, ifNotExists) =>
+ CreateOrReplaceTagExec(catalog, ident, tag, tagOptions, create, replace, ifNotExists) :: Nil
case DropBranch(IcebergCatalogAndIdentifier(catalog, ident), branch, ifExists) =>
DropBranchExec(catalog, ident, branch, ifExists) :: Nil
diff --git a/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestBranchDDL.java b/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestBranchDDL.java
index 1eba38b2f5..0f00603bb4 100644
--- a/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestBranchDDL.java
+++ b/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestBranchDDL.java
@@ -18,6 +18,8 @@
*/
package org.apache.iceberg.spark.extensions;
+import static org.assertj.core.api.Assertions.assertThat;
+
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
@@ -289,4 +291,69 @@ public class TestBranchDDL extends SparkExtensionsTestBase {
df.writeTo(tableName).append();
return validationCatalog.loadTable(tableIdent);
}
+
+ @Test
+ public void createOrReplace() throws NoSuchTableException {
+ Table table = insertRows();
+ long first = table.currentSnapshot().snapshotId();
+ String branchName = "b1";
+ insertRows();
+ long second = table.currentSnapshot().snapshotId();
+ table.manageSnapshots().createBranch(branchName, second).commit();
+
+ sql(
+ "ALTER TABLE %s CREATE OR REPLACE BRANCH %s AS OF VERSION %d",
+ tableName, branchName, first);
+ table.refresh();
+ assertThat(table.refs().get(branchName).snapshotId()).isEqualTo(second);
+ }
+
+ @Test
+ public void createOrReplaceWithNonExistingBranch() throws NoSuchTableException {
+ Table table = insertRows();
+ String branchName = "b1";
+ insertRows();
+ long snapshotId = table.currentSnapshot().snapshotId();
+
+ sql(
+ "ALTER TABLE %s CREATE OR REPLACE BRANCH %s AS OF VERSION %d",
+ tableName, branchName, snapshotId);
+ table.refresh();
+ assertThat(table.refs().get(branchName).snapshotId()).isEqualTo(snapshotId);
+ }
+
+ @Test
+ public void replaceBranch() throws NoSuchTableException {
+ Table table = insertRows();
+ long first = table.currentSnapshot().snapshotId();
+ String branchName = "b1";
+ long expectedMaxRefAgeMs = 1000;
+ table
+ .manageSnapshots()
+ .createBranch(branchName, first)
+ .setMaxRefAgeMs(branchName, expectedMaxRefAgeMs)
+ .commit();
+
+ insertRows();
+ long second = table.currentSnapshot().snapshotId();
+
+ sql("ALTER TABLE %s REPLACE BRANCH %s AS OF VERSION %d", tableName, branchName, second);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ assertThat(ref.snapshotId()).isEqualTo(second);
+ assertThat(ref.maxRefAgeMs()).isEqualTo(expectedMaxRefAgeMs);
+ }
+
+ @Test
+ public void replaceBranchDoesNotExist() throws NoSuchTableException {
+ Table table = insertRows();
+
+ Assertions.assertThatThrownBy(
+ () ->
+ sql(
+ "ALTER TABLE %s REPLACE BRANCH %s AS OF VERSION %d",
+ tableName, "someBranch", table.currentSnapshot().snapshotId()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Branch does not exist: someBranch");
+ }
}
diff --git a/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestTagDDL.java b/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestTagDDL.java
index ec3148de6c..866a965e33 100644
--- a/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestTagDDL.java
+++ b/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestTagDDL.java
@@ -18,6 +18,8 @@
*/
package org.apache.iceberg.spark.extensions;
+import static org.assertj.core.api.Assertions.assertThat;
+
import java.util.List;
import java.util.Locale;
import java.util.Map;
@@ -364,6 +366,18 @@ public class TestTagDDL extends SparkExtensionsTestBase {
Assert.assertNull("The tag needs to be dropped.", table.refs().get(tagName));
}
+ @Test
+ public void createOrReplaceWithNonExistingTag() throws NoSuchTableException {
+ Table table = insertRows();
+ String tagName = "t1";
+ insertRows();
+ long snapshotId = table.currentSnapshot().snapshotId();
+
+ sql("ALTER TABLE %s CREATE OR REPLACE TAG %s AS OF VERSION %d", tableName, tagName, snapshotId);
+ table.refresh();
+ assertThat(table.refs().get(tagName).snapshotId()).isEqualTo(snapshotId);
+ }
+
private Table insertRows() throws NoSuchTableException {
List<SimpleRecord> records =
ImmutableList.of(new SimpleRecord(1, "a"), new SimpleRecord(2, "b"));