Add entityType discriminator and isolate tables from views

iceberg/openhouse/internalcatalog/src/main/java/com/linkedin/openhouse/internal/catalog/mapper/HouseTableMapper.java

@@ -62,7 +62,16 @@ private static boolean isHtsField(String key) {
         && HouseTableSerdeUtils.HTS_FIELD_NAMES.contains(stripOhNamespace(key));
   }
 
+  /**
+   * MapStruct picks this up as an implicit {@code String -> String} conversion for every String
+   * property on the generated mappers, so it must tolerate null. {@code entityType} is the first
+   * genuinely nullable String on {@link HouseTable} (it stays null for legacy tables), which is
+   * what surfaced this; a null {@code tableVersion} would have hit it too.
+   */
   static String stripOhNamespace(String key) {
+    if (key == null) {
+      return null;
+    }
     return IS_OH_PREFIXED.test(key) ? key.substring(OPENHOUSE_NAMESPACE.length()) : key;
   }
 }

iceberg/openhouse/internalcatalog/src/main/java/com/linkedin/openhouse/internal/catalog/mapper/HouseTableSerdeUtils.java

@@ -35,6 +35,8 @@ private HouseTableSerdeUtils() {
     // no-op for util class constructor
   }
 
+  @VisibleForTesting public static final String ENTITY_TYPE_FIELD_NAME = "entityType";
+
   @VisibleForTesting
   public static String getCanonicalFieldName(String htsField) {
     return OPENHOUSE_NAMESPACE + htsField;

iceberg/openhouse/internalcatalog/src/main/java/com/linkedin/openhouse/internal/catalog/model/HouseTable.java

@@ -51,4 +51,11 @@ public class HouseTable {
    * with this table.
    */
   private String storageType;
+
+  /**
+   * As a private non-static field this is picked up automatically by {@link
+   * com.linkedin.openhouse.internal.catalog.mapper.HouseTableSerdeUtils#HTS_FIELD_NAMES}, so it
+   * serializes as the {@code openhouse.entityType} table property.
+   */
+  private String entityType;
 }

iceberg/openhouse/internalcatalog/src/test/java/com/linkedin/openhouse/internal/catalog/OpenHouseInternalTableOperationsTest.java

@@ -2128,4 +2128,66 @@ void testRefreshMetadataMissingFileThrowsInvalidTableMetadataException() {
         InvalidTableMetadataException.class,
         () -> openHouseInternalTableOperations.refreshMetadata(nonExistentPath));
   }
+
+  /**
+   * Backward compatibility: ordinary table commits must NOT start stamping the discriminator.
+   * Adding {@code openhouse.entityType=TABLE} to every table commit would rewrite every table's
+   * metadata.json content and mask the "NULL means table" compatibility contract that all the
+   * legacy-row assertions depend on.
+   */
+  @Test
+  void normalTableCommitDoesNotStampEntityType() {
+    AtomicReference<HouseTable> savedHouseTable = new AtomicReference<>();
+    when(mockHouseTableMapper.toHouseTable(Mockito.any(TableMetadata.class), Mockito.any()))
+        .thenAnswer(
+            invocation -> {
+              TableMetadata tableMetadata = invocation.getArgument(0);
+              // Mirror the real mapper: the pointer's entityType comes solely from the
+              // openhouse.entityType property, so a null here proves the commit did not stamp it.
+              HouseTable mapped =
+                  HouseTable.builder()
+                      .databaseId(TEST_TABLE_IDENTIFIER.namespace().toString())
+                      .tableId(TEST_TABLE_IDENTIFIER.name())
+                      .tableLocation(
+                          tableMetadata.properties().get(getCanonicalFieldName("tableLocation")))
+                      .entityType(
+                          tableMetadata.properties().get(getCanonicalFieldName("entityType")))
+                      .build();
+              savedHouseTable.set(mapped);
+              return mapped;
+            });
+    when(mockHouseTableRepository.save(Mockito.any(HouseTable.class)))
+        .thenAnswer(invocation -> invocation.getArgument(0));
+
+    Map<String, String> properties = new HashMap<>(BASE_TABLE_METADATA.properties());
+    properties.put(getCanonicalFieldName("tableLocation"), TEST_LOCATION);
+    TableMetadata metadata = BASE_TABLE_METADATA.replaceProperties(properties);
+
+    try (MockedStatic<TableMetadataParser> parserMock =
+        Mockito.mockStatic(TableMetadataParser.class, Mockito.CALLS_REAL_METHODS)) {
+      parserMock
+          .when(
+              () ->
+                  TableMetadataParser.write(
+                      Mockito.any(TableMetadata.class),
+                      Mockito.any(org.apache.iceberg.io.OutputFile.class)))
+          .thenAnswer(invocation -> null);
+
+      openHouseInternalTableOperations.doCommit(BASE_TABLE_METADATA, metadata);
+
+      Mockito.verify(mockHouseTableMapper).toHouseTable(tblMetadataCaptor.capture(), Mockito.any());
+      Map<String, String> committedProperties = tblMetadataCaptor.getValue().properties();
+      Assertions.assertFalse(
+          committedProperties.containsKey(getCanonicalFieldName("entityType")),
+          "Ordinary table commits must not write openhouse.entityType, but properties were: "
+              + committedProperties);
+
+      Assertions.assertNull(
+          savedHouseTable.get().getEntityType(),
+          "The saved pointer for an ordinary table commit must keep a null discriminator");
+      // NOTE: the captured-properties assertion above is the load-bearing one. This second
+      // assertion runs through a hand-rolled mirror of the real mapper, so it is null largely
+      // because the captured properties are; it guards the mapping wiring, not the commit itself.
+    }
+  }
 }

iceberg/openhouse/internalcatalog/src/test/java/com/linkedin/openhouse/internal/catalog/mapper/HouseTableMapperTest.java

@@ -7,6 +7,7 @@
 import com.linkedin.openhouse.housetables.client.api.ToggleStatusApi;
 import com.linkedin.openhouse.housetables.client.api.UserTableApi;
 import com.linkedin.openhouse.housetables.client.invoker.ApiClient;
+import com.linkedin.openhouse.housetables.client.model.UserTable;
 import com.linkedin.openhouse.internal.catalog.fileio.FileIOManager;
 import com.linkedin.openhouse.internal.catalog.model.HouseTable;
 import com.linkedin.openhouse.internal.catalog.repository.HouseTableRepository;
@@ -70,4 +71,59 @@ public void simpleMapperTest() {
     Assertions.assertEquals("table", houseTable.getTableId());
     Assertions.assertEquals("local", houseTable.getStorageType());
   }
+
+  /**
+   * The discriminator must survive the generated HTS client model in both directions, otherwise the
+   * pointer would be written or read back without its type.
+   */
+  @Test
+  public void houseTableGeneratedUserTableRoundTripPreservesEntityType() {
+    HouseTable viewHouseTable =
+        HouseTable.builder()
+            .databaseId("d1")
+            .tableId("v1")
+            .tableLocation("/base/d1/v1-uuid/00001-x.metadata.json")
+            .tableVersion("INITIAL_VERSION")
+            .entityType("VIEW")
+            .build();
+
+    UserTable viewUserTable = houseTableMapper.toUserTable(viewHouseTable);
+    Assertions.assertEquals("VIEW", viewUserTable.getEntityType());
+    Assertions.assertEquals(
+        "/base/d1/v1-uuid/00001-x.metadata.json", viewUserTable.getMetadataLocation());
+
+    HouseTable viewBack = houseTableMapper.toHouseTable(viewUserTable);
+    Assertions.assertEquals("VIEW", viewBack.getEntityType());
+    Assertions.assertEquals("/base/d1/v1-uuid/00001-x.metadata.json", viewBack.getTableLocation());
+
+    HouseTable tableHouseTable =
+        HouseTable.builder()
+            .databaseId("d1")
+            .tableId("t1")
+            .tableLocation("/base/d1/t1-uuid/00001-y.metadata.json")
+            .entityType("TABLE")
+            .build();
+    UserTable tableUserTable = houseTableMapper.toUserTable(tableHouseTable);
+    Assertions.assertEquals("TABLE", tableUserTable.getEntityType());
+    Assertions.assertEquals("TABLE", houseTableMapper.toHouseTable(tableUserTable).getEntityType());
+  }
+
+  /** Backward compatibility: an absent discriminator maps to null, never "TABLE". */
+  @Test
+  public void missingEntityTypeMapsToLegacyTableNull() {
+    HadoopFileIO fileIO = new HadoopFileIO(new Configuration());
+    LocalStorage localStorage = mock(LocalStorage.class);
+    when(fileIOManager.getStorage(fileIO)).thenReturn(localStorage);
+    when(localStorage.getType()).thenReturn(StorageType.LOCAL);
+
+    HouseTable fromProperties =
+        houseTableMapper.toHouseTable(ImmutableMap.of("databaseId", "d1", "tableId", "t1"), fileIO);
+    Assertions.assertNull(fromProperties.getEntityType());
+
+    UserTable legacyUserTable =
+        houseTableMapper.toUserTable(
+            HouseTable.builder().databaseId("d1").tableId("t1").tableLocation("loc").build());
+    Assertions.assertNull(legacyUserTable.getEntityType());
+    Assertions.assertNull(houseTableMapper.toHouseTable(legacyUserTable).getEntityType());
+  }
 }

iceberg/openhouse/internalcatalog/src/test/java/com/linkedin/openhouse/internal/catalog/model/HouseTableTest.java

@@ -60,4 +60,28 @@ public void testHouseTableDefaultValues() {
       Assertions.fail(e);
     }
   }
+
+  /**
+   * {@code HTS_FIELD_NAMES} is derived reflectively from HouseTable's private fields, and {@code
+   * HouseTableMapper.extractRawHTSFields} only carries properties whose stripped key is in that
+   * set. So the discriminator is only serialized through Iceberg table properties if it is a real
+   * private field named exactly {@code entityType}, and its canonical property key must be {@code
+   * openhouse.entityType}.
+   */
+  @Test
+  public void testEntityTypeDefaultAndSerdeRegistration() {
+    Assertions.assertNull(
+        HouseTable.builder().build().getEntityType(),
+        "entityType must default to null so ordinary table commits keep writing no discriminator");
+
+    Assertions.assertTrue(
+        HouseTableSerdeUtils.HTS_FIELD_NAMES.contains(HouseTableSerdeUtils.ENTITY_TYPE_FIELD_NAME),
+        "entityType must be reflected into HTS_FIELD_NAMES: "
+            + HouseTableSerdeUtils.HTS_FIELD_NAMES);
+
+    Assertions.assertEquals("entityType", HouseTableSerdeUtils.ENTITY_TYPE_FIELD_NAME);
+    Assertions.assertEquals(
+        "openhouse.entityType",
+        HouseTableSerdeUtils.getCanonicalFieldName(HouseTableSerdeUtils.ENTITY_TYPE_FIELD_NAME));
+  }
 }

iceberg/openhouse/internalcatalog/src/test/java/com/linkedin/openhouse/internal/catalog/repository/HouseTableRepositoryImplTest.java

@@ -754,4 +754,64 @@ public void testRestoreSoftDeletedTablesFailsWhenTableDoesNotExist() {
             htsRepo.restoreTable(
                 HOUSE_TABLE.getDatabaseId(), HOUSE_TABLE.getTableId(), System.currentTimeMillis()));
   }
+
+  /**
+   * This adapter is an HTTP client, not a query owner. It must faithfully carry the discriminator
+   * across the wire in both directions so the internal catalog's Java guards see the real stored
+   * value. It must NOT gain a local view filter of its own — filtering belongs in the HTS query so
+   * page counts stay correct.
+   */
+  @Test
+  public void testRepoPointAndSavePreserveEntityType() {
+    HouseTable viewHouseTable = HOUSE_TABLE.toBuilder().entityType("VIEW").build();
+
+    EntityResponseBodyUserTable getResponse = new EntityResponseBodyUserTable();
+    getResponse.entity(houseTableMapper.toUserTable(viewHouseTable));
+    mockHtsServer.enqueue(
+        new MockResponse()
+            .setResponseCode(200)
+            .setBody((new Gson()).toJson(getResponse))
+            .addHeader("Content-Type", "application/json"));
+
+    HouseTable found =
+        htsRepo
+            .findById(
+                HouseTablePrimaryKey.builder()
+                    .tableId(HOUSE_TABLE.getTableId())
+                    .databaseId(HOUSE_TABLE.getDatabaseId())
+                    .build())
+            .get();
+    Assertions.assertEquals("VIEW", found.getEntityType());
+    Assertions.assertEquals(HOUSE_TABLE.getTableLocation(), found.getTableLocation());
+
+    EntityResponseBodyUserTable putResponse = new EntityResponseBodyUserTable();
+    putResponse.entity(houseTableMapper.toUserTable(viewHouseTable));
+    mockHtsServer.enqueue(
+        new MockResponse()
+            .setResponseCode(201)
+            .setBody((new Gson()).toJson(putResponse))
+            .addHeader("Content-Type", "application/json"));
+
+    HouseTable saved = htsRepo.save(viewHouseTable);
+    Assertions.assertEquals("VIEW", saved.getEntityType());
+    Assertions.assertEquals(HOUSE_TABLE.getTableLocation(), saved.getTableLocation());
+
+    // A legacy table pointer with no discriminator still round trips as null.
+    EntityResponseBodyUserTable legacyResponse = new EntityResponseBodyUserTable();
+    legacyResponse.entity(houseTableMapper.toUserTable(HOUSE_TABLE));
+    mockHtsServer.enqueue(
+        new MockResponse()
+            .setResponseCode(200)
+            .setBody((new Gson()).toJson(legacyResponse))
+            .addHeader("Content-Type", "application/json"));
+    Assertions.assertNull(
+        htsRepo
+            .findById(
+                HouseTablePrimaryKey.builder()
+                    .tableId(HOUSE_TABLE.getTableId())
+                    .databaseId(HOUSE_TABLE.getDatabaseId())
+                    .build())
+            .get()
+            .getEntityType());
+  }
 }

services/common/src/main/java/com/linkedin/openhouse/common/api/validator/ValidatorConstants.java

@@ -19,4 +19,10 @@ private ValidatorConstants() {}
       "Only alphanumerics, hyphen and underscore supported";
   public static final int MAX_ALLOWED_CLUSTERING_COLUMNS = 4;
   public static final String INITIAL_TABLE_VERSION = "INITIAL_VERSION";
+
+  /** A null/absent value is allowed and means a legacy table row. */
+  public static final String ENTITY_TYPE_REGEX = "(?i)^(TABLE|VIEW)$";
+
+  public static final String ENTITY_TYPE_ERROR_MSG =
+      "Only TABLE and VIEW are supported entity types (case-insensitive)";
 }

services/housetables/src/main/java/com/linkedin/openhouse/housetables/api/spec/model/UserTable.java

@@ -2,6 +2,8 @@
 
 import static com.linkedin.openhouse.common.api.validator.ValidatorConstants.ALPHA_NUM_UNDERSCORE_ERROR_MSG;
 import static com.linkedin.openhouse.common.api.validator.ValidatorConstants.ALPHA_NUM_UNDERSCORE_REGEX;
+import static com.linkedin.openhouse.common.api.validator.ValidatorConstants.ENTITY_TYPE_ERROR_MSG;
+import static com.linkedin.openhouse.common.api.validator.ValidatorConstants.ENTITY_TYPE_REGEX;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.gson.Gson;
@@ -61,6 +63,15 @@ public class UserTable {
   @JsonProperty(value = "creationTime")
   private Long creationTime;
 
+  @Schema(
+      description =
+          "Type of the catalog object occupying this (databaseId, tableId) key. Null or 'TABLE' "
+              + "means a table; 'VIEW' means a view. Matched case-insensitively.",
+      example = "TABLE")
+  @JsonProperty(value = "entityType")
+  @Pattern(regexp = ENTITY_TYPE_REGEX, message = ENTITY_TYPE_ERROR_MSG)
+  private String entityType;
+
   @Schema(
       description =
           "Timestamp in milliseconds when the table was soft-deleted. "

services/housetables/src/main/java/com/linkedin/openhouse/housetables/dto/model/UserTableDto.java

@@ -23,6 +23,8 @@ public class UserTableDto {
 
   Long creationTime;
 
+  String entityType;
+
   Long deletedAtMs;
 
   Long purgeAfterMs;

services/housetables/src/main/java/com/linkedin/openhouse/housetables/model/UserTableRow.java

@@ -32,4 +32,7 @@ public class UserTableRow {
   String storageType;
 
   Long creationTime;
+
+  /** Nullable and without a default so existing rows need no backfill. */
+  String entityType;
 }