diff --git a/integrations/java/iceberg-1.2/openhouse-java-itest/src/test/java/com/linkedin/openhouse/javaclient/OpenHouseTableOperationsTest.java b/integrations/java/iceberg-1.2/openhouse-java-itest/src/test/java/com/linkedin/openhouse/javaclient/OpenHouseTableOperationsTest.java index 1ddfd9713..5be88bb17 100644 --- a/integrations/java/iceberg-1.2/openhouse-java-itest/src/test/java/com/linkedin/openhouse/javaclient/OpenHouseTableOperationsTest.java +++ b/integrations/java/iceberg-1.2/openhouse-java-itest/src/test/java/com/linkedin/openhouse/javaclient/OpenHouseTableOperationsTest.java @@ -25,14 +25,19 @@ import java.util.Map; import java.util.Set; import org.apache.commons.compress.utils.Lists; +import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.Schema; +import org.apache.iceberg.SchemaParser; import org.apache.iceberg.Snapshot; import org.apache.iceberg.TableMetadata; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.exceptions.CommitStateUnknownException; import org.apache.iceberg.exceptions.NoSuchTableException; +import org.apache.iceberg.expressions.Expressions; import org.apache.iceberg.io.FileIO; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.types.Types.NestedField; import org.apache.iceberg.util.Tasks; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -494,13 +499,13 @@ private OpenHouseTableOperations refreshableOps(TableApi tableApi) { .build(); } - /** Before any refresh, there is no server-stamped config, so the safe default is null. */ + /** No refresh yet → no config. */ @Test public void testCurrentConfigNullBeforeRefresh() { Assertions.assertNull(refreshableOps(mock(TableApi.class)).currentConfig()); } - /** doRefresh stashes the server-stamped config so subclasses can read it back. */ + /** doRefresh stores the response config. */ @Test public void testDoRefreshCapturesConfig() { TableApi mockTableApi = mock(TableApi.class); @@ -517,7 +522,7 @@ public void testDoRefreshCapturesConfig() { Assertions.assertSame(stamped, ops.currentConfig()); } - /** Absent config on the response => null, the consumer's safe default. */ + /** Missing config on the response is stored as null. */ @Test public void testDoRefreshNullConfigWhenAbsent() { TableApi mockTableApi = mock(TableApi.class); @@ -532,11 +537,7 @@ public void testDoRefreshNullConfigWhenAbsent() { Assertions.assertNull(ops.currentConfig()); } - /** - * The held config is a snapshot of the latest refresh, never sticky: once the server stops - * stamping config, a subsequent refresh must clear the previously-captured value back to null. - * Guards against a stale directive lingering after the server turns it off. - */ + /** A later refresh without config clears the previous value. */ @Test public void testDoRefreshClearsStaleConfig() { TableApi mockTableApi = mock(TableApi.class); @@ -551,7 +552,7 @@ public void testDoRefreshClearsStaleConfig() { when(withoutConfig.getTableLocation()).thenReturn(null); when(withoutConfig.getConfig()).thenReturn(null); - // First refresh stamps config, second refresh stops stamping it. + // Second refresh has no config. when(mockTableApi.getTableV1(anyString(), anyString())) .thenReturn(Mono.just(withConfig)) .thenReturn(Mono.just(withoutConfig)); @@ -565,17 +566,13 @@ public void testDoRefreshClearsStaleConfig() { Assertions.assertNull(ops.currentConfig()); } - /** - * Bridge failures must not ride Iceberg's metadata-read retry. Decode runs before any FileIO - * access; {@link Tasks.UnrecoverableException} keeps {@code Tasks.retry(20)} from re-reading - * storage ~21 times (~90s) to reproduce a deterministic config error. - */ + /** Bad config fails before FileIO so Iceberg does not retry the metadata read. */ @Test public void testMalformedConfigFailsBeforeTouchingStorage() { TableApi mockTableApi = mock(TableApi.class); FileIO mockFileIO = mock(FileIO.class); GetTableResponseBody body = mock(GetTableResponseBody.class); - // Non-null location, so a metadata load would otherwise be attempted. + // Non-null location would otherwise trigger a metadata load. when(body.getTableLocation()).thenReturn("/tmp/does-not-matter/metadata.json"); when(body.getConfig()) .thenReturn( @@ -597,11 +594,7 @@ public void testMalformedConfigFailsBeforeTouchingStorage() { verifyNoInteractions(mockFileIO); } - /** - * Wire contract: a server-stamped config map deserializes on the client (the Iceberg REST {@code - * LoadTableResponse.config} convention — a string map). This is how the value actually arrives on - * a real table-load response. - */ + /** Config arrives as a string map on the table-load JSON. */ @Test public void testConfigDeserializeFromResponse() throws Exception { ObjectMapper mapper = ApiClient.createDefaultObjectMapper(null); @@ -612,14 +605,11 @@ public void testConfigDeserializeFromResponse() throws Exception { GetTableResponseBody body = mapper.readValue(json, GetTableResponseBody.class); Map config = body.getConfig(); Assertions.assertNotNull(config); - // value stays an opaque JSON string; the channel never parses it. + // Channel does not parse the value. Assertions.assertEquals("{\"read\":\"ON\"}", config.get("openhouse.read-bridge")); } - /** - * Unknown future fields must not break deserialization — older clients ignore what they do not - * understand (FAIL_ON_UNKNOWN_PROPERTIES=false), and unknown config keys are simply carried. - */ + /** Unknown JSON fields and unknown config keys are carried, not rejected. */ @Test public void testConfigToleratesUnknownFields() throws Exception { ObjectMapper mapper = ApiClient.createDefaultObjectMapper(null); @@ -632,4 +622,67 @@ public void testConfigToleratesUnknownFields() throws Exception { Assertions.assertNotNull(config); Assertions.assertEquals("whatever", config.get("openhouse.unknown-feature")); } + + @Test + public void constructMetadataRequestBody_stripsOverlayOnExistingIdsKeepsNewColumnDefaults() { + TableMetadata raw = + tableWithSchema( + "file:/tmp/rb-sanitize-ops", + new Schema( + NestedField.optional(1, "id", Types.IntegerType.get()), + NestedField.optional(2, "country", Types.StringType.get()))); + TableMetadata commit = + tableWithSchema( + "file:/tmp/rb-sanitize-ops-c", + new Schema( + NestedField.optional(1, "id", Types.IntegerType.get()), + NestedField.from(NestedField.optional(2, "country", Types.StringType.get())) + .withInitialDefault(Expressions.lit("US")) + .build(), + NestedField.from(NestedField.optional(3, "email", Types.StringType.get())) + .withInitialDefault(Expressions.lit("none")) + .build())); + + OpenHouseTableOperations ops = refreshableOps(mock(TableApi.class)); + ops.stashRawMetadata(raw); + + CreateUpdateTableRequestBody body = ops.constructMetadataRequestBody(raw, commit); + Schema sent = SchemaParser.fromJson(body.getSchema()); + + Assertions.assertNull(sent.findField(2).initialDefault()); + Assertions.assertEquals("none", sent.findField(3).initialDefault()); + Assertions.assertEquals("email", sent.findField(3).name()); + } + + @Test + public void constructMetadataRequestBody_withoutRawLeavesWriterDefaults() { + TableMetadata commit = + tableWithSchema( + "file:/tmp/rb-sanitize-create", + new Schema( + NestedField.from(NestedField.optional(1, "country", Types.StringType.get())) + .withInitialDefault(Expressions.lit("US")) + .build())); + + CreateUpdateTableRequestBody body = + refreshableOps(mock(TableApi.class)).constructMetadataRequestBody(null, commit); + + Assertions.assertEquals( + "US", SchemaParser.fromJson(body.getSchema()).findField(1).initialDefault()); + } + + private static TableMetadata tableWithSchema(String location, Schema schema) { + TableMetadata created = + TableMetadata.newTableMetadata( + schema, PartitionSpec.unpartitioned(), location, Collections.emptyMap()); + return ReadBridge.replaceSchemas( + created, + Collections.singletonMap( + created.currentSchemaId(), + new Schema( + created.currentSchemaId(), + schema.columns(), + schema.getAliases(), + schema.identifierFieldIds()))); + } } diff --git a/integrations/java/iceberg-1.2/openhouse-java-itest/src/test/java/com/linkedin/openhouse/javaclient/ReadBridgeTest.java b/integrations/java/iceberg-1.2/openhouse-java-itest/src/test/java/com/linkedin/openhouse/javaclient/ReadBridgeTest.java index a3f15a9d1..b81993b00 100644 --- a/integrations/java/iceberg-1.2/openhouse-java-itest/src/test/java/com/linkedin/openhouse/javaclient/ReadBridgeTest.java +++ b/integrations/java/iceberg-1.2/openhouse-java-itest/src/test/java/com/linkedin/openhouse/javaclient/ReadBridgeTest.java @@ -1,6 +1,7 @@ package com.linkedin.openhouse.javaclient; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -8,39 +9,41 @@ import java.util.Collections; import java.util.HashMap; import java.util.Map; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.TableMetadata; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.types.Types.NestedField; import org.junit.jupiter.api.Test; -/** - * Unit tests for the client-side read-bridge config decoder ({@link ReadBridge#from}), exercised in - * isolation. Mirrors the server-side encoder {@code ReadBridgeConfigResolver}. - */ +/** Decoder, apply, and sanitize path for {@link ReadBridge}. */ class ReadBridgeTest { private static final String PREFIX = ReadBridge.COLUMN_DEFAULT_PREFIX; @Test void decodesColumnDefaultsByFieldId() { - // Inline calls avoid naming Jackson's JsonNode, which is relocated in the shaded client uber - // (and this module compiles at a source level without `var`). Map config = new HashMap<>(); config.put(PREFIX + "5", "\"US\""); config.put(PREFIX + "7", "0"); - assertEquals(2, ReadBridge.from(config).columnDefaults().size()); - assertEquals("US", ReadBridge.from(config).columnDefaults().get(5).asText()); - assertEquals(0, ReadBridge.from(config).columnDefaults().get(7).asInt()); + ReadBridge bridge = ReadBridge.from(config); + assertEquals(2, bridge.columnDefaults().size()); + // Original JSON strings so apply can bind without a relocated JsonNode. + assertEquals("\"US\"", bridge.columnDefaults().get(5)); + assertEquals("0", bridge.columnDefaults().get(7)); } @Test void inertWhenConfigNullOrNoReadBridgeKeys() { assertSame(ReadBridge.INERT, ReadBridge.from(null)); assertSame(ReadBridge.INERT, ReadBridge.from(Collections.singletonMap("other.key", "x"))); - assertTrue(ReadBridge.INERT.columnDefaults().isEmpty()); + assertTrue(ReadBridge.from(null).columnDefaults().isEmpty()); } @Test void failsLoudOnKnownEntryWithBadFieldId() { - // A non-integer field-id on a key we own can't come from the server encoder (it stamps int - // field-ids and JsonNode values), so it's a bug/corruption and throws rather than degrading. + // Non-integer suffix on a key we own is a bug, not a missing default. Map config = new HashMap<>(); config.put(PREFIX + "5", "\"US\""); config.put(PREFIX + "notAnInt", "\"x\""); @@ -56,12 +59,313 @@ void failsLoudOnKnownEntryWithUnparseableValue() { @Test void ignoresUnknownKeysWithoutFailing() { - // Forward compatibility: a key outside the column-default prefix (e.g. a newer server feature) - // is ignored, never enforced — even if its value would not parse as a default. + // Keys outside the prefix are ignored so a newer server stays readable. Map config = new HashMap<>(); config.put(PREFIX + "5", "\"US\""); config.put("openhouse.read-bridge.some-future-feature.3", "{not a default}"); - assertEquals(1, ReadBridge.from(config).columnDefaults().size()); - assertEquals("US", ReadBridge.from(config).columnDefaults().get(5).asText()); + ReadBridge bridge = ReadBridge.from(config); + assertEquals(1, bridge.columnDefaults().size()); + assertEquals("\"US\"", bridge.columnDefaults().get(5)); + } + + @Test + void applyReturnsSameInstanceWhenNothingToBridge() { + TableMetadata raw = newTable("file:/tmp/rb-inert"); + assertSame(raw, ReadBridge.INERT.apply(raw)); + assertSame(raw, ReadBridge.from(Collections.singletonMap("other.key", "x")).apply(raw)); + } + + @Test + void applySetsInitialDefaultOnMatchingField() { + TableMetadata raw = newTable("file:/tmp/rb-apply"); + Map config = Collections.singletonMap(PREFIX + "2", "\"US\""); + + TableMetadata bridged = ReadBridge.from(config).apply(raw); + + assertEquals("US", bridged.schema().findField(2).initialDefault()); + assertNull(bridged.schema().findField(1).initialDefault()); + // Disk identity is unchanged; only in-memory schemas carry the overlay. + assertEquals(raw.uuid(), bridged.uuid()); + assertEquals(raw.currentSchemaId(), bridged.currentSchemaId()); + assertEquals(raw.metadataFileLocation(), bridged.metadataFileLocation()); + } + + @Test + void applyOverlaysEverySchemaId() { + Schema v0 = + new Schema( + 0, + Types.NestedField.optional(1, "id", Types.IntegerType.get()), + Types.NestedField.optional(2, "country", Types.StringType.get())); + Schema v1 = + new Schema( + 1, + Types.NestedField.optional(1, "id", Types.IntegerType.get()), + Types.NestedField.optional(2, "country", Types.StringType.get()), + Types.NestedField.optional(3, "region", Types.StringType.get())); + TableMetadata raw = + TableMetadata.buildFrom( + TableMetadata.newTableMetadata( + v0, + PartitionSpec.unpartitioned(), + "file:/tmp/rb-multischema", + Collections.emptyMap())) + .addSchema(v1, 3) + .setCurrentSchema(1) + .build(); + + Map config = new HashMap<>(); + config.put(PREFIX + "2", "\"US\""); + config.put(PREFIX + "3", "\"west\""); + + TableMetadata bridged = ReadBridge.from(config).apply(raw); + + assertEquals(2, bridged.schemas().size()); + for (Schema schema : bridged.schemas()) { + assertEquals("US", schema.findField(2).initialDefault()); + } + // Field 3 exists only on schema 1. + assertNull(bridged.schemasById().get(0).findField(3)); + assertEquals("west", bridged.schemasById().get(1).findField(3).initialDefault()); + } + + @Test + void applyIgnoresFieldIdsAbsentFromAllSchemas() { + TableMetadata raw = newTable("file:/tmp/rb-gap"); + Map config = Collections.singletonMap(PREFIX + "99", "\"x\""); + assertSame(raw, ReadBridge.from(config).apply(raw)); + } + + @Test + void applyFailsLoudWhenDefaultCannotBindToColumnType() { + TableMetadata raw = newTable("file:/tmp/rb-bad-bind"); + // Field 1 is int; a string default cannot bind. + Map config = Collections.singletonMap(PREFIX + "1", "\"not-an-int\""); + assertThrows(IllegalStateException.class, () -> ReadBridge.from(config).apply(raw)); + } + + @Test + void applySetsDefaultOnNestedStructField() { + Schema schema = + new Schema( + Types.NestedField.optional(1, "id", Types.IntegerType.get()), + Types.NestedField.optional( + 2, + "address", + Types.StructType.of( + Types.NestedField.optional(3, "country", Types.StringType.get())))); + TableMetadata raw = + TableMetadata.newTableMetadata( + schema, PartitionSpec.unpartitioned(), "file:/tmp/rb-nested", Collections.emptyMap()); + Map config = Collections.singletonMap(PREFIX + "3", "\"US\""); + + TableMetadata bridged = ReadBridge.from(config).apply(raw); + + assertEquals( + "US", bridged.schema().findField(2).type().asStructType().field(3).initialDefault()); + } + + @Test + void sanitizeAfterApplyRestoresOnDiskSchema() { + TableMetadata raw = newTable("file:/tmp/rb-roundtrip"); + Map config = Collections.singletonMap(PREFIX + "2", "\"US\""); + + TableMetadata sanitized = ReadBridge.sanitize(raw, ReadBridge.from(config).apply(raw)); + + assertEquals(raw.schema().asStruct(), sanitized.schema().asStruct()); + } + + private static TableMetadata newTable(String location) { + Schema schema = + new Schema( + Types.NestedField.optional(1, "id", Types.IntegerType.get()), + Types.NestedField.optional(2, "country", Types.StringType.get())); + return TableMetadata.newTableMetadata( + schema, PartitionSpec.unpartitioned(), location, Collections.emptyMap()); + } + + @Test + void sanitizeReturnsSameInstanceWhenRawIsNullOrIdentical() { + TableMetadata raw = newTable("file:/tmp/rb-sanitize-same", twoColumns(null, null)); + assertSame(raw, ReadBridge.sanitize(null, raw)); + assertSame(raw, ReadBridge.sanitize(raw, raw)); + assertSame(null, ReadBridge.sanitize(raw, null)); + } + + @Test + void sanitizeRestoresDefaultsOnFieldIdsThatExistedOnDisk() { + TableMetadata raw = newTable("file:/tmp/rb-sanitize-restore", twoColumns(null, null)); + TableMetadata commit = newTable("file:/tmp/rb-sanitize-restore-c", twoColumns(null, "US")); + + TableMetadata sanitized = ReadBridge.sanitize(raw, commit); + + assertNull(sanitized.schema().findField(1).initialDefault()); + assertNull(sanitized.schema().findField(2).initialDefault()); + assertEquals(raw.schema().asStruct(), sanitized.schema().asStruct()); + } + + @Test + void sanitizeKeepsWriterDefaultsOnNewFieldIds() { + TableMetadata raw = newTable("file:/tmp/rb-sanitize-add", twoColumns(null, null)); + TableMetadata commit = + newTable( + "file:/tmp/rb-sanitize-add-c", + new Schema( + optionalInt(1, "id"), + withInitialDefault(optionalString(2, "country"), "US"), + withInitialDefault(optionalString(3, "email"), "none"))); + + TableMetadata sanitized = ReadBridge.sanitize(raw, commit); + + assertNull(sanitized.schema().findField(2).initialDefault()); + assertEquals("none", sanitized.schema().findField(3).initialDefault()); + assertEquals("email", sanitized.schema().findField(3).name()); + } + + @Test + void sanitizePreservesRenameAndTypeWidenOnExistingIds() { + TableMetadata raw = newTable("file:/tmp/rb-sanitize-evolve", twoColumns(null, null)); + TableMetadata commit = + newTable( + "file:/tmp/rb-sanitize-evolve-c", + new Schema( + NestedField.from(optionalInt(1, "id")).ofType(Types.LongType.get()).build(), + withInitialDefault(optionalString(2, "nation"), "US"))); + + TableMetadata sanitized = ReadBridge.sanitize(raw, commit); + + NestedField id = sanitized.schema().findField(1); + assertEquals(Types.LongType.get(), id.type()); + assertNull(id.initialDefault()); + NestedField nation = sanitized.schema().findField(2); + assertEquals("nation", nation.name()); + assertNull(nation.initialDefault()); + } + + @Test + void sanitizeRestoresWriteDefaultOnExistingIds() { + TableMetadata raw = + newTable( + "file:/tmp/rb-sanitize-write", + new Schema( + optionalInt(1, "id"), + NestedField.from(optionalString(2, "country")) + .withWriteDefault(Expressions.lit("CA")) + .build())); + + TableMetadata commit = + newTable( + "file:/tmp/rb-sanitize-write-c", + new Schema( + optionalInt(1, "id"), + NestedField.from(optionalString(2, "country")) + .withInitialDefault(Expressions.lit("US")) + .withWriteDefault(Expressions.lit("MX")) + .build())); + + TableMetadata sanitized = ReadBridge.sanitize(raw, commit); + + NestedField country = sanitized.schema().findField(2); + assertNull(country.initialDefault()); + assertEquals("CA", country.writeDefault()); + } + + @Test + void sanitizeRestoresNestedExistingIdsAndKeepsNewNestedIds() { + TableMetadata raw = + newTable( + "file:/tmp/rb-sanitize-nested", + new Schema( + optionalInt(1, "id"), + NestedField.optional( + 2, "address", Types.StructType.of(optionalString(3, "country"))))); + + TableMetadata commit = + newTable( + "file:/tmp/rb-sanitize-nested-c", + new Schema( + optionalInt(1, "id"), + NestedField.optional( + 2, + "address", + Types.StructType.of( + withInitialDefault(optionalString(3, "country"), "US"), + withInitialDefault(optionalString(4, "region"), "west"))))); + + TableMetadata sanitized = ReadBridge.sanitize(raw, commit); + + Types.StructType address = sanitized.schema().findField(2).type().asStructType(); + assertNull(address.field(3).initialDefault()); + assertEquals("west", address.field(4).initialDefault()); + } + + @Test + void sanitizeRestoresEverySchemaId() { + TableMetadata raw = newTable("file:/tmp/rb-sanitize-multi", twoColumns(null, null)); + TableMetadata commit = + TableMetadata.buildFrom( + newTable( + "file:/tmp/rb-sanitize-multi-c", + new Schema( + optionalInt(1, "id"), + withInitialDefault(optionalString(2, "country"), "US")))) + .addSchema( + new Schema( + 1, + optionalInt(1, "id"), + withInitialDefault(optionalString(2, "country"), "US"), + withInitialDefault(optionalString(3, "region"), "west")), + 3) + .setCurrentSchema(1) + .build(); + + TableMetadata sanitized = ReadBridge.sanitize(raw, commit); + + assertEquals(2, sanitized.schemas().size()); + for (Schema schema : sanitized.schemas()) { + assertNull(schema.findField(2).initialDefault()); + } + assertNull(sanitized.schemasById().get(0).findField(3)); + assertEquals("west", sanitized.schemasById().get(1).findField(3).initialDefault()); + } + + private static TableMetadata newTable(String location, Schema schema) { + TableMetadata created = + TableMetadata.newTableMetadata( + schema, PartitionSpec.unpartitioned(), location, Collections.emptyMap()); + // newTableMetadata reassigns ids and drops defaults; put this schema back with defaults. + return ReadBridge.replaceSchemas( + created, + Collections.singletonMap( + created.currentSchemaId(), + new Schema( + created.currentSchemaId(), + schema.columns(), + schema.getAliases(), + schema.identifierFieldIds()))); + } + + private static Schema twoColumns(String idDefault, String countryDefault) { + NestedField id = optionalInt(1, "id"); + NestedField country = optionalString(2, "country"); + if (idDefault != null) { + id = withInitialDefault(id, idDefault); + } + if (countryDefault != null) { + country = withInitialDefault(country, countryDefault); + } + return new Schema(id, country); + } + + private static NestedField optionalInt(int id, String name) { + return NestedField.optional(id, name, Types.IntegerType.get()); + } + + private static NestedField optionalString(int id, String name) { + return NestedField.optional(id, name, Types.StringType.get()); + } + + private static NestedField withInitialDefault(NestedField field, String value) { + return NestedField.from(field).withInitialDefault(Expressions.lit(value)).build(); } } 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 52d4cd3fa..d58fd093d 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 @@ -64,16 +64,17 @@ public class OpenHouseTableOperations extends BaseMetastoreTableOperations { private String cluster; /** - * The per-table client {@code config} (Iceberg REST {@code LoadTableResponse.config} convention) - * the OH server stamped onto the most recent table-load response (or {@code null} if none / not - * yet refreshed). A final holder keeps Lombok's all-args constructor unchanged. + * Config from the last refresh, or {@code null}. Atomic so Lombok's constructor stays unchanged. */ private final AtomicReference> config = new AtomicReference<>(); /** - * The server-stamped per-table client config from the last {@code doRefresh}, or {@code null} - * when absent. Subclasses read it to gate read-time behavior. + * On-disk metadata from the last successful load, before apply. Commit restores default slots + * from this copy so overlays cannot persist. */ + private final AtomicReference lastRawMetadata = new AtomicReference<>(); + + /** Config from the last refresh, or {@code null}. */ protected Map currentConfig() { return config.get(); } @@ -114,29 +115,23 @@ public void doRefresh() { WebClientRequestException.class, e -> Mono.error(new WebClientRequestWithMessageException(e))) .blockOptional(); - // Capture the server-stamped per-table config so subclasses can gate read-time behavior via - // currentConfig(); absent => null. Side-channel only: never sent back on writes. + // Keep config from the GET response; it is not a table property. this.config.set(tableResponse.map(GetTableResponseBody::getConfig).orElse(null)); Optional tableLocation = tableResponse.map(GetTableResponseBody::getTableLocation); if (!tableLocation.isPresent() && currentMetadataLocation() != null) { throw new NoSuchTableException( "Cannot find table %s after refresh, maybe another process deleted it", tableName()); } - // Route the parse through loadMetadata() so subclasses can transform metadata as it loads; - // (null, 20) preserves the stock refresh behavior. + // Parse via loadMetadata so ReadBridge can overlay after the file read. super.refreshFromMetadataLocation(tableLocation.orElse(null), null, 20, this::loadMetadata); log.debug("Calling doRefresh succeeded"); } /** - * Loads table metadata from storage and overlays the read-time {@link ReadBridge} behavior the - * server stamped onto {@link #currentConfig()}. + * Decode config, then read the metadata file, then overlay. * - *

Decode runs before the file read so a malformed config never touches storage. - * Bridge failures ({@link IllegalStateException}) are wrapped as {@link - * Tasks.UnrecoverableException} so Iceberg's {@code Tasks.retry(20)} around this loader does not - * re-read the metadata file to reproduce a deterministic error. IO / parse failures from the file - * read remain retryable. + *

Decode first so a bad config never hits storage. {@link IllegalStateException} is wrapped as + * {@link Tasks.UnrecoverableException} so Iceberg's retry loop does not re-read the file. */ protected TableMetadata loadMetadata(String metadataLocation) { final ReadBridge bridge; @@ -147,7 +142,9 @@ protected TableMetadata loadMetadata(String metadataLocation) { } TableMetadata raw = TableMetadataParser.read(io(), metadataLocation); try { - return bridge.apply(raw); + TableMetadata loaded = bridge.apply(raw); + lastRawMetadata.set(raw); + return loaded; } catch (IllegalStateException e) { throw new Tasks.UnrecoverableException(e); } @@ -219,6 +216,8 @@ private void createUpdateTable(TableMetadata base, TableMetadata metadata) { protected CreateUpdateTableRequestBody constructMetadataRequestBody( TableMetadata base, TableMetadata metadata) { + // Iceberg commit() requires base == current(); strip overlays here, not by swapping base. + metadata = ReadBridge.sanitize(lastRawMetadata.get(), metadata); CreateUpdateTableRequestBody createUpdateTableRequestBody = new CreateUpdateTableRequestBody(); createUpdateTableRequestBody.setBaseTableVersion( base == null ? INITIAL_TABLE_VERSION : base.metadataFileLocation()); @@ -263,6 +262,11 @@ && getTableType(base, metadata) return createUpdateTableRequestBody; } + @VisibleForTesting + void stashRawMetadata(TableMetadata raw) { + lastRawMetadata.set(raw); + } + /** * If request is coming from replication process, createUpdateTableRequestBody.tableType should be * REPLICA_TABLE Replication process requests are identified based on difference between table diff --git a/integrations/java/iceberg-1.2/openhouse-java-runtime/src/main/java/com/linkedin/openhouse/javaclient/ReadBridge.java b/integrations/java/iceberg-1.2/openhouse-java-runtime/src/main/java/com/linkedin/openhouse/javaclient/ReadBridge.java index d3a2334af..4ecccfe29 100644 --- a/integrations/java/iceberg-1.2/openhouse-java-runtime/src/main/java/com/linkedin/openhouse/javaclient/ReadBridge.java +++ b/integrations/java/iceberg-1.2/openhouse-java-runtime/src/main/java/com/linkedin/openhouse/javaclient/ReadBridge.java @@ -3,112 +3,137 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ArrayNode; +import com.fasterxml.jackson.databind.node.ObjectNode; +import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; +import java.util.Objects; +import org.apache.iceberg.Schema; +import org.apache.iceberg.SchemaParser; +import org.apache.iceberg.SingleValueParser; import org.apache.iceberg.TableMetadata; +import org.apache.iceberg.TableMetadataParser; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.types.Types.NestedField; /** - * Read-time bridge: overlays Iceberg V3 read semantics onto loaded metadata for tables/clients that - * don't yet carry them natively, using behavior the server delivers in the per-table {@code - * config}. Today it applies per-column initial-defaults; further V3 features can be backported - * through the same entry point as they are added. + * Overlays server-stamped read-time behavior from table {@code config} onto loaded Iceberg + * metadata. * - *

Client end of the read-bridge wire contract — mirror of the server encoder {@code - * ReadBridgeConfigResolver} (services/tables). The contract is flat, namespaced config keys (no - * envelope/POJO): {@code openhouse.read-bridge.column-default. = }. + *

Keys: {@code openhouse.read-bridge.column-default. = }. {@link + * #from} decodes; {@link #apply} overlays. Unknown keys are ignored. A malformed known entry throws + * — that is an encoder or transport bug, not a missing default. * - *

Decode before IO; mark bridge failures unrecoverable

+ *

Apply rebuilds schemas with {@code NestedField.withInitialDefault} and puts them back through + * {@link TableMetadataParser} JSON. Missing field-id on a schema is a gap (NULL); a default that + * cannot bind throws. * - *

{@link #from(Map)} decodes the config; {@link #apply(TableMetadata)} overlays the result onto - * loaded metadata. {@link OpenHouseTableOperations#loadMetadata} calls {@code from} before - * reading the metadata file so a malformed config never touches storage, then {@code apply} after. - * Both steps throw {@link IllegalStateException} on invariant violations; the loader wraps those as - * Iceberg's {@code Tasks.UnrecoverableException} so {@code Tasks.retry(20)} around the metadata - * read does not burn ~90s re-reading the file to reproduce a deterministic failure. - * - *

A read-bridge entry is produced by the server encoder from typed {@code JsonNode}s keyed by - * integer field-id, so its value always round-trips through {@code readTree} and its suffix always - * parses as an int. A decode failure on a known entry is therefore a bug or transport - * corruption, not an expected runtime state, and this fails loud rather than silently degrading to - * NULL. An unknown key (a newer server feature this client doesn't recognize) is ignored, - * preserving forward compatibility. With nothing to bridge, metadata is returned unchanged. - * - *

Note this guarantees only that a stamped value is well-formed, not that it is the - * correct default for its column — that semantic (default-to-schema) consistency is a - * write-time concern owned by whatever server path sources the defaults. + *

{@link #sanitize} restores default slots on field-ids that existed in the last on-disk + * metadata so an overlay cannot persist. New field-ids keep the writer's defaults. */ final class ReadBridge { - /** Mirror of {@code ReadBridgeConfigResolver.COLUMN_DEFAULT_PREFIX}. */ + /** Same prefix the server encoder stamps. */ static final String COLUMN_DEFAULT_PREFIX = "openhouse.read-bridge.column-default."; - /** Nothing to bridge; {@link #apply(TableMetadata)} returns metadata untouched. */ + /** {@link #apply} is a no-op. */ static final ReadBridge INERT = new ReadBridge(Collections.emptyMap()); private static final ObjectMapper MAPPER = new ObjectMapper(); - private final Map columnDefaults; + private static final String FORMAT_VERSION = "format-version"; + private static final String SCHEMA = "schema"; + private static final String SCHEMAS = "schemas"; + private static final String SCHEMA_ID = "schema-id"; + + /** JSON strings, not JsonNodes — Jackson is relocated in the shaded client. */ + private final Map columnDefaults; - private ReadBridge(Map columnDefaults) { + private ReadBridge(Map columnDefaults) { this.columnDefaults = columnDefaults; } /** - * Decodes the read-bridge behavior the server stamped into {@code config}, returning {@link - * #INERT} when there is nothing to bridge. + * Decode stamped config. Returns {@link #INERT} when there is nothing to apply. * - * @throws IllegalStateException if an entry this client owns is malformed (encoder bug or - * transport corruption); unknown keys are ignored. + * @throws IllegalStateException if a key this client owns is malformed */ static ReadBridge from(Map config) { - Map columnDefaults = columnDefaults(config); + Map columnDefaults = decodeColumnDefaults(config); return columnDefaults.isEmpty() ? INERT : new ReadBridge(columnDefaults); } - /** - * Applies the bridged read-time behavior onto {@code raw}, returning the transformed metadata (or - * {@code raw} when there is nothing to bridge). - */ + /** Overlay onto {@code raw}, or return it unchanged. */ TableMetadata apply(TableMetadata raw) { if (columnDefaults.isEmpty()) { return raw; } - // TODO(read-bridge): overlay columnDefaults onto raw.schemas() via withSchemaOverlay; future V3 - // features bridged from config are applied here too. Two failure categories apply there, as - // here: a capability gap we don't yet support degrades to NULL, while an invariant violation - // (e.g. a default that can't bind to its column) fails loud. - return raw; - } - /** The decoded {@code field-id -> initial-default} entries. Package-visible for testing. */ - Map columnDefaults() { - return columnDefaults; + Map overlaidById = new HashMap<>(); + for (Schema schema : raw.schemas()) { + Schema overlaid = overlaySchema(schema, columnDefaults); + if (overlaid != schema) { + overlaidById.put(schema.schemaId(), overlaid); + } + } + if (overlaidById.isEmpty()) { + // Every stamped field-id is missing from every schema — leave metadata as-is. + return raw; + } + return replaceSchemas(raw, overlaidById); } /** - * Decodes {@code field-id -> initial-default} from the {@code - * openhouse.read-bridge.column-default.*} config entries; empty when there are none. On a known - * entry, the server encoder guarantees an integer field-id and a value that round-trips through - * {@code readTree}, so a non-integer field-id or an unparseable value is an encoder bug or - * transport corruption — it throws rather than degrading. Unknown keys are ignored above. + * Restore {@code initialDefault} / {@code writeDefault} on field-ids that existed in {@code raw}. + * Name, type, nullability, doc, order, and new field-ids stay on {@code metadata}. + * + *

A field-id is overlay iff it was on disk at load. Apply only stamps existing ids; V2 + * evolution does not set defaults on those ids. A field-id absent from {@code raw} was added in + * this commit — keep the writer's defaults. */ - private static Map columnDefaults(Map config) { + static TableMetadata sanitize(TableMetadata raw, TableMetadata metadata) { + if (raw == null || metadata == null || raw == metadata) { + return metadata; + } + Map rawById = indexFields(raw); + Map restoredById = new HashMap<>(); + for (Schema schema : metadata.schemas()) { + Schema restored = restoreSchema(schema, rawById); + if (restored != schema) { + restoredById.put(schema.schemaId(), restored); + } + } + if (restoredById.isEmpty()) { + return metadata; + } + return replaceSchemas(metadata, restoredById); + } + + Map columnDefaults() { + return columnDefaults; + } + + private static Map decodeColumnDefaults(Map config) { if (config == null) { return Collections.emptyMap(); } - Map byFieldId = new HashMap<>(); + Map byFieldId = new HashMap<>(); for (Map.Entry entry : config.entrySet()) { if (!entry.getKey().startsWith(COLUMN_DEFAULT_PREFIX)) { continue; } try { int fieldId = Integer.parseInt(entry.getKey().substring(COLUMN_DEFAULT_PREFIX.length())); - byFieldId.put(fieldId, MAPPER.readTree(entry.getValue())); + // Validate JSON; keep the original string so apply can bind without a relocated JsonNode. + MAPPER.readTree(entry.getValue()); + byFieldId.put(fieldId, entry.getValue()); } catch (RuntimeException | JsonProcessingException e) { - // The server encoder stamps an int field-id and a JsonNode value that round-trips through - // readTree, so reaching here means an encoder bug or transport corruption, not an expected - // state. Fail loud so it is caught, rather than silently reading NULL. + // Known keys are stamped as int field-id + JSON; anything else is a bug. throw new IllegalStateException( "read-bridge: unusable " + COLUMN_DEFAULT_PREFIX @@ -121,4 +146,244 @@ private static Map columnDefaults(Map config) } return byFieldId; } + + private static Schema overlaySchema(Schema schema, Map columnDefaults) { + List columns = schema.columns(); + List overlaid = new ArrayList<>(columns.size()); + boolean changed = false; + for (NestedField column : columns) { + NestedField next = overlayField(column, columnDefaults); + overlaid.add(next); + if (next != column) { + changed = true; + } + } + if (!changed) { + return schema; + } + return new Schema( + schema.schemaId(), overlaid, schema.getAliases(), schema.identifierFieldIds()); + } + + private static NestedField overlayField(NestedField field, Map columnDefaults) { + Type type = field.type(); + Type overlaidType = overlayType(type, columnDefaults); + String defaultJson = columnDefaults.get(field.fieldId()); + + if (defaultJson == null && overlaidType == type) { + return field; + } + + NestedField.Builder builder = NestedField.from(field); + if (overlaidType != type) { + builder.ofType(overlaidType); + } + if (defaultJson != null) { + try { + Object value = SingleValueParser.fromJson(overlaidType, defaultJson); + builder.withInitialDefault(Expressions.lit(value)); + } catch (RuntimeException e) { + throw new IllegalStateException( + "read-bridge: cannot bind " + + COLUMN_DEFAULT_PREFIX + + field.fieldId() + + "=" + + defaultJson + + " to " + + field, + e); + } + } + return builder.build(); + } + + private static Type overlayType(Type type, Map columnDefaults) { + if (type.isStructType()) { + List fields = type.asStructType().fields(); + List overlaid = new ArrayList<>(fields.size()); + boolean changed = false; + for (NestedField field : fields) { + NestedField next = overlayField(field, columnDefaults); + overlaid.add(next); + if (next != field) { + changed = true; + } + } + return changed ? Types.StructType.of(overlaid) : type; + } + if (type.isListType()) { + Types.ListType list = type.asListType(); + Type element = overlayType(list.elementType(), columnDefaults); + if (element == list.elementType()) { + return type; + } + return list.isElementRequired() + ? Types.ListType.ofRequired(list.elementId(), element) + : Types.ListType.ofOptional(list.elementId(), element); + } + if (type.isMapType()) { + Types.MapType map = type.asMapType(); + Type key = overlayType(map.keyType(), columnDefaults); + Type value = overlayType(map.valueType(), columnDefaults); + if (key == map.keyType() && value == map.valueType()) { + return type; + } + return map.isValueRequired() + ? Types.MapType.ofRequired(map.keyId(), map.valueId(), key, value) + : Types.MapType.ofOptional(map.keyId(), map.valueId(), key, value); + } + return type; + } + + private static Map indexFields(TableMetadata raw) { + Map byId = new HashMap<>(); + for (Schema schema : raw.schemas()) { + collectFields(schema.asStruct(), byId); + } + Schema current = raw.schema(); + if (current != null) { + collectFields(current.asStruct(), byId); + } + return byId; + } + + private static void collectFields(Type type, Map byId) { + if (type.isStructType()) { + for (NestedField field : type.asStructType().fields()) { + byId.put(field.fieldId(), field); + collectFields(field.type(), byId); + } + } else if (type.isListType()) { + collectFields(type.asListType().elementType(), byId); + } else if (type.isMapType()) { + Types.MapType map = type.asMapType(); + collectFields(map.keyType(), byId); + collectFields(map.valueType(), byId); + } + } + + private static Schema restoreSchema(Schema schema, Map rawById) { + List columns = schema.columns(); + List restored = new ArrayList<>(columns.size()); + boolean changed = false; + for (NestedField column : columns) { + NestedField next = restoreField(column, rawById); + restored.add(next); + if (next != column) { + changed = true; + } + } + if (!changed) { + return schema; + } + return new Schema( + schema.schemaId(), restored, schema.getAliases(), schema.identifierFieldIds()); + } + + private static NestedField restoreField(NestedField field, Map rawById) { + Type type = field.type(); + Type restoredType = restoreType(type, rawById); + NestedField rawField = rawById.get(field.fieldId()); + if (rawField == null) { + if (restoredType == type) { + return field; + } + return NestedField.from(field).ofType(restoredType).build(); + } + boolean defaultsSame = + Objects.equals(field.initialDefaultLiteral(), rawField.initialDefaultLiteral()) + && Objects.equals(field.writeDefaultLiteral(), rawField.writeDefaultLiteral()); + if (defaultsSame && restoredType == type) { + return field; + } + NestedField.Builder builder = NestedField.from(field); + if (restoredType != type) { + builder.ofType(restoredType); + } + if (!defaultsSame) { + builder.withInitialDefault(rawField.initialDefaultLiteral()); + builder.withWriteDefault(rawField.writeDefaultLiteral()); + } + return builder.build(); + } + + private static Type restoreType(Type type, Map rawById) { + if (type.isStructType()) { + List fields = type.asStructType().fields(); + List restored = new ArrayList<>(fields.size()); + boolean changed = false; + for (NestedField field : fields) { + NestedField next = restoreField(field, rawById); + restored.add(next); + if (next != field) { + changed = true; + } + } + return changed ? Types.StructType.of(restored) : type; + } + if (type.isListType()) { + Types.ListType list = type.asListType(); + Type element = restoreType(list.elementType(), rawById); + if (element == list.elementType()) { + return type; + } + return list.isElementRequired() + ? Types.ListType.ofRequired(list.elementId(), element) + : Types.ListType.ofOptional(list.elementId(), element); + } + if (type.isMapType()) { + Types.MapType map = type.asMapType(); + Type key = restoreType(map.keyType(), rawById); + Type value = restoreType(map.valueType(), rawById); + if (key == map.keyType() && value == map.valueType()) { + return type; + } + return map.isValueRequired() + ? Types.MapType.ofRequired(map.keyId(), map.valueId(), key, value) + : Types.MapType.ofOptional(map.keyId(), map.valueId(), key, value); + } + return type; + } + + /** + * Swap rebuilt schemas in by rewriting metadata JSON. The public builder will not replace an + * existing schema-id. + */ + static TableMetadata replaceSchemas( + TableMetadata metadata, Map replacementById) { + try { + ObjectNode root = (ObjectNode) MAPPER.readTree(TableMetadataParser.toJson(metadata)); + JsonNode schemasNode = root.get(SCHEMAS); + if (schemasNode == null || !schemasNode.isArray()) { + throw new IllegalStateException( + "read-bridge: metadata JSON missing required '" + SCHEMAS + "' array"); + } + ArrayNode schemas = (ArrayNode) schemasNode; + for (int i = 0; i < schemas.size(); i++) { + JsonNode schemaNode = schemas.get(i); + if (schemaNode == null || !schemaNode.has(SCHEMA_ID)) { + throw new IllegalStateException( + "read-bridge: schemas[" + i + "] missing '" + SCHEMA_ID + "'"); + } + int schemaId = schemaNode.get(SCHEMA_ID).asInt(); + Schema replacement = replacementById.get(schemaId); + if (replacement != null) { + schemas.set(i, MAPPER.readTree(SchemaParser.toJson(replacement))); + } + } + // v1 also writes the current schema under "schema"; keep it in sync. + if (root.path(FORMAT_VERSION).asInt(/* default= */ 2) == 1) { + Schema currentReplacement = replacementById.get(metadata.currentSchemaId()); + if (currentReplacement != null) { + root.set(SCHEMA, MAPPER.readTree(SchemaParser.toJson(currentReplacement))); + } + } + return TableMetadataParser.fromJson( + metadata.metadataFileLocation(), MAPPER.writeValueAsString(root)); + } catch (IllegalStateException e) { + throw e; + } catch (RuntimeException | JsonProcessingException e) { + throw new IllegalStateException("read-bridge: failed to rebuild table metadata schemas", e); + } + } } diff --git a/services/tables/src/main/java/com/linkedin/openhouse/tables/api/ApiConfig.java b/services/tables/src/main/java/com/linkedin/openhouse/tables/api/ApiConfig.java index 00408f757..4c7655993 100644 --- a/services/tables/src/main/java/com/linkedin/openhouse/tables/api/ApiConfig.java +++ b/services/tables/src/main/java/com/linkedin/openhouse/tables/api/ApiConfig.java @@ -4,12 +4,12 @@ import com.linkedin.openhouse.tables.api.handler.impl.OpenHouseTablesApiHandler; import com.linkedin.openhouse.tables.readbridge.ColumnDefaultsSource; import com.linkedin.openhouse.tables.readbridge.ReadBridgeConfigResolver; -import java.util.Collections; -import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import com.linkedin.openhouse.tables.toggle.TableFeatureToggle; +import org.springframework.beans.factory.ObjectProvider; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -/** Class that holds all the Beans related to a controller. */ +/** Beans related to tables API controllers. */ @Configuration public class ApiConfig { @Bean @@ -17,23 +17,11 @@ public TablesApiHandler tablesApiHandler() { return new OpenHouseTablesApiHandler(); } - /** - * Open-source default {@link ColumnDefaultsSource}: supplies none, so read-bridge stays inert. - */ + /** ObjectProvider so a deployment bean does not collide with an OSS default. */ @Bean - @ConditionalOnMissingBean(ColumnDefaultsSource.class) - public ColumnDefaultsSource columnDefaultsSource() { - return tableDto -> Collections.emptyMap(); - } - - /** - * Server-side encoder that stamps the read-bridge {@code config} from {@link - * ColumnDefaultsSource}. - */ - @Bean - @ConditionalOnMissingBean(ReadBridgeConfigResolver.class) public ReadBridgeConfigResolver readBridgeConfigResolver( - ColumnDefaultsSource columnDefaultsSource) { - return new ReadBridgeConfigResolver(columnDefaultsSource); + ObjectProvider columnDefaultsSource, TableFeatureToggle featureToggle) { + return new ReadBridgeConfigResolver( + columnDefaultsSource.getIfAvailable(() -> ColumnDefaultsSource.NONE), featureToggle); } } diff --git a/services/tables/src/main/java/com/linkedin/openhouse/tables/api/handler/impl/OpenHouseTablesApiHandler.java b/services/tables/src/main/java/com/linkedin/openhouse/tables/api/handler/impl/OpenHouseTablesApiHandler.java index 8de0d0a49..e7f0835c4 100644 --- a/services/tables/src/main/java/com/linkedin/openhouse/tables/api/handler/impl/OpenHouseTablesApiHandler.java +++ b/services/tables/src/main/java/com/linkedin/openhouse/tables/api/handler/impl/OpenHouseTablesApiHandler.java @@ -38,16 +38,9 @@ public class OpenHouseTablesApiHandler implements TablesApiHandler { @Autowired private ReadBridgeConfigResolver readBridgeConfigResolver; - /** - * Stamp the server-resolved, per-table client {@code config} (Iceberg REST {@code - * LoadTableResponse.config} convention) onto a freshly mapped response body. The mapper leaves - * {@code config} null; it is a request-time decision resolved here. - */ - private GetTableResponseBody withConfig( - GetTableResponseBody body, String databaseId, String tableId, TableDto tableDto) { - return body.toBuilder() - .config(readBridgeConfigResolver.resolve(databaseId, tableId, tableDto)) - .build(); + /** Config is request-time; the mapper does not persist it. */ + private GetTableResponseBody withConfig(GetTableResponseBody body, TableDto tableDto) { + return body.toBuilder().config(readBridgeConfigResolver.resolve(tableDto)).build(); } @Override @@ -57,9 +50,7 @@ public ApiResponse getTable( TableDto tableDto = tableService.getTable(databaseId, tableId, actingPrincipal); return ApiResponse.builder() .httpStatus(HttpStatus.OK) - .responseBody( - withConfig( - tablesMapper.toGetTableResponseBody(tableDto), databaseId, tableId, tableDto)) + .responseBody(withConfig(tablesMapper.toGetTableResponseBody(tableDto), tableDto)) .build(); } @@ -111,12 +102,7 @@ public ApiResponse createTable( TableDto tableDto = putResult.getFirst(); return ApiResponse.builder() .httpStatus(HttpStatus.CREATED) - .responseBody( - withConfig( - tablesMapper.toGetTableResponseBody(tableDto), - databaseId, - tableDto.getTableId(), - tableDto)) + .responseBody(withConfig(tablesMapper.toGetTableResponseBody(tableDto), tableDto)) .build(); } @@ -134,9 +120,7 @@ public ApiResponse updateTable( TableDto tableDto = putResult.getFirst(); return ApiResponse.builder() .httpStatus(status) - .responseBody( - withConfig( - tablesMapper.toGetTableResponseBody(tableDto), databaseId, tableId, tableDto)) + .responseBody(withConfig(tablesMapper.toGetTableResponseBody(tableDto), tableDto)) .build(); } diff --git a/services/tables/src/main/java/com/linkedin/openhouse/tables/readbridge/ColumnDefaultsSource.java b/services/tables/src/main/java/com/linkedin/openhouse/tables/readbridge/ColumnDefaultsSource.java index 74444119a..82392fc26 100644 --- a/services/tables/src/main/java/com/linkedin/openhouse/tables/readbridge/ColumnDefaultsSource.java +++ b/services/tables/src/main/java/com/linkedin/openhouse/tables/readbridge/ColumnDefaultsSource.java @@ -2,28 +2,20 @@ import com.fasterxml.jackson.databind.JsonNode; import com.linkedin.openhouse.tables.model.TableDto; +import java.util.Collections; import java.util.Map; /** - * Pluggable input to the open-source {@code read-bridge} feature: the per-column initial-defaults - * to overlay at read time, keyed by Iceberg field-id and valued as Iceberg single-value JSON. - * - *

This is the only part of read-bridge a deployment supplies. The open-source default (see - * {@code ApiConfig}) returns nothing, so the feature is wired but inert until a deployment - * overrides this bean (e.g. li-openhouse derives the defaults from the {@code avro.schema.literal} - * table property). - * - *

Called on every table-load/commit response, so implementations must be cheap. An empty map - * means "nothing to bridge for this table" — no default is declared, or a declared default is of a - * kind this source does not support; either way the column keeps reading {@code NULL} as it does - * today. Throw instead when a default is declared but cannot be honored (e.g. it does not - * bind to its column's type): degrading there would leave the column reading {@code NULL} while the - * table claims to be bridged, hiding a real defect. + * Column defaults for one table, keyed by Iceberg field-id. Values are Iceberg single-value JSON. + * Ramp is {@link ReadBridgeConfigResolver}. */ public interface ColumnDefaultsSource { + + /** Used when no deployment bean is registered. */ + ColumnDefaultsSource NONE = tableDto -> Collections.emptyMap(); + /** - * @param tableDto the already-loaded table state (no extra fetch needed) - * @return field-id -> initial-default as Iceberg single-value JSON; empty/{@code null} = none + * Field-id to default JSON. Empty/null stamps nothing. Throw if a declared default cannot bind. */ Map defaults(TableDto tableDto); } diff --git a/services/tables/src/main/java/com/linkedin/openhouse/tables/readbridge/ReadBridgeConfigResolver.java b/services/tables/src/main/java/com/linkedin/openhouse/tables/readbridge/ReadBridgeConfigResolver.java index 405bb2d3b..1f93de533 100644 --- a/services/tables/src/main/java/com/linkedin/openhouse/tables/readbridge/ReadBridgeConfigResolver.java +++ b/services/tables/src/main/java/com/linkedin/openhouse/tables/readbridge/ReadBridgeConfigResolver.java @@ -2,46 +2,75 @@ import com.fasterxml.jackson.databind.JsonNode; import com.linkedin.openhouse.tables.model.TableDto; +import com.linkedin.openhouse.tables.toggle.TableFeatureToggle; import java.util.Collections; import java.util.HashMap; import java.util.Map; +import lombok.extern.slf4j.Slf4j; /** - * Open-source encoder for the {@code read-bridge} feature: it asks the pluggable {@link - * ColumnDefaultsSource} for a table's column initial-defaults and stamps each as a namespaced entry - * in the per-table {@code config} — {@code openhouse.read-bridge.column-default. = - * }. The client decoder ({@code ReadBridge} in {@code openhouse-java-runtime}) - * reads these entries and overlays the defaults at metadata-load time. - * - *

No envelope/POJO: the flat config map (Iceberg REST {@code LoadTableResponse.config} - * convention) carries the structure directly. Behaviorless by default — the open-source {@link - * ColumnDefaultsSource} bean supplies nothing (see {@code ApiConfig}), so no entries are stamped. A - * deployment delivers the bridge by overriding only {@link ColumnDefaultsSource}. - * - *

Mirror: {@link #COLUMN_DEFAULT_PREFIX} is the shared contract with the client decoder; - * keep it in sync. Further V3 features ride the same {@code openhouse.read-bridge.*} namespace as - * additional keys. + * Builds the per-table {@code config} map the client reads. OpenHouse owns ramp and keys; {@link + * ColumnDefaultsSource} supplies the values. */ +@Slf4j public class ReadBridgeConfigResolver { - /** Config key prefix for a per-column read-time default; suffixed with the Iceberg field-id. */ - public static final String COLUMN_DEFAULT_PREFIX = "openhouse.read-bridge.column-default."; + /** Also names {@code .enabled} and the config key prefix. */ + public static final String COLUMN_DEFAULT_FEATURE_ID = "read-bridge.column-default"; + + /** {@code openhouse.read-bridge.column-default.}. */ + public static final String COLUMN_DEFAULT_PREFIX = "openhouse." + COLUMN_DEFAULT_FEATURE_ID + "."; private final ColumnDefaultsSource columnDefaultsSource; - public ReadBridgeConfigResolver(ColumnDefaultsSource columnDefaultsSource) { + private final TableFeatureToggle featureToggle; + + public ReadBridgeConfigResolver( + ColumnDefaultsSource columnDefaultsSource, TableFeatureToggle featureToggle) { this.columnDefaultsSource = columnDefaultsSource; + this.featureToggle = featureToggle; + } + + /** Per-table config the client applies at load. Empty when nothing is bridged. */ + public Map resolve(TableDto tableDto) { + Map config = new HashMap<>(); + config.putAll(columnDefaultConfig(tableDto)); + return config; } - public Map resolve(String databaseId, String tableId, TableDto tableDto) { + private Map columnDefaultConfig(TableDto tableDto) { + // No source registered: skip the HouseTables lookup. + if (columnDefaultsSource == ColumnDefaultsSource.NONE) { + return Collections.emptyMap(); + } + if (!isColumnDefaultRamped(tableDto)) { + return Collections.emptyMap(); + } Map columnDefaults = columnDefaultsSource.defaults(tableDto); if (columnDefaults == null || columnDefaults.isEmpty()) { - return Collections.emptyMap(); // nothing to bridge -> stamp nothing + return Collections.emptyMap(); } Map config = new HashMap<>(); - // JsonNode.toString() is the single-value JSON (e.g. "US" -> "\"US\"", 0 -> "0"). columnDefaults.forEach( (fieldId, value) -> config.put(COLUMN_DEFAULT_PREFIX + fieldId, value.toString())); return config; } + + /** + * Table property {@code read-bridge.column-default.enabled} overrides HouseTables. A lookup + * failure leaves the table unbridged (same as today's NULL reads). + */ + private boolean isColumnDefaultRamped(TableDto tableDto) { + try { + return featureToggle.isFeatureActivatedWithOverride(tableDto, COLUMN_DEFAULT_FEATURE_ID); + } catch (RuntimeException e) { + log.warn( + "read-bridge: toggle lookup failed for {}.{}; treating {} as not ramped", + tableDto.getDatabaseId(), + tableDto.getTableId(), + COLUMN_DEFAULT_FEATURE_ID, + e); + return false; + } + } } diff --git a/services/tables/src/test/java/com/linkedin/openhouse/tables/readbridge/ReadBridgeConfigResolverTest.java b/services/tables/src/test/java/com/linkedin/openhouse/tables/readbridge/ReadBridgeConfigResolverTest.java index a5e06a02f..e9161b2d3 100644 --- a/services/tables/src/test/java/com/linkedin/openhouse/tables/readbridge/ReadBridgeConfigResolverTest.java +++ b/services/tables/src/test/java/com/linkedin/openhouse/tables/readbridge/ReadBridgeConfigResolverTest.java @@ -3,7 +3,11 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.CALLS_REAL_METHODS; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; import static org.mockito.Mockito.when; import com.fasterxml.jackson.databind.JsonNode; @@ -16,6 +20,7 @@ import com.linkedin.openhouse.tables.dto.mapper.TablesMapper; import com.linkedin.openhouse.tables.model.TableDto; import com.linkedin.openhouse.tables.services.TablesService; +import com.linkedin.openhouse.tables.toggle.TableFeatureToggle; import java.util.Collections; import java.util.LinkedHashMap; import java.util.Map; @@ -25,23 +30,155 @@ public class ReadBridgeConfigResolverTest { - /** Open-source default source: supplies nothing, so the feature is inert. */ - private static final ColumnDefaultsSource NONE = tableDto -> Collections.emptyMap(); + private static final ColumnDefaultsSource NONE = ColumnDefaultsSource.NONE; private static final String PREFIX = ReadBridgeConfigResolver.COLUMN_DEFAULT_PREFIX; + /** Isolates encoding from ramp. */ + private static final TableFeatureToggle ALL_ON = + new TableFeatureToggle() { + @Override + public boolean isFeatureActivated(String databaseId, String tableId, String featureId) { + return true; + } + }; + + private static ReadBridgeConfigResolver resolverFor(ColumnDefaultsSource source) { + return new ReadBridgeConfigResolver(source, ALL_ON); + } + + private static ColumnDefaultsSource oneDefault() { + return tableDto -> Collections.singletonMap(5, TextNode.valueOf("US")); + } + + /** Table with an explicit {@code .enabled} property. */ + private static TableDto tableWithOverride(String value) { + return TableDto.builder() + .databaseId("db") + .tableId("tbl") + .tableProperties( + Collections.singletonMap( + ReadBridgeConfigResolver.COLUMN_DEFAULT_FEATURE_ID + + TableFeatureToggle.ENABLED_PROPERTY_SUFFIX, + value)) + .build(); + } + + /** No source → no HouseTables call. */ @Test - public void testEmptyWhenNoColumnDefaults() { + public void testInertAndSkipsToggleWhenNoSourceSupplied() { + TableFeatureToggle toggle = mock(TableFeatureToggle.class); + ReadBridgeConfigResolver resolver = + new ReadBridgeConfigResolver(ColumnDefaultsSource.NONE, toggle); + + Assertions.assertTrue(resolver.resolve(mock(TableDto.class)).isEmpty()); + verifyNoInteractions(toggle); + } + + /** HouseTables down → unbridged, not a failed read. */ + @Test + public void testToggleLookupFailureDegradesInsteadOfFailingTheRead() { + TableFeatureToggle exploding = + new TableFeatureToggle() { + @Override + public boolean isFeatureActivated(String databaseId, String tableId, String featureId) { + throw new IllegalStateException("housetables is down"); + } + }; + + Map config = + new ReadBridgeConfigResolver(oneDefault(), exploding) + .resolve(TableDto.builder().databaseId("db").tableId("tbl").build()); + + Assertions.assertTrue(config.isEmpty()); + } + + /** Unramped table is not asked for defaults. */ + @Test + public void testUnrampedTableIsNotBridgedAndSourceNotConsulted() { + ColumnDefaultsSource source = mock(ColumnDefaultsSource.class); + TableFeatureToggle allOff = + new TableFeatureToggle() { + @Override + public boolean isFeatureActivated(String databaseId, String tableId, String featureId) { + return false; + } + }; + + Assertions.assertTrue( + new ReadBridgeConfigResolver(source, allOff) + .resolve(TableDto.builder().databaseId("db").tableId("tbl").build()) + .isEmpty()); + // Deriving defaults can be expensive; check the ramp first. + verifyNoInteractions(source); + } + + /** {@code .enabled=true} wins over a server-side off. */ + @Test + public void testTablePropertyOptsInOverServerToggle() { + // Real override method; stub HTS so an accidental call would return false. + TableFeatureToggle toggle = mock(TableFeatureToggle.class, CALLS_REAL_METHODS); + when(toggle.isFeatureActivated(anyString(), anyString(), anyString())).thenReturn(false); + + Map config = + new ReadBridgeConfigResolver(oneDefault(), toggle).resolve(tableWithOverride("true")); + + Assertions.assertEquals("\"US\"", config.get(PREFIX + "5")); + verify(toggle, never()).isFeatureActivated(anyString(), anyString(), anyString()); + } + + /** {@code .enabled=false} wins over a server-side on. */ + @Test + public void testTablePropertyOptsOutOverServerToggle() { + Assertions.assertTrue(resolverFor(oneDefault()).resolve(tableWithOverride("false")).isEmpty()); + } + + /** Ramped table whose source has nothing to stamp. */ + @Test + public void testEmptyWhenSourceReturnsNoDefaults() { + ColumnDefaultsSource emptySource = mock(ColumnDefaultsSource.class); + when(emptySource.defaults(any())).thenReturn(Collections.emptyMap()); + + Assertions.assertTrue( + resolverFor(emptySource) + .resolve(TableDto.builder().databaseId("db").tableId("tbl").build()) + .isEmpty()); + verify(emptySource).defaults(any()); + } + + /** Id, property, and prefix are external contracts; keep them one token. */ + @Test + public void testFeatureIdPropertyAndKeysAreOneToken() { + Assertions.assertEquals( + "read-bridge.column-default", ReadBridgeConfigResolver.COLUMN_DEFAULT_FEATURE_ID); + Assertions.assertEquals( + "read-bridge.column-default.enabled", + ReadBridgeConfigResolver.COLUMN_DEFAULT_FEATURE_ID + + TableFeatureToggle.ENABLED_PROPERTY_SUFFIX); + Assertions.assertEquals( + "openhouse.read-bridge.column-default.", ReadBridgeConfigResolver.COLUMN_DEFAULT_PREFIX); + } + + /** Ramp is per capability, not a blanket read-bridge id. */ + @Test + public void testRolloutIdIsScopedToTheCapabilityNotTheMechanism() { + Assertions.assertNotEquals("read-bridge", ReadBridgeConfigResolver.COLUMN_DEFAULT_FEATURE_ID); + Assertions.assertNotEquals( + "v3-read-bridge", ReadBridgeConfigResolver.COLUMN_DEFAULT_FEATURE_ID); Assertions.assertTrue( - new ReadBridgeConfigResolver(NONE).resolve("db", "tbl", mock(TableDto.class)).isEmpty()); + ReadBridgeConfigResolver.COLUMN_DEFAULT_FEATURE_ID.startsWith("read-bridge.")); + } + + @Test + public void testEmptyWhenNoColumnDefaults() { + Assertions.assertTrue(resolverFor(NONE).resolve(mock(TableDto.class)).isEmpty()); } @Test public void testStampsColumnDefaultEntry() { ColumnDefaultsSource source = tableDto -> Collections.singletonMap(5, TextNode.valueOf("US")); - Map config = - new ReadBridgeConfigResolver(source).resolve("db", "tbl", mock(TableDto.class)); - // value is the single-value JSON for the default ("US" -> "\"US\""). + Map config = resolverFor(source).resolve(mock(TableDto.class)); + // "US" as Iceberg single-value JSON. Assertions.assertEquals("\"US\"", config.get(PREFIX + "5")); } @@ -54,14 +191,13 @@ public void testStampsAllColumnDefaultsAsSeparateEntries() { defaults.put(7, IntNode.valueOf(0)); return defaults; }; - Map config = - new ReadBridgeConfigResolver(source).resolve("db", "tbl", mock(TableDto.class)); + Map config = resolverFor(source).resolve(mock(TableDto.class)); Assertions.assertEquals(2, config.size()); Assertions.assertEquals("\"US\"", config.get(PREFIX + "5")); Assertions.assertEquals("0", config.get(PREFIX + "7")); } - /** getTable stamps the resolver's config onto the response body. */ + /** getTable puts resolver output on the response. */ @Test public void testGetTableStampsResolvedConfig() { TablesService tableService = mock(TablesService.class); @@ -74,7 +210,7 @@ public void testGetTableStampsResolvedConfig() { .thenReturn(GetTableResponseBody.builder().tableId("tbl").databaseId("db").build()); Map resolved = Collections.singletonMap(PREFIX + "5", "\"US\""); - when(resolver.resolve(eq("db"), eq("tbl"), eq(tableDto))).thenReturn(resolved); + when(resolver.resolve(eq(tableDto))).thenReturn(resolved); OpenHouseTablesApiHandler handler = handlerWith(tableService, tablesMapper, resolver); @@ -83,7 +219,7 @@ public void testGetTableStampsResolvedConfig() { Assertions.assertSame(resolved, response.getResponseBody().getConfig()); } - /** With the behaviorless open-source source wired in, getTable leaves config empty. */ + /** OSS source → empty config on getTable. */ @Test public void testGetTableLeavesConfigEmptyWithNoColumnDefaults() { TablesService tableService = mock(TablesService.class); @@ -94,8 +230,7 @@ public void testGetTableLeavesConfigEmptyWithNoColumnDefaults() { when(tablesMapper.toGetTableResponseBody(any())) .thenReturn(GetTableResponseBody.builder().tableId("tbl").databaseId("db").build()); - OpenHouseTablesApiHandler handler = - handlerWith(tableService, tablesMapper, new ReadBridgeConfigResolver(NONE)); + OpenHouseTablesApiHandler handler = handlerWith(tableService, tablesMapper, resolverFor(NONE)); ApiResponse response = handler.getTable("db", "tbl", "principal");