github/dbx
4
0
Fork 0
mirror of https://github.com/t8y2/dbx.git synced 2026-10-19 16:16:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(dameng): use table-scoped statistics, index, and foreign key queries

Browse Source 
Closes #10388
This commit is contained in:
t8y2
2026-09-27 18:13:29 +00:00
parent bfeb2fdb69
commit 654443d6ab
3 changed files with 550 additions and 283 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agents/drivers/dameng/src/main/java/com/dbx/agent/dameng/DamengAgent.java
+228 -193
View File
@@ -2244,52 +2244,124 @@ public final class DamengAgent extends AbstractJdbcAgent {
table
);
}
return unchecked(() -> {
List<IndexInfo> result = new ArrayList<>();
// The LEFT JOIN also matches 'U': a UNIQUE constraint owns its backing index the same
// way a primary key does, and neither can be altered with index-level DDL (#7959).
// The constraint type therefore drives two flags — IS_PK (only 'P') and
// CONSTRAINT_BACKED ('P' or 'U'), the same split `independentIndexes` already uses to
// keep constraint-backed indexes out of the generated table DDL. A unique index
// created with CREATE UNIQUE INDEX has no ALL_CONSTRAINTS row, so it stays false and
// keeps the index-level DDL path.
String sql = """
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,
i.INDEX_TYPE,
CASE WHEN c.CONSTRAINT_TYPE IN ('P', 'U') THEN 1 ELSE 0 END AS CONSTRAINT_BACKED
FROM ALL_INDEXES i
JOIN ALL_IND_COLUMNS ic ON i.INDEX_NAME = ic.INDEX_NAME AND i.OWNER = ic.INDEX_OWNER AND i.TABLE_OWNER = ic.TABLE_OWNER
LEFT JOIN ALL_CONSTRAINTS c ON i.INDEX_NAME = c.INDEX_NAME AND i.TABLE_OWNER = c.OWNER
AND c.CONSTRAINT_TYPE IN ('P', 'U')
WHERE i.TABLE_OWNER = ? AND i.TABLE_NAME = ?
GROUP BY i.INDEX_NAME, i.UNIQUENESS, c.CONSTRAINT_TYPE, i.INDEX_TYPE
ORDER BY i.INDEX_NAME
""".stripIndent().trim();
try (PreparedStatement stmt = requireConnected().prepareStatement(sql)) {
stmt.setString(1, schema);
stmt.setString(2, table);
try (ResultSet rs = stmt.executeQuery()) {
while (rs.next()) {
IndexInfo index = new IndexInfo(
rs.getString(1),
splitNonEmpty(coalesce(rs.getString(2)), ","),
"UNIQUE".equals(rs.getString(3)),
"1".equals(rs.getString(4)),
null,
rs.getString(5),
null,
null
);
index.setConstraint_backed("1".equals(rs.getString(6)));
result.add(index);
}
return unchecked(() -> tableIndexes(schema, table));
}
private List<IndexInfo> tableIndexes(String schema, String table) throws Exception {
try {
return indexesFromSystemCatalog(schema, table);
} catch (Exception systemCatalogError) {
if (isDamengConnectionError(systemCatalogError)) {
throw systemCatalogError;
}
try {
return indexesFromDictionaryViews(schema, table);
} catch (Exception dictionaryViewError) {
dictionaryViewError.addSuppressed(systemCatalogError);
throw dictionaryViewError;
}
}
}
private List<IndexInfo> indexesFromSystemCatalog(String schema, String table) throws Exception {
List<IndexInfo> result = new ArrayList<>();
// ALL_INDEXES applies visibility checks across the complete catalog. Resolve the one table
// first, then inspect only its index children; the official catalog exposes key order via
// SF_GET_INDEX_KEY_SEQ and constraint ownership through SYSCONS.
String sql = """
SELECT /*+ PARALLEL(1) */ index_object.NAME,
LISTAGG(column_object.NAME, ',') WITHIN GROUP (
ORDER BY SF_GET_INDEX_KEY_SEQ(index_metadata.KEYNUM, index_metadata.KEYINFO, column_object.COLID)
) AS COLUMNS,
index_metadata.ISUNIQUE,
MAX(CASE WHEN constraint_metadata.TYPE$ = 'P' THEN 1 ELSE 0 END) AS IS_PK,
index_metadata.TYPE$,
MAX(CASE WHEN constraint_metadata.TYPE$ IN ('P', 'U') THEN 1 ELSE 0 END) AS CONSTRAINT_BACKED,
index_metadata.FLAG
FROM SYS.SYSOBJECTS schema_object
JOIN SYS.SYSOBJECTS table_object ON table_object.SCHID = schema_object.ID
AND table_object.TYPE$ = 'SCHOBJ' AND table_object.SUBTYPE$ = 'UTAB'
JOIN SYS.SYSOBJECTS index_object ON index_object.PID = table_object.ID
AND index_object.SUBTYPE$ = 'INDEX'
JOIN SYS.SYSINDEXES index_metadata ON index_metadata.ID = index_object.ID
JOIN SYS.SYSCOLUMNS column_object ON column_object.ID = table_object.ID
AND SF_COL_IS_IDX_KEY(
index_metadata.KEYNUM,
index_metadata.KEYINFO,
column_object.COLID
) = 1
LEFT JOIN SYS.SYSCONS constraint_metadata ON constraint_metadata.TABLEID = table_object.ID
AND constraint_metadata.INDEXID = index_metadata.ID
AND constraint_metadata.TYPE$ IN ('P', 'U')
WHERE schema_object.TYPE$ = 'SCH' AND schema_object.NAME = ? AND table_object.NAME = ?
GROUP BY index_object.NAME, index_metadata.ISUNIQUE, index_metadata.TYPE$, index_metadata.FLAG
ORDER BY index_object.NAME
""".stripIndent().trim();
try (PreparedStatement stmt = requireConnected().prepareStatement(sql)) {
stmt.setString(1, schema);
stmt.setString(2, table);
try (ResultSet rs = stmt.executeQuery()) {
while (rs.next()) {
boolean isPrimary = "1".equals(rs.getString(4));
boolean constraintBacked = "1".equals(rs.getString(6));
IndexInfo index = new IndexInfo(
rs.getString(1),
splitNonEmpty(coalesce(rs.getString(2)), ","),
"Y".equalsIgnoreCase(rs.getString(3)),
isPrimary,
null,
damengSystemIndexType(rs.getString(5), rs.getInt(7), constraintBacked),
null,
null
);
index.setConstraint_backed(constraintBacked);
result.add(index);
}
}
return result;
});
}
return result;
}
private List<IndexInfo> indexesFromDictionaryViews(String schema, String table) throws Exception {
List<IndexInfo> result = new ArrayList<>();
// The LEFT JOIN also matches 'U': a UNIQUE constraint owns its backing index the same
// way a primary key does, and neither can be altered with index-level DDL (#7959).
String sql = """
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,
i.INDEX_TYPE,
CASE WHEN c.CONSTRAINT_TYPE IN ('P', 'U') THEN 1 ELSE 0 END AS CONSTRAINT_BACKED
FROM ALL_INDEXES i
JOIN ALL_IND_COLUMNS ic ON i.INDEX_NAME = ic.INDEX_NAME AND i.OWNER = ic.INDEX_OWNER AND i.TABLE_OWNER = ic.TABLE_OWNER
LEFT JOIN ALL_CONSTRAINTS c ON i.INDEX_NAME = c.INDEX_NAME AND i.TABLE_OWNER = c.OWNER
AND c.CONSTRAINT_TYPE IN ('P', 'U')
WHERE i.TABLE_OWNER = ? AND i.TABLE_NAME = ?
GROUP BY i.INDEX_NAME, i.UNIQUENESS, c.CONSTRAINT_TYPE, i.INDEX_TYPE
ORDER BY i.INDEX_NAME
""".stripIndent().trim();
try (PreparedStatement stmt = requireConnected().prepareStatement(sql)) {
stmt.setString(1, schema);
stmt.setString(2, table);
try (ResultSet rs = stmt.executeQuery()) {
while (rs.next()) {
IndexInfo index = new IndexInfo(
rs.getString(1),
splitNonEmpty(coalesce(rs.getString(2)), ","),
"UNIQUE".equals(rs.getString(3)),
"1".equals(rs.getString(4)),
null,
rs.getString(5),
null,
null
);
index.setConstraint_backed("1".equals(rs.getString(6)));
result.add(index);
}
}
}
return result;
}
@Override
@@ -2298,33 +2370,108 @@ public final class DamengAgent extends AbstractJdbcAgent {
if (legacyJdbcMetadata) {
return StandardJdbcMetadata.INSTANCE.listForeignKeys(requireConnected(), schema, table);
}
return unchecked(() -> {
List<ForeignKeyInfo> result = new ArrayList<>();
String sql = """
SELECT c.CONSTRAINT_NAME, cc.COLUMN_NAME, rc.TABLE_NAME, rcc.COLUMN_NAME
FROM ALL_CONSTRAINTS c
JOIN ALL_CONS_COLUMNS cc ON c.CONSTRAINT_NAME = cc.CONSTRAINT_NAME AND c.OWNER = cc.OWNER
JOIN ALL_CONSTRAINTS rc ON c.R_CONSTRAINT_NAME = rc.CONSTRAINT_NAME AND c.R_OWNER = rc.OWNER
JOIN ALL_CONS_COLUMNS rcc ON rc.CONSTRAINT_NAME = rcc.CONSTRAINT_NAME AND rc.OWNER = rcc.OWNER
WHERE c.CONSTRAINT_TYPE = 'R' AND c.OWNER = ? AND c.TABLE_NAME = ?
ORDER BY c.CONSTRAINT_NAME
""".stripIndent().trim();
try (PreparedStatement stmt = requireConnected().prepareStatement(sql)) {
stmt.setString(1, schema);
stmt.setString(2, table);
try (ResultSet rs = stmt.executeQuery()) {
while (rs.next()) {
result.add(new ForeignKeyInfo(
rs.getString(1),
rs.getString(2),
rs.getString(3),
rs.getString(4)
));
}
return unchecked(() -> foreignKeys(schema, table));
}
private List<ForeignKeyInfo> foreignKeys(String schema, String table) throws Exception {
try {
return foreignKeysFromSystemCatalog(schema, table);
} catch (Exception systemCatalogError) {
if (isDamengConnectionError(systemCatalogError)) {
throw systemCatalogError;
}
try {
return foreignKeysFromDictionaryViews(schema, table);
} catch (Exception dictionaryViewError) {
dictionaryViewError.addSuppressed(systemCatalogError);
throw dictionaryViewError;
}
}
}
private List<ForeignKeyInfo> foreignKeysFromSystemCatalog(String schema, String table) throws Exception {
List<ForeignKeyInfo> result = new ArrayList<>();
String sql = """
SELECT /*+ MAX_OPT_N_TABLES(5) PARALLEL(1) */ constraint_object.NAME,
local_column.NAME,
referenced_table.NAME,
referenced_column.NAME
FROM SYS.SYSOBJECTS schema_object
JOIN SYS.SYSOBJECTS table_object ON table_object.SCHID = schema_object.ID
AND table_object.TYPE$ = 'SCHOBJ' AND table_object.SUBTYPE$ = 'UTAB'
JOIN SYS.SYSCONS foreign_key ON foreign_key.TABLEID = table_object.ID
AND foreign_key.TYPE$ = 'F'
JOIN SYS.SYSOBJECTS constraint_object ON constraint_object.ID = foreign_key.ID
AND constraint_object.PID = table_object.ID
AND constraint_object.SUBTYPE$ = 'CONS'
JOIN SYS.SYSINDEXES local_index ON local_index.ID = foreign_key.INDEXID
JOIN SYS.SYSCOLUMNS local_column ON local_column.ID = table_object.ID
AND SF_COL_IS_IDX_KEY(local_index.KEYNUM, local_index.KEYINFO, local_column.COLID) = 1
JOIN SYS.SYSOBJECTS referenced_index_object ON referenced_index_object.ID = foreign_key.FINDEXID
AND referenced_index_object.SUBTYPE$ = 'INDEX'
JOIN SYS.SYSINDEXES referenced_index ON referenced_index.ID = referenced_index_object.ID
JOIN SYS.SYSOBJECTS referenced_table ON referenced_table.ID = referenced_index_object.PID
AND referenced_table.TYPE$ = 'SCHOBJ' AND referenced_table.SUBTYPE$ = 'UTAB'
JOIN SYS.SYSCOLUMNS referenced_column ON referenced_column.ID = referenced_table.ID
AND SF_COL_IS_IDX_KEY(
referenced_index.KEYNUM,
referenced_index.KEYINFO,
referenced_column.COLID
) = 1
AND SF_GET_INDEX_KEY_SEQ(local_index.KEYNUM, local_index.KEYINFO, local_column.COLID)
= SF_GET_INDEX_KEY_SEQ(
referenced_index.KEYNUM,
referenced_index.KEYINFO,
referenced_column.COLID
)
WHERE schema_object.TYPE$ = 'SCH' AND schema_object.NAME = ? AND table_object.NAME = ?
ORDER BY constraint_object.NAME,
SF_GET_INDEX_KEY_SEQ(local_index.KEYNUM, local_index.KEYINFO, local_column.COLID)
""".stripIndent().trim();
try (PreparedStatement stmt = requireConnected().prepareStatement(sql)) {
stmt.setString(1, schema);
stmt.setString(2, table);
try (ResultSet rs = stmt.executeQuery()) {
while (rs.next()) {
result.add(new ForeignKeyInfo(
rs.getString(1),
rs.getString(2),
rs.getString(3),
rs.getString(4)
));
}
}
return result;
});
}
return result;
}
private List<ForeignKeyInfo> foreignKeysFromDictionaryViews(String schema, String table) throws Exception {
List<ForeignKeyInfo> result = new ArrayList<>();
String sql = """
SELECT /*+ PARALLEL(1) */ c.CONSTRAINT_NAME, cc.COLUMN_NAME, rc.TABLE_NAME, rcc.COLUMN_NAME
FROM ALL_CONSTRAINTS c
JOIN ALL_CONS_COLUMNS cc ON c.CONSTRAINT_NAME = cc.CONSTRAINT_NAME AND c.OWNER = cc.OWNER
JOIN ALL_CONSTRAINTS rc ON c.R_CONSTRAINT_NAME = rc.CONSTRAINT_NAME AND c.R_OWNER = rc.OWNER
JOIN ALL_CONS_COLUMNS rcc ON rc.CONSTRAINT_NAME = rcc.CONSTRAINT_NAME AND rc.OWNER = rcc.OWNER
AND cc.POSITION = rcc.POSITION
WHERE c.CONSTRAINT_TYPE = 'R' AND c.OWNER = ? AND c.TABLE_NAME = ?
ORDER BY c.CONSTRAINT_NAME, cc.POSITION
""".stripIndent().trim();
try (PreparedStatement stmt = requireConnected().prepareStatement(sql)) {
stmt.setString(1, schema);
stmt.setString(2, table);
try (ResultSet rs = stmt.executeQuery()) {
while (rs.next()) {
result.add(new ForeignKeyInfo(
rs.getString(1),
rs.getString(2),
rs.getString(3),
rs.getString(4)
));
}
}
}
return result;
}
@Override
@@ -2895,134 +3042,22 @@ public final class DamengAgent extends AbstractJdbcAgent {
}
private List<IndexInfo> independentIndexes(String schema, String table) throws Exception {
try {
return independentIndexesFromSystemCatalog(schema, table);
} catch (Exception systemCatalogError) {
if (isDamengConnectionError(systemCatalogError)) {
throw systemCatalogError;
}
try {
return independentIndexesFromDictionaryViews(schema, table);
} catch (Exception dictionaryViewError) {
dictionaryViewError.addSuppressed(systemCatalogError);
throw dictionaryViewError;
}
}
return tableIndexes(schema, table).stream()
.filter(index -> !index.getConstraint_backed())
.toList();
}
private List<IndexInfo> independentIndexesFromSystemCatalog(String schema, String table) throws Exception {
List<IndexInfo> result = new ArrayList<>();
// DM's ALL_INDEXES view performs privilege filtering across the complete catalog and can
// take tens of seconds in installations with very large index dictionaries (#10075).
// The DM JDBC driver and DBeaver instead resolve the requested table id first and read its
// SYS index rows directly. Keep the ALL_* implementation below as a compatibility and
// permission fallback for deployments that do not expose these catalog tables.
String sql = """
SELECT /*+ PARALLEL(1) */ index_object.NAME,
LISTAGG(column_object.NAME, ',') WITHIN GROUP (
ORDER BY SF_GET_INDEX_KEY_SEQ(index_metadata.KEYNUM, index_metadata.KEYINFO, column_object.COLID)
) AS COLUMNS,
index_metadata.ISUNIQUE,
index_metadata.TYPE$,
index_metadata.FLAG
FROM SYS.SYSOBJECTS schema_object
JOIN SYS.SYSOBJECTS table_object ON table_object.SCHID = schema_object.ID
AND table_object.TYPE$ = 'SCHOBJ' AND table_object.SUBTYPE$ = 'UTAB'
JOIN SYS.SYSOBJECTS index_object ON index_object.PID = table_object.ID
AND index_object.SUBTYPE$ = 'INDEX'
JOIN SYS.SYSINDEXES index_metadata ON index_metadata.ID = index_object.ID
JOIN SYS.SYSCOLUMNS column_object ON column_object.ID = table_object.ID
AND SF_COL_IS_IDX_KEY(
index_metadata.KEYNUM,
index_metadata.KEYINFO,
column_object.COLID
) = 1
WHERE schema_object.TYPE$ = 'SCH' AND schema_object.NAME = ? AND table_object.NAME = ?
AND NOT EXISTS (
SELECT 1
FROM SYS.SYSCONS constraint_metadata
WHERE constraint_metadata.TABLEID = table_object.ID
AND constraint_metadata.INDEXID = index_metadata.ID
AND constraint_metadata.TYPE$ IN ('P', 'U')
)
GROUP BY index_object.NAME, index_metadata.ISUNIQUE, index_metadata.TYPE$, index_metadata.FLAG
ORDER BY index_object.NAME
""".stripIndent().trim();
try (PreparedStatement stmt = requireConnected().prepareStatement(sql)) {
stmt.setString(1, schema);
stmt.setString(2, table);
try (ResultSet rs = stmt.executeQuery()) {
while (rs.next()) {
String indexName = rs.getString(1);
if (notBlank(indexName)) {
result.add(new IndexInfo(
indexName,
splitNonEmpty(coalesce(rs.getString(2)), ","),
"Y".equalsIgnoreCase(rs.getString(3)),
false,
null,
damengSystemIndexType(rs.getString(4), rs.getInt(5)),
null,
null
));
}
}
}
}
return result;
}
private List<IndexInfo> independentIndexesFromDictionaryViews(String schema, String table) throws Exception {
List<IndexInfo> result = new ArrayList<>();
// Primary-key and unique-constraint backing indexes are already represented in table DDL.
String sql = """
SELECT /*+ PARALLEL(1) */ i.INDEX_NAME,
LISTAGG(ic.COLUMN_NAME, ',') WITHIN GROUP (ORDER BY ic.COLUMN_POSITION) AS COLUMNS,
i.UNIQUENESS,
i.INDEX_TYPE
FROM ALL_INDEXES i
JOIN ALL_IND_COLUMNS ic ON i.INDEX_NAME = ic.INDEX_NAME AND i.OWNER = ic.INDEX_OWNER AND i.TABLE_OWNER = ic.TABLE_OWNER
WHERE i.TABLE_OWNER = ? AND i.TABLE_NAME = ?
AND NOT EXISTS (
SELECT 1
FROM ALL_CONSTRAINTS c
WHERE c.OWNER = i.TABLE_OWNER
AND c.TABLE_NAME = i.TABLE_NAME
AND c.INDEX_NAME = i.INDEX_NAME
AND c.CONSTRAINT_TYPE IN ('P', 'U')
)
GROUP BY i.INDEX_NAME, i.UNIQUENESS, i.INDEX_TYPE
ORDER BY i.INDEX_NAME
""".stripIndent().trim();
try (PreparedStatement stmt = requireConnected().prepareStatement(sql)) {
stmt.setString(1, schema);
stmt.setString(2, table);
try (ResultSet rs = stmt.executeQuery()) {
while (rs.next()) {
String indexName = rs.getString(1);
if (notBlank(indexName)) {
result.add(new IndexInfo(
indexName,
splitNonEmpty(coalesce(rs.getString(2)), ","),
"UNIQUE".equals(rs.getString(3)),
false,
null,
rs.getString(4),
null,
null
));
}
}
}
}
return result;
}
private static String damengSystemIndexType(String catalogType, int flags) {
if ((flags & 3) != 0) {
private static String damengSystemIndexType(String catalogType, int flags, boolean constraintBacked) {
if ((flags & 1) != 0 || ((flags & 2) != 0 && !constraintBacked)) {
return "INTERNAL";
}
return "ST".equalsIgnoreCase(coalesce(catalogType).trim()) ? "SPATIAL" : catalogType;
String normalized = coalesce(catalogType).trim().toUpperCase(Locale.ROOT);
return switch (normalized) {
case "BT" -> "NORMAL";
case "BM" -> "BITMAP";
case "ST" -> "SPATIAL";
default -> catalogType;
};
}
static String indexDdl(String schema, String table, IndexInfo index) {
agents/drivers/dameng/src/test/java/com/dbx/agent/dameng/DamengAgentMetadataTest.java
+196 -12
View File
@@ -2,6 +2,7 @@ package com.dbx.agent.dameng;
import com.dbx.agent.ColumnInfo;
import com.dbx.agent.DatabaseInfo;
import com.dbx.agent.ForeignKeyInfo;
import com.dbx.agent.IndexInfo;
import com.dbx.agent.MetadataListConstraints;
import com.dbx.agent.ObjectInfo;
@@ -188,17 +189,25 @@ class DamengAgentMetadataTest {
}
@Test
void disablesParallelExecutionForIndexMetadataQuery() {
void usesBoundedSystemCatalogForIndexMetadata() {
DamengAgent agent = new DamengAgent();
TestSupport.setPrivateConnection(agent, JdbcMetadataSqlFake.connection());
agent.listIndexes("APP", "USERS");
String indexesSql = JdbcMetadataSqlFake.statements.stream()
.filter(sql -> sql.contains("ALL_INDEXES"))
.filter(sql -> sql.contains("SYS.SYSINDEXES"))
.findFirst()
.orElseThrow();
Assertions.assertTrue(indexesSql.startsWith("SELECT /*+ PARALLEL(1) */"), indexesSql);
Assertions.assertTrue(indexesSql.contains("schema_object.NAME = ?"), indexesSql);
Assertions.assertTrue(indexesSql.contains("table_object.NAME = ?"), indexesSql);
Assertions.assertTrue(indexesSql.contains("SF_COL_IS_IDX_KEY"), indexesSql);
Assertions.assertTrue(indexesSql.contains("SF_GET_INDEX_KEY_SEQ"), indexesSql);
Assertions.assertTrue(
JdbcMetadataSqlFake.statements.stream().noneMatch(sql -> sql.contains("ALL_INDEXES")),
String.join("\n", JdbcMetadataSqlFake.statements)
);
}
@Test
@@ -209,12 +218,15 @@ class DamengAgentMetadataTest {
agent.listIndexes("APP", "USERS");
String indexesSql = JdbcMetadataSqlFake.statements.stream()
.filter(sql -> sql.contains("ALL_INDEXES") && sql.contains("ALL_CONSTRAINTS"))
.filter(sql -> sql.contains("SYS.SYSINDEXES") && sql.contains("SYS.SYSCONS"))
.findFirst()
.orElseThrow();
Assertions.assertTrue(indexesSql.contains("AND c.CONSTRAINT_TYPE IN ('P', 'U')"), indexesSql);
Assertions.assertTrue(indexesSql.contains("constraint_metadata.TYPE$ IN ('P', 'U')"), indexesSql);
Assertions.assertTrue(indexesSql.contains("AS CONSTRAINT_BACKED"), indexesSql);
Assertions.assertTrue(indexesSql.contains("CASE WHEN c.CONSTRAINT_TYPE = 'P' THEN 1 ELSE 0 END AS IS_PK"), indexesSql);
Assertions.assertTrue(
indexesSql.contains("MAX(CASE WHEN constraint_metadata.TYPE$ = 'P' THEN 1 ELSE 0 END) AS IS_PK"),
indexesSql
);
}
@Test
@@ -222,10 +234,10 @@ class DamengAgentMetadataTest {
DamengAgent agent = new DamengAgent();
TestSupport.setPrivateConnection(agent, indexMetadataConnection(List.of(
// UNIQUE constraint (虚索引) / standalone CREATE UNIQUE INDEX (实索引) / plain index.
Arrays.asList("UX_USERS_EMAIL", "EMAIL", "UNIQUE", "0", "NORMAL", "1"),
Arrays.asList("UX_USERS_CODE", "CODE", "UNIQUE", "0", "NORMAL", "0"),
Arrays.asList("IDX_USERS_NAME", "NAME", "NONUNIQUE", "0", "NORMAL", "0"),
Arrays.asList("PK_USERS", "ID", "UNIQUE", "1", "NORMAL", "1")
Arrays.asList("UX_USERS_EMAIL", "EMAIL", "Y", "0", "BT", "1", 0),
Arrays.asList("UX_USERS_CODE", "CODE", "Y", "0", "BT", "0", 0),
Arrays.asList("IDX_USERS_NAME", "NAME", "N", "0", "BT", "0", 0),
Arrays.asList("PK_USERS", "ID", "Y", "1", "BT", "1", 4)
)));
List<IndexInfo> indexes = agent.listIndexes("APP", "USERS");
@@ -246,6 +258,142 @@ class DamengAgentMetadataTest {
List.of(true, true, false, true),
indexes.stream().map(IndexInfo::getIs_unique).toList()
);
Assertions.assertEquals(
List.of("NORMAL", "NORMAL", "NORMAL", "NORMAL"),
indexes.stream().map(IndexInfo::getIndex_type).toList()
);
}
@Test
void fallsBackToDictionaryViewsForIndexMetadataWithoutSystemCatalogAccess() {
DamengAgent agent = new DamengAgent();
List<String> sqls = new ArrayList<>();
TestSupport.setPrivateConnection(agent, catalogFallbackConnection(
sqls,
"SYS.SYSINDEXES index_metadata",
List.of(),
new SQLException("no SYS index catalog privilege"),
"FROM ALL_INDEXES i",
List.of(Arrays.asList("IDX_USERS_NAME", "NAME", "NONUNIQUE", "0", "NORMAL", "0"))
));
List<IndexInfo> indexes = agent.listIndexes("APP", "USERS");
Assertions.assertEquals(List.of("IDX_USERS_NAME"), indexes.stream().map(IndexInfo::getName).toList());
Assertions.assertTrue(indexOfSql(sqls, "SYS.SYSINDEXES") >= 0, String.join("\n", sqls));
Assertions.assertTrue(
indexOfSql(sqls, "ALL_INDEXES") > indexOfSql(sqls, "SYS.SYSINDEXES"),
String.join("\n", sqls)
);
}
@Test
void acceptsEmptySystemIndexMetadataWithoutRunningExpensiveFallback() {
DamengAgent agent = new DamengAgent();
List<String> sqls = new ArrayList<>();
TestSupport.setPrivateConnection(agent, catalogFallbackConnection(
sqls,
"SYS.SYSINDEXES index_metadata",
List.of(),
null,
"FROM ALL_INDEXES i",
List.of(Arrays.asList("SHOULD_NOT_LOAD", "ID", "NONUNIQUE", "0", "NORMAL", "0"))
));
Assertions.assertTrue(agent.listIndexes("APP", "USERS").isEmpty());
Assertions.assertTrue(sqls.stream().noneMatch(sql -> sql.contains("ALL_INDEXES")), String.join("\n", sqls));
}
@Test
void usesBoundedSystemCatalogForForeignKeyMetadata() {
DamengAgent agent = new DamengAgent();
List<String> sqls = new ArrayList<>();
TestSupport.setPrivateConnection(agent, catalogFallbackConnection(
sqls,
"SYS.SYSCONS foreign_key",
List.of(
List.of("FK_ORDER_CUSTOMER", "TENANT_ID", "CUSTOMERS", "TENANT_ID"),
List.of("FK_ORDER_CUSTOMER", "CUSTOMER_ID", "CUSTOMERS", "ID")
),
null,
"FROM ALL_CONSTRAINTS c",
List.of()
));
List<ForeignKeyInfo> foreignKeys = agent.listForeignKeys("APP", "ORDERS");
Assertions.assertEquals(List.of("TENANT_ID", "CUSTOMER_ID"), foreignKeys.stream().map(ForeignKeyInfo::getColumn).toList());
Assertions.assertEquals(List.of("TENANT_ID", "ID"), foreignKeys.stream().map(ForeignKeyInfo::getRef_column).toList());
String sql = sqls.get(0);
Assertions.assertTrue(sql.contains("foreign_key.TABLEID = table_object.ID"), sql);
Assertions.assertTrue(sql.contains("foreign_key.FINDEXID"), sql);
Assertions.assertTrue(sql.contains("foreign_key.TYPE$ = 'F'"), sql);
Assertions.assertTrue(sql.contains("SF_GET_INDEX_KEY_SEQ"), sql);
Assertions.assertTrue(sql.contains("schema_object.NAME = ?"), sql);
Assertions.assertTrue(sql.contains("table_object.NAME = ?"), sql);
Assertions.assertTrue(sqls.stream().noneMatch(query -> query.contains("ALL_CONSTRAINTS")), String.join("\n", sqls));
}
@Test
void fallsBackToPositionMatchedDictionaryForeignKeysWithoutSystemCatalogAccess() {
DamengAgent agent = new DamengAgent();
List<String> sqls = new ArrayList<>();
TestSupport.setPrivateConnection(agent, catalogFallbackConnection(
sqls,
"SYS.SYSCONS foreign_key",
List.of(),
new SQLException("no SYS constraint catalog privilege"),
"FROM ALL_CONSTRAINTS c",
List.of(List.of("FK_ORDER_CUSTOMER", "CUSTOMER_ID", "CUSTOMERS", "ID"))
));
List<ForeignKeyInfo> foreignKeys = agent.listForeignKeys("APP", "ORDERS");
Assertions.assertEquals(1, foreignKeys.size());
Assertions.assertEquals("FK_ORDER_CUSTOMER", foreignKeys.get(0).getName());
int systemQuery = indexOfSql(sqls, "SYS.SYSCONS foreign_key");
int fallbackQuery = indexOfSql(sqls, "ALL_CONSTRAINTS");
Assertions.assertTrue(fallbackQuery > systemQuery, String.join("\n", sqls));
Assertions.assertTrue(sqls.get(fallbackQuery).contains("cc.POSITION = rcc.POSITION"), sqls.get(fallbackQuery));
}
@Test
void acceptsEmptySystemForeignKeysWithoutRunningExpensiveFallback() {
DamengAgent agent = new DamengAgent();
List<String> sqls = new ArrayList<>();
TestSupport.setPrivateConnection(agent, catalogFallbackConnection(
sqls,
"SYS.SYSCONS foreign_key",
List.of(),
null,
"FROM ALL_CONSTRAINTS c",
List.of(List.of("SHOULD_NOT_LOAD", "ID", "PARENT", "ID"))
));
Assertions.assertTrue(agent.listForeignKeys("APP", "ORDERS").isEmpty());
Assertions.assertTrue(sqls.stream().noneMatch(sql -> sql.contains("ALL_CONSTRAINTS")), String.join("\n", sqls));
}
@Test
void doesNotRetryForeignKeyCatalogAfterConnectionFailure() {
DamengAgent agent = new DamengAgent();
List<String> sqls = new ArrayList<>();
TestSupport.setPrivateConnection(agent, catalogFallbackConnection(
sqls,
"SYS.SYSCONS foreign_key",
List.of(),
new SQLNonTransientConnectionException("connection lost"),
"FROM ALL_CONSTRAINTS c",
List.of()
));
RuntimeException error = Assertions.assertThrows(
RuntimeException.class,
() -> agent.listForeignKeys("APP", "ORDERS")
);
Assertions.assertEquals("connection lost", error.getCause().getMessage());
Assertions.assertTrue(sqls.stream().noneMatch(sql -> sql.contains("ALL_CONSTRAINTS")), String.join("\n", sqls));
}
@Test
@@ -2178,7 +2326,8 @@ class DamengAgentMetadataTest {
List.of(
indexRow("IDX_USERS_NAME", "NAME", "N", "BT", 0),
indexRow("UX_USERS_EMAIL", "EMAIL", "Y", "BT", 0),
indexRow("IDX_USERS_GEO", "GEO", "N", "ST", 0)
indexRow("IDX_USERS_GEO", "GEO", "N", "ST", 0),
Arrays.asList("PK_USERS", "ID", "Y", "1", "BT", "1", 4)
),
sqls
);
@@ -2480,7 +2629,7 @@ class DamengAgentMetadataTest {
}
private static List<Object> indexRow(String name, String columns, String uniqueness, String indexType, int flags) {
return List.of(name, columns, uniqueness, indexType, flags);
return List.of(name, columns, uniqueness, "0", indexType, "0", flags);
}
private static List<List<Object>> dictionaryIndexRows(List<List<Object>> systemRows) {
@@ -2488,10 +2637,45 @@ class DamengAgentMetadataTest {
row.get(0),
row.get(1),
"Y".equals(row.get(2)) ? "UNIQUE" : "NONUNIQUE",
"ST".equals(row.get(3)) ? "SPATIAL" : "NORMAL"
row.get(3),
"ST".equals(row.get(4)) ? "SPATIAL" : "NORMAL",
row.get(5)
)).toList();
}
private static Connection catalogFallbackConnection(
List<String> sqls,
String systemQueryMarker,
List<List<Object>> systemRows,
SQLException systemError,
String fallbackQueryMarker,
List<List<Object>> fallbackRows
) {
return proxy(Connection.class, (method, args) -> {
String name = method.getName();
if ("prepareStatement".equals(name)) {
String sql = (String) args[0];
sqls.add(sql);
if (sql.contains(systemQueryMarker)) {
return systemError == null
? metadataStatement(systemRows)
: failingMetadataStatement(systemError);
}
if (sql.contains(fallbackQueryMarker)) {
return metadataStatement(fallbackRows);
}
throw new AssertionError("Unexpected metadata query: " + sql);
}
if ("close".equals(name)) {
return null;
}
if ("isClosed".equals(name)) {
return false;
}
return defaultValue(method.getReturnType());
});
}
private static int indexOfSql(List<String> sqls, String fragment) {
for (int i = 0; i < sqls.size(); i++) {
if (sqls.get(i).contains(fragment)) {
crates/dbx-core/src/schema/mod.rs
+126 -78
View File
@@ -1760,54 +1760,16 @@ fn oracle_object_statistics_rows_only_sql(schema: &str) -> String {
)
}
fn dameng_object_statistics_dba_segments_sql(schema: &str) -> String {
fn dameng_object_statistics_sql(schema: &str) -> String {
// DM exposes a table-scoped size estimator. It avoids joining the global
// segment and index dictionaries, which can exhaust memory on large catalogs.
format!(
"SELECT t.TABLE_NAME, t.OWNER, t.NUM_ROWS, NVL(s.BYTES, 0) AS TOTAL_BYTES \
"SELECT t.TABLE_NAME, t.OWNER, t.NUM_ROWS, \
NVL(TABLE_USED_PAGES(t.OWNER, t.TABLE_NAME) * PAGE, 0) AS TOTAL_BYTES \
FROM ALL_TABLES t \
LEFT JOIN ( \
SELECT owner, table_name, SUM(bytes) AS BYTES \
FROM ( \
SELECT s.OWNER, s.SEGMENT_NAME AS TABLE_NAME, s.BYTES \
FROM DBA_SEGMENTS s \
WHERE s.OWNER = {} AND s.SEGMENT_TYPE IN ('TABLE','TABLE PARTITION','TABLE SUBPARTITION') \
UNION ALL \
SELECT i.TABLE_OWNER AS OWNER, i.TABLE_NAME, s.BYTES \
FROM ALL_INDEXES i \
JOIN DBA_SEGMENTS s ON s.OWNER = i.OWNER AND s.SEGMENT_NAME = i.INDEX_NAME \
WHERE i.TABLE_OWNER = {} AND s.SEGMENT_TYPE IN ('INDEX','INDEX PARTITION','INDEX SUBPARTITION') \
) \
GROUP BY owner, table_name \
) s ON s.OWNER = t.OWNER AND s.TABLE_NAME = t.TABLE_NAME \
WHERE t.OWNER = {} AND (t.NESTED IS NULL OR t.NESTED = 'NO') \
ORDER BY t.TABLE_NAME",
oracle_owner_filter(schema),
oracle_owner_filter(schema),
oracle_owner_filter(schema),
)
}
fn dameng_object_statistics_user_segments_sql(schema: &str) -> String {
format!(
"SELECT t.TABLE_NAME, t.OWNER, t.NUM_ROWS, NVL(s.BYTES, 0) AS TOTAL_BYTES \
FROM ALL_TABLES t \
LEFT JOIN ( \
SELECT table_name, SUM(bytes) AS BYTES \
FROM ( \
SELECT s.SEGMENT_NAME AS TABLE_NAME, s.BYTES \
FROM USER_SEGMENTS s \
WHERE s.SEGMENT_TYPE IN ('TABLE','TABLE PARTITION','TABLE SUBPARTITION') \
UNION ALL \
SELECT i.TABLE_NAME, s.BYTES \
FROM ALL_INDEXES i \
JOIN USER_SEGMENTS s ON s.SEGMENT_NAME = i.INDEX_NAME \
WHERE i.TABLE_OWNER = {} AND s.SEGMENT_TYPE IN ('INDEX','INDEX PARTITION','INDEX SUBPARTITION') \
) \
GROUP BY table_name \
) s ON s.TABLE_NAME = t.TABLE_NAME \
WHERE t.OWNER = {} AND t.OWNER = USER AND (t.NESTED IS NULL OR t.NESTED = 'NO') \
ORDER BY t.TABLE_NAME",
oracle_owner_filter(schema),
oracle_owner_filter(schema),
)
}
@@ -1837,8 +1799,7 @@ fn oracle_object_statistics_query_plan(schema: &str) -> Vec<ObjectStatisticsAtte
fn dameng_object_statistics_query_plan(schema: &str) -> Vec<ObjectStatisticsAttempt> {
vec![
("dba-segments", dameng_object_statistics_dba_segments_sql(schema), true),
("user-segments", dameng_object_statistics_user_segments_sql(schema), false),
("table-used-pages", dameng_object_statistics_sql(schema), true),
("rows-only", dameng_object_statistics_rows_only_sql(schema), true),
]
}
@@ -3127,26 +3088,26 @@ mod tests {
use super::agent_postgres_extension_fallback_config;
use super::db;
use super::{
clickhouse_metadata_database, dameng_object_statistics_dba_segments_sql,
dameng_object_statistics_rows_only_sql, dameng_object_statistics_user_segments_sql, deduplicate_column_infos,
ephemeral_agent_metadata_session_id, external_driver_statistics_dialect, external_driver_statistics_query_plan,
external_driver_uses_generic_ddl, external_driver_uses_mysql_ddl, filter_mongodb_agent_collections,
filter_mysql_system_databases_for_config, filter_object_infos, filter_table_infos, filter_visible_schema_names,
finalize_object_source, gaussdb_m_view_object_source_sql, gbase8a_object_statistics_sql,
is_agent_postgres_metadata_fallback_config, is_mysql_external_driver_config, is_oracle_external_driver_config,
is_retryable_metadata_error, metadata_error_action, metadata_name_or_comment_matches,
mysql_database_list_timeout, mysql_external_driver_ddl_from_query_result, mysql_external_driver_ddl_sql,
mysql_object_source_ddl_column_index, mysql_object_source_sql, mysql_table_list_source_for_config,
mysql_table_metadata_catalog, normalize_information_schema_table_type, oracle_columns_from_query_result,
oracle_columns_sql, oracle_columns_sql_for_resolved_owner, oracle_completion_synonyms_sql,
oracle_current_schema_from_query_result, oracle_object_statistics_dba_segments_sql,
oracle_object_statistics_from_query_result, oracle_object_statistics_rows_only_sql,
oracle_object_statistics_sql, oracle_object_statistics_user_segments_sql,
oracle_synonym_target_from_query_result, oracle_synonym_target_sql, oracle_table_comment_from_query_result,
oracle_table_comment_sql, oracle_table_comments_sql, presto_like_columns_from_query_result,
presto_like_information_schema_columns_sql, presto_like_information_schema_tables_sql,
presto_like_tables_from_query_result, reference_key_columns_from_indexes, reference_keys_from_indexes,
replace_metadata_runtime, should_append_oracle_style_comment_ddl, should_query_oracle_columns_via_sql_first,
clickhouse_metadata_database, dameng_object_statistics_rows_only_sql, dameng_object_statistics_sql,
deduplicate_column_infos, ephemeral_agent_metadata_session_id, external_driver_statistics_dialect,
external_driver_statistics_query_plan, external_driver_uses_generic_ddl, external_driver_uses_mysql_ddl,
filter_mongodb_agent_collections, filter_mysql_system_databases_for_config, filter_object_infos,
filter_table_infos, filter_visible_schema_names, finalize_object_source, gaussdb_m_view_object_source_sql,
gbase8a_object_statistics_sql, is_agent_postgres_metadata_fallback_config, is_mysql_external_driver_config,
is_oracle_external_driver_config, is_retryable_metadata_error, metadata_error_action,
metadata_name_or_comment_matches, mysql_database_list_timeout, mysql_external_driver_ddl_from_query_result,
mysql_external_driver_ddl_sql, mysql_object_source_ddl_column_index, mysql_object_source_sql,
mysql_table_list_source_for_config, mysql_table_metadata_catalog, normalize_information_schema_table_type,
oracle_columns_from_query_result, oracle_columns_sql, oracle_columns_sql_for_resolved_owner,
oracle_completion_synonyms_sql, oracle_current_schema_from_query_result,
oracle_object_statistics_dba_segments_sql, oracle_object_statistics_from_query_result,
oracle_object_statistics_rows_only_sql, oracle_object_statistics_sql,
oracle_object_statistics_user_segments_sql, oracle_synonym_target_from_query_result, oracle_synonym_target_sql,
oracle_table_comment_from_query_result, oracle_table_comment_sql, oracle_table_comments_sql,
presto_like_columns_from_query_result, presto_like_information_schema_columns_sql,
presto_like_information_schema_tables_sql, presto_like_tables_from_query_result,
reference_key_columns_from_indexes, reference_keys_from_indexes, replace_metadata_runtime,
should_append_oracle_style_comment_ddl, should_query_oracle_columns_via_sql_first,
table_comments_from_query_result, table_name_filter_matches, tdengine_table_comment_like_pattern,
tdengine_table_comment_sql, tdengine_table_comments_sql, uses_mongodb_agent_collection_listing,
visible_schema_filter, ExternalDriverStatisticsDialect, MetadataErrorAction, MysqlTableListSource,
@@ -4223,7 +4184,12 @@ done
assert!(oracle[0].1.contains("t.OWNER = 'DBX_TEST'"));
let dameng = external_driver_statistics_query_plan(ExternalDriverStatisticsDialect::Dameng, "dbx_test");
assert_eq!(dameng[0].1, dameng_object_statistics_dba_segments_sql("dbx_test"));
assert_eq!(
dameng.iter().map(|(source, ..)| *source).collect::<Vec<_>>(),
vec!["table-used-pages", "rows-only"]
);
assert_eq!(dameng[0].1, dameng_object_statistics_sql("dbx_test"));
assert!(dameng.iter().all(|(_, _, accept_empty)| *accept_empty));
let kingbase = external_driver_statistics_query_plan(ExternalDriverStatisticsDialect::Kingbase, "public");
assert_eq!(kingbase.len(), 1);
@@ -6180,24 +6146,106 @@ done
}
#[test]
fn dameng_object_statistics_sql_uses_available_segment_views() {
let dba_sql = dameng_object_statistics_dba_segments_sql("app's");
assert!(dba_sql.contains("DBA_SEGMENTS"));
assert!(dba_sql.contains("ALL_INDEXES"));
assert!(!dba_sql.contains("ALL_SEGMENTS"));
assert!(!dba_sql.contains("ALL_LOBS"));
assert!(dba_sql.contains("OWNER = 'APP''S'"));
assert!(dba_sql.contains("t.NESTED IS NULL OR t.NESTED = 'NO'"));
let user_sql = dameng_object_statistics_user_segments_sql("app's");
assert!(user_sql.contains("USER_SEGMENTS"));
assert!(user_sql.contains("t.OWNER = USER"));
fn dameng_object_statistics_sql_uses_bounded_table_size_function() {
let sql = dameng_object_statistics_sql("app's");
assert!(sql.contains("TABLE_USED_PAGES(t.OWNER, t.TABLE_NAME) * PAGE"));
assert!(sql.contains("OWNER = 'APP''S'"));
assert!(sql.contains("t.NESTED IS NULL OR t.NESTED = 'NO'"));
assert!(!sql.contains("SEGMENTS"));
assert!(!sql.contains("ALL_INDEXES"));
let rows_only_sql = dameng_object_statistics_rows_only_sql("app's");
assert!(rows_only_sql.contains("CAST(NULL AS NUMBER) AS TOTAL_BYTES"));
assert!(!rows_only_sql.contains("SEGMENTS"));
}
fn statistics_query_result(rows: Vec<Vec<serde_json::Value>>) -> db::QueryResult {
db::QueryResult {
columns: vec![
"TABLE_NAME".to_string(),
"OWNER".to_string(),
"NUM_ROWS".to_string(),
"TOTAL_BYTES".to_string(),
],
column_types: Vec::new(),
column_sortables: Vec::new(),
spatial_columns: vec![],
spatial_values: vec![],
rows,
affected_rows: 0,
execution_time_ms: 0,
server_execute_time_us: None,
query_timings_ms: None,
truncated: false,
session_id: None,
has_more: false,
elasticsearch_raw_body: None,
messages: Vec::new(),
}
}
#[tokio::test]
async fn dameng_object_statistics_falls_back_to_rows_when_size_lookup_fails() {
let sqls = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
let recorded_sqls = sqls.clone();
let stats = super::object_statistics_from_query_plan(
"Dameng",
"APP",
super::dameng_object_statistics_query_plan("APP"),
move |sql| {
let recorded_sqls = recorded_sqls.clone();
async move {
recorded_sqls.lock().unwrap().push(sql.clone());
if sql.contains("TABLE_USED_PAGES") {
Err("insufficient privilege".to_string())
} else {
Ok(statistics_query_result(vec![vec![
serde_json::json!("ORDERS"),
serde_json::json!("APP"),
serde_json::Value::Null,
serde_json::Value::Null,
]]))
}
}
},
)
.await
.unwrap();
let sqls = sqls.lock().unwrap();
assert_eq!(sqls.len(), 2);
assert!(sqls[0].contains("TABLE_USED_PAGES"));
assert!(sqls[1].contains("CAST(NULL AS NUMBER) AS TOTAL_BYTES"));
assert_eq!(stats.len(), 1);
assert_eq!(stats[0].estimated_rows, None);
assert_eq!(stats[0].total_bytes, None);
}
#[tokio::test]
async fn dameng_object_statistics_accepts_an_empty_size_result() {
let sqls = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
let recorded_sqls = sqls.clone();
let stats = super::object_statistics_from_query_plan(
"Dameng",
"EMPTY",
super::dameng_object_statistics_query_plan("EMPTY"),
move |sql| {
let recorded_sqls = recorded_sqls.clone();
async move {
recorded_sqls.lock().unwrap().push(sql);
Ok(statistics_query_result(Vec::new()))
}
},
)
.await
.unwrap();
assert!(stats.is_empty());
assert_eq!(sqls.lock().unwrap().len(), 1);
}
#[test]
fn gbase8a_object_statistics_sql_uses_information_schema() {
let gbase_sql = gbase8a_object_statistics_sql("shop's");