yujunjun/datahub
2
0
Fork 0
mirror of https://github.com/datahub-project/datahub.git synced 2025-12-07 08:02:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(readOnly): fixes read only mode for update queries (#15398)

Browse Source
This commit is contained in:
RyanHolstien 2025-11-25 14:31:28 -06:00 committed by GitHub
parent bbfdb51df0
commit 4833bbbd65
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 51 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
metadata-io/src/main/java/com/linkedin/metadata/entity/ebean/EbeanAspectDao.java
View File

@ -194,7 +194,7 @@ public class EbeanAspectDao implements AspectDao, AspectMigrationsDao {
.collect(Collectors.toList());
final List<EbeanAspectV2> results;
if (forUpdate) {
if (forUpdate && canWrite) {
results = server.find(EbeanAspectV2.class).where().idIn(keys).forUpdate().findList();
} else {
results = server.find(EbeanAspectV2.class).where().idIn(keys).findList();
@ -425,7 +425,7 @@ public class EbeanAspectDao implements AspectDao, AspectMigrationsDao {
}
// Add FOR UPDATE clause only once at the end of the entire statement
if (forUpdate) {
if (forUpdate && canWrite) {
sb.append(" FOR UPDATE");
}
@ -484,7 +484,7 @@ public class EbeanAspectDao implements AspectDao, AspectMigrationsDao {
sb.append(")");
if (forUpdate) {
if (forUpdate && canWrite) {
sb.append(" FOR UPDATE");
}
@ -891,7 +891,9 @@ public class EbeanAspectDao implements AspectDao, AspectMigrationsDao {
// forUpdate is required to avoid duplicate key violations (it is used as an indication that the
// max(version) was invalidated
server.find(EbeanAspectV2.class).where().idIn(forUpdateKeys).forUpdate().findList();
if (canWrite) {
server.find(EbeanAspectV2.class).where().idIn(forUpdateKeys).forUpdate().findList();
}
Junction<EbeanAspectV2> queryJunction =
server

66
metadata-io/src/test/java/com/linkedin/metadata/entity/ebean/EbeanAspectDaoTest.java
View File

@ -61,8 +61,17 @@ public class EbeanAspectDaoTest {
EbeanTestUtils.shutdownDatabase(server);
}
@Test
public void testGetNextVersionForUpdate() {
@DataProvider(name = "writabilityConfig")
public Object[][] writabilityConfigProvider() {
return new Object[][] {
{true, "Writable"}, // canWrite = true, description
{false, "ReadOnly"} // canWrite = false, description
};
}
@Test(dataProvider = "writabilityConfig")
public void testGetNextVersionForUpdate(boolean canWrite, String description) {
testDao.setWritable(canWrite);
LoggedSql.start();
testDao.runInTransactionWithRetryUnlocked(
@ -79,13 +88,22 @@ public class EbeanAspectDaoTest {
LoggedSql.stop().stream()
.filter(str -> str.contains("testGetNextVersionForUpdate"))
.toList();
assertEquals(sql.size(), 2, String.format("Found: %s", sql));
assertTrue(
sql.get(0).contains("for update;"), String.format("Did not find `for update` in %s ", sql));
if (canWrite) {
assertEquals(sql.size(), 2, String.format("Found: %s", sql));
assertTrue(
sql.get(0).contains("for update;"),
String.format("Did not find `for update` in %s ", sql));
} else {
assertEquals(sql.size(), 1, String.format("Found: %s", sql));
assertFalse(
sql.get(0).contains("for update;"), String.format("Found `for update` in %s ", sql));
}
}
@Test
public void testGetLatestAspectsForUpdate() throws JsonProcessingException {
@Test(dataProvider = "writabilityConfig")
public void testGetLatestAspectsForUpdate(boolean canWrite, String description)
throws JsonProcessingException {
testDao.setWritable(canWrite);
LoggedSql.start();
testDao.runInTransactionWithRetryUnlocked(
@ -106,12 +124,20 @@ public class EbeanAspectDaoTest {
.toList();
assertEquals(
sql.size(), 1, String.format("Found: %s", new ObjectMapper().writeValueAsString(sql)));
assertTrue(
sql.get(0).contains("for update;"), String.format("Did not find `for update` in %s ", sql));
if (canWrite) {
assertTrue(
sql.get(0).contains("for update;"),
String.format("Did not find `for update` in %s ", sql));
} else {
assertFalse(
sql.get(0).contains("for update;"), String.format("Found `for update` in %s ", sql));
}
}
@Test
public void testbatchGetForUpdate() throws JsonProcessingException {
@Test(dataProvider = "writabilityConfig")
public void testbatchGetForUpdate(boolean canWrite, String description)
throws JsonProcessingException {
testDao.setWritable(canWrite);
LoggedSql.start();
testDao.runInTransactionWithRetryUnlocked(
@ -142,8 +168,14 @@ public class EbeanAspectDaoTest {
.toList();
assertEquals(
sql.size(), 1, String.format("Found: %s", new ObjectMapper().writeValueAsString(sql)));
assertTrue(
sql.get(0).contains("FOR UPDATE;"), String.format("Did not find `for update` in %s ", sql));
if (canWrite) {
assertTrue(
sql.get(0).contains("FOR UPDATE;"),
String.format("Did not find `for update` in %s ", sql));
} else {
assertFalse(
sql.get(0).contains("FOR UPDATE;"), String.format("Found `for update` in %s ", sql));
}
}
@Test
@ -217,14 +249,6 @@ public class EbeanAspectDaoTest {
count3, 2, "Should return count of aspects matching both URN pattern and aspect name");
}
@DataProvider(name = "writabilityConfig")
public Object[][] writabilityConfigProvider() {
return new Object[][] {
{true, "Writable"}, // canWrite = true, description
{false, "ReadOnly"} // canWrite = false, description
};
}
@Test(dataProvider = "writabilityConfig")
public void testUpdateAspectWithWritability(boolean canWrite, String description) {
// Set writability