[#9238] fix(authz): fix list tags for columns error

clients/client-java/src/test/java/org/apache/gravitino/client/integration/test/authorization/TagOperationsAuthorizationIT.java

@@ -47,6 +47,7 @@
 import org.apache.gravitino.tag.SupportsTags;
 import org.apache.gravitino.tag.TagChange;
 import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.MethodOrderer;
 import org.junit.jupiter.api.Order;
@@ -94,6 +95,7 @@ public void startIntegrationTest() throws Exception {
     // grant tester privilege
     List<SecurableObject> securableObjects = new ArrayList<>();
     GravitinoMetalake gravitinoMetalake = client.loadMetalake(METALAKE);
+
     SecurableObject catalogObject =
         SecurableObjects.ofCatalog(CATALOG, ImmutableList.of(Privileges.UseCatalog.allow()));
     securableObjects.add(catalogObject);
@@ -281,6 +283,22 @@ public void testListTagByMetadata() {
 
   @Test
   @Order(8)
+  public void testListColumnTag() {
+    GravitinoMetalake gravitinoMetalake = client.loadMetalake(METALAKE);
+    SupportsTags columnSupportTag =
+        gravitinoMetalake
+            .loadCatalog(CATALOG)
+            .asTableCatalog()
+            .loadTable(NameIdentifier.of(SCHEMA, "table1"))
+            .columns()[0]
+            .supportsTags();
+    String[] tags = columnSupportTag.listTags();
+    Arrays.sort(tags);
+    Assertions.assertArrayEquals(new String[] {"tag1", "tag2", "tag3"}, tags);
+  }
+
+  @Test
+  @Order(9)
   public void testDropTag() {
     GravitinoMetalake gravitinoMetalakeLoadByNormalUser = normalUserClient.loadMetalake(METALAKE);
     assertThrows(

core/src/main/java/org/apache/gravitino/utils/NameIdentifierUtil.java

@@ -370,6 +370,34 @@ public static NameIdentifier getSchemaIdentifier(NameIdentifier ident)
     return NameIdentifier.of(allElems.get(0), allElems.get(1), allElems.get(2));
   }
 
+  /**
+   * Try to get the table {@link NameIdentifier} from the given {@link NameIdentifier}.
+   *
+   * @param ident The {@link NameIdentifier} to check.
+   * @return The table {@link NameIdentifier}
+   * @throws IllegalNameIdentifierException If the given {@link NameIdentifier} does not include
+   *     table name
+   */
+  public static NameIdentifier getTableIdentifier(NameIdentifier ident)
+      throws IllegalNameIdentifierException {
+    NameIdentifier.check(
+        ident.name() != null && !ident.name().isEmpty(),
+        "The name variable in the NameIdentifier must have value.");
+    Namespace.check(
+        ident.namespace() != null && !ident.namespace().isEmpty() && ident.namespace().length() > 2,
+        "Table namespace must be non-null and at least 2 level, the input namespace is %s",
+        ident.namespace());
+
+    List<String> allElems =
+        Stream.concat(Arrays.stream(ident.namespace().levels()), Stream.of(ident.name()))
+            .collect(Collectors.toList());
+    if (allElems.size() < 4) {
+      throw new IllegalNameIdentifierException(
+          "Cannot create a table NameIdentifier less than four elements.");
+    }
+    return NameIdentifier.of(allElems.get(0), allElems.get(1), allElems.get(2), allElems.get(3));
+  }
+
   /**
    * Check the given {@link NameIdentifier} is a metalake identifier. Throw an {@link
    * IllegalNameIdentifierException} if it's not.

server-common/src/main/java/org/apache/gravitino/server/authorization/MetadataAuthzHelper.java

@@ -255,66 +255,10 @@ public static Map<Entity.EntityType, NameIdentifier> spiltMetadataNames(
       String metalake, Entity.EntityType entityType, NameIdentifier nameIdentifier) {
     Map<Entity.EntityType, NameIdentifier> nameIdentifierMap = new HashMap<>();
     nameIdentifierMap.put(Entity.EntityType.METALAKE, NameIdentifierUtil.ofMetalake(metalake));
-    switch (entityType) {
-      case CATALOG:
-        nameIdentifierMap.put(Entity.EntityType.CATALOG, nameIdentifier);
-        break;
-      case SCHEMA:
-        nameIdentifierMap.put(Entity.EntityType.SCHEMA, nameIdentifier);
-        nameIdentifierMap.put(
-            Entity.EntityType.CATALOG, NameIdentifierUtil.getCatalogIdentifier(nameIdentifier));
-        break;
-      case TABLE:
-        nameIdentifierMap.put(Entity.EntityType.TABLE, nameIdentifier);
-        nameIdentifierMap.put(
-            Entity.EntityType.SCHEMA, NameIdentifierUtil.getSchemaIdentifier(nameIdentifier));
-        nameIdentifierMap.put(
-            Entity.EntityType.CATALOG, NameIdentifierUtil.getCatalogIdentifier(nameIdentifier));
-        break;
-      case MODEL:
-        nameIdentifierMap.put(Entity.EntityType.MODEL, nameIdentifier);
-        nameIdentifierMap.put(
-            Entity.EntityType.SCHEMA, NameIdentifierUtil.getSchemaIdentifier(nameIdentifier));
-        nameIdentifierMap.put(
-            Entity.EntityType.CATALOG, NameIdentifierUtil.getCatalogIdentifier(nameIdentifier));
-        break;
-      case MODEL_VERSION:
-        nameIdentifierMap.put(Entity.EntityType.MODEL_VERSION, nameIdentifier);
-        nameIdentifierMap.put(
-            Entity.EntityType.MODEL, NameIdentifierUtil.getModelIdentifier(nameIdentifier));
-        nameIdentifierMap.put(
-            Entity.EntityType.SCHEMA, NameIdentifierUtil.getSchemaIdentifier(nameIdentifier));
-        nameIdentifierMap.put(
-            Entity.EntityType.CATALOG, NameIdentifierUtil.getCatalogIdentifier(nameIdentifier));
-        break;
-      case TOPIC:
-        nameIdentifierMap.put(Entity.EntityType.TOPIC, nameIdentifier);
-        nameIdentifierMap.put(
-            Entity.EntityType.SCHEMA, NameIdentifierUtil.getSchemaIdentifier(nameIdentifier));
-        nameIdentifierMap.put(
-            Entity.EntityType.CATALOG, NameIdentifierUtil.getCatalogIdentifier(nameIdentifier));
-        break;
-      case FILESET:
-        nameIdentifierMap.put(Entity.EntityType.FILESET, nameIdentifier);
-        nameIdentifierMap.put(
-            Entity.EntityType.SCHEMA, NameIdentifierUtil.getSchemaIdentifier(nameIdentifier));
-        nameIdentifierMap.put(
-            Entity.EntityType.CATALOG, NameIdentifierUtil.getCatalogIdentifier(nameIdentifier));
-        break;
-      case METALAKE:
-        nameIdentifierMap.put(entityType, nameIdentifier);
-        break;
-      case ROLE:
-        nameIdentifierMap.put(entityType, nameIdentifier);
-        break;
-      case USER:
-        nameIdentifierMap.put(entityType, nameIdentifier);
-        break;
-      case TAG:
-        nameIdentifierMap.put(entityType, nameIdentifier);
-        break;
-      default:
-        throw new IllegalArgumentException("Unsupported entity type: " + entityType);
+    while (entityType != Entity.EntityType.METALAKE) {
+      nameIdentifierMap.put(entityType, nameIdentifier);
+      entityType = NameIdentifierUtil.parentEntityType(entityType);
+      nameIdentifier = NameIdentifierUtil.parentNameIdentifier(nameIdentifier, entityType);
     }
     return nameIdentifierMap;
   }