[Fix][Connector-V2]Oceanbase vector database is added as the source server (#7832)

xxsc0529 · Hisoka-X · web-flow · commit 258f9317656c · 2024-10-17T10:35:39.000+08:00
Co-authored-by: Jia Fan &lt;fanjiaeminem@qq.com&gt;
diff --git a/docs/en/connector-v2/sink/Jdbc.md b/docs/en/connector-v2/sink/Jdbc.md
@@ -245,7 +245,7 @@ there are some reference value for params above.
 | Snowflake         | net.snowflake.client.jdbc.SnowflakeDriver    | jdbc&#58;snowflake://<account_name>.snowflakecomputing.com         | /                                                  | https://mvnrepository.com/artifact/net.snowflake/snowflake-jdbc                                                               |
 | Vertica           | com.vertica.jdbc.Driver                      | jdbc:vertica://localhost:5433                                      | /                                                  | https://repo1.maven.org/maven2/com/vertica/jdbc/vertica-jdbc/12.0.3-0/vertica-jdbc-12.0.3-0.jar                               |
 | Kingbase          | com.kingbase8.Driver                         | jdbc:kingbase8://localhost:54321/db_test                           | /                                                  | https://repo1.maven.org/maven2/cn/com/kingbase/kingbase8/8.6.0/kingbase8-8.6.0.jar                                            |
-| OceanBase         | com.oceanbase.jdbc.Driver                    | jdbc:oceanbase://localhost:2881                                    | /                                                  | https://repo1.maven.org/maven2/com/oceanbase/oceanbase-client/2.4.11/oceanbase-client-2.4.11.jar                              |
+| OceanBase         | com.oceanbase.jdbc.Driver                    | jdbc:oceanbase://localhost:2881                                    | /                                                  | https://repo1.maven.org/maven2/com/oceanbase/oceanbase-client/2.4.12/oceanbase-client-2.4.12.jar                              |
 | xugu              | com.xugu.cloudjdbc.Driver                    | jdbc:xugu://localhost:5138                                         | /                                                  | https://repo1.maven.org/maven2/com/xugudb/xugu-jdbc/12.2.0/xugu-jdbc-12.2.0.jar                                               |
 | InterSystems IRIS | com.intersystems.jdbc.IRISDriver             | jdbc:IRIS://localhost:1972/%SYS                                    | /                                                  | https://raw.githubusercontent.com/intersystems-community/iris-driver-distribution/main/JDBC/JDK18/intersystems-jdbc-3.8.4.jar |
 | opengauss         | org.opengauss.Driver                         | jdbc:opengauss://localhost:5432/postgres                           | /                                                  | https://repo1.maven.org/maven2/org/opengauss/opengauss-jdbc/5.1.0-og/opengauss-jdbc-5.1.0-og.jar                              |
diff --git a/docs/en/connector-v2/source/Jdbc.md b/docs/en/connector-v2/source/Jdbc.md
@@ -133,7 +133,7 @@ there are some reference value for params above.
 | Redshift          | com.amazon.redshift.jdbc42.Driver                   | jdbc:redshift://localhost:5439/testdb?defaultRowFetchSize=1000         | https://mvnrepository.com/artifact/com.amazon.redshift/redshift-jdbc42                                                        |
 | Vertica           | com.vertica.jdbc.Driver                             | jdbc:vertica://localhost:5433                                          | https://repo1.maven.org/maven2/com/vertica/jdbc/vertica-jdbc/12.0.3-0/vertica-jdbc-12.0.3-0.jar                               |
 | Kingbase          | com.kingbase8.Driver                                | jdbc:kingbase8://localhost:54321/db_test                               | https://repo1.maven.org/maven2/cn/com/kingbase/kingbase8/8.6.0/kingbase8-8.6.0.jar                                            |
-| OceanBase         | com.oceanbase.jdbc.Driver                           | jdbc:oceanbase://localhost:2881                                        | https://repo1.maven.org/maven2/com/oceanbase/oceanbase-client/2.4.11/oceanbase-client-2.4.11.jar                              |
+| OceanBase         | com.oceanbase.jdbc.Driver                           | jdbc:oceanbase://localhost:2881                                        | https://repo1.maven.org/maven2/com/oceanbase/oceanbase-client/2.4.12/oceanbase-client-2.4.12.jar                              |
 | Hive              | org.apache.hive.jdbc.HiveDriver                     | jdbc:hive2://localhost:10000                                           | https://repo1.maven.org/maven2/org/apache/hive/hive-jdbc/3.1.3/hive-jdbc-3.1.3-standalone.jar                                 |
 | xugu              | com.xugu.cloudjdbc.Driver                           | jdbc:xugu://localhost:5138                                             | https://repo1.maven.org/maven2/com/xugudb/xugu-jdbc/12.2.0/xugu-jdbc-12.2.0.jar                                               |
 | InterSystems IRIS | com.intersystems.jdbc.IRISDriver                    | jdbc:IRIS://localhost:1972/%SYS                                        | https://raw.githubusercontent.com/intersystems-community/iris-driver-distribution/main/JDBC/JDK18/intersystems-jdbc-3.8.4.jar |
diff --git a/docs/zh/connector-v2/sink/Jdbc.md b/docs/zh/connector-v2/sink/Jdbc.md
@@ -235,7 +235,7 @@ Sink插件常用参数，请参考 [Sink常用选项](../sink-common-options.md)
 | Snowflake  | net.snowflake.client.jdbc.SnowflakeDriver    | jdbc&#58;snowflake://<account_name>.snowflakecomputing.com         | /                                                  | https://mvnrepository.com/artifact/net.snowflake/snowflake-jdbc                                      |
 | Vertica    | com.vertica.jdbc.Driver                      | jdbc:vertica://localhost:5433                                      | /                                                  | https://repo1.maven.org/maven2/com/vertica/jdbc/vertica-jdbc/12.0.3-0/vertica-jdbc-12.0.3-0.jar      |
 | Kingbase   | com.kingbase8.Driver                         | jdbc:kingbase8://localhost:54321/db_test                           | /                                                  | https://repo1.maven.org/maven2/cn/com/kingbase/kingbase8/8.6.0/kingbase8-8.6.0.jar                   |
-| OceanBase  | com.oceanbase.jdbc.Driver                    | jdbc:oceanbase://localhost:2881                                    | /                                                  | https://repo1.maven.org/maven2/com/oceanbase/oceanbase-client/2.4.11/oceanbase-client-2.4.11.jar     |
+| OceanBase  | com.oceanbase.jdbc.Driver                    | jdbc:oceanbase://localhost:2881                                    | /                                                  | https://repo1.maven.org/maven2/com/oceanbase/oceanbase-client/2.4.12/oceanbase-client-2.4.12.jar     |
 | opengauss  | org.opengauss.Driver                         | jdbc:opengauss://localhost:5432/postgres                           | /                                                  | https://repo1.maven.org/maven2/org/opengauss/opengauss-jdbc/5.1.0-og/opengauss-jdbc-5.1.0-og.jar     |
 
 ## 示例
diff --git a/seatunnel-connectors-v2/connector-jdbc/pom.xml b/seatunnel-connectors-v2/connector-jdbc/pom.xml
@@ -49,7 +49,7 @@
         <postgis.jdbc.version>2.5.1</postgis.jdbc.version>
         <kingbase8.version>8.6.0</kingbase8.version>
         <hive.jdbc.version>3.1.3</hive.jdbc.version>
-        <oceanbase.jdbc.version>2.4.11</oceanbase.jdbc.version>
+        <oceanbase.jdbc.version>2.4.12</oceanbase.jdbc.version>
         <xugu.jdbc.version>12.2.0</xugu.jdbc.version>
         <iris.jdbc.version>3.0.0</iris.jdbc.version>
         <tikv.version>3.2.0</tikv.version>
diff --git a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/oceanbase/OceanBaseMySqlCatalog.java b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/oceanbase/OceanBaseMySqlCatalog.java
@@ -20,26 +20,32 @@
 import org.apache.seatunnel.api.table.catalog.CatalogTable;
 import org.apache.seatunnel.api.table.catalog.Column;
 import org.apache.seatunnel.api.table.catalog.ConstraintKey;
+import org.apache.seatunnel.api.table.catalog.PrimaryKey;
 import org.apache.seatunnel.api.table.catalog.TableIdentifier;
 import org.apache.seatunnel.api.table.catalog.TablePath;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
 import org.apache.seatunnel.api.table.catalog.exception.CatalogException;
 import org.apache.seatunnel.api.table.converter.BasicTypeDefine;
 import org.apache.seatunnel.common.utils.JdbcUrlUtil;
 import org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.AbstractJdbcCatalog;
-import org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.utils.CatalogUtils;
 import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.oceanbase.OceanBaseMySqlTypeConverter;
-import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.oceanbase.OceanBaseMySqlTypeMapper;
 import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.oceanbase.OceanBaseMysqlType;
 
+import org.apache.commons.lang.StringUtils;
+
 import com.google.common.base.Preconditions;
 import lombok.extern.slf4j.Slf4j;
 
 import java.sql.Connection;
 import java.sql.DatabaseMetaData;
+import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.ResultSetMetaData;
 import java.sql.SQLException;
 import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Locale;
@@ -197,12 +203,54 @@ protected String getDropDatabaseSql(String databaseName) {
 
     @Override
     public CatalogTable getTable(String sqlQuery) throws SQLException {
-        Connection defaultConnection = getConnection(defaultUrl);
-        try (Statement statement = defaultConnection.createStatement();
-                ResultSet resultSet = statement.executeQuery(sqlQuery)) {
-            ResultSetMetaData metaData = resultSet.getMetaData();
-            return CatalogUtils.getCatalogTable(
-                    metaData, new OceanBaseMySqlTypeMapper(typeConverter), sqlQuery);
+        try (Connection connection = getConnection(defaultUrl)) {
+            String tableName = null;
+            String databaseName = null;
+            String schemaName = null;
+            String catalogName = "jdbc_catalog";
+            TableSchema.Builder schemaBuilder = TableSchema.builder();
+
+            try (Statement statement = connection.createStatement();
+                    ResultSet resultSet = statement.executeQuery(sqlQuery)) {
+                ResultSetMetaData metaData = resultSet.getMetaData();
+                tableName = metaData.getTableName(1);
+                databaseName = metaData.getCatalogName(1);
+                schemaName = metaData.getSchemaName(1);
+                catalogName = metaData.getCatalogName(1);
+            }
+            databaseName = StringUtils.defaultIfBlank(databaseName, null);
+            schemaName = StringUtils.defaultIfBlank(schemaName, null);
+
+            TablePath tablePath =
+                    StringUtils.isBlank(tableName)
+                            ? TablePath.DEFAULT
+                            : TablePath.of(databaseName, schemaName, tableName);
+
+            try (PreparedStatement ps =
+                            connection.prepareStatement(getSelectColumnsSql(tablePath));
+                    ResultSet columnResultSet = ps.executeQuery();
+                    ResultSet primaryKeys =
+                            connection
+                                    .getMetaData()
+                                    .getPrimaryKeys(catalogName, schemaName, tableName)) {
+                while (primaryKeys.next()) {
+                    String primaryKeyColumnName = primaryKeys.getString("COLUMN_NAME");
+                    schemaBuilder.primaryKey(
+                            PrimaryKey.of(
+                                    primaryKeyColumnName,
+                                    Collections.singletonList(primaryKeyColumnName)));
+                }
+                while (columnResultSet.next()) {
+                    schemaBuilder.column(buildColumn(columnResultSet));
+                }
+            }
+            return CatalogTable.of(
+                    TableIdentifier.of(catalogName, tablePath),
+                    schemaBuilder.build(),
+                    new HashMap<>(),
+                    new ArrayList<>(),
+                    "",
+                    catalogName);
         }
     }
 
diff --git a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oceanbase/OceanBaseMySqlTypeConverter.java b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oceanbase/OceanBaseMySqlTypeConverter.java
@@ -25,6 +25,7 @@
 import org.apache.seatunnel.api.table.type.DecimalType;
 import org.apache.seatunnel.api.table.type.LocalTimeType;
 import org.apache.seatunnel.api.table.type.PrimitiveByteArrayType;
+import org.apache.seatunnel.api.table.type.VectorType;
 import org.apache.seatunnel.common.exception.CommonError;
 import org.apache.seatunnel.connectors.seatunnel.common.source.TypeDefineUtils;
 import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.DatabaseIdentifier;
@@ -100,6 +101,9 @@ public class OceanBaseMySqlTypeConverter
     public static final long POWER_2_32 = (long) Math.pow(2, 32);
     public static final long MAX_VARBINARY_LENGTH = POWER_2_16 - 4;
 
+    private static final String VECTOR_TYPE_NAME = "";
+    private static final String VECTOR_NAME = "VECTOR";
+
     @Override
     public String identifier() {
         return DatabaseIdentifier.OCENABASE;
@@ -289,6 +293,17 @@ public Column convert(BasicTypeDefine typeDefine) {
                 builder.dataType(LocalTimeType.LOCAL_DATE_TIME_TYPE);
                 builder.scale(typeDefine.getScale());
                 break;
+            case VECTOR_TYPE_NAME:
+                String columnType = typeDefine.getColumnType();
+                if (columnType.startsWith("vector(") && columnType.endsWith(")")) {
+                    Integer number =
+                            Integer.parseInt(
+                                    columnType.substring(
+                                            columnType.indexOf("(") + 1, columnType.indexOf(")")));
+                    builder.dataType(VectorType.VECTOR_FLOAT_TYPE);
+                    builder.scale(number);
+                }
+                break;
             default:
                 throw CommonError.convertToSeaTunnelTypeError(
                         DatabaseIdentifier.OCENABASE, mysqlDataType, typeDefine.getName());
@@ -501,6 +516,11 @@ public BasicTypeDefine<OceanBaseMysqlType> reconvert(Column column) {
                     builder.columnType(MYSQL_DATETIME);
                 }
                 break;
+            case FLOAT_VECTOR:
+                builder.nativeType(VECTOR_NAME);
+                builder.columnType(String.format("%s(%s)", VECTOR_NAME, column.getScale()));
+                builder.dataType(VECTOR_NAME);
+                break;
             default:
                 throw CommonError.convertToConnectorTypeError(
                         DatabaseIdentifier.OCENABASE,
diff --git a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oceanbase/OceanBaseMysqlJdbcRowConverter.java b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oceanbase/OceanBaseMysqlJdbcRowConverter.java
@@ -32,6 +32,8 @@
 import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.DatabaseIdentifier;
 import org.apache.seatunnel.connectors.seatunnel.jdbc.utils.JdbcFieldTypeUtils;
 
+import org.apache.commons.lang3.StringUtils;
+
 import java.math.BigDecimal;
 import java.nio.ByteBuffer;
 import java.sql.Date;
@@ -89,12 +91,16 @@ public SeaTunnelRow toInternal(ResultSet rs, TableSchema tableSchema) throws SQL
                     fields[fieldIndex] = JdbcFieldTypeUtils.getFloat(rs, resultSetIndex);
                     break;
                 case FLOAT_VECTOR:
-                    Object[] objects = (Object[]) rs.getObject(fieldIndex);
-                    Float[] arrays = new Float[objects.length];
-                    for (int i = 0; i < objects.length; i++) {
-                        arrays[i] = Float.parseFloat(objects[i].toString());
+                    String result = JdbcFieldTypeUtils.getString(rs, resultSetIndex);
+                    if (StringUtils.isNotBlank(result)) {
+                        result = result.replace("[", "").replace("]", "");
+                        String[] stringArray = result.split(",");
+                        Float[] arrays = new Float[stringArray.length];
+                        for (int i = 0; i < stringArray.length; i++) {
+                            arrays[i] = Float.parseFloat(stringArray[i]);
+                        }
+                        fields[fieldIndex] = BufferUtils.toByteBuffer(arrays);
                     }
-                    fields[fieldIndex] = BufferUtils.toByteBuffer(arrays);
                     break;
                 case DOUBLE:
                     fields[fieldIndex] = JdbcFieldTypeUtils.getDouble(rs, resultSetIndex);
diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-2/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcOceanBaseITBase.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-2/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcOceanBaseITBase.java
@@ -87,6 +87,6 @@ void checkResult(String executeKey, TestContainer container, Container.ExecResul
 
     @Override
     String driverUrl() {
-        return "https://repo1.maven.org/maven2/com/oceanbase/oceanbase-client/2.4.11/oceanbase-client-2.4.11.jar";
+        return "https://repo1.maven.org/maven2/com/oceanbase/oceanbase-client/2.4.12/oceanbase-client-2.4.12.jar";
     }
 }
diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-2/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcOceanBaseMilvusIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-2/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcOceanBaseMilvusIT.java
diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-2/src/test/resources/jdbc_oceanbase_source_and_milvus_sink.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-2/src/test/resources/jdbc_oceanbase_source_and_milvus_sink.conf

Original file line number	Diff line number	Diff line change
`@@ -87,6 +87,6 @@ void checkResult(String executeKey, TestContainer container, Container.ExecResul`
`87`	`87`
`88`	`88`	`@Override`
`89`	`89`	`String driverUrl() {`
`90`		`- return "https://repo1.maven.org/maven2/com/oceanbase/oceanbase-client/2.4.11/oceanbase-client-2.4.11.jar";`
	`90`	`+ return "https://repo1.maven.org/maven2/com/oceanbase/oceanbase-client/2.4.12/oceanbase-client-2.4.12.jar";`
`91`	`91`	`}`
`92`	`92`	`}`