[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

@@ -25,14 +25,19 @@
 import java.util.Map;
 import java.util.Set;
 import org.apache.commons.compress.utils.Lists;
+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.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.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.types.Types.NestedField;
 import org.apache.iceberg.util.Tasks;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
@@ -494,13 +499,13 @@ private OpenHouseTableOperations refreshableOps(TableApi tableApi) {
         .build();
   }
 
-  /** Before any refresh, there is no server-stamped config, so the safe default is null. */
+  /** No refresh yet → no config. */
   @Test
   public void testCurrentConfigNullBeforeRefresh() {
     Assertions.assertNull(refreshableOps(mock(TableApi.class)).currentConfig());
   }
 
-  /** doRefresh stashes the server-stamped config so subclasses can read it back. */
+  /** doRefresh stores the response config. */
   @Test
   public void testDoRefreshCapturesConfig() {
     TableApi mockTableApi = mock(TableApi.class);
@@ -517,7 +522,7 @@ public void testDoRefreshCapturesConfig() {
     Assertions.assertSame(stamped, ops.currentConfig());
   }
 
-  /** Absent config on the response => null, the consumer's safe default. */
+  /** Missing config on the response is stored as null. */
   @Test
   public void testDoRefreshNullConfigWhenAbsent() {
     TableApi mockTableApi = mock(TableApi.class);
@@ -532,11 +537,7 @@ public void testDoRefreshNullConfigWhenAbsent() {
     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.
-   */
+  /** A later refresh without config clears the previous value. */
   @Test
   public void testDoRefreshClearsStaleConfig() {
     TableApi mockTableApi = mock(TableApi.class);
@@ -551,7 +552,7 @@ public void testDoRefreshClearsStaleConfig() {
     when(withoutConfig.getTableLocation()).thenReturn(null);
     when(withoutConfig.getConfig()).thenReturn(null);
 
-    // First refresh stamps config, second refresh stops stamping it.
+    // Second refresh has no config.
     when(mockTableApi.getTableV1(anyString(), anyString()))
         .thenReturn(Mono.just(withConfig))
         .thenReturn(Mono.just(withoutConfig));
@@ -565,17 +566,13 @@ public void testDoRefreshClearsStaleConfig() {
     Assertions.assertNull(ops.currentConfig());
   }
 
-  /**
-   * Bridge failures must not ride Iceberg's metadata-read retry. Decode runs before any FileIO
-   * access; {@link Tasks.UnrecoverableException} keeps {@code Tasks.retry(20)} from re-reading
-   * storage ~21 times (~90s) to reproduce a deterministic config error.
-   */
+  /** Bad config fails before FileIO so Iceberg does not retry the metadata read. */
   @Test
   public void testMalformedConfigFailsBeforeTouchingStorage() {
     TableApi mockTableApi = mock(TableApi.class);
     FileIO mockFileIO = mock(FileIO.class);
     GetTableResponseBody body = mock(GetTableResponseBody.class);
-    // Non-null location, so a metadata load would otherwise be attempted.
+    // Non-null location would otherwise trigger a metadata load.
     when(body.getTableLocation()).thenReturn("/tmp/does-not-matter/metadata.json");
     when(body.getConfig())
         .thenReturn(
@@ -597,11 +594,7 @@ public void testMalformedConfigFailsBeforeTouchingStorage() {
     verifyNoInteractions(mockFileIO);
   }
 
-  /**
-   * 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.
-   */
+  /** Config arrives as a string map on the table-load JSON. */
   @Test
   public void testConfigDeserializeFromResponse() throws Exception {
     ObjectMapper mapper = ApiClient.createDefaultObjectMapper(null);
@@ -612,14 +605,11 @@ public void testConfigDeserializeFromResponse() throws Exception {
     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.
+    // Channel does not parse the value.
     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.
-   */
+  /** Unknown JSON fields and unknown config keys are carried, not rejected. */
   @Test
   public void testConfigToleratesUnknownFields() throws Exception {
     ObjectMapper mapper = ApiClient.createDefaultObjectMapper(null);
@@ -632,4 +622,67 @@ public void testConfigToleratesUnknownFields() throws Exception {
     Assertions.assertNotNull(config);
     Assertions.assertEquals("whatever", config.get("openhouse.unknown-feature"));
   }
+
+  @Test
+  public void constructMetadataRequestBody_stripsOverlayOnExistingIdsKeepsNewColumnDefaults() {
+    TableMetadata raw =
+        tableWithSchema(
+            "file:/tmp/rb-sanitize-ops",
+            new Schema(
+                NestedField.optional(1, "id", Types.IntegerType.get()),
+                NestedField.optional(2, "country", Types.StringType.get())));
+    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.stashRawMetadata(raw);
+
+    CreateUpdateTableRequestBody body = ops.constructMetadataRequestBody(raw, commit);
+    Schema sent = SchemaParser.fromJson(body.getSchema());
+
+    Assertions.assertNull(sent.findField(2).initialDefault());
+    Assertions.assertEquals("none", sent.findField(3).initialDefault());
+    Assertions.assertEquals("email", sent.findField(3).name());
+  }
+
+  @Test
+  public void constructMetadataRequestBody_withoutRawLeavesWriterDefaults() {
+    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());
+  }
+
+  private static TableMetadata tableWithSchema(String location, Schema schema) {
+    TableMetadata created =
+        TableMetadata.newTableMetadata(
+            schema, PartitionSpec.unpartitioned(), location, Collections.emptyMap());
+    return ReadBridge.replaceSchemas(
+        created,
+        Collections.singletonMap(
+            created.currentSchemaId(),
+            new Schema(
+                created.currentSchemaId(),
+                schema.columns(),
+                schema.getAliases(),
+                schema.identifierFieldIds())));
+  }
 }