[default values] Implement ReadBridge.apply without an Iceberg fork

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

@@ -13,6 +13,8 @@
 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.com.fasterxml.jackson.databind.node.ArrayNode;
+import com.linkedin.openhouse.relocated.com.fasterxml.jackson.databind.node.ObjectNode;
 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;
@@ -29,12 +31,14 @@
 import org.apache.iceberg.Files;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
+import org.apache.iceberg.SchemaParser;
 import org.apache.iceberg.Snapshot;
 import org.apache.iceberg.TableMetadata;
 import org.apache.iceberg.TableMetadataParser;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.CommitStateUnknownException;
 import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.io.InputFile;
 import org.apache.iceberg.io.OutputFile;
@@ -577,6 +581,51 @@ public void testDoRefreshKeepsConfigWhenLocationUnchanged() {
     Assertions.assertSame(stamped, ops.currentConfig());
   }
 
+  /**
+   * Skip-reload after a GET that stops stamping must still send overlays from the bound config. The
+   * server drops them; the client must not strip the default-aware signal.
+   */
+  @Test
+  public void testDoRefreshSkipReloadStillSendsStampedDefaults() {
+    String location = writeTempMetadata();
+    Map<String, String> stamped =
+        Collections.singletonMap(ReadBridge.COLUMN_DEFAULT_PREFIX + "2", "\"US\"");
+
+    GetTableResponseBody withConfig = mock(GetTableResponseBody.class);
+    when(withConfig.getTableLocation()).thenReturn(location);
+    when(withConfig.getConfig()).thenReturn(stamped);
+
+    GetTableResponseBody withoutConfig = mock(GetTableResponseBody.class);
+    when(withoutConfig.getTableLocation()).thenReturn(location);
+    when(withoutConfig.getConfig()).thenReturn(null);
+
+    TableApi mockTableApi = mock(TableApi.class);
+    when(mockTableApi.getTableV1(anyString(), anyString()))
+        .thenReturn(Mono.just(withConfig))
+        .thenReturn(Mono.just(withoutConfig));
+
+    OpenHouseTableOperations ops = refreshableOps(mockTableApi, localFileIO());
+    ops.doRefresh();
+    Assertions.assertSame(stamped, ops.currentConfig());
+
+    ops.doRefresh();
+    Assertions.assertSame(stamped, ops.currentConfig());
+
+    TableMetadata commit =
+        tableWithSchema(
+            "file:/tmp/rb-signal-skip-reload",
+            new Schema(
+                NestedField.optional(1, "id", Types.IntegerType.get()),
+                NestedField.from(NestedField.optional(2, "country", Types.StringType.get()))
+                    .withInitialDefault(Expressions.lit("US"))
+                    .build()));
+    Assertions.assertEquals(
+        "US",
+        SchemaParser.fromJson(ops.constructMetadataRequestBody(null, commit).getSchema())
+            .findField(2)
+            .initialDefault());
+  }
+
   /** A later load from a new metadata location binds that response's config. */
   @Test
   public void testDoRefreshBindsNewConfigWhenLocationChanges() {
@@ -741,4 +790,72 @@ public void testConfigToleratesUnknownFields() throws Exception {
     Assertions.assertNotNull(config);
     Assertions.assertEquals("whatever", config.get("openhouse.unknown-feature"));
   }
+
+  @Test
+  public void constructMetadataRequestBody_sendsStampedIdsKeepsUnstampedColumnDefaults() {
+    TableMetadata commit =
+        tableWithSchema(
+            "file:/tmp/rb-sanitize-ops-c",
+            new Schema(
+                NestedField.optional(1, "id", Types.IntegerType.get()),
+                NestedField.from(NestedField.optional(2, "country", Types.StringType.get()))
+                    .withInitialDefault(Expressions.lit("US"))
+                    .build(),
+                NestedField.from(NestedField.optional(3, "email", Types.StringType.get()))
+                    .withInitialDefault(Expressions.lit("none"))
+                    .build()));
+
+    OpenHouseTableOperations ops = refreshableOps(mock(TableApi.class));
+    ops.setCurrentConfig(
+        Collections.singletonMap(ReadBridge.COLUMN_DEFAULT_PREFIX + "2", "\"US\""));
+
+    CreateUpdateTableRequestBody body = ops.constructMetadataRequestBody(null, commit);
+    Schema sent = SchemaParser.fromJson(body.getSchema());
+
+    Assertions.assertEquals("US", sent.findField(2).initialDefault());
+    Assertions.assertEquals("none", sent.findField(3).initialDefault());
+    Assertions.assertEquals("email", sent.findField(3).name());
+  }
+
+  @Test
+  public void constructMetadataRequestBody_withoutConfigLeavesWriterDefaults() {
+    TableMetadata commit =
+        tableWithSchema(
+            "file:/tmp/rb-sanitize-create",
+            new Schema(
+                NestedField.from(NestedField.optional(1, "country", Types.StringType.get()))
+                    .withInitialDefault(Expressions.lit("US"))
+                    .build()));
+
+    CreateUpdateTableRequestBody body =
+        refreshableOps(mock(TableApi.class)).constructMetadataRequestBody(null, commit);
+
+    Assertions.assertEquals(
+        "US", SchemaParser.fromJson(body.getSchema()).findField(1).initialDefault());
+  }
+
+  /**
+   * {@link TableMetadata#newTableMetadata} reassigns ids and drops defaults. Put this schema back
+   * so PUT tests can see writer/overlay defaults.
+   */
+  private static TableMetadata tableWithSchema(String location, Schema schema) {
+    TableMetadata created =
+        TableMetadata.newTableMetadata(
+            schema, PartitionSpec.unpartitioned(), location, Collections.emptyMap());
+    try {
+      ObjectMapper mapper = new ObjectMapper();
+      ObjectNode root = (ObjectNode) mapper.readTree(TableMetadataParser.toJson(created));
+      Schema kept =
+          new Schema(
+              created.currentSchemaId(),
+              schema.columns(),
+              schema.getAliases(),
+              schema.identifierFieldIds());
+      ((ArrayNode) root.get("schemas")).set(0, mapper.readTree(SchemaParser.toJson(kept)));
+      return TableMetadataParser.fromJson(
+          created.metadataFileLocation(), mapper.writeValueAsString(root));
+    } catch (Exception e) {
+      throw new IllegalStateException(e);
+    }
+  }
 }

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

@@ -1,16 +1,21 @@
 package com.linkedin.openhouse.javaclient;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.types.Types;
 import org.junit.jupiter.api.Test;
 
-/** Decoder for {@link ReadBridge#from}. */
+/** Decoder and apply for {@link ReadBridge}. */
 class ReadBridgeTest {
 
   private static final String PREFIX = ReadBridge.COLUMN_DEFAULT_PREFIX;
@@ -21,16 +26,18 @@ void decodesColumnDefaultsByFieldId() {
     Map<String, String> config = new HashMap<>();
     config.put(PREFIX + "5", "\"US\"");
     config.put(PREFIX + "7", "0");
-    assertEquals(2, ReadBridge.from(config).columnDefaults().size());
-    assertEquals("US", ReadBridge.from(config).columnDefaults().get(5).asText());
-    assertEquals(0, ReadBridge.from(config).columnDefaults().get(7).asInt());
+    ReadBridge bridge = ReadBridge.from(config);
+    assertEquals(2, bridge.columnDefaults().size());
+    // Original JSON strings so apply can bind without a relocated JsonNode.
+    assertEquals("\"US\"", bridge.columnDefaults().get(5));
+    assertEquals("0", bridge.columnDefaults().get(7));
   }
 
   @Test
   void inertWhenConfigNullOrNoReadBridgeKeys() {
     assertSame(ReadBridge.INERT, ReadBridge.from(null));
     assertSame(ReadBridge.INERT, ReadBridge.from(Collections.singletonMap("other.key", "x")));
-    assertTrue(ReadBridge.INERT.columnDefaults().isEmpty());
+    assertTrue(ReadBridge.from(null).columnDefaults().isEmpty());
   }
 
   @Test
@@ -55,7 +62,111 @@ void ignoresUnknownKeysWithoutFailing() {
     Map<String, String> config = new HashMap<>();
     config.put(PREFIX + "5", "\"US\"");
     config.put("openhouse.read-bridge.some-future-feature.3", "{not a default}");
-    assertEquals(1, ReadBridge.from(config).columnDefaults().size());
-    assertEquals("US", ReadBridge.from(config).columnDefaults().get(5).asText());
+    ReadBridge bridge = ReadBridge.from(config);
+    assertEquals(1, bridge.columnDefaults().size());
+    assertEquals("\"US\"", bridge.columnDefaults().get(5));
+  }
+
+  @Test
+  void applyReturnsSameInstanceWhenNothingToBridge() {
+    TableMetadata raw = newTable("file:/tmp/rb-inert");
+    assertSame(raw, ReadBridge.INERT.apply(raw));
+    assertSame(raw, ReadBridge.from(Collections.singletonMap("other.key", "x")).apply(raw));
+  }
+
+  @Test
+  void applySetsInitialDefaultOnMatchingField() {
+    TableMetadata raw = newTable("file:/tmp/rb-apply");
+    Map<String, String> config = Collections.singletonMap(PREFIX + "2", "\"US\"");
+
+    TableMetadata bridged = ReadBridge.from(config).apply(raw);
+
+    assertEquals("US", bridged.schema().findField(2).initialDefault());
+    assertNull(bridged.schema().findField(1).initialDefault());
+    assertEquals(raw.uuid(), bridged.uuid());
+    assertEquals(raw.currentSchemaId(), bridged.currentSchemaId());
+    assertEquals(raw.metadataFileLocation(), bridged.metadataFileLocation());
+  }
+
+  @Test
+  void applyOverlaysEverySchemaId() {
+    Schema v0 =
+        new Schema(
+            0,
+            Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+            Types.NestedField.optional(2, "country", Types.StringType.get()));
+    Schema v1 =
+        new Schema(
+            1,
+            Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+            Types.NestedField.optional(2, "country", Types.StringType.get()),
+            Types.NestedField.optional(3, "region", Types.StringType.get()));
+    TableMetadata raw =
+        TableMetadata.buildFrom(
+                TableMetadata.newTableMetadata(
+                    v0,
+                    PartitionSpec.unpartitioned(),
+                    "file:/tmp/rb-multischema",
+                    Collections.emptyMap()))
+            .addSchema(v1, 3)
+            .setCurrentSchema(1)
+            .build();
+
+    Map<String, String> config = new HashMap<>();
+    config.put(PREFIX + "2", "\"US\"");
+    config.put(PREFIX + "3", "\"west\"");
+
+    TableMetadata bridged = ReadBridge.from(config).apply(raw);
+
+    assertEquals(2, bridged.schemas().size());
+    for (Schema schema : bridged.schemas()) {
+      assertEquals("US", schema.findField(2).initialDefault());
+    }
+    assertNull(bridged.schemasById().get(0).findField(3));
+    assertEquals("west", bridged.schemasById().get(1).findField(3).initialDefault());
+  }
+
+  @Test
+  void applyIgnoresFieldIdsAbsentFromAllSchemas() {
+    TableMetadata raw = newTable("file:/tmp/rb-gap");
+    Map<String, String> config = Collections.singletonMap(PREFIX + "99", "\"x\"");
+    assertSame(raw, ReadBridge.from(config).apply(raw));
+  }
+
+  @Test
+  void applyFailsLoudWhenDefaultCannotBindToColumnType() {
+    TableMetadata raw = newTable("file:/tmp/rb-bad-bind");
+    Map<String, String> config = Collections.singletonMap(PREFIX + "1", "\"not-an-int\"");
+    assertThrows(IllegalStateException.class, () -> ReadBridge.from(config).apply(raw));
+  }
+
+  @Test
+  void applySetsDefaultOnNestedStructField() {
+    Schema schema =
+        new Schema(
+            Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+            Types.NestedField.optional(
+                2,
+                "address",
+                Types.StructType.of(
+                    Types.NestedField.optional(3, "country", Types.StringType.get()))));
+    TableMetadata raw =
+        TableMetadata.newTableMetadata(
+            schema, PartitionSpec.unpartitioned(), "file:/tmp/rb-nested", Collections.emptyMap());
+    Map<String, String> config = Collections.singletonMap(PREFIX + "3", "\"US\"");
+
+    TableMetadata bridged = ReadBridge.from(config).apply(raw);
+
+    assertEquals(
+        "US", bridged.schema().findField(2).type().asStructType().field(3).initialDefault());
+  }
+
+  private static TableMetadata newTable(String location) {
+    Schema schema =
+        new Schema(
+            Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+            Types.NestedField.optional(2, "country", Types.StringType.get()));
+    return TableMetadata.newTableMetadata(
+        schema, PartitionSpec.unpartitioned(), location, Collections.emptyMap());
   }
 }

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

@@ -266,6 +266,11 @@ && getTableType(base, metadata)
     return createUpdateTableRequestBody;
   }
 
+  @VisibleForTesting
+  void setCurrentConfig(Map<String, String> value) {
+    config.set(value);
+  }
+
   /**
    * If request is coming from replication process, createUpdateTableRequestBody.tableType should be
    * REPLICA_TABLE Replication process requests are identified based on difference between table