diff --git a/integrations/java/iceberg-1.2/openhouse-java-itest/src/test/java/com/linkedin/openhouse/javaclient/OpenHouseTableOperationsMetadataUpdatesTest.java b/integrations/java/iceberg-1.2/openhouse-java-itest/src/test/java/com/linkedin/openhouse/javaclient/OpenHouseTableOperationsMetadataUpdatesTest.java
new file mode 100644
index 000000000..a168c51a9
--- /dev/null
+++ b/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.
+ *
+ * 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.
+ *
+ *
{@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 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 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 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 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));
+ }
+}
diff --git a/integrations/java/iceberg-1.2/openhouse-java-runtime/src/main/java/com/linkedin/openhouse/javaclient/OpenHouseTableOperations.java b/integrations/java/iceberg-1.2/openhouse-java-runtime/src/main/java/com/linkedin/openhouse/javaclient/OpenHouseTableOperations.java
index 7a3a07fc1..6395577ed 100644
--- a/integrations/java/iceberg-1.2/openhouse-java-runtime/src/main/java/com/linkedin/openhouse/javaclient/OpenHouseTableOperations.java
+++ b/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.
+ *
+ * {@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.
+ *
+ *
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 serializeMetadataUpdates(TableMetadata newMetadata) {
+ try {
+ List changes = newMetadata.changes();
+ if (changes == null || changes.isEmpty()) {
+ return null;
+ }
+ List 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.
*
diff --git a/services/tables/src/main/java/com/linkedin/openhouse/tables/api/spec/v0/request/IcebergSnapshotsRequestBody.java b/services/tables/src/main/java/com/linkedin/openhouse/tables/api/spec/v0/request/IcebergSnapshotsRequestBody.java
index f942c4594..be579d032 100644
--- a/services/tables/src/main/java/com/linkedin/openhouse/tables/api/spec/v0/request/IcebergSnapshotsRequestBody.java
+++ b/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 snapshotRefs;
+ /**
+ * The deltas this commit applies, in Iceberg REST spec form.
+ *
+ * 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[]}.
+ *
+ *
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.
+ *
+ *
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 jsonMetadataUpdates;
+
@Schema(description = "The request body that contains complete metadata")
private CreateUpdateTableRequestBody createUpdateTableRequestBody;
diff --git a/services/tables/src/main/java/com/linkedin/openhouse/tables/audit/TableAuditAspect.java b/services/tables/src/main/java/com/linkedin/openhouse/tables/audit/TableAuditAspect.java
index 243a6ab83..ad721d245 100644
--- a/services/tables/src/main/java/com/linkedin/openhouse/tables/audit/TableAuditAspect.java
+++ b/services/tables/src/main/java/com/linkedin/openhouse/tables/audit/TableAuditAspect.java
@@ -28,6 +28,8 @@
import java.util.regex.Pattern;
import java.util.regex.PatternSyntaxException;
import lombok.extern.slf4j.Slf4j;
+import org.apache.iceberg.MetadataUpdate;
+import org.apache.iceberg.MetadataUpdateParser;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.SnapshotParser;
import org.apache.iceberg.SnapshotRef;
@@ -47,6 +49,13 @@
@Component
public class TableAuditAspect {
+ /**
+ * The {@code type} discriminator the Iceberg REST spec assigns to branch refs in a {@code
+ * set-snapshot-ref} action (the alternative being {@code tag}). Iceberg's {@code SnapshotRefType}
+ * enum is package-private, so the spec's wire value is matched directly.
+ */
+ private static final String BRANCH_REF_TYPE = "branch";
+
@Autowired private ClusterProperties clusterProperties;
@Autowired private AuditHandler tableAuditHandler;
@@ -415,22 +424,26 @@ protected ApiResponse auditPutIcebergSnapshots(
}
/**
- * Extracts snapshot ID and timestamp of the main branch from the request body. The snapshotRefs
- * map contains branch name to JSON-serialized SnapshotRef. We read the main branch's snapshot-id
- * (this is what Iceberg treats as current-snapshot-id — see TableMetadata.Builder.setRef()) and
- * then find the matching snapshot in jsonSnapshots to get its timestamp-ms.
+ * Extracts snapshot ID, timestamp, and branch ref name from the request body.
*
- * Leaves both fields null if the main branch ref is absent (e.g. branch-only commits where
- * main didn't advance, or non-commit operations) or if the matching snapshot can't be found.
+ *
currentSnapshotId and currentSnapshotTimestampMs track the main branch ref for backwards
+ * compatibility. They are null when main is absent from snapshotRefs.
*/
private void extractSnapshotInfo(
IcebergSnapshotsRequestBody requestBody,
TableAuditEvent.TableAuditEventBuilder eventBuilder) {
try {
Map snapshotRefs = requestBody.getSnapshotRefs();
- if (snapshotRefs == null) {
+ List jsonSnapshots = requestBody.getJsonSnapshots();
+
+ extractBranchRefName(requestBody, eventBuilder);
+
+ if (snapshotRefs == null || jsonSnapshots == null || jsonSnapshots.isEmpty()) {
return;
}
+
+ // Extract snapshot ID and timestamp for main branch (backwards-compatible).
+ // Iterate jsonSnapshots in reverse: main's snapshot is typically the most recent.
String mainRefJson = snapshotRefs.get(SnapshotRef.MAIN_BRANCH);
if (mainRefJson == null) {
return;
@@ -438,14 +451,6 @@ private void extractSnapshotInfo(
long mainSnapshotId = SnapshotRefParser.fromJson(mainRefJson).snapshotId();
eventBuilder.currentSnapshotId(mainSnapshotId);
- // Find the matching snapshot in jsonSnapshots to get its timestamp-ms. Iterate in reverse
- // because Iceberg appends snapshots chronologically and main's snapshot is typically the
- // most recent. Skip snapshots whose JSON doesn't contain the target id as a cheap
- // pre-filter before invoking the JSON parser.
- List jsonSnapshots = requestBody.getJsonSnapshots();
- if (jsonSnapshots == null) {
- return;
- }
String mainSnapshotIdStr = Long.toString(mainSnapshotId);
for (int i = jsonSnapshots.size() - 1; i >= 0; i--) {
String snapshotJson = jsonSnapshots.get(i);
@@ -464,6 +469,49 @@ private void extractSnapshotInfo(
}
}
+ /**
+ * Sets branchRefName from the commit's Iceberg REST spec {@code TableUpdate} actions.
+ *
+ * The client sends the deltas it applied, so the branch that was written is stated outright by
+ * a {@code set-snapshot-ref} action rather than inferred. This matters most for operations that
+ * commit no snapshot at all: {@code CREATE BRANCH b} produces a lone {@code set-snapshot-ref}
+ * naming {@code b}, where the resulting table state is indistinguishable from a no-op on main.
+ *
+ *
Only branch-typed refs qualify; a {@code CREATE TAG} carries {@code type: tag} and is
+ * correctly ignored. When several branches move in one commit the first is reported, matching the
+ * order the client applied them.
+ *
+ *
Clients predating {@code jsonMetadataUpdates} omit it, in which case branchRefName is left
+ * unset. The previous behavior guessed by matching refs against the last snapshot in the list,
+ * which returned an arbitrary branch whenever two refs shared a snapshot — exactly what {@code
+ * CREATE BRANCH} produces. An absent field is preferable to a coin-flip one in an audit log.
+ */
+ private void extractBranchRefName(
+ IcebergSnapshotsRequestBody requestBody,
+ TableAuditEvent.TableAuditEventBuilder eventBuilder) {
+ List jsonMetadataUpdates = requestBody.getJsonMetadataUpdates();
+ if (jsonMetadataUpdates == null || jsonMetadataUpdates.isEmpty()) {
+ return;
+ }
+ for (String jsonMetadataUpdate : jsonMetadataUpdates) {
+ MetadataUpdate update;
+ try {
+ update = MetadataUpdateParser.fromJson(jsonMetadataUpdate);
+ } catch (Exception e) {
+ // A single unparseable action must not hide the rest of the commit's updates.
+ log.debug("Skipping unparseable metadata update in audit extraction", e);
+ continue;
+ }
+ if (update instanceof MetadataUpdate.SetSnapshotRef) {
+ MetadataUpdate.SetSnapshotRef setSnapshotRef = (MetadataUpdate.SetSnapshotRef) update;
+ if (BRANCH_REF_TYPE.equalsIgnoreCase(setSnapshotRef.type())) {
+ eventBuilder.branchRefName(setSnapshotRef.name());
+ return;
+ }
+ }
+ }
+ }
+
/** Install the Around advice for getAllDatabases() method in OpenHouseDatabasesApiHandler */
@Around(
"execution("
diff --git a/services/tables/src/main/java/com/linkedin/openhouse/tables/audit/model/TableAuditEvent.java b/services/tables/src/main/java/com/linkedin/openhouse/tables/audit/model/TableAuditEvent.java
index bca9796d9..526bde515 100644
--- a/services/tables/src/main/java/com/linkedin/openhouse/tables/audit/model/TableAuditEvent.java
+++ b/services/tables/src/main/java/com/linkedin/openhouse/tables/audit/model/TableAuditEvent.java
@@ -42,6 +42,8 @@ public class TableAuditEvent extends BaseAuditEvent {
private Long currentSnapshotTimestampMs;
+ private String branchRefName;
+
/** Allowlisted subset of table properties at commit time, not the full property map. */
private Map auditedTableProperties;
}
diff --git a/services/tables/src/test/java/com/linkedin/openhouse/tables/mock/RequestConstants.java b/services/tables/src/test/java/com/linkedin/openhouse/tables/mock/RequestConstants.java
index 7dda9edc3..ab71e8536 100644
--- a/services/tables/src/test/java/com/linkedin/openhouse/tables/mock/RequestConstants.java
+++ b/services/tables/src/test/java/com/linkedin/openhouse/tables/mock/RequestConstants.java
@@ -12,7 +12,9 @@
import com.linkedin.openhouse.tables.api.spec.v0.response.GetDatabaseResponseBody;
import com.linkedin.openhouse.tables.api.spec.v0.response.GetTableResponseBody;
import com.linkedin.openhouse.tables.api.spec.v0.response.components.AclPolicy;
+import java.util.Arrays;
import java.util.Collections;
+import java.util.List;
import java.util.Random;
import java.util.UUID;
@@ -94,11 +96,23 @@ private RequestConstants() {}
public static final String TEST_MAIN_SNAPSHOT_REF_JSON =
"{\"snapshot-id\":2151407017102313398,\"type\":\"branch\"}";
+ /**
+ * The Iceberg REST spec {@code TableUpdate} actions a plain append to main produces: the snapshot
+ * is added, then main is moved onto it. Shaped exactly as {@code MetadataUpdateParser} emits them
+ * client-side, so these fixtures exercise the same bytes a real commit sends.
+ */
+ public static final List TEST_MAIN_APPEND_METADATA_UPDATES =
+ Arrays.asList(
+ "{\"action\":\"add-snapshot\",\"snapshot\":" + TEST_ICEBERG_SNAPSHOT_JSON + "}",
+ "{\"action\":\"set-snapshot-ref\",\"ref-name\":\"main\","
+ + "\"snapshot-id\":2151407017102313398,\"type\":\"branch\"}");
+
public static final IcebergSnapshotsRequestBody TEST_ICEBERG_SNAPSHOTS_REQUEST_BODY =
IcebergSnapshotsRequestBody.builder()
.baseTableVersion("v1")
.jsonSnapshots(Collections.singletonList(TEST_ICEBERG_SNAPSHOT_JSON))
.snapshotRefs(Collections.singletonMap("main", TEST_MAIN_SNAPSHOT_REF_JSON))
+ .jsonMetadataUpdates(TEST_MAIN_APPEND_METADATA_UPDATES)
.createUpdateTableRequestBody(TEST_CREATE_TABLE_REQUEST_BODY)
.build();
@@ -123,6 +137,7 @@ private RequestConstants() {}
.baseTableVersion("INITIAL_VERSION")
.jsonSnapshots(Collections.singletonList(TEST_ICEBERG_SNAPSHOT_JSON))
.snapshotRefs(Collections.singletonMap("main", TEST_MAIN_SNAPSHOT_REF_JSON))
+ .jsonMetadataUpdates(TEST_MAIN_APPEND_METADATA_UPDATES)
.createUpdateTableRequestBody(TEST_CREATE_TABLE_REQUEST_BODY)
.build();
diff --git a/services/tables/src/test/java/com/linkedin/openhouse/tables/mock/audit/IcebergSnapshotsApiHandlerAuditTest.java b/services/tables/src/test/java/com/linkedin/openhouse/tables/mock/audit/IcebergSnapshotsApiHandlerAuditTest.java
index 381034ccd..c6d8c9932 100644
--- a/services/tables/src/test/java/com/linkedin/openhouse/tables/mock/audit/IcebergSnapshotsApiHandlerAuditTest.java
+++ b/services/tables/src/test/java/com/linkedin/openhouse/tables/mock/audit/IcebergSnapshotsApiHandlerAuditTest.java
@@ -12,6 +12,7 @@
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.Map;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
@@ -125,11 +126,207 @@ public void testPutIcebergSnapshotsFailedPathStillHasSnapshotInfo() throws Excep
assertEquals(1669126937912L, actualEvent.getCurrentSnapshotTimestampMs().longValue());
}
+ @Test
+ public void testPutIcebergSnapshotsMainCommitSetsBranchRefNameToMain() throws Exception {
+ mvc.perform(
+ MockMvcRequestBuilders.put(
+ String.format(
+ CURRENT_MAJOR_VERSION_PREFIX
+ + "/databases/d200/tables/tb1/iceberg/v2/snapshots"))
+ .accept(MediaType.APPLICATION_JSON)
+ .contentType(MediaType.APPLICATION_JSON)
+ .content(RequestConstants.TEST_ICEBERG_SNAPSHOTS_REQUEST_BODY.toJson()));
+ Mockito.verify(tableAuditHandler, atLeastOnce()).audit(argCaptor.capture());
+ assertEquals("main", argCaptor.getValue().getBranchRefName());
+ }
+
+ @Test
+ public void testPutIcebergSnapshotsNamedBranchCommitSetsBranchRefName() throws Exception {
+ // Realistic named-branch commit: main ref exists but its snapshot is NOT in jsonSnapshots
+ // (main didn't advance). Only the feature branch got a new snapshot.
+ String newSnapshotJson =
+ "{\n"
+ + " \"snapshot-id\" : 999,\n"
+ + " \"timestamp-ms\" : 5000,\n"
+ + " \"summary\" : {\"operation\": \"append\"},\n"
+ + " \"manifest-list\" : \"/tmp/feature.avro\",\n"
+ + " \"schema-id\" : 0\n"
+ + "}";
+ Map refs = new HashMap<>();
+ refs.put("main", "{\"snapshot-id\":100,\"type\":\"branch\"}"); // main stayed at old snapshot
+ refs.put("feature", "{\"snapshot-id\":999,\"type\":\"branch\"}"); // feature got new snapshot
+
+ IcebergSnapshotsRequestBody requestBody =
+ IcebergSnapshotsRequestBody.builder()
+ .baseTableVersion("v1")
+ .jsonSnapshots(Collections.singletonList(newSnapshotJson))
+ .snapshotRefs(refs)
+ .jsonMetadataUpdates(
+ Collections.singletonList(setSnapshotRef("feature", 999L, "branch")))
+ .createUpdateTableRequestBody(RequestConstants.TEST_CREATE_TABLE_REQUEST_BODY)
+ .build();
+
+ mvc.perform(
+ MockMvcRequestBuilders.put(
+ String.format(
+ CURRENT_MAJOR_VERSION_PREFIX
+ + "/databases/d200/tables/tb1/iceberg/v2/snapshots"))
+ .accept(MediaType.APPLICATION_JSON)
+ .contentType(MediaType.APPLICATION_JSON)
+ .content(requestBody.toJson()));
+ Mockito.verify(tableAuditHandler, atLeastOnce()).audit(argCaptor.capture());
+ TableAuditEvent actualEvent = argCaptor.getValue();
+ assertEquals("feature", actualEvent.getBranchRefName());
+ // main didn't advance, so currentSnapshotId is main's old snapshot and timestamp is null
+ assertEquals(100L, actualEvent.getCurrentSnapshotId().longValue());
+ assertNull(actualEvent.getCurrentSnapshotTimestampMs());
+ }
+
+ /**
+ * {@code ALTER TABLE t CREATE BRANCH b} on a table that already has snapshots. This is the case
+ * the resulting table state cannot express: the ref is created at the current head and no
+ * snapshot is committed, so main and b are indistinguishable in {@code snapshotRefs} — both point
+ * at the same, already-existing snapshot. The commit's {@code set-snapshot-ref} action names b
+ * outright.
+ */
+ @Test
+ public void testPutIcebergSnapshotsCreateBranchAtHeadReportsNewBranchNotMain() throws Exception {
+ Map refs = new HashMap<>();
+ refs.put("main", TEST_HEAD_SNAPSHOT_REF_JSON);
+ refs.put("b", TEST_HEAD_SNAPSHOT_REF_JSON); // same snapshot as main
+
+ IcebergSnapshotsRequestBody requestBody =
+ IcebergSnapshotsRequestBody.builder()
+ .baseTableVersion("v1")
+ .jsonSnapshots(Collections.singletonList(RequestConstants.TEST_ICEBERG_SNAPSHOT_JSON))
+ .snapshotRefs(refs)
+ .jsonMetadataUpdates(
+ Collections.singletonList(setSnapshotRef("b", HEAD_SNAPSHOT_ID, "branch")))
+ .createUpdateTableRequestBody(RequestConstants.TEST_CREATE_TABLE_REQUEST_BODY)
+ .build();
+
+ assertEquals("b", putSnapshots(requestBody).getBranchRefName());
+ }
+
+ /**
+ * The same tie, with the ref map ordered so "main" is encountered first. Under the previous
+ * snapshot-matching heuristic the answer depended on {@link HashMap} iteration order and could
+ * flip between runs; keyed off the commit's declared updates it is fixed.
+ */
+ @Test
+ public void testPutIcebergSnapshotsCreateBranchIsDeterministicRegardlessOfRefOrder()
+ throws Exception {
+ Map refs = new LinkedHashMap<>();
+ refs.put("main", TEST_HEAD_SNAPSHOT_REF_JSON);
+ refs.put("aaa_sorts_first", TEST_HEAD_SNAPSHOT_REF_JSON);
+ refs.put("zzz_sorts_last", TEST_HEAD_SNAPSHOT_REF_JSON);
+
+ IcebergSnapshotsRequestBody requestBody =
+ IcebergSnapshotsRequestBody.builder()
+ .baseTableVersion("v1")
+ .jsonSnapshots(Collections.singletonList(RequestConstants.TEST_ICEBERG_SNAPSHOT_JSON))
+ .snapshotRefs(refs)
+ .jsonMetadataUpdates(
+ Collections.singletonList(
+ setSnapshotRef("zzz_sorts_last", HEAD_SNAPSHOT_ID, "branch")))
+ .createUpdateTableRequestBody(RequestConstants.TEST_CREATE_TABLE_REQUEST_BODY)
+ .build();
+
+ assertEquals("zzz_sorts_last", putSnapshots(requestBody).getBranchRefName());
+ }
+
+ /**
+ * {@code CREATE TAG} carries {@code "type": "tag"}. A tag is not a branch, so branchRefName stays
+ * null rather than reporting a tag name in a field documented as a branch.
+ */
+ @Test
+ public void testPutIcebergSnapshotsTagCommitLeavesBranchRefNameNull() throws Exception {
+ Map refs = new HashMap<>();
+ refs.put("main", TEST_HEAD_SNAPSHOT_REF_JSON);
+ refs.put("v1_release", "{\"snapshot-id\":" + HEAD_SNAPSHOT_ID + ",\"type\":\"tag\"}");
+
+ IcebergSnapshotsRequestBody requestBody =
+ IcebergSnapshotsRequestBody.builder()
+ .baseTableVersion("v1")
+ .jsonSnapshots(Collections.singletonList(RequestConstants.TEST_ICEBERG_SNAPSHOT_JSON))
+ .snapshotRefs(refs)
+ .jsonMetadataUpdates(
+ Collections.singletonList(setSnapshotRef("v1_release", HEAD_SNAPSHOT_ID, "tag")))
+ .createUpdateTableRequestBody(RequestConstants.TEST_CREATE_TABLE_REQUEST_BODY)
+ .build();
+
+ TableAuditEvent actualEvent = putSnapshots(requestBody);
+ assertNull(actualEvent.getBranchRefName());
+ // The tag commit does not move main, but main's snapshot info is still reported.
+ assertEquals(HEAD_SNAPSHOT_ID, actualEvent.getCurrentSnapshotId().longValue());
+ }
+
+ /**
+ * {@code DROP BRANCH b} removes a ref and commits nothing. No branch was written, so
+ * branchRefName stays null; {@code remove-snapshot-ref} is deliberately not treated as a write.
+ */
+ @Test
+ public void testPutIcebergSnapshotsDropBranchLeavesBranchRefNameNull() throws Exception {
+ IcebergSnapshotsRequestBody requestBody =
+ IcebergSnapshotsRequestBody.builder()
+ .baseTableVersion("v1")
+ .jsonSnapshots(Collections.singletonList(RequestConstants.TEST_ICEBERG_SNAPSHOT_JSON))
+ .snapshotRefs(Collections.singletonMap("main", TEST_HEAD_SNAPSHOT_REF_JSON))
+ .jsonMetadataUpdates(
+ Collections.singletonList(
+ "{\"action\":\"remove-snapshot-ref\",\"ref-name\":\"b\"}"))
+ .createUpdateTableRequestBody(RequestConstants.TEST_CREATE_TABLE_REQUEST_BODY)
+ .build();
+
+ assertNull(putSnapshots(requestBody).getBranchRefName());
+ }
+
+ /**
+ * Clients predating {@code jsonMetadataUpdates} omit it. branchRefName is then left unset rather
+ * than guessed — an absent audit field beats one that is wrong on ties.
+ */
+ @Test
+ public void testPutIcebergSnapshotsWithoutMetadataUpdatesLeavesBranchRefNameNull()
+ throws Exception {
+ IcebergSnapshotsRequestBody legacyRequestBody =
+ IcebergSnapshotsRequestBody.builder()
+ .baseTableVersion("v1")
+ .jsonSnapshots(Collections.singletonList(RequestConstants.TEST_ICEBERG_SNAPSHOT_JSON))
+ .snapshotRefs(Collections.singletonMap("main", TEST_HEAD_SNAPSHOT_REF_JSON))
+ .createUpdateTableRequestBody(RequestConstants.TEST_CREATE_TABLE_REQUEST_BODY)
+ .build();
+
+ TableAuditEvent actualEvent = putSnapshots(legacyRequestBody);
+ assertNull(actualEvent.getBranchRefName());
+ // Everything else on the legacy path is unaffected.
+ assertEquals(HEAD_SNAPSHOT_ID, actualEvent.getCurrentSnapshotId().longValue());
+ assertEquals(1669126937912L, actualEvent.getCurrentSnapshotTimestampMs().longValue());
+ }
+
+ /** A malformed action must not hide the well-formed ones around it. */
+ @Test
+ public void testPutIcebergSnapshotsSkipsUnparseableMetadataUpdate() throws Exception {
+ IcebergSnapshotsRequestBody requestBody =
+ IcebergSnapshotsRequestBody.builder()
+ .baseTableVersion("v1")
+ .jsonSnapshots(Collections.singletonList(RequestConstants.TEST_ICEBERG_SNAPSHOT_JSON))
+ .snapshotRefs(Collections.singletonMap("main", TEST_HEAD_SNAPSHOT_REF_JSON))
+ .jsonMetadataUpdates(
+ Arrays.asList(
+ "{\"action\":\"not-a-real-action\"}",
+ "}{ malformed json",
+ setSnapshotRef("feature", HEAD_SNAPSHOT_ID, "branch")))
+ .createUpdateTableRequestBody(RequestConstants.TEST_CREATE_TABLE_REQUEST_BODY)
+ .build();
+
+ assertEquals("feature", putSnapshots(requestBody).getBranchRefName());
+ }
+
@Test
public void testPutIcebergSnapshotsBranchOnlyCommitLeavesSnapshotInfoNull() throws Exception {
- // Simulate a branch-only commit where main is absent from snapshotRefs.
- // In this case the main branch ref doesn't exist, so currentSnapshotId /
- // currentSnapshotTimestampMs should be null.
+ // Simulate a branch-only commit where main is absent from snapshotRefs entirely.
+ // currentSnapshotId / currentSnapshotTimestampMs are null (no main), but branchRefName
+ // is still populated from the ref that received the new snapshot.
IcebergSnapshotsRequestBody branchOnlyRequestBody =
IcebergSnapshotsRequestBody.builder()
.baseTableVersion("v1")
@@ -137,6 +334,8 @@ public void testPutIcebergSnapshotsBranchOnlyCommitLeavesSnapshotInfoNull() thro
.snapshotRefs(
Collections.singletonMap(
"my_branch", "{\"snapshot-id\":2151407017102313398,\"type\":\"branch\"}"))
+ .jsonMetadataUpdates(
+ Collections.singletonList(setSnapshotRef("my_branch", HEAD_SNAPSHOT_ID, "branch")))
.createUpdateTableRequestBody(RequestConstants.TEST_CREATE_TABLE_REQUEST_BODY)
.build();
@@ -150,6 +349,7 @@ public void testPutIcebergSnapshotsBranchOnlyCommitLeavesSnapshotInfoNull() thro
.content(branchOnlyRequestBody.toJson()));
Mockito.verify(tableAuditHandler, atLeastOnce()).audit(argCaptor.capture());
TableAuditEvent actualEvent = argCaptor.getValue();
+ assertEquals("my_branch", actualEvent.getBranchRefName());
assertNull(actualEvent.getCurrentSnapshotId());
assertNull(actualEvent.getCurrentSnapshotTimestampMs());
}
@@ -185,6 +385,8 @@ public void testPutIcebergSnapshotsMainPointsToOlderSnapshot() throws Exception
.baseTableVersion("v1")
.jsonSnapshots(Arrays.asList(olderSnapshotJson, newerSnapshotJson))
.snapshotRefs(refs)
+ .jsonMetadataUpdates(
+ Collections.singletonList(setSnapshotRef("feature", 200L, "branch")))
.createUpdateTableRequestBody(RequestConstants.TEST_CREATE_TABLE_REQUEST_BODY)
.build();
@@ -200,6 +402,34 @@ public void testPutIcebergSnapshotsMainPointsToOlderSnapshot() throws Exception
TableAuditEvent actualEvent = argCaptor.getValue();
assertEquals(100L, actualEvent.getCurrentSnapshotId().longValue());
assertEquals(1000L, actualEvent.getCurrentSnapshotTimestampMs().longValue());
+ // The commit declared it moved feature; main is untouched despite sharing the ref map.
+ assertEquals("feature", actualEvent.getBranchRefName());
+ }
+
+ /** The snapshot id carried by {@link RequestConstants#TEST_ICEBERG_SNAPSHOT_JSON}. */
+ private static final long HEAD_SNAPSHOT_ID = 2151407017102313398L;
+
+ private static final String TEST_HEAD_SNAPSHOT_REF_JSON =
+ "{\"snapshot-id\":" + HEAD_SNAPSHOT_ID + ",\"type\":\"branch\"}";
+
+ /** Builds one Iceberg REST spec {@code set-snapshot-ref} action. */
+ private static String setSnapshotRef(String refName, long snapshotId, String type) {
+ return String.format(
+ "{\"action\":\"set-snapshot-ref\",\"ref-name\":\"%s\",\"snapshot-id\":%d,\"type\":\"%s\"}",
+ refName, snapshotId, type);
+ }
+
+ private TableAuditEvent putSnapshots(IcebergSnapshotsRequestBody requestBody) throws Exception {
+ mvc.perform(
+ MockMvcRequestBuilders.put(
+ String.format(
+ CURRENT_MAJOR_VERSION_PREFIX
+ + "/databases/d200/tables/tb1/iceberg/v2/snapshots"))
+ .accept(MediaType.APPLICATION_JSON)
+ .contentType(MediaType.APPLICATION_JSON)
+ .content(requestBody.toJson()));
+ Mockito.verify(tableAuditHandler, atLeastOnce()).audit(argCaptor.capture());
+ return argCaptor.getValue();
}
@Test
diff --git a/services/tables/src/test/java/com/linkedin/openhouse/tables/model/TableAuditModelConstants.java b/services/tables/src/test/java/com/linkedin/openhouse/tables/model/TableAuditModelConstants.java
index b7486f617..1a9f5eab1 100644
--- a/services/tables/src/test/java/com/linkedin/openhouse/tables/model/TableAuditModelConstants.java
+++ b/services/tables/src/test/java/com/linkedin/openhouse/tables/model/TableAuditModelConstants.java
@@ -225,6 +225,7 @@ public final class TableAuditModelConstants {
.operationType(OperationType.COMMIT)
.currentSnapshotId(2151407017102313398L)
.currentSnapshotTimestampMs(1669126937912L)
+ .branchRefName("main")
.build();
public static final TableAuditEvent TABLE_AUDIT_EVENT_PUT_ICEBERG_SNAPSHOTS_FAILED =
@@ -237,6 +238,7 @@ public final class TableAuditModelConstants {
.operationType(OperationType.COMMIT)
.currentSnapshotId(2151407017102313398L)
.currentSnapshotTimestampMs(1669126937912L)
+ .branchRefName("main")
.build();
public static final TableAuditEvent TABLE_AUDIT_EVENT_PUT_ICEBERG_SNAPSHOTS_CTAS =
@@ -249,6 +251,7 @@ public final class TableAuditModelConstants {
.operationType(OperationType.STAGED_COMMIT)
.currentSnapshotId(2151407017102313398L)
.currentSnapshotTimestampMs(1669126937912L)
+ .branchRefName("main")
.build();
public static final TableAuditEvent TABLE_AUDIT_EVENT_GET_ALL_DATABASES_SUCCESS =