diff --git a/apps/desktop/src/lib/database/databaseObjectCapabilities.ts b/apps/desktop/src/lib/database/databaseObjectCapabilities.ts index c8385e6646..42d2b33dcf 100644 --- a/apps/desktop/src/lib/database/databaseObjectCapabilities.ts +++ b/apps/desktop/src/lib/database/databaseObjectCapabilities.ts @@ -69,7 +69,8 @@ const DATABASE_TYPE_OBJECTS = new Map([ // MV support that the backend cannot route. ["doris", TABLE_VIEW_OBJECTS], ["starrocks", TABLE_VIEW_MV_OBJECTS], - ["hive", TABLE_VIEW_OBJECTS], + // Inceptor/Hive routines can be listed via JDBC plugin fallbacks (system.procedures_v/functions_v). + ["hive", ROUTINE_OBJECTS], ["kyuubi", TABLE_VIEW_OBJECTS], ["impala", TABLE_VIEW_OBJECTS], ["spark", TABLE_VIEW_OBJECTS], 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 136d27fd7c..3382c48924 100644 --- a/plugins/jdbc/src/main/java/app/dbx/jdbc/DbxJdbcPlugin.java +++ b/plugins/jdbc/src/main/java/app/dbx/jdbc/DbxJdbcPlugin.java @@ -46,6 +46,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; import java.util.HashSet; +import java.util.LinkedHashSet; import java.util.LinkedHashMap; import java.util.List; import java.util.Locale; @@ -302,6 +303,11 @@ public final class DbxJdbcPlugin { private static final Pattern ORACLE_UNSUPPORTED_CHARSET_PATTERN = Pattern.compile("(?i)unsupported charset|不支持的字符集"); + // Inceptor/Hive adhoc engine error code. Matched with word boundaries so that + // incidental substrings (ports, durations like 107500 or 10750ms) do not trigger + // the adhoc hint retry. + private static final Pattern HIVE_ADHOC_ERROR_CODE_PATTERN = Pattern.compile("\\b10750\\b"); + // The base Oracle thin driver jar ships converters for a handful of charsets only; // databases such as ZHS16GBK need orai18n.jar, otherwise every metadata call that // reads dictionary comments fails wholesale. @@ -917,6 +923,60 @@ public final class DbxJdbcPlugin { return executeQueryOnConnection(connection, conn, sql, database, schema, maxRows, fetchSize, rowOffset, timeoutSecs); } + private static boolean shouldRetryWithAdhocHint(SQLException error) { + String msg = error == null ? null : error.getMessage(); + if (msg == null) { + return false; + } + String lower = msg.toLowerCase(Locale.ROOT); + return lower.contains("adhoc") || HIVE_ADHOC_ERROR_CODE_PATTERN.matcher(msg).find() || lower.contains("stream query"); + } + + private static boolean isPlainSelectStatement(String sql) { + if (sql == null || sql.isBlank()) { + return false; + } + // Only plain SELECT statements are retried with the hint. WITH ... SELECT is + // excluded because the hint would be injected before the first SELECT inside + // the CTE body instead of the outer query; DML and other statements are not + // silently rewritten and re-executed. + return "SELECT".equals(firstSqlKeyword(sql)); + } + + private static String injectAdhocHint(String sql) { + if (sql == null) { + return null; + } + String trimmed = sql.trim(); + if (trimmed.isEmpty() || trimmed.toLowerCase(Locale.ROOT).contains("adhoc")) { + return trimmed; + } + String body = stripLeadingSqlComments(trimmed); + if (!body.regionMatches(true, 0, "SELECT", 0, 6)) { + return trimmed; + } + int bodyStart = trimmed.length() - body.length(); + return trimmed.substring(0, bodyStart) + body.replaceFirst("(?i)^SELECT\\s+", "SELECT /*+ adhoc */ "); + } + + private static ExecutedStatement executeStatementForResultWithAdhocRetry( + JsonNode connection, + Statement statement, + String sql, + JdbcDriverQuirks quirks + ) throws SQLException { + try { + return executeStatementForResult(statement, sql, quirks); + } catch (SQLException error) { + if (isHive2RoutinesConnection(connection) + && isPlainSelectStatement(sql) + && shouldRetryWithAdhocHint(error)) { + return executeStatementForResult(statement, injectAdhocHint(sql), quirks); + } + throw error; + } + } + private static JsonNode executeQueryOnConnection( JsonNode connection, Connection conn, @@ -937,7 +997,7 @@ public final class DbxJdbcPlugin { applyStatementOptions(statement, maxRows, fetchSize, timeoutSecs, quirks); String trimmedSql = trimStatementSql(sql); String effectiveSql = rewritePhoenixSystemCatalogQuery(connection, conn, trimmedSql); - ExecutedStatement executed = executeStatementForResult(statement, effectiveSql, quirks); + ExecutedStatement executed = executeStatementForResultWithAdhocRetry(connection, statement, effectiveSql, quirks); ObjectNode result = MAPPER.createObjectNode(); ArrayNode columns = MAPPER.createArrayNode(); ArrayNode rows = MAPPER.createArrayNode(); @@ -1114,7 +1174,7 @@ public final class DbxJdbcPlugin { applyStatementOptions(statement, maxRows, fetchSize, timeoutSecs, quirks); String trimmedSql = trimStatementSql(sql); String effectiveSql = rewritePhoenixSystemCatalogQuery(connection, conn, trimmedSql); - ExecutedStatement executed = executeStatementForResult(statement, effectiveSql, quirks); + ExecutedStatement executed = executeStatementForResultWithAdhocRetry(connection, statement, effectiveSql, quirks); ResultSet rs = executed.resultSet(); if (rs == null) { ObjectNode result = MAPPER.createObjectNode(); @@ -2116,6 +2176,18 @@ public final class DbxJdbcPlugin { String schemaPattern = resolveSchemaPattern(meta, database, schema, quirks); Set allowedObjectTypes = normalizedObjectTypes(objectTypes); + boolean loadedRoutinesFromSystemTables = false; + if (isHive2RoutinesConnection(connection)) { + loadedRoutinesFromSystemTables = appendInceptorRoutinesFromSystemTables( + conn, + database, + schema, + filter, + result, + objectTypes + ); + } + if (!kingbase) { String[] tableTypes = constrainedJdbcTableTypes(jdbcTableTypes(meta), objectTypes); if (tableTypes.length > 0) { @@ -2126,7 +2198,7 @@ public final class DbxJdbcPlugin { } } - if (allowedObjectTypes.isEmpty() || allowedObjectTypes.contains("PROCEDURE")) { + if (!loadedRoutinesFromSystemTables && (allowedObjectTypes.isEmpty() || allowedObjectTypes.contains("PROCEDURE"))) { try (ResultSet rs = meta.getProcedures(catalog, schemaPattern, "%")) { while (rs != null && rs.next()) { ObjectNode item = MAPPER.createObjectNode(); @@ -2136,7 +2208,7 @@ public final class DbxJdbcPlugin { putNullable(item, "comment", rs.getString("REMARKS")); result.add(item); } - } catch (SQLException ignored) { + } catch (SQLException | AbstractMethodError | UnsupportedOperationException ignored) { } } @@ -2146,7 +2218,7 @@ public final class DbxJdbcPlugin { procedureNames.add(node.path("name").asText()); } } - if (allowedObjectTypes.isEmpty() || allowedObjectTypes.contains("FUNCTION")) { + if (!loadedRoutinesFromSystemTables && (allowedObjectTypes.isEmpty() || allowedObjectTypes.contains("FUNCTION"))) { try (ResultSet rs = meta.getFunctions(catalog, schemaPattern, "%")) { while (rs != null && rs.next()) { String name = rs.getString("FUNCTION_NAME"); @@ -2159,7 +2231,7 @@ public final class DbxJdbcPlugin { result.add(item); } } - } catch (SQLException ignored) { + } catch (SQLException | AbstractMethodError | UnsupportedOperationException ignored) { } } @@ -2170,6 +2242,118 @@ public final class DbxJdbcPlugin { return filterMetadataNodes(result, filter, limit, offset, objectTypes, "object_type", false); } + private static boolean isHive2RoutinesConnection(JsonNode connection) { + String url = optionalText(connection, "connection_string"); + if (url != null && urlMatchesPrefix(url, "jdbc:hive2:")) { + return true; + } + String driverClass = optionalText(connection, "jdbc_driver_class"); + if (driverClass == null) { + return false; + } + String normalized = driverClass.toLowerCase(Locale.ROOT); + return normalized.contains("inceptor") || normalized.contains("transwarp") || normalized.contains("hive"); + } + + private static boolean appendInceptorRoutinesFromSystemTables( + Connection conn, + String database, + String schema, + String filter, + ArrayNode result, + List objectTypes + ) { + Set allowedTypes = normalizedObjectTypes(objectTypes); + boolean wantAll = allowedTypes.isEmpty(); + boolean wantProcedures = wantAll || allowedTypes.contains("PROCEDURE"); + boolean wantFunctions = wantAll || allowedTypes.contains("FUNCTION"); + if (!wantProcedures && !wantFunctions) { + return false; + } + + String trimmedFilter = filter == null ? "" : filter.trim(); + String likePattern = trimmedFilter.isEmpty() ? "%" : "%" + trimmedFilter + "%"; + + LinkedHashSet candidates = new LinkedHashSet<>(); + String db = emptyToNull(database); + if (db != null) { + candidates.add(db); + } + String sc = emptyToNull(schema); + if (sc != null) { + candidates.add(sc); + } + if (candidates.isEmpty()) { + return false; + } + + int added = 0; + try { + for (String candidateDb : candidates) { + if (wantProcedures) { + String sql = + "SELECT procedure_name FROM system.procedures_v " + + "WHERE lower(database_name) = lower(?) AND lower(procedure_name) LIKE lower(?) " + + "ORDER BY procedure_name"; + try (PreparedStatement ps = conn.prepareStatement(sql)) { + ps.setString(1, candidateDb); + ps.setString(2, likePattern); + try (ResultSet rs = ps.executeQuery()) { + while (rs.next()) { + ObjectNode item = MAPPER.createObjectNode(); + item.put("name", rs.getString(1)); + item.put("object_type", "PROCEDURE"); + putNullable(item, "schema", schema); + item.putNull("comment"); + result.add(item); + added++; + } + } + } + } + + if (wantFunctions) { + String sql = + "SELECT function_name FROM system.functions_v " + + "WHERE lower(database_name) = lower(?) AND lower(function_name) LIKE lower(?) " + + "ORDER BY function_name"; + try (PreparedStatement ps = conn.prepareStatement(sql)) { + ps.setString(1, candidateDb); + ps.setString(2, likePattern); + try (ResultSet rs = ps.executeQuery()) { + while (rs.next()) { + ObjectNode item = MAPPER.createObjectNode(); + item.put("name", rs.getString(1)); + item.put("object_type", "FUNCTION"); + putNullable(item, "schema", schema); + item.putNull("comment"); + result.add(item); + added++; + } + } + } + } + + if (added > 0) { + break; + } + } + } catch (SQLException | AbstractMethodError | UnsupportedOperationException ignored) { + return added > 0; + } + + return added > 0; + } + + private static String stripRoutineSignature(String name) { + if (name == null) { + return null; + } + String trimmed = name.trim(); + int paren = trimmed.indexOf('('); + return paren > 0 ? trimmed.substring(0, paren).trim() : trimmed; + } + private static boolean isMysqlFamilyConnection(JsonNode connection) { String url = jdbcUrl(connection); return urlMatchesPrefix(url, "jdbc:mysql:") || urlMatchesPrefix(url, "jdbc:mariadb:") || urlMatchesPrefix(url, "jdbc:tidb:"); @@ -3733,6 +3917,55 @@ public final class DbxJdbcPlugin { } } + if (isHive2RoutinesConnection(connection)) { + String routineName = stripRoutineSignature(name); + String normalizedType = normalizeObjectType(objectType); + + LinkedHashSet candidates = new LinkedHashSet<>(); + String db = emptyToNull(database); + if (db != null) { + candidates.add(db); + } + String sc = emptyToNull(schema); + if (sc != null) { + candidates.add(sc); + } + if (candidates.isEmpty()) { + throw new SQLException("Object source requires database context for Hive/Inceptor routines"); + } + + for (String candidateDb : candidates) { + String sql; + if ("PROCEDURE".equals(normalizedType)) { + sql = "SELECT full_text FROM system.procedures_v " + + "WHERE lower(database_name) = lower(?) AND procedure_name = ?"; + } else if ("FUNCTION".equals(normalizedType)) { + sql = "SELECT full_text FROM system.functions_v " + + "WHERE lower(database_name) = lower(?) AND function_name = ?"; + } else { + throw new SQLException("Unsupported object_type for Hive/Inceptor routine source: " + objectType); + } + + try (PreparedStatement ps = conn.prepareStatement(sql)) { + ps.setString(1, candidateDb); + ps.setString(2, routineName); + try (ResultSet rs = ps.executeQuery()) { + if (!rs.next()) { + continue; + } + ObjectNode item = MAPPER.createObjectNode(); + item.put("name", name); + item.put("object_type", objectType); + putNullable(item, "schema", emptyToNull(schema) != null ? schema : candidateDb); + putNullable(item, "source", rs.getString(1)); + return item; + } + } + } + + throw new SQLException("Object source not found"); + } + throw new SQLException("Object source is not supported by this JDBC driver"); } 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 7d45aa59e5..eb3e0f0bc1 100644 --- a/plugins/jdbc/src/test/java/app/dbx/jdbc/DbxJdbcPluginTest.java +++ b/plugins/jdbc/src/test/java/app/dbx/jdbc/DbxJdbcPluginTest.java @@ -2133,6 +2133,44 @@ final class DbxJdbcPluginTest { } } + @Test + void hiveAdhocRetryOnlyRewritesSelectStatements() throws Exception { + List executedSql = new ArrayList<>(); + Driver driver = new AdhocFailingHiveDriver(executedSql); + DriverManager.registerDriver(driver); + String connection = """ + { + "connection_string": "jdbc:hive2://adhoc-retry-test:10000/default", + "connect_timeout_secs": 30 + } + """; + try { + JsonNode insertResponse = request("executeQuery", """ + { + "connection": %s, + "sql": "INSERT INTO logs VALUES (1)" + } + """.formatted(connection)); + + assertTrue(insertResponse.has("error"), insertResponse.toString()); + assertTrue(insertResponse.path("error").path("message").asText().contains("10750"), insertResponse.toString()); + assertEquals(List.of("INSERT INTO logs VALUES (1)"), executedSql); + + executedSql.clear(); + JsonNode selectResponse = request("executeQuery", """ + { + "connection": %s, + "sql": "SELECT name FROM users" + } + """.formatted(connection)); + + assertTrue(selectResponse.has("error"), selectResponse.toString()); + assertEquals(List.of("SELECT name FROM users", "SELECT /*+ adhoc */ name FROM users"), executedSql); + } finally { + DriverManager.deregisterDriver(driver); + } + } + @Test void listDataTypesUsesJdbcTypeInfo() throws Exception { JsonNode response = request("listDataTypes", """ @@ -3828,6 +3866,10 @@ final class DbxJdbcPluginTest { new Class[] { Connection.class }, (proxy, method, args) -> switch (method.getName()) { case "getMetaData" -> metadata; + // Hive fallback probes "SHOW DATABASES" via a plain statement before + // falling back to getSchemas; report no rows so the schema fallback + // path stays covered. + case "createStatement" -> emptyQueryResultStatement(); case "isClosed" -> false; case "isValid" -> true; case "getCatalog" -> null; @@ -3837,6 +3879,17 @@ final class DbxJdbcPluginTest { ); } + private static Statement emptyQueryResultStatement() { + return (Statement) Proxy.newProxyInstance( + DbxJdbcPluginTest.class.getClassLoader(), + new Class[] { Statement.class }, + (proxy, method, args) -> switch (method.getName()) { + case "executeQuery" -> rowsResultSet(new String[] { "database_name" }, new Object[0][]); + default -> defaultValue(method.getReturnType()); + } + ); + } + private enum RoutineMetadataBehavior { NULL, ROWS, @@ -4493,6 +4546,79 @@ final class DbxJdbcPluginTest { return null; } + private static final class AdhocFailingHiveDriver implements Driver { + private final List executedSql; + + private AdhocFailingHiveDriver(List executedSql) { + this.executedSql = executedSql; + } + + @Override + public Connection connect(String url, Properties info) { + if (!acceptsURL(url)) { + return null; + } + return (Connection) Proxy.newProxyInstance( + DbxJdbcPluginTest.class.getClassLoader(), + new Class[] { Connection.class }, + (proxy, method, args) -> switch (method.getName()) { + case "createStatement" -> adhocFailingStatement(executedSql); + case "isClosed" -> false; + case "close" -> null; + default -> defaultValue(method.getReturnType()); + } + ); + } + + @Override + public boolean acceptsURL(String url) { + return url != null && url.startsWith("jdbc:hive2://adhoc-retry-test:"); + } + + @Override + public DriverPropertyInfo[] getPropertyInfo(String url, Properties info) { + return new DriverPropertyInfo[0]; + } + + @Override + public int getMajorVersion() { + return 1; + } + + @Override + public int getMinorVersion() { + return 0; + } + + @Override + public boolean jdbcCompliant() { + return false; + } + + @Override + public java.util.logging.Logger getParentLogger() { + return java.util.logging.Logger.getGlobal(); + } + } + + private static Statement adhocFailingStatement(List executedSql) { + return (Statement) Proxy.newProxyInstance( + DbxJdbcPluginTest.class.getClassLoader(), + new Class[] { Statement.class }, + (proxy, method, args) -> { + if ("execute".equals(method.getName()) || "executeQuery".equals(method.getName())) { + executedSql.add((String) args[0]); + throw new SQLException( + "FAILED: Execution Error, return code 10750 from org.apache.hadoop.hive.ql.exec.mr.MapRedTask"); + } + return switch (method.getName()) { + case "setMaxRows", "setFetchSize", "setQueryTimeout", "close" -> null; + default -> defaultValue(method.getReturnType()); + }; + } + ); + } + private static final class Hive2ConnectThrowsUnsupportedDriver implements Driver { @Override public Connection connect(String connectUrl, Properties info) { @@ -4533,7 +4659,9 @@ final class DbxJdbcPluginTest { } } - private static final class Hive2ConnectGoodDriver implements Driver { + // Instantiated reflectively by DbxJdbcPlugin through jdbc_driver_class, so it + // must stay accessible outside this class. + public static final class Hive2ConnectGoodDriver implements Driver { private static final String URL = "jdbc:hive2://hive2-connect-test:10000/default"; @Override