diff --git a/apps/spark/src/test/java/com/linkedin/openhouse/catalog/e2e/RTASJavaTest.java b/apps/spark/src/test/java/com/linkedin/openhouse/catalog/e2e/RTASJavaTest.java index 23cd7ae8d..35040547c 100644 --- a/apps/spark/src/test/java/com/linkedin/openhouse/catalog/e2e/RTASJavaTest.java +++ b/apps/spark/src/test/java/com/linkedin/openhouse/catalog/e2e/RTASJavaTest.java @@ -65,6 +65,9 @@ void testReplaceTransaction() throws Exception { Table table = catalog.buildTable(TABLE_IDENT, SCHEMA).withPartitionSpec(SPEC).create(); table.newAppend().appendFile(FILE_A).commit(); + // RTAS is disabled by default; opt the table in before replacing it. + table.updateProperties().set("replace.enabled", "true").commit(); + String originalLocation = table.location(); String originalMetadataLocation = getMetadataLocation(table); long originalSnapshotId = table.currentSnapshot().snapshotId(); @@ -120,6 +123,9 @@ void testCreateOrReplaceTransaction() throws Exception { // verify that the table was created with one snapshot assertEquals(1, Iterables.size(table.snapshots()), "Should have one snapshot after create"); + // RTAS is disabled by default; opt the table in before replacing it. + table.updateProperties().set("replace.enabled", "true").commit(); + // create or replace on existing table should replace it Transaction replaceTxn = catalog diff --git a/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java b/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java index 2e44c9263..a32274662 100644 --- a/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java +++ b/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java @@ -752,6 +752,12 @@ public void testSnapshotsExpirationAfterReplaceTable() throws Exception { prepareTable(ops, tableName); populateTable(ops, tableName, numInserts); + // RTAS is disabled by default; opt the table in before replacing it. + ops.spark() + .sql( + String.format( + "ALTER TABLE %s SET TBLPROPERTIES ('replace.enabled'='true')", tableName)); + // replace the table using RTAS ops.spark() .sql( diff --git a/iceberg/openhouse/internalcatalog/src/main/java/com/linkedin/openhouse/internal/catalog/CatalogConstants.java b/iceberg/openhouse/internalcatalog/src/main/java/com/linkedin/openhouse/internal/catalog/CatalogConstants.java index 94200ee3a..23bd10183 100644 --- a/iceberg/openhouse/internalcatalog/src/main/java/com/linkedin/openhouse/internal/catalog/CatalogConstants.java +++ b/iceberg/openhouse/internalcatalog/src/main/java/com/linkedin/openhouse/internal/catalog/CatalogConstants.java @@ -30,6 +30,10 @@ public final class CatalogConstants { public static final String EVOLVED_SCHEMA_KEY = "evolved.table.schema"; + public static final String RTAS_ENABLED_TABLE_PROP = "replace.enabled"; + + public static final String WAP_ENABLED_TABLE_PROP = "write.wap.enabled"; + static final String FEATURE_TOGGLE_STOP_CREATE = "stop_create"; static final String CLIENT_TABLE_SCHEMA = "client.table.schema"; 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 b564a432e..499a297d3 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 @@ -4,12 +4,15 @@ import com.linkedin.openhouse.gen.tables.client.api.SnapshotApi; import com.linkedin.openhouse.gen.tables.client.api.TableApi; +import com.linkedin.openhouse.gen.tables.client.invoker.ApiClient; import com.linkedin.openhouse.gen.tables.client.model.CreateUpdateTableRequestBody; +import com.linkedin.openhouse.gen.tables.client.model.GetTableResponseBody; import com.linkedin.openhouse.gen.tables.client.model.History; import com.linkedin.openhouse.gen.tables.client.model.Policies; import com.linkedin.openhouse.gen.tables.client.model.PolicyTag; import com.linkedin.openhouse.gen.tables.client.model.Retention; import com.linkedin.openhouse.javaclient.exception.WebClientWithMessageException; +import com.linkedin.openhouse.relocated.com.fasterxml.jackson.databind.ObjectMapper; import com.linkedin.openhouse.relocated.org.springframework.http.HttpStatus; import com.linkedin.openhouse.relocated.org.springframework.web.reactive.function.client.WebClientRequestException; import com.linkedin.openhouse.relocated.org.springframework.web.reactive.function.client.WebClientResponseException; @@ -479,4 +482,121 @@ public void testPoliciesHistoryExistsUpdate() { Assertions.assertEquals(2, updatedPolicies.getHistory().getVersions()); Assertions.assertEquals(true, updatedPolicies.getSharingEnabled()); } + + private OpenHouseTableOperations refreshableOps(TableApi tableApi) { + return OpenHouseTableOperations.builder() + .tableIdentifier(TableIdentifier.of("db", "tbl")) + .fileIO(mock(FileIO.class)) + .tableApi(tableApi) + .snapshotApi(mock(SnapshotApi.class)) + .cluster("cluster") + .build(); + } + + /** Before any refresh, there is no server-stamped config, so the safe default is null. */ + @Test + public void testCurrentConfigNullBeforeRefresh() { + Assertions.assertNull(refreshableOps(mock(TableApi.class)).currentConfig()); + } + + /** doRefresh stashes the server-stamped config so subclasses can read it back. */ + @Test + public void testDoRefreshCapturesConfig() { + TableApi mockTableApi = mock(TableApi.class); + Map stamped = + Collections.singletonMap("openhouse.read-bridge", "{\"read\":\"ON\"}"); + GetTableResponseBody body = mock(GetTableResponseBody.class); + when(body.getTableLocation()).thenReturn(null); + when(body.getConfig()).thenReturn(stamped); + when(mockTableApi.getTableV1(anyString(), anyString())).thenReturn(Mono.just(body)); + + OpenHouseTableOperations ops = refreshableOps(mockTableApi); + ops.doRefresh(); + + Assertions.assertSame(stamped, ops.currentConfig()); + } + + /** Absent config on the response => null, the consumer's safe default. */ + @Test + public void testDoRefreshNullConfigWhenAbsent() { + TableApi mockTableApi = mock(TableApi.class); + GetTableResponseBody body = mock(GetTableResponseBody.class); + when(body.getTableLocation()).thenReturn(null); + when(body.getConfig()).thenReturn(null); + when(mockTableApi.getTableV1(anyString(), anyString())).thenReturn(Mono.just(body)); + + OpenHouseTableOperations ops = refreshableOps(mockTableApi); + ops.doRefresh(); + + 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. + */ + @Test + public void testDoRefreshClearsStaleConfig() { + TableApi mockTableApi = mock(TableApi.class); + Map stamped = + Collections.singletonMap("openhouse.read-bridge", "{\"read\":\"ON\"}"); + + GetTableResponseBody withConfig = mock(GetTableResponseBody.class); + when(withConfig.getTableLocation()).thenReturn(null); + when(withConfig.getConfig()).thenReturn(stamped); + + GetTableResponseBody withoutConfig = mock(GetTableResponseBody.class); + when(withoutConfig.getTableLocation()).thenReturn(null); + when(withoutConfig.getConfig()).thenReturn(null); + + // First refresh stamps config, second refresh stops stamping it. + when(mockTableApi.getTableV1(anyString(), anyString())) + .thenReturn(Mono.just(withConfig)) + .thenReturn(Mono.just(withoutConfig)); + + OpenHouseTableOperations ops = refreshableOps(mockTableApi); + + ops.doRefresh(); + Assertions.assertSame(stamped, ops.currentConfig()); + + ops.doRefresh(); + Assertions.assertNull(ops.currentConfig()); + } + + /** + * 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. + */ + @Test + public void testConfigDeserializeFromResponse() throws Exception { + ObjectMapper mapper = ApiClient.createDefaultObjectMapper(null); + String json = + "{\"tableId\":\"tbl\",\"databaseId\":\"db\",\"config\":{" + + "\"openhouse.read-bridge\":\"{\\\"read\\\":\\\"ON\\\"}\"}}"; + + 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. + 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. + */ + @Test + public void testConfigToleratesUnknownFields() throws Exception { + ObjectMapper mapper = ApiClient.createDefaultObjectMapper(null); + String json = + "{\"tableId\":\"tbl\",\"databaseId\":\"db\",\"someFutureField\":\"x\",\"config\":{" + + "\"openhouse.unknown-feature\":\"whatever\"}}"; + + GetTableResponseBody body = mapper.readValue(json, GetTableResponseBody.class); + Map config = body.getConfig(); + Assertions.assertNotNull(config); + Assertions.assertEquals("whatever", config.get("openhouse.unknown-feature")); + } } 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 new file mode 100644 index 000000000..cd019f820 --- /dev/null +++ b/integrations/java/iceberg-1.2/openhouse-java-itest/src/test/java/com/linkedin/openhouse/javaclient/ReadBridgeTest.java @@ -0,0 +1,46 @@ +package com.linkedin.openhouse.javaclient; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import org.junit.jupiter.api.Test; + +/** + * Unit tests for the client-side read-bridge config decoder ({@link ReadBridge#columnDefaults}), + * exercised in isolation. Mirrors the server-side encoder {@code ReadBridgeConfigResolver}. + */ +class ReadBridgeTest { + + private static final String PREFIX = ReadBridge.COLUMN_DEFAULT_PREFIX; + + @Test + void decodesColumnDefaultsByFieldId() { + // Inline calls avoid naming Jackson's JsonNode, which is relocated in the shaded client uber + // (and this module compiles at a source level without `var`). + Map config = new HashMap<>(); + config.put(PREFIX + "5", "\"US\""); + config.put(PREFIX + "7", "0"); + assertEquals(2, ReadBridge.columnDefaults(config).size()); + assertEquals("US", ReadBridge.columnDefaults(config).get(5).asText()); + assertEquals(0, ReadBridge.columnDefaults(config).get(7).asInt()); + } + + @Test + void emptyWhenConfigNullOrNoReadBridgeKeys() { + assertTrue(ReadBridge.columnDefaults(null).isEmpty()); + assertTrue(ReadBridge.columnDefaults(Collections.singletonMap("other.key", "x")).isEmpty()); + } + + @Test + void skipsEntriesWithBadFieldIdOrUnparseableValue() { + Map config = new HashMap<>(); + config.put(PREFIX + "5", "\"US\""); + config.put(PREFIX + "notAnInt", "\"x\""); // non-integer field-id -> skipped + config.put(PREFIX + "7", "{bad json"); // unparseable value -> skipped + assertEquals(1, ReadBridge.columnDefaults(config).size()); + assertEquals("US", ReadBridge.columnDefaults(config).get(5).asText()); + } +} 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 0e62ddbbc..a897874f1 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 @@ -18,6 +18,7 @@ import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; import lombok.AccessLevel; import lombok.AllArgsConstructor; @@ -61,6 +62,21 @@ public class OpenHouseTableOperations extends BaseMetastoreTableOperations { @Getter(AccessLevel.PROTECTED) 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. + */ + 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. + */ + protected Map currentConfig() { + return config.get(); + } + @Override protected String tableName() { return tableIdentifier.toString(); @@ -81,10 +97,9 @@ public FileIO io() { @Override public void doRefresh() { log.info("Calling doRefresh for table: {}", tableName()); - Optional tableLocation = + Optional tableResponse = tableApi .getTableV1(tableIdentifier.namespace().toString(), tableIdentifier.name()) - .mapNotNull(GetTableResponseBody::getTableLocation) /* on 404 from table service, resume the stream as empty response. for any other error, surface it! @@ -98,22 +113,28 @@ 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. + 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()); } - // Read through loadMetadata() so subclasses can transform metadata as it loads. The 4-arg call - // with (null, 20) is the stock 1-arg behavior, with the parse step made overridable. + // Route the parse through loadMetadata() so subclasses can transform metadata as it loads; + // (null, 20) preserves the stock refresh behavior. super.refreshFromMetadataLocation(tableLocation.orElse(null), null, 20, this::loadMetadata); log.debug("Calling doRefresh succeeded"); } /** - * Loads the table metadata at the given location. Defaults to the stock parser; subclasses may - * override to transform the metadata (e.g. attach column defaults) as it loads. + * Loads the table metadata from storage, then runs the read-time {@link ReadBridge} over it using + * the per-table behavior the server stamped onto {@link #currentConfig()}. Fail-closed: + * absent/off/unparseable config leaves the raw metadata untouched. */ protected TableMetadata loadMetadata(String metadataLocation) { - return TableMetadataParser.read(io(), metadataLocation); + TableMetadata raw = TableMetadataParser.read(io(), metadataLocation); + return ReadBridge.apply(raw, currentConfig()); } @Override 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 new file mode 100644 index 000000000..c95fbc109 --- /dev/null +++ b/integrations/java/iceberg-1.2/openhouse-java-runtime/src/main/java/com/linkedin/openhouse/javaclient/ReadBridge.java @@ -0,0 +1,78 @@ +package com.linkedin.openhouse.javaclient; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import lombok.extern.slf4j.Slf4j; +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. + * + *

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

Kept out of {@link OpenHouseTableOperations} so the read path stays slim. Pure and + * fail-closed: an entry with a non-integer field-id or an unparseable value is skipped, and nothing + * to apply returns {@code raw} unchanged. + */ +@Slf4j +final class ReadBridge { + + /** Mirror of {@code ReadBridgeConfigResolver.COLUMN_DEFAULT_PREFIX}. */ + static final String COLUMN_DEFAULT_PREFIX = "openhouse.read-bridge.column-default."; + + private static final ObjectMapper MAPPER = new ObjectMapper(); + + private ReadBridge() {} + + /** + * Applies the bridged read-time behavior the server stamped into {@code config} onto {@code raw}, + * returning the transformed metadata (or {@code raw} when there is nothing to bridge). + */ + static TableMetadata apply(TableMetadata raw, Map config) { + Map columnDefaults = columnDefaults(config); + 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. + return raw; + } + + /** + * Decodes {@code field-id -> initial-default} from the {@code + * openhouse.read-bridge.column-default.*} config entries; empty when there are none. Skips any + * entry with a non-integer field-id or an unparseable value (fail-closed). Package-visible for + * unit testing in isolation. + */ + static Map columnDefaults(Map config) { + if (config == null) { + return Collections.emptyMap(); + } + 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())); + } catch (RuntimeException | JsonProcessingException e) { + log.warn( + "Skipping unparseable read-bridge config entry {}={}", + entry.getKey(), + entry.getValue(), + e); + } + } + return byFieldId; + } +} diff --git a/integrations/spark/spark-3.1/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/catalogtest/RTASTest.java b/integrations/spark/spark-3.1/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/catalogtest/RTASTest.java index e831ca706..21f05b45d 100644 --- a/integrations/spark/spark-3.1/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/catalogtest/RTASTest.java +++ b/integrations/spark/spark-3.1/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/catalogtest/RTASTest.java @@ -10,6 +10,7 @@ import org.apache.iceberg.Table; import org.apache.iceberg.catalog.Catalog; import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.exceptions.BadRequestException; import org.apache.iceberg.relocated.com.google.common.collect.Iterables; import org.apache.iceberg.types.Types; import org.apache.spark.sql.Row; @@ -60,6 +61,10 @@ public void testRTAS() throws Exception { spark.sql(String.format("ALTER TABLE %s SET POLICY (HISTORY MAX_AGE=24H)", tableName)); + // RTAS is disabled by default; opt the table in before replacing it. + spark.sql( + String.format("ALTER TABLE %s SET TBLPROPERTIES ('replace.enabled'='true')", tableName)); + String expectedTableLocation = catalog.loadTable(tableIdent).location(); // replace table @@ -111,6 +116,29 @@ public void testRTAS() throws Exception { } } + @Test + public void testRTASFailsWhenReplaceDisabled() throws Exception { + try (SparkSession spark = getSparkSession()) { + // create the table without opting into RTAS; replace is disabled by default + spark.sql( + String.format( + "CREATE TABLE %s USING iceberg AS SELECT * FROM %s", tableName, sourceName)); + + // REPLACE TABLE should be rejected because 'replace.enabled' is not set on the table + BadRequestException exception = + assertThrows( + BadRequestException.class, + () -> + spark.sql( + String.format( + "REPLACE TABLE %s USING iceberg AS SELECT * FROM %s", + tableName, sourceName))); + assertTrue( + exception.getMessage().contains("REPLACE TABLE AS SELECT is not enabled"), + "Expected an RTAS-disabled error but got: " + exception.getMessage()); + } + } + @Test public void testCreateRTAS() throws Exception { try (SparkSession spark = getSparkSession()) { @@ -125,6 +153,10 @@ public void testCreateRTAS() throws Exception { String expectedTableLocation = catalog.loadTable(tableIdent).location(); + // RTAS is disabled by default; opt the table in before replacing it. + spark.sql( + String.format("ALTER TABLE %s SET TBLPROPERTIES ('replace.enabled'='true')", tableName)); + // create or replace table should replace the table spark.sql( String.format( @@ -169,6 +201,10 @@ public void testDataFrameV2Replace() throws Exception { spark.table(sourceName).writeTo(tableName).using("iceberg").create(); + // RTAS is disabled by default; opt the table in before replacing it. + spark.sql( + String.format("ALTER TABLE %s SET TBLPROPERTIES ('replace.enabled'='true')", tableName)); + String expectedTableLocation = catalog.loadTable(tableIdent).location(); spark @@ -226,6 +262,10 @@ public void testDataFrameV2CreateOrReplace() throws Exception { .using("iceberg") .createOrReplace(); + // RTAS is disabled by default; opt the table in before replacing it. + spark.sql( + String.format("ALTER TABLE %s SET TBLPROPERTIES ('replace.enabled'='true')", tableName)); + String expectedTableLocation = catalog.loadTable(tableIdent).location(); spark diff --git a/services/common/src/main/java/com/linkedin/openhouse/common/exception/UnsupportedClientOperationException.java b/services/common/src/main/java/com/linkedin/openhouse/common/exception/UnsupportedClientOperationException.java index b76fbc8da..9df8e42fc 100644 --- a/services/common/src/main/java/com/linkedin/openhouse/common/exception/UnsupportedClientOperationException.java +++ b/services/common/src/main/java/com/linkedin/openhouse/common/exception/UnsupportedClientOperationException.java @@ -20,6 +20,7 @@ public enum Operation { GRANT_ON_UNSHARED_TABLES, ALTER_TABLE_TYPE, GRANT_ON_LOCKED_TABLES, - LOCKED_TABLE_OPERATION + LOCKED_TABLE_OPERATION, + RTAS_DISABLED, } } 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 a3b873e50..00408f757 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 @@ -2,6 +2,10 @@ import com.linkedin.openhouse.tables.api.handler.TablesApiHandler; 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 org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -12,4 +16,24 @@ public class ApiConfig { public TablesApiHandler tablesApiHandler() { return new OpenHouseTablesApiHandler(); } + + /** + * Open-source default {@link ColumnDefaultsSource}: supplies none, so read-bridge stays inert. + */ + @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); + } } 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 f4b7b1a67..8de0d0a49 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 @@ -13,6 +13,7 @@ import com.linkedin.openhouse.tables.api.validator.TablesApiValidator; import com.linkedin.openhouse.tables.dto.mapper.TablesMapper; import com.linkedin.openhouse.tables.model.TableDto; +import com.linkedin.openhouse.tables.readbridge.ReadBridgeConfigResolver; import com.linkedin.openhouse.tables.services.TablesService; import java.util.List; import java.util.stream.Collectors; @@ -35,15 +36,30 @@ public class OpenHouseTablesApiHandler implements TablesApiHandler { @Autowired private ClusterProperties clusterProperties; + @Autowired private ReadBridgeConfigResolver readBridgeConfigResolver; + + /** + * Stamp the server-resolved, per-table client {@code config} (Iceberg REST {@code + * LoadTableResponse.config} convention) onto a freshly mapped response body. The mapper leaves + * {@code config} null; it is a request-time decision resolved here. + */ + private GetTableResponseBody withConfig( + GetTableResponseBody body, String databaseId, String tableId, TableDto tableDto) { + return body.toBuilder() + .config(readBridgeConfigResolver.resolve(databaseId, tableId, tableDto)) + .build(); + } + @Override public ApiResponse getTable( String databaseId, String tableId, String actingPrincipal) { tablesApiValidator.validateGetTable(databaseId, tableId); + TableDto tableDto = tableService.getTable(databaseId, tableId, actingPrincipal); return ApiResponse.builder() .httpStatus(HttpStatus.OK) .responseBody( - tablesMapper.toGetTableResponseBody( - tableService.getTable(databaseId, tableId, actingPrincipal))) + withConfig( + tablesMapper.toGetTableResponseBody(tableDto), databaseId, tableId, tableDto)) .build(); } @@ -92,9 +108,15 @@ public ApiResponse createTable( clusterProperties.getClusterName(), databaseId, createUpdateTableRequestBody); Pair putResult = tableService.putTable(createUpdateTableRequestBody, tableCreator, true); + TableDto tableDto = putResult.getFirst(); return ApiResponse.builder() .httpStatus(HttpStatus.CREATED) - .responseBody(tablesMapper.toGetTableResponseBody(putResult.getFirst())) + .responseBody( + withConfig( + tablesMapper.toGetTableResponseBody(tableDto), + databaseId, + tableDto.getTableId(), + tableDto)) .build(); } @@ -109,9 +131,12 @@ public ApiResponse updateTable( Pair putResult = tableService.putTable(createUpdateTableRequestBody, tableCreatorUpdator, false); HttpStatus status = putResult.getSecond() ? HttpStatus.CREATED : HttpStatus.OK; + TableDto tableDto = putResult.getFirst(); return ApiResponse.builder() .httpStatus(status) - .responseBody(tablesMapper.toGetTableResponseBody(putResult.getFirst())) + .responseBody( + withConfig( + tablesMapper.toGetTableResponseBody(tableDto), databaseId, tableId, tableDto)) .build(); } diff --git a/services/tables/src/main/java/com/linkedin/openhouse/tables/api/spec/v0/response/GetTableResponseBody.java b/services/tables/src/main/java/com/linkedin/openhouse/tables/api/spec/v0/response/GetTableResponseBody.java index c410630e9..273b25348 100644 --- a/services/tables/src/main/java/com/linkedin/openhouse/tables/api/spec/v0/response/GetTableResponseBody.java +++ b/services/tables/src/main/java/com/linkedin/openhouse/tables/api/spec/v0/response/GetTableResponseBody.java @@ -106,6 +106,17 @@ public class GetTableResponseBody { @JsonProperty(access = JsonProperty.Access.READ_ONLY) private String sortOrder; + @Schema( + nullable = true, + description = + "Server-stamped, per-table client configuration overrides, following the Iceberg REST " + + "`LoadTableResponse.config` convention: a string map the server controls at runtime " + + "without a client re-roll. READ_ONLY and advisory — absent/empty means today's " + + "behavior; clients ignore keys they do not understand. Keys are vendor-namespaced " + + "(e.g. `openhouse.read-bridge`). Distinct from the table-governance `policies` object.") + @JsonProperty(access = JsonProperty.Access.READ_ONLY) + private Map config; + public String toJson() { return new Gson().toJson(this); } diff --git a/services/tables/src/main/java/com/linkedin/openhouse/tables/dto/mapper/TablesMapper.java b/services/tables/src/main/java/com/linkedin/openhouse/tables/dto/mapper/TablesMapper.java index 08111cd82..159f6e93b 100644 --- a/services/tables/src/main/java/com/linkedin/openhouse/tables/dto/mapper/TablesMapper.java +++ b/services/tables/src/main/java/com/linkedin/openhouse/tables/dto/mapper/TablesMapper.java @@ -133,6 +133,9 @@ public interface TablesMapper { * com.linkedin.openhouse.tables.api.spec.v0.response.GetTableResponseBody} to be forwarded to * the client. */ + // config is not sourced from TableDto; the API handler stamps the per-table client config onto + // the response separately, so the mapper leaves it null. + @Mapping(target = "config", ignore = true) GetTableResponseBody toGetTableResponseBody(TableDto tableDto); @Mappings({ 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 new file mode 100644 index 000000000..47ab3a9e8 --- /dev/null +++ b/services/tables/src/main/java/com/linkedin/openhouse/tables/readbridge/ColumnDefaultsSource.java @@ -0,0 +1,25 @@ +package com.linkedin.openhouse.tables.readbridge; + +import com.fasterxml.jackson.databind.JsonNode; +import com.linkedin.openhouse.tables.model.TableDto; +import java.util.Map; + +/** + * Pluggable input to the open-source {@code read-bridge} feature: the per-column initial-defaults + * to overlay at read time, keyed by Iceberg field-id and valued as Iceberg single-value JSON. + * + *

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

Called on every table-load/commit response, so implementations must be cheap and must never + * throw — an empty map means "no bridge for this table". + */ +public interface ColumnDefaultsSource { + /** + * @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 + */ + 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 new file mode 100644 index 000000000..405bb2d3b --- /dev/null +++ b/services/tables/src/main/java/com/linkedin/openhouse/tables/readbridge/ReadBridgeConfigResolver.java @@ -0,0 +1,47 @@ +package com.linkedin.openhouse.tables.readbridge; + +import com.fasterxml.jackson.databind.JsonNode; +import com.linkedin.openhouse.tables.model.TableDto; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +/** + * Open-source encoder for the {@code read-bridge} feature: it asks the pluggable {@link + * ColumnDefaultsSource} for a table's column initial-defaults and stamps each as a namespaced entry + * in the per-table {@code config} — {@code openhouse.read-bridge.column-default. = + * }. The client decoder ({@code ReadBridge} in {@code openhouse-java-runtime}) + * reads these entries and overlays the defaults at metadata-load time. + * + *

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

Mirror: {@link #COLUMN_DEFAULT_PREFIX} is the shared contract with the client decoder; + * keep it in sync. Further V3 features ride the same {@code openhouse.read-bridge.*} namespace as + * additional keys. + */ +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."; + + private final ColumnDefaultsSource columnDefaultsSource; + + public ReadBridgeConfigResolver(ColumnDefaultsSource columnDefaultsSource) { + this.columnDefaultsSource = columnDefaultsSource; + } + + public Map resolve(String databaseId, String tableId, TableDto tableDto) { + Map columnDefaults = columnDefaultsSource.defaults(tableDto); + if (columnDefaults == null || columnDefaults.isEmpty()) { + return Collections.emptyMap(); // nothing to bridge -> stamp nothing + } + 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; + } +} diff --git a/services/tables/src/main/java/com/linkedin/openhouse/tables/repository/impl/OpenHouseInternalRepositoryImpl.java b/services/tables/src/main/java/com/linkedin/openhouse/tables/repository/impl/OpenHouseInternalRepositoryImpl.java index 3f4f40487..dced7cdee 100644 --- a/services/tables/src/main/java/com/linkedin/openhouse/tables/repository/impl/OpenHouseInternalRepositoryImpl.java +++ b/services/tables/src/main/java/com/linkedin/openhouse/tables/repository/impl/OpenHouseInternalRepositoryImpl.java @@ -24,6 +24,7 @@ import com.linkedin.openhouse.internal.catalog.fileio.FileIOManager; import com.linkedin.openhouse.internal.catalog.model.SoftDeletedTableDto; import com.linkedin.openhouse.internal.catalog.model.SoftDeletedTablePrimaryKey; +import com.linkedin.openhouse.tables.api.spec.v0.request.components.Policies; import com.linkedin.openhouse.tables.common.TableType; import com.linkedin.openhouse.tables.dto.mapper.TablesMapper; import com.linkedin.openhouse.tables.dto.mapper.iceberg.PartitionSpecMapper; @@ -36,6 +37,7 @@ import com.linkedin.openhouse.tables.repository.SchemaValidator; import io.micrometer.core.instrument.MeterRegistry; import io.opentelemetry.instrumentation.annotations.WithSpan; +import java.util.ArrayList; import java.util.HashSet; import java.util.List; import java.util.Map; @@ -150,6 +152,7 @@ public TableDto save(TableDto tableDto) { tableIdentifier, System.currentTimeMillis() - startTime); } else if (tableDto.isStageReplace() || tableDto.isReplaceCommit()) { + validateReplaceTable(tableIdentifier); PartitionSpec partitionSpec = partitionSpecMapper.toPartitionSpec(tableDto); log.info( "Replacing a user table: {} with schema: {} and partitionSpec: {}", @@ -313,6 +316,55 @@ protected void creationEligibilityCheck(TableDto tableDto) { versionCheck(null, tableDto); } + /** + * RTAS is only permitted when the {@value CatalogConstants#RTAS_ENABLED_TABLE_PROP} table + * property is set. And it is not allowed on table where WAP or replication is enabled. + * + * @param tableIdentifier identifier for the table + */ + protected void validateReplaceTable(TableIdentifier tableIdentifier) { + // A replace is only permitted when RTAS is enabled on the table. + Table table = catalog.loadTable(tableIdentifier); + Map existingProperties = table.properties(); + if (!Boolean.parseBoolean(existingProperties.get(RTAS_ENABLED_TABLE_PROP))) { + throw new UnsupportedClientOperationException( + UnsupportedClientOperationException.Operation.RTAS_DISABLED, + String.format( + "REPLACE TABLE AS SELECT is not enabled for table openhouse.%s.%s. You can enable this feature with 'ALTER TABLE openhouse.%s.%s SET TBLPROPERTIES ('%s'='true')'", + tableIdentifier.namespace(), + tableIdentifier.name(), + tableIdentifier.namespace(), + tableIdentifier.name(), + RTAS_ENABLED_TABLE_PROP)); + } + + // RTAS is incompatible with WAP or replication. + boolean wapEnabled = Boolean.parseBoolean(existingProperties.get(WAP_ENABLED_TABLE_PROP)); + Policies existingPolicies = + policiesMapper.toPoliciesObject(existingProperties.get(POLICIES_KEY)); + boolean replicationEnabled = + existingPolicies != null + && existingPolicies.getReplication() != null + && existingPolicies.getReplication().getConfig() != null + && !existingPolicies.getReplication().getConfig().isEmpty(); + if (wapEnabled || replicationEnabled) { + List conflictingFeatures = new ArrayList<>(); + if (wapEnabled) { + conflictingFeatures.add(String.format("WAP ('%s=true')", WAP_ENABLED_TABLE_PROP)); + } + if (replicationEnabled) { + conflictingFeatures.add("replication"); + } + throw new UnsupportedClientOperationException( + UnsupportedClientOperationException.Operation.RTAS_DISABLED, + String.format( + "REPLACE TABLE AS SELECT cannot be performed on table %s.%s while %s is enabled.", + tableIdentifier.namespace(), + tableIdentifier.name(), + String.join(" and ", conflictingFeatures))); + } + } + /** * Check the eligibility of table updates. Throw exceptions when invalidate behaviors detected for * {@link com.linkedin.openhouse.common.exception.handler.OpenHouseExceptionHandler} to deal with diff --git a/services/tables/src/main/java/com/linkedin/openhouse/tables/services/TablesServiceImpl.java b/services/tables/src/main/java/com/linkedin/openhouse/tables/services/TablesServiceImpl.java index 898a12b9b..89e48225f 100644 --- a/services/tables/src/main/java/com/linkedin/openhouse/tables/services/TablesServiceImpl.java +++ b/services/tables/src/main/java/com/linkedin/openhouse/tables/services/TablesServiceImpl.java @@ -104,6 +104,7 @@ public Pair putTable( Boolean failOnExist) { String databaseId = createUpdateTableRequestBody.getDatabaseId(); String tableId = createUpdateTableRequestBody.getTableId(); + Optional tableDto = openHouseInternalRepository.findById( TableDtoPrimaryKey.builder().databaseId(databaseId).tableId(tableId).build()); diff --git a/services/tables/src/test/java/com/linkedin/openhouse/tables/e2e/h2/SnapshotsControllerTest.java b/services/tables/src/test/java/com/linkedin/openhouse/tables/e2e/h2/SnapshotsControllerTest.java index 763daf0a2..12c9a03cb 100644 --- a/services/tables/src/test/java/com/linkedin/openhouse/tables/e2e/h2/SnapshotsControllerTest.java +++ b/services/tables/src/test/java/com/linkedin/openhouse/tables/e2e/h2/SnapshotsControllerTest.java @@ -13,6 +13,7 @@ import com.linkedin.openhouse.cluster.storage.StorageManager; import com.linkedin.openhouse.cluster.storage.local.LocalStorage; import com.linkedin.openhouse.common.test.cluster.PropertyOverrideContextInitializer; +import com.linkedin.openhouse.internal.catalog.CatalogConstants; import com.linkedin.openhouse.tables.api.spec.v0.request.CreateUpdateTableRequestBody; import com.linkedin.openhouse.tables.api.spec.v0.request.IcebergSnapshotsRequestBody; import com.linkedin.openhouse.tables.api.spec.v0.response.GetTableResponseBody; @@ -312,6 +313,12 @@ public void testPutSnapshotsReplicaTableType(GetTableResponseBody getTableRespon @MethodSource("responseBodyFeeder") public void testPutSnapshotsReplaceCommit(GetTableResponseBody getTableResponseBody) throws Exception { + // Enable RTAS and remove policy + Map propsWithRtas = new HashMap<>(getTableResponseBody.getTableProperties()); + propsWithRtas.put(CatalogConstants.RTAS_ENABLED_TABLE_PROP, "true"); + getTableResponseBody = + getTableResponseBody.toBuilder().tableProperties(propsWithRtas).policies(null).build(); + // Step 1: Create a table MvcResult createResult = RequestAndValidateHelper.createTableAndValidateResponse( diff --git a/services/tables/src/test/java/com/linkedin/openhouse/tables/e2e/h2/TablesControllerTest.java b/services/tables/src/test/java/com/linkedin/openhouse/tables/e2e/h2/TablesControllerTest.java index 777adc6fa..0e4b6d6cd 100644 --- a/services/tables/src/test/java/com/linkedin/openhouse/tables/e2e/h2/TablesControllerTest.java +++ b/services/tables/src/test/java/com/linkedin/openhouse/tables/e2e/h2/TablesControllerTest.java @@ -51,6 +51,7 @@ import io.micrometer.core.instrument.simple.SimpleMeterRegistry; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.LinkedHashMap; @@ -829,8 +830,13 @@ public void testCreateTableWithIncorrectTableTypeThrowsException() throws Except @SneakyThrows @Test public void testStagedReplace() { - GetTableResponseBody stageReplaceTable = + // Enable RTAS and remove policy + GetTableResponseBody baseTable = TableModelConstants.buildGetTableResponseBodyWithDbTbl("d_sr", "t_sr"); + Map propsWithRtas = new HashMap<>(baseTable.getTableProperties()); + propsWithRtas.put(CatalogConstants.RTAS_ENABLED_TABLE_PROP, "true"); + GetTableResponseBody stageReplaceTable = + baseTable.toBuilder().tableProperties(propsWithRtas).policies(null).build(); // Create a real table first MvcResult createResult = @@ -904,6 +910,130 @@ public void testStagedReplace() { RequestAndValidateHelper.deleteTableAndValidateResponse(mvc, stageReplaceTable); } + @SneakyThrows + @Test + public void testStagedReplaceFailsWhenRtasDisabled() { + // Table created without enabling RTAS; replace is disabled by default. + GetTableResponseBody table = + TableModelConstants.buildGetTableResponseBodyWithDbTbl("d_sr_dis", "t_sr_dis"); + MvcResult createResult = + RequestAndValidateHelper.createTableAndValidateResponse(table, mvc, storageManager); + String originalTableLocation = + JsonPath.read(createResult.getResponse().getContentAsString(), "$.tableLocation"); + + // A stageReplace against a table without replaceEnabled should be rejected. + mvc.perform( + MockMvcRequestBuilders.post( + String.format( + ValidationUtilities.CURRENT_MAJOR_VERSION_PREFIX + "/databases/%s/tables/", + table.getDatabaseId())) + .contentType(MediaType.APPLICATION_JSON) + .content( + buildCreateUpdateTableRequestBody(table) + .toBuilder() + .baseTableVersion(originalTableLocation) + .stageReplace(true) + .build() + .toJson()) + .accept(MediaType.APPLICATION_JSON)) + .andExpect(status().isBadRequest()) + .andExpect(jsonPath("$.message", containsString("REPLACE TABLE AS SELECT is not enabled"))); + + RequestAndValidateHelper.deleteTableAndValidateResponse(mvc, table); + } + + @SneakyThrows + @Test + public void testReplaceWithWapEnabledIsRejected() { + // Enable RTAS and remove policy + GetTableResponseBody baseTable = + TableModelConstants.buildGetTableResponseBodyWithDbTbl("d_sr", "t_sr"); + Map props = new HashMap<>(baseTable.getTableProperties()); + props.put(CatalogConstants.RTAS_ENABLED_TABLE_PROP, "true"); + props.put(CatalogConstants.WAP_ENABLED_TABLE_PROP, "true"); + GetTableResponseBody table = + baseTable.toBuilder().tableProperties(props).policies(null).build(); + MvcResult createResult = + RequestAndValidateHelper.createTableAndValidateResponse(table, mvc, storageManager); + String originalTableLocation = + JsonPath.read(createResult.getResponse().getContentAsString(), "$.tableLocation"); + + // A stageReplace whose resulting table enables WAP must be rejected. + mvc.perform( + MockMvcRequestBuilders.post( + String.format( + ValidationUtilities.CURRENT_MAJOR_VERSION_PREFIX + "/databases/%s/tables/", + table.getDatabaseId())) + .contentType(MediaType.APPLICATION_JSON) + .content( + buildCreateUpdateTableRequestBody(table) + .toBuilder() + .tableProperties(props) + .baseTableVersion(originalTableLocation) + .stageReplace(true) + .build() + .toJson()) + .accept(MediaType.APPLICATION_JSON)) + .andExpect(status().isBadRequest()) + .andExpect( + jsonPath("$.message", containsString("REPLACE TABLE AS SELECT cannot be performed"))) + .andExpect(jsonPath("$.message", containsString("WAP"))); + + RequestAndValidateHelper.deleteTableAndValidateResponse(mvc, table); + } + + @SneakyThrows + @Test + public void testReplaceWithReplicationEnabledIsRejected() { + // Enable RTAS + GetTableResponseBody baseTable = + TableModelConstants.buildGetTableResponseBodyWithDbTbl("d_sr", "t_sr"); + Map propsWithRtas = new HashMap<>(baseTable.getTableProperties()); + propsWithRtas.put(CatalogConstants.RTAS_ENABLED_TABLE_PROP, "true"); + GetTableResponseBody table = baseTable.toBuilder().tableProperties(propsWithRtas).build(); + MvcResult createResult = + RequestAndValidateHelper.createTableAndValidateResponse(table, mvc, storageManager); + String originalTableLocation = + JsonPath.read(createResult.getResponse().getContentAsString(), "$.tableLocation"); + + // A stageReplace whose resulting table enables replication must be rejected. + Policies policiesWithReplication = + table + .getPolicies() + .toBuilder() + .replication( + Replication.builder() + .config( + Collections.singletonList( + ReplicationConfig.builder() + .destination("CLUSTER1") + .interval("12H") + .build())) + .build()) + .build(); + mvc.perform( + MockMvcRequestBuilders.post( + String.format( + ValidationUtilities.CURRENT_MAJOR_VERSION_PREFIX + "/databases/%s/tables/", + table.getDatabaseId())) + .contentType(MediaType.APPLICATION_JSON) + .content( + buildCreateUpdateTableRequestBody(table) + .toBuilder() + .policies(policiesWithReplication) + .baseTableVersion(originalTableLocation) + .stageReplace(true) + .build() + .toJson()) + .accept(MediaType.APPLICATION_JSON)) + .andExpect(status().isBadRequest()) + .andExpect( + jsonPath("$.message", containsString("REPLACE TABLE AS SELECT cannot be performed"))) + .andExpect(jsonPath("$.message", containsString("replication"))); + + RequestAndValidateHelper.deleteTableAndValidateResponse(mvc, table); + } + @SneakyThrows @Test public void testStagedCreateDoesntExistInConsecutiveCalls() { diff --git a/services/tables/src/test/java/com/linkedin/openhouse/tables/model/TableModelConstants.java b/services/tables/src/test/java/com/linkedin/openhouse/tables/model/TableModelConstants.java index 90f16f1dd..dcd3eb0bd 100644 --- a/services/tables/src/test/java/com/linkedin/openhouse/tables/model/TableModelConstants.java +++ b/services/tables/src/test/java/com/linkedin/openhouse/tables/model/TableModelConstants.java @@ -560,8 +560,10 @@ public static GetTableResponseBody buildGetTableResponseBody(MvcResult mvcResult new ClusteringSpecConverter() .convertToEntityAttribute(JsonPath.read(content, "$.clustering").toString())) .policies( - new PoliciesSpecConverter() - .convertToEntityAttribute(JsonPath.read(content, "$.policies").toString())) + JsonPath.read(content, "$.policies") == null + ? null + : new PoliciesSpecConverter() + .convertToEntityAttribute(JsonPath.read(content, "$.policies").toString())) .sortOrder(JsonPath.read(content, "$.sortOrder")) .build(); } 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 new file mode 100644 index 000000000..a5e06a02f --- /dev/null +++ b/services/tables/src/test/java/com/linkedin/openhouse/tables/readbridge/ReadBridgeConfigResolverTest.java @@ -0,0 +1,114 @@ +package com.linkedin.openhouse.tables.readbridge; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.IntNode; +import com.fasterxml.jackson.databind.node.TextNode; +import com.linkedin.openhouse.common.api.spec.ApiResponse; +import com.linkedin.openhouse.tables.api.handler.impl.OpenHouseTablesApiHandler; +import com.linkedin.openhouse.tables.api.spec.v0.response.GetTableResponseBody; +import com.linkedin.openhouse.tables.api.validator.TablesApiValidator; +import com.linkedin.openhouse.tables.dto.mapper.TablesMapper; +import com.linkedin.openhouse.tables.model.TableDto; +import com.linkedin.openhouse.tables.services.TablesService; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.springframework.test.util.ReflectionTestUtils; + +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 String PREFIX = ReadBridgeConfigResolver.COLUMN_DEFAULT_PREFIX; + + @Test + public void testEmptyWhenNoColumnDefaults() { + Assertions.assertTrue( + new ReadBridgeConfigResolver(NONE).resolve("db", "tbl", mock(TableDto.class)).isEmpty()); + } + + @Test + public void testStampsColumnDefaultEntry() { + ColumnDefaultsSource source = tableDto -> Collections.singletonMap(5, TextNode.valueOf("US")); + Map config = + new ReadBridgeConfigResolver(source).resolve("db", "tbl", mock(TableDto.class)); + // value is the single-value JSON for the default ("US" -> "\"US\""). + Assertions.assertEquals("\"US\"", config.get(PREFIX + "5")); + } + + @Test + public void testStampsAllColumnDefaultsAsSeparateEntries() { + ColumnDefaultsSource source = + tableDto -> { + Map defaults = new LinkedHashMap<>(); + defaults.put(5, TextNode.valueOf("US")); + defaults.put(7, IntNode.valueOf(0)); + return defaults; + }; + Map config = + new ReadBridgeConfigResolver(source).resolve("db", "tbl", mock(TableDto.class)); + Assertions.assertEquals(2, config.size()); + Assertions.assertEquals("\"US\"", config.get(PREFIX + "5")); + Assertions.assertEquals("0", config.get(PREFIX + "7")); + } + + /** getTable stamps the resolver's config onto the response body. */ + @Test + public void testGetTableStampsResolvedConfig() { + TablesService tableService = mock(TablesService.class); + TablesMapper tablesMapper = mock(TablesMapper.class); + ReadBridgeConfigResolver resolver = mock(ReadBridgeConfigResolver.class); + + TableDto tableDto = mock(TableDto.class); + when(tableService.getTable("db", "tbl", "principal")).thenReturn(tableDto); + when(tablesMapper.toGetTableResponseBody(tableDto)) + .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); + + OpenHouseTablesApiHandler handler = handlerWith(tableService, tablesMapper, resolver); + + ApiResponse response = handler.getTable("db", "tbl", "principal"); + + Assertions.assertSame(resolved, response.getResponseBody().getConfig()); + } + + /** With the behaviorless open-source source wired in, getTable leaves config empty. */ + @Test + public void testGetTableLeavesConfigEmptyWithNoColumnDefaults() { + TablesService tableService = mock(TablesService.class); + TablesMapper tablesMapper = mock(TablesMapper.class); + + TableDto tableDto = mock(TableDto.class); + when(tableService.getTable(anyString(), anyString(), anyString())).thenReturn(tableDto); + when(tablesMapper.toGetTableResponseBody(any())) + .thenReturn(GetTableResponseBody.builder().tableId("tbl").databaseId("db").build()); + + OpenHouseTablesApiHandler handler = + handlerWith(tableService, tablesMapper, new ReadBridgeConfigResolver(NONE)); + + ApiResponse response = handler.getTable("db", "tbl", "principal"); + + Assertions.assertTrue(response.getResponseBody().getConfig().isEmpty()); + } + + private OpenHouseTablesApiHandler handlerWith( + TablesService tableService, TablesMapper tablesMapper, ReadBridgeConfigResolver resolver) { + OpenHouseTablesApiHandler handler = new OpenHouseTablesApiHandler(); + ReflectionTestUtils.setField(handler, "tablesApiValidator", mock(TablesApiValidator.class)); + ReflectionTestUtils.setField(handler, "tableService", tableService); + ReflectionTestUtils.setField(handler, "tablesMapper", tablesMapper); + ReflectionTestUtils.setField(handler, "readBridgeConfigResolver", resolver); + return handler; + } +}