database/doris
2
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

[improvement](jdbc catalog) remove useless jdbc catalog code (#34986) (#35418)

Browse Source
This commit is contained in:
zy-kkk
2024-05-27 14:25:26 +08:00
committed by GitHub
parent e3b4d4e630
commit 2e20e38523
16 changed files with 154 additions and 1639 deletions
Download Patch File Download Diff File Expand all files Collapse all files

247
be/src/vec/exec/vjdbc_connector.cpp
View File

@ -20,7 +20,6 @@
#include <gen_cpp/Types_types.h>
#include <algorithm>
#include <boost/iterator/iterator_facade.hpp>
// IWYU pragma: no_include <bits/std_abs.h>
#include <cmath> // IWYU pragma: keep
#include <memory>
@ -32,7 +31,6 @@
#include "exec/table_connector.h"
#include "gutil/strings/substitute.h"
#include "jni.h"
#include "runtime/define_primitive_type.h"
#include "runtime/descriptors.h"
#include "runtime/runtime_state.h"
#include "runtime/types.h"
@ -46,14 +44,12 @@
#include "vec/exec/jni_connector.h"
#include "vec/exprs/vexpr.h"
#include "vec/functions/simple_function_factory.h"
#include "vec/io/reader_buffer.h"
namespace doris::vectorized {
const char* JDBC_EXECUTOR_FACTORY_CLASS = "org/apache/doris/jdbc/JdbcExecutorFactory";
const char* JDBC_EXECUTOR_CTOR_SIGNATURE = "([B)V";
const char* JDBC_EXECUTOR_STMT_WRITE_SIGNATURE = "(Ljava/util/Map;)I";
const char* JDBC_EXECUTOR_HAS_NEXT_SIGNATURE = "()Z";
const char* JDBC_EXECUTOR_GET_TYPES_SIGNATURE = "()Ljava/util/List;";
const char* JDBC_EXECUTOR_CLOSE_SIGNATURE = "()V";
const char* JDBC_EXECUTOR_TRANSACTION_SIGNATURE = "()V";
@ -69,11 +65,6 @@ JdbcConnector::~JdbcConnector() {
}
}
#define GET_BASIC_JAVA_CLAZZ(JAVA_TYPE, CPP_TYPE) \
RETURN_IF_ERROR(JniUtil::GetGlobalClassRef(env, JAVA_TYPE, &_executor_##CPP_TYPE##_clazz));
#define DELETE_BASIC_JAVA_CLAZZ_REF(CPP_TYPE) env->DeleteGlobalRef(_executor_##CPP_TYPE##_clazz);
Status JdbcConnector::close(Status /*unused*/) {
SCOPED_RAW_TIMER(&_jdbc_statistic._connector_close_timer);
_closed = true;
@ -88,10 +79,6 @@ Status JdbcConnector::close(Status /*unused*/) {
env->CallNonvirtualVoidMethod(_executor_obj, _executor_clazz, _executor_close_id);
env->DeleteGlobalRef(_executor_factory_clazz);
env->DeleteGlobalRef(_executor_clazz);
DELETE_BASIC_JAVA_CLAZZ_REF(object)
DELETE_BASIC_JAVA_CLAZZ_REF(string)
DELETE_BASIC_JAVA_CLAZZ_REF(list)
#undef DELETE_BASIC_JAVA_CLAZZ_REF
RETURN_IF_ERROR(JniUtil::GetJniExceptionMsg(env));
env->DeleteGlobalRef(_executor_obj);
return Status::OK();
@ -128,9 +115,6 @@ Status JdbcConnector::open(RuntimeState* state, bool read) {
env->DeleteLocalRef(jtable_type);
env->ReleaseStringUTFChars(executor_name, executor_name_str);
env->DeleteLocalRef(executor_name);
GET_BASIC_JAVA_CLAZZ("java/util/List", list)
GET_BASIC_JAVA_CLAZZ("java/lang/Object", object)
GET_BASIC_JAVA_CLAZZ("java/lang/String", string)
#undef GET_BASIC_JAVA_CLAZZ
RETURN_IF_ERROR(_register_func_id(env));
@ -244,9 +228,6 @@ Status JdbcConnector::query() {
}
LOG(INFO) << "JdbcConnector::query has exec success: " << _sql_str;
if (_conn_param.table_type != TOdbcTableType::NEBULA) {
RETURN_IF_ERROR(_check_column_type());
}
return Status::OK();
}
@ -386,12 +367,6 @@ Status JdbcConnector::_register_func_id(JNIEnv* env) {
_executor_get_block_address_id));
RETURN_IF_ERROR(
register_id(_executor_clazz, "getCurBlockRows", "()I", _executor_block_rows_id));
RETURN_IF_ERROR(register_id(_executor_list_clazz, "get", "(I)Ljava/lang/Object;",
_executor_get_list_id));
RETURN_IF_ERROR(register_id(_executor_string_clazz, "getBytes", "(Ljava/lang/String;)[B",
_get_bytes_id));
RETURN_IF_ERROR(
register_id(_executor_object_clazz, "toString", "()Ljava/lang/String;", _to_string_id));
RETURN_IF_ERROR(register_id(_executor_clazz, "openTrans", JDBC_EXECUTOR_TRANSACTION_SIGNATURE,
_executor_begin_trans_id));
@ -399,8 +374,6 @@ Status JdbcConnector::_register_func_id(JNIEnv* env) {
_executor_finish_trans_id));
RETURN_IF_ERROR(register_id(_executor_clazz, "rollbackTrans",
JDBC_EXECUTOR_TRANSACTION_SIGNATURE, _executor_abort_trans_id));
RETURN_IF_ERROR(register_id(_executor_clazz, "getResultColumnTypeNames",
JDBC_EXECUTOR_GET_TYPES_SIGNATURE, _executor_get_types_id));
RETURN_IF_ERROR(
register_id(_executor_clazz, "testConnection", "()V", _executor_test_connection_id));
RETURN_IF_ERROR(
@ -408,206 +381,6 @@ Status JdbcConnector::_register_func_id(JNIEnv* env) {
return Status::OK();
}
Status JdbcConnector::_check_column_type() {
SCOPED_RAW_TIMER(&_jdbc_statistic._check_type_timer);
JNIEnv* env = nullptr;
RETURN_IF_ERROR(JniUtil::GetJNIEnv(&env));
jobject type_lists =
env->CallNonvirtualObjectMethod(_executor_obj, _executor_clazz, _executor_get_types_id);
auto column_size = _tuple_desc->slots().size();
for (int column_index = 0, materialized_column_index = 0; column_index < column_size;
++column_index) {
auto slot_desc = _tuple_desc->slots()[column_index];
if (!slot_desc->is_materialized()) {
continue;
}
jobject column_type =
env->CallObjectMethod(type_lists, _executor_get_list_id, materialized_column_index);
const std::string& type_str = _jobject_to_string(env, column_type);
RETURN_IF_ERROR(_check_type(slot_desc, type_str, column_index));
env->DeleteLocalRef(column_type);
materialized_column_index++;
}
env->DeleteLocalRef(type_lists);
return JniUtil::GetJniExceptionMsg(env);
}
/* type mapping: https://doris.apache.org/zh-CN/docs/dev/ecosystem/external-table/jdbc-of-doris?_highlight=jdbc
Doris MYSQL PostgreSQL Oracle SQLServer
BOOLEAN java.lang.Boolean java.lang.Boolean java.lang.Boolean
TINYINT java.lang.Integer java.lang.Short
SMALLINT java.lang.Integer java.lang.Integer java.math.BigDecimal java.lang.Short
INT java.lang.Integer java.lang.Integer java.math.BigDecimal java.lang.Integer
BIGINT java.lang.Long java.lang.Long java.lang.Long
LARGET java.math.BigInteger
DECIMAL java.math.BigDecimal java.math.BigDecimal java.math.BigDecimal java.math.BigDecimal
VARCHAR java.lang.String java.lang.String java.lang.String java.lang.String
DOUBLE java.lang.Double java.lang.Double java.lang.Double java.lang.Double
FLOAT java.lang.Float java.lang.Float java.lang.Float
DATE java.sql.Date java.sql.Date java.sql.Date
DATETIME java.sql.Timestamp java.sql.Timestamp java.sql.Timestamp java.sql.Timestamp
NOTE: because oracle always use number(p,s) to create all numerical type, so it's java type maybe java.math.BigDecimal
*/
Status JdbcConnector::_check_type(SlotDescriptor* slot_desc, const std::string& type_str,
int column_index) {
const std::string error_msg = fmt::format(
"Fail to convert jdbc type of {} to doris type {} on column: {}. You need to "
"check this column type between external table and doris table.",
type_str, slot_desc->type().debug_string(), slot_desc->col_name());
switch (slot_desc->type().type) {
case TYPE_BOOLEAN: {
if (type_str != "java.lang.Boolean" && type_str != "java.lang.Byte" &&
type_str != "java.lang.Integer") {
return Status::InternalError(error_msg);
}
break;
}
case TYPE_TINYINT:
case TYPE_SMALLINT:
case TYPE_INT: {
if (type_str != "java.lang.Short" && type_str != "java.lang.Integer" &&
type_str != "java.math.BigDecimal" && type_str != "java.lang.Byte" &&
type_str != "com.clickhouse.data.value.UnsignedByte" &&
type_str != "com.clickhouse.data.value.UnsignedShort" && type_str != "java.lang.Long") {
return Status::InternalError(error_msg);
}
break;
}
case TYPE_BIGINT:
case TYPE_LARGEINT: {
if (type_str != "java.lang.Long" && type_str != "java.math.BigDecimal" &&
type_str != "java.math.BigInteger" && type_str != "java.lang.String" &&
type_str != "com.clickhouse.data.value.UnsignedInteger" &&
type_str != "com.clickhouse.data.value.UnsignedLong") {
return Status::InternalError(error_msg);
}
break;
}
case TYPE_FLOAT: {
if (type_str != "java.lang.Float" && type_str != "java.math.BigDecimal") {
return Status::InternalError(error_msg);
}
break;
}
case TYPE_DOUBLE: {
if (type_str != "java.lang.Double" && type_str != "java.math.BigDecimal" &&
type_str != "java.lang.String") {
return Status::InternalError(error_msg);
}
break;
}
case TYPE_CHAR:
case TYPE_VARCHAR:
case TYPE_STRING: {
//now here break directly
break;
}
case TYPE_DATE:
case TYPE_DATEV2:
case TYPE_TIMEV2:
case TYPE_DATETIME:
case TYPE_DATETIMEV2: {
if (type_str != "java.sql.Timestamp" && type_str != "java.time.LocalDateTime" &&
type_str != "java.sql.Date" && type_str != "java.time.LocalDate" &&
type_str != "oracle.sql.TIMESTAMP" && type_str != "java.time.OffsetDateTime" &&
type_str != "java.lang.String") {
return Status::InternalError(error_msg);
}
break;
}
case TYPE_DECIMALV2:
case TYPE_DECIMAL32:
case TYPE_DECIMAL64:
case TYPE_DECIMAL128I:
case TYPE_DECIMAL256: {
if (type_str != "java.math.BigDecimal") {
return Status::InternalError(error_msg);
}
break;
}
case TYPE_ARRAY: {
if (type_str != "java.sql.Array" && type_str != "java.lang.String" &&
type_str != "java.lang.Object") {
return Status::InternalError(error_msg);
}
break;
}
case TYPE_JSONB: {
if (type_str != "java.lang.String" && type_str != "org.postgresql.util.PGobject") {
return Status::InternalError(error_msg);
}
_map_column_idx_to_cast_idx_json[column_index] = _input_json_string_types.size();
if (slot_desc->is_nullable()) {
_input_json_string_types.push_back(make_nullable(std::make_shared<DataTypeString>()));
} else {
_input_json_string_types.push_back(std::make_shared<DataTypeString>());
}
str_json_cols.push_back(
_input_json_string_types[_map_column_idx_to_cast_idx_json[column_index]]
->create_column());
break;
}
case TYPE_HLL: {
if (type_str != "java.lang.String") {
return Status::InternalError(error_msg);
}
_map_column_idx_to_cast_idx_hll[column_index] = _input_hll_string_types.size();
if (slot_desc->is_nullable()) {
_input_hll_string_types.push_back(make_nullable(std::make_shared<DataTypeString>()));
} else {
_input_hll_string_types.push_back(std::make_shared<DataTypeString>());
}
str_hll_cols.push_back(
_input_hll_string_types[_map_column_idx_to_cast_idx_hll[column_index]]
->create_column());
break;
}
case TYPE_OBJECT: {
if (type_str != "java.lang.String") {
return Status::InternalError(error_msg);
}
_map_column_idx_to_cast_idx_bitmap[column_index] = _input_bitmap_string_types.size();
if (slot_desc->is_nullable()) {
_input_bitmap_string_types.push_back(make_nullable(std::make_shared<DataTypeString>()));
} else {
_input_bitmap_string_types.push_back(std::make_shared<DataTypeString>());
}
str_bitmap_cols.push_back(
_input_bitmap_string_types[_map_column_idx_to_cast_idx_bitmap[column_index]]
->create_column());
break;
}
default: {
return Status::InternalError(error_msg);
}
}
return Status::OK();
}
std::string JdbcConnector::_jobject_to_string(JNIEnv* env, jobject jobj) {
jobject jstr = env->CallObjectMethod(jobj, _to_string_id);
auto coding = env->NewStringUTF("UTF-8");
const jbyteArray stringJbytes = (jbyteArray)env->CallObjectMethod(jstr, _get_bytes_id, coding);
size_t length = (size_t)env->GetArrayLength(stringJbytes);
jbyte* pBytes = env->GetByteArrayElements(stringJbytes, nullptr);
std::string str = std::string((char*)pBytes, length);
env->ReleaseByteArrayElements(stringJbytes, pBytes, JNI_ABORT);
env->DeleteLocalRef(stringJbytes);
env->DeleteLocalRef(jstr);
env->DeleteLocalRef(coding);
return str;
}
jobject JdbcConnector::_get_reader_params(Block* block, JNIEnv* env, size_t column_size) {
std::ostringstream columns_nullable;
std::ostringstream columns_replace_string;
@ -687,6 +460,13 @@ Status JdbcConnector::_cast_string_to_special(Block* block, JNIEnv* env, size_t
Status JdbcConnector::_cast_string_to_hll(const SlotDescriptor* slot_desc, Block* block,
int column_index, int rows) {
_map_column_idx_to_cast_idx_hll[column_index] = _input_hll_string_types.size();
if (slot_desc->is_nullable()) {
_input_hll_string_types.push_back(make_nullable(std::make_shared<DataTypeString>()));
} else {
_input_hll_string_types.push_back(std::make_shared<DataTypeString>());
}
DataTypePtr _target_data_type = slot_desc->get_data_type_ptr();
std::string _target_data_type_name = _target_data_type->get_name();
DataTypePtr _cast_param_data_type = _target_data_type;
@ -724,6 +504,13 @@ Status JdbcConnector::_cast_string_to_hll(const SlotDescriptor* slot_desc, Block
Status JdbcConnector::_cast_string_to_bitmap(const SlotDescriptor* slot_desc, Block* block,
int column_index, int rows) {
_map_column_idx_to_cast_idx_bitmap[column_index] = _input_bitmap_string_types.size();
if (slot_desc->is_nullable()) {
_input_bitmap_string_types.push_back(make_nullable(std::make_shared<DataTypeString>()));
} else {
_input_bitmap_string_types.push_back(std::make_shared<DataTypeString>());
}
DataTypePtr _target_data_type = slot_desc->get_data_type_ptr();
std::string _target_data_type_name = _target_data_type->get_name();
DataTypePtr _cast_param_data_type = _target_data_type;
@ -762,6 +549,12 @@ Status JdbcConnector::_cast_string_to_bitmap(const SlotDescriptor* slot_desc, Bl
// Deprecated, this code is retained only for compatibility with query problems that may be encountered when upgrading the version that maps JSON to JSONB to this version, and will be deleted in subsequent versions.
Status JdbcConnector::_cast_string_to_json(const SlotDescriptor* slot_desc, Block* block,
int column_index, int rows) {
_map_column_idx_to_cast_idx_json[column_index] = _input_json_string_types.size();
if (slot_desc->is_nullable()) {
_input_json_string_types.push_back(make_nullable(std::make_shared<DataTypeString>()));
} else {
_input_json_string_types.push_back(std::make_shared<DataTypeString>());
}
DataTypePtr _target_data_type = slot_desc->get_data_type_ptr();
std::string _target_data_type_name = _target_data_type->get_name();
DataTypePtr _cast_param_data_type = _target_data_type;

13
be/src/vec/exec/vjdbc_connector.h
View File

@ -121,9 +121,6 @@ protected:
private:
Status _register_func_id(JNIEnv* env);
Status _check_column_type();
Status _check_type(SlotDescriptor*, const std::string& type_str, int column_index);
std::string _jobject_to_string(JNIEnv* env, jobject jobj);
jobject _get_reader_params(Block* block, JNIEnv* env, size_t column_size);
@ -139,9 +136,6 @@ private:
bool _closed = false;
jclass _executor_factory_clazz;
jclass _executor_clazz;
jclass _executor_list_clazz;
jclass _executor_object_clazz;
jclass _executor_string_clazz;
jobject _executor_obj;
jmethodID _executor_factory_ctor_id;
jmethodID _executor_ctor_id;
@ -150,11 +144,7 @@ private:
jmethodID _executor_has_next_id;
jmethodID _executor_get_block_address_id;
jmethodID _executor_block_rows_id;
jmethodID _executor_get_types_id;
jmethodID _executor_close_id;
jmethodID _executor_get_list_id;
jmethodID _get_bytes_id;
jmethodID _to_string_id;
jmethodID _executor_begin_trans_id;
jmethodID _executor_finish_trans_id;
jmethodID _executor_abort_trans_id;
@ -163,15 +153,12 @@ private:
std::map<int, int> _map_column_idx_to_cast_idx_hll;
std::vector<DataTypePtr> _input_hll_string_types;
std::vector<MutableColumnPtr> str_hll_cols; // for hll type to save data like string
std::map<int, int> _map_column_idx_to_cast_idx_bitmap;
std::vector<DataTypePtr> _input_bitmap_string_types;
std::vector<MutableColumnPtr> str_bitmap_cols; // for bitmap type to save data like string
std::map<int, int> _map_column_idx_to_cast_idx_json;
std::vector<DataTypePtr> _input_json_string_types;
std::vector<MutableColumnPtr> str_json_cols; // for json type to save data like string
JdbcStatistic _jdbc_statistic;
};

4
fe/be-java-extensions/java-common/pom.xml
View File

@ -34,10 +34,6 @@ under the License.
</properties>
<dependencies>
<dependency>
<groupId>com.vesoft</groupId>
<artifactId>client</artifactId>
</dependency>
<dependency>
<groupId>org.apache.doris</groupId>
<artifactId>fe-common</artifactId>

57
fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/utils/UdfUtils.java
View File

@ -24,9 +24,6 @@ import org.apache.doris.catalog.Type;
import org.apache.doris.common.Pair;
import org.apache.doris.common.exception.InternalException;
import com.vesoft.nebula.client.graph.data.DateTimeWrapper;
import com.vesoft.nebula.client.graph.data.DateWrapper;
import com.vesoft.nebula.client.graph.data.ValueWrapper;
import org.apache.log4j.Logger;
import sun.misc.Unsafe;
@ -38,8 +35,6 @@ import java.net.URL;
import java.net.URLClassLoader;
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.Set;
public class UdfUtils {
@ -231,56 +226,4 @@ public class UdfUtils {
}
return bytes;
}
// only used by nebula-graph
// transfer to an object that can copy to the block
public static Object convertObject(ValueWrapper value) {
try {
if (value.isLong()) {
return value.asLong();
}
if (value.isBoolean()) {
return value.asBoolean();
}
if (value.isDouble()) {
return value.asDouble();
}
if (value.isString()) {
return value.asString();
}
if (value.isTime()) {
return value.asTime().toString();
}
if (value.isDate()) {
DateWrapper date = value.asDate();
return LocalDate.of(date.getYear(), date.getMonth(), date.getDay());
}
if (value.isDateTime()) {
DateTimeWrapper dateTime = value.asDateTime();
return LocalDateTime.of(dateTime.getYear(), dateTime.getMonth(), dateTime.getDay(),
dateTime.getHour(), dateTime.getMinute(), dateTime.getSecond(), dateTime.getMicrosec() * 1000);
}
if (value.isVertex()) {
return value.asNode().toString();
}
if (value.isEdge()) {
return value.asRelationship().toString();
}
if (value.isPath()) {
return value.asPath().toString();
}
if (value.isList()) {
return value.asList().toString();
}
if (value.isSet()) {
return value.asSet().toString();
}
if (value.isMap()) {
return value.asMap().toString();
}
return null;
} catch (Exception e) {
return null;
}
}
}

17
fe/be-java-extensions/jdbc-scanner/pom.xml
View File

@ -40,28 +40,11 @@ under the License.
<artifactId>java-common</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>com.oracle.database.jdbc</groupId>
<artifactId>ojdbc8</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>com.zaxxer</groupId>
<artifactId>HikariCP</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>com.clickhouse</groupId>
<artifactId>clickhouse-jdbc</artifactId>
<classifier>all</classifier>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>com.oracle.ojdbc</groupId>
<artifactId>orai18n</artifactId>
<version>19.3.0.0</version>
<scope>provided</scope>
</dependency>
</dependencies>
<build>

12
fe/be-java-extensions/jdbc-scanner/src/main/java/org/apache/doris/jdbc/BaseJdbcExecutor.java
View File

@ -26,7 +26,6 @@ import org.apache.doris.common.jni.vec.VectorColumn;
import org.apache.doris.common.jni.vec.VectorTable;
import org.apache.doris.thrift.TJdbcExecutorCtorParams;
import org.apache.doris.thrift.TJdbcOperation;
import org.apache.doris.thrift.TOdbcTableType;
import com.google.common.base.Preconditions;
import com.zaxxer.hikari.HikariDataSource;
@ -57,14 +56,12 @@ public abstract class BaseJdbcExecutor implements JdbcExecutor {
private static final TBinaryProtocol.Factory PROTOCOL_FACTORY = new TBinaryProtocol.Factory();
private HikariDataSource hikariDataSource = null;
private final byte[] hikariDataSourceLock = new byte[0];
private TOdbcTableType tableType;
private JdbcDataSourceConfig config;
private Connection conn = null;
protected PreparedStatement preparedStatement = null;
protected Statement stmt = null;
protected ResultSet resultSet = null;
protected ResultSetMetaData resultSetMetaData = null;
protected List<String> resultColumnTypeNames = null;
protected List<Object[]> block = null;
protected VectorTable outputTable = null;
protected int batchSizeNum = 0;
@ -78,7 +75,6 @@ public abstract class BaseJdbcExecutor implements JdbcExecutor {
} catch (TException e) {
throw new InternalException(e.getMessage());
}
tableType = request.table_type;
this.config = new JdbcDataSourceConfig()
.setCatalogId(request.catalog_id)
.setJdbcUser(request.jdbc_user)
@ -175,11 +171,7 @@ public abstract class BaseJdbcExecutor implements JdbcExecutor {
resultSet = ((PreparedStatement) stmt).executeQuery();
resultSetMetaData = resultSet.getMetaData();
int columnCount = resultSetMetaData.getColumnCount();
resultColumnTypeNames = new ArrayList<>(columnCount);
block = new ArrayList<>(columnCount);
for (int i = 0; i < columnCount; ++i) {
resultColumnTypeNames.add(resultSetMetaData.getColumnClassName(i + 1));
}
return columnCount;
} catch (SQLException e) {
throw new UdfRuntimeException("JDBC executor sql has error: ", e);
@ -283,10 +275,6 @@ public abstract class BaseJdbcExecutor implements JdbcExecutor {
}
}
public List<String> getResultColumnTypeNames() {
return resultColumnTypeNames;
}
public int getCurBlockRows() {
return curBlockRows;
}

216
fe/be-java-extensions/jdbc-scanner/src/main/java/org/apache/doris/jdbc/ClickHouseJdbcExecutor.java
View File

@ -22,6 +22,8 @@ import org.apache.doris.common.jni.vec.ColumnType.Type;
import org.apache.doris.common.jni.vec.ColumnValueConverter;
import org.apache.doris.common.jni.vec.VectorTable;
import com.google.common.collect.Lists;
import java.lang.reflect.Array;
import java.math.BigDecimal;
import java.math.BigInteger;
@ -30,11 +32,7 @@ import java.sql.SQLException;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.function.Function;
import java.util.stream.Collectors;
public class ClickHouseJdbcExecutor extends BaseJdbcExecutor {
@ -89,7 +87,7 @@ public class ClickHouseJdbcExecutor extends BaseJdbcExecutor {
case STRING:
return resultSet.getObject(columnIndex + 1, String.class);
case ARRAY:
return resultSet.getObject(columnIndex + 1);
return convertArrayToList(resultSet.getArray(columnIndex + 1).getArray());
default:
throw new IllegalArgumentException("Unsupported column type: " + type.getType());
}
@ -99,114 +97,128 @@ public class ClickHouseJdbcExecutor extends BaseJdbcExecutor {
protected ColumnValueConverter getOutputConverter(ColumnType columnType, String replaceString) {
if (columnType.getType() == Type.ARRAY) {
return createConverter(
(Object input) -> convertArray(input, columnType.getChildTypes().get(0)),
(Object input) -> convertArray((List<?>) input, columnType.getChildTypes().get(0)),
List.class);
} else {
return null;
}
}
private <T, U> List<U> convertArray(T[] input, Function<T, U> converter) {
if (input == null) {
return Collections.emptyList();
private List<Object> convertArrayToList(Object array) {
if (array == null) {
return null;
}
return Arrays.stream(input)
.map(converter)
.collect(Collectors.toList());
}
private List<?> convertArray(Object input, ColumnType childType) {
if (input == null) {
return Collections.emptyList();
}
if (childType.isArray()) {
ColumnType subType = childType.getChildTypes().get(0);
Object[] array = (Object[]) input;
List<Object> convertedList = new ArrayList<>();
for (Object subArray : array) {
convertedList.add(convertArray(subArray, subType));
}
return convertedList;
}
if (input instanceof Object[]) {
Object[] arrayInput = (Object[]) input;
switch (childType.getType()) {
case SMALLINT:
return input instanceof Byte[]
? convertArray((Byte[]) input,
byteValue -> byteValue != null ? (short) (byte) byteValue : null)
: convertArray((Number[]) arrayInput,
number -> number != null ? number.shortValue() : null);
case INT:
return input instanceof Short[]
? convertArray((Short[]) input,
shortValue -> shortValue != null ? (int) (short) shortValue : null)
: convertArray((Number[]) arrayInput, number -> number != null ? number.intValue() : null);
case BIGINT:
return input instanceof Integer[]
? convertArray((Integer[]) input,
intValue -> intValue != null ? (long) (int) intValue : null)
: convertArray((Number[]) arrayInput, number -> number != null ? number.longValue() : null);
case LARGEINT:
return input instanceof Long[]
? convertArray((Long[]) input,
longValue -> longValue != null ? BigInteger.valueOf(longValue) : null)
: convertArray((Number[]) arrayInput,
number -> number != null ? BigInteger.valueOf(number.longValue()) : null);
case STRING:
if (input instanceof InetAddress[]) {
return convertArray((InetAddress[]) input,
inetAddress -> inetAddress != null ? inetAddress.getHostAddress() : null);
} else {
return convertArray(arrayInput, element -> element != null ? element.toString() : null);
}
default:
return Arrays.asList(arrayInput);
}
} else {
return convertPrimitiveArray(input, childType);
}
}
private List<?> convertPrimitiveArray(Object input, ColumnType childType) {
int length = Array.getLength(input);
int length = Array.getLength(array);
List<Object> list = new ArrayList<>(length);
for (int i = 0; i < length; i++) {
Object element = Array.get(input, i);
switch (childType.getType()) {
case SMALLINT:
if (input instanceof byte[]) {
list.add((short) (byte) element);
} else {
list.add(element);
}
break;
case INT:
if (input instanceof short[]) {
list.add((int) (short) element);
} else {
list.add(element);
}
break;
case BIGINT:
if (input instanceof int[]) {
list.add((long) (int) element);
} else {
list.add(element);
}
break;
case LARGEINT:
if (input instanceof long[]) {
list.add(BigInteger.valueOf((long) element));
} else {
list.add(element);
}
break;
default:
list.add(element);
break;
}
Object element = Array.get(array, i);
list.add(element);
}
return list;
}
private List<?> convertArray(List<?> array, ColumnType type) {
if (array == null) {
return null;
}
switch (type.getType()) {
case SMALLINT: {
List<Short> result = Lists.newArrayList();
for (Object element : array) {
if (element == null) {
result.add(null);
} else {
if (element instanceof Byte) {
result.add(((Byte) element).shortValue());
} else if (element instanceof Number) {
result.add(((Number) element).shortValue());
} else {
throw new IllegalArgumentException("Unsupported element type: " + element.getClass());
}
}
}
return result;
}
case INT: {
List<Integer> result = Lists.newArrayList();
for (Object element : array) {
if (element == null) {
result.add(null);
} else {
if (element instanceof Short) {
result.add(((Short) element).intValue());
} else if (element instanceof Number) {
result.add(((Number) element).intValue());
} else {
throw new IllegalArgumentException("Unsupported element type: " + element.getClass());
}
}
}
return result;
}
case BIGINT: {
List<Long> result = Lists.newArrayList();
for (Object element : array) {
if (element == null) {
result.add(null);
} else {
if (element instanceof Integer) {
result.add(((Integer) element).longValue());
} else if (element instanceof Number) {
result.add(((Number) element).longValue());
} else {
throw new IllegalArgumentException("Unsupported element type: " + element.getClass());
}
}
}
return result;
}
case LARGEINT: {
List<BigInteger> result = Lists.newArrayList();
for (Object element : array) {
if (element == null) {
result.add(null);
} else {
if (element instanceof BigDecimal) {
result.add(((BigDecimal) element).toBigInteger());
} else if (element instanceof Number) {
result.add(BigInteger.valueOf(((Number) element).longValue()));
} else {
throw new IllegalArgumentException("Unsupported element type: " + element.getClass());
}
}
}
return result;
}
case STRING: {
List<String> result = Lists.newArrayList();
for (Object element : array) {
if (element == null) {
result.add(null);
} else if (element instanceof InetAddress) {
result.add(((InetAddress) element).getHostAddress());
} else {
result.add(element.toString());
}
}
return result;
}
case ARRAY:
List<List<?>> resultArray = Lists.newArrayList();
for (Object element : array) {
if (element == null) {
resultArray.add(null);
} else {
resultArray.add(
Lists.newArrayList(convertArray((List<?>) element, type.getChildTypes().get(0))));
}
}
return resultArray;
default:
return array;
}
}
}

1123
fe/be-java-extensions/jdbc-scanner/src/main/java/org/apache/doris/jdbc/DefaultJdbcExecutor.java
View File

File diff suppressed because it is too large Load Diff

2
fe/be-java-extensions/jdbc-scanner/src/main/java/org/apache/doris/jdbc/JdbcExecutorFactory.java
View File

@ -42,7 +42,7 @@ public class JdbcExecutorFactory {
case PRESTO:
return "org/apache/doris/jdbc/TrinoJdbcExecutor";
default:
return "org/apache/doris/jdbc/DefaultJdbcExecutor";
throw new IllegalArgumentException("Unsupported jdbc type: " + type);
}
}
}

15
fe/be-java-extensions/preload-extensions/pom.xml
View File

@ -199,25 +199,10 @@ under the License.
</exclusions>
</dependency>
<!-- For JDBC Scanner PreLoad-->
<dependency>
<groupId>com.oracle.database.jdbc</groupId>
<artifactId>ojdbc8</artifactId>
</dependency>
<dependency>
<groupId>com.zaxxer</groupId>
<artifactId>HikariCP</artifactId>
</dependency>
<dependency>
<groupId>com.clickhouse</groupId>
<artifactId>clickhouse-jdbc</artifactId>
<classifier>all</classifier>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>com.oracle.ojdbc</groupId>
<artifactId>orai18n</artifactId>
<version>19.3.0.0</version>
</dependency>
<dependency>
<groupId>org.apache.doris</groupId>
<artifactId>hive-catalog-shade</artifactId>

4
fe/fe-core/src/main/java/org/apache/doris/catalog/JdbcResource.java
View File

@ -65,7 +65,6 @@ import java.util.Map;
public class JdbcResource extends Resource {
private static final Logger LOG = LogManager.getLogger(JdbcResource.class);
public static final String JDBC_NEBULA = "jdbc:nebula";
public static final String JDBC_MYSQL = "jdbc:mysql";
public static final String JDBC_MARIADB = "jdbc:mariadb";
public static final String JDBC_POSTGRESQL = "jdbc:postgresql";
@ -78,7 +77,6 @@ public class JdbcResource extends Resource {
public static final String JDBC_OCEANBASE = "jdbc:oceanbase";
public static final String JDBC_DB2 = "jdbc:db2";
public static final String NEBULA = "NEBULA";
public static final String MYSQL = "MYSQL";
public static final String POSTGRESQL = "POSTGRESQL";
public static final String ORACLE = "ORACLE";
@ -323,8 +321,6 @@ public class JdbcResource extends Resource {
return PRESTO;
} else if (url.startsWith(JDBC_OCEANBASE)) {
return OCEANBASE;
} else if (url.startsWith(JDBC_NEBULA)) {
return NEBULA;
} else if (url.startsWith(JDBC_DB2)) {
return DB2;
}

1
fe/fe-core/src/main/java/org/apache/doris/catalog/JdbcTable.java
View File

@ -92,7 +92,6 @@ public class JdbcTable extends Table {
static {
Map<String, TOdbcTableType> tempMap = new CaseInsensitiveMap();
tempMap.put("nebula", TOdbcTableType.NEBULA);
tempMap.put("mysql", TOdbcTableType.MYSQL);
tempMap.put("postgresql", TOdbcTableType.POSTGRESQL);
tempMap.put("sqlserver", TOdbcTableType.SQLSERVER);

23
fe/fe-core/src/main/java/org/apache/doris/datasource/jdbc/source/JdbcScanNode.java
View File

@ -98,7 +98,6 @@ public class JdbcScanNode extends ExternalScanNode {
@Override
public void init(Analyzer analyzer) throws UserException {
super.init(analyzer);
getGraphQueryString();
}
/**
@ -112,25 +111,6 @@ public class JdbcScanNode extends ExternalScanNode {
cardinality = (long) statsDeriveResult.getRowCount();
}
private boolean isNebula() {
return jdbcType == TOdbcTableType.NEBULA;
}
private void getGraphQueryString() {
if (!isNebula()) {
return;
}
for (Expr expr : conjuncts) {
FunctionCallExpr functionCallExpr = (FunctionCallExpr) expr;
if ("g".equals(functionCallExpr.getFnName().getFunction())) {
graphQueryString = functionCallExpr.getChild(0).getStringValue();
break;
}
}
// clean conjusts cause graph sannnode no need conjuncts
conjuncts = Lists.newArrayList();
}
private void createJdbcFilters() {
if (conjuncts.isEmpty()) {
return;
@ -194,9 +174,6 @@ public class JdbcScanNode extends ExternalScanNode {
}
private String getJdbcQueryStr() {
if (isNebula()) {
return graphQueryString;
}
StringBuilder sql = new StringBuilder("SELECT ");
// Oracle use the where clause to do top n

20
fe/pom.xml
View File

@ -259,7 +259,6 @@ under the License.
<json-simple.version>1.1.1</json-simple.version>
<junit.version>5.8.2</junit.version>
<hikaricp.version>4.0.3</hikaricp.version>
<clickhouse.version>0.4.6</clickhouse.version>
<thrift.version>0.16.0</thrift.version>
<tomcat-embed-core.version>8.5.86</tomcat-embed-core.version>
<log4j2.version>2.18.0</log4j2.version>
@ -335,7 +334,6 @@ under the License.
<project.scm.id>github</project.scm.id>
<spring.version>2.7.13</spring.version>
<orc.version>1.8.4</orc.version>
<ojdbc8.version>12.2.0.1</ojdbc8.version>
<zookeeper.version>3.9.1</zookeeper.version>
<velocity-engine-core.version>2.3</velocity-engine-core.version>
<ranger-plugins-common.version>2.4.0</ranger-plugins-common.version>
@ -345,7 +343,6 @@ under the License.
<jettison.version>1.5.4</jettison.version>
<jetty.version>9.4.53.v20231009</jetty.version>
<immutables.version>2.9.3</immutables.version>
<vesoft.client.version>3.0.0</vesoft.client.version>
<!--todo waiting release-->
<quartz.version>2.3.2</quartz.version>
<!-- paimon -->
@ -741,11 +738,6 @@ under the License.
<artifactId>json-simple</artifactId>
<version>${json-simple.version}</version>
</dependency>
<dependency>
<groupId>com.oracle.database.jdbc</groupId>
<artifactId>ojdbc8</artifactId>
<version>${ojdbc8.version}</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.junit.jupiter/junit-jupiter-engine -->
<dependency>
<groupId>org.junit.jupiter</groupId>
@ -1536,13 +1528,6 @@ under the License.
<artifactId>HikariCP</artifactId>
<version>${hikaricp.version}</version>
</dependency>
<dependency>
<groupId>com.clickhouse</groupId>
<artifactId>clickhouse-jdbc</artifactId>
<version>${clickhouse.version}</version>
<scope>provided</scope>
<classifier>all</classifier>
</dependency>
<!-- https://mvnrepository.com/artifact/joda-time/joda-time -->
<dependency>
<groupId>joda-time</groupId>
@ -1555,11 +1540,6 @@ under the License.
<artifactId>tomcat-embed-core</artifactId>
<version>${tomcat-embed-core.version}</version>
</dependency>
<dependency>
<groupId>com.vesoft</groupId>
<artifactId>client</artifactId>
<version>${vesoft.client.version}</version>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-netty</artifactId>

1
gensrc/thrift/Types.thrift
View File

@ -399,7 +399,6 @@ enum TOdbcTableType {
PRESTO,
OCEANBASE,
OCEANBASE_ORACLE,
NEBULA,
DB2
}

38
regression-test/suites/external_table_p0/jdbc/test_jdbc_query_mysql.groovy
View File

@ -949,25 +949,25 @@ suite("test_jdbc_query_mysql", "p0,external,mysql,external_docker,external_docke
}
sql """alter resource $jdbcResourceMysql57 properties("password" = "123456")"""
// test for type check
sql """ drop table if exists ${exMysqlTypeTable} """
sql """
CREATE EXTERNAL TABLE ${exMysqlTypeTable} (
`id` bigint NOT NULL,
`count_value` varchar(100) NULL
) ENGINE=JDBC
COMMENT "JDBC Mysql 外部表"
PROPERTIES (
"resource" = "$jdbcResourceMysql57",
"table" = "ex_tb2",
"table_type"="mysql"
);
"""
test {
sql """select * from ${exMysqlTypeTable} order by id"""
exception "Fail to convert jdbc type of java.lang.Integer to doris type BIGINT on column: id"
}
// // test for type check
// sql """ drop table if exists ${exMysqlTypeTable} """
// sql """
// CREATE EXTERNAL TABLE ${exMysqlTypeTable} (
// `id` bigint NOT NULL,
// `count_value` varchar(100) NULL
// ) ENGINE=JDBC
// COMMENT "JDBC Mysql 外部表"
// PROPERTIES (
// "resource" = "$jdbcResourceMysql57",
// "table" = "ex_tb2",
// "table_type"="mysql"
// );
// """
//
// test {
// sql """select * from ${exMysqlTypeTable} order by id"""
// exception "Fail to convert jdbc type of java.lang.Integer to doris type BIGINT on column: id"
// }
}
}