From 6877cbe910cc19a534b83f437555cb0b0a1c0aed Mon Sep 17 00:00:00 2001 From: Christian Bush Date: Wed, 12 Aug 2026 10:01:29 -0700 Subject: [PATCH 1/9] Tighten read-bridge comments to why, not design essays. Co-authored-by: Cursor --- .../OpenHouseTableOperationsTest.java | 35 +++------- .../openhouse/javaclient/ReadBridgeTest.java | 14 ++-- .../javaclient/OpenHouseTableOperations.java | 25 ++------ .../openhouse/javaclient/ReadBridge.java | 64 ++++--------------- 4 files changed, 33 insertions(+), 105 deletions(-) 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..a2aeda4d4 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 @@ -494,13 +494,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 +517,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 +532,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 +547,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 +561,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 +589,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 +600,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); 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..a5d3345a1 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 @@ -10,18 +10,14 @@ import java.util.Map; 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 for {@link ReadBridge#from}. */ 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`). + // Avoid naming JsonNode: it is relocated in the shaded client, and this module has no `var`. Map config = new HashMap<>(); config.put(PREFIX + "5", "\"US\""); config.put(PREFIX + "7", "0"); @@ -39,8 +35,7 @@ void inertWhenConfigNullOrNoReadBridgeKeys() { @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,8 +51,7 @@ 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}"); 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..83371c149 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,11 @@ 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. - */ + /** Config from the last refresh, or {@code null}. */ protected Map currentConfig() { return config.get(); } @@ -114,29 +109,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; 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..7a9e8cd07 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 @@ -9,41 +9,19 @@ import org.apache.iceberg.TableMetadata; /** - * 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. = }. - * - *

Decode before IO; mark bridge failures unrecoverable

- * - *

{@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. + *

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. */ 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(); @@ -55,44 +33,28 @@ private ReadBridge(Map 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); 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. + // TODO(read-bridge): overlay columnDefaults onto schemas. return raw; } - /** The decoded {@code field-id -> initial-default} entries. Package-visible for testing. */ Map columnDefaults() { return columnDefaults; } - /** - * 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. - */ private static Map columnDefaults(Map config) { if (config == null) { return Collections.emptyMap(); @@ -106,9 +68,7 @@ private static Map columnDefaults(Map config) int fieldId = Integer.parseInt(entry.getKey().substring(COLUMN_DEFAULT_PREFIX.length())); byFieldId.put(fieldId, MAPPER.readTree(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 From 9d95a7f5e91154d61a4da23072b14a8b71316efa Mon Sep 17 00:00:00 2001 From: Christian Bush Date: Tue, 11 Aug 2026 19:07:50 -0700 Subject: [PATCH 2/9] Move read-bridge column-default ramp into OpenHouse Keep the deployment-specific ColumnDefaultsSource optional and data-only; OpenHouse owns the read-bridge.column-default feature id, self-serve enabled property, and fail-open toggle lookup before asking for defaults. --- .../openhouse/tables/api/ApiConfig.java | 36 ++-- .../impl/OpenHouseTablesApiHandler.java | 22 +-- .../readbridge/ColumnDefaultsSource.java | 39 +++- .../readbridge/ReadBridgeConfigResolver.java | 171 ++++++++++++++++-- .../ReadBridgeConfigResolverTest.java | 163 ++++++++++++++++- 5 files changed, 364 insertions(+), 67 deletions(-) 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..510c5e44f 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,8 +4,8 @@ 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; @@ -18,22 +18,26 @@ public TablesApiHandler tablesApiHandler() { } /** - * Open-source default {@link ColumnDefaultsSource}: supplies none, so read-bridge stays inert. + * Server-side encoder that stamps the read-bridge {@code config}. + * + *

{@link ColumnDefaultsSource} is the column-default capability's single extension point, and + * it is resolved here rather than declared as an overridable default bean. A deployment supplies + * one; with none present that capability is inert and never consults the feature toggle. Each + * capability is wired, and rolled out, on its own. + * + *

Deliberately not a {@code @ConditionalOnMissingBean} default bean. Spring Boot documents + * that condition as safe only inside auto-configuration, and this is an ordinary + * {@code @Configuration}: a component-scanned override happens to work, because {@code + * ConfigurationClassPostProcessor} finishes scanning before it evaluates {@code @Bean} + * conditions, but a deployment declaring its source with {@code @Bean} in a configuration class + * parsed after this one would get a competing no-op bean and need {@code @Primary} to avoid a + * {@code NoUniqueBeanDefinitionException}. With {@link ObjectProvider} no default bean is ever + * registered, so exactly one bean of the type exists however it was declared. */ @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..5abf91557 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 @@ -43,11 +43,8 @@ public class OpenHouseTablesApiHandler implements TablesApiHandler { * 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(); + private GetTableResponseBody withConfig(GetTableResponseBody body, TableDto tableDto) { + return body.toBuilder().config(readBridgeConfigResolver.resolve(tableDto)).build(); } @Override @@ -57,9 +54,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 +106,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 +124,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..d5452e153 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,25 +2,44 @@ 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). + *

This is the only part of read-bridge a deployment supplies, and it supplies data only + * — not policy. Whether a table is bridged at all is decided by {@link ReadBridgeConfigResolver} + * from the open-source feature toggle, so an implementation neither consults nor knows about the + * ramp. Deriving the defaults — from whatever a deployment treats as the authority on a column's + * declared default — is the one deployment-specific step, and the only reason this interface + * exists. * - *

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. + *

No open-source default bean exists: {@code ApiConfig} resolves the type through an {@code + * ObjectProvider} and falls back to {@link #NONE}, so the feature is wired but completely inert out + * of the box — including skipping the toggle lookup entirely. + * + *

{@code JsonNode} rather than {@code String} is deliberate. It makes a stamped value + * well-formed by construction at the only place that produces one, which is what entitles + * the client decoder to treat a malformed entry as a bug and fail loud instead of degrading. + * + *

Called only for tables the ramp has activated, so an implementation may do real work (parsing + * a schema, say) without paying it on every table load fleet-wide. 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. */ public interface ColumnDefaultsSource { + + /** + * Supplies nothing. The value {@code ApiConfig} falls back to when a deployment supplies no + * source; {@link ReadBridgeConfigResolver} recognises it and short-circuits before the toggle. + */ + 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 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..627ce8032 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,38 +2,162 @@ 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. + * Open-source encoder for the {@code read-bridge} feature: for a table the ramp has activated, it + * asks the pluggable {@link ColumnDefaultsSource} for that 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}. + * convention) carries the structure directly. + * + *

Who decides what

+ * + * This class owns the policy — the feature id, the ramp, and the wire keys — and a + * deployment supplies only the data, via {@link ColumnDefaultsSource}. Keeping the ramp + * here means every deployment inherits it, the self-service property {@code read-bridge.enabled} is + * documented alongside the {@code openhouse.read-bridge.*} keys it controls, and a deployment's + * source is never asked to derive defaults for a table that is not bridged. + * + *

What capabilities share, and what they don't

+ * + * Capabilities bridged through this class share the infrastructure and nothing else: the + * per-table {@code config} channel, the {@code openhouse.read-bridge.*} namespace, and the client's + * decode/apply path. Rollout is never shared. Each capability has its own source, feature id, + * self-service table property and cluster kill switch, so it can be ramped, paused or killed + * without touching any other. There is deliberately no single switch, toggle id or property meaning + * "all of read-bridge". + * + *

{@link #resolve(TableDto)} therefore contains no cross-capability gate at all — it only merges + * what each capability decided for itself. A shared short-circuit there would couple rollouts that + * are meant to be independent: a deployment supplying a deletion-vector source but no + * column-default source must still get deletion vectors. + * + *

Combining rollouts later

+ * + * Independence is the default, not the ceiling. A superset ramp — say {@code v3-read-bridge}, + * activating every capability at once for tables that want the whole V3 read surface — is a natural + * later addition, and nothing here blocks it: it would be one more feature id consulted alongside + * the capability's own. The mechanism-wide ids ({@code read-bridge}, {@code v3-read-bridge}) are + * left unused today so one of them can take that role without colliding with a capability. It will + * need an explicit precedence rule; the sane one is that the more specific wins, so a table setting + * {@code read-bridge.column-default.enabled=false} stays opted out of that capability even while + * opted into the superset. + * + *

Gating, cheapest check first (per capability)

+ * + *
    + *
  1. No source supplied for the capability ({@link ColumnDefaultsSource#NONE}) — structurally + * inert, and notably makes no toggle lookup, so open-source and dev deployments add nothing + * to the table-load path. + *
  2. {@link TableFeatureToggle#isFeatureActivatedWithOverride} — the per-table ramp: an explicit + * {@code read-bridge.column-default.enabled} table property opts a table in or out, otherwise + * the server-managed toggle decides. Rules match database and table as globs, so a {@code *} + * / {@code *} rule ramps or un-ramps the fleet as data, taking effect immediately. That is + * the kill switch; there is deliberately no cluster property duplicating it, which would only + * add a second place to look and a slower one, since it would need a redeploy to change. + *
+ * + *

The toggle is on the read path, so it fails open

+ * + * Consulting the ramp here puts a blocking HouseTables call on every table load, which is a path it + * is not otherwise on — elsewhere toggles gate writes and table-property changes. A HouseTables + * blip must therefore not fail table reads, so a lookup failure is logged and treated as "not + * bridged". + * + *

That is safe for exactly the same reason old clients may ignore unknown keys: not bridging + * leaves the reader at today's behavior. The two are the same property of a capability, used twice. + * A capability where ignoring is unsafe — deletion vectors, where skipping means returning deleted + * rows — must NOT reuse this fail-open block; for those, a lookup failure has to fail the read, + * because serving data that is silently wrong is worse than serving an error. + * + *

The override-honoring form is the correct one here and should stay that way: read-bridge is a + * rollout, not an authorization gate. Features that decide whether a user may write a preserved + * property must keep using the server-only {@code isFeatureActivated}, because the table property + * this form honors is writable by the very user being gated. + * + *

Adding a capability

+ * + * Add a source interface, a {@code Config} method that owns its own source check, kill + * switch, ramp and keys, and one merge line in {@link #resolve(TableDto)}. Nothing in the existing + * capability changes. Deliberately not generalised into a capability registry yet: with a single + * implementation that interface would be a guess. + * + *

Ignoring is not always safe. The client ignores config keys it does not recognise, so a + * capability may only be bridged this way if ignoring it leaves the client at today's behavior. + * That holds for column defaults — an old client reads {@code NULL}, exactly as it does now. It + * would NOT hold for something like deletion vectors, where ignoring the key means returning + * deleted rows: a silent correctness violation rather than a missed improvement. A capability of + * that kind cannot rely on the ignore rule and must not be stamped for a client too old to honor + * it, which means gating on the client version advertised in the {@code User-Agent} header. * *

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. + * keep it in sync. */ +@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."; + /** + * Feature id for the column-default capability's ramp. Also names its self-service table property + * ({@code read-bridge.column-default.enabled}) and its config keys, below. + * + *

Per capability, NOT one id for all of read-bridge. Capabilities bridged through this + * namespace differ wildly in risk and readiness — deletion vectors must be rampable separately + * from column defaults, not dragged along by them. The id is also baked into a user-facing table + * property, so splitting it later means migrating properties customers have already set; it costs + * nothing to get right while nothing is ramped. The bare {@code read-bridge} id is left free for + * a future superset ramp. + */ + public static final String COLUMN_DEFAULT_FEATURE_ID = "read-bridge.column-default"; + + /** + * Config key prefix for a per-column read-time default; suffixed with the Iceberg field-id. + * Derived from the feature id so the ramp, the property and the wire keys cannot drift apart. + */ + 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; } - public Map resolve(String databaseId, String tableId, TableDto tableDto) { + /** + * Resolves the per-table client {@code config} for {@code tableDto}, empty when nothing is + * bridged. Purely a merge of independently-gated capabilities; see the class javadoc for why + * there is no shared gate here. + * + *

Takes the DTO alone: it already carries the database and table ids, and passing them + * separately invites the call sites to disagree about where they came from. + */ + public Map resolve(TableDto tableDto) { + Map config = new HashMap<>(); + config.putAll(columnDefaultConfig(tableDto)); + return config; + } + + /** The column-default capability: its own source, ramp and keys. */ + private Map columnDefaultConfig(TableDto tableDto) { + if (columnDefaultsSource == ColumnDefaultsSource.NONE) { + // Nothing can be bridged for this capability: skip the toggle lookup, which is a remote + // HouseTables call on the table-load path. + return Collections.emptyMap(); + } + if (!isColumnDefaultRamped(tableDto)) { + return Collections.emptyMap(); // not ramped for this table -> stamp nothing + } Map columnDefaults = columnDefaultsSource.defaults(tableDto); if (columnDefaults == null || columnDefaults.isEmpty()) { return Collections.emptyMap(); // nothing to bridge -> stamp nothing @@ -44,4 +168,23 @@ public Map resolve(String databaseId, String tableId, TableDto t (fieldId, value) -> config.put(COLUMN_DEFAULT_PREFIX + fieldId, value.toString())); return config; } + + /** + * Whether the column-default capability is ramped for this table, failing open on a toggle-lookup + * failure so a HouseTables blip degrades bridging rather than failing the read. Safe only because + * not bridging is today's behavior; see the class javadoc. + */ + 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..110a74759 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 @@ -4,6 +4,7 @@ import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verifyNoInteractions; import static org.mockito.Mockito.when; import com.fasterxml.jackson.databind.JsonNode; @@ -16,6 +17,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; @@ -26,21 +28,164 @@ 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; + /** A toggle that ramps everything, so a test isolates the encoder rather than the 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")); + } + + /** A table carrying an explicit self-service opt-in/opt-out 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(); + } + + /** Gate 1: no deployment-supplied source => inert, and crucially no toggle lookup at all. */ @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()); + // The toggle is a remote HouseTables call on the table-load path; it must not be made. + verifyNoInteractions(toggle); + } + + /** + * The ramp lookup is a blocking HouseTables call, and this is the table-load path — a path + * toggles are not otherwise on. A HouseTables outage must degrade bridging, not fail reads. Sound + * only because not bridging is exactly today's behavior; a capability where ignoring is unsafe + * (deletion vectors) would have to fail the read instead. + */ + @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()); + } + + /** Gate 3: a table the ramp has not activated is not bridged, and its source is never asked. */ + @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 (a deployment may parse a schema); gate first. + verifyNoInteractions(source); + } + + /** The self-service property opts a table in even when the server-managed ramp says no. */ + @Test + public void testTablePropertyOptsInOverServerToggle() { + TableFeatureToggle allOff = + new TableFeatureToggle() { + @Override + public boolean isFeatureActivated(String databaseId, String tableId, String featureId) { + return false; + } + }; + Map config = + new ReadBridgeConfigResolver(oneDefault(), allOff).resolve(tableWithOverride("true")); + + Assertions.assertEquals("\"US\"", config.get(PREFIX + "5")); + } + + /** ...and opts it out even when the server-managed ramp says yes. */ + @Test + public void testTablePropertyOptsOutOverServerToggle() { + Assertions.assertTrue(resolverFor(oneDefault()).resolve(tableWithOverride("false")).isEmpty()); + } + + /** + * The capability's feature id, its self-service property and its wire keys are one token. Pinned + * as literals because all three are external contracts: the id is stored in HouseTables toggle + * rules, the property is set on customer tables, and the prefix is mirrored by the client + * decoder. Deriving them from each other keeps them consistent; asserting the literals keeps a + * refactor from silently renaming all three at once. + */ + @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); + } + + /** + * Rollout is per capability, never for read-bridge as a whole: capabilities share only the + * transport. A table opted out of column defaults must not thereby be opted out of a capability + * added later, and vice versa. Pinned because the id is baked into a customer-set property, so + * splitting it after the fact means a migration. + * + *

The bare "read-bridge" id staying unclaimed is also the room a future superset ramp (e.g. + * "v3-read-bridge", activating every capability at once) needs in order to exist without + * colliding with a capability's own 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)); + Map config = resolverFor(source).resolve(mock(TableDto.class)); // value is the single-value JSON for the default ("US" -> "\"US\""). Assertions.assertEquals("\"US\"", config.get(PREFIX + "5")); } @@ -54,8 +199,7 @@ 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")); @@ -74,7 +218,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); @@ -94,8 +238,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"); From 2b30b0f88a8964f7754278cdb9750e89c7f65e08 Mon Sep 17 00:00:00 2001 From: Christian Bush Date: Tue, 11 Aug 2026 22:13:05 -0700 Subject: [PATCH 3/9] Fix read-bridge policy docs and pin empty-source / opt-in-skip-HTS Correct the self-service property name and drop the contradictory "cluster kill switch" claim. Add coverage for a real source that returns no defaults, and assert table-property opt-in never calls the server toggle. Testing Done: - :services:tables:test --tests '*ReadBridgeConfigResolverTest' --- .../readbridge/ReadBridgeConfigResolver.java | 14 ++++---- .../ReadBridgeConfigResolverTest.java | 35 ++++++++++++++----- 2 files changed, 34 insertions(+), 15 deletions(-) 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 627ce8032..e0c5563e5 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 @@ -23,18 +23,18 @@ * * This class owns the policy — the feature id, the ramp, and the wire keys — and a * deployment supplies only the data, via {@link ColumnDefaultsSource}. Keeping the ramp - * here means every deployment inherits it, the self-service property {@code read-bridge.enabled} is - * documented alongside the {@code openhouse.read-bridge.*} keys it controls, and a deployment's - * source is never asked to derive defaults for a table that is not bridged. + * here means every deployment inherits it, the self-service property {@code + * read-bridge.column-default.enabled} is documented alongside the {@code openhouse.read-bridge.*} + * keys it controls, and a deployment's source is never asked to derive defaults for a table that is + * not bridged. * *

What capabilities share, and what they don't

* * Capabilities bridged through this class share the infrastructure and nothing else: the * per-table {@code config} channel, the {@code openhouse.read-bridge.*} namespace, and the client's - * decode/apply path. Rollout is never shared. Each capability has its own source, feature id, - * self-service table property and cluster kill switch, so it can be ramped, paused or killed - * without touching any other. There is deliberately no single switch, toggle id or property meaning - * "all of read-bridge". + * decode/apply path. Rollout is never shared. Each capability has its own source, feature id, and + * self-service table property, so it can be ramped, paused or killed without touching any other. + * There is deliberately no single switch, toggle id or property meaning "all of read-bridge". * *

{@link #resolve(TableDto)} therefore contains no cross-capability gate at all — it only merges * what each capability decided for itself. A shared short-circuit there would couple rollouts that 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 110a74759..9e2140924 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,10 @@ 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; @@ -120,17 +123,17 @@ public boolean isFeatureActivated(String databaseId, String tableId, String feat /** The self-service property opts a table in even when the server-managed ramp says no. */ @Test public void testTablePropertyOptsInOverServerToggle() { - TableFeatureToggle allOff = - new TableFeatureToggle() { - @Override - public boolean isFeatureActivated(String databaseId, String tableId, String featureId) { - return false; - } - }; + // CALLS_REAL_METHODS so the override-honoring default reads the table property; stub the + // server-side form so an accidental HTS 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(), allOff).resolve(tableWithOverride("true")); + new ReadBridgeConfigResolver(oneDefault(), toggle).resolve(tableWithOverride("true")); Assertions.assertEquals("\"US\"", config.get(PREFIX + "5")); + // Explicit opt-in is decided from the table property alone; no HouseTables round-trip. + verify(toggle, never()).isFeatureActivated(anyString(), anyString(), anyString()); } /** ...and opts it out even when the server-managed ramp says yes. */ @@ -139,6 +142,22 @@ public void testTablePropertyOptsOutOverServerToggle() { Assertions.assertTrue(resolverFor(oneDefault()).resolve(tableWithOverride("false")).isEmpty()); } + /** + * Source present and table ramped, but the source has nothing to stamp — still empty config. Not + * the same as {@link ColumnDefaultsSource#NONE}: the toggle ran and the source was asked. + */ + @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()); + } + /** * The capability's feature id, its self-service property and its wire keys are one token. Pinned * as literals because all three are external contracts: the id is stored in HouseTables toggle From cd3cd5ec77a643f3608f23bccc6804d896fdc481 Mon Sep 17 00:00:00 2001 From: Christian Bush Date: Tue, 11 Aug 2026 22:20:14 -0700 Subject: [PATCH 4/9] Docs: describe exact-match HTS ramp, not glob */* BaseTableFeatureToggle looks up (databaseId, tableId, featureId) exactly; claiming a * / * fleet kill switch was inaccurate. Also note HTS is only hit when the self-service property is absent. --- .../readbridge/ReadBridgeConfigResolver.java | 21 +++++++++++-------- 1 file changed, 12 insertions(+), 9 deletions(-) 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 e0c5563e5..34729cb2d 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 @@ -59,19 +59,22 @@ * inert, and notably makes no toggle lookup, so open-source and dev deployments add nothing * to the table-load path. *

  • {@link TableFeatureToggle#isFeatureActivatedWithOverride} — the per-table ramp: an explicit - * {@code read-bridge.column-default.enabled} table property opts a table in or out, otherwise - * the server-managed toggle decides. Rules match database and table as globs, so a {@code *} - * / {@code *} rule ramps or un-ramps the fleet as data, taking effect immediately. That is - * the kill switch; there is deliberately no cluster property duplicating it, which would only - * add a second place to look and a slower one, since it would need a redeploy to change. + * {@code read-bridge.column-default.enabled} table property opts a table in or out without a + * HouseTables call; when absent, the server-managed toggle decides via an exact {@code + * (databaseId, tableId, featureId)} lookup ({@link + * com.linkedin.openhouse.tables.toggle.BaseTableFeatureToggle}). There is no glob / {@code *} + * matcher today — fleet ramp means writing {@code ACTIVE} rows (or setting the table + * property) per table. That HTS row (or the property) is the kill switch; there is + * deliberately no cluster property duplicating it, which would only add a second place to + * look and a slower one, since it would need a redeploy to change. * * *

    The toggle is on the read path, so it fails open

    * - * Consulting the ramp here puts a blocking HouseTables call on every table load, which is a path it - * is not otherwise on — elsewhere toggles gate writes and table-property changes. A HouseTables - * blip must therefore not fail table reads, so a lookup failure is logged and treated as "not - * bridged". + * Consulting the ramp here can put a blocking HouseTables call on table load when the self-service + * property is absent — a path toggles are not otherwise on (elsewhere they gate writes and + * table-property changes). A HouseTables blip must therefore not fail table reads, so a lookup + * failure is logged and treated as "not bridged". * *

    That is safe for exactly the same reason old clients may ignore unknown keys: not bridging * leaves the reader at today's behavior. The two are the same property of a capability, used twice. From e9bfef3a254d57d3d3828af5310dbb79f4a9d8eb Mon Sep 17 00:00:00 2001 From: Christian Bush Date: Tue, 11 Aug 2026 22:32:27 -0700 Subject: [PATCH 5/9] Trim read-bridge policy comments to short why-notes Replace essay javadoc with brief ownership, contract, ObjectProvider, and fail-open notes so the PR description carries the design narrative. --- .../openhouse/tables/api/ApiConfig.java | 19 +-- .../readbridge/ColumnDefaultsSource.java | 36 +---- .../readbridge/ReadBridgeConfigResolver.java | 138 ++---------------- 3 files changed, 19 insertions(+), 174 deletions(-) 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 510c5e44f..580686da0 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 @@ -9,7 +9,7 @@ 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 @@ -18,21 +18,8 @@ public TablesApiHandler tablesApiHandler() { } /** - * Server-side encoder that stamps the read-bridge {@code config}. - * - *

    {@link ColumnDefaultsSource} is the column-default capability's single extension point, and - * it is resolved here rather than declared as an overridable default bean. A deployment supplies - * one; with none present that capability is inert and never consults the feature toggle. Each - * capability is wired, and rolled out, on its own. - * - *

    Deliberately not a {@code @ConditionalOnMissingBean} default bean. Spring Boot documents - * that condition as safe only inside auto-configuration, and this is an ordinary - * {@code @Configuration}: a component-scanned override happens to work, because {@code - * ConfigurationClassPostProcessor} finishes scanning before it evaluates {@code @Bean} - * conditions, but a deployment declaring its source with {@code @Bean} in a configuration class - * parsed after this one would get a competing no-op bean and need {@code @Primary} to avoid a - * {@code NoUniqueBeanDefinitionException}. With {@link ObjectProvider} no default bean is ever - * registered, so exactly one bean of the type exists however it was declared. + * Prefer {@link ObjectProvider} over a {@code @ConditionalOnMissingBean} noop so a deployment + * {@code @Bean} source cannot collide with an OSS default. */ @Bean public ReadBridgeConfigResolver readBridgeConfigResolver( 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 d5452e153..5793d1a78 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 @@ -6,43 +6,17 @@ 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, and it supplies data only - * — not policy. Whether a table is bridged at all is decided by {@link ReadBridgeConfigResolver} - * from the open-source feature toggle, so an implementation neither consults nor knows about the - * ramp. Deriving the defaults — from whatever a deployment treats as the authority on a column's - * declared default — is the one deployment-specific step, and the only reason this interface - * exists. - * - *

    No open-source default bean exists: {@code ApiConfig} resolves the type through an {@code - * ObjectProvider} and falls back to {@link #NONE}, so the feature is wired but completely inert out - * of the box — including skipping the toggle lookup entirely. - * - *

    {@code JsonNode} rather than {@code String} is deliberate. It makes a stamped value - * well-formed by construction at the only place that produces one, which is what entitles - * the client decoder to treat a malformed entry as a bug and fail loud instead of degrading. - * - *

    Called only for tables the ramp has activated, so an implementation may do real work (parsing - * a schema, say) without paying it on every table load fleet-wide. 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. + * Deployment-supplied column defaults (data only). Keyed by Iceberg field-id; values are Iceberg + * single-value JSON. Policy/ramp lives in {@link ReadBridgeConfigResolver}. */ public interface ColumnDefaultsSource { - /** - * Supplies nothing. The value {@code ApiConfig} falls back to when a deployment supplies no - * source; {@link ReadBridgeConfigResolver} recognises it and short-circuits before the toggle. - */ + /** Sentinel when no deployment bean is registered; resolver short-circuits before HTS. */ 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 + * @return field-id → default JSON; empty/null means nothing to stamp. Throw if a declared default + * cannot bind (do not silently omit). */ 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 34729cb2d..e0042ac6a 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 @@ -9,122 +9,16 @@ import lombok.extern.slf4j.Slf4j; /** - * Open-source encoder for the {@code read-bridge} feature: for a table the ramp has activated, it - * asks the pluggable {@link ColumnDefaultsSource} for that 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. - * - *

    Who decides what

    - * - * This class owns the policy — the feature id, the ramp, and the wire keys — and a - * deployment supplies only the data, via {@link ColumnDefaultsSource}. Keeping the ramp - * here means every deployment inherits it, the self-service property {@code - * read-bridge.column-default.enabled} is documented alongside the {@code openhouse.read-bridge.*} - * keys it controls, and a deployment's source is never asked to derive defaults for a table that is - * not bridged. - * - *

    What capabilities share, and what they don't

    - * - * Capabilities bridged through this class share the infrastructure and nothing else: the - * per-table {@code config} channel, the {@code openhouse.read-bridge.*} namespace, and the client's - * decode/apply path. Rollout is never shared. Each capability has its own source, feature id, and - * self-service table property, so it can be ramped, paused or killed without touching any other. - * There is deliberately no single switch, toggle id or property meaning "all of read-bridge". - * - *

    {@link #resolve(TableDto)} therefore contains no cross-capability gate at all — it only merges - * what each capability decided for itself. A shared short-circuit there would couple rollouts that - * are meant to be independent: a deployment supplying a deletion-vector source but no - * column-default source must still get deletion vectors. - * - *

    Combining rollouts later

    - * - * Independence is the default, not the ceiling. A superset ramp — say {@code v3-read-bridge}, - * activating every capability at once for tables that want the whole V3 read surface — is a natural - * later addition, and nothing here blocks it: it would be one more feature id consulted alongside - * the capability's own. The mechanism-wide ids ({@code read-bridge}, {@code v3-read-bridge}) are - * left unused today so one of them can take that role without colliding with a capability. It will - * need an explicit precedence rule; the sane one is that the more specific wins, so a table setting - * {@code read-bridge.column-default.enabled=false} stays opted out of that capability even while - * opted into the superset. - * - *

    Gating, cheapest check first (per capability)

    - * - *
      - *
    1. No source supplied for the capability ({@link ColumnDefaultsSource#NONE}) — structurally - * inert, and notably makes no toggle lookup, so open-source and dev deployments add nothing - * to the table-load path. - *
    2. {@link TableFeatureToggle#isFeatureActivatedWithOverride} — the per-table ramp: an explicit - * {@code read-bridge.column-default.enabled} table property opts a table in or out without a - * HouseTables call; when absent, the server-managed toggle decides via an exact {@code - * (databaseId, tableId, featureId)} lookup ({@link - * com.linkedin.openhouse.tables.toggle.BaseTableFeatureToggle}). There is no glob / {@code *} - * matcher today — fleet ramp means writing {@code ACTIVE} rows (or setting the table - * property) per table. That HTS row (or the property) is the kill switch; there is - * deliberately no cluster property duplicating it, which would only add a second place to - * look and a slower one, since it would need a redeploy to change. - *
    - * - *

    The toggle is on the read path, so it fails open

    - * - * Consulting the ramp here can put a blocking HouseTables call on table load when the self-service - * property is absent — a path toggles are not otherwise on (elsewhere they gate writes and - * table-property changes). A HouseTables blip must therefore not fail table reads, so a lookup - * failure is logged and treated as "not bridged". - * - *

    That is safe for exactly the same reason old clients may ignore unknown keys: not bridging - * leaves the reader at today's behavior. The two are the same property of a capability, used twice. - * A capability where ignoring is unsafe — deletion vectors, where skipping means returning deleted - * rows — must NOT reuse this fail-open block; for those, a lookup failure has to fail the read, - * because serving data that is silently wrong is worse than serving an error. - * - *

    The override-honoring form is the correct one here and should stay that way: read-bridge is a - * rollout, not an authorization gate. Features that decide whether a user may write a preserved - * property must keep using the server-only {@code isFeatureActivated}, because the table property - * this form honors is writable by the very user being gated. - * - *

    Adding a capability

    - * - * Add a source interface, a {@code Config} method that owns its own source check, kill - * switch, ramp and keys, and one merge line in {@link #resolve(TableDto)}. Nothing in the existing - * capability changes. Deliberately not generalised into a capability registry yet: with a single - * implementation that interface would be a guess. - * - *

    Ignoring is not always safe. The client ignores config keys it does not recognise, so a - * capability may only be bridged this way if ignoring it leaves the client at today's behavior. - * That holds for column defaults — an old client reads {@code NULL}, exactly as it does now. It - * would NOT hold for something like deletion vectors, where ignoring the key means returning - * deleted rows: a silent correctness violation rather than a missed improvement. A capability of - * that kind cannot rely on the ignore rule and must not be stamped for a client too old to honor - * it, which means gating on the client version advertised in the {@code User-Agent} header. - * - *

    Mirror: {@link #COLUMN_DEFAULT_PREFIX} is the shared contract with the client decoder; - * keep it in sync. + * Stamps per-table {@code config} for read-bridge capabilities. Owns policy (feature id, ramp, + * keys); deployments supply data via {@link ColumnDefaultsSource}. */ @Slf4j public class ReadBridgeConfigResolver { - /** - * Feature id for the column-default capability's ramp. Also names its self-service table property - * ({@code read-bridge.column-default.enabled}) and its config keys, below. - * - *

    Per capability, NOT one id for all of read-bridge. Capabilities bridged through this - * namespace differ wildly in risk and readiness — deletion vectors must be rampable separately - * from column defaults, not dragged along by them. The id is also baked into a user-facing table - * property, so splitting it later means migrating properties customers have already set; it costs - * nothing to get right while nothing is ramped. The bare {@code read-bridge} id is left free for - * a future superset ramp. - */ + /** Capability id; also names {@code .enabled} and the config key prefix below. */ public static final String COLUMN_DEFAULT_FEATURE_ID = "read-bridge.column-default"; - /** - * Config key prefix for a per-column read-time default; suffixed with the Iceberg field-id. - * Derived from the feature id so the ramp, the property and the wire keys cannot drift apart. - */ + /** Client contract: {@code openhouse.read-bridge.column-default.}. */ public static final String COLUMN_DEFAULT_PREFIX = "openhouse." + COLUMN_DEFAULT_FEATURE_ID + "."; private final ColumnDefaultsSource columnDefaultsSource; @@ -137,45 +31,35 @@ public ReadBridgeConfigResolver( this.featureToggle = featureToggle; } - /** - * Resolves the per-table client {@code config} for {@code tableDto}, empty when nothing is - * bridged. Purely a merge of independently-gated capabilities; see the class javadoc for why - * there is no shared gate here. - * - *

    Takes the DTO alone: it already carries the database and table ids, and passing them - * separately invites the call sites to disagree about where they came from. - */ + /** Merges independently gated capabilities; empty when nothing is bridged. */ public Map resolve(TableDto tableDto) { Map config = new HashMap<>(); config.putAll(columnDefaultConfig(tableDto)); return config; } - /** The column-default capability: its own source, ramp and keys. */ private Map columnDefaultConfig(TableDto tableDto) { + // No deployment source → skip HTS entirely. if (columnDefaultsSource == ColumnDefaultsSource.NONE) { - // Nothing can be bridged for this capability: skip the toggle lookup, which is a remote - // HouseTables call on the table-load path. return Collections.emptyMap(); } if (!isColumnDefaultRamped(tableDto)) { - return Collections.emptyMap(); // not ramped for this table -> stamp nothing + 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; } /** - * Whether the column-default capability is ramped for this table, failing open on a toggle-lookup - * failure so a HouseTables blip degrades bridging rather than failing the read. Safe only because - * not bridging is today's behavior; see the class javadoc. + * Uses {@link TableFeatureToggle#isFeatureActivatedWithOverride} so {@code + * read-bridge.column-default.enabled} can opt in/out without HTS. Fail-open on lookup errors: not + * bridging equals today's NULL reads. */ private boolean isColumnDefaultRamped(TableDto tableDto) { try { From e0a6cf22b5e2cf4fd938fdfe6edf06786ce3351c Mon Sep 17 00:00:00 2001 From: Christian Bush Date: Tue, 11 Aug 2026 22:33:04 -0700 Subject: [PATCH 6/9] Shorten withConfig javadoc to a one-liner --- .../tables/api/handler/impl/OpenHouseTablesApiHandler.java | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) 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 5abf91557..12091bae1 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,11 +38,7 @@ 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. - */ + /** Request-time {@code config} stamp; mapper leaves it null. */ private GetTableResponseBody withConfig(GetTableResponseBody body, TableDto tableDto) { return body.toBuilder().config(readBridgeConfigResolver.resolve(tableDto)).build(); } From 27e689c24b7962f8df0e2439f10a9892304f85fc Mon Sep 17 00:00:00 2001 From: Christian Bush Date: Wed, 12 Aug 2026 10:02:30 -0700 Subject: [PATCH 7/9] Tighten read-bridge policy comments to why, not design essays. Co-authored-by: Cursor --- .../openhouse/tables/api/ApiConfig.java | 5 +- .../impl/OpenHouseTablesApiHandler.java | 2 +- .../readbridge/ColumnDefaultsSource.java | 9 ++- .../readbridge/ReadBridgeConfigResolver.java | 17 +++--- .../ReadBridgeConfigResolverTest.java | 57 +++++-------------- 5 files changed, 29 insertions(+), 61 deletions(-) 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 580686da0..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 @@ -17,10 +17,7 @@ public TablesApiHandler tablesApiHandler() { return new OpenHouseTablesApiHandler(); } - /** - * Prefer {@link ObjectProvider} over a {@code @ConditionalOnMissingBean} noop so a deployment - * {@code @Bean} source cannot collide with an OSS default. - */ + /** ObjectProvider so a deployment bean does not collide with an OSS default. */ @Bean public ReadBridgeConfigResolver readBridgeConfigResolver( ObjectProvider columnDefaultsSource, TableFeatureToggle 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 12091bae1..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,7 +38,7 @@ public class OpenHouseTablesApiHandler implements TablesApiHandler { @Autowired private ReadBridgeConfigResolver readBridgeConfigResolver; - /** Request-time {@code config} stamp; mapper leaves it null. */ + /** 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(); } 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 5793d1a78..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 @@ -6,17 +6,16 @@ import java.util.Map; /** - * Deployment-supplied column defaults (data only). Keyed by Iceberg field-id; values are Iceberg - * single-value JSON. Policy/ramp lives in {@link ReadBridgeConfigResolver}. + * Column defaults for one table, keyed by Iceberg field-id. Values are Iceberg single-value JSON. + * Ramp is {@link ReadBridgeConfigResolver}. */ public interface ColumnDefaultsSource { - /** Sentinel when no deployment bean is registered; resolver short-circuits before HTS. */ + /** Used when no deployment bean is registered. */ ColumnDefaultsSource NONE = tableDto -> Collections.emptyMap(); /** - * @return field-id → default JSON; empty/null means nothing to stamp. Throw if a declared default - * cannot bind (do not silently omit). + * 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 e0042ac6a..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 @@ -9,16 +9,16 @@ import lombok.extern.slf4j.Slf4j; /** - * Stamps per-table {@code config} for read-bridge capabilities. Owns policy (feature id, ramp, - * keys); deployments supply data via {@link ColumnDefaultsSource}. + * Builds the per-table {@code config} map the client reads. OpenHouse owns ramp and keys; {@link + * ColumnDefaultsSource} supplies the values. */ @Slf4j public class ReadBridgeConfigResolver { - /** Capability id; also names {@code .enabled} and the config key prefix below. */ + /** Also names {@code .enabled} and the config key prefix. */ public static final String COLUMN_DEFAULT_FEATURE_ID = "read-bridge.column-default"; - /** Client contract: {@code openhouse.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; @@ -31,7 +31,7 @@ public ReadBridgeConfigResolver( this.featureToggle = featureToggle; } - /** Merges independently gated capabilities; empty when nothing is bridged. */ + /** 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)); @@ -39,7 +39,7 @@ public Map resolve(TableDto tableDto) { } private Map columnDefaultConfig(TableDto tableDto) { - // No deployment source → skip HTS entirely. + // No source registered: skip the HouseTables lookup. if (columnDefaultsSource == ColumnDefaultsSource.NONE) { return Collections.emptyMap(); } @@ -57,9 +57,8 @@ private Map columnDefaultConfig(TableDto tableDto) { } /** - * Uses {@link TableFeatureToggle#isFeatureActivatedWithOverride} so {@code - * read-bridge.column-default.enabled} can opt in/out without HTS. Fail-open on lookup errors: not - * bridging equals today's NULL reads. + * 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 { 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 9e2140924..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 @@ -30,12 +30,11 @@ public class ReadBridgeConfigResolverTest { - /** Open-source default source: supplies nothing, so the feature is inert. */ private static final ColumnDefaultsSource NONE = ColumnDefaultsSource.NONE; private static final String PREFIX = ReadBridgeConfigResolver.COLUMN_DEFAULT_PREFIX; - /** A toggle that ramps everything, so a test isolates the encoder rather than the ramp. */ + /** Isolates encoding from ramp. */ private static final TableFeatureToggle ALL_ON = new TableFeatureToggle() { @Override @@ -52,7 +51,7 @@ private static ColumnDefaultsSource oneDefault() { return tableDto -> Collections.singletonMap(5, TextNode.valueOf("US")); } - /** A table carrying an explicit self-service opt-in/opt-out property. */ + /** Table with an explicit {@code .enabled} property. */ private static TableDto tableWithOverride(String value) { return TableDto.builder() .databaseId("db") @@ -65,7 +64,7 @@ private static TableDto tableWithOverride(String value) { .build(); } - /** Gate 1: no deployment-supplied source => inert, and crucially no toggle lookup at all. */ + /** No source → no HouseTables call. */ @Test public void testInertAndSkipsToggleWhenNoSourceSupplied() { TableFeatureToggle toggle = mock(TableFeatureToggle.class); @@ -73,16 +72,10 @@ public void testInertAndSkipsToggleWhenNoSourceSupplied() { new ReadBridgeConfigResolver(ColumnDefaultsSource.NONE, toggle); Assertions.assertTrue(resolver.resolve(mock(TableDto.class)).isEmpty()); - // The toggle is a remote HouseTables call on the table-load path; it must not be made. verifyNoInteractions(toggle); } - /** - * The ramp lookup is a blocking HouseTables call, and this is the table-load path — a path - * toggles are not otherwise on. A HouseTables outage must degrade bridging, not fail reads. Sound - * only because not bridging is exactly today's behavior; a capability where ignoring is unsafe - * (deletion vectors) would have to fail the read instead. - */ + /** HouseTables down → unbridged, not a failed read. */ @Test public void testToggleLookupFailureDegradesInsteadOfFailingTheRead() { TableFeatureToggle exploding = @@ -100,7 +93,7 @@ public boolean isFeatureActivated(String databaseId, String tableId, String feat Assertions.assertTrue(config.isEmpty()); } - /** Gate 3: a table the ramp has not activated is not bridged, and its source is never asked. */ + /** Unramped table is not asked for defaults. */ @Test public void testUnrampedTableIsNotBridgedAndSourceNotConsulted() { ColumnDefaultsSource source = mock(ColumnDefaultsSource.class); @@ -116,15 +109,14 @@ public boolean isFeatureActivated(String databaseId, String tableId, String feat new ReadBridgeConfigResolver(source, allOff) .resolve(TableDto.builder().databaseId("db").tableId("tbl").build()) .isEmpty()); - // Deriving defaults can be expensive (a deployment may parse a schema); gate first. + // Deriving defaults can be expensive; check the ramp first. verifyNoInteractions(source); } - /** The self-service property opts a table in even when the server-managed ramp says no. */ + /** {@code .enabled=true} wins over a server-side off. */ @Test public void testTablePropertyOptsInOverServerToggle() { - // CALLS_REAL_METHODS so the override-honoring default reads the table property; stub the - // server-side form so an accidental HTS call would return false. + // 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); @@ -132,20 +124,16 @@ public void testTablePropertyOptsInOverServerToggle() { new ReadBridgeConfigResolver(oneDefault(), toggle).resolve(tableWithOverride("true")); Assertions.assertEquals("\"US\"", config.get(PREFIX + "5")); - // Explicit opt-in is decided from the table property alone; no HouseTables round-trip. verify(toggle, never()).isFeatureActivated(anyString(), anyString(), anyString()); } - /** ...and opts it out even when the server-managed ramp says yes. */ + /** {@code .enabled=false} wins over a server-side on. */ @Test public void testTablePropertyOptsOutOverServerToggle() { Assertions.assertTrue(resolverFor(oneDefault()).resolve(tableWithOverride("false")).isEmpty()); } - /** - * Source present and table ramped, but the source has nothing to stamp — still empty config. Not - * the same as {@link ColumnDefaultsSource#NONE}: the toggle ran and the source was asked. - */ + /** Ramped table whose source has nothing to stamp. */ @Test public void testEmptyWhenSourceReturnsNoDefaults() { ColumnDefaultsSource emptySource = mock(ColumnDefaultsSource.class); @@ -158,13 +146,7 @@ public void testEmptyWhenSourceReturnsNoDefaults() { verify(emptySource).defaults(any()); } - /** - * The capability's feature id, its self-service property and its wire keys are one token. Pinned - * as literals because all three are external contracts: the id is stored in HouseTables toggle - * rules, the property is set on customer tables, and the prefix is mirrored by the client - * decoder. Deriving them from each other keeps them consistent; asserting the literals keeps a - * refactor from silently renaming all three at once. - */ + /** Id, property, and prefix are external contracts; keep them one token. */ @Test public void testFeatureIdPropertyAndKeysAreOneToken() { Assertions.assertEquals( @@ -177,16 +159,7 @@ public void testFeatureIdPropertyAndKeysAreOneToken() { "openhouse.read-bridge.column-default.", ReadBridgeConfigResolver.COLUMN_DEFAULT_PREFIX); } - /** - * Rollout is per capability, never for read-bridge as a whole: capabilities share only the - * transport. A table opted out of column defaults must not thereby be opted out of a capability - * added later, and vice versa. Pinned because the id is baked into a customer-set property, so - * splitting it after the fact means a migration. - * - *

    The bare "read-bridge" id staying unclaimed is also the room a future superset ramp (e.g. - * "v3-read-bridge", activating every capability at once) needs in order to exist without - * colliding with a capability's own id. - */ + /** Ramp is per capability, not a blanket read-bridge id. */ @Test public void testRolloutIdIsScopedToTheCapabilityNotTheMechanism() { Assertions.assertNotEquals("read-bridge", ReadBridgeConfigResolver.COLUMN_DEFAULT_FEATURE_ID); @@ -205,7 +178,7 @@ public void testEmptyWhenNoColumnDefaults() { public void testStampsColumnDefaultEntry() { ColumnDefaultsSource source = tableDto -> Collections.singletonMap(5, TextNode.valueOf("US")); Map config = resolverFor(source).resolve(mock(TableDto.class)); - // value is the single-value JSON for the default ("US" -> "\"US\""). + // "US" as Iceberg single-value JSON. Assertions.assertEquals("\"US\"", config.get(PREFIX + "5")); } @@ -224,7 +197,7 @@ public void testStampsAllColumnDefaultsAsSeparateEntries() { 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); @@ -246,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); From 2f134b5e869b92579331934da58213b20c785fbb Mon Sep 17 00:00:00 2001 From: Christian Bush Date: Wed, 12 Aug 2026 15:56:31 -0700 Subject: [PATCH 8/9] Restore on-disk column defaults before OpenHouse commits. Read-bridge overlays must not persist: a later write would stamp initial-default onto disk and survive ramp-off. Restore default slots for field-ids that existed at load; keep writer defaults on new ids so V2 schema evolution still works. --- .../OpenHouseTableOperationsTest.java | 68 ++++++ .../openhouse/javaclient/ReadBridgeTest.java | 194 ++++++++++++++++- .../javaclient/OpenHouseTableOperations.java | 17 +- .../openhouse/javaclient/ReadBridge.java | 197 ++++++++++++++++++ 4 files changed, 474 insertions(+), 2 deletions(-) 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 a2aeda4d4..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; @@ -617,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 a5d3345a1..085d2ea2b 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,9 +9,15 @@ 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; -/** Decoder for {@link ReadBridge#from}. */ +/** Decoder and sanitize path for {@link ReadBridge}. */ class ReadBridgeTest { private static final String PREFIX = ReadBridge.COLUMN_DEFAULT_PREFIX; @@ -58,4 +65,189 @@ void ignoresUnknownKeysWithoutFailing() { assertEquals(1, ReadBridge.from(config).columnDefaults().size()); assertEquals("US", ReadBridge.from(config).columnDefaults().get(5).asText()); } + + @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 83371c149..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 @@ -68,6 +68,12 @@ public class OpenHouseTableOperations extends BaseMetastoreTableOperations { */ private final AtomicReference> config = new AtomicReference<>(); + /** + * 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(); @@ -136,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); } @@ -208,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()); @@ -252,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 7a9e8cd07..a75a35386 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,10 +3,21 @@ 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.TableMetadata; +import org.apache.iceberg.TableMetadataParser; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.types.Types.NestedField; /** * Overlays server-stamped read-time behavior from table {@code config} onto loaded Iceberg @@ -15,6 +26,9 @@ *

    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. + * + *

    {@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 { @@ -26,6 +40,11 @@ final class ReadBridge { private static final ObjectMapper MAPPER = new ObjectMapper(); + 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"; + private final Map columnDefaults; private ReadBridge(Map columnDefaults) { @@ -51,6 +70,32 @@ TableMetadata apply(TableMetadata raw) { return raw; } + /** + * 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. + */ + 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; } @@ -81,4 +126,156 @@ private static Map columnDefaults(Map config) } return byFieldId; } + + 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); + } + } } From b85fd39f60696c829cfc6094c0256c233159b46c Mon Sep 17 00:00:00 2001 From: Christian Bush Date: Tue, 11 Aug 2026 19:07:57 -0700 Subject: [PATCH 9/9] Implement ReadBridge.apply without an Iceberg fork Overlay server-stamped column defaults onto every schema-id at metadata load using NestedField.initialDefault and a public TableMetadataParser JSON rebuild, keeping decode on the from()/apply() split from #668. --- .../openhouse/javaclient/ReadBridgeTest.java | 134 ++++++++++++++++-- .../openhouse/javaclient/ReadBridge.java | 126 ++++++++++++++-- 2 files changed, 243 insertions(+), 17 deletions(-) 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 085d2ea2b..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 @@ -17,27 +17,28 @@ import org.apache.iceberg.types.Types.NestedField; import org.junit.jupiter.api.Test; -/** Decoder and sanitize path for {@link ReadBridge}. */ +/** Decoder, apply, and sanitize path for {@link ReadBridge}. */ class ReadBridgeTest { private static final String PREFIX = ReadBridge.COLUMN_DEFAULT_PREFIX; @Test void decodesColumnDefaultsByFieldId() { - // Avoid naming JsonNode: it is relocated in the shaded client, and this module has no `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 @@ -62,8 +63,125 @@ void ignoresUnknownKeysWithoutFailing() { 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 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 a75a35386..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 @@ -13,8 +13,10 @@ 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; @@ -27,6 +29,10 @@ * #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. * + *

    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 #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. */ @@ -45,9 +51,10 @@ final class ReadBridge { private static final String SCHEMAS = "schemas"; private static final String SCHEMA_ID = "schema-id"; - private final Map columnDefaults; + /** 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; } @@ -57,7 +64,7 @@ private ReadBridge(Map columnDefaults) { * @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); } @@ -66,8 +73,19 @@ TableMetadata apply(TableMetadata raw) { if (columnDefaults.isEmpty()) { return raw; } - // TODO(read-bridge): overlay columnDefaults onto schemas. - return raw; + + 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); } /** @@ -96,22 +114,24 @@ static TableMetadata sanitize(TableMetadata raw, TableMetadata metadata) { return replaceSchemas(metadata, restoredById); } - Map columnDefaults() { + Map columnDefaults() { return columnDefaults; } - private static Map columnDefaults(Map config) { + 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) { // Known keys are stamped as int field-id + JSON; anything else is a bug. throw new IllegalStateException( @@ -127,6 +147,94 @@ 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()) {