[apache#2856] fix(core): Fix the namespace missing issue when get ent…

…ities from kv storage (apache#2971)

### What changes were proposed in this pull request?

This PR fixes the namespace missing issue when entities are gotten from
kv storage.

### Why are the changes needed?

Fix: apache#2856 

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

Modified UTs.

build.gradle.kts

@@ -539,7 +539,8 @@ tasks.rat {
     "**/LICENSE.*",
     "**/NOTICE.*",
     "ROADMAP.md",
-    "clients/client-python/.pytest_cache/*"
+    "clients/client-python/.pytest_cache/*",
+    "clients/client-python/gravitino.egg-info/*"
   )
 
   // Add .gitignore excludes to the Apache Rat exclusion list.

core/src/main/java/com/datastrato/gravitino/EntitySerDe.java

@@ -24,15 +24,17 @@ public interface EntitySerDe {
    *
    * @param bytes the byte array to deserialize
    * @param clazz the class of the entity
+   * @param namespace the namespace to use when deserializing the entity
    * @return the deserialized entity
    * @param <T> The type of entity
    * @throws IOException if the deserialization fails
    */
-  default <T extends Entity> T deserialize(byte[] bytes, Class<T> clazz) throws IOException {
+  default <T extends Entity> T deserialize(byte[] bytes, Class<T> clazz, Namespace namespace)
+      throws IOException {
     ClassLoader loader =
         Optional.ofNullable(Thread.currentThread().getContextClassLoader())
             .orElse(getClass().getClassLoader());
-    return deserialize(bytes, clazz, loader);
+    return deserialize(bytes, clazz, loader, namespace);
   }
 
   /**
@@ -41,10 +43,12 @@ default <T extends Entity> T deserialize(byte[] bytes, Class<T> clazz) throws IO
    * @param bytes the byte array to deserialize
    * @param clazz the class of the entity
    * @param classLoader the class loader to use
+   * @param namespace the namespace to use when deserializing the entity
    * @return the deserialized entity
    * @param <T> The type of entity
    * @throws IOException if the deserialization fails
    */
-  <T extends Entity> T deserialize(byte[] bytes, Class<T> clazz, ClassLoader classLoader)
+  <T extends Entity> T deserialize(
+      byte[] bytes, Class<T> clazz, ClassLoader classLoader, Namespace namespace)
       throws IOException;
 }

core/src/main/java/com/datastrato/gravitino/catalog/CatalogManager.java

@@ -528,7 +528,7 @@ private void checkMetalakeExists(NameIdentifier ident) throws NoSuchMetalakeExce
   private CatalogWrapper loadCatalogInternal(NameIdentifier ident) throws NoSuchCatalogException {
     try {
       CatalogEntity entity = store.get(ident, EntityType.CATALOG, CatalogEntity.class);
-      return createCatalogWrapper(entity.withNamespace(ident.namespace()));
+      return createCatalogWrapper(entity);
 
     } catch (NoSuchEntityException ne) {
       LOG.warn("Catalog {} does not exist", ident, ne);

core/src/main/java/com/datastrato/gravitino/meta/CatalogEntity.java

@@ -12,7 +12,6 @@
 import com.datastrato.gravitino.HasIdentifier;
 import com.datastrato.gravitino.Namespace;
 import com.datastrato.gravitino.connector.CatalogInfo;
-import com.datastrato.gravitino.proto.CatalogEntitySerDe;
 import com.google.common.base.Objects;
 import java.util.Collections;
 import java.util.HashMap;
@@ -124,18 +123,6 @@ public CatalogInfo toCatalogInfo() {
     return new CatalogInfo(id, name, type, provider, comment, properties, auditInfo, namespace);
   }
 
-  /**
-   * Sets the namespace of the catalog entity. because the {@link CatalogEntitySerDe} does not
-   * serialize the namespace field
-   *
-   * @param namespace the namespace of the catalog entity.
-   * @return the instance of the source catalog entity.
-   */
-  public CatalogEntity withNamespace(Namespace namespace) {
-    this.namespace = namespace;
-    return this;
-  }
-
   /** Builder class for creating instances of {@link CatalogEntity}. */
   public static class Builder {
 
@@ -263,6 +250,7 @@ public boolean equals(Object o) {
     CatalogEntity that = (CatalogEntity) o;
     return Objects.equal(id, that.id)
         && Objects.equal(name, that.name)
+        && Objects.equal(namespace, that.namespace)
         && type == that.type
         && Objects.equal(provider, that.provider)
         && Objects.equal(comment, that.comment)

core/src/main/java/com/datastrato/gravitino/meta/FilesetEntity.java

@@ -166,6 +166,7 @@ public boolean equals(Object o) {
     FilesetEntity that = (FilesetEntity) o;
     return Objects.equals(id, that.id)
         && Objects.equals(name, that.name)
+        && Objects.equals(namespace, that.namespace)
         && Objects.equals(comment, that.comment)
         && Objects.equals(type, that.type)
         && Objects.equals(storageLocation, that.storageLocation)

core/src/main/java/com/datastrato/gravitino/meta/GroupEntity.java

@@ -120,6 +120,7 @@ public boolean equals(Object o) {
     GroupEntity that = (GroupEntity) o;
     return Objects.equals(id, that.id)
         && Objects.equals(name, that.name)
+        && Objects.equals(namespace, that.namespace)
         && Objects.equals(auditInfo, that.auditInfo)
         && Objects.equals(roles, that.roles);
   }

core/src/main/java/com/datastrato/gravitino/meta/RoleEntity.java

@@ -152,6 +152,7 @@ public boolean equals(Object o) {
     RoleEntity that = (RoleEntity) o;
     return Objects.equals(id, that.id)
         && Objects.equals(name, that.name)
+        && Objects.equals(namespace, that.namespace)
         && Objects.equals(auditInfo, that.auditInfo)
         && Objects.equals(properties, that.properties)
         && Objects.equals(securableObject, that.securableObject)

core/src/main/java/com/datastrato/gravitino/meta/SchemaEntity.java

@@ -145,6 +145,7 @@ public boolean equals(Object o) {
     SchemaEntity schema = (SchemaEntity) o;
     return Objects.equal(id, schema.id)
         && Objects.equal(name, schema.name)
+        && Objects.equal(namespace, schema.namespace)
         && Objects.equal(comment, schema.comment)
         && Objects.equal(properties, schema.properties)
         && Objects.equal(auditInfo, schema.auditInfo);

core/src/main/java/com/datastrato/gravitino/meta/TableEntity.java

@@ -109,6 +109,7 @@ public boolean equals(Object o) {
     TableEntity baseTable = (TableEntity) o;
     return Objects.equal(id, baseTable.id)
         && Objects.equal(name, baseTable.name)
+        && Objects.equal(namespace, baseTable.namespace)
         && Objects.equal(auditInfo, baseTable.auditInfo);
   }
 

core/src/main/java/com/datastrato/gravitino/meta/TopicEntity.java

@@ -135,6 +135,7 @@ public boolean equals(Object o) {
     TopicEntity that = (TopicEntity) o;
     return Objects.equals(id, that.id)
         && Objects.equals(name, that.name)
+        && Objects.equals(namespace, that.namespace)
         && Objects.equals(comment, that.comment)
         && Objects.equals(auditInfo, that.auditInfo)
         && Objects.equals(properties, that.properties);

core/src/main/java/com/datastrato/gravitino/meta/UserEntity.java

@@ -124,6 +124,7 @@ public boolean equals(Object o) {
     UserEntity that = (UserEntity) o;
     return Objects.equals(id, that.id)
         && Objects.equals(name, that.name)
+        && Objects.equals(namespace, that.namespace)
         && Objects.equals(auditInfo, that.auditInfo)
         && Objects.equals(roles, that.roles);
   }

core/src/main/java/com/datastrato/gravitino/proto/AuditInfoSerDe.java

@@ -5,6 +5,7 @@
 
 package com.datastrato.gravitino.proto;
 
+import com.datastrato.gravitino.Namespace;
 import java.util.Optional;
 
 /** A class for serializing and deserializing AuditInfo objects. */
@@ -41,7 +42,7 @@ public AuditInfo serialize(com.datastrato.gravitino.meta.AuditInfo auditInfo) {
    * @return The deserialized AuditInfo object.
    */
   @Override
-  public com.datastrato.gravitino.meta.AuditInfo deserialize(AuditInfo p) {
+  public com.datastrato.gravitino.meta.AuditInfo deserialize(AuditInfo p, Namespace namespace) {
     com.datastrato.gravitino.meta.AuditInfo.Builder builder =
         com.datastrato.gravitino.meta.AuditInfo.builder();
 

core/src/main/java/com/datastrato/gravitino/proto/BaseMetalakeSerDe.java

@@ -5,6 +5,7 @@
 
 package com.datastrato.gravitino.proto;
 
+import com.datastrato.gravitino.Namespace;
 import com.datastrato.gravitino.meta.AuditInfo;
 
 /** A class for serializing and deserializing BaseMetalake objects. */
@@ -52,13 +53,13 @@ public Metalake serialize(com.datastrato.gravitino.meta.BaseMetalake baseMetalak
    * @return The deserialized BaseMetalake object.
    */
   @Override
-  public com.datastrato.gravitino.meta.BaseMetalake deserialize(Metalake p) {
+  public com.datastrato.gravitino.meta.BaseMetalake deserialize(Metalake p, Namespace namespace) {
     com.datastrato.gravitino.meta.BaseMetalake.Builder builder =
         com.datastrato.gravitino.meta.BaseMetalake.builder();
     builder
         .withId(p.getId())
         .withName(p.getName())
-        .withAuditInfo(new AuditInfoSerDe().deserialize(p.getAuditInfo()));
+        .withAuditInfo(new AuditInfoSerDe().deserialize(p.getAuditInfo(), namespace));
 
     if (p.hasComment()) {
       builder.withComment(p.getComment());

core/src/main/java/com/datastrato/gravitino/proto/CatalogEntitySerDe.java

@@ -5,6 +5,7 @@
 
 package com.datastrato.gravitino.proto;
 
+import com.datastrato.gravitino.Namespace;
 import com.datastrato.gravitino.meta.AuditInfo;
 import com.datastrato.gravitino.meta.CatalogEntity;
 
@@ -49,13 +50,14 @@ public Catalog serialize(CatalogEntity catalogEntity) {
    * @return The deserialized CatalogEntity object.
    */
   @Override
-  public CatalogEntity deserialize(Catalog p) {
+  public CatalogEntity deserialize(Catalog p, Namespace namespace) {
     CatalogEntity.Builder builder = CatalogEntity.builder();
     builder
         .withId(p.getId())
         .withName(p.getName())
+        .withNamespace(namespace)
         .withProvider(p.getProvider())
-        .withAuditInfo(new AuditInfoSerDe().deserialize(p.getAuditInfo()));
+        .withAuditInfo(new AuditInfoSerDe().deserialize(p.getAuditInfo(), namespace));
 
     if (p.hasComment()) {
       builder.withComment(p.getComment());

core/src/main/java/com/datastrato/gravitino/proto/FilesetEntitySerDe.java

@@ -4,6 +4,7 @@
  */
 package com.datastrato.gravitino.proto;
 
+import com.datastrato.gravitino.Namespace;
 import com.datastrato.gravitino.meta.FilesetEntity;
 
 public class FilesetEntitySerDe implements ProtoSerDe<FilesetEntity, Fileset> {
@@ -31,13 +32,14 @@ public Fileset serialize(FilesetEntity filesetEntity) {
   }
 
   @Override
-  public FilesetEntity deserialize(Fileset p) {
+  public FilesetEntity deserialize(Fileset p, Namespace namespace) {
     FilesetEntity.Builder builder =
         FilesetEntity.builder()
             .withId(p.getId())
             .withName(p.getName())
+            .withNamespace(namespace)
             .withStorageLocation(p.getStorageLocation())
-            .withAuditInfo(new AuditInfoSerDe().deserialize(p.getAuditInfo()))
+            .withAuditInfo(new AuditInfoSerDe().deserialize(p.getAuditInfo(), namespace))
             .withFilesetType(
                 com.datastrato.gravitino.file.Fileset.Type.valueOf(p.getType().name()));
 

core/src/main/java/com/datastrato/gravitino/proto/GroupEntitySerDe.java

@@ -4,6 +4,7 @@
  */
 package com.datastrato.gravitino.proto;
 
+import com.datastrato.gravitino.Namespace;
 import com.datastrato.gravitino.meta.GroupEntity;
 import java.util.Collection;
 
@@ -25,12 +26,13 @@ public Group serialize(GroupEntity groupEntity) {
   }
 
   @Override
-  public GroupEntity deserialize(Group group) {
+  public GroupEntity deserialize(Group group, Namespace namespace) {
     GroupEntity.Builder builder =
         GroupEntity.builder()
             .withId(group.getId())
             .withName(group.getName())
-            .withAuditInfo(new AuditInfoSerDe().deserialize(group.getAuditInfo()));
+            .withNamespace(namespace)
+            .withAuditInfo(new AuditInfoSerDe().deserialize(group.getAuditInfo(), namespace));
 
     if (group.getRolesCount() > 0) {
       builder.withRoles(group.getRolesList());

core/src/main/java/com/datastrato/gravitino/proto/ProtoEntitySerDe.java

@@ -6,6 +6,7 @@
 
 import com.datastrato.gravitino.Entity;
 import com.datastrato.gravitino.EntitySerDe;
+import com.datastrato.gravitino.Namespace;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
 import com.google.protobuf.Any;
@@ -91,7 +92,8 @@ public <T extends Entity> byte[] serialize(T t) throws IOException {
   }
 
   @Override
-  public <T extends Entity> T deserialize(byte[] bytes, Class<T> clazz, ClassLoader classLoader)
+  public <T extends Entity> T deserialize(
+      byte[] bytes, Class<T> clazz, ClassLoader classLoader, Namespace namespace)
       throws IOException {
     Any any = Any.parseFrom(bytes);
     Class<? extends Message> protoClass = getProtoClass(clazz, classLoader);
@@ -101,7 +103,7 @@ public <T extends Entity> T deserialize(byte[] bytes, Class<T> clazz, ClassLoade
     }
 
     Message anyMessage = any.unpack(protoClass);
-    return fromProto(anyMessage, clazz, classLoader);
+    return fromProto(anyMessage, clazz, classLoader, namespace);
   }
 
   private <T extends Entity, M extends Message> ProtoSerDe<T, M> getProtoSerde(
@@ -151,9 +153,9 @@ private <T extends Entity, M extends Message> M toProto(T t, ClassLoader classLo
   }
 
   private <T extends Entity, M extends Message> T fromProto(
-      M m, Class<T> entityClass, ClassLoader classLoader) throws IOException {
+      M m, Class<T> entityClass, ClassLoader classLoader, Namespace namespace) throws IOException {
     ProtoSerDe<T, Message> protoSerDe = getProtoSerde(entityClass, classLoader);
-    return protoSerDe.deserialize(m);
+    return protoSerDe.deserialize(m, namespace);
   }
 
   private Class<?> loadClass(String className, ClassLoader classLoader) throws IOException {

core/src/main/java/com/datastrato/gravitino/proto/ProtoSerDe.java

@@ -4,6 +4,7 @@
  */
 package com.datastrato.gravitino.proto;
 
+import com.datastrato.gravitino.Namespace;
 import com.google.protobuf.Message;
 
 /**
@@ -26,7 +27,8 @@ public interface ProtoSerDe<T, M extends Message> {
    * Deserializes the provided Protocol Buffer message into its corresponding entity representation.
    *
    * @param p The Protocol Buffer message to be deserialized.
+   * @param namespace The namespace to be specified for entity deserialization.
    * @return The entity representing the deserialized Protocol Buffer message.
    */
-  T deserialize(M p);
+  T deserialize(M p, Namespace namespace);
 }

core/src/main/java/com/datastrato/gravitino/proto/RoleEntitySerDe.java

@@ -4,6 +4,7 @@
  */
 package com.datastrato.gravitino.proto;
 
+import com.datastrato.gravitino.Namespace;
 import com.datastrato.gravitino.authorization.Privileges;
 import com.datastrato.gravitino.authorization.SecurableObjects;
 import com.datastrato.gravitino.meta.RoleEntity;
@@ -44,17 +45,18 @@ public Role serialize(RoleEntity roleEntity) {
    * @return The entity representing the deserialized Protocol Buffer message.
    */
   @Override
-  public RoleEntity deserialize(Role role) {
+  public RoleEntity deserialize(Role role, Namespace namespace) {
     RoleEntity.Builder builder =
         RoleEntity.builder()
             .withId(role.getId())
             .withName(role.getName())
+            .withNamespace(namespace)
             .withPrivileges(
                 role.getPrivilegesList().stream()
                     .map(Privileges::fromString)
                     .collect(Collectors.toList()))
             .withSecurableObject(SecurableObjects.parse(role.getSecurableObject()))
-            .withAuditInfo(new AuditInfoSerDe().deserialize(role.getAuditInfo()));
+            .withAuditInfo(new AuditInfoSerDe().deserialize(role.getAuditInfo(), namespace));
 
     if (!role.getPropertiesMap().isEmpty()) {
       builder.withProperties(role.getPropertiesMap());

core/src/main/java/com/datastrato/gravitino/proto/SchemaEntitySerDe.java

@@ -4,6 +4,7 @@
  */
 package com.datastrato.gravitino.proto;
 
+import com.datastrato.gravitino.Namespace;
 import com.datastrato.gravitino.meta.SchemaEntity;
 
 public class SchemaEntitySerDe implements ProtoSerDe<SchemaEntity, Schema> {
@@ -27,12 +28,13 @@ public Schema serialize(SchemaEntity schemaEntity) {
   }
 
   @Override
-  public SchemaEntity deserialize(Schema p) {
+  public SchemaEntity deserialize(Schema p, Namespace namespace) {
     SchemaEntity.Builder builder =
         SchemaEntity.builder()
             .withId(p.getId())
             .withName(p.getName())
-            .withAuditInfo(new AuditInfoSerDe().deserialize(p.getAuditInfo()));
+            .withNamespace(namespace)
+            .withAuditInfo(new AuditInfoSerDe().deserialize(p.getAuditInfo(), namespace));
 
     if (p.hasComment()) {
       builder.withComment(p.getComment());

core/src/main/java/com/datastrato/gravitino/proto/TableEntitySerDe.java

@@ -4,6 +4,7 @@
  */
 package com.datastrato.gravitino.proto;
 
+import com.datastrato.gravitino.Namespace;
 import com.datastrato.gravitino.meta.TableEntity;
 
 public class TableEntitySerDe implements ProtoSerDe<TableEntity, Table> {
@@ -17,11 +18,12 @@ public Table serialize(TableEntity tableEntity) {
   }
 
   @Override
-  public TableEntity deserialize(Table p) {
+  public TableEntity deserialize(Table p, Namespace namespace) {
     return TableEntity.builder()
         .withId(p.getId())
         .withName(p.getName())
-        .withAuditInfo(new AuditInfoSerDe().deserialize(p.getAuditInfo()))
+        .withNamespace(namespace)
+        .withAuditInfo(new AuditInfoSerDe().deserialize(p.getAuditInfo(), namespace))
         .build();
   }
 }