From c62e323e16db67cdb78c91c2a09fc3d16650b45d Mon Sep 17 00:00:00 2001 From: t8y2 <1156263951@qq.com> Date: Thu, 18 Jun 2026 16:47:46 +0800 Subject: [PATCH] fix(jdbc): improve goldendb metadata handling --- apps/desktop/src/lib/jdbcDialect.ts | 1 + crates/dbx-core/src/plugins.rs | 108 +++++++++++++----- packages/app-tests/jdbcDialect.test.ts | 20 ++++ .../main/java/app/dbx/jdbc/DbxJdbcPlugin.java | 20 +++- .../java/app/dbx/jdbc/DbxJdbcPluginTest.java | 86 ++++++++++++++ 5 files changed, 200 insertions(+), 35 deletions(-) create mode 100644 packages/app-tests/jdbcDialect.test.ts diff --git a/apps/desktop/src/lib/jdbcDialect.ts b/apps/desktop/src/lib/jdbcDialect.ts index a7a9132c3..ace47499b 100644 --- a/apps/desktop/src/lib/jdbcDialect.ts +++ b/apps/desktop/src/lib/jdbcDialect.ts @@ -7,6 +7,7 @@ const JDBC_DIALECT_MATCHERS: Array<{ type: DatabaseType; patterns: RegExp[] }> = { type: "databend", patterns: [/jdbc:databend:/i, /com\.databend\.jdbc\.DatabendDriver/i, /databend-jdbc/i] }, { type: "starrocks", patterns: [/starrocks/i] }, { type: "doris", patterns: [/doris/i] }, + { type: "goldendb", patterns: [/jdbc:goldendb:/i, /goldendb/i] }, { type: "hive", patterns: [/org\.apache\.hive\.jdbc\.HiveDriver/i, /hive-jdbc/i] }, { type: "mysql", patterns: [/jdbc:mysql:/i, /mysql/i, /mariadb/i, /kyuubi/i, /hive2/i] }, { type: "postgres", patterns: [/jdbc:postgresql:/i, /postgres/i] }, diff --git a/crates/dbx-core/src/plugins.rs b/crates/dbx-core/src/plugins.rs index 9d7d40761..76e5445a8 100644 --- a/crates/dbx-core/src/plugins.rs +++ b/crates/dbx-core/src/plugins.rs @@ -313,25 +313,16 @@ impl PluginDriverSession { process.stdin.write_all(&line).await.map_err(|err| err.to_string())?; process.stdin.flush().await.map_err(|err| err.to_string())?; - let response_line = match read_plugin_line(&mut process.stdout, "response").await { - Ok(line) => line, - Err(err) if err.contains("end of stream") => { - let status = process.child.try_wait().map_err(|err| err.to_string())?; - return Err(match status { - Some(status) => format!("Plugin '{}' exited with status {}", self.plugin.manifest.id, status), - None => format!("Plugin '{}' closed stdout without a response", self.plugin.manifest.id), - }); - } - Err(err) => return Err(err), - }; - if response_line.is_empty() { - let status = process.child.try_wait().map_err(|err| err.to_string())?; - return Err(match status { + let stdout = &mut process.stdout; + let child = &mut process.child; + read_decoded_plugin_response(&self.plugin, request_id, stdout, || { + let status = child.try_wait().map_err(|err| err.to_string())?; + Ok(match status { Some(status) => format!("Plugin '{}' exited with status {}", self.plugin.manifest.id, status), None => format!("Plugin '{}' closed stdout without a response", self.plugin.manifest.id), - }); - } - decode_plugin_response(&self.plugin, request_id, &response_line) + }) + }) + .await } async fn kill(&self) { @@ -368,14 +359,17 @@ where let stdout = child.stdout.take().ok_or("Plugin stdout unavailable")?; let mut stderr = child.stderr.take().ok_or("Plugin stderr unavailable")?; let mut reader = BufReader::new(stdout); - let response_line_result = read_plugin_line(&mut reader, "response").await; + let response_result = read_decoded_plugin_response(plugin, request.id, &mut reader, || { + Ok(format!("Plugin '{}' exited without a response", plugin.manifest.id)) + }) + .await; let mut stderr_bytes = Vec::new(); stderr.read_to_end(&mut stderr_bytes).await.map_err(|err| err.to_string())?; let stderr_text = String::from_utf8_lossy(&stderr_bytes).into_owned(); let status = child.wait().await.map_err(|err| err.to_string())?; - let response_line = match response_line_result { - Ok(line) => line, + let response = match response_result { + Ok(response) => response, Err(err) if err.contains("end of stream") => { let stderr = stderr_text.trim().to_string(); return Err(if stderr.is_empty() { @@ -386,14 +380,6 @@ where } Err(err) => return Err(err), }; - if response_line.is_empty() { - let stderr = stderr_text.trim().to_string(); - return Err(if stderr.is_empty() { - format!("Plugin '{}' exited without a response", plugin.manifest.id) - } else { - format!("Plugin '{}' exited without a response: {stderr}", plugin.manifest.id) - }); - } if !status.success() { let stderr = stderr_text.trim().to_string(); return Err(if stderr.is_empty() { @@ -403,7 +389,7 @@ where }); } - decode_plugin_response(plugin, request.id, &response_line) + Ok(response) } fn encode_plugin_request_line(request: &PluginRequest) -> Result, String> { @@ -453,6 +439,42 @@ where Ok(String::from_utf8_lossy(&bytes).into_owned()) } +async fn read_decoded_plugin_response( + plugin: &InstalledPlugin, + request_id: u64, + reader: &mut R, + end_of_stream_message: F, +) -> Result +where + T: DeserializeOwned, + R: tokio::io::AsyncBufRead + Unpin, + F: FnMut() -> Result, +{ + let mut end_of_stream_message = end_of_stream_message; + const MAX_NOISY_LINES: usize = 20; + for _ in 0..MAX_NOISY_LINES { + let line = match read_plugin_line(reader, "response").await { + Ok(line) => line, + Err(err) if err.contains("end of stream") => return Err(end_of_stream_message()?), + Err(err) => return Err(err), + }; + match decode_plugin_response(plugin, request_id, &line) { + Ok(response) => return Ok(response), + Err(err) if err.starts_with(&format!("Failed to parse plugin '{}' response:", plugin.manifest.id)) => { + log::warn!("[plugin:{}] ignored non-protocol stdout: {}", plugin.manifest.id, line.trim_end()); + } + Err(err) => return Err(err), + } + } + match read_plugin_line(reader, "response").await { + Ok(_) => { + Err(format!("Plugin '{}' wrote too many non-protocol stdout lines before its response", plugin.manifest.id)) + } + Err(err) if err.contains("end of stream") => Err(end_of_stream_message()?), + Err(err) => Err(err), + } +} + fn decode_plugin_response(plugin: &InstalledPlugin, request_id: u64, response_line: &str) -> Result where T: DeserializeOwned, @@ -487,7 +509,8 @@ fn resolve_plugin_executable(plugin_dir: &Path, executable: &str) -> PathBuf { #[cfg(test)] mod tests { - use super::read_plugin_line; + use super::{read_decoded_plugin_response, read_plugin_line, InstalledPlugin, PluginManifest}; + use std::path::PathBuf; use tokio::io::BufReader; #[tokio::test] @@ -499,4 +522,29 @@ mod tests { assert_eq!(line, format!("{{\"error\":{}}}\n", "\u{fffd}\u{fffd}")); } + + #[tokio::test] + async fn skips_noisy_plugin_stdout_before_json_response() { + let plugin = InstalledPlugin { + manifest: PluginManifest { + id: "jdbc".to_string(), + name: "JDBC".to_string(), + version: "test".to_string(), + protocol_version: 1, + description: String::new(), + executable: None, + drivers: Vec::new(), + }, + path: PathBuf::new(), + }; + let bytes = b"driver banner\n{\"id\":1,\"result\":{\"ok\":true}}\n"; + let mut reader = BufReader::new(std::io::Cursor::new(bytes)); + + let result: serde_json::Value = + read_decoded_plugin_response(&plugin, 1, &mut reader, || Ok("closed".to_string())) + .await + .expect("response should decode after noisy line"); + + assert_eq!(result["ok"], true); + } } diff --git a/packages/app-tests/jdbcDialect.test.ts b/packages/app-tests/jdbcDialect.test.ts new file mode 100644 index 000000000..4dd750883 --- /dev/null +++ b/packages/app-tests/jdbcDialect.test.ts @@ -0,0 +1,20 @@ +import { strict as assert } from "node:assert"; +import { test } from "vitest"; +import { effectiveDatabaseTypeForConnection, inferJdbcDialect } from "../../apps/desktop/src/lib/jdbcDialect.ts"; + +test("infers GoldenDB for generic JDBC connections", () => { + assert.equal( + inferJdbcDialect({ + db_type: "jdbc", + connection_string: "jdbc:goldendb://127.0.0.1:3306/app", + }), + "goldendb", + ); + assert.equal( + effectiveDatabaseTypeForConnection({ + db_type: "jdbc", + jdbc_driver_class: "com.goldendb.jdbc.Driver", + }), + "goldendb", + ); +}); diff --git a/plugins/jdbc/src/main/java/app/dbx/jdbc/DbxJdbcPlugin.java b/plugins/jdbc/src/main/java/app/dbx/jdbc/DbxJdbcPlugin.java index 7d810cde7..c001d90de 100644 --- a/plugins/jdbc/src/main/java/app/dbx/jdbc/DbxJdbcPlugin.java +++ b/plugins/jdbc/src/main/java/app/dbx/jdbc/DbxJdbcPlugin.java @@ -973,7 +973,7 @@ public final class DbxJdbcPlugin { appendColumns(result, meta, null, schemaPattern, table, primaryKeys); } if (quirks.useCatalogFallbackSql()) { - mergeShowFullColumnComments(conn, result, schemaPattern, table); + mergeShowFullColumnMetadata(conn, result, schemaPattern, table); } return result; } @@ -1148,19 +1148,29 @@ public final class DbxJdbcPlugin { } } - private static void mergeShowFullColumnComments(Connection conn, ArrayNode result, String schema, String table) { + private static void mergeShowFullColumnMetadata(Connection conn, ArrayNode result, String schema, String table) { String target = qualifiedJdbcTableName(schema, table); try (Statement statement = conn.createStatement(); ResultSet rs = statement.executeQuery("SHOW FULL COLUMNS FROM " + target)) { int fieldIndex = resultSetColumnIndex(rs, "Field"); + int typeIndex = resultSetColumnIndex(rs, "Type"); + int extraIndex = resultSetColumnIndex(rs, "Extra"); int commentIndex = resultSetColumnIndex(rs, "Comment"); - if (fieldIndex <= 0 || commentIndex <= 0) { + if (fieldIndex <= 0) { return; } while (rs.next()) { String name = rs.getString(fieldIndex); - String comment = rs.getString(commentIndex); if (name != null) { - putNullablePreferValue(columnNode(result, name), "comment", comment); + ObjectNode item = columnNode(result, name); + if (typeIndex > 0) { + putNullablePreferValue(item, "data_type", rs.getString(typeIndex)); + } + if (extraIndex > 0) { + putNullablePreferValue(item, "extra", rs.getString(extraIndex)); + } + if (commentIndex > 0) { + putNullablePreferValue(item, "comment", rs.getString(commentIndex)); + } } } } catch (SQLException | AbstractMethodError | UnsupportedOperationException ignored) { diff --git a/plugins/jdbc/src/test/java/app/dbx/jdbc/DbxJdbcPluginTest.java b/plugins/jdbc/src/test/java/app/dbx/jdbc/DbxJdbcPluginTest.java index ea31195c7..bee094732 100644 --- a/plugins/jdbc/src/test/java/app/dbx/jdbc/DbxJdbcPluginTest.java +++ b/plugins/jdbc/src/test/java/app/dbx/jdbc/DbxJdbcPluginTest.java @@ -2,6 +2,8 @@ package app.dbx.jdbc; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ArrayNode; +import com.fasterxml.jackson.databind.node.ObjectNode; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; @@ -557,6 +559,30 @@ final class DbxJdbcPluginTest { assertEquals(true, response.path("result").path(0).path("is_primary_key").asBoolean()); } + @Test + void showFullColumnsMetadataCompletesMysqlCompatibleTypesAndComments() throws Exception { + Method method = DbxJdbcPlugin.class.getDeclaredMethod( + "mergeShowFullColumnMetadata", + Connection.class, + ArrayNode.class, + String.class, + String.class + ); + method.setAccessible(true); + ArrayNode columns = MAPPER.createArrayNode(); + ObjectNode column = columns.addObject(); + column.put("name", "name"); + column.put("data_type", "varchar"); + column.putNull("extra"); + column.putNull("comment"); + + method.invoke(null, showFullColumnsConnection(), columns, "app", "people"); + + assertEquals("varchar(32)", columns.path(0).path("data_type").asText()); + assertEquals("auto_increment", columns.path(0).path("extra").asText()); + assertEquals("姓名", columns.path(0).path("comment").asText()); + } + @Test void oracleMetadataObjectTypeAcceptsPackageBodyAliases() throws Exception { Method method = DbxJdbcPlugin.class.getDeclaredMethod("oracleMetadataObjectType", String.class); @@ -744,6 +770,66 @@ final class DbxJdbcPluginTest { ); } + private static Connection showFullColumnsConnection() { + return (Connection) Proxy.newProxyInstance( + DbxJdbcPluginTest.class.getClassLoader(), + new Class[] { Connection.class }, + (proxy, method, args) -> switch (method.getName()) { + case "createStatement" -> showFullColumnsStatement(); + case "isClosed" -> false; + case "close" -> null; + default -> defaultValue(method.getReturnType()); + } + ); + } + + private static Statement showFullColumnsStatement() { + return (Statement) Proxy.newProxyInstance( + DbxJdbcPluginTest.class.getClassLoader(), + new Class[] { Statement.class }, + (proxy, method, args) -> switch (method.getName()) { + case "executeQuery" -> showFullColumnsResultSet(); + case "close" -> null; + default -> defaultValue(method.getReturnType()); + } + ); + } + + private static ResultSet showFullColumnsResultSet() { + String[] labels = { "Field", "Type", "Extra", "Comment" }; + String[][] rows = { { "name", "varchar(32)", "auto_increment", "姓名" } }; + return (ResultSet) Proxy.newProxyInstance( + DbxJdbcPluginTest.class.getClassLoader(), + new Class[] { ResultSet.class }, + new java.lang.reflect.InvocationHandler() { + private int index = -1; + + @Override + public Object invoke(Object proxy, Method method, Object[] args) { + return switch (method.getName()) { + case "next" -> ++index < rows.length; + case "getMetaData" -> resultSetMeta(labels); + case "getString" -> rows[index][((Integer) args[0]) - 1]; + case "close" -> null; + default -> defaultValue(method.getReturnType()); + }; + } + } + ); + } + + private static ResultSetMetaData resultSetMeta(String[] labels) { + return (ResultSetMetaData) Proxy.newProxyInstance( + DbxJdbcPluginTest.class.getClassLoader(), + new Class[] { ResultSetMetaData.class }, + (proxy, method, args) -> switch (method.getName()) { + case "getColumnCount" -> labels.length; + case "getColumnLabel", "getColumnName" -> labels[((Integer) args[0]) - 1]; + default -> defaultValue(method.getReturnType()); + } + ); + } + private static final class BrokenResultSetDriver implements Driver { private final String urlPrefix; private final boolean executeReturnsResultSet;