This is an automated email from the ASF dual-hosted git repository.
nzw921rx 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 403efeee0b [Fix][Transform-V2] Preserve exact numeric comparisons in
SQL predicates (#12318)
403efeee0b is described below
commit 403efeee0b3806c77e55c38368abdd41fbf828bd
Author: Goutam Adwant <[email protected]>
AuthorDate: Tue Oct 27 10:09:31 2026 -0700
[Fix][Transform-V2] Preserve exact numeric comparisons in SQL predicates
(#12318)
Signed-off-by: Goutam Adwant <[email protected]>
---
.../introduction/concepts/incompatible-changes.md | 26 ++
docs/en/transforms/sql.md | 40 +++
.../introduction/concepts/incompatible-changes.md | 21 ++
docs/zh/transforms/sql.md | 33 ++
.../src/test/resources/sql_transform.conf | 15 +-
.../transform/sql/zeta/ZetaSQLFilter.java | 62 +++-
.../transform/sql/SQLNumericComparisonTest.java | 341 +++++++++++++++++++++
7 files changed, 530 insertions(+), 8 deletions(-)
diff --git a/docs/en/introduction/concepts/incompatible-changes.md
b/docs/en/introduction/concepts/incompatible-changes.md
index 3984ea546a..9b699028c5 100644
--- a/docs/en/introduction/concepts/incompatible-changes.md
+++ b/docs/en/introduction/concepts/incompatible-changes.md
@@ -266,6 +266,32 @@ The Zeta SQL ARRAY function now declares TINYINT elements
as `ARRAY<TINYINT>`, m
values and previously ran successfully will now fail fast at startup with a
validation
error.
+### SQL Numeric Predicate Precision
+
+- **Behavior change: exact numeric operands no longer compare through
`double`**
+ - **Affected component**: `seatunnel-transforms-v2`, `Sql` transform (`ZETA`
and `INTERNAL`).
+ - **Description**: Comparisons and `IN`/`NOT IN` checks preserve precision
when both
+ evaluated operands are integral or `DECIMAL`, including comparisons in
searched
+ and simple `CASE` expressions. For example, `BIGINT` `9007199254740993` no
longer
+ equals the integer literal `9007199254740992`.
+ - **Impact**: Filters can select different rows, and `CASE` expressions can
choose
+ different branches. Comparisons involving `FLOAT` or `DOUBLE` retain their
existing
+ behavior. Literals containing a decimal point still evaluate as `DOUBLE`.
Unary minus
+ preserves `DECIMAL` precision for both `-decimal_col` and
+ `-CAST('...' AS DECIMAL(p, s))`. Each `IN` element independently selects
the
+ exact or floating path, so mixing exact and floating operands can still
match
+ distinct large values and does not guarantee transitive equality.
+ - **Migration Guide**: Review affected predicates and `CASE` expressions,
and reconcile
+ previously written data that relied on rounded comparison results. For
exact decimal
+ constants, use quoted strings with sufficient scale, such as
+ `CAST('-123456789012345678.99' AS DECIMAL(38, 2))` or
+ `-CAST('123456789012345678.99' AS DECIMAL(38, 2))`. Existing CAST rounding
is unchanged:
+ string/`DECIMAL` inputs use
+ `CEILING` when reducing scale, so negation before or after CAST can
produce different results.
+ `FLOAT`/`DOUBLE` inputs use `HALF_UP` and cannot recover
+ precision already lost to floating-point evaluation. See [Numeric
comparisons](../../transforms/sql.md#numeric-comparisons)
+ for examples and limits. No configuration, output schema or state format
changes are required.
+
### DataValidator Array Length
The `LENGTH` rule now counts array elements rather than characters in the
diff --git a/docs/en/transforms/sql.md b/docs/en/transforms/sql.md
index 6b6691b219..be4c6ffff3 100644
--- a/docs/en/transforms/sql.md
+++ b/docs/en/transforms/sql.md
@@ -32,6 +32,46 @@ or `select c_row.c_inner_row.column_b` to query the inline
struct column that na
The SQL engine used by this transform. Supported values are `ZETA` and
`INTERNAL`. If this option is not configured, `ZETA` is used.
+## Numeric comparisons
+
+Comparisons preserve numeric precision when both evaluated operands are
integral
+types (`TINYINT`, `SMALLINT`, `INT`, `BIGINT`) or `DECIMAL`. This applies to
`=`,
+`!=`, `<>`, `<`, `<=`, `>` and `>=`, numeric `IN` and `NOT IN` membership
checks,
+and comparisons in searched and simple `CASE` expressions.
+For example, a `BIGINT` value of `9007199254740993` does not match
+`WHERE id = 9007199254740992`.
+
+If either operand is `FLOAT` or `DOUBLE`, comparisons continue to use
+floating-point conversion, retaining the existing behavior for NaN, infinity
+and signed zero. Numeric literals containing a decimal point are evaluated as
+`DOUBLE`, not `DECIMAL`: comparing a `DECIMAL` column to
+`123456789012345678.98` can still match `123456789012345678.99`. Integer
literals
+evaluated as integral values, such as `-9007199254740993`, use the exact path
+when the other operand is also exact. Unary minus preserves `DECIMAL`
precision,
+so both `-decimal_col` and `-CAST('...' AS DECIMAL(p, s))` retain exact decimal
+comparison semantics when the other operand is also exact.
+
+For an exact decimal constant, use a quoted string with sufficient scale, for
+example `amount = CAST('123456789012345678.99' AS DECIMAL(38, 2))`.
+For a negative constant, either `CAST('-123456789012345678.99' AS DECIMAL(38,
2))`
+or `-CAST('123456789012345678.99' AS DECIMAL(38, 2))` preserves decimal
precision.
+Existing CAST rounding is unchanged: converting a string or `DECIMAL` to
+`DECIMAL(p, s)` uses `CEILING` (toward positive infinity) if scale is reduced,
+so `CAST('1.231' AS DECIMAL(10, 2))` is `1.24` and
+`CAST('-1.239' AS DECIMAL(10, 2))` is `-1.23`. Negating after CAST instead
gives
+`-1.24` for `-CAST('1.239' AS DECIMAL(10, 2))`: reducing scale rounds before
+negation, so the two forms are not interchangeable in that case. Converting
+`FLOAT` or `DOUBLE` uses `HALF_UP`; casting a fractional numeric literal
cannot recover precision
+already lost during its evaluation as `DOUBLE`.
+
+The exact-or-floating choice is made for each operand pair, including each
+element of an `IN` list. For `id = 9007199254740993`,
+`id IN (9007199254740992)` is false, but
+`id IN (9007199254740992, 9007199254740992.0)` is true because the second
+comparison uses `DOUBLE`. Equality is therefore not necessarily transitive
+across mixed exact and floating types; use exact operands consistently when
+precision matters. These rules apply to both `ZETA` and `INTERNAL`.
+
## Example
The data read from source is a table like this:
diff --git a/docs/zh/introduction/concepts/incompatible-changes.md
b/docs/zh/introduction/concepts/incompatible-changes.md
index 3a654648af..7a5ceb6b78 100644
--- a/docs/zh/introduction/concepts/incompatible-changes.md
+++ b/docs/zh/introduction/concepts/incompatible-changes.md
@@ -224,6 +224,27 @@ Zeta SQL ARRAY 函数现在将 TINYINT 元素声明为 `ARRAY<TINYINT>`，与实
必须满足 min <= max。注意 `row.num = 0`（空 Source）仍然有效。此前设置了无效值且成功运行的现有作业，
将在启动时快速抛出校验错误并失败。
+### SQL 数值条件的精度
+
+- **行为变更：精确数值操作数不再通过 `double` 进行比较**
+ - **影响范围**：`seatunnel-transforms-v2`，`Sql` 转换（`ZETA` 和 `INTERNAL`）。
+ - **变更说明**：当求值后的两个操作数均为整型或 `DECIMAL` 时，比较及 `IN`/`NOT IN` 判断保留精度，
+ 包括搜索式和简单式 `CASE` 表达式中的比较。例如，`BIGINT` 值 `9007199254740993`
+ 不再等于整数字面量 `9007199254740992`。
+ - **影响**：过滤条件可能选择不同的行，`CASE` 表达式也可能选择不同的分支。
+ 涉及 `FLOAT` 或 `DOUBLE` 的比较保持原有行为。包含小数点的字面量仍求值为 `DOUBLE`，
+ 一元负号会保留 `DECIMAL` 精度，`-decimal_col` 和 `-CAST('...' AS DECIMAL(p, s))` 均如此。
+ 每个 `IN` 元素独立选择精确或浮点比较，
+ 因此混用精确类型和浮点类型时，仍可能匹配不同的大数值，且等号不保证传递性。
+ - **迁移指南**：请检查相关条件和 `CASE` 表达式，并核对、修正依赖此前舍入比较结果的已写入数据。
+ 如需精确的小数常量，请使用带引号的字符串并保留足够的小数位，例如
+ `CAST('-123456789012345678.99' AS DECIMAL(38, 2))` 或
+ `-CAST('123456789012345678.99' AS DECIMAL(38, 2))`。
+ 现有 CAST 舍入行为不变：字符串/`DECIMAL` 输入在减少小数位时使用 `CEILING`，
+ 因此在 CAST 之前或之后取负可能产生不同结果；
+ `FLOAT`/`DOUBLE` 输入使用 `HALF_UP`，且无法恢复浮点求值时已丢失的精度。
+ 示例与限制见[数值比较](../../transforms/sql.md#数值比较)。无需更改配置、输出结构或状态格式。
+
### DataValidator 数组长度
`LENGTH` 规则现在计算数组元素个数，而非数组的 Java 标识字符串长度。
diff --git a/docs/zh/transforms/sql.md b/docs/zh/transforms/sql.md
index f92ed74b4e..1760ebc0ea 100644
--- a/docs/zh/transforms/sql.md
+++ b/docs/zh/transforms/sql.md
@@ -32,6 +32,39 @@ SQL 转换使用内存中的 SQL 引擎，我们可以通过 SQL 函数和 SQL
该 Transform 使用的 SQL 引擎。支持 `ZETA` 和 `INTERNAL`。如果不配置，默认使用 `ZETA`。
+## 数值比较
+
+当求值后的两个操作数均为整型（`TINYINT`、`SMALLINT`、`INT`、`BIGINT`）或 `DECIMAL` 时，比较保留数值精度。
+此规则适用于 `=`、`!=`、`<>`、`<`、`<=`、`>`、`>=`、数值的 `IN` 和 `NOT IN` 判断，
+以及搜索式和简单式 `CASE` 表达式中的比较。
+例如，`BIGINT` 值 `9007199254740993` 不会匹配 `WHERE id = 9007199254740992`。
+
+当任一操作数为 `FLOAT` 或 `DOUBLE` 时，仍使用浮点转换进行比较，保留 NaN、无穷大和带符号零的原有行为。
+包含小数点的数值字面量求值为 `DOUBLE`，而非 `DECIMAL`：因此，将 `DECIMAL` 列与
+`123456789012345678.98` 比较时，仍可能匹配 `123456789012345678.99`。
+求值为整型的整数字面量（如 `-9007199254740993`）在另一操作数也是精确类型时使用精确比较。
+对 `DECIMAL` 表达式应用一元负号会保留 `DECIMAL` 精度，因此在另一操作数也是精确类型时，
+`-decimal_col` 和 `-CAST('...' AS DECIMAL(p, s))` 均保留精确的小数比较语义。
+
+如需精确的小数常量，请使用带引号的字符串并保留足够的小数位，例如
+`amount = CAST('123456789012345678.99' AS DECIMAL(38, 2))`。
+对于负数常量，`CAST('-123456789012345678.99' AS DECIMAL(38, 2))` 和
+`-CAST('123456789012345678.99' AS DECIMAL(38, 2))` 均保留小数精度。
+现有 CAST 舍入行为不变：将字符串或 `DECIMAL` 转换为 `DECIMAL(p, s)` 时，
+如果减少小数位，则采用 `CEILING`（向正无穷方向舍入）。因此，
+`CAST('1.231' AS DECIMAL(10, 2))` 为 `1.24`，
+`CAST('-1.239' AS DECIMAL(10, 2))` 为 `-1.23`。如果在 CAST 之后取负，
+`-CAST('1.239' AS DECIMAL(10, 2))` 则为 `-1.24`：减少小数位时先舍入再取负，
+因此这两种写法在该情况下不能互换。
+转换 `FLOAT` 或 `DOUBLE` 时采用 `HALF_UP`；对带小数点的数值字面量进行 CAST，
+无法恢复其求值为 `DOUBLE` 时已丢失的精度。
+
+每一对操作数独立选择精确或浮点比较，包括 `IN` 列表中的每个元素。
+当 `id = 9007199254740993` 时，`id IN (9007199254740992)` 为假，
+但 `id IN (9007199254740992, 9007199254740992.0)` 为真，因为第二次比较使用 `DOUBLE`。
+因此，在精确类型与浮点类型混用时，等号不一定满足传递性；需要精度保证时，请统一使用精确类型的操作数。
+以上规则适用于 `ZETA` 和 `INTERNAL` 两种 SQL 引擎设置。
+
## 示例
源端数据读取的表格如下：
diff --git
a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/sql_transform.conf
b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/sql_transform.conf
index 4bba0cacba..b2de7b4c8c 100644
---
a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/sql_transform.conf
+++
b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/sql_transform.conf
@@ -26,9 +26,12 @@ source {
FakeSource {
plugin_output = "fake"
row.num = 100
+ bigint.fake.mode = "template"
+ bigint.template = [9007199254740993]
schema = {
fields {
id = "int"
+ large_id = "bigint"
name = "string"
age = "int"
c_time = "time"
@@ -52,7 +55,7 @@ transform {
plugin_input = "fake"
plugin_output = "fake1"
# the query table name must same as field 'plugin_input'
- query = "select id, regexp_replace(name, '.+', 'b') as name, age+1 as age,
pi() as pi, c_time, c_timestamp, c_date, c_map, c_array, c_decimal, c_row from
dual"
+ query = "select id, regexp_replace(name, '.+', 'b') as name, age+1 as age,
pi() as pi, c_time, c_timestamp, c_date, c_map, c_array, c_decimal, c_row,
large_id, case large_id when 9007199254740992 then 'rounded' else 'exact' end
as comparison_result from dual where large_id > 9007199254740992 and large_id
not in (9007199254740992)"
}
# The SQL transform support base function and criteria operation
# But the complex SQL unsupported yet, include: multi source table/rows JOIN
and AGGREGATE operation and the like
@@ -70,6 +73,16 @@ sink {
}
]
field_rules = [
+ {
+ field_name = large_id
+ field_type = bigint
+ field_value = [{equals_to = 9007199254740993}]
+ },
+ {
+ field_name = comparison_result
+ field_type = string
+ field_value = [{equals_to = "exact"}]
+ },
{
field_name = id
field_type = int
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLFilter.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLFilter.java
index 9446e323d0..8bf3a145fc 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLFilter.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLFilter.java
@@ -22,6 +22,7 @@ import
org.apache.seatunnel.shade.org.apache.commons.lang3.tuple.Pair;
import org.apache.seatunnel.api.table.type.BasicType;
import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated;
import org.apache.seatunnel.transform.exception.TransformException;
+import org.apache.seatunnel.transform.sql.zeta.functions.NumericFunction;
import net.sf.jsqlparser.expression.Expression;
import net.sf.jsqlparser.expression.Function;
@@ -41,6 +42,7 @@ import
net.sf.jsqlparser.expression.operators.relational.NotEqualsTo;
import
net.sf.jsqlparser.expression.operators.relational.ParenthesedExpressionList;
import net.sf.jsqlparser.schema.Column;
+import java.math.BigDecimal;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
@@ -154,7 +156,13 @@ public class ZetaSQLFilter {
}
if (leftValue != null) {
if (leftValue instanceof Number && rightValue instanceof
Number) {
- if (((Number) leftValue).doubleValue() == ((Number)
rightValue).doubleValue()) {
+ boolean equal =
+ areExactNumbers(leftValue, rightValue)
+ ? compareExactNumbers((Number) leftValue,
(Number) rightValue)
+ == 0
+ : ((Number) leftValue).doubleValue()
+ == ((Number)
rightValue).doubleValue();
+ if (equal) {
return !inExpression.isNot();
}
} else if (leftValue instanceof OffsetDateTime
@@ -277,7 +285,9 @@ public class ZetaSQLFilter {
return false;
}
if (leftVal instanceof Number && rightVal instanceof Number) {
- return ((Number) leftVal).doubleValue() == ((Number)
rightVal).doubleValue();
+ return areExactNumbers(leftVal, rightVal)
+ ? compareExactNumbers((Number) leftVal, (Number) rightVal)
== 0
+ : ((Number) leftVal).doubleValue() == ((Number)
rightVal).doubleValue();
}
if (leftVal instanceof OffsetDateTime && rightVal instanceof
OffsetDateTime) {
return ((OffsetDateTime) leftVal).isEqual((OffsetDateTime)
rightVal);
@@ -292,7 +302,9 @@ public class ZetaSQLFilter {
return rightVal != null;
}
if (leftVal instanceof Number && rightVal instanceof Number) {
- return ((Number) leftVal).doubleValue() != ((Number)
rightVal).doubleValue();
+ return areExactNumbers(leftVal, rightVal)
+ ? compareExactNumbers((Number) leftVal, (Number) rightVal)
!= 0
+ : ((Number) leftVal).doubleValue() != ((Number)
rightVal).doubleValue();
}
if (leftVal instanceof OffsetDateTime && rightVal instanceof
OffsetDateTime) {
return !((OffsetDateTime) leftVal).isEqual((OffsetDateTime)
rightVal);
@@ -307,7 +319,9 @@ public class ZetaSQLFilter {
return false;
}
if (leftVal instanceof Number && rightVal instanceof Number) {
- return ((Number) leftVal).doubleValue() > ((Number)
rightVal).doubleValue();
+ return areExactNumbers(leftVal, rightVal)
+ ? compareExactNumbers((Number) leftVal, (Number) rightVal)
> 0
+ : ((Number) leftVal).doubleValue() > ((Number)
rightVal).doubleValue();
}
if (leftVal instanceof String && rightVal instanceof String) {
return ((String) leftVal).compareTo((String) rightVal) > 0;
@@ -338,7 +352,9 @@ public class ZetaSQLFilter {
return false;
}
if (leftVal instanceof Number && rightVal instanceof Number) {
- return ((Number) leftVal).doubleValue() >= ((Number)
rightVal).doubleValue();
+ return areExactNumbers(leftVal, rightVal)
+ ? compareExactNumbers((Number) leftVal, (Number) rightVal)
>= 0
+ : ((Number) leftVal).doubleValue() >= ((Number)
rightVal).doubleValue();
}
if (leftVal instanceof String && rightVal instanceof String) {
return ((String) leftVal).compareTo((String) rightVal) >= 0;
@@ -384,7 +400,9 @@ public class ZetaSQLFilter {
return ((LocalTime) leftVal).isBefore((LocalTime) rightVal);
}
if (leftVal instanceof Number && rightVal instanceof Number) {
- return ((Number) leftVal).doubleValue() < ((Number)
rightVal).doubleValue();
+ return areExactNumbers(leftVal, rightVal)
+ ? compareExactNumbers((Number) leftVal, (Number) rightVal)
< 0
+ : ((Number) leftVal).doubleValue() < ((Number)
rightVal).doubleValue();
}
if (leftVal instanceof String && rightVal instanceof String) {
return ((String) leftVal).compareTo((String) rightVal) < 0;
@@ -418,7 +436,9 @@ public class ZetaSQLFilter {
return ((LocalTime) leftVal).isBefore((LocalTime) rightVal) ||
leftVal.equals(rightVal);
}
if (leftVal instanceof Number && rightVal instanceof Number) {
- return ((Number) leftVal).doubleValue() <= ((Number)
rightVal).doubleValue();
+ return areExactNumbers(leftVal, rightVal)
+ ? compareExactNumbers((Number) leftVal, (Number) rightVal)
<= 0
+ : ((Number) leftVal).doubleValue() <= ((Number)
rightVal).doubleValue();
}
if (leftVal instanceof String && rightVal instanceof String) {
return ((String) leftVal).compareTo((String) rightVal) <= 0;
@@ -430,6 +450,34 @@ public class ZetaSQLFilter {
leftVal.getClass().getSimpleName(),
rightVal.getClass().getSimpleName()));
}
+ private static boolean areExactNumbers(Object left, Object right) {
+ return isExactNumber(left) && isExactNumber(right);
+ }
+
+ /**
+ * Recognizes integral wrappers and BigDecimal. Float/Double and other
Number subtypes
+ * (including BigInteger) retain the legacy double path. Callers use
native floating operators,
+ * not Double.compare, to preserve NaN, infinity and signed-zero behavior.
+ */
+ private static boolean isExactNumber(Object value) {
+ return value instanceof Byte
+ || value instanceof Short
+ || value instanceof Integer
+ || value instanceof Long
+ || value instanceof BigDecimal;
+ }
+
+ /** Compares exact operands; callers must first establish {@link
#areExactNumbers}. */
+ private static int compareExactNumbers(Number left, Number right) {
+ // Integral pairs need no allocation. Mixed decimal/integral pairs may
allocate a BigDecimal
+ // per comparison (including each IN item) to preserve precision
instead of using double.
+ if (left instanceof BigDecimal || right instanceof BigDecimal) {
+ return NumericFunction.toBigDecimal(left)
+ .compareTo(NumericFunction.toBigDecimal(right));
+ }
+ return Long.compare(left.longValue(), right.longValue());
+ }
+
private boolean andExpr(AndExpression andExpression, Object[] inputFields)
{
Expression leftExpr = andExpression.getLeftExpression();
boolean leftRes = executeFilter(leftExpr, inputFields);
diff --git
a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/SQLNumericComparisonTest.java
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/SQLNumericComparisonTest.java
new file mode 100644
index 0000000000..f9d3628480
--- /dev/null
+++
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/SQLNumericComparisonTest.java
@@ -0,0 +1,341 @@
+/*
+ * 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.transform.sql;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.DecimalType;
+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.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Stream;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Exercises SQLTransform deliberately: parsing and operand evaluation
determine numeric types, and
+ * both searched and simple CASE expressions route back through the comparison
filter.
+ */
+class SQLNumericComparisonTest {
+ private static final String[] OPERATORS = {
+ "=", "!=", "<>", ">", ">=", "<", "<=", "IN", "NOT IN"
+ };
+
+ static Stream<Arguments> exactComparisons() {
+ List<Arguments> cases = new ArrayList<>();
+ addPair(cases, 9007199254740993L, 9007199254740992L, 1);
+ addPair(cases, -9007199254740993L, -9007199254740992L, -1);
+ addPair(cases, Long.MAX_VALUE, Long.MAX_VALUE - 1, 1);
+ addPair(cases, Long.MIN_VALUE, Long.MIN_VALUE + 1, -1);
+ addPair(cases, Long.MIN_VALUE, Long.MAX_VALUE, -1);
+ addPair(
+ cases,
+ new BigDecimal("123456789012345678.99"),
+ new BigDecimal("123456789012345678.98"),
+ 1);
+ addPair(
+ cases,
+ new BigDecimal("1.00000000000000000002"),
+ new BigDecimal("1.00000000000000000001"),
+ 1);
+ addPair(
+ cases,
+ new BigDecimal("-123456789012345678.99"),
+ new BigDecimal("-123456789012345678.98"),
+ -1);
+ addPair(
+ cases,
+ new BigDecimal("-1.00000000000000000002"),
+ new BigDecimal("-1.00000000000000000001"),
+ -1);
+ addPair(cases, new BigDecimal("9007199254740993"), 9007199254740992L,
1);
+ addPair(cases, new BigDecimal("2.00"), 2L, 0);
+ addPair(cases, new BigDecimal("2.0"), new BigDecimal("2.00"), 0);
+ addPair(cases, (byte) 1, (short) 2, -1);
+ addPair(cases, Integer.MAX_VALUE, (long) Integer.MAX_VALUE, 0);
+ return cases.stream();
+ }
+
+ private static void addPair(List<Arguments> cases, Number a, Number b, int
comparison) {
+ for (String operator : OPERATORS) {
+ cases.add(Arguments.of(a, b, operator, accepts(operator,
comparison)));
+ cases.add(Arguments.of(b, a, operator, accepts(operator,
-comparison)));
+ }
+ }
+
+ @ParameterizedTest
+ @MethodSource("exactComparisons")
+ void comparesExactNumbers(Number a, Number b, String operator, boolean
expected) {
+ assertEquals(expected, matches(predicate(operator), a, b));
+ }
+
+ static Stream<Arguments> floatingComparisons() {
+ List<Arguments> cases = new ArrayList<>();
+ Number[][] pairs = {
+ {Double.NaN, Double.NaN},
+ {Double.NaN, 1L},
+ {1L, Double.NaN},
+ {Double.POSITIVE_INFINITY, Double.POSITIVE_INFINITY},
+ {Double.NEGATIVE_INFINITY, 1L},
+ {-0.0d, 0.0d},
+ {-0.0f, 0L},
+ {0.1f, 0.1d},
+ {9007199254740993L, 9007199254740992d},
+ {new BigDecimal("0.1"), 0.1d},
+ {Float.NaN, 0f}
+ };
+ for (Number[] pair : pairs) {
+ for (String op : OPERATORS) {
+ double a = pair[0].doubleValue();
+ double b = pair[1].doubleValue();
+ boolean expected;
+ switch (op) {
+ case "=":
+ case "IN":
+ expected = a == b;
+ break;
+ case "!=":
+ case "<>":
+ case "NOT IN":
+ expected = a != b;
+ break;
+ case ">":
+ expected = a > b;
+ break;
+ case ">=":
+ expected = a >= b;
+ break;
+ case "<":
+ expected = a < b;
+ break;
+ default:
+ expected = a <= b;
+ }
+ cases.add(Arguments.of(pair[0], pair[1], op, expected));
+ }
+ }
+ return cases.stream();
+ }
+
+ @ParameterizedTest
+ @MethodSource("floatingComparisons")
+ void preservesFloatingPointComparisons(Number a, Number b, String
operator, boolean expected) {
+ assertEquals(expected, matches(predicate(operator), a, b));
+ }
+
+ @Test
+ void comparesLargeIntegerLiteralAndMultipleInValues() {
+ assertEquals(false, matches("a = 9007199254740992", 9007199254740993L,
0L));
+ assertEquals(true, matches("a IN (0, 9007199254740993)",
9007199254740993L, 0L));
+ assertEquals(false, matches("a IN (0, 9007199254740992)",
9007199254740993L, 0L));
+ assertEquals(true, matches("a NOT IN (0, 9007199254740992)",
9007199254740993L, 0L));
+ }
+
+ @Test
+ void preservesFloatingComparisonForFractionalLiterals() {
+ BigDecimal value = new BigDecimal("123456789012345678.99");
+ assertEquals(true, matches("a = 123456789012345678.98", value, 0L));
+ assertEquals(false, matches("a <> 123456789012345678.98", value, 0L));
+ assertEquals(false, matches("a > 123456789012345678.98", value, 0L));
+ }
+
+ @Test
+ void comparesExplicitDecimalCastsExactly() {
+ BigDecimal value = new BigDecimal("123456789012345678.99");
+ SQLTransform transform =
+ new SQLTransform(
+ ReadonlyConfig.fromMap(
+ Collections.singletonMap(
+ "query",
+ "select CAST('123456789012345678.99'
AS DECIMAL(38, 2)) as exact_value, "
+ +
"CAST('123456789012345678.99' AS DECIMAL(38, 4)) as scaled_value from dual")),
+ CatalogTableUtil.getCatalogTable(
+ "test",
+ new SeaTunnelRowType(
+ new String[] {"a"},
+ new SeaTunnelDataType<?>[] {new
DecimalType(38, 2)})));
+ List<SeaTunnelRow> output = transform.transformRow(new
SeaTunnelRow(new Object[] {value}));
+ assertEquals(1, output.size());
+ assertEquals(value, output.get(0).getField(0));
+ assertEquals(new BigDecimal("123456789012345678.9900"),
output.get(0).getField(1));
+ assertEquals(
+ true, matches("a = CAST('123456789012345678.99' AS DECIMAL(38,
2))", value, 0L));
+ assertEquals(
+ false, matches("a = CAST('123456789012345678.98' AS
DECIMAL(38, 2))", value, 0L));
+ assertEquals(
+ true, matches("a > CAST('123456789012345678.98' AS DECIMAL(38,
2))", value, 0L));
+ assertEquals(
+ true,
+ matches("a = CAST('9007199254740993' AS DECIMAL(38, 0))",
9007199254740993L, 0L));
+ assertEquals(
+ false,
+ matches("a = CAST('9007199254740992' AS DECIMAL(38, 0))",
9007199254740993L, 0L));
+ }
+
+ @Test
+ void comparesNegativeIntegerLiteralsExactly() {
+ assertEquals(true, matches("a = -9007199254740993",
-9007199254740993L, 0L));
+ assertEquals(false, matches("a = -9007199254740992",
-9007199254740993L, 0L));
+ assertEquals(true, matches("a < -9007199254740992",
-9007199254740993L, 0L));
+ }
+
+ @Test
+ void preservesPerElementComparisonForMixedInLists() {
+ assertEquals(false, matches("a = 9007199254740992", 9007199254740993L,
0L));
+ assertEquals(true, matches("a = 9007199254740992.0",
9007199254740993L, 0L));
+ assertEquals(false, matches("a IN (9007199254740992)",
9007199254740993L, 0L));
+ assertEquals(
+ true,
+ matches("a IN (9007199254740992, 9007199254740992.0)",
9007199254740993L, 0L));
+ assertEquals(
+ true,
+ matches("a IN (9007199254740992.0, 9007199254740992)",
9007199254740993L, 0L));
+ assertEquals(
+ false,
+ matches("a NOT IN (9007199254740992, 9007199254740992.0)",
9007199254740993L, 0L));
+ }
+
+ @Test
+ void comparesNegatedDecimalOperandsExactly() {
+ BigDecimal value = new BigDecimal("-123456789012345678.99");
+ assertEquals(
+ false, matches("a = -CAST('123456789012345678.98' AS
DECIMAL(38, 2))", value, 0L));
+ assertEquals(
+ false, matches("-a = CAST('123456789012345678.98' AS
DECIMAL(38, 2))", value, 0L));
+ assertEquals(
+ true, matches("a = -CAST('123456789012345678.99' AS
DECIMAL(38, 2))", value, 0L));
+ assertEquals(
+ true, matches("-a = CAST('123456789012345678.99' AS
DECIMAL(38, 2))", value, 0L));
+ assertEquals(
+ false, matches("a = CAST('-123456789012345678.98' AS
DECIMAL(38, 2))", value, 0L));
+ assertEquals(
+ true, matches("a = CAST('-123456789012345678.99' AS
DECIMAL(38, 2))", value, 0L));
+ assertEquals(true, matches("a = -123456789012345678.98", value, 0L));
+ }
+
+ @Test
+ void preservesDecimalCastScaleRounding() {
+ assertEquals(
+ true, matches("a = CAST('1.231' AS DECIMAL(10, 2))", new
BigDecimal("1.24"), 0L));
+ assertEquals(
+ true, matches("a = CAST('-1.239' AS DECIMAL(10, 2))", new
BigDecimal("-1.23"), 0L));
+ assertEquals(
+ true, matches("a = CAST(1.231 AS DECIMAL(10, 2))", new
BigDecimal("1.23"), 0L));
+ assertEquals(
+ true, matches("a = CAST(-1.239 AS DECIMAL(10, 2))", new
BigDecimal("-1.24"), 0L));
+ }
+
+ @Test
+ void comparesExactNumbersInCaseExpressions() {
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"a", "b"},
+ new SeaTunnelDataType<?>[] {BasicType.LONG_TYPE,
BasicType.LONG_TYPE});
+ SQLTransform transform =
+ new SQLTransform(
+ ReadonlyConfig.fromMap(
+ Collections.singletonMap(
+ "query",
+ "select case when a > b then 'higher'
else 'lower' end as ordering, "
+ + "case a when b then 'same'
else 'different' end as equality from dual")),
+ CatalogTableUtil.getCatalogTable("test", rowType));
+ List<SeaTunnelRow> output =
+ transform.transformRow(
+ new SeaTunnelRow(new Object[] {9007199254740993L,
9007199254740992L}));
+ assertEquals("higher", output.get(0).getField(0));
+ assertEquals("different", output.get(0).getField(1));
+ }
+
+ @Test
+ void leavesNullAndStringHandlingUnchanged() {
+ assertEquals(false, matches("a = b", null, 1L));
+ assertEquals(false, matches("a > b", 1L, null));
+ assertEquals(true, matches("a IS NULL", null, 1L));
+ assertEquals(true, matches("a = b", "same", "same"));
+ assertEquals(true, matches("a < b", "a", "b"));
+ }
+
+ private static String predicate(String operator) {
+ return "a " + operator + (operator.endsWith("IN") ? " (b)" : " b");
+ }
+
+ private static boolean accepts(String op, int comparison) {
+ switch (op) {
+ case "=":
+ case "IN":
+ return comparison == 0;
+ case "!=":
+ case "<>":
+ case "NOT IN":
+ return comparison != 0;
+ case ">":
+ return comparison > 0;
+ case ">=":
+ return comparison >= 0;
+ case "<":
+ return comparison < 0;
+ default:
+ return comparison <= 0;
+ }
+ }
+
+ private static boolean matches(String predicate, Object a, Object b) {
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"a", "b"}, new SeaTunnelDataType<?>[]
{type(a), type(b)});
+ SQLTransform transform =
+ new SQLTransform(
+ ReadonlyConfig.fromMap(
+ Collections.singletonMap(
+ "query", "select a, b from dual where
" + predicate)),
+ CatalogTableUtil.getCatalogTable("test", rowType));
+ List<SeaTunnelRow> output = transform.transformRow(new
SeaTunnelRow(new Object[] {a, b}));
+ if (output == null || output.isEmpty()) {
+ return false;
+ }
+ assertEquals(1, output.size());
+ assertEquals(a, output.get(0).getField(0));
+ assertEquals(b, output.get(0).getField(1));
+ return true;
+ }
+
+ private static SeaTunnelDataType<?> type(Object value) {
+ if (value instanceof BigDecimal) {
+ BigDecimal decimal = (BigDecimal) value;
+ return new DecimalType(38, Math.max(0, decimal.scale()));
+ }
+ if (value instanceof Byte) return BasicType.BYTE_TYPE;
+ if (value instanceof Short) return BasicType.SHORT_TYPE;
+ if (value instanceof Integer) return BasicType.INT_TYPE;
+ if (value instanceof Float) return BasicType.FLOAT_TYPE;
+ if (value instanceof Double) return BasicType.DOUBLE_TYPE;
+ if (value instanceof String) return BasicType.STRING_TYPE;
+ return BasicType.LONG_TYPE;
+ }
+}
