Make entity lookups by id honor the specified entity type

extension/persistence/eclipselink/src/main/java/org/apache/polaris/extension/persistence/impl/eclipselink/PolarisEclipseLinkMetaStoreSessionImpl.java

@@ -309,7 +309,8 @@ public void deleteFromEntitiesInCurrentTxn(
       @Nonnull PolarisCallContext callCtx, @Nonnull PolarisEntityCore entity) {
 
     // delete it
-    this.store.deleteFromEntities(localSession.get(), entity.getCatalogId(), entity.getId());
+    this.store.deleteFromEntities(
+        localSession.get(), entity.getCatalogId(), entity.getId(), entity.getTypeCode());
   }
 
   /** {@inheritDoc} */
@@ -360,7 +361,8 @@ public void deleteAllInCurrentTxn(@Nonnull PolarisCallContext callCtx) {
   @Override
   public @Nullable PolarisBaseEntity lookupEntityInCurrentTxn(
       @Nonnull PolarisCallContext callCtx, long catalogId, long entityId, int typeCode) {
-    return ModelEntity.toEntity(this.store.lookupEntity(localSession.get(), catalogId, entityId));
+    return ModelEntity.toEntity(
+        this.store.lookupEntity(localSession.get(), catalogId, entityId, typeCode));
   }
 
   @Override

extension/persistence/eclipselink/src/main/java/org/apache/polaris/extension/persistence/impl/eclipselink/PolarisEclipseLinkStore.java

@@ -84,7 +84,8 @@ void writeToEntities(EntityManager session, PolarisBaseEntity entity) {
     diagnosticServices.check(session != null, "session_is_null");
     checkInitialized();
 
-    ModelEntity model = lookupEntity(session, entity.getCatalogId(), entity.getId());
+    ModelEntity model =
+        lookupEntity(session, entity.getCatalogId(), entity.getId(), entity.getTypeCode());
     if (model != null) {
       // Update if the same entity already exists
       model.update(entity);
@@ -129,11 +130,11 @@ void writeToGrantRecords(EntityManager session, PolarisGrantRecord grantRec) {
     session.persist(ModelGrantRecord.fromGrantRecord(grantRec));
   }
 
-  void deleteFromEntities(EntityManager session, long catalogId, long entityId) {
+  void deleteFromEntities(EntityManager session, long catalogId, long entityId, int typeCode) {
     diagnosticServices.check(session != null, "session_is_null");
     checkInitialized();
 
-    ModelEntity model = lookupEntity(session, catalogId, entityId);
+    ModelEntity model = lookupEntity(session, catalogId, entityId, typeCode);
     diagnosticServices.check(model != null, "entity_not_found");
 
     session.remove(model);
@@ -203,14 +204,15 @@ void deleteAll(EntityManager session) {
     LOGGER.debug("All entities deleted.");
   }
 
-  ModelEntity lookupEntity(EntityManager session, long catalogId, long entityId) {
+  ModelEntity lookupEntity(EntityManager session, long catalogId, long entityId, long typeCode) {
     diagnosticServices.check(session != null, "session_is_null");
     checkInitialized();
 
     return session
         .createQuery(
-            "SELECT m from ModelEntity m where m.catalogId=:catalogId and m.id=:id",
+            "SELECT m from ModelEntity m where m.catalogId=:catalogId and m.id=:id and m.typeCode=:typeCode",
             ModelEntity.class)
+        .setParameter("typeCode", typeCode)
         .setParameter("catalogId", catalogId)
         .setParameter("id", entityId)
         .getResultStream()

polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/TreeMapTransactionalPersistenceImpl.java

@@ -217,7 +217,12 @@ public void deleteAllInCurrentTxn(@Nonnull PolarisCallContext callCtx) {
   @Override
   public @Nullable PolarisBaseEntity lookupEntityInCurrentTxn(
       @Nonnull PolarisCallContext callCtx, long catalogId, long entityId, int typeCode) {
-    return this.store.getSliceEntities().read(this.store.buildKeyComposite(catalogId, entityId));
+    PolarisBaseEntity entity =
+        this.store.getSliceEntities().read(this.store.buildKeyComposite(catalogId, entityId));
+    if (entity != null && entity.getTypeCode() != typeCode) {
+      return null;
+    }
+    return entity;
   }
 
   /** {@inheritDoc} */

polaris-core/src/testFixtures/java/org/apache/polaris/core/persistence/BasePolarisMetaStoreManagerTest.java

@@ -273,6 +273,12 @@ protected void testRename() {
     polarisTestMetaStoreManager.testRename();
   }
 
+  /** test entity lookup */
+  @Test
+  protected void testLookup() {
+    polarisTestMetaStoreManager.testLookup();
+  }
+
   /** Test the set of functions for the entity cache */
   @Test
   protected void testEntityCache() {

polaris-core/src/testFixtures/java/org/apache/polaris/core/persistence/PolarisTestMetaStoreManager.java

@@ -192,6 +192,24 @@ private PolarisBaseEntity ensureExistsById(
     return entity;
   }
 
+  /**
+   * Validate that the specified identity identified by the pair catalogId, entityId does not exist.
+   *
+   * @param catalogId catalog id of that entity
+   * @param entityId the entity id
+   * @param expectedType its expected type
+   */
+  private void ensureNotExistsById(long catalogId, long entityId, PolarisEntityType expectedType) {
+
+    PolarisBaseEntity entity =
+        polarisMetaStoreManager
+            .loadEntity(this.polarisCallContext, catalogId, entityId, expectedType)
+            .getEntity();
+
+    // assert entity was not found
+    Assertions.assertThat(entity).isNull();
+  }
+
   /**
    * Check if the specified grant record exists
    *
@@ -2564,6 +2582,68 @@ public void testRename() {
     this.renameEntity(List.of(catalog, N1, N1_N2), N1_N2_T1, List.of(catalog, N5), "T7");
   }
 
+  /** Play with looking up entities */
+  public void testLookup() {
+    // load all principals
+    List<EntityNameLookupRecord> principals =
+        polarisMetaStoreManager
+            .listEntities(
+                this.polarisCallContext,
+                null,
+                PolarisEntityType.PRINCIPAL,
+                PolarisEntitySubType.NULL_SUBTYPE)
+            .getEntities();
+
+    // ensure not null, one element only
+    Assertions.assertThat(principals).isNotNull().hasSize(1);
+
+    // get catalog list information
+    EntityNameLookupRecord principalListInfo = principals.get(0);
+
+    PolarisBaseEntity principal =
+        this.ensureExistsById(
+            null,
+            principalListInfo.getId(),
+            true,
+            PolarisEntityConstants.getRootPrincipalName(),
+            PolarisEntityType.PRINCIPAL,
+            PolarisEntitySubType.NULL_SUBTYPE);
+
+    this.ensureNotExistsById(
+        PolarisEntityConstants.getNullId(), principal.getId(), PolarisEntityType.PRINCIPAL_ROLE);
+    this.ensureNotExistsById(
+        PolarisEntityConstants.getNullId(), principal.getId(), PolarisEntityType.CATALOG);
+    this.ensureNotExistsById(
+        PolarisEntityConstants.getNullId(), principal.getId(), PolarisEntityType.CATALOG_ROLE);
+
+    // create new catalog
+    PolarisBaseEntity catalog =
+        new PolarisBaseEntity(
+            PolarisEntityConstants.getNullId(),
+            polarisMetaStoreManager.generateNewEntityId(this.polarisCallContext).getId(),
+            PolarisEntityType.CATALOG,
+            PolarisEntitySubType.NULL_SUBTYPE,
+            PolarisEntityConstants.getRootEntityId(),
+            "test");
+    CreateCatalogResult catalogCreated =
+        polarisMetaStoreManager.createCatalog(this.polarisCallContext, catalog, List.of());
+    Assertions.assertThat(catalogCreated).isNotNull();
+    catalog = catalogCreated.getCatalog();
+
+    // now create all objects
+    PolarisBaseEntity N1 = this.createEntity(List.of(catalog), PolarisEntityType.NAMESPACE, "N1");
+    PolarisBaseEntity N1_N2 =
+        this.createEntity(List.of(catalog, N1), PolarisEntityType.NAMESPACE, "N2");
+    PolarisBaseEntity T1 =
+        this.createEntity(
+            List.of(catalog, N1, N1_N2),
+            PolarisEntityType.TABLE_LIKE,
+            PolarisEntitySubType.ICEBERG_TABLE,
+            "T1");
+
+    this.ensureNotExistsById(catalog.getId(), T1.getId(), PolarisEntityType.NAMESPACE);
+  }
+
   /** Test the set of functions for the entity cache */
   public void testEntityCache() {
     // create test catalog