mirror of
https://github.com/t8y2/dbx.git
synced 2026-10-21 05:20:02 +00:00
refactor(hive): remove retired Java agent
This commit is contained in:
@@ -1,11 +0,0 @@
|
||||
dependencies {
|
||||
implementation fileTree(dir: 'libs', include: ['*.jar'])
|
||||
implementation 'org.apache.hive:hive-jdbc:4.0.1:standalone'
|
||||
implementation 'org.slf4j:slf4j-nop:1.7.30'
|
||||
}
|
||||
|
||||
tasks.named('shadowJar') {
|
||||
manifest {
|
||||
attributes('Agent-Label': 'Apache Hive', 'Main-Class': 'com.dbx.agent.hive.HiveAgent')
|
||||
}
|
||||
}
|
||||
@@ -1,293 +0,0 @@
|
||||
package com.dbx.agent.hive;
|
||||
|
||||
import com.dbx.agent.AbstractJdbcAgent;
|
||||
import com.dbx.agent.ColumnInfo;
|
||||
import com.dbx.agent.ConnectParams;
|
||||
import com.dbx.agent.DatabaseInfo;
|
||||
import com.dbx.agent.ForeignKeyInfo;
|
||||
import com.dbx.agent.IndexInfo;
|
||||
import com.dbx.agent.JdbcIdentifiers;
|
||||
import com.dbx.agent.MultiSessionJsonRpcServer;
|
||||
import com.dbx.agent.TableInfo;
|
||||
import com.dbx.agent.TriggerInfo;
|
||||
import com.dbx.agent.ExecuteQueryOptions;
|
||||
import com.dbx.agent.QueryPageOptions;
|
||||
import com.dbx.agent.QueryPageResult;
|
||||
import com.dbx.agent.QueryResult;
|
||||
import java.sql.Connection;
|
||||
import java.sql.DatabaseMetaData;
|
||||
import java.sql.ResultSet;
|
||||
import java.sql.Statement;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.Comparator;
|
||||
import java.util.List;
|
||||
import java.util.Locale;
|
||||
|
||||
public final class HiveAgent extends AbstractJdbcAgent {
|
||||
|
||||
@Override
|
||||
protected String driverClass() {
|
||||
return "org.apache.hive.jdbc.HiveDriver";
|
||||
}
|
||||
|
||||
@Override
|
||||
protected String buildJdbcUrl(ConnectParams params) {
|
||||
return buildUrl(params);
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<DatabaseInfo> listDatabases() {
|
||||
return unchecked(() -> {
|
||||
List<DatabaseInfo> result = new ArrayList<>();
|
||||
try (java.sql.Statement stmt = requireConnected().createStatement();
|
||||
ResultSet rs = stmt.executeQuery("SHOW DATABASES")) {
|
||||
while (rs.next()) {
|
||||
result.add(new DatabaseInfo(rs.getString(1)));
|
||||
}
|
||||
}
|
||||
result.sort(Comparator.comparing(DatabaseInfo::getName));
|
||||
return result;
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<String> listSchemas() {
|
||||
List<String> result = new ArrayList<>();
|
||||
for (DatabaseInfo database : listDatabases()) {
|
||||
result.add(database.getName());
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<TableInfo> listTables(String schema) {
|
||||
return unchecked(() -> {
|
||||
List<TableInfo> result = new ArrayList<>();
|
||||
useSchema(schema);
|
||||
try (java.sql.Statement stmt = requireConnected().createStatement();
|
||||
ResultSet rs = stmt.executeQuery("SHOW TABLES")) {
|
||||
while (rs.next()) {
|
||||
result.add(new TableInfo(rs.getString(1), "TABLE", null));
|
||||
}
|
||||
}
|
||||
result.sort(Comparator.comparing(TableInfo::getName));
|
||||
return result;
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<ColumnInfo> getColumns(String schema, String table) {
|
||||
return unchecked(() -> {
|
||||
try {
|
||||
return getColumnsFromDescribe(schema, table);
|
||||
} catch (Exception ignored) {
|
||||
return getColumnsFromMetadata(requireConnected(), schema, table);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getTableDdl(String schema, String table) {
|
||||
return unchecked(() -> {
|
||||
String qualifiedName = schema == null || schema.trim().isEmpty()
|
||||
? JdbcIdentifiers.INSTANCE.backtick(table)
|
||||
: JdbcIdentifiers.INSTANCE.backtick(schema) + "." + JdbcIdentifiers.INSTANCE.backtick(table);
|
||||
try (Statement stmt = requireConnected().createStatement();
|
||||
ResultSet rs = stmt.executeQuery("SHOW CREATE TABLE " + qualifiedName)) {
|
||||
StringBuilder ddl = new StringBuilder();
|
||||
while (rs.next()) {
|
||||
String line = rs.getString(1);
|
||||
if (line != null) {
|
||||
ddl.append(line).append('\n');
|
||||
}
|
||||
}
|
||||
return ddl.toString();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<IndexInfo> listIndexes(String schema, String table) {
|
||||
return Collections.emptyList();
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<ForeignKeyInfo> listForeignKeys(String schema, String table) {
|
||||
return Collections.emptyList();
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<TriggerInfo> listTriggers(String schema, String table) {
|
||||
return Collections.emptyList();
|
||||
}
|
||||
|
||||
@Override
|
||||
public String setSchemaSQL(String schema) {
|
||||
return "USE " + JdbcIdentifiers.INSTANCE.backtick(schema);
|
||||
}
|
||||
|
||||
@Override
|
||||
public QueryResult executeQuery(String sql, String schema, ExecuteQueryOptions options) {
|
||||
// Hive JDBC fetches rows via Thrift FetchResults in batches controlled
|
||||
// by fetchSize. A large fetchSize can fail with "Error retrieving next
|
||||
// row" when the result contains large binary/struct columns. Use a
|
||||
// small fetchSize (50) so each Thrift batch stays within limits.
|
||||
return super.executeQuery(sql, schema, new ExecuteQueryOptions(
|
||||
options.getMaxRows(),
|
||||
50,
|
||||
options.getTimeoutSecs()
|
||||
));
|
||||
}
|
||||
|
||||
@Override
|
||||
public QueryPageResult executeQueryPage(String sql, String schema, QueryPageOptions options) {
|
||||
return super.executeQueryPage(sql, schema, withSmallFetchSize(options));
|
||||
}
|
||||
|
||||
@Override
|
||||
public QueryPageResult startTableRead(String sql, String schema, QueryPageOptions options) {
|
||||
return super.startTableRead(sql, schema, withSmallFetchSize(options));
|
||||
}
|
||||
|
||||
private static QueryPageOptions withSmallFetchSize(QueryPageOptions options) {
|
||||
return new QueryPageOptions(
|
||||
options.getPageSize(), 50, options.getMaxRows(), options.getTimeoutSecs()
|
||||
);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected Object resultValue(ResultSet rs, int index, int sqlType) {
|
||||
return unchecked(() -> {
|
||||
Object value = rs.getObject(index);
|
||||
return rs.wasNull() ? null : value == null ? null : value.toString();
|
||||
});
|
||||
}
|
||||
|
||||
private void useSchema(String schema) throws Exception {
|
||||
applySchemaContext(requireConnected(), schema);
|
||||
}
|
||||
|
||||
private List<ColumnInfo> getColumnsFromDescribe(String schema, String table) throws Exception {
|
||||
List<ColumnInfo> result = new ArrayList<>();
|
||||
useSchema(schema);
|
||||
try (java.sql.Statement stmt = requireConnected().createStatement();
|
||||
ResultSet rs = stmt.executeQuery("DESCRIBE " + JdbcIdentifiers.INSTANCE.backtick(table))) {
|
||||
while (rs.next()) {
|
||||
String colName = trimToNull(rs.getString(1));
|
||||
if (colName == null || colName.startsWith("#")) {
|
||||
continue;
|
||||
}
|
||||
result.add(new ColumnInfo(
|
||||
colName,
|
||||
trimToEmpty(rs.getString(2)),
|
||||
true,
|
||||
null,
|
||||
false,
|
||||
null,
|
||||
optionalComment(rs),
|
||||
null,
|
||||
null,
|
||||
null
|
||||
));
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
private static List<ColumnInfo> getColumnsFromMetadata(Connection conn, String schema, String table) throws Exception {
|
||||
List<ColumnInfo> result = new ArrayList<>();
|
||||
DatabaseMetaData meta = conn.getMetaData();
|
||||
try (ResultSet rs = meta.getColumns(null, trimToNull(schema), table, "%")) {
|
||||
while (rs.next()) {
|
||||
String name = rs.getString("COLUMN_NAME");
|
||||
if (trimToNull(name) == null) {
|
||||
continue;
|
||||
}
|
||||
result.add(new ColumnInfo(
|
||||
name,
|
||||
trimToEmpty(rs.getString("TYPE_NAME")),
|
||||
rs.getInt("NULLABLE") != DatabaseMetaData.columnNoNulls,
|
||||
rs.getString("COLUMN_DEF"),
|
||||
false,
|
||||
null,
|
||||
rs.getString("REMARKS"),
|
||||
intOrNull(rs, "COLUMN_SIZE"),
|
||||
intOrNull(rs, "DECIMAL_DIGITS"),
|
||||
characterLength(rs)
|
||||
));
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
static String buildUrl(ConnectParams params) {
|
||||
String connectionString = trimToEmpty(params.getConnection_string());
|
||||
if (!connectionString.isEmpty()) {
|
||||
return connectionString;
|
||||
}
|
||||
String url = "jdbc:hive2://" + params.getHost() + ":" + params.getPort() + "/" + params.getDatabase();
|
||||
return appendHiveUrlParams(url, params.getUrl_params());
|
||||
}
|
||||
|
||||
private static String appendHiveUrlParams(String url, String urlParams) {
|
||||
String params = trimHiveUrlParams(urlParams);
|
||||
if (params.isEmpty()) {
|
||||
return url;
|
||||
}
|
||||
// Hive JDBC authentication options use semicolon-separated URL properties.
|
||||
return url + (url.endsWith(";") ? "" : ";") + params;
|
||||
}
|
||||
|
||||
private static String trimHiveUrlParams(String urlParams) {
|
||||
String value = trimToEmpty(urlParams);
|
||||
while (value.startsWith("?") || value.startsWith("&") || value.startsWith(";")) {
|
||||
value = value.substring(1);
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
private static String trimToNull(String value) {
|
||||
if (value == null) {
|
||||
return null;
|
||||
}
|
||||
String trimmed = value.trim();
|
||||
return trimmed.isEmpty() ? null : trimmed;
|
||||
}
|
||||
|
||||
private static String trimToEmpty(String value) {
|
||||
return value == null ? "" : value.trim();
|
||||
}
|
||||
|
||||
private static Integer intOrNull(ResultSet rs, String column) throws Exception {
|
||||
Object value = rs.getObject(column);
|
||||
return value instanceof Number ? ((Number) value).intValue() : null;
|
||||
}
|
||||
|
||||
private static Integer characterLength(ResultSet rs) throws Exception {
|
||||
String typeName = rs.getString("TYPE_NAME");
|
||||
if (typeName == null) {
|
||||
return null;
|
||||
}
|
||||
String normalized = typeName.toLowerCase(Locale.ROOT);
|
||||
if (!normalized.contains("char") && !normalized.contains("text") && !normalized.contains("string")) {
|
||||
return null;
|
||||
}
|
||||
return intOrNull(rs, "COLUMN_SIZE");
|
||||
}
|
||||
|
||||
private static String optionalComment(ResultSet rs) {
|
||||
return unchecked(() -> {
|
||||
try {
|
||||
String comment = trimToNull(rs.getString(3));
|
||||
return comment;
|
||||
} catch (Exception e) {
|
||||
return null;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
public static void main(String[] args) {
|
||||
new MultiSessionJsonRpcServer(HiveAgent::new).run();
|
||||
}
|
||||
}
|
||||
@@ -1,77 +0,0 @@
|
||||
package com.dbx.agent.hive;
|
||||
|
||||
import com.dbx.agent.DatabaseAgent;
|
||||
import com.dbx.agent.ExecuteQueryOptions;
|
||||
import com.dbx.agent.QueryPageOptions;
|
||||
import com.dbx.agent.QueryResult;
|
||||
import com.dbx.agent.test.JdbcAgentFake;
|
||||
import com.dbx.agent.test.JdbcFakeExecutionBehaviorTest;
|
||||
import com.dbx.agent.test.TestSupport;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.Collections;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
|
||||
class HiveAgentExecutionTest extends JdbcFakeExecutionBehaviorTest {
|
||||
@Override
|
||||
protected DatabaseAgent createAgent() {
|
||||
return new HiveAgent();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected String resultSetSql() {
|
||||
return "MSCK REPAIR TABLE sample";
|
||||
}
|
||||
|
||||
// Override: HiveAgent.executeQuery forces fetchSize=50 to keep Thrift
|
||||
// FetchResults batches small (prevents "Error retrieving next row" on
|
||||
// tables with large binary/struct columns).
|
||||
@Override
|
||||
@Test
|
||||
protected void executesNonSelectStatementsThatReturnResultSets() {
|
||||
DatabaseAgent agent = createAgent();
|
||||
TestSupport.setPrivateConnection(agent, JdbcAgentFake.connection());
|
||||
|
||||
QueryResult result = agent.executeQuery(resultSetSql(), null, new ExecuteQueryOptions());
|
||||
|
||||
assertEquals(Collections.singletonList("VALUE"), result.getColumns());
|
||||
assertEquals(Collections.singletonList(Collections.singletonList("row-value")), result.getRows());
|
||||
assertEquals(Arrays.asList("setMaxRows:10001", "setFetchSize:50", "execute"), JdbcAgentFake.calls);
|
||||
}
|
||||
|
||||
@Override
|
||||
@Test
|
||||
protected void appliesExecutionRowAndFetchLimitsToJdbcStatements() {
|
||||
DatabaseAgent agent = createAgent();
|
||||
TestSupport.setPrivateConnection(agent, JdbcAgentFake.connection());
|
||||
|
||||
QueryResult result = agent.executeQuery(resultSetSql(), null, new ExecuteQueryOptions(1, 25));
|
||||
|
||||
assertEquals(Collections.singletonList(Collections.singletonList("row-value")), result.getRows());
|
||||
assertEquals(Arrays.asList("setMaxRows:2", "setFetchSize:50", "execute"), JdbcAgentFake.calls);
|
||||
}
|
||||
|
||||
@Override
|
||||
@Test
|
||||
protected void pagedQueryExecutionPassesCallerFetchSizeToStatement() {
|
||||
DatabaseAgent agent = createAgent();
|
||||
TestSupport.setPrivateConnection(agent, JdbcAgentFake.connection());
|
||||
|
||||
agent.executeQueryPage(resultSetSql(), null, new QueryPageOptions(100, 500, 1000));
|
||||
|
||||
assertEquals(Arrays.asList("setFetchSize:50", "execute"), JdbcAgentFake.calls);
|
||||
}
|
||||
|
||||
@Override
|
||||
@Test
|
||||
protected void tableReadPassesCallerFetchSizeToStatement() {
|
||||
DatabaseAgent agent = createAgent();
|
||||
TestSupport.setPrivateConnection(agent, JdbcAgentFake.connection());
|
||||
|
||||
agent.startTableRead(resultSetSql(), null, new QueryPageOptions(100, 500, 1000));
|
||||
|
||||
assertEquals(Arrays.asList("setFetchSize:50", "execute"), JdbcAgentFake.calls);
|
||||
}
|
||||
}
|
||||
@@ -1,186 +0,0 @@
|
||||
package com.dbx.agent.hive;
|
||||
|
||||
import com.dbx.agent.test.TestSupport;
|
||||
import com.dbx.agent.test.JdbcMetadataSqlFake;
|
||||
import com.dbx.agent.ColumnInfo;
|
||||
import java.lang.reflect.InvocationHandler;
|
||||
import java.lang.reflect.Method;
|
||||
import java.lang.reflect.Proxy;
|
||||
import java.sql.Connection;
|
||||
import java.sql.DatabaseMetaData;
|
||||
import java.util.List;
|
||||
import java.sql.ResultSet;
|
||||
import java.sql.SQLException;
|
||||
import java.sql.Statement;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.Map;
|
||||
import org.junit.jupiter.api.Assertions;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
class HiveAgentMetadataTest {
|
||||
@Test
|
||||
void quotesSchemaAndTableIdentifiersInMetadataSql() {
|
||||
HiveAgent agent = new HiveAgent();
|
||||
Connection fake = JdbcMetadataSqlFake.connection();
|
||||
TestSupport.setPrivateConnection(agent, fake);
|
||||
|
||||
agent.listTables("bad`schema");
|
||||
agent.getColumns("bad`schema", "bad`table");
|
||||
|
||||
Assertions.assertEquals(
|
||||
List.of(
|
||||
"USE `bad``schema`",
|
||||
"SHOW TABLES",
|
||||
"USE `bad``schema`",
|
||||
"DESCRIBE `bad``table`"
|
||||
),
|
||||
JdbcMetadataSqlFake.statements
|
||||
);
|
||||
}
|
||||
|
||||
@Test
|
||||
void fallsBackToJdbcMetadataWhenDescribeCannotCreateResultSet() {
|
||||
HiveAgent agent = new HiveAgent();
|
||||
Connection fake = describeFailingMetadataConnection();
|
||||
TestSupport.setPrivateConnection(agent, fake);
|
||||
|
||||
List<ColumnInfo> columns = agent.getColumns("default", "events");
|
||||
|
||||
Assertions.assertEquals(2, columns.size());
|
||||
Assertions.assertEquals("id", columns.get(0).getName());
|
||||
Assertions.assertEquals("bigint", columns.get(0).getData_type());
|
||||
Assertions.assertFalse(columns.get(0).getIs_nullable());
|
||||
Assertions.assertEquals("name", columns.get(1).getName());
|
||||
Assertions.assertEquals("string", columns.get(1).getData_type());
|
||||
Assertions.assertEquals(Integer.valueOf(255), columns.get(1).getCharacter_maximum_length());
|
||||
}
|
||||
|
||||
private static Connection describeFailingMetadataConnection() {
|
||||
DatabaseMetaData meta = proxy(DatabaseMetaData.class, (method, args) -> {
|
||||
if ("getColumns".equals(method.getName())) {
|
||||
return rows(
|
||||
row(
|
||||
"COLUMN_NAME", "id",
|
||||
"TYPE_NAME", "bigint",
|
||||
"NULLABLE", DatabaseMetaData.columnNoNulls,
|
||||
"COLUMN_DEF", null,
|
||||
"REMARKS", "identifier",
|
||||
"COLUMN_SIZE", 19,
|
||||
"DECIMAL_DIGITS", 0
|
||||
),
|
||||
row(
|
||||
"COLUMN_NAME", "name",
|
||||
"TYPE_NAME", "string",
|
||||
"NULLABLE", DatabaseMetaData.columnNullable,
|
||||
"COLUMN_DEF", null,
|
||||
"REMARKS", null,
|
||||
"COLUMN_SIZE", 255,
|
||||
"DECIMAL_DIGITS", 0
|
||||
)
|
||||
);
|
||||
}
|
||||
return defaultValue(method.getReturnType());
|
||||
});
|
||||
return proxy(Connection.class, (method, args) -> {
|
||||
String name = method.getName();
|
||||
if ("createStatement".equals(name)) {
|
||||
return describeFailingStatement();
|
||||
}
|
||||
if ("getMetaData".equals(name)) {
|
||||
return meta;
|
||||
}
|
||||
if ("isClosed".equals(name)) {
|
||||
return false;
|
||||
}
|
||||
return defaultValue(method.getReturnType());
|
||||
});
|
||||
}
|
||||
|
||||
private static Statement describeFailingStatement() {
|
||||
return proxy(Statement.class, (method, args) -> {
|
||||
String name = method.getName();
|
||||
if ("execute".equals(name)) {
|
||||
return false;
|
||||
}
|
||||
if ("executeQuery".equals(name)) {
|
||||
throw new SQLException("Could not create ResultSet: Required field 'type' is unset! Struct:TPrimitiveTypeEntry(type:null)");
|
||||
}
|
||||
return defaultValue(method.getReturnType());
|
||||
});
|
||||
}
|
||||
|
||||
@SafeVarargs
|
||||
private static ResultSet rows(Map<String, Object>... rows) {
|
||||
return proxy(ResultSet.class, new MethodHandler() {
|
||||
private int index = -1;
|
||||
|
||||
@Override
|
||||
public Object handle(Method method, Object[] args) {
|
||||
String name = method.getName();
|
||||
if ("next".equals(name)) {
|
||||
index += 1;
|
||||
return index < rows.length;
|
||||
}
|
||||
if ("close".equals(name)) {
|
||||
return null;
|
||||
}
|
||||
Object value = rows[index].get(args[0]);
|
||||
if ("getString".equals(name)) {
|
||||
return value == null ? null : String.valueOf(value);
|
||||
}
|
||||
if ("getObject".equals(name)) {
|
||||
return value;
|
||||
}
|
||||
if ("getInt".equals(name)) {
|
||||
return value instanceof Number ? ((Number) value).intValue() : 0;
|
||||
}
|
||||
return defaultValue(method.getReturnType());
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private static Map<String, Object> row(Object... values) {
|
||||
Map<String, Object> row = new LinkedHashMap<>();
|
||||
for (int i = 0; i < values.length; i += 2) {
|
||||
row.put(String.valueOf(values[i]), values[i + 1]);
|
||||
}
|
||||
return row;
|
||||
}
|
||||
|
||||
private static <T> T proxy(Class<T> type, MethodHandler handler) {
|
||||
InvocationHandler invocationHandler = (proxy, method, args) -> handler.handle(method, args);
|
||||
return type.cast(Proxy.newProxyInstance(type.getClassLoader(), new Class<?>[]{type}, invocationHandler));
|
||||
}
|
||||
|
||||
private static Object defaultValue(Class<?> type) {
|
||||
if (Boolean.TYPE.equals(type)) {
|
||||
return false;
|
||||
}
|
||||
if (Byte.TYPE.equals(type)) {
|
||||
return (byte) 0;
|
||||
}
|
||||
if (Short.TYPE.equals(type)) {
|
||||
return (short) 0;
|
||||
}
|
||||
if (Integer.TYPE.equals(type)) {
|
||||
return 0;
|
||||
}
|
||||
if (Long.TYPE.equals(type)) {
|
||||
return 0L;
|
||||
}
|
||||
if (Float.TYPE.equals(type)) {
|
||||
return 0f;
|
||||
}
|
||||
if (Double.TYPE.equals(type)) {
|
||||
return 0.0d;
|
||||
}
|
||||
if (Character.TYPE.equals(type)) {
|
||||
return '\0';
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
private interface MethodHandler {
|
||||
Object handle(Method method, Object[] args) throws Throwable;
|
||||
}
|
||||
}
|
||||
@@ -1,127 +0,0 @@
|
||||
package com.dbx.agent.hive;
|
||||
|
||||
import com.dbx.agent.ConnectParams;
|
||||
import com.dbx.agent.test.TestSupport;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.lang.reflect.InvocationHandler;
|
||||
import java.lang.reflect.Method;
|
||||
import java.lang.reflect.Proxy;
|
||||
import java.sql.Connection;
|
||||
import java.sql.ResultSet;
|
||||
import java.sql.Statement;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
|
||||
class HiveAgentTest {
|
||||
@Test
|
||||
void hiveJdbcStandaloneRuntimeClassesAreAvailable() throws ClassNotFoundException {
|
||||
Class.forName("org.apache.hive.jdbc.HiveDriver");
|
||||
Class.forName("org.apache.hive.org.apache.thrift.protocol.TProtocol");
|
||||
}
|
||||
|
||||
@Test
|
||||
void buildUrlAppendsKerberosUrlParams() {
|
||||
ConnectParams params = new ConnectParams();
|
||||
params.setHost("hive.example.com");
|
||||
params.setPort(10000);
|
||||
params.setDatabase("default");
|
||||
params.setUrl_params(";principal=hive/hive.example.com@EXAMPLE.COM;auth=kerberos");
|
||||
|
||||
assertEquals(
|
||||
"jdbc:hive2://hive.example.com:10000/default;principal=hive/hive.example.com@EXAMPLE.COM;auth=kerberos",
|
||||
HiveAgent.buildUrl(params)
|
||||
);
|
||||
}
|
||||
|
||||
@Test
|
||||
void buildUrlUsesCustomJdbcUrlAsIs() {
|
||||
ConnectParams params = new ConnectParams();
|
||||
params.setConnection_string(
|
||||
"jdbc:hive2://zk1.example.com,zk2.example.com/default;serviceDiscoveryMode=zooKeeper;zooKeeperNamespace=hiveserver2;principal=hive/_HOST@EXAMPLE.COM"
|
||||
);
|
||||
params.setUrl_params("principal=hive/ignored@EXAMPLE.COM");
|
||||
|
||||
assertEquals(
|
||||
"jdbc:hive2://zk1.example.com,zk2.example.com/default;serviceDiscoveryMode=zooKeeper;zooKeeperNamespace=hiveserver2;principal=hive/_HOST@EXAMPLE.COM",
|
||||
HiveAgent.buildUrl(params)
|
||||
);
|
||||
}
|
||||
|
||||
@Test
|
||||
void getTableDdlUsesHiveShowCreateTable() {
|
||||
HiveAgent agent = new HiveAgent();
|
||||
List<String> queries = new ArrayList<>();
|
||||
String expectedDdl = "CREATE TABLE `hive_test`.`cleaned_data_table`(col string)\n"
|
||||
+ "ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'\n"
|
||||
+ "LOCATION 'hdfs://warehouse/cleaned_data_table'";
|
||||
ResultSet resultSet = proxy(ResultSet.class, new InvocationHandler() {
|
||||
private int row = -1;
|
||||
|
||||
@Override
|
||||
public Object invoke(Object proxy, Method method, Object[] args) {
|
||||
if ("next".equals(method.getName())) {
|
||||
return ++row == 0;
|
||||
}
|
||||
if ("getString".equals(method.getName())) {
|
||||
return expectedDdl;
|
||||
}
|
||||
return defaultValue(method.getReturnType());
|
||||
}
|
||||
});
|
||||
Statement statement = proxy(Statement.class, (proxy, method, args) -> {
|
||||
if ("executeQuery".equals(method.getName())) {
|
||||
queries.add((String) args[0]);
|
||||
return resultSet;
|
||||
}
|
||||
return defaultValue(method.getReturnType());
|
||||
});
|
||||
Connection connection = proxy(Connection.class, (proxy, method, args) -> {
|
||||
if ("createStatement".equals(method.getName())) {
|
||||
return statement;
|
||||
}
|
||||
if ("isClosed".equals(method.getName())) {
|
||||
return false;
|
||||
}
|
||||
return defaultValue(method.getReturnType());
|
||||
});
|
||||
TestSupport.setPrivateConnection(agent, connection);
|
||||
|
||||
assertEquals(expectedDdl + "\n", agent.getTableDdl("hive_test", "cleaned_data_table"));
|
||||
assertEquals(List.of("SHOW CREATE TABLE `hive_test`.`cleaned_data_table`"), queries);
|
||||
}
|
||||
|
||||
private static <T> T proxy(Class<T> type, InvocationHandler handler) {
|
||||
return type.cast(Proxy.newProxyInstance(type.getClassLoader(), new Class<?>[]{type}, handler));
|
||||
}
|
||||
|
||||
private static Object defaultValue(Class<?> type) {
|
||||
if (type == boolean.class) {
|
||||
return false;
|
||||
}
|
||||
if (type == int.class) {
|
||||
return 0;
|
||||
}
|
||||
if (type == long.class) {
|
||||
return 0L;
|
||||
}
|
||||
if (type == float.class) {
|
||||
return 0F;
|
||||
}
|
||||
if (type == double.class) {
|
||||
return 0D;
|
||||
}
|
||||
if (type == byte.class) {
|
||||
return (byte) 0;
|
||||
}
|
||||
if (type == short.class) {
|
||||
return (short) 0;
|
||||
}
|
||||
if (type == char.class) {
|
||||
return '\0';
|
||||
}
|
||||
return null;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user