Iceberg REST Phase 1 [2/3]: Runtime facade and unit coverage

cluster/configs/src/main/java/com/linkedin/openhouse/cluster/configs/ClusterProperties.java

@@ -96,4 +96,7 @@ public class ClusterProperties {
   // string
   @Value("${cluster.tables.allowed-client-name-values:}")
   private List<String> allowedClientNameValues;
+
+  @Value("${cluster.tables.iceberg-rest.enabled:false}")
+  private boolean clusterTablesIcebergRestEnabled;
 }

docs/iceberg-rest-catalog.md

@@ -0,0 +1,65 @@
+# Iceberg REST catalog
+
+OpenHouse exposes a read-only Apache Iceberg REST Catalog facade for new clients while preserving
+the existing OpenHouse APIs and business behavior.
+
+## Enablement
+
+The facade is disabled by default. Enable it with:
+
+```properties
+cluster.tables.iceberg-rest.enabled=true
+```
+
+`GET /v1/config` returns the `iceberg` route prefix and advertises only the implemented endpoints:
+
+- `GET /v1/{prefix}/namespaces/{namespace}/tables`
+- `GET /v1/{prefix}/namespaces/{namespace}/tables/{table}`
+- `HEAD /v1/{prefix}/namespaces/{namespace}/tables/{table}`
+
+## Architecture
+
+The OpenAPI-generated interfaces own the HTTP contract. `IcebergRestCatalogController` is a thin
+Spring MVC adapter, and `IcebergRestApiHandler` translates the Iceberg protocol to existing
+`TablesApiHandler` and `OpenHouseInternalCatalog` behavior. The facade does not add business rules
+or change the existing OpenHouse endpoints.
+
+Iceberg response types use a narrowly scoped Spring `HttpMessageConverter`. Errors are translated
+by controller-scoped advice into the standard Iceberg error envelope.
+
+## Compatibility and limitations
+
+- Only single-level namespaces are supported.
+- The optional `warehouse` configuration hint does not select a different OpenHouse warehouse.
+- List responses support opaque continuation tokens and page sizes from 1 through 1000.
+- Table loads return all snapshots. The `snapshots=refs` projection is explicitly unsupported.
+- The Iceberg 1.11 `referenced-by` query parameter is accepted and ignored.
+- Access delegation may be requested, but this read-only version does not vend credentials.
+- Conditional ETag responses are not currently emitted.
+- Namespace, table-write, view, transaction, credential, and OAuth endpoints are not advertised.
+
+Existing OpenHouse APIs remain supported. Client migrations can therefore be incremental.
+
+## Observability and audit
+
+Spring Boot records the facade through the standard `http.server.requests` metrics, including URI,
+status, and latency. Table reads delegate through `TablesApiHandler`, retaining existing
+authorization, lock visibility, and table-read audit behavior.
+
+## Contract maintenance
+
+`spec/iceberg-rest-catalog-open-api.yaml` is the full Apache Iceberg REST OpenAPI. OpenHouse support
+is opt-in:
+
+```yaml
+operationId: listTables
+x-openhouse-support: supported
+```
+
+Operations without the annotation are unsupported. The build codegens Spring interfaces only for
+`supported` operations and generates `IcebergRestOpenHouseSupport.SUPPORTED_ENDPOINTS` for
+`/v1/config`. Marking a new operation `supported` (or changing a supported signature) fails
+compilation until the facade implements it.
+
+To upgrade Iceberg OpenAPI: merge the newer upstream YAML into the checked-in file, add
+`x-openhouse-support: supported` where needed, then compile.

infra/recipes/docker-compose/oh-only/docker-compose.yml

@@ -4,6 +4,8 @@ services:
     extends:
       file: ../common/oh-services.yml
       service: openhouse-tables
+    environment:
+      CLUSTER_TABLES_ICEBERGREST_ENABLED: "true"
     volumes:
       - ./:/var/config/
     depends_on:

services/tables/src/main/java/com/linkedin/openhouse/tables/api/handler/IcebergRestApiHandler.java

@@ -0,0 +1,27 @@
+package com.linkedin.openhouse.tables.api.handler;
+
+import com.linkedin.openhouse.tables.generated.iceberg.model.CatalogConfig;
+import com.linkedin.openhouse.tables.generated.iceberg.model.ListTablesResponse;
+import org.apache.iceberg.rest.responses.LoadTableResponse;
+
+/** Protocol adapter between the generated Iceberg REST API and existing OpenHouse behavior. */
+public interface IcebergRestApiHandler {
+
+  String ICEBERG_REST_PREFIX = "iceberg";
+
+  CatalogConfig getConfig(String warehouse);
+
+  ListTablesResponse listTables(
+      String prefix, String namespace, String pageToken, Integer pageSize);
+
+  LoadTableResponse loadTable(
+      String prefix,
+      String namespace,
+      String table,
+      String accessDelegation,
+      String ifNoneMatch,
+      String snapshots,
+      String referencedBy);
+
+  void tableExists(String prefix, String namespace, String table);
+}

services/tables/src/main/java/com/linkedin/openhouse/tables/api/handler/impl/OpenHouseIcebergRestApiHandler.java

@@ -0,0 +1,172 @@
+package com.linkedin.openhouse.tables.api.handler.impl;
+
+import static com.linkedin.openhouse.common.security.AuthenticationUtils.extractAuthenticatedUserPrincipal;
+
+import com.linkedin.openhouse.common.api.spec.ApiResponse;
+import com.linkedin.openhouse.common.exception.NoSuchUserTableException;
+import com.linkedin.openhouse.internal.catalog.OpenHouseInternalCatalog;
+import com.linkedin.openhouse.tables.api.handler.IcebergRestApiHandler;
+import com.linkedin.openhouse.tables.api.handler.TablesApiHandler;
+import com.linkedin.openhouse.tables.api.spec.v0.response.GetAllTablesResponseBody;
+import com.linkedin.openhouse.tables.api.spec.v0.response.GetTableResponseBody;
+import com.linkedin.openhouse.tables.generated.iceberg.IcebergRestOpenHouseSupport;
+import com.linkedin.openhouse.tables.generated.iceberg.model.CatalogConfig;
+import com.linkedin.openhouse.tables.generated.iceberg.model.ListTablesResponse;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.stream.Collectors;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NoSuchNamespaceException;
+import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.rest.CatalogHandlers;
+import org.apache.iceberg.rest.RESTUtil;
+import org.apache.iceberg.rest.responses.LoadTableResponse;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.data.domain.Page;
+import org.springframework.stereotype.Component;
+
+/** Default Iceberg REST adapter backed by existing OpenHouse API handlers and catalog behavior. */
+@Component
+@ConditionalOnProperty(value = "cluster.tables.iceberg-rest.enabled", havingValue = "true")
+public class OpenHouseIcebergRestApiHandler implements IcebergRestApiHandler {
+
+  static final int DEFAULT_PAGE_SIZE = 100;
+  static final int MAX_PAGE_SIZE = 1000;
+  private static final String PAGE_TOKEN_VERSION = "v1";
+
+  private final TablesApiHandler tablesApiHandler;
+  private final OpenHouseInternalCatalog openHouseInternalCatalog;
+
+  public OpenHouseIcebergRestApiHandler(
+      TablesApiHandler tablesApiHandler, OpenHouseInternalCatalog openHouseInternalCatalog) {
+    this.tablesApiHandler = tablesApiHandler;
+    this.openHouseInternalCatalog = openHouseInternalCatalog;
+  }
+
+  @Override
+  public CatalogConfig getConfig(String warehouse) {
+    return new CatalogConfig(
+            Collections.singletonMap("prefix", ICEBERG_REST_PREFIX), Collections.emptyMap())
+        .endpoints(IcebergRestOpenHouseSupport.SUPPORTED_ENDPOINTS);
+  }
+
+  @Override
+  public ListTablesResponse listTables(
+      String prefix, String namespace, String pageToken, Integer pageSize) {
+    validatePrefix(prefix);
+    Namespace icebergNamespace = decodeSingleLevelNamespace(namespace);
+    PageCursor cursor = decodePageToken(pageToken, pageSize);
+    ApiResponse<GetAllTablesResponseBody> response =
+        tablesApiHandler.searchTables(
+            icebergNamespace.level(0), cursor.getPage(), cursor.getPageSize(), "tableId");
+    Page<GetTableResponseBody> page = response.getResponseBody().getPageResults();
+    LinkedHashSet<TableIdentifier> identifiers =
+        page.getContent().stream()
+            .map(table -> TableIdentifier.of(icebergNamespace, table.getTableId()))
+            .collect(Collectors.toCollection(LinkedHashSet::new));
+    String nextPageToken =
+        page.hasNext() ? encodePageToken(cursor.getPage() + 1, cursor.getPageSize()) : null;
+    return new ListTablesResponse().identifiers(identifiers).nextPageToken(nextPageToken);
+  }
+
+  @Override
+  public LoadTableResponse loadTable(
+      String prefix,
+      String namespace,
+      String table,
+      String accessDelegation,
+      String ifNoneMatch,
+      String snapshots,
+      String referencedBy) {
+    validatePrefix(prefix);
+    if (snapshots != null && !"all".equals(snapshots)) {
+      throw new UnsupportedOperationException(
+          "The snapshots=refs projection is not supported by this catalog");
+    }
+    // Iceberg 1.11 loadTable may send referenced-by for view-load chains; Phase 1 ignores it.
+
+    Namespace icebergNamespace = decodeSingleLevelNamespace(namespace);
+    String databaseId = icebergNamespace.level(0);
+    try {
+      tablesApiHandler.getTable(databaseId, table, extractAuthenticatedUserPrincipal());
+    } catch (NoSuchUserTableException e) {
+      throw new NoSuchTableException("Table does not exist: %s.%s", databaseId, table);
+    }
+
+    return CatalogHandlers.loadTable(
+        openHouseInternalCatalog, TableIdentifier.of(icebergNamespace, table));
+  }
+
+  @Override
+  public void tableExists(String prefix, String namespace, String table) {
+    validatePrefix(prefix);
+    Namespace icebergNamespace = decodeSingleLevelNamespace(namespace);
+    String databaseId = icebergNamespace.level(0);
+    try {
+      tablesApiHandler.getTable(databaseId, table, extractAuthenticatedUserPrincipal());
+    } catch (NoSuchUserTableException e) {
+      throw new NoSuchTableException("Table does not exist: %s.%s", databaseId, table);
+    }
+  }
+
+  private static void validatePrefix(String prefix) {
+    if (!ICEBERG_REST_PREFIX.equals(prefix)) {
+      throw new IllegalArgumentException("Unsupported Iceberg REST prefix");
+    }
+  }
+
+  private static Namespace decodeSingleLevelNamespace(String encodedNamespace) {
+    Namespace namespace = RESTUtil.decodeNamespace(encodedNamespace);
+    if (namespace.isEmpty() || namespace.levels().length != 1) {
+      throw new NoSuchNamespaceException("Only single-level namespaces are supported");
+    }
+    return namespace;
+  }
+
+  private static PageCursor decodePageToken(String pageToken, Integer requestedPageSize) {
+    if (pageToken == null) {
+      return new PageCursor(0, validatePageSize(requestedPageSize));
+    }
+
+    try {
+      String decoded = new String(Base64.getUrlDecoder().decode(pageToken), StandardCharsets.UTF_8);
+      String[] parts = decoded.split(":", -1);
+      if (parts.length != 3 || !PAGE_TOKEN_VERSION.equals(parts[0])) {
+        throw new IllegalArgumentException("Invalid Iceberg REST page token");
+      }
+      int page = Integer.parseInt(parts[1]);
+      int pageSize = validatePageSize(Integer.parseInt(parts[2]));
+      if (page < 1 || (requestedPageSize != null && requestedPageSize != pageSize)) {
+        throw new IllegalArgumentException("Invalid Iceberg REST page token");
+      }
+      return new PageCursor(page, pageSize);
+    } catch (IllegalArgumentException e) {
+      throw new IllegalArgumentException("Invalid Iceberg REST page token", e);
+    }
+  }
+
+  private static int validatePageSize(Integer requestedPageSize) {
+    int pageSize = requestedPageSize == null ? DEFAULT_PAGE_SIZE : requestedPageSize;
+    if (pageSize < 1 || pageSize > MAX_PAGE_SIZE) {
+      throw new IllegalArgumentException(
+          String.format("page-size must be between 1 and %s", MAX_PAGE_SIZE));
+    }
+    return pageSize;
+  }
+
+  private static String encodePageToken(int page, int pageSize) {
+    String value = String.format("%s:%s:%s", PAGE_TOKEN_VERSION, page, pageSize);
+    return Base64.getUrlEncoder()
+        .withoutPadding()
+        .encodeToString(value.getBytes(StandardCharsets.UTF_8));
+  }
+
+  @lombok.Value
+  private static class PageCursor {
+    int page;
+    int pageSize;
+  }
+}

services/tables/src/main/java/com/linkedin/openhouse/tables/controller/IcebergRestCatalogController.java

@@ -0,0 +1,75 @@
+package com.linkedin.openhouse.tables.controller;
+
+import com.linkedin.openhouse.tables.api.handler.IcebergRestApiHandler;
+import com.linkedin.openhouse.tables.generated.iceberg.api.CatalogApiApi;
+import com.linkedin.openhouse.tables.generated.iceberg.api.ConfigurationApiApi;
+import com.linkedin.openhouse.tables.generated.iceberg.model.CatalogConfig;
+import com.linkedin.openhouse.tables.generated.iceberg.model.ListTablesResponse;
+import io.swagger.v3.oas.annotations.Hidden;
+import java.util.Optional;
+import org.apache.iceberg.rest.responses.LoadTableResponse;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.context.request.NativeWebRequest;
+
+/**
+ * Thin Spring MVC adapter for the generated read-only Iceberg REST contract.
+ *
+ * <p>Protocol translation and orchestration live in {@link IcebergRestApiHandler}; existing
+ * OpenHouse handlers and services remain the source of business behavior.
+ */
+@Hidden
+@RestController
+@ConditionalOnProperty(value = "cluster.tables.iceberg-rest.enabled", havingValue = "true")
+public class IcebergRestCatalogController implements CatalogApiApi, ConfigurationApiApi {
+
+  private final IcebergRestApiHandler icebergRestApiHandler;
+
+  public IcebergRestCatalogController(IcebergRestApiHandler icebergRestApiHandler) {
+    this.icebergRestApiHandler = icebergRestApiHandler;
+  }
+
+  @Override
+  public Optional<NativeWebRequest> getRequest() {
+    return Optional.empty();
+  }
+
+  @Override
+  public ResponseEntity<CatalogConfig> getConfig(String warehouse) {
+    return ResponseEntity.ok(icebergRestApiHandler.getConfig(warehouse));
+  }
+
+  @Override
+  public ResponseEntity<ListTablesResponse> listTables(
+      String prefix, String namespace, String pageToken, Integer pageSize) {
+    return ResponseEntity.ok(
+        icebergRestApiHandler.listTables(prefix, namespace, pageToken, pageSize));
+  }
+
+  @Override
+  public ResponseEntity<LoadTableResponse> loadTable(
+      String prefix,
+      String namespace,
+      String table,
+      String xIcebergAccessDelegation,
+      String ifNoneMatch,
+      String snapshots,
+      String referencedBy) {
+    return ResponseEntity.ok(
+        icebergRestApiHandler.loadTable(
+            prefix,
+            namespace,
+            table,
+            xIcebergAccessDelegation,
+            ifNoneMatch,
+            snapshots,
+            referencedBy));
+  }
+
+  @Override
+  public ResponseEntity<Void> tableExists(String prefix, String namespace, String table) {
+    icebergRestApiHandler.tableExists(prefix, namespace, table);
+    return ResponseEntity.noContent().build();
+  }
+}