xiaoxue_admin
/
dbx
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(dameng): serialize metadata queries

This commit is contained in:
zipg 2026-07-13 23:57:45 +08:00 committed by GitHub
parent a9ca66a829
commit 4475ccb7fc
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 86 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
agents/drivers/dameng/src/main/java/com/dbx/agent/dameng/DamengAgent.java
View File

@ -598,7 +598,7 @@ public final class DamengAgent extends BaseDatabaseAgent {
default -> throw new IllegalArgumentException("Unsupported object type: " + objectType);
};
String source;
String sql = "SELECT DBMS_METADATA.GET_DDL(?, ?, ?) FROM DUAL";
String sql = "SELECT /*+ PARALLEL(1) */ DBMS_METADATA.GET_DDL(?, ?, ?) FROM DUAL";
try (PreparedStatement stmt = requireConnected().prepareStatement(sql)) {
stmt.setString(1, dbmsType);
stmt.setString(2, name);
@ -614,18 +614,22 @@ public final class DamengAgent extends BaseDatabaseAgent {
@Override
public String getTableDdl(String schema, String table) {
return unchecked(() -> {
String sql = "SELECT DBMS_METADATA.GET_DDL(?, ?, ?) FROM DUAL";
String sql = "SELECT /*+ PARALLEL(1) */ DBMS_METADATA.GET_DDL(?, ?, ?) FROM DUAL";
String ddl = null;
try (PreparedStatement stmt = requireConnected().prepareStatement(sql)) {
stmt.setString(1, "TABLE");
stmt.setString(2, table);
stmt.setString(3, schema);
try (ResultSet rs = stmt.executeQuery()) {
if (rs.next()) {
String ddl = appendTableAndColumnComments(coalesce(readTextColumn(rs, 1)), schema, table);
return appendIndependentIndexDdl(ddl, schema, table);
ddl = coalesce(readTextColumn(rs, 1));
}
}
}
if (ddl != null) {
ddl = appendTableAndColumnComments(ddl, schema, table);
return appendIndependentIndexDdl(ddl, schema, table);
}
throw new IllegalArgumentException("Table not found: " + schema + "." + table);
});
}
@ -635,7 +639,7 @@ public final class DamengAgent extends BaseDatabaseAgent {
return unchecked(() -> {
Set<String> pkColumns = new java.util.HashSet<>();
String pkSql = """
SELECT cols.COLUMN_NAME FROM ALL_CONS_COLUMNS cols
SELECT /*+ PARALLEL(1) */ cols.COLUMN_NAME FROM ALL_CONS_COLUMNS cols
JOIN ALL_CONSTRAINTS cons ON cols.CONSTRAINT_NAME = cons.CONSTRAINT_NAME AND cols.OWNER = cons.OWNER
WHERE cons.CONSTRAINT_TYPE = 'P' AND cons.OWNER = ? AND cons.TABLE_NAME = ?
""".stripIndent().trim();
@ -654,7 +658,7 @@ public final class DamengAgent extends BaseDatabaseAgent {
// DATA_DEFAULT is a LONG column — it must be selected first and read first
// in JDBC, otherwise the data is truncated.
String colSql = """
SELECT c.DATA_DEFAULT,
SELECT /*+ PARALLEL(1) */ c.DATA_DEFAULT,
c.COLUMN_NAME,
c.DATA_TYPE,
c.NULLABLE,
@ -709,7 +713,7 @@ public final class DamengAgent extends BaseDatabaseAgent {
private Set<String> identityColumns(String schema, String table) {
Set<String> result = new java.util.HashSet<>();
String sql = """
SELECT c.NAME
SELECT /*+ PARALLEL(1) */ c.NAME
FROM SYS.SYSCOLUMNS c
JOIN SYS.SYSOBJECTS t ON c.ID = t.ID
JOIN SYS.SYSOBJECTS s ON t.SCHID = s.ID
@ -737,7 +741,7 @@ public final class DamengAgent extends BaseDatabaseAgent {
return unchecked(() -> {
List<IndexInfo> result = new ArrayList<>();
String sql = """
SELECT i.INDEX_NAME,
SELECT /*+ PARALLEL(1) */ i.INDEX_NAME,
LISTAGG(ic.COLUMN_NAME, ',') WITHIN GROUP (ORDER BY ic.COLUMN_POSITION) AS COLUMNS,
i.UNIQUENESS,
CASE WHEN c.CONSTRAINT_TYPE = 'P' THEN 1 ELSE 0 END AS IS_PK,
@ -1031,18 +1035,18 @@ public final class DamengAgent extends BaseDatabaseAgent {
Map<String, String> comments = new HashMap<>();
queryColumnComments(
comments,
"SELECT COLUMN_NAME, COMMENTS FROM USER_COL_COMMENTS WHERE TABLE_NAME = ?",
"SELECT /*+ PARALLEL(1) */ COLUMN_NAME, COMMENTS FROM USER_COL_COMMENTS WHERE TABLE_NAME = ?",
table
);
queryColumnComments(
comments,
"SELECT COLNAME, COMMENT$ FROM SYS.SYSCOLUMNCOMMENTS WHERE SCHNAME = ? AND TVNAME = ?",
"SELECT /*+ PARALLEL(1) */ COLNAME, COMMENT$ FROM SYS.SYSCOLUMNCOMMENTS WHERE SCHNAME = ? AND TVNAME = ?",
schema,
table
);
queryColumnComments(
comments,
"SELECT COLUMN_NAME, COMMENTS FROM ALL_COL_COMMENTS WHERE UPPER(OWNER) = UPPER(?) AND UPPER(TABLE_NAME) = UPPER(?)",
"SELECT /*+ PARALLEL(1) */ COLUMN_NAME, COMMENTS FROM ALL_COL_COMMENTS WHERE UPPER(OWNER) = UPPER(?) AND UPPER(TABLE_NAME) = UPPER(?)",
schema,
table
);
@ -1111,7 +1115,7 @@ public final class DamengAgent extends BaseDatabaseAgent {
List<IndexInfo> result = new ArrayList<>();
// Primary-key and unique-constraint backing indexes are already represented in table DDL.
String sql = """
SELECT i.INDEX_NAME,
SELECT /*+ PARALLEL(1) */ i.INDEX_NAME,
LISTAGG(ic.COLUMN_NAME, ',') WITHIN GROUP (ORDER BY ic.COLUMN_POSITION) AS COLUMNS,
i.UNIQUENESS,
i.INDEX_TYPE
@ -1180,7 +1184,7 @@ public final class DamengAgent extends BaseDatabaseAgent {
private String tableComment(String schema, String table) throws Exception {
String sql = """
SELECT COMMENTS
SELECT /*+ PARALLEL(1) */ COMMENTS
FROM ALL_TAB_COMMENTS
WHERE OWNER = ? AND TABLE_NAME = ?
""".stripIndent().trim();

73
agents/drivers/dameng/src/test/java/com/dbx/agent/dameng/DamengAgentMetadataTest.java
View File

@ -34,6 +34,21 @@ class DamengAgentMetadataTest {
.findFirst()
.orElseThrow();
Assertions.assertTrue(columnsSql.contains("LEFT JOIN ALL_COL_COMMENTS"), columnsSql);
Assertions.assertTrue(columnsSql.startsWith("SELECT /*+ PARALLEL(1) */"), columnsSql);
}
@Test
void disablesParallelExecutionForIndexMetadataQuery() {
DamengAgent agent = new DamengAgent();
TestSupport.setPrivateConnection(agent, JdbcMetadataSqlFake.connection());
agent.listIndexes("APP", "USERS");
String indexesSql = JdbcMetadataSqlFake.statements.stream()
.filter(sql -> sql.contains("ALL_INDEXES"))
.findFirst()
.orElseThrow();
Assertions.assertTrue(indexesSql.startsWith("SELECT /*+ PARALLEL(1) */"), indexesSql);
}
@Test
@ -203,6 +218,43 @@ class DamengAgentMetadataTest {
Assertions.assertTrue(ddl.contains("COMMENT ON COLUMN \"APP\".\"USERS\".\"ID\" IS 'id comment';"), ddl);
}
@Test
void closesDbmsMetadataResultBeforeLoadingSupplementalDdlMetadata() {
DamengAgent agent = new DamengAgent();
TestSupport.setPrivateConnection(agent, metadataConnection());
Assertions.assertDoesNotThrow(() -> agent.getTableDdl("APP", "USERS"));
}
@Test
void disablesParallelExecutionForTableDdlMetadataQueries() {
DamengAgent agent = new DamengAgent();
List<String> sqls = new ArrayList<>();
TestSupport.setPrivateConnection(agent, metadataConnection(
"id comment",
null,
false,
List.of(),
sqls
));
agent.getTableDdl("APP", "USERS");
List<String> ddlMetadataSql = sqls.stream()
.filter(sql -> sql.contains("DBMS_METADATA.GET_DDL")
|| sql.contains("ALL_TAB_COLUMNS")
|| sql.contains("ALL_CONS_COLUMNS")
|| sql.contains("SYS.SYSCOLUMNS")
|| sql.contains("ALL_TAB_COMMENTS")
|| sql.contains("ALL_INDEXES"))
.toList();
Assertions.assertFalse(ddlMetadataSql.isEmpty());
Assertions.assertTrue(
ddlMetadataSql.stream().allMatch(sql -> sql.startsWith("SELECT /*+ PARALLEL(1) */")),
String.join("\n", ddlMetadataSql)
);
}
@Test
void readsFullTableDdlFromCharacterStreamWhenGetStringIsTruncated() {
DamengAgent agent = new DamengAgent();
@ -316,15 +368,19 @@ class DamengAgentMetadataTest {
List<String> sqls,
String dbmsMetadataDdl
) {
boolean[] dbmsMetadataResultOpen = {false};
return proxy(Connection.class, (method, args) -> {
String name = method.getName();
if ("prepareStatement".equals(name)) {
String sql = (String) args[0];
if (dbmsMetadataResultOpen[0]) {
throw new AssertionError("Supplemental metadata query started before DBMS_METADATA ResultSet closed: " + sql);
}
if (sqls != null) {
sqls.add(sql);
}
if (sql.contains("DBMS_METADATA.GET_DDL")) {
return dbmsMetadataStatement(dbmsMetadataDdl);
return dbmsMetadataStatement(dbmsMetadataDdl, dbmsMetadataResultOpen);
}
if (sql.contains("SYS.SYSOBJECTS") && sql.contains("TYPE$ = 'SCH'")) {
return metadataStatement(List.of(List.of("APP"), List.of("EMPTY_SCHEMA")));
@ -335,7 +391,7 @@ class DamengAgentMetadataTest {
if (sql.contains("SYS.SYSCOLUMNS")) {
return metadataStatement(List.of(List.of("ID")));
}
if (sql.startsWith("SELECT COMMENTS")) {
if (sql.contains("SELECT /*+ PARALLEL(1) */ COMMENTS")) {
return metadataStatement(List.of(List.of("用户示例表")));
}
if (sql.contains("USER_COL_COMMENTS")) {
@ -410,7 +466,7 @@ class DamengAgentMetadataTest {
return List.of(name, columns, uniqueness, indexType);
}
private static PreparedStatement dbmsMetadataStatement(String ddl) {
private static PreparedStatement dbmsMetadataStatement(String ddl, boolean[] resultOpen) {
List<String> params = new ArrayList<>();
return proxy(PreparedStatement.class, (method, args) -> {
String name = method.getName();
@ -419,7 +475,11 @@ class DamengAgentMetadataTest {
if ("INDEX".equals(objectType)) {
throw new AssertionError("Dameng table DDL should generate index DDL from metadata");
}
return metadataResultSet(List.of(List.of(new LongText(ddl, ddl.substring(0, Math.min(ddl.length(), 64))))));
resultOpen[0] = true;
return metadataResultSet(
List.of(List.of(new LongText(ddl, ddl.substring(0, Math.min(ddl.length(), 64))))),
() -> resultOpen[0] = false
);
}
if ("setString".equals(name)) {
int index = ((Integer) args[0]) - 1;
@ -486,6 +546,10 @@ class DamengAgentMetadataTest {
}
private static ResultSet metadataResultSet(List<List<Object>> rows) {
return metadataResultSet(rows, () -> {});
}
private static ResultSet metadataResultSet(List<List<Object>> rows, Runnable onClose) {
int[] index = {-1};
return proxy(ResultSet.class, (method, args) -> {
String name = method.getName();
@ -527,6 +591,7 @@ class DamengAgentMetadataTest {
};
}
if ("close".equals(name)) {
onClose.run();
return null;
}
return defaultValue(method.getReturnType());