You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by lz...@apache.org on 2022/12/02 10:33:26 UTC
[flink-table-store] branch master updated: [FLINK-30269] Validate table name for metadata table
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-table-store.git
The following commit(s) were added to refs/heads/master by this push:
new 2c8d043a [FLINK-30269] Validate table name for metadata table
2c8d043a is described below
commit 2c8d043acc7e24e9b8f0cbf86beea62002351bf7
Author: shammon <zj...@gmail.com>
AuthorDate: Fri Dec 2 18:33:20 2022 +0800
[FLINK-30269] Validate table name for metadata table
This closes #415
---
.../flink/table/store/connector/CatalogTableITCase.java | 15 +++++++++++++++
.../flink/table/store/file/catalog/AbstractCatalog.java | 6 ++++++
2 files changed, 21 insertions(+)
diff --git a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/CatalogTableITCase.java b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/CatalogTableITCase.java
index 7d9f97b2..d3d6c7e3 100644
--- a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/CatalogTableITCase.java
+++ b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/CatalogTableITCase.java
@@ -24,6 +24,7 @@ import org.junit.Test;
import java.util.List;
+import static org.apache.flink.table.store.file.catalog.Catalog.METADATA_TABLE_SPLITTER;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -55,4 +56,18 @@ public class CatalogTableITCase extends CatalogITCaseBase {
List<Row> result = sql("SELECT * FROM T$options");
assertThat(result).containsExactly(Row.of("snapshot.time-retained", "5 h"));
}
+
+ @Test
+ public void testCreateMetaTable() {
+ assertThatThrownBy(() -> sql("CREATE TABLE T$snapshots (a INT, b INT)"))
+ .hasRootCauseMessage(
+ String.format(
+ "Table name[%s] cannot contain '%s' separator",
+ "T$snapshots", METADATA_TABLE_SPLITTER));
+ assertThatThrownBy(() -> sql("CREATE TABLE T$aa$bb (a INT, b INT)"))
+ .hasRootCauseMessage(
+ String.format(
+ "Table name[%s] cannot contain '%s' separator",
+ "T$aa$bb", METADATA_TABLE_SPLITTER));
+ }
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/catalog/AbstractCatalog.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/catalog/AbstractCatalog.java
index 5b07fc78..6cbbdcf0 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/catalog/AbstractCatalog.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/catalog/AbstractCatalog.java
@@ -34,6 +34,12 @@ public abstract class AbstractCatalog implements Catalog {
@Override
public Path getTableLocation(ObjectPath tablePath) {
+ if (tablePath.getObjectName().contains(METADATA_TABLE_SPLITTER)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Table name[%s] cannot contain '%s' separator",
+ tablePath.getObjectName(), METADATA_TABLE_SPLITTER));
+ }
return new Path(databasePath(tablePath.getDatabaseName()), tablePath.getObjectName());
}