From 41f36f20fcc8b3a5136d3ad7b3ac97f055d347f3 Mon Sep 17 00:00:00 2001 From: Christian Bush Date: Sun, 2 Aug 2026 21:07:24 -0700 Subject: [PATCH 1/2] Carry Iceberg REST TableUpdate deltas on the snapshots commit request MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds an optional `jsonMetadataUpdates` field to IcebergSnapshotsRequestBody holding a commit's deltas as Iceberg REST spec `TableUpdate` actions, and populates it client-side from `TableMetadata.changes()`. Why: `jsonSnapshots` and `snapshotRefs` carry complete replacement state, so the request describes what the table now looks like but never what the commit actually changed. The server recovers the deltas by diffing the incoming snapshot set against prior state (OpenHouseInternalTableOperations), and some changes cannot be recovered at all — `CREATE BRANCH b` adds a ref at the current head and commits no snapshot, so the resulting state is indistinguishable from a no-op on main. The Iceberg REST spec already models this: `CommitTableRequest.updates[]` is a list of `TableUpdate` actions, where that operation is a single `set-snapshot-ref` with `ref-name: b`, `type: branch`, and no `add-snapshot`. The client has the list for free via `TableMetadata.changes()` — the same one every REST catalog sends — and `MetadataUpdateParser` emits the spec wire format verbatim. This ships the spec shape rather than a lookalike, so when OpenHouse adopts the REST commit endpoint the field is promoted to `updates` and the full-state fields retire, with no re-modeling. Additive and advisory: the server still builds table metadata from jsonSnapshots/snapshotRefs and is not changed here, clients predating the field omit it, and consumers must tolerate null/empty. Serialization failures are swallowed and individual unserializable actions skipped, so the field can never fail a commit. One source file serves both the iceberg-1.2 and iceberg-1.5 runtimes. No consumer yet — the audit path and the server-side delta handling land separately. Tests assert `changes()` yields the expected spec actions per operation: CREATE BRANCH emits only a branch-typed `set-snapshot-ref`, CREATE TAG is tag-typed, DROP BRANCH is a `remove-snapshot-ref`, and an append reports both `add-snapshot` and the ref that moved. Note `changes()` accumulates across builds within a session, so the fixture discards construction history to match production, where the base always comes from a refresh parsed off disk. --- ...useTableOperationsMetadataUpdatesTest.java | 183 ++++++++++++++++++ .../javaclient/OpenHouseTableOperations.java | 42 ++++ .../request/IcebergSnapshotsRequestBody.java | 27 +++ 3 files changed, 252 insertions(+) create mode 100644 integrations/java/iceberg-1.2/openhouse-java-itest/src/test/java/com/linkedin/openhouse/javaclient/OpenHouseTableOperationsMetadataUpdatesTest.java 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; From 3962fe331700f8596d2332d6e68e88e481947501 Mon Sep 17 00:00:00 2001 From: Christian Bush Date: Sun, 2 Aug 2026 21:15:45 -0700 Subject: [PATCH 2/2] Add branchRefName to TableAuditEvent for Git for Data observability MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Table operations against named Iceberg branches are not currently observable: TableAuditEvent records currentSnapshotId and currentSnapshotTimestampMs but carries no signal for which branch ref a commit wrote. Add branchRefName, populated from the `set-snapshot-ref` action in the request's jsonMetadataUpdates. The commit states which ref it moved and whether that ref is a branch or a tag, so the field reports what happened rather than inferring it. Inferring it from the resulting table state does not work. The obvious approach — match the last snapshot in jsonSnapshots to the ref pointing at it — fails on the operation this field exists to observe: `CREATE BRANCH b` creates a ref at the current head and commits no snapshot, so main and b both point at the last snapshot and the answer depends on HashMap iteration order. It would also treat tags as branches, since SnapshotRefParser parses both and ref type is absent from that comparison, and it assumes the client always serializes snapshots chronologically. None of those hold for ref-only commits. Clients predating jsonMetadataUpdates omit it; branchRefName is then left unset rather than guessed, since an absent audit field is preferable to one that is wrong roughly half the time. Unparseable actions are skipped individually so a single bad entry cannot hide the rest, and extraction stays best-effort — it cannot fail the request. currentSnapshotId and currentSnapshotTimestampMs continue to track the main ref unchanged, for backwards compatibility. This reports what the client declared, not what the server committed. Making it authoritative means having doCommit surface the refs it actually changed, which it already computes internally; that is a separate change. Tests cover CREATE BRANCH at head, determinism across ref iteration order, CREATE TAG, DROP BRANCH, clients omitting the field, and malformed actions. --- .../tables/audit/TableAuditAspect.java | 78 ++++-- .../tables/audit/model/TableAuditEvent.java | 2 + .../tables/mock/RequestConstants.java | 15 ++ .../IcebergSnapshotsApiHandlerAuditTest.java | 236 +++++++++++++++++- .../model/TableAuditModelConstants.java | 3 + 5 files changed, 316 insertions(+), 18 deletions(-) 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 =