This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 6e23e4fadc [Improve][Connector-V2] Improve YashanDB vector type
conversion (#11397)
6e23e4fadc is described below
commit 6e23e4fadcbaf2595e4782a24fd116abb0e7ab33
Author: Tlinian <[email protected]>
AuthorDate: Tue Jul 21 22:49:11 2026 +0800
[Improve][Connector-V2] Improve YashanDB vector type conversion (#11397)
---
docs/en/connectors/sink/Jdbc.md | 2 +-
docs/en/connectors/source/Jdbc.md | 2 +-
docs/zh/connectors/sink/Jdbc.md | 4 +-
docs/zh/connectors/source/Jdbc.md | 4 +-
seatunnel-connectors-v2/connector-jdbc/pom.xml | 2 +-
.../dialect/yashandb/YashanDbJdbcRowConverter.java | 237 +++++++++++++++++++--
.../dialect/yashandb/YashanDbTypeConverter.java | 35 ++-
.../yashandb/YashanDbJdbcRowConverterTest.java | 174 +++++++++++++++
.../yashandb/YashanDbTypeConverterTest.java | 28 ++-
.../connectors/seatunnel/jdbc/JdbcYashanDbIT.java | 2 +-
10 files changed, 441 insertions(+), 49 deletions(-)
diff --git a/docs/en/connectors/sink/Jdbc.md b/docs/en/connectors/sink/Jdbc.md
index 81a34bf5e3..f79a90e0cc 100644
--- a/docs/en/connectors/sink/Jdbc.md
+++ b/docs/en/connectors/sink/Jdbc.md
@@ -557,7 +557,7 @@ The following values are starting points. Confirm the
driver artifact and versio
| 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
|
| Highgo | com.highgo.jdbc.Driver |
jdbc:highgo://localhost:5866/highgo | /
|
https://repo1.maven.org/maven2/com/highgo/HgdbJdbc/6.2.3/HgdbJdbc-6.2.3.jar
|
| Dsql | org.postgresql.Driver |
jdbc:postgresql://Amazon Aurora DSQL Cluster Endpoint:5432/postgres |
org.postgresql.xa.PGXADataSource |
https://mvnrepository.com/artifact/org.postgresql/postgresql
|
-| YashanDB | com.yashandb.jdbc.Driver |
jdbc:yasdb://localhost:1688/SYS | /
|
https://mvnrepository.com/artifact/com.yashandb/yashandb-jdbc
|
+| YashanDB | com.yashandb.jdbc.Driver |
jdbc:yasdb://localhost:1688/SYS | /
|
https://repo1.maven.org/maven2/com/yashandb/yashandb-jdbc/1.10.7/yashandb-jdbc-1.10.7.jar
|
## Common patterns
diff --git a/docs/en/connectors/source/Jdbc.md
b/docs/en/connectors/source/Jdbc.md
index ef485b54e3..1521a6d2ee 100644
--- a/docs/en/connectors/source/Jdbc.md
+++ b/docs/en/connectors/source/Jdbc.md
@@ -314,7 +314,7 @@ The following values are starting points. Confirm the
driver artifact, license,
| Highgo | com.highgo.jdbc.Driver |
jdbc:highgo://localhost:5866/highgo |
https://repo1.maven.org/maven2/com/highgo/HgdbJdbc/6.2.3/HgdbJdbc-6.2.3.jar
|
| Presto | com.facebook.presto.jdbc.PrestoDriver |
jdbc:presto://localhost:8080/presto |
https://repo1.maven.org/maven2/com/facebook/presto/presto-jdbc/0.279/presto-jdbc-0.279.jar
|
| Trino | io.trino.jdbc.TrinoDriver |
jdbc:trino://localhost:8080/trino |
https://repo1.maven.org/maven2/io/trino/trino-jdbc/460/trino-jdbc-460.jar
|
-| YashanDB | com.yashandb.jdbc.Driver |
jdbc:yasdb://localhost:1688/SYS |
https://mvnrepository.com/artifact/com.yashandb/yashandb-jdbc
|
+| YashanDB | com.yashandb.jdbc.Driver |
jdbc:yasdb://localhost:1688/SYS |
https://repo1.maven.org/maven2/com/yashandb/yashandb-jdbc/1.10.7/yashandb-jdbc-1.10.7.jar
|
## Common patterns
diff --git a/docs/zh/connectors/sink/Jdbc.md b/docs/zh/connectors/sink/Jdbc.md
index 73f6530ae5..353721f8c4 100644
--- a/docs/zh/connectors/sink/Jdbc.md
+++ b/docs/zh/connectors/sink/Jdbc.md
@@ -550,8 +550,8 @@ Amazon Aurora DSQL 所在的区域。 该参考仅适用于 dialect="dsql"
| 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
|
| Highgo | com.highgo.jdbc.Driver |
jdbc:highgo://localhost:5866/highgo | /
|
https://repo1.maven.org/maven2/com/highgo/HgdbJdbc/6.2.3/HgdbJdbc-6.2.3.jar
|
-| Dsql | org.postgresql.Driver |
jdbc:postgresql://Amazon Aurora DSQL Cluster Endpoint:5432/postgres |
org.postgresql.xa.PGXADataSource |
https://mvnrepository.com/artifact/org.postgresql/postgresql
|
-| YashanDB | com.yashandb.jdbc.Driver |
jdbc:yasdb://localhost:1688/SYS | /
|
https://mvnrepository.com/artifact/com.yashandb/yashandb-jdbc
|
+| Dsql | org.postgresql.Driver |
jdbc:postgresql://Amazon Aurora DSQL Cluster Endpoint:5432/postgres |
org.postgresql.xa.PGXADataSource |
https://mvnrepository.com/artifact/org.postgresql/postgresql
|
+| YashanDB | com.yashandb.jdbc.Driver |
jdbc:yasdb://localhost:1688/SYS | /
|
https://repo1.maven.org/maven2/com/yashandb/yashandb-jdbc/1.10.7/yashandb-jdbc-1.10.7.jar
|
## 常用模式
diff --git a/docs/zh/connectors/source/Jdbc.md
b/docs/zh/connectors/source/Jdbc.md
index c88fa2001d..cb5d185789 100644
--- a/docs/zh/connectors/source/Jdbc.md
+++ b/docs/zh/connectors/source/Jdbc.md
@@ -286,7 +286,7 @@ SeaTunnel 为 `query` 推断主键时，会继承结果集第一列所属底层
下表仅作为起点。部署前应向数据库厂商确认驱动制品、许可证、数据库版本兼容性和 Java 版本兼容性。
| 数据源 | 驱动 | URL
| Maven
|
-|-------------|---------------------------------------------------|--------------------------------------------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------|
+|-------------|---------------------------------------------------|--------------------------------------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------|
| mysql | com.mysql.cj.jdbc.Driver |
jdbc:mysql://localhost:3306/test |
https://mvnrepository.com/artifact/mysql/mysql-connector-java
|
| postgresql | org.postgresql.Driver |
jdbc:postgresql://localhost:5432/postgres |
https://mvnrepository.com/artifact/org.postgresql/postgresql
|
| dm | dm.jdbc.driver.DmDriver |
jdbc:dm://localhost:5236 |
https://mvnrepository.com/artifact/com.dameng/DmJdbcDriver18
|
@@ -313,7 +313,7 @@ SeaTunnel 为 `query` 推断主键时，会继承结果集第一列所属底层
| Highgo | com.highgo.jdbc.Driver |
jdbc:highgo://localhost:5866/highgo |
https://repo1.maven.org/maven2/com/highgo/HgdbJdbc/6.2.3/HgdbJdbc-6.2.3.jar
|
| Presto | com.facebook.presto.jdbc.PrestoDriver |
jdbc:presto://localhost:8080/presto |
https://repo1.maven.org/maven2/com/facebook/presto/presto-jdbc/0.279/presto-jdbc-0.279.jar
|
| Trino | io.trino.jdbc.TrinoDriver |
jdbc:trino://localhost:8080/trino |
https://repo1.maven.org/maven2/io/trino/trino-jdbc/460/trino-jdbc-460.jar
|
-| YashanDB | com.yashandb.jdbc.Driver |
jdbc:yasdb://localhost:1688/SYS |
https://mvnrepository.com/artifact/com.yashandb/yashandb-jdbc
|
+| YashanDB | com.yashandb.jdbc.Driver |
jdbc:yasdb://localhost:1688/SYS |
https://repo1.maven.org/maven2/com/yashandb/yashandb-jdbc/1.10.7/yashandb-jdbc-1.10.7.jar
|
## 常用模式
diff --git a/seatunnel-connectors-v2/connector-jdbc/pom.xml
b/seatunnel-connectors-v2/connector-jdbc/pom.xml
index e42478cf46..a7696429fd 100644
--- a/seatunnel-connectors-v2/connector-jdbc/pom.xml
+++ b/seatunnel-connectors-v2/connector-jdbc/pom.xml
@@ -60,7 +60,7 @@
<trino.version>460</trino.version>
<aws.sdk.version>2.31.30</aws.sdk.version>
<duckdb.version>1.3.1.0</duckdb.version>
- <yashandb.jdbc.version>1.9.24</yashandb.jdbc.version>
+ <yashandb.jdbc.version>1.10.7</yashandb.jdbc.version>
</properties>
<dependencyManagement>
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/yashandb/YashanDbJdbcRowConverter.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/yashandb/YashanDbJdbcRowConverter.java
index a7332635e2..ae05689c13 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/yashandb/YashanDbJdbcRowConverter.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/yashandb/YashanDbJdbcRowConverter.java
@@ -20,18 +20,21 @@ package
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.yashandb
import org.apache.seatunnel.shade.org.apache.commons.lang3.StringUtils;
import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.type.ArrayType;
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
import org.apache.seatunnel.api.table.type.SqlType;
import org.apache.seatunnel.common.exception.CommonError;
-import org.apache.seatunnel.common.utils.VectorUtils;
import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.converter.AbstractJdbcRowConverter;
import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.DatabaseIdentifier;
import org.apache.seatunnel.connectors.seatunnel.jdbc.utils.JdbcFieldTypeUtils;
+import com.yashandb.vector.Vector;
+
import javax.annotation.Nullable;
+import java.nio.ByteBuffer;
import java.sql.Date;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
@@ -107,16 +110,7 @@ public class YashanDbJdbcRowConverter extends
AbstractJdbcRowConverter {
case FLOAT16_VECTOR:
case BFLOAT16_VECTOR:
case FLOAT_VECTOR:
- 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] = VectorUtils.toByteBuffer(arrays);
- }
+ fields[fieldIndex] =
convertToVectorBuffer(rs.getObject(resultSetIndex));
break;
case BYTES:
fields[fieldIndex] = JdbcFieldTypeUtils.getBytes(rs,
resultSetIndex);
@@ -138,6 +132,26 @@ public class YashanDbJdbcRowConverter extends
AbstractJdbcRowConverter {
return new SeaTunnelRow(fields);
}
+ @Override
+ protected void setNullToStatementByDataType(
+ PreparedStatement statement,
+ SeaTunnelDataType<?> seaTunnelDataType,
+ int statementIndex,
+ @Nullable String sourceType)
+ throws SQLException {
+ SqlType sqlType = seaTunnelDataType.getSqlType();
+ if (SqlType.ARRAY.equals(sqlType)) {
+ statement.setNull(statementIndex, java.sql.Types.ARRAY);
+ return;
+ }
+ if (isFloatVectorType(sqlType)) {
+ statement.setNull(statementIndex, java.sql.Types.VARCHAR);
+ return;
+ }
+ super.setNullToStatementByDataType(
+ statement, seaTunnelDataType, statementIndex, sourceType);
+ }
+
@Override
protected void setValueToStatementByDataType(
Object value,
@@ -146,20 +160,201 @@ public class YashanDbJdbcRowConverter extends
AbstractJdbcRowConverter {
int statementIndex,
@Nullable String sourceType)
throws SQLException {
- if (seaTunnelDataType.getSqlType().equals(SqlType.ARRAY)
- ||
seaTunnelDataType.getSqlType().equals(SqlType.FLOAT16_VECTOR)
- || seaTunnelDataType.getSqlType().equals(SqlType.FLOAT_VECTOR)
- ||
seaTunnelDataType.getSqlType().equals(SqlType.BFLOAT16_VECTOR)) {
- Object[] array = (Object[]) value;
- if (array == null) {
+ SqlType sqlType = seaTunnelDataType.getSqlType();
+ if (sqlType.equals(SqlType.ARRAY)) {
+ if (value == null) {
statement.setNull(statementIndex, java.sql.Types.ARRAY);
+ return;
+ }
+ Object[] array = (Object[]) value;
+ setArray(statement, (ArrayType<?, ?>) seaTunnelDataType,
statementIndex, array);
+ return;
+ }
+ if (isFloatVectorType(sqlType)) {
+ if (value == null) {
+ statement.setNull(statementIndex, java.sql.Types.VARCHAR);
+ } else if (value instanceof ByteBuffer) {
+ statement.setObject(
+ statementIndex,
+ Vector.ofFloat32Values(toFloat32VectorValues(value,
sqlType)));
+ } else if (value instanceof Object[]) {
+ statement.setObject(
+ statementIndex,
+ Vector.ofFloat32Values(toFloat32VectorValues(value,
sqlType)));
} else {
- String arrayStr = Arrays.toString(array);
- statement.setString(statementIndex, arrayStr);
+ throw new SQLException(
+ "Unsupported YashanDB vector value type: " +
value.getClass().getName());
+ }
+ return;
+ }
+ super.setValueToStatementByDataType(
+ value, statement, seaTunnelDataType, statementIndex,
sourceType);
+ }
+
+ private static boolean isFloatVectorType(SqlType sqlType) {
+ return SqlType.FLOAT_VECTOR.equals(sqlType)
+ || SqlType.FLOAT16_VECTOR.equals(sqlType)
+ || SqlType.BFLOAT16_VECTOR.equals(sqlType);
+ }
+
+ private static ByteBuffer convertToVectorBuffer(Object vectorValue) {
+ if (vectorValue == null) {
+ return null;
+ }
+ if (vectorValue instanceof Vector) {
+ return toByteBuffer(((Vector) vectorValue).toFloatArray());
+ }
+ String vectorString = vectorValue.toString();
+ if (StringUtils.isBlank(vectorString)) {
+ return null;
+ }
+ vectorString = vectorString.replace("[", "").replace("]", "");
+ String[] values = vectorString.split(",");
+ return toByteBuffer(parseFloatArray(values));
+ }
+
+ private static float[] parseFloatArray(String[] values) {
+ float[] floatArray = new float[values.length];
+ for (int i = 0; i < values.length; i++) {
+ floatArray[i] = Float.parseFloat(values[i].trim());
+ }
+ return floatArray;
+ }
+
+ private static ByteBuffer toByteBuffer(float[] floatArray) {
+ ByteBuffer byteBuffer = ByteBuffer.allocate(floatArray.length *
Float.BYTES);
+ for (float value : floatArray) {
+ byteBuffer.putFloat(value);
+ }
+ byteBuffer.flip();
+ return byteBuffer;
+ }
+
+ private static float[] toPrimitiveFloatArray(Object value) {
+ if (value instanceof ByteBuffer) {
+ ByteBuffer byteBuffer = ((ByteBuffer) value).duplicate();
+ byteBuffer.rewind();
+ float[] floatArray = new float[byteBuffer.remaining() /
Float.BYTES];
+ for (int i = 0; i < floatArray.length; i++) {
+ floatArray[i] = byteBuffer.getFloat();
+ }
+ return floatArray;
+ }
+
+ Object[] objectArray = (Object[]) value;
+ float[] floatArray = new float[objectArray.length];
+ for (int i = 0; i < objectArray.length; i++) {
+ floatArray[i] = ((Number) objectArray[i]).floatValue();
+ }
+ return floatArray;
+ }
+
+ private static float[] toFloat32VectorValues(Object value, SqlType sqlType)
+ throws SQLException {
+ // YashanDB Vector writes use float32 values for SeaTunnel vector
inputs. FLOAT16 and
+ // BFLOAT16 buffers keep their original 16-bit precision; this path
widens them to
+ // float32 only at the JDBC boundary so the element count is preserved.
+ if (SqlType.FLOAT16_VECTOR.equals(sqlType)) {
+ return toFloat16Array(value);
+ }
+ if (SqlType.BFLOAT16_VECTOR.equals(sqlType)) {
+ return toBFloat16Array(value);
+ }
+ return toPrimitiveFloatArray(value);
+ }
+
+ private static float[] toFloat16Array(Object value) throws SQLException {
+ short[] shortArray = toPrimitiveShortArray(value);
+ float[] floatArray = new float[shortArray.length];
+ for (int i = 0; i < shortArray.length; i++) {
+ floatArray[i] = float16ToFloat(shortArray[i]);
+ }
+ return floatArray;
+ }
+
+ private static float[] toBFloat16Array(Object value) throws SQLException {
+ short[] shortArray = toPrimitiveShortArray(value);
+ float[] floatArray = new float[shortArray.length];
+ for (int i = 0; i < shortArray.length; i++) {
+ floatArray[i] = bfloat16ToFloat(shortArray[i]);
+ }
+ return floatArray;
+ }
+
+ private static short[] toPrimitiveShortArray(Object value) throws
SQLException {
+ if (value instanceof ByteBuffer) {
+ ByteBuffer byteBuffer = ((ByteBuffer) value).duplicate();
+ byteBuffer.rewind();
+ // SeaTunnel FLOAT16_VECTOR and BFLOAT16_VECTOR ByteBuffers are
encoded as one
+ // big-endian 2-byte short per vector element via
VectorUtils.toByteBuffer(Short[]).
+ if (byteBuffer.remaining() % Short.BYTES != 0) {
+ throw new SQLException(
+ "Invalid YashanDB 2-byte vector buffer length: " +
byteBuffer.remaining());
+ }
+ short[] shortArray = new short[byteBuffer.remaining() /
Short.BYTES];
+ for (int i = 0; i < shortArray.length; i++) {
+ shortArray[i] = byteBuffer.getShort();
+ }
+ return shortArray;
+ }
+
+ Object[] objectArray = (Object[]) value;
+ short[] shortArray = new short[objectArray.length];
+ for (int i = 0; i < objectArray.length; i++) {
+ shortArray[i] = ((Number) objectArray[i]).shortValue();
+ }
+ return shortArray;
+ }
+
+ private static float float16ToFloat(short value) {
+ int half = value & 0xffff;
+ int sign = (half & 0x8000) << 16;
+ int exponent = (half >>> 10) & 0x1f;
+ int mantissa = half & 0x03ff;
+ int bits;
+ if (exponent == 0) {
+ if (mantissa == 0) {
+ bits = sign;
+ } else {
+ while ((mantissa & 0x0400) == 0) {
+ mantissa <<= 1;
+ exponent--;
+ }
+ exponent++;
+ mantissa &= ~0x0400;
+ bits = sign | ((exponent + 112) << 23) | (mantissa << 13);
+ }
+ } else if (exponent == 0x1f) {
+ bits = sign | 0x7f800000 | (mantissa << 13);
+ } else {
+ bits = sign | ((exponent + 112) << 23) | (mantissa << 13);
+ }
+ return Float.intBitsToFloat(bits);
+ }
+
+ private static float bfloat16ToFloat(short value) {
+ return Float.intBitsToFloat((value & 0xffff) << 16);
+ }
+
+ private static void setArray(
+ PreparedStatement statement,
+ ArrayType<?, ?> arrayType,
+ int statementIndex,
+ Object[] array)
+ throws SQLException {
+ SqlType elementSqlType = arrayType.getElementType().getSqlType();
+ if
(YashanDbTypeConverter.SUPPORT_FLOAT32_VECTOR.contains(elementSqlType)) {
+ statement.setObject(
+ statementIndex,
Vector.ofFloat32Values(toPrimitiveFloatArray(array)));
+ } else if
(YashanDbTypeConverter.SUPPORT_FLOAT64_VECTOR.contains(elementSqlType)) {
+ double[] doubleArray = new double[array.length];
+ for (int i = 0; i < array.length; i++) {
+ doubleArray[i] = ((Number) array[i]).doubleValue();
}
+ statement.setObject(statementIndex,
Vector.ofFloat64Values(doubleArray));
} else {
- super.setValueToStatementByDataType(
- value, statement, seaTunnelDataType, statementIndex,
sourceType);
+ String arrayStr = Arrays.toString(array);
+ statement.setString(statementIndex, arrayStr);
}
}
}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/yashandb/YashanDbTypeConverter.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/yashandb/YashanDbTypeConverter.java
index f30fdb3264..163646a8b0 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/yashandb/YashanDbTypeConverter.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/yashandb/YashanDbTypeConverter.java
@@ -35,7 +35,8 @@ import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.DatabaseI
import com.google.auto.service.AutoService;
import lombok.extern.slf4j.Slf4j;
-import java.util.HashSet;
+import java.util.Collections;
+import java.util.EnumSet;
import java.util.Set;
// reference
@@ -132,20 +133,14 @@ public class YashanDbTypeConverter implements
TypeConverter<BasicTypeDefine> {
public static final long MAX_NVARCHAR_LENGTH = 32767;
public static final long MAX_UROWID_LENGTH = 4000;
public static final int MAX_JSON_LENGTH = 32 * 1024 * 1024;
-
public static final long BYTES_4GB = (long) Math.pow(2, 32);
public static final YashanDbTypeConverter INSTANCE = new
YashanDbTypeConverter();
- private static final Set<SqlType> SUPPORT_VECTOR = new HashSet<>();
-
- static {
- SUPPORT_VECTOR.add(SqlType.BIGINT);
- SUPPORT_VECTOR.add(SqlType.DOUBLE);
- SUPPORT_VECTOR.add(SqlType.FLOAT);
- SUPPORT_VECTOR.add(SqlType.INT);
- SUPPORT_VECTOR.add(SqlType.SMALLINT);
- SUPPORT_VECTOR.add(SqlType.TINYINT);
- }
+ static final Set<SqlType> SUPPORT_FLOAT32_VECTOR =
+ Collections.unmodifiableSet(
+ EnumSet.of(SqlType.FLOAT, SqlType.INT, SqlType.SMALLINT,
SqlType.TINYINT));
+ static final Set<SqlType> SUPPORT_FLOAT64_VECTOR =
+ Collections.unmodifiableSet(EnumSet.of(SqlType.BIGINT,
SqlType.DOUBLE));
public YashanDbTypeConverter() {}
@@ -301,10 +296,6 @@ public class YashanDbTypeConverter implements
TypeConverter<BasicTypeDefine> {
builder.dataType(BasicType.STRING_TYPE);
break;
case VECTOR_NAME:
- String columnType = typeDefine.getColumnType().toUpperCase();
- if ("FLOAT64".equals(columnType)) {
- log.warn("{} vector convert to float32 vector.",
columnType);
- }
builder.dataType(VectorType.VECTOR_FLOAT_TYPE);
builder.scale(typeDefine.getScale());
builder.columnLength(typeDefine.getLength());
@@ -418,13 +409,21 @@ public class YashanDbTypeConverter implements
TypeConverter<BasicTypeDefine> {
break;
case ARRAY:
// YashanDB does not have a native ARRAY column type, store as
vector
- if (SUPPORT_VECTOR.contains(
+ if (SUPPORT_FLOAT32_VECTOR.contains(
+ ((ArrayType)
column.getDataType()).getElementType().getSqlType())) {
+ log.warn(
+ "The column {} with type ARRAY will be converted
to VECTOR in YashanDB",
+ column.getName());
+ builder.columnType(
+ String.format("%s(%s,FLOAT32)", VECTOR_NAME,
column.getColumnLength()));
+ builder.dataType(VECTOR_NAME);
+ } else if (SUPPORT_FLOAT64_VECTOR.contains(
((ArrayType)
column.getDataType()).getElementType().getSqlType())) {
log.warn(
"The column {} with type ARRAY will be converted
to VECTOR in YashanDB",
column.getName());
builder.columnType(
- String.format("%s(%s)", VECTOR_NAME,
column.getColumnLength()));
+ String.format("%s(%s,FLOAT64)", VECTOR_NAME,
column.getColumnLength()));
builder.dataType(VECTOR_NAME);
} else {
throw new UnsupportedOperationException(
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/yashandb/YashanDbJdbcRowConverterTest.java
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/yashandb/YashanDbJdbcRowConverterTest.java
new file mode 100644
index 0000000000..2e0aa021d8
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/yashandb/YashanDbJdbcRowConverterTest.java
@@ -0,0 +1,174 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.yashandb;
+
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.type.ArrayType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.VectorType;
+import org.apache.seatunnel.common.utils.VectorUtils;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+
+import com.yashandb.vector.Vector;
+
+import java.nio.ByteBuffer;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Types;
+import java.util.Collections;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class YashanDbJdbcRowConverterTest {
+
+ private final YashanDbJdbcRowConverter converter = new
YashanDbJdbcRowConverter();
+
+ @Test
+ public void testReadFloatVectorFromYashanDbVectorObject() throws
SQLException {
+ ResultSet resultSet = mock(ResultSet.class);
+ when(resultSet.getObject(1)).thenReturn(Vector.ofFloat32Values(new
float[] {1.25f, 2.5f}));
+
+ SeaTunnelRow row =
+ converter.toInternal(resultSet,
createSchema(VectorType.VECTOR_FLOAT_TYPE));
+
+ ByteBuffer buffer = ((ByteBuffer) row.getField(0)).duplicate();
+ Assertions.assertEquals(1.25f, buffer.getFloat());
+ Assertions.assertEquals(2.5f, buffer.getFloat());
+ }
+
+ @Test
+ public void testWriteFloatVectorUsesYashanDbVectorApi() throws
SQLException {
+ PreparedStatement statement = mock(PreparedStatement.class);
+ ByteBuffer buffer = ByteBuffer.allocate(2 * Float.BYTES);
+ buffer.putFloat(1.25f);
+ buffer.putFloat(2.5f);
+ buffer.flip();
+
+ converter.toExternal(
+ createSchema(VectorType.VECTOR_FLOAT_TYPE),
+ new SeaTunnelRow(new Object[] {buffer}),
+ statement);
+
+ ArgumentCaptor<Vector> vectorCaptor =
ArgumentCaptor.forClass(Vector.class);
+ verify(statement).setObject(eq(1), vectorCaptor.capture());
+ Assertions.assertArrayEquals(
+ new float[] {1.25f, 2.5f},
vectorCaptor.getValue().toFloatArray());
+ verify(statement, never()).setNull(1, Types.NULL);
+ }
+
+ @Test
+ public void testWriteFloat16VectorDecodesTwoByteElements() throws
SQLException {
+ PreparedStatement statement = mock(PreparedStatement.class);
+
+ converter.toExternal(
+ createSchema(VectorType.VECTOR_FLOAT16_TYPE),
+ new SeaTunnelRow(
+ new Object[] {
+ VectorUtils.toByteBuffer(new Short[] {0x3c00,
(short) 0xc000})
+ }),
+ statement);
+
+ ArgumentCaptor<Vector> vectorCaptor =
ArgumentCaptor.forClass(Vector.class);
+ verify(statement).setObject(eq(1), vectorCaptor.capture());
+ Assertions.assertArrayEquals(
+ new float[] {1.0f, -2.0f},
vectorCaptor.getValue().toFloatArray());
+ }
+
+ @Test
+ public void testWriteBFloat16VectorDecodesTwoByteElements() throws
SQLException {
+ PreparedStatement statement = mock(PreparedStatement.class);
+
+ converter.toExternal(
+ createSchema(VectorType.VECTOR_BFLOAT16_TYPE),
+ new SeaTunnelRow(
+ new Object[] {
+ VectorUtils.toByteBuffer(new Short[] {0x3f80,
(short) 0xc000})
+ }),
+ statement);
+
+ ArgumentCaptor<Vector> vectorCaptor =
ArgumentCaptor.forClass(Vector.class);
+ verify(statement).setObject(eq(1), vectorCaptor.capture());
+ Assertions.assertArrayEquals(
+ new float[] {1.0f, -2.0f},
vectorCaptor.getValue().toFloatArray());
+ }
+
+ @Test
+ public void testWriteNullVectorUsesExplicitJdbcType() throws SQLException {
+ PreparedStatement statement = mock(PreparedStatement.class);
+
+ converter.toExternal(
+ createSchema(VectorType.VECTOR_FLOAT_TYPE),
+ new SeaTunnelRow(new Object[] {null}),
+ statement);
+
+ verify(statement).setNull(1, Types.VARCHAR);
+ verify(statement, never()).setObject(eq(1), any());
+ }
+
+ @Test
+ public void testWriteFloat32ArrayUsesYashanDbVectorApi() throws
SQLException {
+ PreparedStatement statement = mock(PreparedStatement.class);
+
+ converter.toExternal(
+ createSchema(ArrayType.INT_ARRAY_TYPE),
+ new SeaTunnelRow(new Object[] {new Integer[] {1, 2}}),
+ statement);
+
+ ArgumentCaptor<Vector> vectorCaptor =
ArgumentCaptor.forClass(Vector.class);
+ verify(statement).setObject(eq(1), vectorCaptor.capture());
+ Assertions.assertArrayEquals(
+ new float[] {1.0f, 2.0f},
vectorCaptor.getValue().toFloatArray());
+ }
+
+ @Test
+ public void testWriteFloat64ArrayUsesYashanDbVectorApi() throws
SQLException {
+ PreparedStatement statement = mock(PreparedStatement.class);
+
+ converter.toExternal(
+ createSchema(ArrayType.DOUBLE_ARRAY_TYPE),
+ new SeaTunnelRow(new Object[] {new Double[] {1.25d, 2.5d}}),
+ statement);
+
+ ArgumentCaptor<Vector> vectorCaptor =
ArgumentCaptor.forClass(Vector.class);
+ verify(statement).setObject(eq(1), vectorCaptor.capture());
+ Assertions.assertArrayEquals(
+ new double[] {1.25d, 2.5d},
vectorCaptor.getValue().toDoubleArray());
+ }
+
+ private static TableSchema createSchema(SeaTunnelDataType<?> dataType) {
+ return TableSchema.builder()
+ .columns(
+ Collections.singletonList(
+ PhysicalColumn.builder()
+ .name("vector_col")
+ .dataType(dataType)
+ .build()))
+ .build();
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/yashandb/YashanDbTypeConverterTest.java
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/yashandb/YashanDbTypeConverterTest.java
index cf7e1c7c60..ce881127a6 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/yashandb/YashanDbTypeConverterTest.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/yashandb/YashanDbTypeConverterTest.java
@@ -459,6 +459,18 @@ public class YashanDbTypeConverterTest {
Column column = INSTANCE.convert(typeDefine);
Assertions.assertEquals(VectorType.VECTOR_FLOAT_TYPE,
column.getDataType());
Assertions.assertEquals(3, column.getScale());
+
+ typeDefine =
+ BasicTypeDefine.builder()
+ .name("test")
+ .columnType("VECTOR(3,FLOAT64)")
+ .dataType("VECTOR")
+ .scale(3)
+ .length(3L)
+ .build();
+ column = INSTANCE.convert(typeDefine);
+ Assertions.assertEquals(VectorType.VECTOR_FLOAT_TYPE,
column.getDataType());
+ Assertions.assertEquals(3, column.getScale());
}
// ============================ reconvert() tests
============================
@@ -733,7 +745,7 @@ public class YashanDbTypeConverterTest {
@Test
public void testReconvertArrayToVector() {
- // ARRAY<INT> -> VECTOR
+ // ARRAY<INT> -> VECTOR FLOAT32
Column column =
PhysicalColumn.builder()
.name("test")
@@ -742,7 +754,19 @@ public class YashanDbTypeConverterTest {
.nullable(true)
.build();
BasicTypeDefine<?> typeDefine = INSTANCE.reconvert(column);
- Assertions.assertEquals("VECTOR(10)", typeDefine.getColumnType());
+ Assertions.assertEquals("VECTOR(10,FLOAT32)",
typeDefine.getColumnType());
+ Assertions.assertEquals("VECTOR", typeDefine.getDataType());
+
+ // ARRAY<DOUBLE> -> VECTOR FLOAT64
+ column =
+ PhysicalColumn.builder()
+ .name("test")
+ .dataType(ArrayType.DOUBLE_ARRAY_TYPE)
+ .columnLength(10L)
+ .nullable(true)
+ .build();
+ typeDefine = INSTANCE.reconvert(column);
+ Assertions.assertEquals("VECTOR(10,FLOAT64)",
typeDefine.getColumnType());
Assertions.assertEquals("VECTOR", typeDefine.getDataType());
// ARRAY<STRING> -> unsupported
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-7/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcYashanDbIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-7/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcYashanDbIT.java
index c79e909f58..8f6608d43b 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-7/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcYashanDbIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-7/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcYashanDbIT.java
@@ -132,7 +132,7 @@ public class JdbcYashanDbIT extends AbstractJdbcIT {
@Override
String driverUrl() {
- return
"https://repo1.maven.org/maven2/com/yashandb/yashandb-jdbc/1.9.24/yashandb-jdbc-1.9.24.jar";;
+ return
"https://repo1.maven.org/maven2/com/yashandb/yashandb-jdbc/1.10.7/yashandb-jdbc-1.10.7.jar";;
}
@Override
