Carry Iceberg REST TableUpdate deltas on the snapshots commit request, and use them for audit branchRefName

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

@@ -0,0 +1,183 @@
+package com.linkedin.openhouse.javaclient;
+
+import com.linkedin.openhouse.relocated.com.fasterxml.jackson.databind.JsonNode;
+import com.linkedin.openhouse.relocated.com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.Collections;
+import java.util.List;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.SnapshotParser;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Verifies that {@link OpenHouseTableOperations#serializeMetadataUpdates} emits Iceberg REST spec
+ * {@code TableUpdate} actions for the operations OpenHouse commits, and in particular that a
+ * ref-only operation is distinguishable from a data write.
+ *
+ * <p>These assertions are the load-bearing premise of the audit path: the server can only report
+ * which branch a commit wrote because the client states it here.
+ */
+public class OpenHouseTableOperationsMetadataUpdatesTest {
+
+  private static final ObjectMapper MAPPER = new ObjectMapper();
+
+  private static final Schema SCHEMA =
+      new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get()));
+
+  private static final String SNAPSHOT_JSON =
+      "{\"snapshot-id\":42,"
+          + "\"timestamp-ms\":1669126937912,"
+          + "\"summary\":{\"operation\":\"append\"},"
+          + "\"manifest-list\":\"/tmp/snap-42.avro\","
+          + "\"schema-id\":0}";
+
+  /**
+   * A table with one snapshot on main, with the construction history discarded.
+   *
+   * <p>{@code discardChanges()} matters: {@link TableMetadata#changes()} accumulates across builds
+   * within a session, so metadata assembled in-test would otherwise still carry its {@code
+   * assign-uuid} / {@code add-schema} / {@code add-spec} creation updates. In production the base
+   * comes from {@code doRefresh}, i.e. parsed off disk with no changes attached, so each commit's
+   * {@code changes()} is exactly that commit's delta. This reproduces that starting condition.
+   */
+  private static TableMetadata tableWithOneSnapshot() {
+    TableMetadata empty =
+        TableMetadata.newTableMetadata(
+            SCHEMA,
+            PartitionSpec.unpartitioned(),
+            SortOrder.unsorted(),
+            "/tmp/tbl",
+            Collections.emptyMap());
+    Snapshot snapshot = SnapshotParser.fromJson(SNAPSHOT_JSON);
+    // setBranchSnapshot adds the snapshot and points the ref at it in one step.
+    return TableMetadata.buildFrom(empty)
+        .setBranchSnapshot(snapshot, SnapshotRef.MAIN_BRANCH)
+        .discardChanges()
+        .build();
+  }
+
+  private static JsonNode parse(String json) throws Exception {
+    return MAPPER.readTree(json);
+  }
+
+  /**
+   * CREATE BRANCH adds a ref at the existing head and commits no snapshot. The resulting table
+   * state is ambiguous — main and the new branch point at the same snapshot — but the update list
+   * names the branch explicitly and contains no {@code add-snapshot}.
+   */
+  @Test
+  public void testCreateBranchEmitsOnlySetSnapshotRefNamingTheNewBranch() throws Exception {
+    TableMetadata base = tableWithOneSnapshot();
+    TableMetadata afterCreateBranch =
+        TableMetadata.buildFrom(base)
+            .setRef("feature_a", SnapshotRef.branchBuilder(42L).build())
+            .build();
+
+    List<String> updates = OpenHouseTableOperations.serializeMetadataUpdates(afterCreateBranch);
+
+    Assertions.assertNotNull(updates);
+    Assertions.assertEquals(1, updates.size(), "CREATE BRANCH must not report a snapshot write");
+    JsonNode update = parse(updates.get(0));
+    Assertions.assertEquals("set-snapshot-ref", update.get("action").asText());
+    Assertions.assertEquals("feature_a", update.get("ref-name").asText());
+    Assertions.assertEquals("branch", update.get("type").asText());
+    Assertions.assertEquals(42L, update.get("snapshot-id").asLong());
+  }
+
+  /** A tag carries {@code type: tag}, so consumers can tell it apart from a branch. */
+  @Test
+  public void testCreateTagEmitsTagTypedSetSnapshotRef() throws Exception {
+    TableMetadata base = tableWithOneSnapshot();
+    TableMetadata afterCreateTag =
+        TableMetadata.buildFrom(base)
+            .setRef("v1_release", SnapshotRef.tagBuilder(42L).build())
+            .build();
+
+    List<String> updates = OpenHouseTableOperations.serializeMetadataUpdates(afterCreateTag);
+
+    Assertions.assertNotNull(updates);
+    Assertions.assertEquals(1, updates.size());
+    JsonNode update = parse(updates.get(0));
+    Assertions.assertEquals("set-snapshot-ref", update.get("action").asText());
+    Assertions.assertEquals("v1_release", update.get("ref-name").asText());
+    Assertions.assertEquals("tag", update.get("type").asText());
+  }
+
+  /** DROP BRANCH is a removal, never a write. */
+  @Test
+  public void testDropBranchEmitsRemoveSnapshotRef() throws Exception {
+    TableMetadata withBranch =
+        TableMetadata.buildFrom(tableWithOneSnapshot())
+            .setRef("feature_a", SnapshotRef.branchBuilder(42L).build())
+            .discardChanges()
+            .build();
+    TableMetadata afterDropBranch =
+        TableMetadata.buildFrom(withBranch).removeRef("feature_a").build();
+
+    List<String> updates = OpenHouseTableOperations.serializeMetadataUpdates(afterDropBranch);
+
+    Assertions.assertNotNull(updates);
+    Assertions.assertEquals(1, updates.size());
+    JsonNode update = parse(updates.get(0));
+    Assertions.assertEquals("remove-snapshot-ref", update.get("action").asText());
+    Assertions.assertEquals("feature_a", update.get("ref-name").asText());
+  }
+
+  /**
+   * An append to a named branch reports both the new snapshot and the ref that moved, so a data
+   * write remains distinguishable from the ref-only case above.
+   */
+  @Test
+  public void testAppendToBranchEmitsAddSnapshotAndSetSnapshotRef() throws Exception {
+    TableMetadata base = tableWithOneSnapshot();
+    Snapshot newSnapshot =
+        SnapshotParser.fromJson(
+            "{\"snapshot-id\":43,"
+                + "\"parent-snapshot-id\":42,"
+                + "\"timestamp-ms\":1669126937999,"
+                + "\"summary\":{\"operation\":\"append\"},"
+                + "\"manifest-list\":\"/tmp/snap-43.avro\","
+                + "\"schema-id\":0}");
+    TableMetadata afterAppend =
+        TableMetadata.buildFrom(base).setBranchSnapshot(newSnapshot, "feature_a").build();
+
+    List<String> updates = OpenHouseTableOperations.serializeMetadataUpdates(afterAppend);
+
+    Assertions.assertNotNull(updates);
+    Assertions.assertEquals(
+        2, updates.size(), "append reports exactly the new snapshot and the ref that moved");
+    boolean sawAddSnapshot = false;
+    boolean sawBranchRef = false;
+    for (String json : updates) {
+      JsonNode update = parse(json);
+      String action = update.get("action").asText();
+      if ("add-snapshot".equals(action)) {
+        sawAddSnapshot = true;
+      } else if ("set-snapshot-ref".equals(action)
+          && "feature_a".equals(update.get("ref-name").asText())) {
+        sawBranchRef = true;
+        Assertions.assertEquals("branch", update.get("type").asText());
+        Assertions.assertEquals(43L, update.get("snapshot-id").asLong());
+      }
+    }
+    Assertions.assertTrue(sawAddSnapshot, "append must report add-snapshot");
+    Assertions.assertTrue(sawBranchRef, "append must report the branch it moved");
+  }
+
+  /**
+   * Metadata read straight off disk carries no changes. The field is omitted entirely rather than
+   * reported as an empty list, so consumers see "not stated" rather than "nothing happened".
+   */
+  @Test
+  public void testMetadataWithNoChangesYieldsNull() {
+    TableMetadata noChanges =
+        TableMetadata.buildFrom(tableWithOneSnapshot()).discardChanges().build();
+    Assertions.assertNull(OpenHouseTableOperations.serializeMetadataUpdates(noChanges));
+  }
+}

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

@@ -26,6 +26,8 @@
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.iceberg.BaseMetastoreTableOperations;
+import org.apache.iceberg.MetadataUpdate;
+import org.apache.iceberg.MetadataUpdateParser;
 import org.apache.iceberg.SchemaParser;
 import org.apache.iceberg.SnapshotParser;
 import org.apache.iceberg.SnapshotRefParser;
@@ -372,6 +374,7 @@ private void commitSnapshots(
             .collect(
                 Collectors.toMap(Map.Entry::getKey, e -> SnapshotRefParser.toJson(e.getValue()))));
     icebergSnapshotsRequestBody.createUpdateTableRequestBody(createUpdateTableRequestBody);
+    icebergSnapshotsRequestBody.jsonMetadataUpdates(serializeMetadataUpdates(newMetadata));
 
     snapshotApi
         .putSnapshotsV1(
@@ -387,6 +390,45 @@ private void commitSnapshots(
         .block();
   }
 
+  /**
+   * Serializes the deltas this commit applies into Iceberg REST spec {@code TableUpdate} JSON.
+   *
+   * <p>{@link TableMetadata#changes()} is the same delta list every Iceberg REST catalog sends as
+   * {@code CommitTableRequest.updates[]}, and {@code MetadataUpdateParser} emits the spec wire
+   * format verbatim. It states what the commit did rather than what the table now looks like, so a
+   * ref-only operation such as {@code CREATE BRANCH b} is visible as a lone {@code
+   * set-snapshot-ref} naming {@code b} — something no amount of inspecting the resulting snapshot
+   * list can recover.
+   *
+   * <p>Advisory only today: the server builds metadata from the full-state fields, so this method
+   * returns null rather than propagating any failure. It must never be able to fail a commit.
+   *
+   * @return spec-shaped update actions, or null when there is nothing trustworthy to report
+   */
+  @VisibleForTesting
+  static List<String> serializeMetadataUpdates(TableMetadata newMetadata) {
+    try {
+      List<MetadataUpdate> changes = newMetadata.changes();
+      if (changes == null || changes.isEmpty()) {
+        return null;
+      }
+      List<String> serialized = new ArrayList<>(changes.size());
+      for (MetadataUpdate change : changes) {
+        // MetadataUpdateParser rejects update types it does not recognize. Skip those rather than
+        // dropping the whole list, so one unknown action cannot blind the rest.
+        try {
+          serialized.add(MetadataUpdateParser.toJson(change));
+        } catch (RuntimeException e) {
+          log.debug("Skipping unserializable metadata update {}", change.getClass().getName(), e);
+        }
+      }
+      return serialized.isEmpty() ? null : serialized;
+    } catch (RuntimeException e) {
+      log.warn("Failed to serialize metadata updates; omitting from commit request", e);
+      return null;
+    }
+  }
+
   /**
    * A wrapper for a remote REST call to put snapshot.
    *

services/tables/src/main/java/com/linkedin/openhouse/tables/api/spec/v0/request/IcebergSnapshotsRequestBody.java

@@ -32,6 +32,33 @@ public class IcebergSnapshotsRequestBody {
               + "Key is the branch name, and value is the SnapshotRef.")
   private Map<String, String> snapshotRefs;
 
+  /**
+   * The deltas this commit applies, in Iceberg REST spec form.
+   *
+   * <p>Each element is one {@code TableUpdate} from the Iceberg REST catalog spec (an object with
+   * an {@code action} discriminator, e.g. {@code add-snapshot}, {@code set-snapshot-ref}, {@code
+   * remove-snapshot-ref}), serialized by {@code MetadataUpdateParser} so the wire format is
+   * byte-identical to {@code CommitTableRequest.updates[]}.
+   *
+   * <p>Unlike {@link #jsonSnapshots} and {@link #snapshotRefs} — which carry complete replacement
+   * state and therefore force the server to rediscover what changed by diffing — this field states
+   * the change directly. A {@code CREATE BRANCH b} that commits no new snapshot appears here as a
+   * single {@code set-snapshot-ref} naming {@code b}, which is otherwise unknowable server-side.
+   *
+   * <p>Optional and advisory in this release: the server still builds table metadata from {@code
+   * jsonSnapshots}/{@code snapshotRefs}, and clients predating this field simply omit it. Consumers
+   * must tolerate null/empty. This is the forward-compatible shape — when OpenHouse adopts the REST
+   * {@code CommitTableRequest} endpoint, this field is promoted to {@code updates} and the
+   * full-state fields retire.
+   */
+  @Schema(
+      description =
+          "Optional. Iceberg REST spec TableUpdate actions describing the deltas this commit "
+              + "applies, each serialized by MetadataUpdateParser and wire-compatible with "
+              + "CommitTableRequest.updates[]. Advisory only: table metadata is still built from "
+              + "jsonSnapshots/snapshotRefs. Older clients omit this field.")
+  private List<String> jsonMetadataUpdates;
+
   @Schema(description = "The request body that contains complete metadata")
   private CreateUpdateTableRequestBody createUpdateTableRequestBody;