fix(ebean-dao): honor isTestMode in batchUpsert and create paths

prepareMultiColumnInsert hardcoded getTableName(urn), silently dropping the
isTestMode flag both callers passed. Both batchUpsert and create therefore
always wrote to the production table even when the caller requested test
mode, which made test-table dual-write features (e.g. the shadow write in
metadata-graph-assets) effectively double-write to production.

Thread isTestMode into prepareMultiColumnInsert and route between
getTableName / getTestTableName at the SQL build site, matching the pattern
already used by the per-aspect add path and the helpers in SQLStatementUtils.

Adds a regression test for the batchUpsert + isTestMode=true case and the
missing metadata_entity_foo_test schema rows in both test SQL fixtures.

Author: jphui

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

@@ -256,7 +256,7 @@ public <ASPECT_UNION extends RecordTemplate> int create(
 
     // Use comprehensive helper to prepare SqlUpdate with all common logic
     SqlUpdate sqlUpdate = prepareMultiColumnInsert(urn, aspectValues, aspectCreateLambdas,
-        auditStamp, ingestionTrackingContext, onDuplicateKeyClause);
+        auditStamp, ingestionTrackingContext, onDuplicateKeyClause, isTestMode);
 
     return sqlUpdate.execute();
   }
@@ -294,7 +294,7 @@ public <ASPECT_UNION extends RecordTemplate> int batchUpsert(
 
     // Use comprehensive helper to prepare SqlUpdate with all common logic
     SqlUpdate sqlUpdate = prepareMultiColumnInsert(urn, aspectValues, aspectUpdateLambdas,
-        auditStamp, ingestionTrackingContext, onDuplicateKeyClause);
+        auditStamp, ingestionTrackingContext, onDuplicateKeyClause, isTestMode);
 
     return sqlUpdate.execute();
   }
@@ -889,7 +889,8 @@ private SqlUpdate prepareMultiColumnInsert(
       @Nonnull List<? extends BaseLocalDAO.AspectUpdateLambda<? extends RecordTemplate>> aspectLambdas,
       @Nonnull AuditStamp auditStamp,
       @Nullable IngestionTrackingContext ingestionTrackingContext,
-      @Nonnull String onDuplicateKeyClause) {
+      @Nonnull String onDuplicateKeyClause,
+      boolean isTestMode) {
 
     // Validate that aspectValues and aspectLambdas have the same size
     if (aspectValues.size() != aspectLambdas.size()) {
@@ -942,7 +943,8 @@ private SqlUpdate prepareMultiColumnInsert(
 
     // Build complete SQL statement with ON DUPLICATE KEY clause
     String insertStatement = insertIntoSql.toString() + insertSqlValues.toString() + onDuplicateKeyClause;
-    insertStatement = String.format(insertStatement, getTableName(urn));
+    insertStatement = String.format(insertStatement,
+        isTestMode ? getTestTableName(urn) : getTableName(urn));
 
     // Create SqlUpdate and set all parameters
     SqlUpdate sqlUpdate = _server.createSqlUpdate(insertStatement);

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

@@ -701,12 +701,40 @@ public void testBatchUpsertWithIngestionTrackingContext() {
     assertEquals("{\"value\":\"tracked_value\"}", results.get(0).getMetadata());
 
     // Verify IngestionTrackingContext fields are persisted and readable
-    assertEquals("test-emitter", results.get(0).getEmitter(), 
+    assertEquals("test-emitter", results.get(0).getEmitter(),
         "Emitter from IngestionTrackingContext should be persisted");
-    assertEquals(Long.valueOf(emitTime), results.get(0).getEmitTime(), 
+    assertEquals(Long.valueOf(emitTime), results.get(0).getEmitTime(),
         "EmitTime from IngestionTrackingContext should be persisted");
   }
 
+  /**
+   * Tests that batchUpsert() with isTestMode=true writes to the test table rather than the
+   * production table. Regression coverage for the previously-dropped flag in
+   * {@code prepareMultiColumnInsert}, which used to hardcode {@code getTableName(urn)}.
+   */
+  @Test
+  public void testBatchUpsertWithTestModeWritesToTestTable() {
+    // Arrange
+    FooUrn fooUrn = makeFooUrn(310);
+    AspectFoo foo = new AspectFoo().setValue("test_mode_value");
+    List<BaseLocalDAO.AspectUpdateContext<RecordTemplate>> updateContexts =
+        Collections.singletonList(new BaseLocalDAO.AspectUpdateContext<>(null, foo,
+            new BaseLocalDAO.AspectUpdateLambda<>(foo)));
+    AuditStamp auditStamp = makeAuditStamp("actor", _now);
+
+    // Act
+    int result = _ebeanLocalAccessFoo.batchUpsert(fooUrn, updateContexts, auditStamp, null, true);
+
+    // Assert - row should land in the test table, readable via batchGetUnion with isTestMode=true
+    assertEquals(result, 1);
+    AspectKey<FooUrn, AspectFoo> aspectKey = new AspectKey<>(AspectFoo.class, fooUrn, 0L);
+    List<EbeanMetadataAspect> testTableResults = _ebeanLocalAccessFoo.batchGetUnion(
+        Collections.singletonList(aspectKey), 1, 0, false, true);
+    assertEquals(1, testTableResults.size());
+    assertEquals("{\"value\":\"test_mode_value\"}", testTableResults.get(0).getMetadata());
+    assertEquals(fooUrn.toString(), testTableResults.get(0).getKey().getUrn());
+  }
+
   // ==================== readDeletionInfoBatch tests ====================
 
   /**

dao-impl/ebean-dao/src/test/resources/ebean-local-access-create-all-with-non-dollar-virtual-column-names.sql

@@ -1,4 +1,5 @@
 DROP TABLE IF EXISTS metadata_entity_foo;
+DROP TABLE IF EXISTS metadata_entity_foo_test;
 DROP TABLE IF EXISTS metadata_entity_bar;
 DROP TABLE IF EXISTS metadata_entity_burger;
 DROP TABLE IF EXISTS metadata_aspect;
@@ -16,6 +17,16 @@ CREATE TABLE IF NOT EXISTS metadata_entity_foo (
     CONSTRAINT pk_metadata_entity_foo PRIMARY KEY (urn)
 );
 
+-- initialize foo test entity table (mirror of metadata_entity_foo, used for isTestMode writes)
+CREATE TABLE IF NOT EXISTS metadata_entity_foo_test (
+    urn VARCHAR(100) NOT NULL,
+    lastmodifiedon TIMESTAMP NOT NULL,
+    lastmodifiedby VARCHAR(255) NOT NULL,
+    createdfor VARCHAR(255),
+    deleted_ts datetime(6) DEFAULT NULL,
+    CONSTRAINT pk_metadata_entity_foo_test PRIMARY KEY (urn)
+);
+
 -- initialize bar entity table
 CREATE TABLE IF NOT EXISTS metadata_entity_bar (
     urn VARCHAR(100) NOT NULL,
@@ -68,6 +79,7 @@ CREATE TABLE IF NOT EXISTS metadata_relationship_belongsto (
 );
 
 ALTER TABLE metadata_entity_foo ADD a_urn JSON;
+ALTER TABLE metadata_entity_foo_test ADD a_urn JSON;
 ALTER TABLE metadata_entity_bar ADD a_urn JSON;
 
 ALTER TABLE metadata_entity_foo ADD COLUMN i_urn0fooId VARCHAR(255)
@@ -78,6 +90,7 @@ ALTER TABLE metadata_entity_foo ADD a_status JSON;
 
 -- add foo aspect to foo entity
 ALTER TABLE metadata_entity_foo ADD a_aspectfoo JSON;
+ALTER TABLE metadata_entity_foo_test ADD a_aspectfoo JSON;
 
 -- add foo aspect to bar entity
 ALTER TABLE metadata_entity_bar ADD a_aspectfoo JSON;

dao-impl/ebean-dao/src/test/resources/ebean-local-access-create-all.sql

@@ -1,4 +1,5 @@
 DROP TABLE IF EXISTS metadata_entity_foo;
+DROP TABLE IF EXISTS metadata_entity_foo_test;
 DROP TABLE IF EXISTS metadata_entity_bar;
 DROP TABLE IF EXISTS metadata_entity_burger;
 DROP TABLE IF EXISTS metadata_aspect;
@@ -16,6 +17,16 @@ CREATE TABLE IF NOT EXISTS metadata_entity_foo (
     CONSTRAINT pk_metadata_entity_foo PRIMARY KEY (urn)
 );
 
+-- initialize foo test entity table (mirror of metadata_entity_foo, used for isTestMode writes)
+CREATE TABLE IF NOT EXISTS metadata_entity_foo_test (
+    urn VARCHAR(100) NOT NULL,
+    lastmodifiedon TIMESTAMP NOT NULL,
+    lastmodifiedby VARCHAR(255) NOT NULL,
+    createdfor VARCHAR(255),
+    deleted_ts datetime(6) DEFAULT NULL,
+    CONSTRAINT pk_metadata_entity_foo_test PRIMARY KEY (urn)
+);
+
 -- initialize bar entity table
 CREATE TABLE IF NOT EXISTS metadata_entity_bar (
     urn VARCHAR(100) NOT NULL,
@@ -68,6 +79,7 @@ CREATE TABLE IF NOT EXISTS metadata_relationship_belongsto (
 );
 
 ALTER TABLE metadata_entity_foo ADD a_urn JSON;
+ALTER TABLE metadata_entity_foo_test ADD a_urn JSON;
 ALTER TABLE metadata_entity_bar ADD a_urn JSON;
 
 ALTER TABLE metadata_entity_foo ADD COLUMN i_urn$fooId VARCHAR(255)
@@ -78,6 +90,7 @@ ALTER TABLE metadata_entity_foo ADD a_status JSON;
 
 -- add foo aspect to foo entity
 ALTER TABLE metadata_entity_foo ADD a_aspectfoo JSON;
+ALTER TABLE metadata_entity_foo_test ADD a_aspectfoo JSON;
 
 -- add foo aspect to bar entity
 ALTER TABLE metadata_entity_bar ADD a_aspectfoo JSON;