This is an automated email from the ASF dual-hosted git repository. czweng pushed a commit to branch release-0.3 in repository https://gitbox.apache.org/repos/asf/flink-table-store.git
The following commit(s) were added to refs/heads/release-0.3 by this push: new 8bcec69b [FLINK-30608] support rename table 8bcec69b is described below commit 8bcec69b5f9d0bd6a270889e8e7947270c0e3c44 Author: tsreaper <tsreape...@gmail.com> AuthorDate: Mon Mar 6 14:45:57 2023 +0800 [FLINK-30608] support rename table This closes #572. --- docs/content/docs/how-to/altering-tables.md | 30 +++++++++++++ .../flink/table/store/connector/FlinkCatalog.java | 11 ++++- .../store/connector/FileSystemCatalogITCase.java | 52 +++++++++++++++++++++- .../flink/table/store/file/catalog/Catalog.java | 13 ++++++ .../store/file/catalog/FileSystemCatalog.java | 20 +++++++++ .../apache/flink/table/store/hive/HiveCatalog.java | 27 +++++++++++ .../flink/table/store/hive/HiveCatalogITCase.java | 35 +++++++++++++++ .../flink/table/store/spark/SparkCatalog.java | 11 ++++- .../flink/table/store/spark/SparkReadTestBase.java | 15 +++++++ .../store/spark/SparkSchemaEvolutionITCase.java | 34 ++++++++++++-- 10 files changed, 239 insertions(+), 9 deletions(-) diff --git a/docs/content/docs/how-to/altering-tables.md b/docs/content/docs/how-to/altering-tables.md index 48048872..090860c0 100644 --- a/docs/content/docs/how-to/altering-tables.md +++ b/docs/content/docs/how-to/altering-tables.md @@ -54,6 +54,36 @@ ALTER TABLE my_table SET TBLPROPERTIES ( {{< /tabs >}} +## Rename Table Name + +The following SQL rename the table name to new name. + +{{< tabs "rename-table-name" >}} + +{{< tab "Flink" >}} + +```sql +ALTER TABLE my_table RENAME TO my_table_new; +``` + +{{< /tab >}} + +{{< tab "Spark3" >}} + +```sql +ALTER TABLE my_table RENAME TO my_table_new; +``` + +{{< /tab >}} + +{{< /tabs >}} + +{{< hint info >}} + +If you use object storage, such as S3 or OSS, please use this syntax carefully, because the renaming of object storage is not atomic, and only partial files may be moved in case of failure. + +{{< /hint >}} + ## Removing Table Properties The following SQL removes `write-buffer-size` table property. diff --git a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/FlinkCatalog.java b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/FlinkCatalog.java index 55e5f30e..9769e868 100644 --- a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/FlinkCatalog.java +++ b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/FlinkCatalog.java @@ -314,8 +314,15 @@ public class FlinkCatalog extends AbstractCatalog { @Override public final void renameTable( ObjectPath tablePath, String newTableName, boolean ignoreIfNotExists) - throws CatalogException { - throw new UnsupportedOperationException(); + throws CatalogException, TableNotExistException, TableAlreadyExistException { + ObjectPath toTable = new ObjectPath(tablePath.getDatabaseName(), newTableName); + try { + catalog.renameTable(tablePath, toTable, ignoreIfNotExists); + } catch (Catalog.TableNotExistException e) { + throw new TableNotExistException(getName(), tablePath); + } catch (Catalog.TableAlreadyExistException e) { + throw new TableAlreadyExistException(getName(), toTable); + } } @Override diff --git a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/FileSystemCatalogITCase.java b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/FileSystemCatalogITCase.java index 69ee10c2..932511e9 100644 --- a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/FileSystemCatalogITCase.java +++ b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/FileSystemCatalogITCase.java @@ -18,25 +18,37 @@ package org.apache.flink.table.store.connector; +import org.apache.flink.core.fs.Path; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.store.file.catalog.Catalog; import org.apache.flink.table.store.file.utils.BlockingIterator; import org.apache.flink.table.store.kafka.KafkaTableTestBase; import org.apache.flink.types.Row; +import org.apache.flink.util.CloseableIterator; +import org.junit.Assert; import org.junit.Before; import org.junit.Test; +import java.io.File; import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import java.util.UUID; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy; /** ITCase for {@link FlinkCatalog}. */ public class FileSystemCatalogITCase extends KafkaTableTestBase { + private String path; + private static final String DB_NAME = "default"; + @Before public void before() throws IOException { - String path = TEMPORARY_FOLDER.newFolder().toURI().toString(); + path = TEMPORARY_FOLDER.newFolder().toURI().toString(); tEnv.executeSql( String.format( "CREATE CATALOG fs WITH ('type'='table-store', 'warehouse'='%s')", path)); @@ -50,6 +62,34 @@ public class FileSystemCatalogITCase extends KafkaTableTestBase { innerTestWriteRead(); } + @Test + public void testRenameTable() throws Exception { + tEnv.executeSql("CREATE TABLE t1 (a INT)").await(); + tEnv.executeSql("CREATE TABLE t2 (a INT)").await(); + tEnv.executeSql("INSERT INTO t1 VALUES(1),(2)").await(); + // the source table do not exist. + assertThatThrownBy(() -> tEnv.executeSql("ALTER TABLE t3 RENAME TO t4")) + .hasMessage("Table `fs`.`default`.`t3` doesn't exist or is a temporary table."); + + // the target table has existed. + assertThatThrownBy(() -> tEnv.executeSql("ALTER TABLE t1 RENAME TO t2")) + .hasMessage("Could not execute ALTER TABLE fs.default.t1 RENAME TO fs.default.t2"); + + tEnv.executeSql("ALTER TABLE t1 RENAME TO t3").await(); + Assert.assertEquals(Arrays.asList(Row.of("t2"), Row.of("t3")), collect("SHOW TABLES")); + + ObjectPath objectPath = new ObjectPath(DB_NAME, "t3"); + Catalog catalog = + ((FlinkCatalog) tEnv.getCatalog(tEnv.getCurrentCatalog()).get()).catalog(); + Path tablePath = catalog.getTableLocation(objectPath); + Assert.assertEquals(tablePath.toString(), path + DB_NAME + ".db" + File.separator + "t3"); + + BlockingIterator<Row, Row> iterator = + BlockingIterator.of(tEnv.from("t3").execute().collect()); + List<Row> result = iterator.collectAndClose(2); + assertThat(result).containsExactlyInAnyOrder(Row.of(1), Row.of(2)); + } + @Test public void testLogWriteRead() throws Exception { String topic = UUID.randomUUID().toString(); @@ -107,4 +147,14 @@ public class FileSystemCatalogITCase extends KafkaTableTestBase { List<Row> result = iterator.collectAndClose(2); assertThat(result).containsExactlyInAnyOrder(Row.of("1", "2", "3"), Row.of("4", "5", "6")); } + + private List<Row> collect(String sql) throws Exception { + List<Row> result = new ArrayList<>(); + try (CloseableIterator<Row> it = tEnv.executeSql(sql).collect()) { + while (it.hasNext()) { + result.add(it.next()); + } + } + return result; + } } diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/catalog/Catalog.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/catalog/Catalog.java index 919f6067..3789187a 100644 --- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/catalog/Catalog.java +++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/catalog/Catalog.java @@ -150,6 +150,19 @@ public interface Catalog extends AutoCloseable { void createTable(ObjectPath tablePath, UpdateSchema tableSchema, boolean ignoreIfExists) throws TableAlreadyExistException, DatabaseNotExistException; + /** + * Rename a table. + * + * @param fromTable the name of the table which need to rename + * @param toTable the new table + * @param ignoreIfNotExists Flag to specify behavior when the table does not exist: if set to + * false, throw an exception, if set to true, do nothing. + * @throws TableNotExistException if the from table does not exist + * @throws TableAlreadyExistException if the to table already exists + */ + void renameTable(ObjectPath fromTable, ObjectPath toTable, boolean ignoreIfNotExists) + throws TableNotExistException, TableAlreadyExistException; + /** * Modify an existing table from {@link SchemaChange}s. * diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/catalog/FileSystemCatalog.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/catalog/FileSystemCatalog.java index 6dae077e..c5a9a80a 100644 --- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/catalog/FileSystemCatalog.java +++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/catalog/FileSystemCatalog.java @@ -163,6 +163,26 @@ public class FileSystemCatalog extends AbstractCatalog { uncheck(() -> new SchemaManager(path).commitNewVersion(table)); } + @Override + public void renameTable(ObjectPath fromTable, ObjectPath toTable, boolean ignoreIfNotExists) + throws TableNotExistException, TableAlreadyExistException { + Path fromPath = getTableLocation(fromTable); + if (!tableExists(fromPath)) { + if (ignoreIfNotExists) { + return; + } + + throw new TableNotExistException(fromTable); + } + + Path toPath = getTableLocation(toTable); + if (tableExists(toPath)) { + throw new TableAlreadyExistException(toTable); + } + + uncheck(() -> fs.rename(fromPath, toPath)); + } + @Override public void alterTable( ObjectPath tablePath, List<SchemaChange> changes, boolean ignoreIfNotExists) diff --git a/flink-table-store-hive/flink-table-store-hive-catalog/src/main/java/org/apache/flink/table/store/hive/HiveCatalog.java b/flink-table-store-hive/flink-table-store-hive-catalog/src/main/java/org/apache/flink/table/store/hive/HiveCatalog.java index 90269ee4..e9db107c 100644 --- a/flink-table-store-hive/flink-table-store-hive-catalog/src/main/java/org/apache/flink/table/store/hive/HiveCatalog.java +++ b/flink-table-store-hive/flink-table-store-hive-catalog/src/main/java/org/apache/flink/table/store/hive/HiveCatalog.java @@ -239,6 +239,33 @@ public class HiveCatalog extends AbstractCatalog { } } + @Override + public void renameTable(ObjectPath fromTable, ObjectPath toTable, boolean ignoreIfNotExists) + throws TableNotExistException, TableAlreadyExistException { + if (!tableStoreTableExists(fromTable)) { + if (ignoreIfNotExists) { + return; + } else { + throw new TableNotExistException(fromTable); + } + } + + if (tableExists(toTable)) { + throw new TableAlreadyExistException(toTable); + } + + try { + String fromDB = fromTable.getDatabaseName(); + String fromTableName = fromTable.getObjectName(); + Table table = client.getTable(fromDB, fromTableName); + table.setDbName(toTable.getDatabaseName()); + table.setTableName(toTable.getObjectName()); + client.alter_table(fromDB, fromTableName, table); + } catch (TException e) { + throw new RuntimeException("Failed to rename table " + fromTable.getFullName(), e); + } + } + @Override public void alterTable( ObjectPath tablePath, List<SchemaChange> changes, boolean ignoreIfNotExists) diff --git a/flink-table-store-hive/flink-table-store-hive-catalog/src/test/java/org/apache/flink/table/store/hive/HiveCatalogITCase.java b/flink-table-store-hive/flink-table-store-hive-catalog/src/test/java/org/apache/flink/table/store/hive/HiveCatalogITCase.java index 3ac047b4..0cde2fe7 100644 --- a/flink-table-store-hive/flink-table-store-hive-catalog/src/test/java/org/apache/flink/table/store/hive/HiveCatalogITCase.java +++ b/flink-table-store-hive/flink-table-store-hive-catalog/src/test/java/org/apache/flink/table/store/hive/HiveCatalogITCase.java @@ -23,7 +23,9 @@ import org.apache.flink.core.fs.Path; import org.apache.flink.table.api.EnvironmentSettings; import org.apache.flink.table.api.TableEnvironment; import org.apache.flink.table.api.internal.TableEnvironmentImpl; +import org.apache.flink.table.catalog.ObjectPath; import org.apache.flink.table.store.connector.FlinkCatalog; +import org.apache.flink.table.store.file.catalog.Catalog; import org.apache.flink.table.store.file.catalog.CatalogLock; import org.apache.flink.types.Row; import org.apache.flink.util.CloseableIterator; @@ -42,6 +44,7 @@ import org.junit.Test; import org.junit.rules.TemporaryFolder; import org.junit.runner.RunWith; +import java.io.File; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -276,6 +279,38 @@ public class HiveCatalogITCase { } } + @Test + public void testRenameTable() throws Exception { + tEnv.executeSql("CREATE TABLE t1 (a INT)").await(); + tEnv.executeSql("CREATE TABLE t2 (a INT)").await(); + tEnv.executeSql("INSERT INTO t1 SELECT 1"); + // the source table do not exist. + assertThatThrownBy(() -> tEnv.executeSql("ALTER TABLE t3 RENAME TO t4")) + .hasMessage( + "Table `my_hive`.`test_db`.`t3` doesn't exist or is a temporary table."); + + // the target table has existed. + assertThatThrownBy(() -> tEnv.executeSql("ALTER TABLE t1 RENAME TO t2")) + .hasMessage( + "Could not execute ALTER TABLE my_hive.test_db.t1 RENAME TO my_hive.test_db.t2"); + + tEnv.executeSql("ALTER TABLE t1 RENAME TO t3").await(); + List<String> tables = hiveShell.executeQuery("SHOW TABLES"); + Assert.assertTrue(tables.contains("t3")); + Assert.assertFalse(tables.contains("t1")); + + ObjectPath objectPath = new ObjectPath("test_db", "t3"); + Catalog catalog = + ((FlinkCatalog) tEnv.getCatalog(tEnv.getCurrentCatalog()).get()).catalog(); + org.apache.flink.core.fs.Path tablePath = catalog.getTableLocation(objectPath); + Assert.assertEquals(tablePath.toString(), path + "test_db.db" + File.separator + "t3"); + + // TODO: the hiverunner (4.0) has a bug ,it can not rename the table path correctly , + // we should upgrade it to the 6.0 later ,and update the test case for query. + assertThatThrownBy(() -> tEnv.executeSql("SELECT * FROM t3")) + .hasMessageContaining("SQL validation failed. There is no table stored in"); + } + @Test public void testHiveLock() throws InterruptedException { tEnv.executeSql("CREATE TABLE t (a INT)"); diff --git a/flink-table-store-spark/src/main/java/org/apache/flink/table/store/spark/SparkCatalog.java b/flink-table-store-spark/src/main/java/org/apache/flink/table/store/spark/SparkCatalog.java index 8c925f4f..98d4e49b 100644 --- a/flink-table-store-spark/src/main/java/org/apache/flink/table/store/spark/SparkCatalog.java +++ b/flink-table-store-spark/src/main/java/org/apache/flink/table/store/spark/SparkCatalog.java @@ -357,7 +357,14 @@ public class SparkCatalog implements TableCatalog, SupportsNamespaces { } @Override - public void renameTable(Identifier oldIdent, Identifier newIdent) { - throw new UnsupportedOperationException(); + public void renameTable(Identifier oldIdent, Identifier newIdent) + throws NoSuchTableException, TableAlreadyExistsException { + try { + catalog.renameTable(objectPath(oldIdent), objectPath(newIdent), false); + } catch (Catalog.TableNotExistException e) { + throw new NoSuchTableException(oldIdent); + } catch (Catalog.TableAlreadyExistException e) { + throw new TableAlreadyExistsException(newIdent); + } } } diff --git a/flink-table-store-spark/src/test/java/org/apache/flink/table/store/spark/SparkReadTestBase.java b/flink-table-store-spark/src/test/java/org/apache/flink/table/store/spark/SparkReadTestBase.java index 15ec0c6f..99e1af9d 100644 --- a/flink-table-store-spark/src/test/java/org/apache/flink/table/store/spark/SparkReadTestBase.java +++ b/flink-table-store-spark/src/test/java/org/apache/flink/table/store/spark/SparkReadTestBase.java @@ -40,7 +40,9 @@ import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Row; import org.apache.spark.sql.SparkSession; import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; import java.io.File; import java.io.IOException; @@ -71,7 +73,11 @@ public abstract class SparkReadTestBase { spark = SparkSession.builder().master("local[2]").getOrCreate(); spark.conf().set("spark.sql.catalog.tablestore", SparkCatalog.class.getName()); spark.conf().set("spark.sql.catalog.tablestore.warehouse", warehousePath.toString()); + spark.sql("USE tablestore"); + } + @BeforeEach + public void beforeEach() throws Exception { // flink sink tablePath1 = new Path(warehousePath, "default.db/t1"); SimpleTableTestHelper testHelper1 = new SimpleTableTestHelper(tablePath1, rowType1()); @@ -129,6 +135,15 @@ public abstract class SparkReadTestBase { testHelper2.commit(); } + @AfterEach + public void afterEach() { + List<Row> tables = spark.sql("show tables").collectAsList(); + tables.forEach( + table -> { + spark.sql("DROP TABLE " + table.getString(0) + "." + table.getString(1)); + }); + } + protected static SimpleTableTestHelper createTestHelper(Path tablePath) throws Exception { RowType rowType = new RowType( diff --git a/flink-table-store-spark/src/test/java/org/apache/flink/table/store/spark/SparkSchemaEvolutionITCase.java b/flink-table-store-spark/src/test/java/org/apache/flink/table/store/spark/SparkSchemaEvolutionITCase.java index 22a096dd..d75956a5 100644 --- a/flink-table-store-spark/src/test/java/org/apache/flink/table/store/spark/SparkSchemaEvolutionITCase.java +++ b/flink-table-store-spark/src/test/java/org/apache/flink/table/store/spark/SparkSchemaEvolutionITCase.java @@ -105,6 +105,36 @@ public class SparkSchemaEvolutionITCase extends SparkReadTestBase { "java.lang.IllegalArgumentException: ADD COLUMN cannot specify NOT NULL."); } + @Test + public void testRenameTable() { + assertThatThrownBy(() -> spark.sql("ALTER TABLE t3 RENAME TO t4")) + .isInstanceOf(AnalysisException.class) + .hasMessageContaining("Table or view not found: t3"); + + assertThatThrownBy(() -> spark.sql("ALTER TABLE t1 RENAME TO t2")) + .isInstanceOf(AnalysisException.class) + .hasMessageContaining("Table default.t2 already exists"); + + spark.sql("ALTER TABLE t1 RENAME TO t3"); + List<Row> tables = spark.sql("SHOW TABLES").collectAsList(); + assertThat(tables.stream().map(Row::toString)) + .containsExactlyInAnyOrder("[default,t2,false]", "[default,t3,false]"); + + List<Row> afterRename = + spark.sql("SHOW CREATE TABLE tablestore.default.t3").collectAsList(); + assertThat(afterRename.toString()) + .isEqualTo( + "[[CREATE TABLE t3 (\n" + + " `a` INT NOT NULL,\n" + + " `b` BIGINT,\n" + + " `c` STRING)\n" + + buildTableProperties("default.db/t3") + + "]]"); + + List<Row> data = spark.sql("SELECT * FROM t3").collectAsList(); + assertThat(data.toString()).isEqualTo("[[1,2,1], [5,6,3]]"); + } + @Test public void testRenameColumn() throws Exception { Path tablePath = new Path(warehousePath, "default.db/testRenameColumn"); @@ -149,7 +179,6 @@ public class SparkSchemaEvolutionITCase extends SparkReadTestBase { @Test public void testRenamePartitionKey() { - spark.sql("USE tablestore"); spark.sql( "CREATE TABLE default.testRenamePartitionKey (\n" + "a BIGINT,\n" @@ -255,7 +284,6 @@ public class SparkSchemaEvolutionITCase extends SparkReadTestBase { @Test public void testDropPartitionKey() { - spark.sql("USE tablestore"); spark.sql( "CREATE TABLE default.testDropPartitionKey (\n" + "a BIGINT,\n" @@ -295,7 +323,6 @@ public class SparkSchemaEvolutionITCase extends SparkReadTestBase { @Test public void testDropPrimaryKey() { - spark.sql("USE tablestore"); spark.sql( "CREATE TABLE default.testDropPrimaryKey (\n" + "a BIGINT,\n" @@ -385,7 +412,6 @@ public class SparkSchemaEvolutionITCase extends SparkReadTestBase { @Test public void testAlterPrimaryKeyNullability() { - spark.sql("USE tablestore"); spark.sql( "CREATE TABLE default.testAlterPkNullability (\n" + "a BIGINT,\n"