forked from apache/hive
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
HIVE-27235: Iceberg: CREATE TAG SQL implementation (Butao Zhang, revi…
…ewed by Denys Kuzmenko) Closes apache#4372
- Loading branch information
1 parent
21adc21
commit 133a29b
Showing
27 changed files
with
491 additions
and
96 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
55 changes: 55 additions & 0 deletions
55
iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTagExec.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,55 @@ | ||
| /* | ||
| * 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.mr.hive; | ||
|
|
||
| import org.apache.hadoop.hive.ql.parse.AlterTableSnapshotRefSpec; | ||
| import org.apache.iceberg.ManageSnapshots; | ||
| import org.apache.iceberg.Table; | ||
| import org.apache.iceberg.util.SnapshotUtil; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| public class IcebergTagExec { | ||
|
|
||
| private static final Logger LOG = LoggerFactory.getLogger(IcebergTagExec.class); | ||
|
|
||
| private IcebergTagExec() { | ||
| } | ||
|
|
||
| public static void createTag(Table table, AlterTableSnapshotRefSpec.CreateSnapshotRefSpec createTagSpec) { | ||
| String tagName = createTagSpec.getRefName(); | ||
| Long snapshotId = null; | ||
| if (createTagSpec.getSnapshotId() != null) { | ||
| snapshotId = createTagSpec.getSnapshotId(); | ||
| } else if (createTagSpec.getAsOfTime() != null) { | ||
| snapshotId = SnapshotUtil.snapshotIdAsOfTime(table, createTagSpec.getAsOfTime()); | ||
| } else { | ||
| snapshotId = table.currentSnapshot().snapshotId(); | ||
| } | ||
| LOG.info("Creating tag {} on iceberg table {} with snapshotId {}", tagName, table.name(), snapshotId); | ||
| ManageSnapshots manageSnapshots = table.manageSnapshots(); | ||
| manageSnapshots.createTag(tagName, snapshotId); | ||
| if (createTagSpec.getMaxRefAgeMs() != null) { | ||
| manageSnapshots.setMaxRefAgeMs(tagName, createTagSpec.getMaxRefAgeMs()); | ||
| } | ||
|
|
||
| manageSnapshots.commit(); | ||
| } | ||
| } |
135 changes: 135 additions & 0 deletions
135
...iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergTagOperation.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,135 @@ | ||
| /* | ||
| * 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.mr.hive; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.concurrent.TimeUnit; | ||
| import org.apache.iceberg.SnapshotRef; | ||
| import org.apache.iceberg.Table; | ||
| import org.junit.Assert; | ||
| import org.junit.Test; | ||
|
|
||
| import static org.apache.iceberg.mr.hive.HiveIcebergTestUtils.timestampAfterSnapshot; | ||
|
|
||
| public class TestHiveIcebergTagOperation extends HiveIcebergStorageHandlerWithEngineBase { | ||
| @Test | ||
| public void testCreateTagWithDefaultConfig() throws InterruptedException, IOException { | ||
| Table table = | ||
| testTables.createTableWithVersions(shell, "customers", HiveIcebergStorageHandlerTestUtils.CUSTOMER_SCHEMA, | ||
| fileFormat, HiveIcebergStorageHandlerTestUtils.CUSTOMER_RECORDS, 2); | ||
|
|
||
| String tagName = "test_tag_1"; | ||
| shell.executeStatement(String.format("ALTER TABLE customers CREATE TAG %s", tagName)); | ||
| table.refresh(); | ||
| SnapshotRef ref = table.refs().get(tagName); | ||
| Assert.assertEquals(table.currentSnapshot().snapshotId(), ref.snapshotId()); | ||
| Assert.assertNull(ref.maxRefAgeMs()); | ||
|
|
||
| // creating a tag which is already exists will fail | ||
| try { | ||
| shell.executeStatement(String.format("ALTER TABLE customers CREATE TAG %s", tagName)); | ||
| } catch (Throwable e) { | ||
| while (e.getCause() != null) { | ||
| e = e.getCause(); | ||
| } | ||
| Assert.assertTrue(e.getMessage().contains("Ref test_tag_1 already exists")); | ||
| } | ||
| } | ||
|
|
||
| @Test | ||
| public void testCreateTagWithSnapshotId() throws InterruptedException, IOException { | ||
| Table table = | ||
| testTables.createTableWithVersions(shell, "customers", HiveIcebergStorageHandlerTestUtils.CUSTOMER_SCHEMA, | ||
| fileFormat, HiveIcebergStorageHandlerTestUtils.CUSTOMER_RECORDS, 2); | ||
|
|
||
| String tagName = "test_tag_1"; | ||
| Long snapshotId = table.history().get(0).snapshotId(); | ||
| shell.executeStatement(String.format("ALTER TABLE customers CREATE TAG %s FOR SYSTEM_VERSION AS OF %d", | ||
| tagName, snapshotId)); | ||
| table.refresh(); | ||
| SnapshotRef ref = table.refs().get(tagName); | ||
| Assert.assertEquals(snapshotId.longValue(), ref.snapshotId()); | ||
| Assert.assertNull(ref.maxRefAgeMs()); | ||
| } | ||
|
|
||
| @Test | ||
| public void testCreateTagWithTimeStamp() throws InterruptedException, IOException { | ||
| Table table = | ||
| testTables.createTableWithVersions(shell, "customers", HiveIcebergStorageHandlerTestUtils.CUSTOMER_SCHEMA, | ||
| fileFormat, HiveIcebergStorageHandlerTestUtils.CUSTOMER_RECORDS, 2); | ||
|
|
||
| String tagName = "test_tag_1"; | ||
| Long snapshotId = table.history().get(0).snapshotId(); | ||
|
|
||
| shell.executeStatement(String.format("ALTER TABLE customers CREATE TAG %s FOR SYSTEM_TIME AS OF '%s'", | ||
| tagName, timestampAfterSnapshot(table, 0))); | ||
| table.refresh(); | ||
| SnapshotRef ref = table.refs().get(tagName); | ||
| Assert.assertEquals(snapshotId.longValue(), ref.snapshotId()); | ||
| } | ||
|
|
||
| @Test | ||
| public void testCreateTagWithMaxRefAge() throws InterruptedException, IOException { | ||
| Table table = | ||
| testTables.createTableWithVersions(shell, "customers", HiveIcebergStorageHandlerTestUtils.CUSTOMER_SCHEMA, | ||
| fileFormat, HiveIcebergStorageHandlerTestUtils.CUSTOMER_RECORDS, 2); | ||
|
|
||
| String tagName = "test_tag_1"; | ||
| long maxRefAge = 5L; | ||
| shell.executeStatement(String.format("ALTER TABLE customers CREATE TAG %s RETAIN %d DAYS", tagName, maxRefAge)); | ||
| table.refresh(); | ||
| SnapshotRef ref = table.refs().get(tagName); | ||
| Assert.assertEquals(table.currentSnapshot().snapshotId(), ref.snapshotId()); | ||
| Assert.assertEquals(TimeUnit.DAYS.toMillis(maxRefAge), ref.maxRefAgeMs().longValue()); | ||
| } | ||
|
|
||
| @Test | ||
| public void testCreateTagWithAllCustomConfig() throws IOException, InterruptedException { | ||
| Table table = | ||
| testTables.createTableWithVersions(shell, "customers", HiveIcebergStorageHandlerTestUtils.CUSTOMER_SCHEMA, | ||
| fileFormat, HiveIcebergStorageHandlerTestUtils.CUSTOMER_RECORDS, 2); | ||
|
|
||
| String tagName = "test_tag_1"; | ||
| Long snapshotId = table.history().get(0).snapshotId(); | ||
| long maxRefAge = 5L; | ||
| shell.executeStatement(String.format("ALTER TABLE customers CREATE TAG %s FOR SYSTEM_VERSION AS OF %d RETAIN" + | ||
| " %d DAYS", | ||
| tagName, snapshotId, maxRefAge)); | ||
| table.refresh(); | ||
| SnapshotRef ref = table.refs().get(tagName); | ||
| Assert.assertEquals(snapshotId.longValue(), ref.snapshotId()); | ||
| Assert.assertEquals(TimeUnit.DAYS.toMillis(maxRefAge), ref.maxRefAgeMs().longValue()); | ||
| } | ||
|
|
||
| @Test | ||
| public void testCreateTagWithNonIcebergTable() { | ||
| shell.executeStatement("create table nonice_tbl (id int, name string)"); | ||
|
|
||
| String tagName = "test_tag_1"; | ||
| try { | ||
| shell.executeStatement(String.format("ALTER TABLE nonice_tbl CREATE TAG %s", tagName)); | ||
| } catch (Throwable e) { | ||
| while (e.getCause() != null) { | ||
| e = e.getCause(); | ||
| } | ||
| Assert.assertTrue(e.getMessage().contains("Not an iceberg table")); | ||
| } | ||
| } | ||
| } |
3 changes: 3 additions & 0 deletions
3
iceberg/iceberg-handler/src/test/queries/negative/alter_table_create_tag_negative.q
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,3 @@ | ||
| create table ice_tbl (id int, name string) Stored by Iceberg; | ||
|
|
||
| alter table ice_tbl create tag test_branch_1; |
19 changes: 19 additions & 0 deletions
19
iceberg/iceberg-handler/src/test/queries/positive/alter_table_create_tag.q
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,19 @@ | ||
| -- SORT_QUERY_RESULTS | ||
| set hive.explain.user=false; | ||
|
|
||
| create table iceTbl (id int, name string) Stored by Iceberg; | ||
|
|
||
| -- creating tag requires table to have current snapshot. here insert some values to generate current snapshot | ||
| insert into iceTbl values(1, 'jack'); | ||
|
|
||
| -- create tag with default values based on the current snapshotId | ||
| explain alter table iceTbl create tag test_tag_1; | ||
| alter table iceTbl create tag test_tag_1; | ||
| select name, max_reference_age_in_ms from default.iceTbl.refs where type='TAG'; | ||
|
|
||
| -- create a tag which could be retained 5 days based on the current snapshotId | ||
| insert into iceTbl values(2, 'bob'); | ||
| explain alter table iceTbl create tag test_tag_2 retain 5 days; | ||
| alter table iceTbl create tag test_tag_2 retain 5 days; | ||
| select name, max_reference_age_in_ms from default.iceTbl.refs where type='TAG'; | ||
|
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
12 changes: 12 additions & 0 deletions
12
iceberg/iceberg-handler/src/test/results/negative/alter_table_create_tag_negative.q.out
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,12 @@ | ||
| PREHOOK: query: create table ice_tbl (id int, name string) Stored by Iceberg | ||
| PREHOOK: type: CREATETABLE | ||
| PREHOOK: Output: database:default | ||
| PREHOOK: Output: default@ice_tbl | ||
| POSTHOOK: query: create table ice_tbl (id int, name string) Stored by Iceberg | ||
| POSTHOOK: type: CREATETABLE | ||
| POSTHOOK: Output: database:default | ||
| POSTHOOK: Output: default@ice_tbl | ||
| PREHOOK: query: alter table ice_tbl create tag test_branch_1 | ||
| PREHOOK: type: ALTERTABLE_CREATETAG | ||
| PREHOOK: Input: default@ice_tbl | ||
| FAILED: Execution Error, return code 40000 from org.apache.hadoop.hive.ql.ddl.DDLTask. java.lang.UnsupportedOperationException: Cannot alter create tag on iceberg table default.ice_tbl which has no snapshot |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.