This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12337-128994ca0c6b6262b14037f5f90a390d513a4720 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 50315b321bc68d910352d78f0fcbff80872451f5 Author: Goutam Adwant <[email protected]> AuthorDate: Wed Oct 28 13:28:59 2026 +0000 [Fix][Transform-V2] Widen vector operands before arithmetic (#12337) Signed-off-by: Goutam Adwant <[email protected]> --- .../introduction/concepts/incompatible-changes.md | 14 ++++ docs/en/transforms/calcite.md | 8 ++ docs/en/transforms/sql-functions.md | 9 ++ .../introduction/concepts/incompatible-changes.md | 11 +++ docs/zh/transforms/calcite.md | 6 ++ docs/zh/transforms/sql-functions.md | 6 ++ .../calcite_builtin_functions.conf | 24 +++++- .../test/resources/sql_transform/func_vector.conf | 51 ++++++++++++ .../calcite/udf/CosineDistanceFunction.java | 8 +- .../calcite/udf/InnerProductFunction.java | 4 +- .../transform/calcite/udf/L1DistanceFunction.java | 2 +- .../transform/calcite/udf/L2DistanceFunction.java | 2 +- .../transform/calcite/udf/VectorNormFunction.java | 2 +- .../calcite/udf/VectorNormalizeFunction.java | 2 +- .../sql/zeta/functions/VectorFunction.java | 20 +++-- .../transform/calcite/CalciteSQLEngineTest.java | 36 ++++++++ .../transform/calcite/udf/VectorUdfTest.java | 69 +++++++++++++++ .../transform/sql/SQLVectorFunctionTest.java | 35 ++++++++ .../sql/zeta/functions/VectorFunctionTest.java | 97 ++++++++++++++++++++++ 19 files changed, 389 insertions(+), 17 deletions(-) diff --git a/docs/en/introduction/concepts/incompatible-changes.md b/docs/en/introduction/concepts/incompatible-changes.md index 0b50a23ed2..39bdf4731f 100644 --- a/docs/en/introduction/concepts/incompatible-changes.md +++ b/docs/en/introduction/concepts/incompatible-changes.md @@ -241,6 +241,20 @@ The Zeta SQL ARRAY function now declares TINYINT elements as `ARRAY<TINYINT>`, m (`existingConfigMap`) and restrict the member port (5801) with a `NetworkPolicy`. Restart the pods after upgrading so the new configuration is applied. +### SQL and Calcite Vector Arithmetic + +- **Behavior correction**: `VECTOR_NORM`, `INNER_PRODUCT`, `COSINE_DISTANCE`, + `L1_DISTANCE`, `L2_DISTANCE` and `VECTOR_NORMALIZE` now widen float elements before + intermediate multiplication or subtraction. Finite inputs no longer produce + incorrect infinity, NaN or zero merely because an intermediate float operation + overflows or underflows. Results can also differ in their low-order digits. +- **Migration**: Review thresholds and previously computed vector metrics or normalized + embeddings before replaying data. No options, types, schemas or state formats change. + Null inputs and actual zero vectors retain their handling. No non-finite input + validation or sanitization is added; corrected finite intermediates can also + affect results when mixed with non-finite elements. Normalized output elements + remain float, and random-projection arithmetic is unaffected. + ### SQL ARRAY_MAX and ARRAY_MIN Precision - **Behavior correction: exact integral and DECIMAL array extrema** diff --git a/docs/en/transforms/calcite.md b/docs/en/transforms/calcite.md index 53ba3107da..06fc6414f3 100644 --- a/docs/en/transforms/calcite.md +++ b/docs/en/transforms/calcite.md @@ -388,6 +388,14 @@ transform { ### Vector Operations +`VECTOR_NORM`, `INNER_PRODUCT`, `COSINE_DISTANCE`, `L1_DISTANCE`, `L2_DISTANCE` and +`VECTOR_NORMALIZE` use double-precision intermediate products and differences for +float vector elements, avoiding float overflow or underflow for finite inputs. +Normalized output elements remain float, so output rounding and underflow remain +possible. Null inputs and actual zero vectors retain their handling. No non-finite +input validation or sanitization is added; corrected finite intermediates can also +affect results when mixed with non-finite elements. + Use built-in vector UDFs to compute distances, reduce dimensions, or normalize vectors in a data pipeline (e.g., between Milvus/Qdrant source and sink). ```hocon diff --git a/docs/en/transforms/sql-functions.md b/docs/en/transforms/sql-functions.md index 09e1d85b02..0a3d10d52c 100644 --- a/docs/en/transforms/sql-functions.md +++ b/docs/en/transforms/sql-functions.md @@ -1443,6 +1443,15 @@ SELECT * FROM dual ## Vector Functions +`VECTOR_NORM`, `INNER_PRODUCT`, `COSINE_DISTANCE`, `L1_DISTANCE`, `L2_DISTANCE` and +`VECTOR_NORMALIZE` perform intermediate products and differences in double precision +for float vector elements. This avoids float overflow or underflow for finite large +or small elements. `VECTOR_NORMALIZE` still returns float elements; rounding and +underflow in the output representation remain possible. Null inputs and actual zero +vectors retain their handling. No non-finite input validation or sanitization is +added; corrected finite intermediates can also affect results when mixed with +non-finite elements. + Vector functions do not consume or modify their input vector buffers. The same vector field can be used in multiple expressions or as both arguments to a distance function, and remains available to downstream transforms and sinks. ### VECTOR_DIMS diff --git a/docs/zh/introduction/concepts/incompatible-changes.md b/docs/zh/introduction/concepts/incompatible-changes.md index b648d2c65b..24b14e119e 100644 --- a/docs/zh/introduction/concepts/incompatible-changes.md +++ b/docs/zh/introduction/concepts/incompatible-changes.md @@ -194,6 +194,17 @@ Zeta SQL ARRAY 函数现在将 TINYINT 元素声明为 `ARRAY<TINYINT>`，与实 并使用 `schema_save_mode = "ERROR_WHEN_SCHEMA_NOT_EXIST"` 保留该表结构。不要对手动定义的目标表使用 `RECREATE_SCHEMA`。任意指定索引前缀长度可能拒绝前缀相同但完整值不同的源主键，因此无法保持原有主键语义。 +### SQL 和 Calcite 向量运算 + +- **行为修正**：`VECTOR_NORM`、`INNER_PRODUCT`、`COSINE_DISTANCE`、`L1_DISTANCE`、 + `L2_DISTANCE` 和 `VECTOR_NORMALIZE` 在中间乘法或减法之前将 float 元素提升为 double。 + 有限输入不再仅因中间 float 运算溢出或下溢而得到错误的无穷大、NaN 或零， + 结果的低位数字也可能发生变化。 +- **迁移建议**：重放数据前，请检查阈值以及此前计算的向量指标或归一化向量。 + 配置项、类型、模式和状态格式均不变。空输入和实际零向量的处理方式不变。 + 未增加非有限输入的校验或清理策略；修正有限中间结果也可能影响与非有限元素混合时的结果。 + 归一化输出元素仍为 float，随机投影运算不受影响。 + ### Helm Chart：默认关闭 Zeta REST API v1 - **行为变更：Kubernetes Helm chart 不再默认开启未鉴权的 Zeta REST API v1** diff --git a/docs/zh/transforms/calcite.md b/docs/zh/transforms/calcite.md index 79bc1ecc33..b2bd2a1f52 100644 --- a/docs/zh/transforms/calcite.md +++ b/docs/zh/transforms/calcite.md @@ -388,6 +388,12 @@ transform { ### 向量运算 +`VECTOR_NORM`、`INNER_PRODUCT`、`COSINE_DISTANCE`、`L1_DISTANCE`、`L2_DISTANCE` 和 +`VECTOR_NORMALIZE` 对 float 向量元素使用 double 精度计算中间乘积和差值， +避免有限输入的 float 中间运算溢出或下溢。归一化输出元素仍为 float， +因此输出舍入或下溢仍可能发生。空输入和实际零向量的处理方式保持不变。 +未增加非有限输入的校验或清理策略；修正有限中间结果也可能影响与非有限元素混合时的结果。 + 使用内置向量 UDF 在数据管线中计算距离、降维或归一化（例如 Milvus/Qdrant 源与目标之间的处理）。 ```hocon diff --git a/docs/zh/transforms/sql-functions.md b/docs/zh/transforms/sql-functions.md index 460841af23..c21c958995 100644 --- a/docs/zh/transforms/sql-functions.md +++ b/docs/zh/transforms/sql-functions.md @@ -1437,6 +1437,12 @@ SELECT * FROM dual ## 向量函数 +`VECTOR_NORM`、`INNER_PRODUCT`、`COSINE_DISTANCE`、`L1_DISTANCE`、`L2_DISTANCE` 和 +`VECTOR_NORMALIZE` 对 float 向量元素使用 double 精度计算中间乘积和差值， +避免有限的大数或小数在 float 中间运算中溢出或下溢。`VECTOR_NORMALIZE` 仍返回 float +元素，输出表示仍可能发生舍入或下溢。空输入和实际零向量的处理方式保持不变。 +未增加非有限输入的校验或清理策略；修正有限中间结果也可能影响与非有限元素混合时的结果。 + 向量函数不会消耗或修改输入向量缓冲区。同一个向量字段可以用于多个表达式，也可以同时作为距离函数的两个参数，并且仍可供下游转换和接收器读取。 ### VECTOR_DIMS diff --git a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/calcite_transform/calcite_builtin_functions.conf b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/calcite_transform/calcite_builtin_functions.conf index ead73d13f4..8fece4eb88 100644 --- a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/calcite_transform/calcite_builtin_functions.conf +++ b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/calcite_transform/calcite_builtin_functions.conf @@ -37,7 +37,19 @@ transform { Calcite { plugin_input = "fake" plugin_output = "calcite_out" - sql = "SELECT id, UPPER(name) AS name_upper, ABS(age) AS abs_age, COALESCE(name, 'unknown') AS safe_name FROM fake" + # Big-endian float encodings: Float.MAX_VALUE, -Float.MAX_VALUE and Float.MIN_VALUE. + sql = """SELECT id, UPPER(name) AS name_upper, ABS(age) AS abs_age, COALESCE(name, 'unknown') AS safe_name, + VECTOR_NORM(X'7F7FFFFF') > 3.0e38 AND VECTOR_NORM(X'7F7FFFFF') < 4.0e38 AS large_norm, + INNER_PRODUCT(X'7F7FFFFF', X'7F7FFFFF') > 1.0e77 AND INNER_PRODUCT(X'7F7FFFFF', X'7F7FFFFF') < 2.0e77 AS large_product, + COSINE_DISTANCE(X'7F7FFFFF', X'FF7FFFFF') > 1.999 AND COSINE_DISTANCE(X'7F7FFFFF', X'FF7FFFFF') < 2.001 AS large_cosine, + L1_DISTANCE(X'7F7FFFFF', X'FF7FFFFF') > 6.0e38 AND L1_DISTANCE(X'7F7FFFFF', X'FF7FFFFF') < 7.0e38 AS large_l1, + L2_DISTANCE(X'7F7FFFFF', X'FF7FFFFF') > 6.0e38 AND L2_DISTANCE(X'7F7FFFFF', X'FF7FFFFF') < 7.0e38 AS large_l2, + VECTOR_NORM(X'00000001') > 1.0e-45 AND VECTOR_NORM(X'00000001') < 2.0e-45 AS tiny_norm, + INNER_PRODUCT(X'00000001', X'00000001') > 1.0e-90 AND INNER_PRODUCT(X'00000001', X'00000001') < 3.0e-90 AS tiny_product, + COSINE_DISTANCE(X'00000001', X'00000001') > -0.001 AND COSINE_DISTANCE(X'00000001', X'00000001') < 0.001 AS tiny_cosine, + VECTOR_NORM(VECTOR_NORMALIZE(X'7F7FFFFF')) = 1.0 AS large_normalized, + VECTOR_NORM(VECTOR_NORMALIZE(X'00000001')) = 1.0 AS tiny_normalized + FROM fake""" } } @@ -52,6 +64,16 @@ sink { } ] field_rules = [ + {field_name = "large_norm", field_type = "boolean", field_value = [{equals_to = true}]}, + {field_name = "large_product", field_type = "boolean", field_value = [{equals_to = true}]}, + {field_name = "large_cosine", field_type = "boolean", field_value = [{equals_to = true}]}, + {field_name = "large_l1", field_type = "boolean", field_value = [{equals_to = true}]}, + {field_name = "large_l2", field_type = "boolean", field_value = [{equals_to = true}]}, + {field_name = "tiny_norm", field_type = "boolean", field_value = [{equals_to = true}]}, + {field_name = "tiny_product", field_type = "boolean", field_value = [{equals_to = true}]}, + {field_name = "tiny_cosine", field_type = "boolean", field_value = [{equals_to = true}]}, + {field_name = "large_normalized", field_type = "boolean", field_value = [{equals_to = true}]}, + {field_name = "tiny_normalized", field_type = "boolean", field_value = [{equals_to = true}]}, { field_name = name_upper field_type = string diff --git a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/sql_transform/func_vector.conf b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/sql_transform/func_vector.conf index 5fd28c53fb..4c5879b0f2 100644 --- a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/sql_transform/func_vector.conf +++ b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/sql_transform/func_vector.conf @@ -25,6 +25,20 @@ env { } source { + FakeSource { + plugin_output = "arithmetic_input" + schema = { + fields { + large = "array<float>" + opposite = "array<float>" + tiny = "array<float>" + } + } + rows = [{ + fields = [[3.0e38, -3.0e38], [-3.0e38, 3.0e38], [1.0e-30, -1.0e-30]] + kind = INSERT + }] + } FakeSource { plugin_output = "fake" schema = { @@ -53,6 +67,22 @@ source { } transform { + Sql { + plugin_input = "arithmetic_input" + plugin_output = "arithmetic_result" + query = """SELECT + CASE WHEN VECTOR_NORM(large) > 4.0e38 AND VECTOR_NORM(large) < 5.0e38 THEN 1 ELSE 0 END AS large_norm, + CASE WHEN INNER_PRODUCT(large, opposite) > -2.0e77 AND INNER_PRODUCT(large, opposite) < -1.0e77 THEN 1 ELSE 0 END AS large_product, + CASE WHEN COSINE_DISTANCE(large, opposite) > 1.999 AND COSINE_DISTANCE(large, opposite) < 2.001 THEN 1 ELSE 0 END AS large_cosine, + CASE WHEN L1_DISTANCE(large, opposite) > 1.0e39 AND L1_DISTANCE(large, opposite) < 1.3e39 THEN 1 ELSE 0 END AS large_l1, + CASE WHEN L2_DISTANCE(large, opposite) > 8.0e38 AND L2_DISTANCE(large, opposite) < 9.0e38 THEN 1 ELSE 0 END AS large_l2, + CASE WHEN VECTOR_NORM(tiny) > 1.0e-30 AND VECTOR_NORM(tiny) < 2.0e-30 THEN 1 ELSE 0 END AS tiny_norm, + CASE WHEN INNER_PRODUCT(tiny, tiny) > 1.0e-60 AND INNER_PRODUCT(tiny, tiny) < 3.0e-60 THEN 1 ELSE 0 END AS tiny_product, + CASE WHEN COSINE_DISTANCE(tiny, tiny) > -0.001 AND COSINE_DISTANCE(tiny, tiny) < 0.001 THEN 1 ELSE 0 END AS tiny_cosine, + CASE WHEN VECTOR_NORM(VECTOR_NORMALIZE(large)) > 0.999 AND VECTOR_NORM(VECTOR_NORMALIZE(large)) < 1.001 THEN 1 ELSE 0 END AS large_normalized, + CASE WHEN VECTOR_NORM(VECTOR_NORMALIZE(tiny)) > 0.999 AND VECTOR_NORM(VECTOR_NORMALIZE(tiny)) < 1.001 THEN 1 ELSE 0 END AS tiny_normalized + FROM dual""" + } Sql { plugin_input = "fake" plugin_output = "vector_buffers" @@ -78,6 +108,27 @@ transform { } sink { + Assert { + plugin_input = "arithmetic_result" + rules { + field_rules = [ + {field_name = "large_norm", field_type = "int", field_value = [{equals_to = 1}]}, + {field_name = "large_product", field_type = "int", field_value = [{equals_to = 1}]}, + {field_name = "large_cosine", field_type = "int", field_value = [{equals_to = 1}]}, + {field_name = "large_l1", field_type = "int", field_value = [{equals_to = 1}]}, + {field_name = "large_l2", field_type = "int", field_value = [{equals_to = 1}]}, + {field_name = "tiny_norm", field_type = "int", field_value = [{equals_to = 1}]}, + {field_name = "tiny_product", field_type = "int", field_value = [{equals_to = 1}]}, + {field_name = "tiny_cosine", field_type = "int", field_value = [{equals_to = 1}]}, + {field_name = "large_normalized", field_type = "int", field_value = [{equals_to = 1}]}, + {field_name = "tiny_normalized", field_type = "int", field_value = [{equals_to = 1}]} + ] + row_rules = [ + {rule_type = MAX_ROW, rule_value = 1}, + {rule_type = MIN_ROW, rule_value = 1} + ] + } + } Assert { plugin_input = "fake1" rules = { diff --git a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/calcite/udf/CosineDistanceFunction.java b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/calcite/udf/CosineDistanceFunction.java index 9f19dabf2b..130b88c628 100644 --- a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/calcite/udf/CosineDistanceFunction.java +++ b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/calcite/udf/CosineDistanceFunction.java @@ -52,9 +52,11 @@ public class CosineDistanceFunction implements CalciteUdf { vector1.length, vector2.length)); } double dotProduct = - IntStream.range(0, vector1.length).mapToDouble(i -> vector1[i] * vector2[i]).sum(); - double norm1 = Arrays.stream(vector1).mapToDouble(v -> v * v).sum(); - double norm2 = Arrays.stream(vector2).mapToDouble(v -> v * v).sum(); + IntStream.range(0, vector1.length) + .mapToDouble(i -> (double) vector1[i] * vector2[i]) + .sum(); + double norm1 = Arrays.stream(vector1).mapToDouble(v -> (double) v * v).sum(); + double norm2 = Arrays.stream(vector2).mapToDouble(v -> (double) v * v).sum(); if (norm1 == 0.0 || norm2 == 0.0) { return 1.0; } diff --git a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/calcite/udf/InnerProductFunction.java b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/calcite/udf/InnerProductFunction.java index 3481b07dab..ec54ac51c9 100644 --- a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/calcite/udf/InnerProductFunction.java +++ b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/calcite/udf/InnerProductFunction.java @@ -49,6 +49,8 @@ public class InnerProductFunction implements CalciteUdf { "Vectors must have the same dimension: %d vs %d", vector1.length, vector2.length)); } - return IntStream.range(0, vector1.length).mapToDouble(i -> vector1[i] * vector2[i]).sum(); + return IntStream.range(0, vector1.length) + .mapToDouble(i -> (double) vector1[i] * vector2[i]) + .sum(); } } diff --git a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/calcite/udf/L1DistanceFunction.java b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/calcite/udf/L1DistanceFunction.java index cd23ea03f6..1ce91d2ac7 100644 --- a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/calcite/udf/L1DistanceFunction.java +++ b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/calcite/udf/L1DistanceFunction.java @@ -50,7 +50,7 @@ public class L1DistanceFunction implements CalciteUdf { vector1.length, vector2.length)); } return IntStream.range(0, vector1.length) - .mapToDouble(i -> Math.abs(vector1[i] - vector2[i])) + .mapToDouble(i -> Math.abs((double) vector1[i] - vector2[i])) .sum(); } } diff --git a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/calcite/udf/L2DistanceFunction.java b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/calcite/udf/L2DistanceFunction.java index 25fa89e46f..0a52e12a3b 100644 --- a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/calcite/udf/L2DistanceFunction.java +++ b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/calcite/udf/L2DistanceFunction.java @@ -53,7 +53,7 @@ public class L2DistanceFunction implements CalciteUdf { IntStream.range(0, vector1.length) .mapToDouble( i -> { - double diff = vector1[i] - vector2[i]; + double diff = (double) vector1[i] - vector2[i]; return diff * diff; }) .sum(); diff --git a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/calcite/udf/VectorNormFunction.java b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/calcite/udf/VectorNormFunction.java index 4288b1839c..04a04857a9 100644 --- a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/calcite/udf/VectorNormFunction.java +++ b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/calcite/udf/VectorNormFunction.java @@ -42,6 +42,6 @@ public class VectorNormFunction implements CalciteUdf { return null; } Float[] vector = VectorUtils.toFloatArray(ByteBuffer.wrap(v)); - return Math.sqrt(Arrays.stream(vector).mapToDouble(val -> val * val).sum()); + return Math.sqrt(Arrays.stream(vector).mapToDouble(val -> (double) val * val).sum()); } } diff --git a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/calcite/udf/VectorNormalizeFunction.java b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/calcite/udf/VectorNormalizeFunction.java index 9ef0f4702f..f42a20c3d6 100644 --- a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/calcite/udf/VectorNormalizeFunction.java +++ b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/calcite/udf/VectorNormalizeFunction.java @@ -44,7 +44,7 @@ public class VectorNormalizeFunction implements CalciteUdf { double magnitude = 0.0; for (Float value : vector) { if (value != null) { - magnitude += value * value; + magnitude += (double) value * value; } } magnitude = Math.sqrt(magnitude); diff --git a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/functions/VectorFunction.java b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/functions/VectorFunction.java index d75dc21534..cddabf0286 100644 --- a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/functions/VectorFunction.java +++ b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/functions/VectorFunction.java @@ -55,9 +55,11 @@ public class VectorFunction { vector1.length, vector2.length)); } double dotProduct = - IntStream.range(0, vector1.length).mapToDouble(i -> vector1[i] * vector2[i]).sum(); - double norm1 = Arrays.stream(vector1).mapToDouble(v -> v * v).sum(); - double norm2 = Arrays.stream(vector2).mapToDouble(v -> v * v).sum(); + IntStream.range(0, vector1.length) + .mapToDouble(i -> (double) vector1[i] * vector2[i]) + .sum(); + double norm1 = Arrays.stream(vector1).mapToDouble(v -> (double) v * v).sum(); + double norm2 = Arrays.stream(vector2).mapToDouble(v -> (double) v * v).sum(); if (norm1 == 0.0 || norm2 == 0.0) { return 1.0; } @@ -88,7 +90,9 @@ public class VectorFunction { "Vectors must have the same dimension: %d vs %d", v1.length, v2.length)); } - return IntStream.range(0, v1.length).mapToDouble(i -> Math.abs(v1[i] - v2[i])).sum(); + return IntStream.range(0, v1.length) + .mapToDouble(i -> Math.abs((double) v1[i] - v2[i])) + .sum(); } public static Object l2Distance(List<Object> args) { @@ -117,7 +121,7 @@ public class VectorFunction { IntStream.range(0, v1.length) .mapToDouble( i -> { - double diff = v1[i] - v2[i]; + double diff = (double) v1[i] - v2[i]; return diff * diff; }) .sum(); @@ -153,7 +157,7 @@ public class VectorFunction { return null; } Float[] vector = convertToFloatArray(arg); - return Math.sqrt(Arrays.stream(vector).mapToDouble(v -> v * v).sum()); + return Math.sqrt(Arrays.stream(vector).mapToDouble(v -> (double) v * v).sum()); } public static Object innerProduct(List<Object> args) { @@ -179,7 +183,7 @@ public class VectorFunction { v1.length, v2.length)); } - return IntStream.range(0, v1.length).mapToDouble(i -> v1[i] * v2[i]).sum(); + return IntStream.range(0, v1.length).mapToDouble(i -> (double) v1[i] * v2[i]).sum(); } private static Float[] convertToFloatArray(Object obj) { @@ -292,7 +296,7 @@ public class VectorFunction { double magnitude = 0.0; for (Float value : vector) { if (value != null) { - magnitude += value * value; + magnitude += (double) value * value; } } magnitude = Math.sqrt(magnitude); diff --git a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/calcite/CalciteSQLEngineTest.java b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/calcite/CalciteSQLEngineTest.java index efbb2e43e7..19ff74249a 100644 --- a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/calcite/CalciteSQLEngineTest.java +++ b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/calcite/CalciteSQLEngineTest.java @@ -4057,6 +4057,42 @@ class CalciteSQLEngineTest { }); } + @Test + void testFiniteVectorArithmetic() { + double value = Float.MAX_VALUE; + String[] expressions = { + "VECTOR_NORM(vec1)", + "INNER_PRODUCT(vec1, vec2)", + "COSINE_DISTANCE(vec1, vec2)", + "L1_DISTANCE(vec1, vec2)", + "L2_DISTANCE(vec1, vec2)", + "VECTOR_NORM(VECTOR_NORMALIZE(vec1))" + }; + double[] expected = {value, -value * value, 2.0, 2 * value, 2 * value, 1.0}; + for (int i = 0; i < expressions.length; i++) { + CalciteSQLEngine engine = + createAndInit( + "SELECT " + expressions[i] + " AS vector_result FROM t", + "t", + twoVectorRowType()); + try { + Object actual = + singleField( + engine, + new Object[] { + floatVec(Float.MAX_VALUE), floatVec(-Float.MAX_VALUE) + }); + Assertions.assertEquals( + expected[i], + ((Number) actual).doubleValue(), + Math.abs(expected[i]) * 1e-14, + expressions[i]); + } finally { + engine.close(); + } + } + } + private SeaTunnelRowType singleVectorRowType() { return new SeaTunnelRowType( new String[] {"vec"}, new SeaTunnelDataType[] {VectorType.VECTOR_FLOAT_TYPE}); diff --git a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/calcite/udf/VectorUdfTest.java b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/calcite/udf/VectorUdfTest.java index 519a717515..36242cfdf3 100644 --- a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/calcite/udf/VectorUdfTest.java +++ b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/calcite/udf/VectorUdfTest.java @@ -21,11 +21,80 @@ import org.apache.seatunnel.common.utils.VectorUtils; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import java.math.BigDecimal; import java.nio.ByteBuffer; class VectorUdfTest { + @ParameterizedTest + @ValueSource( + floats = {Float.MAX_VALUE, 1e20f, 1e-30f, Float.MIN_NORMAL, Float.MIN_VALUE, 3f, 0.1f}) + void testFiniteVectorArithmetic(float value) { + byte[] left = toBytes(new Float[] {value, -value}); + byte[] right = toBytes(new Float[] {-value, value}); + BigDecimal exact = new BigDecimal((double) value); + double squaredNorm = exact.multiply(exact).multiply(BigDecimal.valueOf(2)).doubleValue(); + double norm = Math.hypot(value, value); + double distance = exact.multiply(BigDecimal.valueOf(4)).doubleValue(); + Assertions.assertAll( + () -> Assertions.assertEquals(norm, VectorNormFunction.eval(left), norm * 1e-14), + () -> + Assertions.assertEquals( + -squaredNorm, + InnerProductFunction.eval(left, right), + squaredNorm * 1e-14), + () -> Assertions.assertEquals(2.0, CosineDistanceFunction.eval(left, right), 1e-14), + () -> + Assertions.assertEquals( + distance, L1DistanceFunction.eval(left, right), distance * 1e-14), + () -> + Assertions.assertEquals( + 2 * norm, L2DistanceFunction.eval(left, right), norm * 1e-14), + () -> { + Float[] normalized = fromBytes(VectorNormalizeFunction.eval(left)); + Assertions.assertEquals(1 / Math.sqrt(2), normalized[0], 1e-7); + Assertions.assertEquals(-1 / Math.sqrt(2), normalized[1], 1e-7); + }); + } + + @Test + void testLargeProductsCancelWithoutOverflow() { + byte[] left = toBytes(new Float[] {Float.MAX_VALUE, Float.MAX_VALUE}); + byte[] right = toBytes(new Float[] {Float.MAX_VALUE, -Float.MAX_VALUE}); + Assertions.assertEquals(0.0, InnerProductFunction.eval(left, right)); + Assertions.assertEquals(1.0, CosineDistanceFunction.eval(left, right)); + } + + @Test + void testVectorArithmeticSpecialValues() { + byte[] zero = toBytes(new Float[] {0f, -0f}); + byte[] empty = toBytes(new Float[] {}); + Assertions.assertEquals(0.0, VectorNormFunction.eval(zero)); + Assertions.assertEquals(1.0, CosineDistanceFunction.eval(zero, zero)); + Assertions.assertEquals(1.0, CosineDistanceFunction.eval(empty, empty)); + Assertions.assertSame(zero, VectorNormalizeFunction.eval(zero)); + Assertions.assertSame(empty, VectorNormalizeFunction.eval(empty)); + Assertions.assertTrue( + Double.isNaN( + CosineDistanceFunction.eval( + toBytes(new Float[] {Float.MIN_VALUE}), + toBytes(new Float[] {Float.POSITIVE_INFINITY})))); + for (float value : + new float[] {Float.NaN, Float.POSITIVE_INFINITY, Float.NEGATIVE_INFINITY}) { + byte[] vector = toBytes(new Float[] {value}); + Assertions.assertTrue(Double.isNaN(CosineDistanceFunction.eval(vector, vector))); + Assertions.assertTrue(Double.isNaN(L1DistanceFunction.eval(vector, vector))); + Assertions.assertTrue(Double.isNaN(L2DistanceFunction.eval(vector, vector))); + double norm = VectorNormFunction.eval(vector); + Assertions.assertTrue( + Float.isNaN(value) ? Double.isNaN(norm) : norm == Double.POSITIVE_INFINITY); + Assertions.assertTrue(Float.isNaN(fromBytes(VectorNormalizeFunction.eval(vector))[0])); + } + } + private static byte[] toBytes(Float[] floats) { ByteBuffer buf = VectorUtils.toByteBuffer(floats); byte[] bytes = new byte[buf.remaining()]; diff --git a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/SQLVectorFunctionTest.java b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/SQLVectorFunctionTest.java index 3e512e3b29..7fe0c50c09 100644 --- a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/SQLVectorFunctionTest.java +++ b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/SQLVectorFunctionTest.java @@ -46,6 +46,41 @@ public class SQLVectorFunctionTest { new String[] {"id", "vector_field", "vector_field2"}; private CatalogTable catalogTable; + @Test + public void testFiniteVectorArithmeticThroughSql() { + double value = Float.MAX_VALUE; + String[] expressions = { + "VECTOR_NORM(vector_field)", + "INNER_PRODUCT(vector_field, vector_field2)", + "COSINE_DISTANCE(vector_field, vector_field2)", + "L1_DISTANCE(vector_field, vector_field2)", + "L2_DISTANCE(vector_field, vector_field2)", + "VECTOR_NORM(VECTOR_NORMALIZE(vector_field))" + }; + double[] expected = {value, -value * value, 2.0, 2 * value, 2 * value, 1.0}; + for (int i = 0; i < expressions.length; i++) { + SQLTransform transform = + new SQLTransform( + ReadonlyConfig.fromMap( + Collections.singletonMap( + "query", + "SELECT " + expressions[i] + " AS result FROM dual")), + catalogTable); + SeaTunnelRow input = + new SeaTunnelRow( + new Object[] { + 1, + VectorUtils.toByteBuffer(new Float[] {Float.MAX_VALUE}), + VectorUtils.toByteBuffer(new Float[] {-Float.MAX_VALUE}) + }); + Assertions.assertEquals( + expected[i], + (double) transform.transformRow(input).get(0).getField(0), + Math.abs(expected[i]) * 1e-14, + expressions[i]); + } + } + @BeforeEach void setUp() { SeaTunnelRowType rowType = diff --git a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/zeta/functions/VectorFunctionTest.java b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/zeta/functions/VectorFunctionTest.java index a5d3fc705e..9c4672ffe3 100644 --- a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/zeta/functions/VectorFunctionTest.java +++ b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/zeta/functions/VectorFunctionTest.java @@ -31,7 +31,9 @@ import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.MethodSource; +import org.junit.jupiter.params.provider.ValueSource; +import java.math.BigDecimal; import java.nio.BufferUnderflowException; import java.nio.ByteBuffer; import java.nio.ByteOrder; @@ -43,6 +45,101 @@ import java.util.stream.Stream; public class VectorFunctionTest { + @ParameterizedTest + @ValueSource( + floats = {Float.MAX_VALUE, 1e20f, 1e-30f, Float.MIN_NORMAL, Float.MIN_VALUE, 3f, 0.1f}) + public void testFiniteVectorArithmetic(float value) { + Float[] left = {value, -value}; + Float[] right = {-value, value}; + BigDecimal exact = new BigDecimal((double) value); + double squaredNorm = exact.multiply(exact).multiply(BigDecimal.valueOf(2)).doubleValue(); + double norm = Math.hypot(value, value); + double distance = exact.multiply(BigDecimal.valueOf(4)).doubleValue(); + Assertions.assertAll( + () -> + Assertions.assertEquals( + norm, + (double) VectorFunction.vectorNorm(Collections.singletonList(left)), + norm * 1e-14), + () -> + Assertions.assertEquals( + -squaredNorm, + (double) VectorFunction.innerProduct(Arrays.asList(left, right)), + squaredNorm * 1e-14), + () -> + Assertions.assertEquals( + 2.0, + (double) VectorFunction.cosineDistance(Arrays.asList(left, right)), + 1e-14), + () -> + Assertions.assertEquals( + distance, + (double) VectorFunction.l1Distance(Arrays.asList(left, right)), + distance * 1e-14), + () -> + Assertions.assertEquals( + 2 * norm, + (double) VectorFunction.l2Distance(Arrays.asList(left, right)), + norm * 1e-14), + () -> { + Float[] normalized = + VectorUtils.toFloatArray( + (ByteBuffer) VectorFunction.vectorNormalize(left)); + Assertions.assertEquals(1 / Math.sqrt(2), normalized[0], 1e-7); + Assertions.assertEquals(-1 / Math.sqrt(2), normalized[1], 1e-7); + }); + } + + @Test + public void testLargeProductsCancelWithoutOverflow() { + Float[] left = {Float.MAX_VALUE, Float.MAX_VALUE}; + Float[] right = {Float.MAX_VALUE, -Float.MAX_VALUE}; + Assertions.assertEquals(0.0, VectorFunction.innerProduct(Arrays.asList(left, right))); + Assertions.assertEquals(1.0, VectorFunction.cosineDistance(Arrays.asList(left, right))); + } + + @Test + public void testVectorArithmeticSpecialValues() { + Float[] zero = {0f, -0f}; + Float[] empty = {}; + Assertions.assertEquals(0.0, VectorFunction.vectorNorm(Collections.singletonList(zero))); + Assertions.assertEquals(1.0, VectorFunction.cosineDistance(Arrays.asList(zero, zero))); + Assertions.assertEquals(1.0, VectorFunction.cosineDistance(Arrays.asList(empty, empty))); + Assertions.assertSame(zero, VectorFunction.vectorNormalize(zero)); + Assertions.assertSame(empty, VectorFunction.vectorNormalize(empty)); + Assertions.assertTrue( + Double.isNaN( + (double) + VectorFunction.cosineDistance( + Arrays.asList( + new Float[] {Float.MIN_VALUE}, + new Float[] {Float.POSITIVE_INFINITY})))); + Assertions.assertNull(VectorFunction.vectorNorm(Collections.singletonList(null))); + Assertions.assertNull(VectorFunction.cosineDistance(Arrays.asList(zero, null))); + Assertions.assertNull(VectorFunction.innerProduct(Arrays.asList(null, zero))); + Assertions.assertNull(VectorFunction.l1Distance(Arrays.asList(zero, null))); + Assertions.assertNull(VectorFunction.l2Distance(Arrays.asList(null, zero))); + for (float value : + new float[] {Float.NaN, Float.POSITIVE_INFINITY, Float.NEGATIVE_INFINITY}) { + Float[] vector = {value}; + Assertions.assertTrue( + Double.isNaN( + (double) VectorFunction.cosineDistance(Arrays.asList(vector, vector)))); + Assertions.assertTrue( + Double.isNaN( + (double) VectorFunction.l1Distance(Arrays.asList(vector, vector)))); + Assertions.assertTrue( + Double.isNaN( + (double) VectorFunction.l2Distance(Arrays.asList(vector, vector)))); + double norm = (double) VectorFunction.vectorNorm(Collections.singletonList(vector)); + Assertions.assertTrue( + Float.isNaN(value) ? Double.isNaN(norm) : norm == Double.POSITIVE_INFINITY); + Float[] normalized = + VectorUtils.toFloatArray((ByteBuffer) VectorFunction.vectorNormalize(vector)); + Assertions.assertTrue(Float.isNaN(normalized[0])); + } + } + @ParameterizedTest @MethodSource("vectorBuffers") public void testVectorReadsPreserveBuffer(ByteBuffer buffer) {
