[default values] Move read-bridge column-default ramp into OpenHouse

services/tables/src/main/java/com/linkedin/openhouse/tables/api/ApiConfig.java

@@ -4,12 +4,12 @@
 import com.linkedin.openhouse.tables.api.handler.impl.OpenHouseTablesApiHandler;
 import com.linkedin.openhouse.tables.readbridge.ColumnDefaultsSource;
 import com.linkedin.openhouse.tables.readbridge.ReadBridgeConfigResolver;
-import java.util.Collections;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import com.linkedin.openhouse.tables.toggle.TableFeatureToggle;
+import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
-/** Class that holds all the Beans related to a controller. */
+/** Beans related to tables API controllers. */
 @Configuration
 public class ApiConfig {
   @Bean
@@ -18,22 +18,13 @@ public TablesApiHandler tablesApiHandler() {
   }
 
   /**
-   * Open-source default {@link ColumnDefaultsSource}: supplies none, so read-bridge stays inert.
+   * Prefer {@link ObjectProvider} over a {@code @ConditionalOnMissingBean} noop so a deployment
+   * {@code @Bean} source cannot collide with an OSS default.
    */
   @Bean
-  @ConditionalOnMissingBean(ColumnDefaultsSource.class)
-  public ColumnDefaultsSource columnDefaultsSource() {
-    return tableDto -> Collections.emptyMap();
-  }
-
-  /**
-   * Server-side encoder that stamps the read-bridge {@code config} from {@link
-   * ColumnDefaultsSource}.
-   */
-  @Bean
-  @ConditionalOnMissingBean(ReadBridgeConfigResolver.class)
   public ReadBridgeConfigResolver readBridgeConfigResolver(
-      ColumnDefaultsSource columnDefaultsSource) {
-    return new ReadBridgeConfigResolver(columnDefaultsSource);
+      ObjectProvider<ColumnDefaultsSource> columnDefaultsSource, TableFeatureToggle featureToggle) {
+    return new ReadBridgeConfigResolver(
+        columnDefaultsSource.getIfAvailable(() -> ColumnDefaultsSource.NONE), featureToggle);
   }
 }

services/tables/src/main/java/com/linkedin/openhouse/tables/api/handler/impl/OpenHouseTablesApiHandler.java

@@ -38,16 +38,9 @@ public class OpenHouseTablesApiHandler implements TablesApiHandler {
 
   @Autowired private ReadBridgeConfigResolver readBridgeConfigResolver;
 
-  /**
-   * Stamp the server-resolved, per-table client {@code config} (Iceberg REST {@code
-   * LoadTableResponse.config} convention) onto a freshly mapped response body. The mapper leaves
-   * {@code config} null; it is a request-time decision resolved here.
-   */
-  private GetTableResponseBody withConfig(
-      GetTableResponseBody body, String databaseId, String tableId, TableDto tableDto) {
-    return body.toBuilder()
-        .config(readBridgeConfigResolver.resolve(databaseId, tableId, tableDto))
-        .build();
+  /** Request-time {@code config} stamp; mapper leaves it null. */
+  private GetTableResponseBody withConfig(GetTableResponseBody body, TableDto tableDto) {
+    return body.toBuilder().config(readBridgeConfigResolver.resolve(tableDto)).build();
   }
 
   @Override
@@ -57,9 +50,7 @@ public ApiResponse<GetTableResponseBody> getTable(
     TableDto tableDto = tableService.getTable(databaseId, tableId, actingPrincipal);
     return ApiResponse.<GetTableResponseBody>builder()
         .httpStatus(HttpStatus.OK)
-        .responseBody(
-            withConfig(
-                tablesMapper.toGetTableResponseBody(tableDto), databaseId, tableId, tableDto))
+        .responseBody(withConfig(tablesMapper.toGetTableResponseBody(tableDto), tableDto))
         .build();
   }
 
@@ -111,12 +102,7 @@ public ApiResponse<GetTableResponseBody> createTable(
     TableDto tableDto = putResult.getFirst();
     return ApiResponse.<GetTableResponseBody>builder()
         .httpStatus(HttpStatus.CREATED)
-        .responseBody(
-            withConfig(
-                tablesMapper.toGetTableResponseBody(tableDto),
-                databaseId,
-                tableDto.getTableId(),
-                tableDto))
+        .responseBody(withConfig(tablesMapper.toGetTableResponseBody(tableDto), tableDto))
         .build();
   }
 
@@ -134,9 +120,7 @@ public ApiResponse<GetTableResponseBody> updateTable(
     TableDto tableDto = putResult.getFirst();
     return ApiResponse.<GetTableResponseBody>builder()
         .httpStatus(status)
-        .responseBody(
-            withConfig(
-                tablesMapper.toGetTableResponseBody(tableDto), databaseId, tableId, tableDto))
+        .responseBody(withConfig(tablesMapper.toGetTableResponseBody(tableDto), tableDto))
         .build();
   }
 

services/tables/src/main/java/com/linkedin/openhouse/tables/readbridge/ColumnDefaultsSource.java

@@ -2,28 +2,21 @@
 
 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.
- *
- * <p>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).
- *
- * <p>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 <em>is</em> 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 {
+
+  /** 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 -&gt; initial-default as Iceberg single-value JSON; empty/{@code null} = none
+   * Field-id → Iceberg single-value JSON. Empty/null stamps nothing. Omit a field that cannot bind
+   * (today's NULL); do not throw — this is the table-load path.
    */
   Map<Integer, JsonNode> defaults(TableDto tableDto);
 }

services/tables/src/main/java/com/linkedin/openhouse/tables/readbridge/ReadBridgeConfigResolver.java

@@ -2,46 +2,86 @@
 
 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.<fieldId> =
- * <single-value-json>}. The client decoder ({@code ReadBridge} in {@code openhouse-java-runtime})
- * reads these entries and overlays the defaults at metadata-load time.
- *
- * <p>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}.
- *
- * <p><b>Mirror:</b> {@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.
+ * 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 {
 
-  /** 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.";
+  /** Capability id; also names {@code <id>.enabled} and the config key prefix below. */
+  public static final String COLUMN_DEFAULT_FEATURE_ID = "read-bridge.column-default";
+
+  /** Client contract: {@code openhouse.read-bridge.column-default.<fieldId>}. */
+  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;
+  }
+
+  /** Merges independently gated capabilities; empty when nothing is bridged. */
+  public Map<String, String> resolve(TableDto tableDto) {
+    Map<String, String> config = new HashMap<>();
+    config.putAll(columnDefaultConfig(tableDto));
+    return config;
   }
 
-  public Map<String, String> resolve(String databaseId, String tableId, TableDto tableDto) {
-    Map<Integer, JsonNode> columnDefaults = columnDefaultsSource.defaults(tableDto);
+  private Map<String, String> columnDefaultConfig(TableDto tableDto) {
+    // No deployment source → skip HTS entirely.
+    if (columnDefaultsSource == ColumnDefaultsSource.NONE) {
+      return Collections.emptyMap();
+    }
+    if (!isColumnDefaultRamped(tableDto)) {
+      return Collections.emptyMap();
+    }
+    Map<Integer, JsonNode> columnDefaults;
+    try {
+      columnDefaults = columnDefaultsSource.defaults(tableDto);
+    } catch (RuntimeException e) {
+      log.warn(
+          "read-bridge: column-defaults source failed for {}.{}; treating as not bridged",
+          tableDto.getDatabaseId(),
+          tableDto.getTableId(),
+          e);
+      return Collections.emptyMap();
+    }
     if (columnDefaults == null || columnDefaults.isEmpty()) {
-      return Collections.emptyMap(); // nothing to bridge -> stamp nothing
+      return Collections.emptyMap();
     }
     Map<String, String> 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;
   }
+
+  /**
+   * 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 {
+      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;
+    }
+  }
 }