apache · mchades · Apr 23, 2024 · Apr 23, 2024
diff --git a/api/src/main/java/com/datastrato/gravitino/exceptions/GravitinoRuntimeException.java b/api/src/main/java/com/datastrato/gravitino/exceptions/GravitinoRuntimeException.java
@@ -27,7 +27,7 @@ public GravitinoRuntimeException(String message) {
    */
   @FormatMethod
   public GravitinoRuntimeException(@FormatString String message, Object... args) {
-    super(String.format(message, args));
+    super(args.length == 0 ? message : String.format(message, args));
   }
 
   /**

diff --git a/...c/test/java/com/datastrato/gravitino/catalog/hadoop/integration/test/HadoopCatalogIT.java b/...c/test/java/com/datastrato/gravitino/catalog/hadoop/integration/test/HadoopCatalogIT.java
@@ -231,6 +231,17 @@ public void testExternalFileset() throws IOException {
         "Should throw IllegalArgumentException when storage location is null");
   }
 
+  @Test
+  void testNameSpec() {
+    String illegalName = "/%~?*";
+
+    NameIdentifier nameIdentifier =
+        NameIdentifier.of(metalakeName, catalogName, schemaName, illegalName);
+
+    Assertions.assertThrows(
+        NoSuchFilesetException.class, () -> catalog.asFilesetCatalog().loadFileset(nameIdentifier));
+  }
+
   @Test
   public void testLoadFileset() throws IOException {
     // create fileset

diff --git a/...src/test/java/com/datastrato/gravitino/catalog/mysql/integration/test/CatalogMysqlIT.java b/...src/test/java/com/datastrato/gravitino/catalog/mysql/integration/test/CatalogMysqlIT.java
@@ -1248,6 +1248,47 @@ void testMySQLTableNameCaseSensitive() {
         "TABLENAME", table_comment, Arrays.asList(newColumns), properties, indexes, table);
   }
 
+  @Test
+  void testNameSpec() {
+    // test operate illegal schema name from MySQL
+    String testSchemaName = "//";
+    String sql = String.format("CREATE DATABASE `%s`", testSchemaName);
+    mysqlService.executeQuery(sql);
+
+    NameIdentifier schemaIdent = NameIdentifier.of(metalakeName, catalogName, testSchemaName);
+    Schema schema = catalog.asSchemas().loadSchema(schemaIdent);
+    Assertions.assertEquals(testSchemaName, schema.name());
+
+    NameIdentifier[] schemaIdents =
+        catalog.asSchemas().listSchemas(Namespace.of(metalakeName, catalogName));
+    Assertions.assertTrue(
+        Arrays.stream(schemaIdents).anyMatch(s -> s.name().equals(testSchemaName)));
+
+    Assertions.assertTrue(catalog.asSchemas().dropSchema(schemaIdent, false));
+    Assertions.assertFalse(catalog.asSchemas().schemaExists(schemaIdent));
+
+    // test operate illegal table name from MySQL
+    mysqlService.executeQuery(sql);
+    String testTableName = "//";
+    sql = String.format("CREATE TABLE `%s`.`%s` (id int)", testSchemaName, testTableName);
+    mysqlService.executeQuery(sql);
+    NameIdentifier tableIdent =
+        NameIdentifier.of(metalakeName, catalogName, testSchemaName, testTableName);
+
+    Table table = catalog.asTableCatalog().loadTable(tableIdent);
+    Assertions.assertEquals(testTableName, table.name());
+
+    NameIdentifier[] tableIdents =
+        catalog
+            .asTableCatalog()
+            .listTables(Namespace.of(metalakeName, catalogName, testSchemaName));
+    Assertions.assertTrue(Arrays.stream(tableIdents).anyMatch(t -> t.name().equals(testTableName)));
+
+    Assertions.assertTrue(catalog.asTableCatalog().dropTable(tableIdent));
+    Assertions.assertFalse(catalog.asTableCatalog().tableExists(tableIdent));
+    Assertions.assertFalse(catalog.asTableCatalog().purgeTable(tableIdent));
+  }
+
   @Test
   void testMySQLSchemaNameCaseSensitive() {
     Column col1 = Column.of("col_1", Types.LongType.get(), "id", false, false, null);

diff --git a/...src/test/java/com/datastrato/gravitino/catalog/kafka/integration/test/CatalogKafkaIT.java b/...src/test/java/com/datastrato/gravitino/catalog/kafka/integration/test/CatalogKafkaIT.java
@@ -26,6 +26,7 @@
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
 import java.time.Duration;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
@@ -358,6 +359,37 @@ public void testDropTopic() throws ExecutionException, InterruptedException {
         "Topic should not exist after dropping");
   }
 
+  @Test
+  public void testNameSpec() throws ExecutionException, InterruptedException {
+    // create topic in Kafka with special characters
+    String illegalName = "test.topic";
+    adminClient.createTopics(ImmutableList.of(new NewTopic(illegalName, 1, (short) 1))).all().get();
+
+    NameIdentifier ident =
+        NameIdentifier.of(METALAKE_NAME, CATALOG_NAME, DEFAULT_SCHEMA_NAME, illegalName);
+    IllegalArgumentException exception =
+        Assertions.assertThrows(
+            IllegalArgumentException.class,
+            () ->
+                catalog
+                    .asTopicCatalog()
+                    .createTopic(ident, "comment", null, Collections.emptyMap()));
+    Assertions.assertTrue(exception.getMessage().contains("Illegal name: test.topic"));
+
+    Topic loadedTopic = catalog.asTopicCatalog().loadTopic(ident);
+    Assertions.assertEquals(illegalName, loadedTopic.name());
+
+    NameIdentifier[] topics =
+        catalog
+            .asTopicCatalog()
+            .listTopics(Namespace.ofTopic(METALAKE_NAME, CATALOG_NAME, DEFAULT_SCHEMA_NAME));
+    Assertions.assertTrue(
+        Arrays.stream(topics).anyMatch(topic -> topic.name().equals(illegalName)));
+
+    Assertions.assertTrue(catalog.asTopicCatalog().dropTopic(ident));
+    Assertions.assertFalse(catalog.asTopicCatalog().topicExists(ident));
+  }
+
   private void assertTopicWithKafka(Topic createdTopic)
       throws ExecutionException, InterruptedException {
     // get topic from Kafka directly

diff --git a/clients/client-java/src/main/java/com/datastrato/gravitino/client/BaseSchemaCatalog.java b/clients/client-java/src/main/java/com/datastrato/gravitino/client/BaseSchemaCatalog.java
@@ -21,6 +21,7 @@
 import com.datastrato.gravitino.rel.Schema;
 import com.datastrato.gravitino.rel.SchemaChange;
 import com.datastrato.gravitino.rel.SupportsSchemas;
+import com.datastrato.gravitino.rest.RESTUtils;
 import com.google.common.annotations.VisibleForTesting;
 import java.util.Arrays;
 import java.util.Collections;
@@ -95,7 +96,8 @@ public Schema createSchema(NameIdentifier ident, String comment, Map<String, Str
       throws NoSuchCatalogException, SchemaAlreadyExistsException {
     NameIdentifier.checkSchema(ident);
 
-    SchemaCreateRequest req = new SchemaCreateRequest(ident.name(), comment, properties);
+    SchemaCreateRequest req =
+        new SchemaCreateRequest(RESTUtils.encodeString(ident.name()), comment, properties);
     req.validate();
 
     SchemaResponse resp =
@@ -123,7 +125,7 @@ public Schema loadSchema(NameIdentifier ident) throws NoSuchSchemaException {
 
     SchemaResponse resp =
         restClient.get(
-            formatSchemaRequestPath(ident.namespace()) + "/" + ident.name(),
+            formatSchemaRequestPath(ident.namespace()) + "/" + RESTUtils.encodeString(ident.name()),
             SchemaResponse.class,
             Collections.emptyMap(),
             ErrorHandlers.schemaErrorHandler());
@@ -154,7 +156,7 @@ public Schema alterSchema(NameIdentifier ident, SchemaChange... changes)
 
     SchemaResponse resp =
         restClient.put(
-            formatSchemaRequestPath(ident.namespace()) + "/" + ident.name(),
+            formatSchemaRequestPath(ident.namespace()) + "/" + RESTUtils.encodeString(ident.name()),
             updatesRequest,
             SchemaResponse.class,
             Collections.emptyMap(),
@@ -179,7 +181,9 @@ public boolean dropSchema(NameIdentifier ident, boolean cascade) throws NonEmpty
     try {
       DropResponse resp =
           restClient.delete(
-              formatSchemaRequestPath(ident.namespace()) + "/" + ident.name(),
+              formatSchemaRequestPath(ident.namespace())
+                  + "/"
+                  + RESTUtils.encodeString(ident.name()),
               Collections.singletonMap("cascade", String.valueOf(cascade)),
               DropResponse.class,
               Collections.emptyMap(),

diff --git a/clients/client-java/src/main/java/com/datastrato/gravitino/client/FilesetCatalog.java b/clients/client-java/src/main/java/com/datastrato/gravitino/client/FilesetCatalog.java
@@ -19,6 +19,7 @@
 import com.datastrato.gravitino.exceptions.NoSuchSchemaException;
 import com.datastrato.gravitino.file.Fileset;
 import com.datastrato.gravitino.file.FilesetChange;
+import com.datastrato.gravitino.rest.RESTUtils;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import java.util.Arrays;
@@ -88,7 +89,9 @@ public Fileset loadFileset(NameIdentifier ident) throws NoSuchFilesetException {
 
     FilesetResponse resp =
         restClient.get(
-            formatFilesetRequestPath(ident.namespace()) + "/" + ident.name(),
+            formatFilesetRequestPath(ident.namespace())
+                + "/"
+                + RESTUtils.encodeString(ident.name()),
             FilesetResponse.class,
             Collections.emptyMap(),
             ErrorHandlers.filesetErrorHandler());
@@ -126,7 +129,7 @@ public Fileset createFileset(
 
     FilesetCreateRequest req =
         FilesetCreateRequest.builder()
-            .name(ident.name())
+            .name(RESTUtils.encodeString(ident.name()))
             .comment(comment)
             .type(type)
             .storageLocation(storageLocation)
@@ -208,7 +211,7 @@ static String formatFilesetRequestPath(Namespace ns) {
     return new StringBuilder()
         .append(formatSchemaRequestPath(schemaNs))
         .append("/")
-        .append(ns.level(2))
+        .append(RESTUtils.encodeString(ns.level(2)))
         .append("/filesets")
         .toString();
   }

diff --git a/clients/client-java/src/main/java/com/datastrato/gravitino/client/RelationalCatalog.java b/clients/client-java/src/main/java/com/datastrato/gravitino/client/RelationalCatalog.java
@@ -28,6 +28,7 @@
 import com.datastrato.gravitino.rel.expressions.sorts.SortOrder;
 import com.datastrato.gravitino.rel.expressions.transforms.Transform;
 import com.datastrato.gravitino.rel.indexes.Index;
+import com.datastrato.gravitino.rest.RESTUtils;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import java.util.Arrays;
@@ -100,7 +101,7 @@ public Table loadTable(NameIdentifier ident) throws NoSuchTableException {
 
     TableResponse resp =
         restClient.get(
-            formatTableRequestPath(ident.namespace()) + "/" + ident.name(),
+            formatTableRequestPath(ident.namespace()) + "/" + RESTUtils.encodeString(ident.name()),
             TableResponse.class,
             Collections.emptyMap(),
             ErrorHandlers.tableErrorHandler());
@@ -137,7 +138,7 @@ public Table createTable(
 
     TableCreateRequest req =
         new TableCreateRequest(
-            ident.name(),
+            RESTUtils.encodeString(ident.name()),
             comment,
             toDTOs(columns),
             properties,
@@ -182,7 +183,7 @@ public Table alterTable(NameIdentifier ident, TableChange... changes)
 
     TableResponse resp =
         restClient.put(
-            formatTableRequestPath(ident.namespace()) + "/" + ident.name(),
+            formatTableRequestPath(ident.namespace()) + "/" + RESTUtils.encodeString(ident.name()),
             updatesRequest,
             TableResponse.class,
             Collections.emptyMap(),
@@ -205,7 +206,9 @@ public boolean dropTable(NameIdentifier ident) {
     try {
       DropResponse resp =
           restClient.delete(
-              formatTableRequestPath(ident.namespace()) + "/" + ident.name(),
+              formatTableRequestPath(ident.namespace())
+                  + "/"
+                  + RESTUtils.encodeString(ident.name()),
               DropResponse.class,
               Collections.emptyMap(),
               ErrorHandlers.tableErrorHandler());
@@ -233,7 +236,9 @@ public boolean purgeTable(NameIdentifier ident) throws UnsupportedOperationExcep
     try {
       DropResponse resp =
           restClient.delete(
-              formatTableRequestPath(ident.namespace()) + "/" + ident.name(),
+              formatTableRequestPath(ident.namespace())
+                  + "/"
+                  + RESTUtils.encodeString(ident.name()),
               params,
               DropResponse.class,
               Collections.emptyMap(),
@@ -255,7 +260,7 @@ static String formatTableRequestPath(Namespace ns) {
     return new StringBuilder()
         .append(formatSchemaRequestPath(schemaNs))
         .append("/")
-        .append(ns.level(2))
+        .append(RESTUtils.encodeString(ns.level(2)))
         .append("/tables")
         .toString();
   }

diff --git a/core/src/main/java/com/datastrato/gravitino/catalog/CapabilityHelpers.java b/core/src/main/java/com/datastrato/gravitino/catalog/CapabilityHelpers.java
@@ -79,6 +79,37 @@ public static NameIdentifier applyCapabilities(
     return NameIdentifier.of(namespace, name);
   }
 
+  public static NameIdentifier[] applyCaseSensitive(
+      NameIdentifier[] idents, Capability.Scope scope, OperationDispatcher operationDispatcher) {
+    return Arrays.stream(idents)
+        .map(ident -> applyCaseSensitive(ident, scope, operationDispatcher))
+        .toArray(NameIdentifier[]::new);
+  }
+
+  public static NameIdentifier applyCaseSensitive(
+      NameIdentifier ident, Capability.Scope scope, OperationDispatcher operationDispatcher) {
+    Capability capabilities = operationDispatcher.getCatalogCapability(ident);
+    Namespace namespace = applyCaseSensitive(ident.namespace(), scope, operationDispatcher);
+
+    String name = applyCaseSensitiveOnName(scope, ident.name(), capabilities);
+    return NameIdentifier.of(namespace, name);
+  }
+
+  public static Namespace applyCaseSensitive(
+      Namespace namespace, Capability.Scope identScope, OperationDispatcher operationDispatcher) {
+    String metalake = namespace.level(0);
+    String catalog = namespace.level(1);
+    if (identScope == Capability.Scope.TABLE
+        || identScope == Capability.Scope.FILESET
+        || identScope == Capability.Scope.TOPIC) {
+      String schema = namespace.level(namespace.length() - 1);
+      Capability capabilities = operationDispatcher.getCatalogCapability(namespace);
+      schema = applyCaseSensitiveOnName(Capability.Scope.SCHEMA, schema, capabilities);
+      return Namespace.of(metalake, catalog, schema);
+    }
+    return namespace;
+  }
+
   public static Transform[] applyCapabilities(Transform[] transforms, Capability capabilities) {
     return Arrays.stream(transforms)
         .map(t -> applyCapabilities(t, capabilities))

diff --git a/core/src/main/java/com/datastrato/gravitino/catalog/FilesetNormalizeDispatcher.java b/core/src/main/java/com/datastrato/gravitino/catalog/FilesetNormalizeDispatcher.java
@@ -5,6 +5,7 @@
 package com.datastrato.gravitino.catalog;
 
 import static com.datastrato.gravitino.catalog.CapabilityHelpers.applyCapabilities;
+import static com.datastrato.gravitino.catalog.CapabilityHelpers.applyCaseSensitive;
 
 import com.datastrato.gravitino.NameIdentifier;
 import com.datastrato.gravitino.Namespace;
@@ -26,21 +27,26 @@ public FilesetNormalizeDispatcher(FilesetOperationDispatcher dispatcher) {
 
   @Override
   public NameIdentifier[] listFilesets(Namespace namespace) throws NoSuchSchemaException {
-    Capability capability = dispatcher.getCatalogCapability(namespace);
-    Namespace standardizedNamespace =
-        applyCapabilities(namespace, Capability.Scope.FILESET, capability);
-    NameIdentifier[] identifiers = dispatcher.listFilesets(standardizedNamespace);
-    return applyCapabilities(identifiers, Capability.Scope.FILESET, capability);
+    // The constraints of the name spec may be more strict than underlying catalog,
+    // and for compatibility reasons, we only apply case-sensitive capabilities here.
+    Namespace caseSensitiveNs = applyCaseSensitive(namespace, Capability.Scope.FILESET, dispatcher);
+    NameIdentifier[] identifiers = dispatcher.listFilesets(caseSensitiveNs);
+    return applyCaseSensitive(identifiers, Capability.Scope.FILESET, dispatcher);
   }
 
   @Override
   public Fileset loadFileset(NameIdentifier ident) throws NoSuchFilesetException {
-    return dispatcher.loadFileset(normalizeNameIdentifier(ident));
+    // The constraints of the name spec may be more strict than underlying catalog,
+    // and for compatibility reasons, we only apply case-sensitive capabilities here.
+    return dispatcher.loadFileset(applyCaseSensitive(ident, Capability.Scope.FILESET, dispatcher));
   }
 
   @Override
   public boolean filesetExists(NameIdentifier ident) {
-    return dispatcher.filesetExists(normalizeNameIdentifier(ident));
+    // The constraints of the name spec may be more strict than underlying catalog,
+    // and for compatibility reasons, we only apply case-sensitive capabilities here.
+    return dispatcher.filesetExists(
+        applyCaseSensitive(ident, Capability.Scope.FILESET, dispatcher));
   }
 
   @Override
@@ -60,13 +66,17 @@ public Fileset alterFileset(NameIdentifier ident, FilesetChange... changes)
       throws NoSuchFilesetException, IllegalArgumentException {
     Capability capability = dispatcher.getCatalogCapability(ident);
     return dispatcher.alterFileset(
-        applyCapabilities(ident, Capability.Scope.FILESET, capability),
+        // The constraints of the name spec may be more strict than underlying catalog,
+        // and for compatibility reasons, we only apply case-sensitive capabilities here.
+        applyCaseSensitive(ident, Capability.Scope.FILESET, dispatcher),
         applyCapabilities(capability, changes));
   }
 
   @Override
   public boolean dropFileset(NameIdentifier ident) {
-    return dispatcher.dropFileset(normalizeNameIdentifier(ident));
+    // The constraints of the name spec may be more strict than underlying catalog,
+    // and for compatibility reasons, we only apply case-sensitive capabilities here.
+    return dispatcher.dropFileset(applyCaseSensitive(ident, Capability.Scope.FILESET, dispatcher));
   }
 
   private NameIdentifier normalizeNameIdentifier(NameIdentifier ident) {