perf(ebean-dao): move SharedSchemaCache pre-warm to constructor (#618)

PR #615 introduced SharedSchemaCache with pre-warm calls inside
EbeanLocalAccess.ensureSchemaUpToDate(). In deploys where schema
migrations run as a separate job (LinkedIn's prod pattern, and likely
others), ensureSchemaUpToDate() is gated off, so pre-warm never runs
and the first request after JVM startup pays the inline
information_schema query cost (the "Inline schema cache miss" log
line).

Moves the pre-warm calls to the EbeanLocalAccess constructor so they
always run, regardless of whether schema evolution is enabled.
refreshTable() already has its own try/catch, so a transient DB
hiccup during construction logs a warning rather than failing boot.

Also drops the unused tableColumns field (dead code, replaced by the
shared cache in #615).

Two supporting changes:

1. SharedSchemaCache.clearRegistry() now also calls clearCaches() on
   each held instance before wiping the REGISTRY map. Previously
   clearing only the static map left stale entries readable through
   any reference (e.g. EbeanLocalAccess.validator) that callers were
   already holding -- a footgun that the @VisibleForTesting name did
   not suggest.

2. EbeanLocalDAOTest.addIndex() helper now calls clearRegistry()
   after its dynamic ALTER TABLE ADD COLUMN. With pre-warm now
   running in the constructor, the cache snapshots the schema before
   addIndex modifies it; the explicit invalidation keeps subsequent
   columnExists()/indexExists() lookups in sync with the actual
   table state.

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: mridul111998

dao-impl/ebean-dao/src/main/java/com/linkedin/metadata/dao/EbeanLocalAccess.java

@@ -42,7 +42,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
@@ -83,9 +82,6 @@ public class EbeanLocalAccess<URN extends Urn> implements IEbeanLocalAccess<URN>
   private static final String DEFAULT_ACTOR = "urn:li:principal:UNKNOWN";
   private static final String EBEAN_SERVER_CONFIG = "EbeanServerConfig";
 
-  // key: table_name,
-  // value: Set(column1, column2, column3 ...)
-  private final Map<String, Set<String>> tableColumns = new ConcurrentHashMap<>();
   private final SchemaValidatorUtil validator;
 
   public EbeanLocalAccess(EbeanServer server, ServerConfig serverConfig, @Nonnull Class<URN> urnClass,
@@ -97,6 +93,13 @@ public EbeanLocalAccess(EbeanServer server, ServerConfig serverConfig, @Nonnull
     _schemaEvolutionManager = createSchemaEvolutionManager(serverConfig);
     _nonDollarVirtualColumnsEnabled = nonDollarVirtualColumnsEnabled;
     validator = buildValidator(server, serverConfig);
+    // Pre-warm the shared cache for both the prod and test tables so the first request after
+    // JVM start never pays the inline information_schema query cost. Done here rather than in
+    // ensureSchemaUpToDate() because consumers that run schema migrations as a separate job
+    // (e.g., LinkedIn's prod deploy pattern) gate ensureSchemaUpToDate() off, so a pre-warm
+    // there would never run and the cold-start cache miss would still hit the request path.
+    validator.registerAndPreWarm(getTableName(_entityType));
+    validator.registerAndPreWarm(getTestTableName(_entityType));
   }
 
   private static SchemaValidatorUtil buildValidator(EbeanServer server, ServerConfig serverConfig) {
@@ -135,9 +138,7 @@ public void configureOptionalForceIndex(@Nullable String indexName,
 
   public void ensureSchemaUpToDate() {
     _schemaEvolutionManager.ensureSchemaUpToDate();
-    // Pre-warm the shared cache for both the prod and test tables so the first request is never slow.
-    validator.registerAndPreWarm(getTableName(_entityType));
-    validator.registerAndPreWarm(getTestTableName(_entityType));
+    // Cache pre-warm moved to constructor — see comment there for rationale.
     validateForceIndex();
   }
 
@@ -254,7 +255,7 @@ public <ASPECT_UNION extends RecordTemplate> int create(
     String onDuplicateKeyClause = buildOnDuplicateKeyForCreate(urn, aspectCreateLambdas);
 
     // Use comprehensive helper to prepare SqlUpdate with all common logic
-    SqlUpdate sqlUpdate = prepareMultiColumnInsert(urn, aspectValues, aspectCreateLambdas, 
+    SqlUpdate sqlUpdate = prepareMultiColumnInsert(urn, aspectValues, aspectCreateLambdas,
         auditStamp, ingestionTrackingContext, onDuplicateKeyClause);
 
     return sqlUpdate.execute();
@@ -282,7 +283,7 @@ public <ASPECT_UNION extends RecordTemplate> int batchUpsert(
     // Extract parallel lists from contexts for prepareMultiColumnInsert
     List<RecordTemplate> aspectValues = new ArrayList<>();
     List<BaseLocalDAO.AspectUpdateLambda<? extends RecordTemplate>> aspectUpdateLambdas = new ArrayList<>();
-    
+
     for (BaseLocalDAO.AspectUpdateContext<RecordTemplate> ctx : updateContexts) {
       aspectValues.add(ctx.getNewValue());
       aspectUpdateLambdas.add(ctx.getLambda());
@@ -292,7 +293,7 @@ public <ASPECT_UNION extends RecordTemplate> int batchUpsert(
     String onDuplicateKeyClause = buildOnDuplicateKeyForUpsert(urn, aspectUpdateLambdas);
 
     // Use comprehensive helper to prepare SqlUpdate with all common logic
-    SqlUpdate sqlUpdate = prepareMultiColumnInsert(urn, aspectValues, aspectUpdateLambdas, 
+    SqlUpdate sqlUpdate = prepareMultiColumnInsert(urn, aspectValues, aspectUpdateLambdas,
         auditStamp, ingestionTrackingContext, onDuplicateKeyClause);
 
     return sqlUpdate.execute();
@@ -889,21 +890,21 @@ private SqlUpdate prepareMultiColumnInsert(
       @Nonnull AuditStamp auditStamp,
       @Nullable IngestionTrackingContext ingestionTrackingContext,
       @Nonnull String onDuplicateKeyClause) {
-    
+
     // Validate that aspectValues and aspectLambdas have the same size
     if (aspectValues.size() != aspectLambdas.size()) {
       throw new IllegalArgumentException(
           String.format("Aspect values size (%d) must match aspect lambdas size (%d)",
               aspectValues.size(), aspectLambdas.size()));
     }
-    
+
     // Validate that no aspect values are null
     aspectValues.forEach(aspectValue -> {
       if (aspectValue == null) {
         throw new IllegalArgumentException("Aspect value cannot be null");
       }
     });
-    
+
     // Extract audit information
     final long timestamp = auditStamp.hasTime() ? auditStamp.getTime() : System.currentTimeMillis();
     final String actor = auditStamp.hasActor() ? auditStamp.getActor().toString() : DEFAULT_ACTOR;
@@ -969,7 +970,7 @@ private SqlUpdate prepareMultiColumnInsert(
     if (urnExtraction) {
       sqlUpdate.setParameter("a_urn", toJsonString(urn));
     }
-    
+
     // Set common parameters
     sqlUpdate.setParameter("urn", urn.toString())
         .setParameter("lastmodifiedon", utcTimestamp)
@@ -987,13 +988,13 @@ private SqlUpdate prepareMultiColumnInsert(
    * @return the ON DUPLICATE KEY UPDATE clause string
    */
   private String buildOnDuplicateKeyForCreate(
-      @Nonnull URN urn, 
+      @Nonnull URN urn,
       @Nonnull List<? extends BaseLocalDAO.AspectUpdateLambda<? extends RecordTemplate>> aspectLambdas) {
-    
+
     List<String> classNames = aspectLambdas.stream()
         .map(lambda -> lambda.getAspectClass().getCanonicalName())
         .collect(Collectors.toList());
-    
+
     StringBuilder onDuplicateKey = new StringBuilder();
     onDuplicateKey.append(ON_DUPLICATE_KEY_UPDATE);
     for (int i = 0; i < classNames.size(); i++) {
@@ -1016,13 +1017,13 @@ private String buildOnDuplicateKeyForCreate(
    * @return the ON DUPLICATE KEY UPDATE clause string
    */
   private String buildOnDuplicateKeyForUpsert(
-      @Nonnull URN urn, 
+      @Nonnull URN urn,
       @Nonnull List<? extends BaseLocalDAO.AspectUpdateLambda<? extends RecordTemplate>> aspectLambdas) {
-    
+
     List<String> classNames = aspectLambdas.stream()
         .map(lambda -> lambda.getAspectClass().getCanonicalName())
         .collect(Collectors.toList());
-    
+
     StringBuilder onDuplicateKey = new StringBuilder();
     onDuplicateKey.append(ON_DUPLICATE_KEY_UPDATE);
     for (int i = 0; i < classNames.size(); i++) {

dao-impl/ebean-dao/src/main/java/com/linkedin/metadata/dao/utils/SharedSchemaCache.java

@@ -210,6 +210,13 @@ void clearCaches() {
 
   @VisibleForTesting
   public static void clearRegistry() {
+    // Invalidate the in-memory caches on every held instance before dropping references from
+    // the registry. Callers that already hold a reference to a SharedSchemaCache (e.g. via
+    // EbeanLocalAccess.validator) will then see the cache miss on their next read instead of
+    // a stale entry.
+    for (SharedSchemaCache cache : REGISTRY.values()) {
+      cache.clearCaches();
+    }
     REGISTRY.clear();
   }
 }

dao-impl/ebean-dao/src/test/java/com/linkedin/metadata/dao/EbeanLocalDAOTest.java

@@ -4133,9 +4133,11 @@ private void addIndex(Urn urn, String aspectName, String pathName, Object val) {
     String checkColumnExistance = String.format("SELECT * FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = '%s' AND"
         + " TABLE_NAME = '%s' AND COLUMN_NAME = '%s'", getDatabaseName(), getTableName(urn), fullIndexColumnName);
 
+    boolean schemaChanged = false;
     if (_server.createSqlQuery(checkColumnExistance).findList().isEmpty()) {
       String sqlUpdate = String.format("ALTER TABLE %s ADD COLUMN %s VARCHAR(255);", getTableName(urn), fullIndexColumnName);
       _server.execute(Ebean.createSqlUpdate(sqlUpdate));
+      schemaChanged = true;
     }
 
     checkColumnExistance = String.format("SELECT * FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = '%s' AND"
@@ -4144,6 +4146,15 @@ private void addIndex(Urn urn, String aspectName, String pathName, Object val) {
     if (aspectColumnName != null && _server.createSqlQuery(checkColumnExistance).findList().isEmpty()) {
       String sqlUpdate = String.format("ALTER TABLE %s ADD COLUMN %s VARCHAR(255);", getTableName(urn), aspectColumnName);
       _server.execute(Ebean.createSqlUpdate(sqlUpdate));
+      schemaChanged = true;
+    }
+    // EbeanLocalAccess pre-warms the SharedSchemaCache in its constructor (capturing the schema
+    // at DAO construction time). The ALTER TABLE above adds a new virtual/aspect column AFTER
+    // construction, so the held cache is now stale. Clear the registry to invalidate every
+    // held instance's caches; subsequent columnExists()/indexExists() lookups then re-query
+    // information_schema and see the new column.
+    if (schemaChanged) {
+      SharedSchemaCache.clearRegistry();
     }
 
     // finally, we need to update the newly added column with the passed-in value.