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 5d2e89108d [Fix][Transform-V2] Fix silent value corruption in Zeta SQL
numeric functions (#11697)
5d2e89108d is described below
commit 5d2e89108dc18ccd18da2f5fd26eb274cd2d4a55
Author: Sepuri Sai Krishna <[email protected]>
AuthorDate: Fri Aug 14 09:43:47 2026 +0530
[Fix][Transform-V2] Fix silent value corruption in Zeta SQL numeric
functions (#11697)
---
.../introduction/concepts/incompatible-changes.md | 19 +++++
.../introduction/concepts/incompatible-changes.md | 18 +++++
.../test/resources/sql_transform/func_numeric.conf | 8 +-
.../seatunnel/transform/sql/zeta/ZetaSQLType.java | 13 +--
.../sql/zeta/functions/NumericFunction.java | 52 +++++++++---
.../transform/sql/SQLNumericFunctionsTest.java | 5 +-
.../transform/sql/zeta/NumericFunctionTest.java | 92 ++++++++++++++++++++++
.../sql/zeta/functions/NumericFunctionTest.java | 12 ++-
8 files changed, 195 insertions(+), 24 deletions(-)
diff --git a/docs/en/introduction/concepts/incompatible-changes.md
b/docs/en/introduction/concepts/incompatible-changes.md
index d1d3159d42..7ad6a2af1b 100644
--- a/docs/en/introduction/concepts/incompatible-changes.md
+++ b/docs/en/introduction/concepts/incompatible-changes.md
@@ -145,6 +145,25 @@ You need to check this document before you upgrade to
related version.
- Adjusted SQL Transform date & time functions:
- `DATEDIFF(<start>, <end>, 'MONTH')` now returns the total number of months
between the two dates across years (for example, from `2023-01-01` to
`2024-03-01` returns `14` instead of `15`).
- `WEEK(<datetime>)` now returns the ISO week number directly (previous
behavior added an extra `+1` to the ISO week value).
+- **[BREAKING]** SQL Transform `CEIL` / `CEILING`, `FLOOR` and `TRUNC` /
`TRUNCATE` now return the data type of their
+ argument, as their documentation has always specified. Previously `CEIL` and
`FLOOR` declared `INT` and `TRUNC`
+ declared `DOUBLE` regardless of the input type, which silently produced
wrong values:
+
+ | Expression | Input | Previous result | Current result |
+ |------------|-------|-----------------|----------------|
+ | `CEIL(bigint_col)` | `9007199254740993` | `1` | `9007199254740993` |
+ | `FLOOR(double_col)` | `1.0E18` | `2147483647` | `1.0E18` |
+ | `TRUNC(bigint_col)` | `9007199254740993` | declared `DOUBLE`, returned a
`Long` | `9007199254740993` |
+
+ **Migration Guide**: If a downstream sink column was created against the old
`INT` / `DOUBLE` output type, widen it to
+ match the source column type (for example `BIGINT` for `CEIL(bigint_col)`),
or wrap the expression in an explicit
+ `CAST(... AS INT)` to keep the previous schema. Expressions over `INT`
columns are unaffected.
+- **[BREAKING]** SQL Transform `ROUND`, `TRUNC` / `TRUNCATE` and `MOD` no
longer round-trip their arguments through
+ `double`, so `DECIMAL` and large `BIGINT` values keep full precision. For
example
+ `ROUND(CAST('12345678901234567890.987654321' AS DECIMAL(38,9)), 2)`
previously returned
+ `12345678901234567000.00` and now returns `12345678901234567890.99`, and
`MOD(9007199254740993, 2)` previously
+ returned `0` and now returns `1`. Jobs that (intentionally or not) depended
on the old lossy values will see
+ different — now correct — output.
### Engine Behavior Changes
diff --git a/docs/zh/introduction/concepts/incompatible-changes.md
b/docs/zh/introduction/concepts/incompatible-changes.md
index a9b52589db..a44dc4e8e0 100644
--- a/docs/zh/introduction/concepts/incompatible-changes.md
+++ b/docs/zh/introduction/concepts/incompatible-changes.md
@@ -138,6 +138,24 @@
| `DynamicCompile` | `compile_pattern = ABSOLUTE_PATH` 但 `absolute_path` 为空
| 运行时文件读取失败 | 使用 `ABSOLUTE_PATH` 模式时提供 `absolute_path` |
**迁移指南**：升级前请对照上表检查您的转换配置。如果现有配置匹配了"新增拒绝的配置"中的情况，请在升级前修改。提交时的错误消息会清楚标明哪个选项无效及原因。
+- **[BREAKING]** SQL 转换的 `CEIL` / `CEILING`、`FLOOR` 与 `TRUNC` / `TRUNCATE`
现在返回与参数相同的数据类型，
+ 这与文档中一直声明的行为一致。此前无论输入类型如何，`CEIL` 和 `FLOOR` 都声明为 `INT`、`TRUNC` 声明为 `DOUBLE`，
+ 从而静默产生错误的值：
+
+ | 表达式 | 输入 | 以前的结果 | 现在的结果 |
+ |-------|------|-----------|-----------|
+ | `CEIL(bigint_col)` | `9007199254740993` | `1` | `9007199254740993` |
+ | `FLOOR(double_col)` | `1.0E18` | `2147483647` | `1.0E18` |
+ | `TRUNC(bigint_col)` | `9007199254740993` | 声明为 `DOUBLE`，实际返回 `Long` |
`9007199254740993` |
+
+ **迁移指南**：如果下游 Sink 的列是按旧的 `INT` / `DOUBLE` 输出类型创建的，请将其放宽为与源列一致的类型
+ （例如 `CEIL(bigint_col)` 对应 `BIGINT`），或使用显式的 `CAST(... AS INT)` 保持原有 schema。
+ 对 `INT` 列的表达式不受影响。
+- **[BREAKING]** SQL 转换的 `ROUND`、`TRUNC` / `TRUNCATE` 和 `MOD` 不再将参数经由 `double`
中转，
+ 因此 `DECIMAL` 和大 `BIGINT` 值可以保留完整精度。例如
+ `ROUND(CAST('12345678901234567890.987654321' AS DECIMAL(38,9)), 2)` 以前返回
+ `12345678901234567000.00`，现在返回
`12345678901234567890.99`；`MOD(9007199254740993, 2)` 以前返回 `0`，
+ 现在返回 `1`。依赖旧的精度丢失结果的作业，其输出会发生变化（现在是正确的）。
### 引擎行为变更
diff --git
a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/sql_transform/func_numeric.conf
b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/sql_transform/func_numeric.conf
index 046a1cf995..56b4e37e65 100644
---
a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/sql_transform/func_numeric.conf
+++
b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/sql_transform/func_numeric.conf
@@ -157,9 +157,9 @@ sink {
},
{
field_name = "c5_1"
- field_type = "int"
+ field_type = "double"
field_value = [
- {equals_to = 14}
+ {equals_to = 14.0}
]
},
{
@@ -171,9 +171,9 @@ sink {
},
{
field_name = "c5_2"
- field_type = "int"
+ field_type = "double"
field_value = [
- {equals_to = 13}
+ {equals_to = 13.0}
]
},
{
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLType.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLType.java
index 7da359c8f5..5fbf4c44d6 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLType.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLType.java
@@ -362,9 +362,6 @@ public class ZetaSQLType {
case ZetaSQLFunction.LOCATE:
case ZetaSQLFunction.INSTR:
case ZetaSQLFunction.POSITION:
- case ZetaSQLFunction.CEIL:
- case ZetaSQLFunction.CEILING:
- case ZetaSQLFunction.FLOOR:
case ZetaSQLFunction.DAY_OF_MONTH:
case ZetaSQLFunction.DAY_OF_WEEK:
case ZetaSQLFunction.DAY_OF_YEAR:
@@ -410,8 +407,6 @@ public class ZetaSQLType {
case ZetaSQLFunction.POWER:
case ZetaSQLFunction.RAND:
case ZetaSQLFunction.RANDOM:
- case ZetaSQLFunction.TRUNC:
- case ZetaSQLFunction.TRUNCATE:
case ZetaSQLFunction.COSINE_DISTANCE:
case ZetaSQLFunction.L1_DISTANCE:
case ZetaSQLFunction.L2_DISTANCE:
@@ -462,10 +457,18 @@ public class ZetaSQLType {
"unsupported datetime format type");
}
}
+ // These functions all return the type of their first
argument. CEIL/FLOOR/TRUNC are
+ // documented that way as well; declaring INT/DOUBLE for them
would truncate BIGINT
+ // and DECIMAL results.
case ZetaSQLFunction.ABS:
case ZetaSQLFunction.DATEADD:
case ZetaSQLFunction.TIMESTAMPADD:
case ZetaSQLFunction.ROUND:
+ case ZetaSQLFunction.CEIL:
+ case ZetaSQLFunction.CEILING:
+ case ZetaSQLFunction.FLOOR:
+ case ZetaSQLFunction.TRUNC:
+ case ZetaSQLFunction.TRUNCATE:
case ZetaSQLFunction.NULLIF:
return
getExpressionType(function.getParameters().getExpressions().get(0));
case ZetaSQLFunction.IFNULL:
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/functions/NumericFunction.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/functions/NumericFunction.java
index 1be6a44ef3..fb9b46ba66 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/functions/NumericFunction.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/functions/NumericFunction.java
@@ -152,6 +152,26 @@ public class NumericFunction {
return Math.atan2(arg.doubleValue(), arg2.doubleValue());
}
+ /**
+ * Converts a numeric value to {@link BigDecimal} without losing precision.
+ *
+ * <p>{@link BigDecimal} values are used as-is and integral types are
widened through {@code
+ * longValue()}. Only the floating point types are read through {@code
doubleValue()}, where the
+ * value carries no more precision than a {@code double} to begin with.
+ */
+ private static BigDecimal toBigDecimal(Number value) {
+ if (value instanceof BigDecimal) {
+ return (BigDecimal) value;
+ }
+ if (value instanceof Byte
+ || value instanceof Short
+ || value instanceof Integer
+ || value instanceof Long) {
+ return BigDecimal.valueOf(value.longValue());
+ }
+ return BigDecimal.valueOf(value.doubleValue());
+ }
+
public static Number mod(List<Object> args) {
Number leftValue = (Number) args.get(0);
if (leftValue == null) {
@@ -161,12 +181,12 @@ public class NumericFunction {
if (rightValue == null) {
return null;
}
- if (rightValue.doubleValue() == 0) {
+ BigDecimal leftBD = toBigDecimal(leftValue);
+ BigDecimal rightBD = toBigDecimal(rightValue);
+ if (rightBD.signum() == 0) {
throw new TransformException(
CommonErrorCodeDeprecated.UNSUPPORTED_OPERATION, "Mod by
zero");
}
- BigDecimal leftBD = BigDecimal.valueOf(leftValue.doubleValue());
- BigDecimal rightBD = BigDecimal.valueOf(rightValue.doubleValue());
BigDecimal[] res = leftBD.divideAndRemainder(rightBD);
if (rightValue instanceof Integer) {
return res[1].intValue();
@@ -190,7 +210,14 @@ public class NumericFunction {
rightValue.getClass().getName(), ZetaSQLFunction.MOD));
}
- public static Integer ceil(List<Object> args) {
+ /**
+ * Returns the smallest value greater than or equal to the argument.
+ *
+ * <p>As documented for {@code CEIL}, the result keeps the data type of
the argument. Narrowing
+ * the result to {@code int} would silently corrupt {@code BIGINT}, {@code
DOUBLE} and {@code
+ * DECIMAL} inputs whose value does not fit into 32 bits.
+ */
+ public static Number ceil(List<Object> args) {
Number v1 = (Number) args.get(0);
if (v1 == null) {
return null;
@@ -199,7 +226,7 @@ public class NumericFunction {
if (args.size() >= 2) {
v2 = (Number) args.get(1);
}
- return round(v1, v2, RoundingMode.CEILING).intValue();
+ return round(v1, v2, RoundingMode.CEILING);
}
private static Number round(Number v1, Number v2, RoundingMode
roundingMode) {
@@ -225,8 +252,9 @@ public class NumericFunction {
}
case "BIGDECIMAL":
{
- BigDecimal bd = BigDecimal.valueOf(v1.doubleValue());
- v1 = bd.setScale(scale, roundingMode);
+ // Must not round-trip through double: a DECIMAL carries
more digits than a
+ // double can hold, so doubleValue() would silently drop
the low-order ones.
+ v1 = ((BigDecimal) v1).setScale(scale, roundingMode);
break;
}
case "DOUBLE":
@@ -282,7 +310,13 @@ public class NumericFunction {
return Math.exp(v1.doubleValue());
}
- public static Integer floor(List<Object> args) {
+ /**
+ * Returns the largest value less than or equal to the argument.
+ *
+ * <p>As documented for {@code FLOOR}, the result keeps the data type of
the argument. See
+ * {@link #ceil(List)} for why the result must not be narrowed to {@code
int}.
+ */
+ public static Number floor(List<Object> args) {
Number v1 = (Number) args.get(0);
if (v1 == null) {
return null;
@@ -291,7 +325,7 @@ public class NumericFunction {
if (args.size() >= 2) {
v2 = (Number) args.get(1);
}
- return round(v1, v2, RoundingMode.FLOOR).intValue();
+ return round(v1, v2, RoundingMode.FLOOR);
}
public static Double ln(List<Object> args) {
diff --git
a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/SQLNumericFunctionsTest.java
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/SQLNumericFunctionsTest.java
index 8c71f4a1be..223ba4ecb0 100644
---
a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/SQLNumericFunctionsTest.java
+++
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/SQLNumericFunctionsTest.java
@@ -65,8 +65,9 @@ public class SQLNumericFunctionsTest {
Assertions.assertEquals(3, outRow.getField(0));
Assertions.assertEquals(-1, outRow.getField(1));
- Assertions.assertEquals(2, outRow.getField(2));
- Assertions.assertEquals(1, outRow.getField(3));
+ // CEIL/FLOOR keep the type of their argument, so a DOUBLE column
yields a DOUBLE
+ Assertions.assertEquals(2d, outRow.getField(2));
+ Assertions.assertEquals(1d, outRow.getField(3));
}
@Test
diff --git
a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/zeta/NumericFunctionTest.java
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/zeta/NumericFunctionTest.java
index 07a194feea..66e26f4d69 100644
---
a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/zeta/NumericFunctionTest.java
+++
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/zeta/NumericFunctionTest.java
@@ -93,4 +93,96 @@ public class NumericFunctionTest {
public void testSignNullReturnsNull() {
Assertions.assertNull(NumericFunction.sign(Collections.singletonList(null)));
}
+
+ /** BIGINT values outside the {@code int} range must survive CEIL/FLOOR
unchanged. */
+ @Test
+ public void testCeilAndFloorKeepBigIntValue() {
+ long value = 9007199254740993L;
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"long_v"}, new SeaTunnelDataType[]
{BasicType.LONG_TYPE});
+ SeaTunnelRow inputRow = new SeaTunnelRow(new Object[] {value});
+
+ SQLEngine sqlEngine =
SQLEngineFactory.getSQLEngine(SQLEngineFactory.EngineType.ZETA);
+ sqlEngine.init(
+ "test",
+ null,
+ rowType,
+ "select CEIL(long_v) as ceil_v, FLOOR(long_v) as floor_v,
TRUNC(long_v) as trunc_v from test");
+
+ SeaTunnelRowType outRowType = sqlEngine.typeMapping(null);
+ Assertions.assertEquals(BasicType.LONG_TYPE,
outRowType.getFieldType(0));
+ Assertions.assertEquals(BasicType.LONG_TYPE,
outRowType.getFieldType(1));
+ Assertions.assertEquals(BasicType.LONG_TYPE,
outRowType.getFieldType(2));
+
+ SeaTunnelRow outRow = sqlEngine.transformBySQL(inputRow,
rowType).get(0);
+ Assertions.assertEquals(value, outRow.getField(0));
+ Assertions.assertEquals(value, outRow.getField(1));
+ Assertions.assertEquals(value, outRow.getField(2));
+ }
+
+ /** DOUBLE inputs keep their type through CEIL/FLOOR instead of being
narrowed to int. */
+ @Test
+ public void testCeilAndFloorKeepDoubleType() {
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"double_v"}, new SeaTunnelDataType[]
{BasicType.DOUBLE_TYPE});
+ SeaTunnelRow inputRow = new SeaTunnelRow(new Object[] {1.0e18d});
+
+ SQLEngine sqlEngine =
SQLEngineFactory.getSQLEngine(SQLEngineFactory.EngineType.ZETA);
+ sqlEngine.init(
+ "test",
+ null,
+ rowType,
+ "select CEIL(double_v) as ceil_v, FLOOR(double_v) as floor_v
from test");
+
+ SeaTunnelRowType outRowType = sqlEngine.typeMapping(null);
+ Assertions.assertEquals(BasicType.DOUBLE_TYPE,
outRowType.getFieldType(0));
+ Assertions.assertEquals(BasicType.DOUBLE_TYPE,
outRowType.getFieldType(1));
+
+ SeaTunnelRow outRow = sqlEngine.transformBySQL(inputRow,
rowType).get(0);
+ Assertions.assertEquals(1.0e18d, outRow.getField(0));
+ Assertions.assertEquals(1.0e18d, outRow.getField(1));
+ }
+
+ /** DECIMAL values carry more digits than a double, so ROUND/TRUNC must
stay exact. */
+ @Test
+ public void testRoundAndTruncKeepDecimalPrecision() {
+ DecimalType decimalType = new DecimalType(38, 9);
+ BigDecimal value = new BigDecimal("12345678901234567890.987654321");
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"decimal_v"}, new SeaTunnelDataType[]
{decimalType});
+ SeaTunnelRow inputRow = new SeaTunnelRow(new Object[] {value});
+
+ SQLEngine sqlEngine =
SQLEngineFactory.getSQLEngine(SQLEngineFactory.EngineType.ZETA);
+ sqlEngine.init(
+ "test",
+ null,
+ rowType,
+ "select ROUND(decimal_v, 2) as round_v, TRUNC(decimal_v, 2) as
trunc_v, CEIL(decimal_v) as ceil_v from test");
+
+ SeaTunnelRowType outRowType = sqlEngine.typeMapping(null);
+ Assertions.assertEquals(decimalType, outRowType.getFieldType(0));
+ Assertions.assertEquals(decimalType, outRowType.getFieldType(1));
+ Assertions.assertEquals(decimalType, outRowType.getFieldType(2));
+
+ SeaTunnelRow outRow = sqlEngine.transformBySQL(inputRow,
rowType).get(0);
+ Assertions.assertEquals(new BigDecimal("12345678901234567890.99"),
outRow.getField(0));
+ Assertions.assertEquals(new BigDecimal("12345678901234567890.98"),
outRow.getField(1));
+ Assertions.assertEquals(new BigDecimal("12345678901234567891"),
outRow.getField(2));
+ }
+
+ /** MOD must not lose the low-order bits of a BIGINT dividend. */
+ @Test
+ public void testModKeepsBigIntPrecision() {
+ Assertions.assertEquals(1,
NumericFunction.mod(Arrays.asList(9007199254740993L, 2)));
+ Assertions.assertEquals(1L,
NumericFunction.mod(Arrays.asList(9007199254740993L, 2L)));
+ Assertions.assertEquals(
+ new BigDecimal("0.987654321"),
+ NumericFunction.mod(
+ Arrays.asList(
+ new
BigDecimal("12345678901234567890.987654321"),
+ new BigDecimal("1"))));
+ }
}
diff --git
a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/zeta/functions/NumericFunctionTest.java
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/zeta/functions/NumericFunctionTest.java
index a14f0a0421..cfed353864 100644
---
a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/zeta/functions/NumericFunctionTest.java
+++
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/zeta/functions/NumericFunctionTest.java
@@ -158,11 +158,15 @@ public class NumericFunctionTest {
@Test
public void testCeilFloorRoundAndTrunc() {
- Assertions.assertEquals(2, NumericFunction.ceil(Arrays.asList(1.2d)));
- Assertions.assertEquals(-1,
NumericFunction.ceil(Arrays.asList(-1.8d)));
+ // CEIL/FLOOR return the type of their argument, so a DOUBLE argument
yields a DOUBLE
+ Assertions.assertEquals(2d, NumericFunction.ceil(Arrays.asList(1.2d)));
+ Assertions.assertEquals(-1d,
NumericFunction.ceil(Arrays.asList(-1.8d)));
- Assertions.assertEquals(1, NumericFunction.floor(Arrays.asList(1.8d)));
- Assertions.assertEquals(-2,
NumericFunction.floor(Arrays.asList(-1.2d)));
+ Assertions.assertEquals(1d,
NumericFunction.floor(Arrays.asList(1.8d)));
+ Assertions.assertEquals(-2d,
NumericFunction.floor(Arrays.asList(-1.2d)));
+
+ Assertions.assertEquals(2, NumericFunction.ceil(Arrays.asList(2)));
+ Assertions.assertEquals(2L, NumericFunction.floor(Arrays.asList(2L)));
Assertions.assertEquals(3L,
NumericFunction.round(Arrays.asList(2.6d)).longValue());
Assertions.assertEquals(2L,
NumericFunction.round(Arrays.asList(2.4d)).longValue());
