[Feature][Transform-V2] Support vector series sql function (apache#9765)

Author: liuwei178

docs/en/transform-v2/sql-functions.md

@@ -1155,4 +1155,70 @@ SELECT * FROM dual
 	LATERAL VIEW EXPLODE ( SPLIT ( pk_id, ';' ) ) AS pk_id
 	LATERAL VIEW OUTER EXPLODE ( age ) AS age
 	LATERAL VIEW OUTER EXPLODE ( ARRAY(1,1) ) AS num
-```
+```
+
+## Vector Functions
+
+### VECTOR_DIMS
+
+```VECTOR_DIMS(vector) -> INT```
+
+Returns an INT value representing the number of dimensions (elements) in the vector.
+
+Example:
+
+VECTOR_DIMS(vector)
+
+### VECTOR_NORM
+
+```VECTOR_NORM(vector) -> DOUBLE```
+
+Calculates the L2 norm (Euclidean norm) of a vector, which represents the length or magnitude of the vector.
+
+Example:
+
+VECTOR_NORM(vector)
+
+### INNER_PRODUCT
+
+```INNER_PRODUCT(vector1, vector2) -> DOUBLE```
+
+Calculates the inner product (dot product) of two vectors, which is used to measure the similarity and projection between the vectors.
+
+Example:
+
+INNER_PRODUCT(vector1, vector2)
+
+### COSINE_DISTANCE
+
+```COSINE_DISTANCE(vector1, vector2) -> DOUBLE```
+
+Returns a DOUBLE value between 0 and 1:
+
+0: Identical vectors (completely similar)
+
+1: Orthogonal vectors (completely dissimilar)
+
+Example:
+
+COSINE_DISTANCE(vector1, vector2)
+
+### L1_DISTANCE
+
+```L1_DISTANCE(vector1, vector2) -> DOUBLE```
+
+Calculates the Manhattan (L1) distance between two vectors.
+
+Example:
+
+L1_DISTANCE(vector1, vector2)
+
+### L2_DISTANCE
+
+```L2_DISTANCE(vector1, vector2) -> DOUBLE```
+
+Calculates the Euclidean (L2) distance between two vectors.
+
+Example:
+
+L2_DISTANCE(vector1, vector2)

docs/zh/transform-v2/sql-functions.md

@@ -1150,3 +1150,69 @@ SELECT * FROM dual
 	LATERAL VIEW OUTER EXPLODE ( age ) AS age
 	LATERAL VIEW OUTER EXPLODE ( ARRAY(1,1) ) AS num
 ```
+
+## 向量函数
+
+### VECTOR_DIMS
+
+```VECTOR_DIMS(vector) -> INT```
+
+返回一个INT值，表示向量中的维数（元素）。
+
+示例:
+
+VECTOR_DIMS(vector)
+
+### VECTOR_NORM
+
+```VECTOR_NORM(vector) -> DOUBLE```
+
+计算向量的L2范数（欧几里得范数），表示向量的长度或大小。
+
+示例:
+
+VECTOR_NORM(vector)
+
+### INNER_PRODUCT
+
+```INNER_PRODUCT(vector1, vector2) -> DOUBLE```
+
+计算两个向量的内积（点积），用于测量向量之间的相似性和投影。
+
+示例:
+
+INNER_PRODUCT(vector1, vector2)
+
+### COSINE_DISTANCE
+
+```COSINE_DISTANCE(vector1, vector2) -> DOUBLE```
+
+返回介于 0 和 1 之间的 DOUBLE 值：
+
+0：相同的向量（完全相似）
+
+1：正交向量（完全不同）
+
+示例:
+
+COSINE_DISTANCE(vector1, vector2)
+
+### L1_DISTANCE
+
+```L1_DISTANCE(vector1, vector2) -> DOUBLE```
+
+计算两个向量之间的曼哈顿（L1）距离。
+
+示例:
+
+L1_DISTANCE(vector1, vector2)
+
+### L2_DISTANCE
+
+```L2_DISTANCE(vector1, vector2) -> DOUBLE```
+
+计算两个向量之间的欧几里得（L2）距离。
+
+示例:
+
+L2_DISTANCE(vector1, vector2)

seatunnel-common/src/main/java/org/apache/seatunnel/common/utils/BufferUtils.java renamed to seatunnel-common/src/main/java/org/apache/seatunnel/common/utils/VectorUtils.java

@@ -35,8 +35,10 @@
 
 import java.nio.Buffer;
 import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Map;
 
-public class BufferUtils {
+public class VectorUtils {
 
     public static ByteBuffer toByteBuffer(Short[] shortArray) {
         ByteBuffer byteBuffer = ByteBuffer.allocate(shortArray.length * 2);
@@ -127,4 +129,46 @@ public static Integer[] toIntArray(ByteBuffer byteBuffer) {
 
         return intArray;
     }
+
+    public static Float[] convertSparseVectorToFloatArray(Map<?, ?> sparseVector) {
+        if (sparseVector.isEmpty()) {
+            return new Float[0];
+        }
+        int maxIndex = -1;
+        for (Map.Entry<?, ?> entry : sparseVector.entrySet()) {
+            Object key = entry.getKey();
+            if (!(key instanceof Integer)) {
+                throw new IllegalArgumentException(
+                        String.format(
+                                "Sparse vector key must be Integer, but got: %s,",
+                                key.getClass().getName()));
+            }
+            int index = (Integer) key;
+            if (index < 0) {
+                throw new IllegalArgumentException(
+                        String.format("Sparse vector index cannot be negative: %d", index));
+            }
+            // prevent OOM
+            if (index > 1000000) {
+                throw new IllegalArgumentException(
+                        String.format("Sparse vector index too large: %d", index));
+            }
+            maxIndex = Math.max(maxIndex, index);
+        }
+        Float[] denseVector = new Float[maxIndex + 1];
+        Arrays.fill(denseVector, 0.0f);
+        for (Map.Entry<?, ?> entry : sparseVector.entrySet()) {
+            Object key = entry.getKey();
+            Object value = entry.getValue();
+            if (!(value instanceof Number)) {
+                throw new IllegalArgumentException(
+                        String.format(
+                                "Sparse vector value must be a Number, but got: %s",
+                                value.getClass().getName()));
+            }
+            int index = (Integer) key;
+            denseVector[index] = ((Number) value).floatValue();
+        }
+        return denseVector;
+    }
 }

seatunnel-common/src/test/java/org/apache/seatunnel/common/utils/BufferUtilsTest.java renamed to seatunnel-common/src/test/java/org/apache/seatunnel/common/utils/VectorUtilsTest.java

@@ -22,40 +22,40 @@
 
 import java.nio.ByteBuffer;
 
-public class BufferUtilsTest {
+public class VectorUtilsTest {
 
     @Test
     public void testToByteBufferAndToShortArray() {
         Short[] shortArray = {1, 2, 3, 4, 5};
-        ByteBuffer byteBuffer = BufferUtils.toByteBuffer(shortArray);
-        Short[] resultArray = BufferUtils.toShortArray(byteBuffer);
+        ByteBuffer byteBuffer = VectorUtils.toByteBuffer(shortArray);
+        Short[] resultArray = VectorUtils.toShortArray(byteBuffer);
 
         Assertions.assertArrayEquals(shortArray, resultArray, "Short array conversion failed");
     }
 
     @Test
     public void testToByteBufferAndToFloatArray() {
         Float[] floatArray = {1.1f, 2.2f, 3.3f, 4.4f, 5.5f};
-        ByteBuffer byteBuffer = BufferUtils.toByteBuffer(floatArray);
-        Float[] resultArray = BufferUtils.toFloatArray(byteBuffer);
+        ByteBuffer byteBuffer = VectorUtils.toByteBuffer(floatArray);
+        Float[] resultArray = VectorUtils.toFloatArray(byteBuffer);
 
         Assertions.assertArrayEquals(floatArray, resultArray, "Float array conversion failed");
     }
 
     @Test
     public void testToByteBufferAndToDoubleArray() {
         Double[] doubleArray = {1.1, 2.2, 3.3, 4.4, 5.5};
-        ByteBuffer byteBuffer = BufferUtils.toByteBuffer(doubleArray);
-        Double[] resultArray = BufferUtils.toDoubleArray(byteBuffer);
+        ByteBuffer byteBuffer = VectorUtils.toByteBuffer(doubleArray);
+        Double[] resultArray = VectorUtils.toDoubleArray(byteBuffer);
 
         Assertions.assertArrayEquals(doubleArray, resultArray, "Double array conversion failed");
     }
 
     @Test
     public void testToByteBufferAndToIntArray() {
         Integer[] intArray = {1, 2, 3, 4, 5};
-        ByteBuffer byteBuffer = BufferUtils.toByteBuffer(intArray);
-        Integer[] resultArray = BufferUtils.toIntArray(byteBuffer);
+        ByteBuffer byteBuffer = VectorUtils.toByteBuffer(intArray);
+        Integer[] resultArray = VectorUtils.toIntArray(byteBuffer);
 
         Assertions.assertArrayEquals(intArray, resultArray, "Integer array conversion failed");
     }
@@ -64,26 +64,26 @@ public void testToByteBufferAndToIntArray() {
     public void testEmptyArrayConversion() {
         // Test empty arrays
         Short[] shortArray = {};
-        ByteBuffer shortBuffer = BufferUtils.toByteBuffer(shortArray);
-        Short[] shortResultArray = BufferUtils.toShortArray(shortBuffer);
+        ByteBuffer shortBuffer = VectorUtils.toByteBuffer(shortArray);
+        Short[] shortResultArray = VectorUtils.toShortArray(shortBuffer);
         Assertions.assertArrayEquals(
                 shortArray, shortResultArray, "Empty Short array conversion failed");
 
         Float[] floatArray = {};
-        ByteBuffer floatBuffer = BufferUtils.toByteBuffer(floatArray);
-        Float[] floatResultArray = BufferUtils.toFloatArray(floatBuffer);
+        ByteBuffer floatBuffer = VectorUtils.toByteBuffer(floatArray);
+        Float[] floatResultArray = VectorUtils.toFloatArray(floatBuffer);
         Assertions.assertArrayEquals(
                 floatArray, floatResultArray, "Empty Float array conversion failed");
 
         Double[] doubleArray = {};
-        ByteBuffer doubleBuffer = BufferUtils.toByteBuffer(doubleArray);
-        Double[] doubleResultArray = BufferUtils.toDoubleArray(doubleBuffer);
+        ByteBuffer doubleBuffer = VectorUtils.toByteBuffer(doubleArray);
+        Double[] doubleResultArray = VectorUtils.toDoubleArray(doubleBuffer);
         Assertions.assertArrayEquals(
                 doubleArray, doubleResultArray, "Empty Double array conversion failed");
 
         Integer[] intArray = {};
-        ByteBuffer intBuffer = BufferUtils.toByteBuffer(intArray);
-        Integer[] intResultArray = BufferUtils.toIntArray(intBuffer);
+        ByteBuffer intBuffer = VectorUtils.toByteBuffer(intArray);
+        Integer[] intResultArray = VectorUtils.toIntArray(intBuffer);
         Assertions.assertArrayEquals(
                 intArray, intResultArray, "Empty Integer array conversion failed");
     }

seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/serialize/ElasticsearchRowSerializer.java

@@ -24,7 +24,7 @@
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
 import org.apache.seatunnel.common.exception.CommonError;
 import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated;
-import org.apache.seatunnel.common.utils.BufferUtils;
+import org.apache.seatunnel.common.utils.VectorUtils;
 import org.apache.seatunnel.connectors.seatunnel.elasticsearch.dto.ElasticsearchClusterInfo;
 import org.apache.seatunnel.connectors.seatunnel.elasticsearch.dto.IndexInfo;
 import org.apache.seatunnel.connectors.seatunnel.elasticsearch.exception.ElasticsearchConnectorException;
@@ -218,7 +218,7 @@ private Object convertValue(String fieldName, Object value) {
             // Check if this field is configured as a vectorization field
             if (vectorizationFields != null && vectorizationFields.contains(fieldName)) {
                 ByteBuffer buffer = (ByteBuffer) value;
-                Float[] floats = BufferUtils.toFloatArray(buffer);
+                Float[] floats = VectorUtils.toFloatArray(buffer);
 
                 // Use the configured dimension or calculate it from the buffer size
                 int dimension = vectorDimension > 0 ? vectorDimension : buffer.remaining() / 4;
@@ -232,7 +232,7 @@ private Object convertValue(String fieldName, Object value) {
             } else {
                 // Default behavior for ByteBuffer fields not specified as vectorization fields
                 ByteBuffer buffer = (ByteBuffer) value;
-                Float[] floats = BufferUtils.toFloatArray(buffer);
+                Float[] floats = VectorUtils.toFloatArray(buffer);
                 int floatCount = buffer.remaining() / 4;
 
                 for (int i = 0; i < floatCount; i++) {

seatunnel-connectors-v2/connector-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/utils/FakeDataRandomUtils.java

@@ -19,7 +19,7 @@
 
 import org.apache.seatunnel.api.table.catalog.Column;
 import org.apache.seatunnel.api.table.type.DecimalType;
-import org.apache.seatunnel.common.utils.BufferUtils;
+import org.apache.seatunnel.common.utils.VectorUtils;
 import org.apache.seatunnel.connectors.seatunnel.fake.config.FakeConfig;
 
 import org.apache.commons.collections4.CollectionUtils;
@@ -231,7 +231,7 @@ public ByteBuffer randomFloatVector(Column column) {
                     RandomUtils.nextFloat(
                             fakeConfig.getVectorFloatMin(), fakeConfig.getVectorFloatMax());
         }
-        return BufferUtils.toByteBuffer(floatVector);
+        return VectorUtils.toByteBuffer(floatVector);
     }
 
     public ByteBuffer randomFloat16Vector(Column column) {
@@ -244,7 +244,7 @@ public ByteBuffer randomFloat16Vector(Column column) {
                             fakeConfig.getVectorFloatMin(), fakeConfig.getVectorFloatMax());
             float16Vector[i] = floatToFloat16(value);
         }
-        return BufferUtils.toByteBuffer(float16Vector);
+        return VectorUtils.toByteBuffer(float16Vector);
     }
 
     public ByteBuffer randomBFloat16Vector(Column column) {
@@ -257,7 +257,7 @@ public ByteBuffer randomBFloat16Vector(Column column) {
                             fakeConfig.getVectorFloatMin(), fakeConfig.getVectorFloatMax());
             bfloat16Vector[i] = floatToBFloat16(value);
         }
-        return BufferUtils.toByteBuffer(bfloat16Vector);
+        return VectorUtils.toByteBuffer(bfloat16Vector);
     }
 
     public Map<Integer, Float> randomSparseFloatVector(Column column) {

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oceanbase/OceanBaseMysqlJdbcRowConverter.java

@@ -25,7 +25,7 @@
 import org.apache.seatunnel.api.table.type.SqlType;
 import org.apache.seatunnel.common.exception.CommonError;
 import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated;
-import org.apache.seatunnel.common.utils.BufferUtils;
+import org.apache.seatunnel.common.utils.VectorUtils;
 import org.apache.seatunnel.connectors.seatunnel.jdbc.exception.JdbcConnectorErrorCode;
 import org.apache.seatunnel.connectors.seatunnel.jdbc.exception.JdbcConnectorException;
 import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.converter.AbstractJdbcRowConverter;
@@ -101,7 +101,7 @@ public SeaTunnelRow toInternal(ResultSet rs, TableSchema tableSchema) throws SQL
                         for (int i = 0; i < stringArray.length; i++) {
                             arrays[i] = Float.parseFloat(stringArray[i]);
                         }
-                        fields[fieldIndex] = BufferUtils.toByteBuffer(arrays);
+                        fields[fieldIndex] = VectorUtils.toByteBuffer(arrays);
                     }
                     break;
                 case DOUBLE:
@@ -188,7 +188,7 @@ public PreparedStatement toExternal(
                         if (row.getField(fieldIndex) instanceof ByteBuffer) {
                             ByteBuffer byteBuffer = (ByteBuffer) row.getField(fieldIndex);
                             // Convert ByteBuffer to Float[]
-                            Float[] floatArray = BufferUtils.toFloatArray(byteBuffer);
+                            Float[] floatArray = VectorUtils.toFloatArray(byteBuffer);
                             StringBuilder vector = new StringBuilder();
                             vector.append("[");
                             for (Float aFloat : floatArray) {

seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/utils/sink/MilvusSinkConverter.java

@@ -28,8 +28,8 @@
 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.utils.BufferUtils;
 import org.apache.seatunnel.common.utils.JsonUtils;
+import org.apache.seatunnel.common.utils.VectorUtils;
 import org.apache.seatunnel.connectors.seatunnel.milvus.exception.MilvusConnectionErrorCode;
 import org.apache.seatunnel.connectors.seatunnel.milvus.exception.MilvusConnectorException;
 
@@ -74,7 +74,7 @@ public Object convertBySeaTunnelType(
                 return value.toString();
             case FLOAT_VECTOR:
                 ByteBuffer floatVectorBuffer = (ByteBuffer) value;
-                Float[] floats = BufferUtils.toFloatArray(floatVectorBuffer);
+                Float[] floats = VectorUtils.toFloatArray(floatVectorBuffer);
                 return Arrays.stream(floats).collect(Collectors.toList());
             case BINARY_VECTOR:
             case BFLOAT16_VECTOR: