[Feature][Transform V2] Add vector dimension reduction transform (apache#9783)

Author: CosmosNi

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

@@ -1221,4 +1221,44 @@ Calculates the Euclidean (L2) distance between two vectors.
 
 Example:
 
-L2_DISTANCE(vector1, vector2)
+L2_DISTANCE(vector1, vector2)
+
+### VECTOR_REDUCE
+
+```VECTOR_REDUCE(vector_field, target_dimension, method)```
+
+Generic vector dimension reduction function that supports multiple reduction methods.
+
+**Parameters:**
+- `vector_field`: The vector field to reduce (VECTOR type)
+- `target_dimension`: The target dimension (INTEGER, must be smaller than source dimension)
+- `method`: The reduction method (STRING):
+  - **'TRUNCATE'**: Truncates the vector by keeping only the first N elements. This is the simplest and fastest dimension reduction method, but may lose important information in the truncated dimensions.
+  - **'RANDOM_PROJECTION'**: Uses Gaussian random projection with normally distributed random matrix. This method preserves relative distances between vectors while reducing dimensionality, following the Johnson-Lindenstrauss lemma.
+  - **'SPARSE_RANDOM_PROJECTION'**: Uses sparse random projection where matrix elements are mostly zero (±√3, 0). This is more computationally efficient than regular random projection while maintaining similar distance preservation properties.
+
+**Returns:** VECTOR type with reduced dimensions
+
+**Example:**
+```sql
+SELECT id, VECTOR_REDUCE(embedding, 256, 'TRUNCATE') as reduced_embedding FROM table
+SELECT id, VECTOR_REDUCE(embedding, 128, 'RANDOM_PROJECTION') as reduced_embedding FROM table
+SELECT id, VECTOR_REDUCE(embedding, 64, 'SPARSE_RANDOM_PROJECTION') as reduced_embedding FROM table
+```
+
+### VECTOR_NORMALIZE
+
+```VECTOR_NORMALIZE(vector_field)```
+
+Normalizes a vector to unit length (magnitude = 1). This is useful for computing cosine similarity.
+
+**Parameters:**
+- `vector_field`: The vector field to normalize (VECTOR type)
+
+**Returns:** VECTOR type - the normalized vector
+
+**Example:**
+```sql
+SELECT id, VECTOR_NORMALIZE(embedding) as normalized_embedding FROM table
+```
+

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

@@ -1215,4 +1215,43 @@ L1_DISTANCE(vector1, vector2)
 
 示例:
 
-L2_DISTANCE(vector1, vector2)
+L2_DISTANCE(vector1, vector2)
+
+### VECTOR_REDUCE
+
+```VECTOR_REDUCE(vector_field, target_dimension, method)```
+
+通用向量降维函数，支持多种降维方法。
+
+**参数:**
+- `vector_field`: 要降维的向量字段 (VECTOR 类型)
+- `target_dimension`: 目标维度 (INTEGER，必须小于源维度)
+- `method`: 降维方法 (STRING)：
+  - **'TRUNCATE'**: 截断法，通过保留前N个元素来缩减向量维度。这是最简单、最快速的降维方法，但可能会丢失被截断维度中的重要信息。
+  - **'RANDOM_PROJECTION'**: 随机投影法，使用高斯随机投影和正态分布的随机矩阵。该方法在降维的同时保持向量间的相对距离，遵循Johnson-Lindenstrauss引理。
+  - **'SPARSE_RANDOM_PROJECTION'**: 稀疏随机投影法，矩阵元素大多为零（±√3, 0）。比常规随机投影在计算上更高效，同时保持相似的距离保持特性。
+
+**返回值:** 降维后的 VECTOR 类型
+
+**示例:**
+```sql
+SELECT id, VECTOR_REDUCE(embedding, 256, 'TRUNCATE') as reduced_embedding FROM table
+SELECT id, VECTOR_REDUCE(embedding, 128, 'RANDOM_PROJECTION') as reduced_embedding FROM table
+SELECT id, VECTOR_REDUCE(embedding, 64, 'SPARSE_RANDOM_PROJECTION') as reduced_embedding FROM table
+```
+
+### VECTOR_NORMALIZE
+
+```VECTOR_NORMALIZE(vector_field)```
+
+将向量归一化为单位长度（模长 = 1）。这对于计算余弦相似度很有用。
+
+**参数:**
+- `vector_field`: 要归一化的向量字段 (VECTOR 类型)
+
+**返回值:** VECTOR 类型 - 归一化后的向量
+
+**示例:**
+```sql
+SELECT id, VECTOR_NORMALIZE(embedding) as normalized_embedding FROM table
+```

seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/java/org/apache/seatunnel/e2e/transform/TestSQLIT.java

@@ -86,6 +86,18 @@ public void testSQLTransform(TestContainer container) throws IOException, Interr
         Assertions.assertEquals(0, multiIfSql.getExitCode());
     }
 
+    @TestTemplate
+    @DisabledOnContainer(
+            value = {},
+            type = {EngineType.SPARK},
+            disabledReason = "Vector functions are not supported in Spark engine")
+    public void testVectorFunctions(TestContainer container)
+            throws IOException, InterruptedException {
+        Container.ExecResult vectorFunctionResult =
+                container.executeJob("/sql_transform/func_vector.conf");
+        Assertions.assertEquals(0, vectorFunctionResult.getExitCode());
+    }
+
     @TestTemplate
     public void testSQLTransformMultiTable(TestContainer container)
             throws IOException, InterruptedException {

seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/sql_transform/func_vector.conf

@@ -0,0 +1,142 @@
+#
+# 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.
+#
+######
+###### This config file is a demonstration of vector functions in SQL transform
+######
+
+env {
+  parallelism = 1
+  job.mode = "BATCH"
+  checkpoint.interval = 10000
+}
+
+source {
+  FakeSource {
+    plugin_output = "fake"
+    schema = {
+      fields {
+        id = "int"
+        name = "string"
+        vector_field = "array<float>"
+        vector_field2 = "array<float>"
+      }
+    }
+    rows = [
+      {
+        fields = [1, "test1", [1.0, 2.0, 3.0, 4.0, 5.0], [1.0, 2.0, 3.0, 4.0, 5.0]]
+        kind = INSERT
+      },
+      {
+        fields = [2, "test2", [2.0, 4.0, 6.0, 8.0, 10.0], [0.6, 0.8, 0.0, 0.0, 0.0]]
+        kind = INSERT
+      },
+      {
+        fields = [3, "test3", [3.0, 4.0, 0.0, 0.0, 0.0], [3.0, 4.0, 0.0, 0.0, 0.0]]
+        kind = INSERT
+      }
+    ]
+  }
+}
+
+transform {
+  Sql {
+    plugin_input = "fake"
+    plugin_output = "fake1"
+    query = """SELECT
+      id,
+      name,
+      VECTOR_DIMS(vector_field) as original_dim,
+      VECTOR_DIMS(VECTOR_REDUCE(vector_field, 3, 'TRUNCATE')) as truncated_dim,
+      VECTOR_DIMS(VECTOR_REDUCE(vector_field, 3, 'RANDOM_PROJECTION')) as projected_dim,
+      VECTOR_DIMS(VECTOR_REDUCE(vector_field, 3, 'SPARSE_RANDOM_PROJECTION')) as sparse_projected_dim,
+      VECTOR_DIMS(VECTOR_NORMALIZE(vector_field)) as normalized_dim
+    FROM dual"""
+  }
+}
+
+sink {
+  Assert {
+    plugin_input = "fake1"
+    rules = {
+      field_rules = [
+        {
+          field_name = "id"
+          field_type = "int"
+          field_value = [
+            {
+              rule_type = NOT_NULL
+            }
+          ]
+        },
+        {
+          field_name = "name"
+          field_type = "string"
+          field_value = [
+            {
+              rule_type = NOT_NULL
+            }
+          ]
+        },
+        {
+          field_name = "original_dim"
+          field_type = "int"
+          field_value = [
+            {equals_to = 5}
+          ]
+        },
+        {
+          field_name = "truncated_dim"
+          field_type = "int"
+          field_value = [
+            {equals_to = 3}
+          ]
+        },
+        {
+          field_name = "projected_dim"
+          field_type = "int"
+          field_value = [
+            {equals_to = 3}
+          ]
+        },
+        {
+          field_name = "sparse_projected_dim"
+          field_type = "int"
+          field_value = [
+            {equals_to = 3}
+          ]
+        },
+        {
+          field_name = "normalized_dim"
+          field_type = "int"
+          field_value = [
+            {equals_to = 5}
+          ]
+        }
+      ]
+      row_rules = [
+        {
+          rule_type = MAX_ROW
+          rule_value = 3
+        },
+        {
+          rule_type = MIN_ROW
+          rule_value = 3
+        }
+      ]
+    }
+  }
+}

seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLFunction.java

@@ -212,6 +212,9 @@ public class ZetaSQLFunction {
     public static final String VECTOR_NORM = "VECTOR_NORM";
     public static final String INNER_PRODUCT = "INNER_PRODUCT";
 
+    public static final String VECTOR_REDUCE = "VECTOR_REDUCE";
+    public static final String VECTOR_NORMALIZE = "VECTOR_NORMALIZE";
+
     private final SeaTunnelRowType inputRowType;
 
     private final ZetaSQLType zetaSQLType;
@@ -619,6 +622,11 @@ public Object executeFunctionExpr(
                 return VectorFunction.vectorNorm(args);
             case INNER_PRODUCT:
                 return VectorFunction.innerProduct(args);
+            case VECTOR_REDUCE:
+                return VectorFunction.vectorReduce(
+                        args.get(0), (Integer) args.get(1), (String) args.get(2));
+            case VECTOR_NORMALIZE:
+                return VectorFunction.vectorNormalize(args.get(0));
             default:
                 for (ZetaUDF udf : udfList) {
                     if (udf.functionName().equalsIgnoreCase(functionName)) {

seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLType.java

@@ -25,6 +25,7 @@
 import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
 import org.apache.seatunnel.api.table.type.SqlType;
+import org.apache.seatunnel.api.table.type.VectorType;
 import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated;
 import org.apache.seatunnel.transform.exception.TransformException;
 import org.apache.seatunnel.transform.sql.zeta.functions.ArrayFunction;
@@ -489,6 +490,10 @@ private SeaTunnelDataType<?> getFunctionType(Function function) {
             case ZetaSQLFunction.MOD:
                 // Result has the same type as second argument
                 return getExpressionType(function.getParameters().getExpressions().get(1));
+                // Vector functions
+            case ZetaSQLFunction.VECTOR_REDUCE:
+            case ZetaSQLFunction.VECTOR_NORMALIZE:
+                return VectorType.VECTOR_FLOAT_TYPE;
             default:
                 for (ZetaUDF udf : udfList) {
                     if (udf.functionName().equalsIgnoreCase(function.getName())) {