Tables: add FeatureFlagsResolver seam to stamp featureFlags server-side

apps/spark/src/test/java/com/linkedin/openhouse/catalog/e2e/RTASJavaTest.java

@@ -65,6 +65,9 @@ void testReplaceTransaction() throws Exception {
       Table table = catalog.buildTable(TABLE_IDENT, SCHEMA).withPartitionSpec(SPEC).create();
       table.newAppend().appendFile(FILE_A).commit();
 
+      // RTAS is disabled by default; opt the table in before replacing it.
+      table.updateProperties().set("replace.enabled", "true").commit();
+
       String originalLocation = table.location();
       String originalMetadataLocation = getMetadataLocation(table);
       long originalSnapshotId = table.currentSnapshot().snapshotId();
@@ -120,6 +123,9 @@ void testCreateOrReplaceTransaction() throws Exception {
       // verify that the table was created with one snapshot
       assertEquals(1, Iterables.size(table.snapshots()), "Should have one snapshot after create");
 
+      // RTAS is disabled by default; opt the table in before replacing it.
+      table.updateProperties().set("replace.enabled", "true").commit();
+
       // create or replace on existing table should replace it
       Transaction replaceTxn =
           catalog

apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java

@@ -752,6 +752,12 @@ public void testSnapshotsExpirationAfterReplaceTable() throws Exception {
       prepareTable(ops, tableName);
       populateTable(ops, tableName, numInserts);
 
+      // RTAS is disabled by default; opt the table in before replacing it.
+      ops.spark()
+          .sql(
+              String.format(
+                  "ALTER TABLE %s SET TBLPROPERTIES ('replace.enabled'='true')", tableName));
+
       // replace the table using RTAS
       ops.spark()
           .sql(

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

@@ -30,6 +30,10 @@ public final class CatalogConstants {
 
   public static final String EVOLVED_SCHEMA_KEY = "evolved.table.schema";
 
+  public static final String RTAS_ENABLED_TABLE_PROP = "replace.enabled";
+
+  public static final String WAP_ENABLED_TABLE_PROP = "write.wap.enabled";
+
   static final String FEATURE_TOGGLE_STOP_CREATE = "stop_create";
 
   static final String CLIENT_TABLE_SCHEMA = "client.table.schema";

integrations/java/iceberg-1.2/openhouse-java-itest/src/test/java/com/linkedin/openhouse/javaclient/OpenHouseTableOperationsTest.java

@@ -4,12 +4,15 @@
 
 import com.linkedin.openhouse.gen.tables.client.api.SnapshotApi;
 import com.linkedin.openhouse.gen.tables.client.api.TableApi;
+import com.linkedin.openhouse.gen.tables.client.invoker.ApiClient;
 import com.linkedin.openhouse.gen.tables.client.model.CreateUpdateTableRequestBody;
+import com.linkedin.openhouse.gen.tables.client.model.GetTableResponseBody;
 import com.linkedin.openhouse.gen.tables.client.model.History;
 import com.linkedin.openhouse.gen.tables.client.model.Policies;
 import com.linkedin.openhouse.gen.tables.client.model.PolicyTag;
 import com.linkedin.openhouse.gen.tables.client.model.Retention;
 import com.linkedin.openhouse.javaclient.exception.WebClientWithMessageException;
+import com.linkedin.openhouse.relocated.com.fasterxml.jackson.databind.ObjectMapper;
 import com.linkedin.openhouse.relocated.org.springframework.http.HttpStatus;
 import com.linkedin.openhouse.relocated.org.springframework.web.reactive.function.client.WebClientRequestException;
 import com.linkedin.openhouse.relocated.org.springframework.web.reactive.function.client.WebClientResponseException;
@@ -479,4 +482,121 @@ public void testPoliciesHistoryExistsUpdate() {
     Assertions.assertEquals(2, updatedPolicies.getHistory().getVersions());
     Assertions.assertEquals(true, updatedPolicies.getSharingEnabled());
   }
+
+  private OpenHouseTableOperations refreshableOps(TableApi tableApi) {
+    return OpenHouseTableOperations.builder()
+        .tableIdentifier(TableIdentifier.of("db", "tbl"))
+        .fileIO(mock(FileIO.class))
+        .tableApi(tableApi)
+        .snapshotApi(mock(SnapshotApi.class))
+        .cluster("cluster")
+        .build();
+  }
+
+  /** Before any refresh, there is no server-stamped config, so the safe default is null. */
+  @Test
+  public void testCurrentConfigNullBeforeRefresh() {
+    Assertions.assertNull(refreshableOps(mock(TableApi.class)).currentConfig());
+  }
+
+  /** doRefresh stashes the server-stamped config so subclasses can read it back. */
+  @Test
+  public void testDoRefreshCapturesConfig() {
+    TableApi mockTableApi = mock(TableApi.class);
+    Map<String, String> stamped =
+        Collections.singletonMap("openhouse.read-bridge", "{\"read\":\"ON\"}");
+    GetTableResponseBody body = mock(GetTableResponseBody.class);
+    when(body.getTableLocation()).thenReturn(null);
+    when(body.getConfig()).thenReturn(stamped);
+    when(mockTableApi.getTableV1(anyString(), anyString())).thenReturn(Mono.just(body));
+
+    OpenHouseTableOperations ops = refreshableOps(mockTableApi);
+    ops.doRefresh();
+
+    Assertions.assertSame(stamped, ops.currentConfig());
+  }
+
+  /** Absent config on the response => null, the consumer's safe default. */
+  @Test
+  public void testDoRefreshNullConfigWhenAbsent() {
+    TableApi mockTableApi = mock(TableApi.class);
+    GetTableResponseBody body = mock(GetTableResponseBody.class);
+    when(body.getTableLocation()).thenReturn(null);
+    when(body.getConfig()).thenReturn(null);
+    when(mockTableApi.getTableV1(anyString(), anyString())).thenReturn(Mono.just(body));
+
+    OpenHouseTableOperations ops = refreshableOps(mockTableApi);
+    ops.doRefresh();
+
+    Assertions.assertNull(ops.currentConfig());
+  }
+
+  /**
+   * The held config is a snapshot of the latest refresh, never sticky: once the server stops
+   * stamping config, a subsequent refresh must clear the previously-captured value back to null.
+   * Guards against a stale directive lingering after the server turns it off.
+   */
+  @Test
+  public void testDoRefreshClearsStaleConfig() {
+    TableApi mockTableApi = mock(TableApi.class);
+    Map<String, String> stamped =
+        Collections.singletonMap("openhouse.read-bridge", "{\"read\":\"ON\"}");
+
+    GetTableResponseBody withConfig = mock(GetTableResponseBody.class);
+    when(withConfig.getTableLocation()).thenReturn(null);
+    when(withConfig.getConfig()).thenReturn(stamped);
+
+    GetTableResponseBody withoutConfig = mock(GetTableResponseBody.class);
+    when(withoutConfig.getTableLocation()).thenReturn(null);
+    when(withoutConfig.getConfig()).thenReturn(null);
+
+    // First refresh stamps config, second refresh stops stamping it.
+    when(mockTableApi.getTableV1(anyString(), anyString()))
+        .thenReturn(Mono.just(withConfig))
+        .thenReturn(Mono.just(withoutConfig));
+
+    OpenHouseTableOperations ops = refreshableOps(mockTableApi);
+
+    ops.doRefresh();
+    Assertions.assertSame(stamped, ops.currentConfig());
+
+    ops.doRefresh();
+    Assertions.assertNull(ops.currentConfig());
+  }
+
+  /**
+   * Wire contract: a server-stamped config map deserializes on the client (the Iceberg REST {@code
+   * LoadTableResponse.config} convention — a string map). This is how the value actually arrives on
+   * a real table-load response.
+   */
+  @Test
+  public void testConfigDeserializeFromResponse() throws Exception {
+    ObjectMapper mapper = ApiClient.createDefaultObjectMapper(null);
+    String json =
+        "{\"tableId\":\"tbl\",\"databaseId\":\"db\",\"config\":{"
+            + "\"openhouse.read-bridge\":\"{\\\"read\\\":\\\"ON\\\"}\"}}";
+
+    GetTableResponseBody body = mapper.readValue(json, GetTableResponseBody.class);
+    Map<String, String> config = body.getConfig();
+    Assertions.assertNotNull(config);
+    // value stays an opaque JSON string; the channel never parses it.
+    Assertions.assertEquals("{\"read\":\"ON\"}", config.get("openhouse.read-bridge"));
+  }
+
+  /**
+   * Unknown future fields must not break deserialization — older clients ignore what they do not
+   * understand (FAIL_ON_UNKNOWN_PROPERTIES=false), and unknown config keys are simply carried.
+   */
+  @Test
+  public void testConfigToleratesUnknownFields() throws Exception {
+    ObjectMapper mapper = ApiClient.createDefaultObjectMapper(null);
+    String json =
+        "{\"tableId\":\"tbl\",\"databaseId\":\"db\",\"someFutureField\":\"x\",\"config\":{"
+            + "\"openhouse.unknown-feature\":\"whatever\"}}";
+
+    GetTableResponseBody body = mapper.readValue(json, GetTableResponseBody.class);
+    Map<String, String> config = body.getConfig();
+    Assertions.assertNotNull(config);
+    Assertions.assertEquals("whatever", config.get("openhouse.unknown-feature"));
+  }
 }

integrations/java/iceberg-1.2/openhouse-java-itest/src/test/java/com/linkedin/openhouse/javaclient/ReadBridgeTest.java

@@ -0,0 +1,46 @@
+package com.linkedin.openhouse.javaclient;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Unit tests for the client-side read-bridge config decoder ({@link ReadBridge#columnDefaults}),
+ * exercised in isolation. Mirrors the server-side encoder {@code ReadBridgeConfigResolver}.
+ */
+class ReadBridgeTest {
+
+  private static final String PREFIX = ReadBridge.COLUMN_DEFAULT_PREFIX;
+
+  @Test
+  void decodesColumnDefaultsByFieldId() {
+    // Inline calls avoid naming Jackson's JsonNode, which is relocated in the shaded client uber
+    // (and this module compiles at a source level without `var`).
+    Map<String, String> config = new HashMap<>();
+    config.put(PREFIX + "5", "\"US\"");
+    config.put(PREFIX + "7", "0");
+    assertEquals(2, ReadBridge.columnDefaults(config).size());
+    assertEquals("US", ReadBridge.columnDefaults(config).get(5).asText());
+    assertEquals(0, ReadBridge.columnDefaults(config).get(7).asInt());
+  }
+
+  @Test
+  void emptyWhenConfigNullOrNoReadBridgeKeys() {
+    assertTrue(ReadBridge.columnDefaults(null).isEmpty());
+    assertTrue(ReadBridge.columnDefaults(Collections.singletonMap("other.key", "x")).isEmpty());
+  }
+
+  @Test
+  void skipsEntriesWithBadFieldIdOrUnparseableValue() {
+    Map<String, String> config = new HashMap<>();
+    config.put(PREFIX + "5", "\"US\"");
+    config.put(PREFIX + "notAnInt", "\"x\""); // non-integer field-id -> skipped
+    config.put(PREFIX + "7", "{bad json"); // unparseable value -> skipped
+    assertEquals(1, ReadBridge.columnDefaults(config).size());
+    assertEquals("US", ReadBridge.columnDefaults(config).get(5).asText());
+  }
+}

integrations/java/iceberg-1.2/openhouse-java-runtime/src/main/java/com/linkedin/openhouse/javaclient/OpenHouseTableOperations.java

@@ -18,6 +18,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 import lombok.AccessLevel;
 import lombok.AllArgsConstructor;
@@ -61,6 +62,21 @@ public class OpenHouseTableOperations extends BaseMetastoreTableOperations {
   @Getter(AccessLevel.PROTECTED)
   private String cluster;
 
+  /**
+   * The per-table client {@code config} (Iceberg REST {@code LoadTableResponse.config} convention)
+   * the OH server stamped onto the most recent table-load response (or {@code null} if none / not
+   * yet refreshed). A final holder keeps Lombok's all-args constructor unchanged.
+   */
+  private final AtomicReference<Map<String, String>> config = new AtomicReference<>();
+
+  /**
+   * The server-stamped per-table client config from the last {@code doRefresh}, or {@code null}
+   * when absent. Subclasses read it to gate read-time behavior.
+   */
+  protected Map<String, String> currentConfig() {
+    return config.get();
+  }
+
   @Override
   protected String tableName() {
     return tableIdentifier.toString();
@@ -81,10 +97,9 @@ public FileIO io() {
   @Override
   public void doRefresh() {
     log.info("Calling doRefresh for table: {}", tableName());
-    Optional<String> tableLocation =
+    Optional<GetTableResponseBody> tableResponse =
         tableApi
             .getTableV1(tableIdentifier.namespace().toString(), tableIdentifier.name())
-            .mapNotNull(GetTableResponseBody::getTableLocation)
             /*
              on 404 from table service, resume the stream as empty response.
              for any other error, surface it!
@@ -98,22 +113,28 @@ public void doRefresh() {
                 WebClientRequestException.class,
                 e -> Mono.error(new WebClientRequestWithMessageException(e)))
             .blockOptional();
+    // Capture the server-stamped per-table config so subclasses can gate read-time behavior via
+    // currentConfig(); absent => null. Side-channel only: never sent back on writes.
+    this.config.set(tableResponse.map(GetTableResponseBody::getConfig).orElse(null));
+    Optional<String> tableLocation = tableResponse.map(GetTableResponseBody::getTableLocation);
     if (!tableLocation.isPresent() && currentMetadataLocation() != null) {
       throw new NoSuchTableException(
           "Cannot find table %s after refresh, maybe another process deleted it", tableName());
     }
-    // Read through loadMetadata() so subclasses can transform metadata as it loads. The 4-arg call
-    // with (null, 20) is the stock 1-arg behavior, with the parse step made overridable.
+    // Route the parse through loadMetadata() so subclasses can transform metadata as it loads;
+    // (null, 20) preserves the stock refresh behavior.
     super.refreshFromMetadataLocation(tableLocation.orElse(null), null, 20, this::loadMetadata);
     log.debug("Calling doRefresh succeeded");
   }
 
   /**
-   * Loads the table metadata at the given location. Defaults to the stock parser; subclasses may
-   * override to transform the metadata (e.g. attach column defaults) as it loads.
+   * Loads the table metadata from storage, then runs the read-time {@link ReadBridge} over it using
+   * the per-table behavior the server stamped onto {@link #currentConfig()}. Fail-closed:
+   * absent/off/unparseable config leaves the raw metadata untouched.
    */
   protected TableMetadata loadMetadata(String metadataLocation) {
-    return TableMetadataParser.read(io(), metadataLocation);
+    TableMetadata raw = TableMetadataParser.read(io(), metadataLocation);
+    return ReadBridge.apply(raw, currentConfig());
   }
 
   @Override

integrations/java/iceberg-1.2/openhouse-java-runtime/src/main/java/com/linkedin/openhouse/javaclient/ReadBridge.java

@@ -0,0 +1,78 @@
+package com.linkedin.openhouse.javaclient;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.iceberg.TableMetadata;
+
+/**
+ * Read-time bridge: overlays Iceberg V3 read semantics onto loaded metadata for tables/clients that
+ * don't yet carry them natively, using behavior the server delivers in the per-table {@code
+ * config}. Today it applies per-column initial-defaults; further V3 features can be backported
+ * through the same entry point as they are added.
+ *
+ * <p>Client end of the read-bridge wire contract — mirror of the server encoder {@code
+ * ReadBridgeConfigResolver} (services/tables). The contract is flat, namespaced config keys (no
+ * envelope/POJO): {@code openhouse.read-bridge.column-default.<fieldId> = <single-value-json>}.
+ *
+ * <p>Kept out of {@link OpenHouseTableOperations} so the read path stays slim. Pure and
+ * fail-closed: an entry with a non-integer field-id or an unparseable value is skipped, and nothing
+ * to apply returns {@code raw} unchanged.
+ */
+@Slf4j
+final class ReadBridge {
+
+  /** Mirror of {@code ReadBridgeConfigResolver.COLUMN_DEFAULT_PREFIX}. */
+  static final String COLUMN_DEFAULT_PREFIX = "openhouse.read-bridge.column-default.";
+
+  private static final ObjectMapper MAPPER = new ObjectMapper();
+
+  private ReadBridge() {}
+
+  /**
+   * Applies the bridged read-time behavior the server stamped into {@code config} onto {@code raw},
+   * returning the transformed metadata (or {@code raw} when there is nothing to bridge).
+   */
+  static TableMetadata apply(TableMetadata raw, Map<String, String> config) {
+    Map<Integer, JsonNode> columnDefaults = columnDefaults(config);
+    if (columnDefaults.isEmpty()) {
+      return raw;
+    }
+    // TODO(read-bridge): overlay columnDefaults onto raw.schemas() via withSchemaOverlay; future V3
+    // features bridged from config are applied here too.
+    return raw;
+  }
+
+  /**
+   * Decodes {@code field-id -> initial-default} from the {@code
+   * openhouse.read-bridge.column-default.*} config entries; empty when there are none. Skips any
+   * entry with a non-integer field-id or an unparseable value (fail-closed). Package-visible for
+   * unit testing in isolation.
+   */
+  static Map<Integer, JsonNode> columnDefaults(Map<String, String> config) {
+    if (config == null) {
+      return Collections.emptyMap();
+    }
+    Map<Integer, JsonNode> byFieldId = new HashMap<>();
+    for (Map.Entry<String, String> entry : config.entrySet()) {
+      if (!entry.getKey().startsWith(COLUMN_DEFAULT_PREFIX)) {
+        continue;
+      }
+      try {
+        int fieldId = Integer.parseInt(entry.getKey().substring(COLUMN_DEFAULT_PREFIX.length()));
+        byFieldId.put(fieldId, MAPPER.readTree(entry.getValue()));
+      } catch (RuntimeException | JsonProcessingException e) {
+        log.warn(
+            "Skipping unparseable read-bridge config entry {}={}",
+            entry.getKey(),
+            entry.getValue(),
+            e);
+      }
+    }
+    return byFieldId;
+  }
+}