You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by GitBox <gi...@apache.org> on 2023/01/18 15:16:03 UTC
[GitHub] [iceberg] hililiwei opened a new pull request, #6617: Spark: Spark SQL Extensions for create branch
hililiwei opened a new pull request, #6617:
URL: https://github.com/apache/iceberg/pull/6617
## What is the purpose of the change
Implement the syntax in the following documents:
https://docs.google.com/document/d/1tbATFPrKF3vNlzkgZQdaW8CAJmbjvryfrlg6C2Ci_aA/edit
Further split ref DDL into smaller PR.
### DDLs for Branch and Retention
#### Create Branches
```sql
ALTER TABLE tableName
{CREATE | REPLACE} BRANCH branchName [AS OF {VERSION snapshotId}]
[WITH SNAPSHOT RETENTION {[num_snapshots SNAPSHOTS]
[interval {MONTHS | DAYS | HOURS | MINUTES}]}]}]
[RETAIN interval {MONTHS | DAYS | HOURS | MINUTES}]
ALTER TABLE prod.db.my_table
CREATE BRANCH test_branch AS OF VERSION 100 WITH SNAPSHOT RETENTION 5
SNAPSHOTS 7 DAYS
RETAIN 12 MONTHS
```
cc @jackye1995 @amogh-jahagirdar
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] jackye1995 commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by GitBox <gi...@apache.org>.
jackye1995 commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1073744274
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestSnapshotRefSQL.java:
##########
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.extensions;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.AssertHelpers;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.spark.SparkCatalogConfig;
+import org.apache.iceberg.spark.source.SimpleRecord;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runners.Parameterized;
+
+public class TestSnapshotRefSQL extends SparkExtensionsTestBase {
Review Comment:
I think this class name is not updated from the original PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] hililiwei commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by "hililiwei (via GitHub)" <gi...@apache.org>.
hililiwei commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1084149504
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateBranch.java:
##########
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.extensions;
+
+import java.util.IllegalFormatConversionException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.iceberg.AssertHelpers;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.spark.SparkCatalogConfig;
+import org.apache.iceberg.spark.source.SimpleRecord;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
+import org.apache.spark.sql.catalyst.parser.extensions.IcebergParseException;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runners.Parameterized;
+
+public class TestCreateBranch extends SparkExtensionsTestBase {
+
+ @Parameterized.Parameters(name = "catalogName = {0}, implementation = {1}, config = {2}")
+ public static Object[][] parameters() {
+ return new Object[][] {
+ {
+ SparkCatalogConfig.SPARK.catalogName(),
+ SparkCatalogConfig.SPARK.implementation(),
+ SparkCatalogConfig.SPARK.properties()
+ }
+ };
+ }
+
+ public TestCreateBranch(String catalogName, String implementation, Map<String, String> config) {
+ super(catalogName, implementation, config);
+ }
+
+ @After
+ public void removeTable() {
+ sql("DROP TABLE IF EXISTS %s", tableName);
+ }
+
+ @Test
+ public void testCreateBranch() throws NoSuchTableException {
+ Table table = createDefaultTableAndInsert2Row();
+ long snapshotId = table.currentSnapshot().snapshotId();
+ String branchName = "b1";
+ Integer minSnapshotsToKeep = 2;
+ long maxSnapshotAge = 2L;
+ long maxRefAge = 10L;
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s AS OF VERSION %d RETAIN %d DAYS WITH SNAPSHOT RETENTION %d SNAPSHOTS %d days",
+ tableName, branchName, snapshotId, maxRefAge, minSnapshotsToKeep, maxSnapshotAge);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertEquals(minSnapshotsToKeep, ref.minSnapshotsToKeep());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxSnapshotAge), ref.maxSnapshotAgeMs().longValue());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxRefAge), ref.maxRefAgeMs().longValue());
+
+ AssertHelpers.assertThrows(
+ "Cannot create an existing branch",
+ IllegalArgumentException.class,
+ "already exists",
+ () -> sql("ALTER TABLE %s CREATE BRANCH %s", tableName, branchName));
+ }
+
+ @Test
+ public void testCreateBranchUseDefaultConfig() throws NoSuchTableException {
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql("ALTER TABLE %s CREATE BRANCH %s", tableName, branchName);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertNull(ref.minSnapshotsToKeep());
+ Assert.assertNull(ref.maxSnapshotAgeMs());
+ Assert.assertNull(ref.maxRefAgeMs());
+ }
+
+ @Test
+ public void testCreateBranchUseCustomMinSnapshotsToKeep() throws NoSuchTableException {
+ Integer minSnapshotsToKeep = 2;
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s WITH SNAPSHOT RETENTION %d SNAPSHOTS",
+ tableName, branchName, minSnapshotsToKeep);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertEquals(minSnapshotsToKeep, ref.minSnapshotsToKeep());
+ Assert.assertNull(ref.maxSnapshotAgeMs());
+ Assert.assertNull(ref.maxRefAgeMs());
+ }
+
+ @Test
+ public void testCreateBranchUseCustomMaxSnapshotAge() throws NoSuchTableException {
+ long maxSnapshotAge = 2L;
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s WITH SNAPSHOT RETENTION %d DAYS",
+ tableName, branchName, maxSnapshotAge);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertNull(ref.minSnapshotsToKeep());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxSnapshotAge), ref.maxSnapshotAgeMs().longValue());
+ Assert.assertNull(ref.maxRefAgeMs());
+ }
+
+ @Test
+ public void testCreateBranchUseCustomMinSnapshotsToKeepAndMaxSnapshotAge()
+ throws NoSuchTableException {
+ Integer minSnapshotsToKeep = 2;
+ long maxSnapshotAge = 2L;
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s WITH SNAPSHOT RETENTION %d SNAPSHOTS %d DAYS",
+ tableName, branchName, minSnapshotsToKeep, maxSnapshotAge);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertEquals(minSnapshotsToKeep, ref.minSnapshotsToKeep());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxSnapshotAge), ref.maxSnapshotAgeMs().longValue());
+ Assert.assertNull(ref.maxRefAgeMs());
+
+ AssertHelpers.assertThrows(
+ "Illegal statement",
+ IcebergParseException.class,
+ "no viable alternative at input 'WITH SNAPSHOT RETENTION'",
+ () ->
+ sql("ALTER TABLE %s CREATE BRANCH %s WITH SNAPSHOT RETENTION", tableName, branchName));
+ }
+
+ @Test
+ public void testCreateBranchUseCustomMaxRefAge() throws NoSuchTableException {
+ long maxRefAge = 10L;
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql("ALTER TABLE %s CREATE BRANCH %s RETAIN %d DAYS", tableName, branchName, maxRefAge);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertNull(ref.minSnapshotsToKeep());
+ Assert.assertNull(ref.maxSnapshotAgeMs());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxRefAge), ref.maxRefAgeMs().longValue());
+
+ AssertHelpers.assertThrows(
+ "Illegal statement",
+ IcebergParseException.class,
+ "mismatched input",
+ () -> sql("ALTER TABLE %s CREATE BRANCH %s RETAIN", tableName, branchName));
+
+ AssertHelpers.assertThrows(
+ "Illegal statement",
+ IllegalFormatConversionException.class,
+ "d != java.lang.String",
+ () -> sql("ALTER TABLE %s CREATE BRANCH %s RETAIN %d DAYS", tableName, branchName, "abc"));
+
+ AssertHelpers.assertThrows(
+ "Illegal statement",
+ IcebergParseException.class,
+ "mismatched input 'SECONDS' expecting {'DAYS', 'HOURS', 'MINUTES'}",
Review Comment:
I tried using the fourth param to test different units of time.
```
@Parameterized.Parameters(
name = "catalogName = {0}, implementation = {1}, config = {2}, timeUnit = {3}")
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] jackye1995 commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by GitBox <gi...@apache.org>.
jackye1995 commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1073729039
##########
spark/v3.3/spark-extensions/src/main/antlr/org.apache.spark.sql.catalyst.parser.extensions/IcebergSqlExtensions.g4:
##########
@@ -168,16 +175,41 @@ fieldList
;
nonReserved
- : ADD | ALTER | AS | ASC | BY | CALL | DESC | DROP | FIELD | FIRST | LAST | NULLS | ORDERED | PARTITION | TABLE | WRITE
- | DISTRIBUTED | LOCALLY | UNORDERED | REPLACE | WITH | IDENTIFIER_KW | FIELDS | SET
+ : ADD | ALTER | AS | ASC | BRANCH | BY | CALL | CREATE | DESC | DROP | FIELD | FIRST | LAST | NULLS | OF | ORDERED | PARTITION | TABLE | WRITE
+ | DISTRIBUTED | LOCALLY | UNORDERED | REPLACE | RETAIN | VERSION | WITH | IDENTIFIER_KW | FIELDS | SET | SNAPSHOT | SNAPSHOTS
| TRUE | FALSE
| MAP
;
+snapshotId
+ : number
+ ;
+
+numSnapshots
+ : number
+ ;
+
+snapshotRetain
+ : number
+ ;
+
+snapshotRefRetain
+ : number
+ ;
+
+snapshotRefRetainTimeUnit
Review Comment:
If we only allow a few values MONTHS, DAYS, HOURS, MINUTES, can we just list them here as keywords instead of using a generic identifier?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] hililiwei commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by GitBox <gi...@apache.org>.
hililiwei commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1082045421
##########
spark/v3.3/spark-extensions/src/main/antlr/org.apache.spark.sql.catalyst.parser.extensions/IcebergSqlExtensions.g4:
##########
@@ -168,34 +175,77 @@ fieldList
;
nonReserved
- : ADD | ALTER | AS | ASC | BY | CALL | DESC | DROP | FIELD | FIRST | LAST | NULLS | ORDERED | PARTITION | TABLE | WRITE
- | DISTRIBUTED | LOCALLY | UNORDERED | REPLACE | WITH | IDENTIFIER_KW | FIELDS | SET
+ : ADD | ALTER | AS | ASC | BRANCH | BY | CALL | CREATE | DAYS | DESC | DROP | FIELD | FIRST | HOURS | LAST | NULLS | OF | ORDERED | PARTITION | TABLE | WRITE
+ | DISTRIBUTED | LOCALLY | MINUTES | MONTHS | UNORDERED | REPLACE | RETAIN | VERSION | WITH | IDENTIFIER_KW | FIELDS | SET | SNAPSHOT | SNAPSHOTS
| TRUE | FALSE
| MAP
;
+snapshotId
+ : number
+ ;
+
+numSnapshots
+ : number
+ ;
+
+snapshotRetain
+ : number
+ ;
+
+snapshotRefRetain
+ : number
+ ;
+
+snapshotRefRetainTimeUnit
+ : timeUnit
+ ;
+
+snapshotRetainTimeUnit
+ : timeUnit
+ ;
+
+timeUnit
+ : MONTHS
+ | DAYS
+ | HOURS
+ | MINUTES
Review Comment:
Should we support it? I prefer at least the minute-level.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1083573324
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateBranch.java:
##########
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.extensions;
+
+import java.util.IllegalFormatConversionException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.iceberg.AssertHelpers;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.spark.SparkCatalogConfig;
+import org.apache.iceberg.spark.source.SimpleRecord;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
+import org.apache.spark.sql.catalyst.parser.extensions.IcebergParseException;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runners.Parameterized;
+
+public class TestCreateBranch extends SparkExtensionsTestBase {
+
+ @Parameterized.Parameters(name = "catalogName = {0}, implementation = {1}, config = {2}")
+ public static Object[][] parameters() {
+ return new Object[][] {
+ {
+ SparkCatalogConfig.SPARK.catalogName(),
+ SparkCatalogConfig.SPARK.implementation(),
+ SparkCatalogConfig.SPARK.properties()
+ }
+ };
+ }
+
+ public TestCreateBranch(String catalogName, String implementation, Map<String, String> config) {
+ super(catalogName, implementation, config);
+ }
+
+ @After
+ public void removeTable() {
+ sql("DROP TABLE IF EXISTS %s", tableName);
+ }
+
+ @Test
+ public void testCreateBranch() throws NoSuchTableException {
+ Table table = createDefaultTableAndInsert2Row();
+ long snapshotId = table.currentSnapshot().snapshotId();
+ String branchName = "b1";
+ Integer minSnapshotsToKeep = 2;
+ long maxSnapshotAge = 2L;
+ long maxRefAge = 10L;
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s AS OF VERSION %d RETAIN %d DAYS WITH SNAPSHOT RETENTION %d SNAPSHOTS %d days",
+ tableName, branchName, snapshotId, maxRefAge, minSnapshotsToKeep, maxSnapshotAge);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertEquals(minSnapshotsToKeep, ref.minSnapshotsToKeep());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxSnapshotAge), ref.maxSnapshotAgeMs().longValue());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxRefAge), ref.maxRefAgeMs().longValue());
+
+ AssertHelpers.assertThrows(
+ "Cannot create an existing branch",
+ IllegalArgumentException.class,
+ "already exists",
+ () -> sql("ALTER TABLE %s CREATE BRANCH %s", tableName, branchName));
+ }
+
+ @Test
+ public void testCreateBranchUseDefaultConfig() throws NoSuchTableException {
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql("ALTER TABLE %s CREATE BRANCH %s", tableName, branchName);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertNull(ref.minSnapshotsToKeep());
+ Assert.assertNull(ref.maxSnapshotAgeMs());
+ Assert.assertNull(ref.maxRefAgeMs());
+ }
+
+ @Test
+ public void testCreateBranchUseCustomMinSnapshotsToKeep() throws NoSuchTableException {
+ Integer minSnapshotsToKeep = 2;
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s WITH SNAPSHOT RETENTION %d SNAPSHOTS",
+ tableName, branchName, minSnapshotsToKeep);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertEquals(minSnapshotsToKeep, ref.minSnapshotsToKeep());
+ Assert.assertNull(ref.maxSnapshotAgeMs());
+ Assert.assertNull(ref.maxRefAgeMs());
+ }
+
+ @Test
+ public void testCreateBranchUseCustomMaxSnapshotAge() throws NoSuchTableException {
+ long maxSnapshotAge = 2L;
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s WITH SNAPSHOT RETENTION %d DAYS",
+ tableName, branchName, maxSnapshotAge);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertNull(ref.minSnapshotsToKeep());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxSnapshotAge), ref.maxSnapshotAgeMs().longValue());
+ Assert.assertNull(ref.maxRefAgeMs());
+ }
+
+ @Test
+ public void testCreateBranchUseCustomMinSnapshotsToKeepAndMaxSnapshotAge()
+ throws NoSuchTableException {
+ Integer minSnapshotsToKeep = 2;
+ long maxSnapshotAge = 2L;
Review Comment:
When using variables like this, it really helps to name them with a unit. That makes it easier to validate the uses, like `%d DAYS", ... maxSnapshotAgeDays, ...` and `TimeUnit.DAYS.toMillis(maxSnapshotAgeDays)`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] hililiwei commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by "hililiwei (via GitHub)" <gi...@apache.org>.
hililiwei commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1083616314
##########
spark/v3.3/spark-extensions/src/main/antlr/org.apache.spark.sql.catalyst.parser.extensions/IcebergSqlExtensions.g4:
##########
@@ -168,34 +175,76 @@ fieldList
;
nonReserved
- : ADD | ALTER | AS | ASC | BY | CALL | DESC | DROP | FIELD | FIRST | LAST | NULLS | ORDERED | PARTITION | TABLE | WRITE
- | DISTRIBUTED | LOCALLY | UNORDERED | REPLACE | WITH | IDENTIFIER_KW | FIELDS | SET
+ : ADD | ALTER | AS | ASC | BRANCH | BY | CALL | CREATE | DAYS | DESC | DROP | FIELD | FIRST | HOURS | LAST | NULLS | OF | ORDERED | PARTITION | TABLE | WRITE
+ | DISTRIBUTED | LOCALLY | MINUTES | MONTHS | UNORDERED | REPLACE | RETAIN | VERSION | WITH | IDENTIFIER_KW | FIELDS | SET | SNAPSHOT | SNAPSHOTS
| TRUE | FALSE
| MAP
;
+snapshotId
+ : number
+ ;
+
+numSnapshots
+ : number
+ ;
+
+snapshotRetain
+ : number
+ ;
+
+snapshotRefRetain
Review Comment:
@jackye1995 asked the same question.
I originally added snapshotRefRetain and snapshotRetain to make the statement parsing code more readable. Removing it is technically feasible. In the new version, I have removed (including create branch).
ref: https://github.com/apache/iceberg/pull/6637#discussion_r1083582974
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] jackye1995 commented on pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by "jackye1995 (via GitHub)" <gi...@apache.org>.
jackye1995 commented on PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#issuecomment-1399009614
I think this PR is mostly ready to go. I see there is a comment in design doc from @flyrain:
```
"VERSION" is used in Iceberg to indicate any table changes including table property changes, which doesn't create a new snapshot. So versions and snapshots are mismatch. Can we call it "SNAPSHOT" to remove the ambiguity?
```
Requesting one more review from him
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] jackye1995 commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by GitBox <gi...@apache.org>.
jackye1995 commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1082118919
##########
spark/v3.3/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/IcebergSqlExtensionsAstBuilder.scala:
##########
@@ -267,6 +286,12 @@ class IcebergSqlExtensionsAstBuilder(delegate: ParserInterface) extends IcebergS
private def typedVisit[T](ctx: ParseTree): T = {
ctx.accept(this).asInstanceOf[T]
}
+
+ private def timeRetain(unit: String, duration: Long): Long = unit.toUpperCase(Locale.ENGLISH) match {
+ case "MONTHS" => 30 * TimeUnit.DAYS.toMillis(duration)
+ case "DAYS" | "HOURS" | "MINUTES" => TimeUnit.valueOf(unit.toUpperCase(Locale.ENGLISH)).toMillis(duration)
Review Comment:
do we still need to use match statement? The parser will only allow these values, so we should be able to directly parse it to TimeUnit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] jackye1995 commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by GitBox <gi...@apache.org>.
jackye1995 commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1073733920
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestSnapshotRefSQL.java:
##########
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.extensions;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.AssertHelpers;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.spark.SparkCatalogConfig;
+import org.apache.iceberg.spark.source.SimpleRecord;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runners.Parameterized;
+
+public class TestSnapshotRefSQL extends SparkExtensionsTestBase {
+
+ @Parameterized.Parameters(name = "catalogName = {0}, implementation = {1}, config = {2}")
+ public static Object[][] parameters() {
+ return new Object[][] {
+ {
+ SparkCatalogConfig.SPARK.catalogName(),
+ SparkCatalogConfig.SPARK.implementation(),
+ SparkCatalogConfig.SPARK.properties()
+ }
+ };
+ }
+
+ public TestSnapshotRefSQL(String catalogName, String implementation, Map<String, String> config) {
+ super(catalogName, implementation, config);
+ }
+
+ @After
+ public void removeTable() {
+ sql("DROP TABLE IF EXISTS %s", tableName);
+ }
+
+ @Test
+ public void testCreateBranch() throws NoSuchTableException {
+ Table table = createDefaultTableAndInsert2Row();
+ long snapshotId = table.currentSnapshot().snapshotId();
+ String branchName = "b1";
+ Integer minSnapshotsToKeep = 2;
+ long maxSnapshotAge = 2L;
+ long maxRefAge = 10L;
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s AS OF VERSION %d WITH SNAPSHOT RETENTION %d SNAPSHOTS %d DAYS RETAIN %d DAYS",
+ tableName, branchName, snapshotId, minSnapshotsToKeep, maxSnapshotAge, maxRefAge);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertEquals(minSnapshotsToKeep, ref.minSnapshotsToKeep());
+ Assert.assertEquals(maxSnapshotAge * 24 * 60 * 60 * 1000L, ref.maxSnapshotAgeMs().longValue());
+ Assert.assertEquals(maxRefAge * 24 * 60 * 60 * 1000L, ref.maxRefAgeMs().longValue());
+
+ AssertHelpers.assertThrows(
+ "Cannot create an existing branch",
+ IllegalArgumentException.class,
+ "already exists",
+ () -> sql("ALTER TABLE %s CREATE BRANCH %s", tableName, branchName));
+
+ String branchName2 = "b2";
Review Comment:
can we separate these to different test cases?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] jackye1995 commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by GitBox <gi...@apache.org>.
jackye1995 commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1082011227
##########
spark/v3.3/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/IcebergSqlExtensionsAstBuilder.scala:
##########
@@ -267,6 +285,16 @@ class IcebergSqlExtensionsAstBuilder(delegate: ParserInterface) extends IcebergS
private def typedVisit[T](ctx: ParseTree): T = {
ctx.accept(this).asInstanceOf[T]
}
+
+ private val timeUnit = (unit: String) => {
+ unit.toUpperCase(Locale.ENGLISH) match {
+ case "MONTHS" => 30 * 24 * 60 * 60 * 1000L
Review Comment:
can use `TimeUnit.MONTHS.toMillis(...)`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1083572778
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateBranch.java:
##########
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.extensions;
+
+import java.util.IllegalFormatConversionException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.iceberg.AssertHelpers;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.spark.SparkCatalogConfig;
+import org.apache.iceberg.spark.source.SimpleRecord;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
+import org.apache.spark.sql.catalyst.parser.extensions.IcebergParseException;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runners.Parameterized;
+
+public class TestCreateBranch extends SparkExtensionsTestBase {
+
+ @Parameterized.Parameters(name = "catalogName = {0}, implementation = {1}, config = {2}")
+ public static Object[][] parameters() {
+ return new Object[][] {
+ {
+ SparkCatalogConfig.SPARK.catalogName(),
+ SparkCatalogConfig.SPARK.implementation(),
+ SparkCatalogConfig.SPARK.properties()
+ }
+ };
+ }
+
+ public TestCreateBranch(String catalogName, String implementation, Map<String, String> config) {
+ super(catalogName, implementation, config);
+ }
+
+ @After
+ public void removeTable() {
+ sql("DROP TABLE IF EXISTS %s", tableName);
+ }
+
+ @Test
+ public void testCreateBranch() throws NoSuchTableException {
+ Table table = createDefaultTableAndInsert2Row();
+ long snapshotId = table.currentSnapshot().snapshotId();
+ String branchName = "b1";
+ Integer minSnapshotsToKeep = 2;
+ long maxSnapshotAge = 2L;
+ long maxRefAge = 10L;
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s AS OF VERSION %d RETAIN %d DAYS WITH SNAPSHOT RETENTION %d SNAPSHOTS %d days",
+ tableName, branchName, snapshotId, maxRefAge, minSnapshotsToKeep, maxSnapshotAge);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertEquals(minSnapshotsToKeep, ref.minSnapshotsToKeep());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxSnapshotAge), ref.maxSnapshotAgeMs().longValue());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxRefAge), ref.maxRefAgeMs().longValue());
+
+ AssertHelpers.assertThrows(
+ "Cannot create an existing branch",
+ IllegalArgumentException.class,
+ "already exists",
+ () -> sql("ALTER TABLE %s CREATE BRANCH %s", tableName, branchName));
+ }
+
+ @Test
+ public void testCreateBranchUseDefaultConfig() throws NoSuchTableException {
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql("ALTER TABLE %s CREATE BRANCH %s", tableName, branchName);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertNull(ref.minSnapshotsToKeep());
+ Assert.assertNull(ref.maxSnapshotAgeMs());
+ Assert.assertNull(ref.maxRefAgeMs());
+ }
+
+ @Test
+ public void testCreateBranchUseCustomMinSnapshotsToKeep() throws NoSuchTableException {
+ Integer minSnapshotsToKeep = 2;
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s WITH SNAPSHOT RETENTION %d SNAPSHOTS",
+ tableName, branchName, minSnapshotsToKeep);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertEquals(minSnapshotsToKeep, ref.minSnapshotsToKeep());
+ Assert.assertNull(ref.maxSnapshotAgeMs());
+ Assert.assertNull(ref.maxRefAgeMs());
+ }
+
+ @Test
+ public void testCreateBranchUseCustomMaxSnapshotAge() throws NoSuchTableException {
+ long maxSnapshotAge = 2L;
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s WITH SNAPSHOT RETENTION %d DAYS",
+ tableName, branchName, maxSnapshotAge);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertNull(ref.minSnapshotsToKeep());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxSnapshotAge), ref.maxSnapshotAgeMs().longValue());
+ Assert.assertNull(ref.maxRefAgeMs());
+ }
+
+ @Test
+ public void testCreateBranchUseCustomMinSnapshotsToKeepAndMaxSnapshotAge()
+ throws NoSuchTableException {
+ Integer minSnapshotsToKeep = 2;
+ long maxSnapshotAge = 2L;
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s WITH SNAPSHOT RETENTION %d SNAPSHOTS %d DAYS",
+ tableName, branchName, minSnapshotsToKeep, maxSnapshotAge);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertEquals(minSnapshotsToKeep, ref.minSnapshotsToKeep());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxSnapshotAge), ref.maxSnapshotAgeMs().longValue());
+ Assert.assertNull(ref.maxRefAgeMs());
+
+ AssertHelpers.assertThrows(
+ "Illegal statement",
+ IcebergParseException.class,
+ "no viable alternative at input 'WITH SNAPSHOT RETENTION'",
+ () ->
+ sql("ALTER TABLE %s CREATE BRANCH %s WITH SNAPSHOT RETENTION", tableName, branchName));
+ }
+
+ @Test
+ public void testCreateBranchUseCustomMaxRefAge() throws NoSuchTableException {
+ long maxRefAge = 10L;
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql("ALTER TABLE %s CREATE BRANCH %s RETAIN %d DAYS", tableName, branchName, maxRefAge);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertNull(ref.minSnapshotsToKeep());
+ Assert.assertNull(ref.maxSnapshotAgeMs());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxRefAge), ref.maxRefAgeMs().longValue());
+
+ AssertHelpers.assertThrows(
+ "Illegal statement",
+ IcebergParseException.class,
+ "mismatched input",
+ () -> sql("ALTER TABLE %s CREATE BRANCH %s RETAIN", tableName, branchName));
+
+ AssertHelpers.assertThrows(
+ "Illegal statement",
+ IllegalFormatConversionException.class,
+ "d != java.lang.String",
Review Comment:
This looks suspicious. Why is this not a mismatched input?
It looks like this isn't reaching the parser and is instead failing in String.format?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1083572053
##########
spark/v3.3/spark-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ExtendedDataSourceV2Strategy.scala:
##########
@@ -61,6 +62,9 @@ case class ExtendedDataSourceV2Strategy(spark: SparkSession) extends Strategy wi
case AddPartitionField(IcebergCatalogAndIdentifier(catalog, ident), transform, name) =>
AddPartitionFieldExec(catalog, ident, transform, name) :: Nil
+ case CreateBranch(IcebergCatalogAndIdentifier(catalog, ident), _, _, _, _, _) =>
+ CreateBranchExec(catalog, ident, plan.asInstanceOf[CreateBranch]) :: Nil
Review Comment:
Why does this pass the logical plan rather than passing the necessary information? Is it just to avoid a longer line?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] amogh-jahagirdar commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by GitBox <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1073943012
##########
spark/v3.3/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/CreateBranch.scala:
##########
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.catalyst.plans.logical
+
+import org.apache.spark.sql.catalyst.expressions.Attribute
+
+case class CreateBranch(table: Seq[String], branch: String, snapshotId: Option[Long], numSnapshots: Option[Long],
+ snapshotRetain: Option[Long], snapshotRefRetain: Option[Long]) extends LeafCommand {
+
+ import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
+
+ override lazy val output: Seq[Attribute] = Nil
+
+ override def simpleString(maxFields: Int): String = {
+ s"Create branch:${branch} for table:${table.quoted} "
Review Comment:
Nit: spacing in the message
`"Create branch: ${branch} for table: ${table.quoted}"`
##########
spark/v3.3/spark-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/CreateBranchExec.scala:
##########
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.v2
+
+import org.apache.iceberg.spark.source.SparkTable
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.plans.logical.CreateBranch
+import org.apache.spark.sql.connector.catalog.Identifier
+import org.apache.spark.sql.connector.catalog.TableCatalog
+
+case class CreateBranchExec(
+ catalog: TableCatalog,
+ ident: Identifier,
+ createBranch: CreateBranch) extends LeafV2CommandExec {
+
+ import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
+
+ override lazy val output: Seq[Attribute] = Nil
+
+ override protected def run(): Seq[InternalRow] = {
+ catalog.loadTable(ident) match {
+ case iceberg: SparkTable =>
+
+ val snapshotId = createBranch.snapshotId.getOrElse(iceberg.table.currentSnapshot().snapshotId())
+ iceberg.table.manageSnapshots()
+ .createBranch(createBranch.branch, snapshotId)
+ .setMinSnapshotsToKeep(createBranch.branch, createBranch.numSnapshots.getOrElse(1L).toInt)
+ // 5 days
+ .setMaxSnapshotAgeMs(createBranch.branch, createBranch.snapshotRetain.getOrElse(5 * 24 * 60 * 60 * 1000L))
+ .setMaxRefAgeMs(createBranch.branch, createBranch.snapshotRefRetain.getOrElse(Long.MaxValue))
+ .commit()
+
+ case table =>
+ throw new UnsupportedOperationException(s"Cannot add branch to non-Iceberg table: $table")
Review Comment:
I think this case will be common for all the reference based operations. We may want to see about extracting to a common parent. Not needed at this point, but we may revisit in later DDL implementations.
##########
spark/v3.3/spark-extensions/src/main/antlr/org.apache.spark.sql.catalyst.parser.extensions/IcebergSqlExtensions.g4:
##########
@@ -73,6 +73,13 @@ statement
| ALTER TABLE multipartIdentifier WRITE writeSpec #setWriteDistributionAndOrdering
| ALTER TABLE multipartIdentifier SET IDENTIFIER_KW FIELDS fieldList #setIdentifierFields
| ALTER TABLE multipartIdentifier DROP IDENTIFIER_KW FIELDS fieldList #dropIdentifierFields
+ | ALTER TABLE multipartIdentifier CREATE BRANCH identifier (AS OF VERSION snapshotId)? (snapshotRetentionClause)? (RETAIN snapshotRefRetain snapshotRefRetainTimeUnit)? #createBranch
Review Comment:
Agreed, after re-reading the syntax I felt this way too. `RETAIN` should be close to the actual branch clause so that association of what the clauses is responsible for (the retention of the reference itself) is there.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] hililiwei commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by "hililiwei (via GitHub)" <gi...@apache.org>.
hililiwei commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1084152486
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateBranch.java:
##########
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.extensions;
+
+import java.util.IllegalFormatConversionException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.iceberg.AssertHelpers;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.spark.SparkCatalogConfig;
+import org.apache.iceberg.spark.source.SimpleRecord;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
+import org.apache.spark.sql.catalyst.parser.extensions.IcebergParseException;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runners.Parameterized;
+
+public class TestCreateBranch extends SparkExtensionsTestBase {
+
+ @Parameterized.Parameters(name = "catalogName = {0}, implementation = {1}, config = {2}")
+ public static Object[][] parameters() {
+ return new Object[][] {
+ {
+ SparkCatalogConfig.SPARK.catalogName(),
+ SparkCatalogConfig.SPARK.implementation(),
+ SparkCatalogConfig.SPARK.properties()
+ }
+ };
+ }
+
+ public TestCreateBranch(String catalogName, String implementation, Map<String, String> config) {
+ super(catalogName, implementation, config);
+ }
+
+ @After
+ public void removeTable() {
+ sql("DROP TABLE IF EXISTS %s", tableName);
+ }
+
+ @Test
+ public void testCreateBranch() throws NoSuchTableException {
+ Table table = createDefaultTableAndInsert2Row();
+ long snapshotId = table.currentSnapshot().snapshotId();
+ String branchName = "b1";
+ Integer minSnapshotsToKeep = 2;
+ long maxSnapshotAge = 2L;
+ long maxRefAge = 10L;
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s AS OF VERSION %d RETAIN %d DAYS WITH SNAPSHOT RETENTION %d SNAPSHOTS %d days",
+ tableName, branchName, snapshotId, maxRefAge, minSnapshotsToKeep, maxSnapshotAge);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertEquals(minSnapshotsToKeep, ref.minSnapshotsToKeep());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxSnapshotAge), ref.maxSnapshotAgeMs().longValue());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxRefAge), ref.maxRefAgeMs().longValue());
+
+ AssertHelpers.assertThrows(
+ "Cannot create an existing branch",
+ IllegalArgumentException.class,
+ "already exists",
+ () -> sql("ALTER TABLE %s CREATE BRANCH %s", tableName, branchName));
+ }
+
+ @Test
+ public void testCreateBranchUseDefaultConfig() throws NoSuchTableException {
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql("ALTER TABLE %s CREATE BRANCH %s", tableName, branchName);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
Review Comment:
updated in https://github.com/apache/iceberg/pull/6637.
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateBranch.java:
##########
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.extensions;
+
+import java.util.IllegalFormatConversionException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.iceberg.AssertHelpers;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.spark.SparkCatalogConfig;
+import org.apache.iceberg.spark.source.SimpleRecord;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
+import org.apache.spark.sql.catalyst.parser.extensions.IcebergParseException;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runners.Parameterized;
+
+public class TestCreateBranch extends SparkExtensionsTestBase {
+
+ @Parameterized.Parameters(name = "catalogName = {0}, implementation = {1}, config = {2}")
+ public static Object[][] parameters() {
+ return new Object[][] {
+ {
+ SparkCatalogConfig.SPARK.catalogName(),
+ SparkCatalogConfig.SPARK.implementation(),
+ SparkCatalogConfig.SPARK.properties()
+ }
+ };
+ }
+
+ public TestCreateBranch(String catalogName, String implementation, Map<String, String> config) {
+ super(catalogName, implementation, config);
+ }
+
+ @After
+ public void removeTable() {
+ sql("DROP TABLE IF EXISTS %s", tableName);
+ }
+
+ @Test
+ public void testCreateBranch() throws NoSuchTableException {
+ Table table = createDefaultTableAndInsert2Row();
+ long snapshotId = table.currentSnapshot().snapshotId();
+ String branchName = "b1";
+ Integer minSnapshotsToKeep = 2;
+ long maxSnapshotAge = 2L;
+ long maxRefAge = 10L;
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s AS OF VERSION %d RETAIN %d DAYS WITH SNAPSHOT RETENTION %d SNAPSHOTS %d days",
+ tableName, branchName, snapshotId, maxRefAge, minSnapshotsToKeep, maxSnapshotAge);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
Review Comment:
updated in https://github.com/apache/iceberg/pull/6637.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] hililiwei commented on pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by "hililiwei (via GitHub)" <gi...@apache.org>.
hililiwei commented on PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#issuecomment-1400474788
hi @rdblue, thank you for your review. I tried to address your comments in #6637. I can raise a separate PR if you think it's necessary, Of course.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] hililiwei commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by "hililiwei (via GitHub)" <gi...@apache.org>.
hililiwei commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1084159331
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateBranch.java:
##########
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.extensions;
+
+import java.util.IllegalFormatConversionException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.iceberg.AssertHelpers;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.spark.SparkCatalogConfig;
+import org.apache.iceberg.spark.source.SimpleRecord;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
+import org.apache.spark.sql.catalyst.parser.extensions.IcebergParseException;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runners.Parameterized;
+
+public class TestCreateBranch extends SparkExtensionsTestBase {
+
+ @Parameterized.Parameters(name = "catalogName = {0}, implementation = {1}, config = {2}")
+ public static Object[][] parameters() {
+ return new Object[][] {
+ {
+ SparkCatalogConfig.SPARK.catalogName(),
+ SparkCatalogConfig.SPARK.implementation(),
+ SparkCatalogConfig.SPARK.properties()
+ }
+ };
+ }
+
+ public TestCreateBranch(String catalogName, String implementation, Map<String, String> config) {
+ super(catalogName, implementation, config);
+ }
+
+ @After
+ public void removeTable() {
+ sql("DROP TABLE IF EXISTS %s", tableName);
+ }
+
+ @Test
+ public void testCreateBranch() throws NoSuchTableException {
+ Table table = createDefaultTableAndInsert2Row();
+ long snapshotId = table.currentSnapshot().snapshotId();
+ String branchName = "b1";
+ Integer minSnapshotsToKeep = 2;
+ long maxSnapshotAge = 2L;
+ long maxRefAge = 10L;
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s AS OF VERSION %d RETAIN %d DAYS WITH SNAPSHOT RETENTION %d SNAPSHOTS %d days",
+ tableName, branchName, snapshotId, maxRefAge, minSnapshotsToKeep, maxSnapshotAge);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertEquals(minSnapshotsToKeep, ref.minSnapshotsToKeep());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxSnapshotAge), ref.maxSnapshotAgeMs().longValue());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxRefAge), ref.maxRefAgeMs().longValue());
+
+ AssertHelpers.assertThrows(
+ "Cannot create an existing branch",
+ IllegalArgumentException.class,
+ "already exists",
+ () -> sql("ALTER TABLE %s CREATE BRANCH %s", tableName, branchName));
+ }
+
+ @Test
+ public void testCreateBranchUseDefaultConfig() throws NoSuchTableException {
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql("ALTER TABLE %s CREATE BRANCH %s", tableName, branchName);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertNull(ref.minSnapshotsToKeep());
+ Assert.assertNull(ref.maxSnapshotAgeMs());
+ Assert.assertNull(ref.maxRefAgeMs());
+ }
+
+ @Test
+ public void testCreateBranchUseCustomMinSnapshotsToKeep() throws NoSuchTableException {
+ Integer minSnapshotsToKeep = 2;
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s WITH SNAPSHOT RETENTION %d SNAPSHOTS",
+ tableName, branchName, minSnapshotsToKeep);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertEquals(minSnapshotsToKeep, ref.minSnapshotsToKeep());
+ Assert.assertNull(ref.maxSnapshotAgeMs());
+ Assert.assertNull(ref.maxRefAgeMs());
+ }
+
+ @Test
+ public void testCreateBranchUseCustomMaxSnapshotAge() throws NoSuchTableException {
+ long maxSnapshotAge = 2L;
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s WITH SNAPSHOT RETENTION %d DAYS",
+ tableName, branchName, maxSnapshotAge);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertNull(ref.minSnapshotsToKeep());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxSnapshotAge), ref.maxSnapshotAgeMs().longValue());
+ Assert.assertNull(ref.maxRefAgeMs());
+ }
+
+ @Test
+ public void testCreateBranchUseCustomMinSnapshotsToKeepAndMaxSnapshotAge()
+ throws NoSuchTableException {
+ Integer minSnapshotsToKeep = 2;
+ long maxSnapshotAge = 2L;
Review Comment:
Changed the unit to a variable. Using a fixed unit suffix is not appropriate, any suggestions?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] jackye1995 commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by GitBox <gi...@apache.org>.
jackye1995 commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1073740501
##########
spark/v3.3/spark-extensions/src/main/antlr/org.apache.spark.sql.catalyst.parser.extensions/IcebergSqlExtensions.g4:
##########
@@ -73,6 +73,13 @@ statement
| ALTER TABLE multipartIdentifier WRITE writeSpec #setWriteDistributionAndOrdering
| ALTER TABLE multipartIdentifier SET IDENTIFIER_KW FIELDS fieldList #setIdentifierFields
| ALTER TABLE multipartIdentifier DROP IDENTIFIER_KW FIELDS fieldList #dropIdentifierFields
+ | ALTER TABLE multipartIdentifier CREATE BRANCH identifier (AS OF VERSION snapshotId)? (snapshotRetentionClause)? (RETAIN snapshotRefRetain snapshotRefRetainTimeUnit)? #createBranch
Review Comment:
feels more natural for the `RETAIN` clause to be before the `snapshotRetentionClause`? Because clauses like:
```
CREATE BRANCH branch AS OF VERSION 12345
WITH SNAPSHOT RETENTION 2 SNAPSHOTS 2 DAYS
RETAIN 2 DAYS
```
vs
```
CREATE BRANCH branch AS OF VERSION 12345
RETAIN 2 DAYS
WITH SNAPSHOT RETENTION 2 SNAPSHOTS 2 DAYS
```
Making `RETAIN` after `WITH SNAPSHOT RETENTION` makes me feel a bit lost about what that clause is doing. By moving it closer to `CREATE BRANCH`, it is clearer that `RETAIN` is describing expiration of the branch.
What do you think?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] jackye1995 commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by GitBox <gi...@apache.org>.
jackye1995 commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1082011726
##########
spark/v3.3/spark-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/CreateBranchExec.scala:
##########
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.v2
+
+import org.apache.iceberg.spark.source.SparkTable
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.plans.logical.CreateBranch
+import org.apache.spark.sql.connector.catalog.Identifier
+import org.apache.spark.sql.connector.catalog.TableCatalog
+
+case class CreateBranchExec(
+ catalog: TableCatalog,
+ ident: Identifier,
+ createBranch: CreateBranch) extends LeafV2CommandExec {
+
+ import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
+
+ override lazy val output: Seq[Attribute] = Nil
+
+ override protected def run(): Seq[InternalRow] = {
+ catalog.loadTable(ident) match {
+ case iceberg: SparkTable =>
+
+ val snapshotId = createBranch.snapshotId.getOrElse(iceberg.table.currentSnapshot().snapshotId())
+ iceberg.table.manageSnapshots()
+ .createBranch(createBranch.branch, snapshotId)
+ .setMinSnapshotsToKeep(createBranch.branch, createBranch.numSnapshots.getOrElse(1L).toInt)
+ // 5 days
+ .setMaxSnapshotAgeMs(createBranch.branch, createBranch.snapshotRetain.getOrElse(5 * 24 * 60 * 60 * 1000L))
+ .setMaxRefAgeMs(createBranch.branch, createBranch.snapshotRefRetain.getOrElse(Long.MaxValue))
+ .commit()
+
+ case table =>
+ throw new UnsupportedOperationException(s"Cannot add branch to non-Iceberg table: $table")
Review Comment:
This seems like an existing pattern for all extensions, so I think it is probably fine to leave it like this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] jackye1995 commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by GitBox <gi...@apache.org>.
jackye1995 commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1082010234
##########
spark/v3.3/spark-extensions/src/main/antlr/org.apache.spark.sql.catalyst.parser.extensions/IcebergSqlExtensions.g4:
##########
@@ -73,6 +73,13 @@ statement
| ALTER TABLE multipartIdentifier WRITE writeSpec #setWriteDistributionAndOrdering
| ALTER TABLE multipartIdentifier SET IDENTIFIER_KW FIELDS fieldList #setIdentifierFields
| ALTER TABLE multipartIdentifier DROP IDENTIFIER_KW FIELDS fieldList #dropIdentifierFields
+ | ALTER TABLE multipartIdentifier CREATE BRANCH identifier (AS OF VERSION snapshotId)? (snapshotRetentionClause)? (RETAIN snapshotRefRetain snapshotRefRetainTimeUnit)? #createBranch
+ ;
+
+snapshotRetentionClause
+ : WITH SNAPSHOT RETENTION numSnapshots SNAPSHOTS
+ | WITH SNAPSHOT RETENTION snapshotRetain snapshotRetainTimeUnit
+ | WITH SNAPSHOT RETENTION numSnapshots SNAPSHOTS snapshotRetain snapshotRetainTimeUnit
Review Comment:
I see, I will leave it here to see if anyone has better suggestions. I am not an Antlr expert 😝
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1083572778
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateBranch.java:
##########
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.extensions;
+
+import java.util.IllegalFormatConversionException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.iceberg.AssertHelpers;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.spark.SparkCatalogConfig;
+import org.apache.iceberg.spark.source.SimpleRecord;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
+import org.apache.spark.sql.catalyst.parser.extensions.IcebergParseException;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runners.Parameterized;
+
+public class TestCreateBranch extends SparkExtensionsTestBase {
+
+ @Parameterized.Parameters(name = "catalogName = {0}, implementation = {1}, config = {2}")
+ public static Object[][] parameters() {
+ return new Object[][] {
+ {
+ SparkCatalogConfig.SPARK.catalogName(),
+ SparkCatalogConfig.SPARK.implementation(),
+ SparkCatalogConfig.SPARK.properties()
+ }
+ };
+ }
+
+ public TestCreateBranch(String catalogName, String implementation, Map<String, String> config) {
+ super(catalogName, implementation, config);
+ }
+
+ @After
+ public void removeTable() {
+ sql("DROP TABLE IF EXISTS %s", tableName);
+ }
+
+ @Test
+ public void testCreateBranch() throws NoSuchTableException {
+ Table table = createDefaultTableAndInsert2Row();
+ long snapshotId = table.currentSnapshot().snapshotId();
+ String branchName = "b1";
+ Integer minSnapshotsToKeep = 2;
+ long maxSnapshotAge = 2L;
+ long maxRefAge = 10L;
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s AS OF VERSION %d RETAIN %d DAYS WITH SNAPSHOT RETENTION %d SNAPSHOTS %d days",
+ tableName, branchName, snapshotId, maxRefAge, minSnapshotsToKeep, maxSnapshotAge);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertEquals(minSnapshotsToKeep, ref.minSnapshotsToKeep());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxSnapshotAge), ref.maxSnapshotAgeMs().longValue());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxRefAge), ref.maxRefAgeMs().longValue());
+
+ AssertHelpers.assertThrows(
+ "Cannot create an existing branch",
+ IllegalArgumentException.class,
+ "already exists",
+ () -> sql("ALTER TABLE %s CREATE BRANCH %s", tableName, branchName));
+ }
+
+ @Test
+ public void testCreateBranchUseDefaultConfig() throws NoSuchTableException {
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql("ALTER TABLE %s CREATE BRANCH %s", tableName, branchName);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertNull(ref.minSnapshotsToKeep());
+ Assert.assertNull(ref.maxSnapshotAgeMs());
+ Assert.assertNull(ref.maxRefAgeMs());
+ }
+
+ @Test
+ public void testCreateBranchUseCustomMinSnapshotsToKeep() throws NoSuchTableException {
+ Integer minSnapshotsToKeep = 2;
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s WITH SNAPSHOT RETENTION %d SNAPSHOTS",
+ tableName, branchName, minSnapshotsToKeep);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertEquals(minSnapshotsToKeep, ref.minSnapshotsToKeep());
+ Assert.assertNull(ref.maxSnapshotAgeMs());
+ Assert.assertNull(ref.maxRefAgeMs());
+ }
+
+ @Test
+ public void testCreateBranchUseCustomMaxSnapshotAge() throws NoSuchTableException {
+ long maxSnapshotAge = 2L;
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s WITH SNAPSHOT RETENTION %d DAYS",
+ tableName, branchName, maxSnapshotAge);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertNull(ref.minSnapshotsToKeep());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxSnapshotAge), ref.maxSnapshotAgeMs().longValue());
+ Assert.assertNull(ref.maxRefAgeMs());
+ }
+
+ @Test
+ public void testCreateBranchUseCustomMinSnapshotsToKeepAndMaxSnapshotAge()
+ throws NoSuchTableException {
+ Integer minSnapshotsToKeep = 2;
+ long maxSnapshotAge = 2L;
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s WITH SNAPSHOT RETENTION %d SNAPSHOTS %d DAYS",
+ tableName, branchName, minSnapshotsToKeep, maxSnapshotAge);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertEquals(minSnapshotsToKeep, ref.minSnapshotsToKeep());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxSnapshotAge), ref.maxSnapshotAgeMs().longValue());
+ Assert.assertNull(ref.maxRefAgeMs());
+
+ AssertHelpers.assertThrows(
+ "Illegal statement",
+ IcebergParseException.class,
+ "no viable alternative at input 'WITH SNAPSHOT RETENTION'",
+ () ->
+ sql("ALTER TABLE %s CREATE BRANCH %s WITH SNAPSHOT RETENTION", tableName, branchName));
+ }
+
+ @Test
+ public void testCreateBranchUseCustomMaxRefAge() throws NoSuchTableException {
+ long maxRefAge = 10L;
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql("ALTER TABLE %s CREATE BRANCH %s RETAIN %d DAYS", tableName, branchName, maxRefAge);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertNull(ref.minSnapshotsToKeep());
+ Assert.assertNull(ref.maxSnapshotAgeMs());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxRefAge), ref.maxRefAgeMs().longValue());
+
+ AssertHelpers.assertThrows(
+ "Illegal statement",
+ IcebergParseException.class,
+ "mismatched input",
+ () -> sql("ALTER TABLE %s CREATE BRANCH %s RETAIN", tableName, branchName));
+
+ AssertHelpers.assertThrows(
+ "Illegal statement",
+ IllegalFormatConversionException.class,
+ "d != java.lang.String",
Review Comment:
This looks suspicious. Why is this not a mismatched input? Is this even reaching the parser?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1083572534
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateBranch.java:
##########
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.extensions;
+
+import java.util.IllegalFormatConversionException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.iceberg.AssertHelpers;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.spark.SparkCatalogConfig;
+import org.apache.iceberg.spark.source.SimpleRecord;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
+import org.apache.spark.sql.catalyst.parser.extensions.IcebergParseException;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runners.Parameterized;
+
+public class TestCreateBranch extends SparkExtensionsTestBase {
+
+ @Parameterized.Parameters(name = "catalogName = {0}, implementation = {1}, config = {2}")
+ public static Object[][] parameters() {
+ return new Object[][] {
+ {
+ SparkCatalogConfig.SPARK.catalogName(),
+ SparkCatalogConfig.SPARK.implementation(),
+ SparkCatalogConfig.SPARK.properties()
+ }
+ };
+ }
+
+ public TestCreateBranch(String catalogName, String implementation, Map<String, String> config) {
+ super(catalogName, implementation, config);
+ }
+
+ @After
+ public void removeTable() {
+ sql("DROP TABLE IF EXISTS %s", tableName);
+ }
+
+ @Test
+ public void testCreateBranch() throws NoSuchTableException {
+ Table table = createDefaultTableAndInsert2Row();
+ long snapshotId = table.currentSnapshot().snapshotId();
Review Comment:
This should use a snapshot other than the default to test the clause.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1083570497
##########
spark/v3.3/spark-extensions/src/main/antlr/org.apache.spark.sql.catalyst.parser.extensions/IcebergSqlExtensions.g4:
##########
@@ -168,34 +175,76 @@ fieldList
;
nonReserved
- : ADD | ALTER | AS | ASC | BY | CALL | DESC | DROP | FIELD | FIRST | LAST | NULLS | ORDERED | PARTITION | TABLE | WRITE
- | DISTRIBUTED | LOCALLY | UNORDERED | REPLACE | WITH | IDENTIFIER_KW | FIELDS | SET
+ : ADD | ALTER | AS | ASC | BRANCH | BY | CALL | CREATE | DAYS | DESC | DROP | FIELD | FIRST | HOURS | LAST | NULLS | OF | ORDERED | PARTITION | TABLE | WRITE
+ | DISTRIBUTED | LOCALLY | MINUTES | MONTHS | UNORDERED | REPLACE | RETAIN | VERSION | WITH | IDENTIFIER_KW | FIELDS | SET | SNAPSHOT | SNAPSHOTS
| TRUE | FALSE
| MAP
;
+snapshotId
+ : number
+ ;
+
+numSnapshots
+ : number
+ ;
+
+snapshotRetain
+ : number
+ ;
+
+snapshotRefRetain
Review Comment:
Why are there so many aliases for number? Are these rules useful?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] jackye1995 commented on pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by GitBox <gi...@apache.org>.
jackye1995 commented on PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#issuecomment-1397789606
Ping some people for thoughts around the syntax: @rdblue @RussellSpitzer @nastra
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] jackye1995 commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by GitBox <gi...@apache.org>.
jackye1995 commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1082118486
##########
spark/v3.3/spark-extensions/src/main/antlr/org.apache.spark.sql.catalyst.parser.extensions/IcebergSqlExtensions.g4:
##########
@@ -168,34 +175,77 @@ fieldList
;
nonReserved
- : ADD | ALTER | AS | ASC | BY | CALL | DESC | DROP | FIELD | FIRST | LAST | NULLS | ORDERED | PARTITION | TABLE | WRITE
- | DISTRIBUTED | LOCALLY | UNORDERED | REPLACE | WITH | IDENTIFIER_KW | FIELDS | SET
+ : ADD | ALTER | AS | ASC | BRANCH | BY | CALL | CREATE | DAYS | DESC | DROP | FIELD | FIRST | HOURS | LAST | NULLS | OF | ORDERED | PARTITION | TABLE | WRITE
+ | DISTRIBUTED | LOCALLY | MINUTES | MONTHS | UNORDERED | REPLACE | RETAIN | VERSION | WITH | IDENTIFIER_KW | FIELDS | SET | SNAPSHOT | SNAPSHOTS
| TRUE | FALSE
| MAP
;
+snapshotId
+ : number
+ ;
+
+numSnapshots
+ : number
+ ;
+
+snapshotRetain
+ : number
+ ;
+
+snapshotRefRetain
+ : number
+ ;
+
+snapshotRefRetainTimeUnit
+ : timeUnit
+ ;
+
+snapshotRetainTimeUnit
+ : timeUnit
+ ;
+
+timeUnit
+ : MONTHS
+ | DAYS
+ | HOURS
+ | MINUTES
Review Comment:
sure, I am fine with minute level. We can always add more if needed.
##########
spark/v3.3/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/IcebergSqlExtensionsAstBuilder.scala:
##########
@@ -267,6 +286,12 @@ class IcebergSqlExtensionsAstBuilder(delegate: ParserInterface) extends IcebergS
private def typedVisit[T](ctx: ParseTree): T = {
ctx.accept(this).asInstanceOf[T]
}
+
+ private def timeRetain(unit: String, duration: Long): Long = unit.toUpperCase(Locale.ENGLISH) match {
+ case "MONTHS" => 30 * TimeUnit.DAYS.toMillis(duration)
Review Comment:
hmm, good point, there is not exactly always 30 days in a month. That's probably why it is not supported by `TimeUnit`. So I would say in that case let's remove MONTHS from the time unit in parser and just support DAYS, HOURS and MINUTES.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1083573422
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateBranch.java:
##########
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.extensions;
+
+import java.util.IllegalFormatConversionException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.iceberg.AssertHelpers;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.spark.SparkCatalogConfig;
+import org.apache.iceberg.spark.source.SimpleRecord;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
+import org.apache.spark.sql.catalyst.parser.extensions.IcebergParseException;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runners.Parameterized;
+
+public class TestCreateBranch extends SparkExtensionsTestBase {
+
+ @Parameterized.Parameters(name = "catalogName = {0}, implementation = {1}, config = {2}")
+ public static Object[][] parameters() {
+ return new Object[][] {
+ {
+ SparkCatalogConfig.SPARK.catalogName(),
+ SparkCatalogConfig.SPARK.implementation(),
+ SparkCatalogConfig.SPARK.properties()
+ }
+ };
+ }
+
+ public TestCreateBranch(String catalogName, String implementation, Map<String, String> config) {
+ super(catalogName, implementation, config);
+ }
+
+ @After
+ public void removeTable() {
+ sql("DROP TABLE IF EXISTS %s", tableName);
+ }
+
+ @Test
+ public void testCreateBranch() throws NoSuchTableException {
+ Table table = createDefaultTableAndInsert2Row();
+ long snapshotId = table.currentSnapshot().snapshotId();
+ String branchName = "b1";
+ Integer minSnapshotsToKeep = 2;
+ long maxSnapshotAge = 2L;
+ long maxRefAge = 10L;
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s AS OF VERSION %d RETAIN %d DAYS WITH SNAPSHOT RETENTION %d SNAPSHOTS %d days",
+ tableName, branchName, snapshotId, maxRefAge, minSnapshotsToKeep, maxSnapshotAge);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertEquals(minSnapshotsToKeep, ref.minSnapshotsToKeep());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxSnapshotAge), ref.maxSnapshotAgeMs().longValue());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxRefAge), ref.maxRefAgeMs().longValue());
+
+ AssertHelpers.assertThrows(
+ "Cannot create an existing branch",
+ IllegalArgumentException.class,
+ "already exists",
+ () -> sql("ALTER TABLE %s CREATE BRANCH %s", tableName, branchName));
+ }
+
+ @Test
+ public void testCreateBranchUseDefaultConfig() throws NoSuchTableException {
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql("ALTER TABLE %s CREATE BRANCH %s", tableName, branchName);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertNull(ref.minSnapshotsToKeep());
+ Assert.assertNull(ref.maxSnapshotAgeMs());
+ Assert.assertNull(ref.maxRefAgeMs());
+ }
+
+ @Test
+ public void testCreateBranchUseCustomMinSnapshotsToKeep() throws NoSuchTableException {
+ Integer minSnapshotsToKeep = 2;
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s WITH SNAPSHOT RETENTION %d SNAPSHOTS",
+ tableName, branchName, minSnapshotsToKeep);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertEquals(minSnapshotsToKeep, ref.minSnapshotsToKeep());
+ Assert.assertNull(ref.maxSnapshotAgeMs());
+ Assert.assertNull(ref.maxRefAgeMs());
+ }
+
+ @Test
+ public void testCreateBranchUseCustomMaxSnapshotAge() throws NoSuchTableException {
+ long maxSnapshotAge = 2L;
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s WITH SNAPSHOT RETENTION %d DAYS",
+ tableName, branchName, maxSnapshotAge);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertNull(ref.minSnapshotsToKeep());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxSnapshotAge), ref.maxSnapshotAgeMs().longValue());
+ Assert.assertNull(ref.maxRefAgeMs());
+ }
+
+ @Test
+ public void testCreateBranchUseCustomMinSnapshotsToKeepAndMaxSnapshotAge()
+ throws NoSuchTableException {
+ Integer minSnapshotsToKeep = 2;
+ long maxSnapshotAge = 2L;
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s WITH SNAPSHOT RETENTION %d SNAPSHOTS %d DAYS",
+ tableName, branchName, minSnapshotsToKeep, maxSnapshotAge);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertEquals(minSnapshotsToKeep, ref.minSnapshotsToKeep());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxSnapshotAge), ref.maxSnapshotAgeMs().longValue());
+ Assert.assertNull(ref.maxRefAgeMs());
+
+ AssertHelpers.assertThrows(
+ "Illegal statement",
+ IcebergParseException.class,
+ "no viable alternative at input 'WITH SNAPSHOT RETENTION'",
+ () ->
+ sql("ALTER TABLE %s CREATE BRANCH %s WITH SNAPSHOT RETENTION", tableName, branchName));
+ }
+
+ @Test
+ public void testCreateBranchUseCustomMaxRefAge() throws NoSuchTableException {
+ long maxRefAge = 10L;
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql("ALTER TABLE %s CREATE BRANCH %s RETAIN %d DAYS", tableName, branchName, maxRefAge);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertNull(ref.minSnapshotsToKeep());
+ Assert.assertNull(ref.maxSnapshotAgeMs());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxRefAge), ref.maxRefAgeMs().longValue());
+
+ AssertHelpers.assertThrows(
+ "Illegal statement",
+ IcebergParseException.class,
+ "mismatched input",
+ () -> sql("ALTER TABLE %s CREATE BRANCH %s RETAIN", tableName, branchName));
+
+ AssertHelpers.assertThrows(
+ "Illegal statement",
+ IllegalFormatConversionException.class,
+ "d != java.lang.String",
+ () -> sql("ALTER TABLE %s CREATE BRANCH %s RETAIN %d DAYS", tableName, branchName, "abc"));
+
+ AssertHelpers.assertThrows(
+ "Illegal statement",
+ IcebergParseException.class,
+ "mismatched input 'SECONDS' expecting {'DAYS', 'HOURS', 'MINUTES'}",
Review Comment:
`HOURS` and `MINUTES` are never tested, but should be.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1083572437
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateBranch.java:
##########
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.extensions;
+
+import java.util.IllegalFormatConversionException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.iceberg.AssertHelpers;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.spark.SparkCatalogConfig;
+import org.apache.iceberg.spark.source.SimpleRecord;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
+import org.apache.spark.sql.catalyst.parser.extensions.IcebergParseException;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runners.Parameterized;
+
+public class TestCreateBranch extends SparkExtensionsTestBase {
+
+ @Parameterized.Parameters(name = "catalogName = {0}, implementation = {1}, config = {2}")
+ public static Object[][] parameters() {
+ return new Object[][] {
+ {
+ SparkCatalogConfig.SPARK.catalogName(),
+ SparkCatalogConfig.SPARK.implementation(),
+ SparkCatalogConfig.SPARK.properties()
+ }
+ };
+ }
+
+ public TestCreateBranch(String catalogName, String implementation, Map<String, String> config) {
+ super(catalogName, implementation, config);
+ }
+
+ @After
+ public void removeTable() {
+ sql("DROP TABLE IF EXISTS %s", tableName);
+ }
+
+ @Test
+ public void testCreateBranch() throws NoSuchTableException {
+ Table table = createDefaultTableAndInsert2Row();
+ long snapshotId = table.currentSnapshot().snapshotId();
+ String branchName = "b1";
+ Integer minSnapshotsToKeep = 2;
+ long maxSnapshotAge = 2L;
+ long maxRefAge = 10L;
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s AS OF VERSION %d RETAIN %d DAYS WITH SNAPSHOT RETENTION %d SNAPSHOTS %d days",
+ tableName, branchName, snapshotId, maxRefAge, minSnapshotsToKeep, maxSnapshotAge);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertEquals(minSnapshotsToKeep, ref.minSnapshotsToKeep());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxSnapshotAge), ref.maxSnapshotAgeMs().longValue());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxRefAge), ref.maxRefAgeMs().longValue());
+
+ AssertHelpers.assertThrows(
+ "Cannot create an existing branch",
+ IllegalArgumentException.class,
+ "already exists",
+ () -> sql("ALTER TABLE %s CREATE BRANCH %s", tableName, branchName));
+ }
+
+ @Test
+ public void testCreateBranchUseDefaultConfig() throws NoSuchTableException {
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql("ALTER TABLE %s CREATE BRANCH %s", tableName, branchName);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
Review Comment:
This should assert the state of the branch. I think it would use the current snapshot of main, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1083571839
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateBranch.java:
##########
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.extensions;
+
+import java.util.IllegalFormatConversionException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.iceberg.AssertHelpers;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.spark.SparkCatalogConfig;
+import org.apache.iceberg.spark.source.SimpleRecord;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
+import org.apache.spark.sql.catalyst.parser.extensions.IcebergParseException;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runners.Parameterized;
+
+public class TestCreateBranch extends SparkExtensionsTestBase {
+
+ @Parameterized.Parameters(name = "catalogName = {0}, implementation = {1}, config = {2}")
Review Comment:
Why is this testing with multiple catalogs? This is a table-level operation that shouldn't be affected by the catalog, so it should test just one.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] jackye1995 commented on pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by "jackye1995 (via GitHub)" <gi...@apache.org>.
jackye1995 commented on PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#issuecomment-1399207001
Looks like there are quite a few duplicates for #6637 without merging this one. In that case given there are 2 committer votes, I will first merge this one to unblock that PR, and will request @flyrain 's review on the next PR. Thanks for the work @hililiwei , and thanks for all the reviews @yyanyy and @amogh-jahagirdar !
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] jackye1995 commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by GitBox <gi...@apache.org>.
jackye1995 commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1073741499
##########
spark/v3.3/spark-extensions/src/main/antlr/org.apache.spark.sql.catalyst.parser.extensions/IcebergSqlExtensions.g4:
##########
@@ -73,6 +73,13 @@ statement
| ALTER TABLE multipartIdentifier WRITE writeSpec #setWriteDistributionAndOrdering
| ALTER TABLE multipartIdentifier SET IDENTIFIER_KW FIELDS fieldList #setIdentifierFields
| ALTER TABLE multipartIdentifier DROP IDENTIFIER_KW FIELDS fieldList #dropIdentifierFields
+ | ALTER TABLE multipartIdentifier CREATE BRANCH identifier (AS OF VERSION snapshotId)? (snapshotRetentionClause)? (RETAIN snapshotRefRetain snapshotRefRetainTimeUnit)? #createBranch
+ ;
+
+snapshotRetentionClause
+ : WITH SNAPSHOT RETENTION numSnapshots SNAPSHOTS
+ | WITH SNAPSHOT RETENTION snapshotRetain snapshotRetainTimeUnit
+ | WITH SNAPSHOT RETENTION numSnapshots SNAPSHOTS snapshotRetain snapshotRetainTimeUnit
Review Comment:
Can we simplify this to be `WITH SNAPSHOT RETENTION (numSnapshots SNAPSHOTS)? (snapshotRetain snapshotRetainTimeUnit)?`?
##########
spark/v3.3/spark-extensions/src/main/antlr/org.apache.spark.sql.catalyst.parser.extensions/IcebergSqlExtensions.g4:
##########
@@ -73,6 +73,13 @@ statement
| ALTER TABLE multipartIdentifier WRITE writeSpec #setWriteDistributionAndOrdering
| ALTER TABLE multipartIdentifier SET IDENTIFIER_KW FIELDS fieldList #setIdentifierFields
| ALTER TABLE multipartIdentifier DROP IDENTIFIER_KW FIELDS fieldList #dropIdentifierFields
+ | ALTER TABLE multipartIdentifier CREATE BRANCH identifier (AS OF VERSION snapshotId)? (snapshotRetentionClause)? (RETAIN snapshotRefRetain snapshotRefRetainTimeUnit)? #createBranch
+ ;
+
+snapshotRetentionClause
+ : WITH SNAPSHOT RETENTION numSnapshots SNAPSHOTS
+ | WITH SNAPSHOT RETENTION snapshotRetain snapshotRetainTimeUnit
+ | WITH SNAPSHOT RETENTION numSnapshots SNAPSHOTS snapshotRetain snapshotRetainTimeUnit
Review Comment:
Can we simplify these 3 cases to be `WITH SNAPSHOT RETENTION (numSnapshots SNAPSHOTS)? (snapshotRetain snapshotRetainTimeUnit)?`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] hililiwei commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by GitBox <gi...@apache.org>.
hililiwei commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1082056581
##########
spark/v3.3/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/IcebergSqlExtensionsAstBuilder.scala:
##########
@@ -267,6 +285,16 @@ class IcebergSqlExtensionsAstBuilder(delegate: ParserInterface) extends IcebergS
private def typedVisit[T](ctx: ParseTree): T = {
ctx.accept(this).asInstanceOf[T]
}
+
+ private val timeUnit = (unit: String) => {
+ unit.toUpperCase(Locale.ENGLISH) match {
+ case "MONTHS" => 30 * 24 * 60 * 60 * 1000L
Review Comment:
Modified. PTAL. Thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] hililiwei commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by GitBox <gi...@apache.org>.
hililiwei commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1082173399
##########
spark/v3.3/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/IcebergSqlExtensionsAstBuilder.scala:
##########
@@ -267,6 +286,12 @@ class IcebergSqlExtensionsAstBuilder(delegate: ParserInterface) extends IcebergS
private def typedVisit[T](ctx: ParseTree): T = {
ctx.accept(this).asInstanceOf[T]
}
+
+ private def timeRetain(unit: String, duration: Long): Long = unit.toUpperCase(Locale.ENGLISH) match {
+ case "MONTHS" => 30 * TimeUnit.DAYS.toMillis(duration)
Review Comment:
Yes, it does. Updated.
##########
spark/v3.3/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/IcebergSqlExtensionsAstBuilder.scala:
##########
@@ -267,6 +286,12 @@ class IcebergSqlExtensionsAstBuilder(delegate: ParserInterface) extends IcebergS
private def typedVisit[T](ctx: ParseTree): T = {
ctx.accept(this).asInstanceOf[T]
}
+
+ private def timeRetain(unit: String, duration: Long): Long = unit.toUpperCase(Locale.ENGLISH) match {
+ case "MONTHS" => 30 * TimeUnit.DAYS.toMillis(duration)
+ case "DAYS" | "HOURS" | "MINUTES" => TimeUnit.valueOf(unit.toUpperCase(Locale.ENGLISH)).toMillis(duration)
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] hililiwei commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by "hililiwei (via GitHub)" <gi...@apache.org>.
hililiwei commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1084144228
##########
spark/v3.3/spark-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ExtendedDataSourceV2Strategy.scala:
##########
@@ -61,6 +62,9 @@ case class ExtendedDataSourceV2Strategy(spark: SparkSession) extends Strategy wi
case AddPartitionField(IcebergCatalogAndIdentifier(catalog, ident), transform, name) =>
AddPartitionFieldExec(catalog, ident, transform, name) :: Nil
+ case CreateBranch(IcebergCatalogAndIdentifier(catalog, ident), _, _, _, _, _) =>
+ CreateBranchExec(catalog, ident, plan.asInstanceOf[CreateBranch]) :: Nil
Review Comment:
I passed in the `CreateBranch` instance directly to reduce the argument count.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] hililiwei commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by "hililiwei (via GitHub)" <gi...@apache.org>.
hililiwei commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1084152065
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateBranch.java:
##########
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.extensions;
+
+import java.util.IllegalFormatConversionException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.iceberg.AssertHelpers;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.spark.SparkCatalogConfig;
+import org.apache.iceberg.spark.source.SimpleRecord;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
+import org.apache.spark.sql.catalyst.parser.extensions.IcebergParseException;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runners.Parameterized;
+
+public class TestCreateBranch extends SparkExtensionsTestBase {
+
+ @Parameterized.Parameters(name = "catalogName = {0}, implementation = {1}, config = {2}")
+ public static Object[][] parameters() {
+ return new Object[][] {
+ {
+ SparkCatalogConfig.SPARK.catalogName(),
+ SparkCatalogConfig.SPARK.implementation(),
+ SparkCatalogConfig.SPARK.properties()
+ }
+ };
+ }
+
+ public TestCreateBranch(String catalogName, String implementation, Map<String, String> config) {
+ super(catalogName, implementation, config);
+ }
+
+ @After
+ public void removeTable() {
+ sql("DROP TABLE IF EXISTS %s", tableName);
+ }
+
+ @Test
+ public void testCreateBranch() throws NoSuchTableException {
+ Table table = createDefaultTableAndInsert2Row();
+ long snapshotId = table.currentSnapshot().snapshotId();
Review Comment:
Makes sense. updated in #6637.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] jackye1995 merged pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by "jackye1995 (via GitHub)" <gi...@apache.org>.
jackye1995 merged PR #6617:
URL: https://github.com/apache/iceberg/pull/6617
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] jackye1995 commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by GitBox <gi...@apache.org>.
jackye1995 commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1082118109
##########
spark/v3.3/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/IcebergSqlExtensionsAstBuilder.scala:
##########
@@ -267,6 +286,12 @@ class IcebergSqlExtensionsAstBuilder(delegate: ParserInterface) extends IcebergS
private def typedVisit[T](ctx: ParseTree): T = {
ctx.accept(this).asInstanceOf[T]
}
+
+ private def timeRetain(unit: String, duration: Long): Long = unit.toUpperCase(Locale.ENGLISH) match {
+ case "MONTHS" => 30 * TimeUnit.DAYS.toMillis(duration)
Review Comment:
hmm, good point, there is not exactly always 30 days in a month. That's probably why it is not supported by `TimeUnit`. So I would say in that case let's remove MONTHS from the time unit in parser and just support DAYS, HOURS, MINUTES and SECONDS
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] jackye1995 commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by GitBox <gi...@apache.org>.
jackye1995 commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1082010680
##########
spark/v3.3/spark-extensions/src/main/antlr/org.apache.spark.sql.catalyst.parser.extensions/IcebergSqlExtensions.g4:
##########
@@ -168,34 +175,77 @@ fieldList
;
nonReserved
- : ADD | ALTER | AS | ASC | BY | CALL | DESC | DROP | FIELD | FIRST | LAST | NULLS | ORDERED | PARTITION | TABLE | WRITE
- | DISTRIBUTED | LOCALLY | UNORDERED | REPLACE | WITH | IDENTIFIER_KW | FIELDS | SET
+ : ADD | ALTER | AS | ASC | BRANCH | BY | CALL | CREATE | DAYS | DESC | DROP | FIELD | FIRST | HOURS | LAST | NULLS | OF | ORDERED | PARTITION | TABLE | WRITE
+ | DISTRIBUTED | LOCALLY | MINUTES | MONTHS | UNORDERED | REPLACE | RETAIN | VERSION | WITH | IDENTIFIER_KW | FIELDS | SET | SNAPSHOT | SNAPSHOTS
| TRUE | FALSE
| MAP
;
+snapshotId
+ : number
+ ;
+
+numSnapshots
+ : number
+ ;
+
+snapshotRetain
+ : number
+ ;
+
+snapshotRefRetain
+ : number
+ ;
+
+snapshotRefRetainTimeUnit
+ : timeUnit
+ ;
+
+snapshotRetainTimeUnit
+ : timeUnit
+ ;
+
+timeUnit
+ : MONTHS
+ | DAYS
+ | HOURS
+ | MINUTES
Review Comment:
missing SECONDS?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] hililiwei commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by GitBox <gi...@apache.org>.
hililiwei commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1080806243
##########
spark/v3.3/spark-extensions/src/main/antlr/org.apache.spark.sql.catalyst.parser.extensions/IcebergSqlExtensions.g4:
##########
@@ -73,6 +73,13 @@ statement
| ALTER TABLE multipartIdentifier WRITE writeSpec #setWriteDistributionAndOrdering
| ALTER TABLE multipartIdentifier SET IDENTIFIER_KW FIELDS fieldList #setIdentifierFields
| ALTER TABLE multipartIdentifier DROP IDENTIFIER_KW FIELDS fieldList #dropIdentifierFields
+ | ALTER TABLE multipartIdentifier CREATE BRANCH identifier (AS OF VERSION snapshotId)? (snapshotRetentionClause)? (RETAIN snapshotRefRetain snapshotRefRetainTimeUnit)? #createBranch
Review Comment:
Sounds reasonable to me.
##########
spark/v3.3/spark-extensions/src/main/antlr/org.apache.spark.sql.catalyst.parser.extensions/IcebergSqlExtensions.g4:
##########
@@ -73,6 +73,13 @@ statement
| ALTER TABLE multipartIdentifier WRITE writeSpec #setWriteDistributionAndOrdering
| ALTER TABLE multipartIdentifier SET IDENTIFIER_KW FIELDS fieldList #setIdentifierFields
| ALTER TABLE multipartIdentifier DROP IDENTIFIER_KW FIELDS fieldList #dropIdentifierFields
+ | ALTER TABLE multipartIdentifier CREATE BRANCH identifier (AS OF VERSION snapshotId)? (snapshotRetentionClause)? (RETAIN snapshotRefRetain snapshotRefRetainTimeUnit)? #createBranch
+ ;
+
+snapshotRetentionClause
+ : WITH SNAPSHOT RETENTION numSnapshots SNAPSHOTS
+ | WITH SNAPSHOT RETENTION snapshotRetain snapshotRetainTimeUnit
+ | WITH SNAPSHOT RETENTION numSnapshots SNAPSHOTS snapshotRetain snapshotRetainTimeUnit
Review Comment:
This skips illegal statements : `WITH SNAPSHOT RETENTION`
or we can use
`WITH SNAPSHOT RETENTION ((numSnapshots SNAPSHOTS)? (snapshotRetain snapshotRetainTimeUnit)? | numSnapshots SNAPSHOTS snapshotRetain snapshotRetainTimeUnit )`?
But it's not intuitive
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] jackye1995 commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by GitBox <gi...@apache.org>.
jackye1995 commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1073732351
##########
spark/v3.3/spark-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/CreateBranchExec.scala:
##########
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.v2
+
+import org.apache.iceberg.spark.source.SparkTable
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.plans.logical.CreateBranch
+import org.apache.spark.sql.connector.catalog.Identifier
+import org.apache.spark.sql.connector.catalog.TableCatalog
+
+case class CreateBranchExec(
+ catalog: TableCatalog,
+ ident: Identifier,
+ createBranch: CreateBranch) extends LeafV2CommandExec {
+
+ import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
+
+ override lazy val output: Seq[Attribute] = Nil
+
+ override protected def run(): Seq[InternalRow] = {
+ catalog.loadTable(ident) match {
+ case iceberg: SparkTable =>
+
+ val snapshotId = createBranch.snapshotId.getOrElse(iceberg.table.currentSnapshot().snapshotId())
+ iceberg.table.manageSnapshots()
+ .createBranch(createBranch.branch, snapshotId)
+ .setMinSnapshotsToKeep(createBranch.branch, createBranch.numSnapshots.getOrElse(1L).toInt)
Review Comment:
instead of using a hard-coded default, can we only set these config values if they are present? So that the table property default will be used during actual snapshot expiration
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1083572578
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateBranch.java:
##########
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.extensions;
+
+import java.util.IllegalFormatConversionException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.iceberg.AssertHelpers;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.spark.SparkCatalogConfig;
+import org.apache.iceberg.spark.source.SimpleRecord;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
+import org.apache.spark.sql.catalyst.parser.extensions.IcebergParseException;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runners.Parameterized;
+
+public class TestCreateBranch extends SparkExtensionsTestBase {
+
+ @Parameterized.Parameters(name = "catalogName = {0}, implementation = {1}, config = {2}")
+ public static Object[][] parameters() {
+ return new Object[][] {
+ {
+ SparkCatalogConfig.SPARK.catalogName(),
+ SparkCatalogConfig.SPARK.implementation(),
+ SparkCatalogConfig.SPARK.properties()
+ }
+ };
+ }
+
+ public TestCreateBranch(String catalogName, String implementation, Map<String, String> config) {
+ super(catalogName, implementation, config);
+ }
+
+ @After
+ public void removeTable() {
+ sql("DROP TABLE IF EXISTS %s", tableName);
+ }
+
+ @Test
+ public void testCreateBranch() throws NoSuchTableException {
+ Table table = createDefaultTableAndInsert2Row();
+ long snapshotId = table.currentSnapshot().snapshotId();
+ String branchName = "b1";
+ Integer minSnapshotsToKeep = 2;
+ long maxSnapshotAge = 2L;
+ long maxRefAge = 10L;
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s AS OF VERSION %d RETAIN %d DAYS WITH SNAPSHOT RETENTION %d SNAPSHOTS %d days",
+ tableName, branchName, snapshotId, maxRefAge, minSnapshotsToKeep, maxSnapshotAge);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
Review Comment:
This needs an assertion about the snapshot referenced by the branch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] hililiwei commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by "hililiwei (via GitHub)" <gi...@apache.org>.
hililiwei commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1083616874
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateBranch.java:
##########
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.extensions;
+
+import java.util.IllegalFormatConversionException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.iceberg.AssertHelpers;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.spark.SparkCatalogConfig;
+import org.apache.iceberg.spark.source.SimpleRecord;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
+import org.apache.spark.sql.catalyst.parser.extensions.IcebergParseException;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runners.Parameterized;
+
+public class TestCreateBranch extends SparkExtensionsTestBase {
+
+ @Parameterized.Parameters(name = "catalogName = {0}, implementation = {1}, config = {2}")
Review Comment:
yes, we just test `SparkCatalogConfig.SPARK` catalog.
```
@Parameterized.Parameters(name = "catalogName = {0}, implementation = {1}, config = {2}")
public static Object[][] parameters() {
return new Object[][] {
{
SparkCatalogConfig.SPARK.catalogName(),
SparkCatalogConfig.SPARK.implementation(),
SparkCatalogConfig.SPARK.properties()
}
};
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] hililiwei commented on a diff in pull request #6617: Spark: Spark SQL Extensions for create branch
Posted by "hililiwei (via GitHub)" <gi...@apache.org>.
hililiwei commented on code in PR #6617:
URL: https://github.com/apache/iceberg/pull/6617#discussion_r1084151082
##########
spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateBranch.java:
##########
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.extensions;
+
+import java.util.IllegalFormatConversionException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.iceberg.AssertHelpers;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.spark.SparkCatalogConfig;
+import org.apache.iceberg.spark.source.SimpleRecord;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
+import org.apache.spark.sql.catalyst.parser.extensions.IcebergParseException;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runners.Parameterized;
+
+public class TestCreateBranch extends SparkExtensionsTestBase {
+
+ @Parameterized.Parameters(name = "catalogName = {0}, implementation = {1}, config = {2}")
+ public static Object[][] parameters() {
+ return new Object[][] {
+ {
+ SparkCatalogConfig.SPARK.catalogName(),
+ SparkCatalogConfig.SPARK.implementation(),
+ SparkCatalogConfig.SPARK.properties()
+ }
+ };
+ }
+
+ public TestCreateBranch(String catalogName, String implementation, Map<String, String> config) {
+ super(catalogName, implementation, config);
+ }
+
+ @After
+ public void removeTable() {
+ sql("DROP TABLE IF EXISTS %s", tableName);
+ }
+
+ @Test
+ public void testCreateBranch() throws NoSuchTableException {
+ Table table = createDefaultTableAndInsert2Row();
+ long snapshotId = table.currentSnapshot().snapshotId();
+ String branchName = "b1";
+ Integer minSnapshotsToKeep = 2;
+ long maxSnapshotAge = 2L;
+ long maxRefAge = 10L;
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s AS OF VERSION %d RETAIN %d DAYS WITH SNAPSHOT RETENTION %d SNAPSHOTS %d days",
+ tableName, branchName, snapshotId, maxRefAge, minSnapshotsToKeep, maxSnapshotAge);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertEquals(minSnapshotsToKeep, ref.minSnapshotsToKeep());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxSnapshotAge), ref.maxSnapshotAgeMs().longValue());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxRefAge), ref.maxRefAgeMs().longValue());
+
+ AssertHelpers.assertThrows(
+ "Cannot create an existing branch",
+ IllegalArgumentException.class,
+ "already exists",
+ () -> sql("ALTER TABLE %s CREATE BRANCH %s", tableName, branchName));
+ }
+
+ @Test
+ public void testCreateBranchUseDefaultConfig() throws NoSuchTableException {
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql("ALTER TABLE %s CREATE BRANCH %s", tableName, branchName);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertNull(ref.minSnapshotsToKeep());
+ Assert.assertNull(ref.maxSnapshotAgeMs());
+ Assert.assertNull(ref.maxRefAgeMs());
+ }
+
+ @Test
+ public void testCreateBranchUseCustomMinSnapshotsToKeep() throws NoSuchTableException {
+ Integer minSnapshotsToKeep = 2;
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s WITH SNAPSHOT RETENTION %d SNAPSHOTS",
+ tableName, branchName, minSnapshotsToKeep);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertEquals(minSnapshotsToKeep, ref.minSnapshotsToKeep());
+ Assert.assertNull(ref.maxSnapshotAgeMs());
+ Assert.assertNull(ref.maxRefAgeMs());
+ }
+
+ @Test
+ public void testCreateBranchUseCustomMaxSnapshotAge() throws NoSuchTableException {
+ long maxSnapshotAge = 2L;
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s WITH SNAPSHOT RETENTION %d DAYS",
+ tableName, branchName, maxSnapshotAge);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertNull(ref.minSnapshotsToKeep());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxSnapshotAge), ref.maxSnapshotAgeMs().longValue());
+ Assert.assertNull(ref.maxRefAgeMs());
+ }
+
+ @Test
+ public void testCreateBranchUseCustomMinSnapshotsToKeepAndMaxSnapshotAge()
+ throws NoSuchTableException {
+ Integer minSnapshotsToKeep = 2;
+ long maxSnapshotAge = 2L;
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql(
+ "ALTER TABLE %s CREATE BRANCH %s WITH SNAPSHOT RETENTION %d SNAPSHOTS %d DAYS",
+ tableName, branchName, minSnapshotsToKeep, maxSnapshotAge);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertEquals(minSnapshotsToKeep, ref.minSnapshotsToKeep());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxSnapshotAge), ref.maxSnapshotAgeMs().longValue());
+ Assert.assertNull(ref.maxRefAgeMs());
+
+ AssertHelpers.assertThrows(
+ "Illegal statement",
+ IcebergParseException.class,
+ "no viable alternative at input 'WITH SNAPSHOT RETENTION'",
+ () ->
+ sql("ALTER TABLE %s CREATE BRANCH %s WITH SNAPSHOT RETENTION", tableName, branchName));
+ }
+
+ @Test
+ public void testCreateBranchUseCustomMaxRefAge() throws NoSuchTableException {
+ long maxRefAge = 10L;
+ Table table = createDefaultTableAndInsert2Row();
+ String branchName = "b1";
+ sql("ALTER TABLE %s CREATE BRANCH %s RETAIN %d DAYS", tableName, branchName, maxRefAge);
+ table.refresh();
+ SnapshotRef ref = table.refs().get(branchName);
+ Assert.assertNotNull(ref);
+ Assert.assertNull(ref.minSnapshotsToKeep());
+ Assert.assertNull(ref.maxSnapshotAgeMs());
+ Assert.assertEquals(TimeUnit.DAYS.toMillis(maxRefAge), ref.maxRefAgeMs().longValue());
+
+ AssertHelpers.assertThrows(
+ "Illegal statement",
+ IcebergParseException.class,
+ "mismatched input",
+ () -> sql("ALTER TABLE %s CREATE BRANCH %s RETAIN", tableName, branchName));
+
+ AssertHelpers.assertThrows(
+ "Illegal statement",
+ IllegalFormatConversionException.class,
+ "d != java.lang.String",
Review Comment:
fixed in the latest version. #6637
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org