[Fix][Transform-V2] Fix datetime/numeric/string function edge cases and update tests (apache#10168)

Co-authored-by: zengyi <zengyi@chinatelecom.cn>

Author: yzeng1618

docs/en/concept/incompatible-changes.md

@@ -13,6 +13,10 @@ You need to check this document before you upgrade to related version.
 
 ### Transform Changes
 
+- 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).
+
 ### Engine Behavior Changes
 
-### Dependency Upgrades
+### Dependency Upgrades

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

@@ -380,7 +380,7 @@ sink {
           field_name = "c4_10"
           field_type = "int"
           field_value = [
-            {equals_to = 15}
+            {equals_to = 14}
           ]
         },
         {

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

@@ -186,7 +186,7 @@ public static Long datediff(List<Object> args) {
                 break;
             case "MONTH":
                 if (date1 != null && date2 != null) {
-                    return (long) Period.between(date1, date2).getMonths();
+                    return Period.between(date1, date2).toTotalMonths();
                 }
                 break;
             case "WEEK":
@@ -500,6 +500,7 @@ public static Integer extract(List<Object> args) {
                 }
                 break;
             case "DOW":
+            case "DAYOFWEEK":
                 if (datetime instanceof LocalDate) {
                     return ((LocalDate) datetime).getDayOfWeek().getValue() % 7;
                 }
@@ -550,8 +551,6 @@ public static Integer extract(List<Object> args) {
                     return (year > 0) ? (year - 1) / 1000 + 1 : year / 1000;
                 }
                 break;
-            case "DAYOFWEEK":
-                return dayOfWeek(args);
             default:
                 throw new TransformException(
                         CommonErrorCodeDeprecated.UNSUPPORTED_OPERATION,
@@ -688,7 +687,7 @@ public static Integer week(List<Object> args) {
         }
         LocalDate localDate = convertToLocalDate(datetime);
         WeekFields weekFields = WeekFields.ISO;
-        return localDate.get(weekFields.weekOfYear()) + 1;
+        return localDate.get(weekFields.weekOfYear());
     }
 
     public static Integer year(List<Object> args) {

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

@@ -208,7 +208,7 @@ private static Number round(Number v1, Number v2, RoundingMode roundingMode) {
         c:
         switch (t.toUpperCase()) {
             case "INTEGER":
-            case "SHOT":
+            case "SHORT":
             case "LONG":
                 {
                     if (scale < 0) {
@@ -262,10 +262,10 @@ private static Number round(Number v1, Number v2, RoundingMode roundingMode) {
     }
 
     private static Number convertTo(String valueType, Number column) {
-        switch (valueType) {
+        switch (valueType.toUpperCase()) {
             case "INTEGER":
                 return column.intValue();
-            case "SHOT":
+            case "SHORT":
                 return column.shortValue();
             case "LONG":
                 return column.longValue();
@@ -403,10 +403,10 @@ public static Number round(List<Object> args) {
         return round(v1, v2, RoundingMode.HALF_UP);
     }
 
-    public static int sign(List<Object> args) {
+    public static Integer sign(List<Object> args) {
         Number v1 = (Number) args.get(0);
         if (v1 == null) {
-            return 0;
+            return null;
         }
         if (v1 instanceof Integer) {
             return Integer.signum((Integer) v1);

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

@@ -50,7 +50,7 @@ public class StringFunction {
 
     public static Integer ascii(List<Object> args) {
         String arg = (String) args.get(0);
-        if (arg == null) {
+        if (arg == null || arg.isEmpty()) {
             return null;
         } else {
             return (int) arg.charAt(0);
@@ -231,6 +231,9 @@ public static String left(List<Object> args) {
             return null;
         }
         int count = ((Number) args.get(1)).intValue();
+        if (count < 0) {
+            return "";
+        }
         if (count > arg.length()) {
             count = arg.length();
         }
@@ -243,6 +246,9 @@ public static String right(List<Object> args) {
             return null;
         }
         int count = ((Number) args.get(1)).intValue();
+        if (count < 0) {
+            return "";
+        }
         int length = arg.length();
         if (count > length) {
             count = length;

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

@@ -107,9 +107,6 @@ public static Object castAs(List<Object> args) {
         if (v1 == null) {
             return null;
         }
-        if (v1.equals(v2)) {
-            return null;
-        }
         switch (v2) {
             case "VARCHAR":
             case "STRING":

seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/zeta/DateTimeFunctionTest.java

@@ -24,13 +24,16 @@
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
 import org.apache.seatunnel.transform.sql.SQLEngine;
 import org.apache.seatunnel.transform.sql.SQLEngineFactory;
+import org.apache.seatunnel.transform.sql.zeta.functions.DateTimeFunction;
 
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
+import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.time.ZoneId;
 import java.util.ArrayList;
+import java.util.Arrays;
 
 public class DateTimeFunctionTest {
 
@@ -176,4 +179,14 @@ public void testFromUnixtimeFunctionWithIntegerInput() {
         // Both Integer and Long inputs should produce the same result
         Assertions.assertEquals(fieldInt.toString(), fieldLong.toString());
     }
+
+    @Test
+    public void testDateDiffMonthAcrossYearUsesTotalMonths() {
+        LocalDate start = LocalDate.of(2023, 1, 1);
+        LocalDate end = LocalDate.of(2024, 3, 1);
+
+        Long months = DateTimeFunction.datediff(Arrays.asList(start, end, "MONTH"));
+
+        Assertions.assertEquals(14L, months);
+    }
 }

seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/zeta/NumericFunctionTest.java

@@ -30,6 +30,7 @@
 import org.junit.jupiter.api.Test;
 
 import java.math.BigDecimal;
+import java.util.Arrays;
 import java.util.Collections;
 
 public class NumericFunctionTest {
@@ -78,4 +79,18 @@ public void testTrimScale() {
         Assertions.assertEquals("0", NumericFunction.trimScale(Collections.singletonList(0)));
         Assertions.assertNull(NumericFunction.trimScale(Collections.singletonList((Object) null)));
     }
+
+    @Test
+    public void testRoundShortNegativeScale() {
+        short shortValue = 123;
+
+        Number result = NumericFunction.round(Arrays.asList(shortValue, -1));
+
+        Assertions.assertEquals(120, result.intValue());
+    }
+
+    @Test
+    public void testSignNullReturnsNull() {
+        Assertions.assertNull(NumericFunction.sign(Collections.singletonList(null)));
+    }
 }

seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/zeta/functions/StringFunctionTest.java

@@ -121,4 +121,28 @@ public void testSubstringWithTemporal() {
         // Should extract time part from formatted string "15:30:45"
         Assertions.assertEquals("15:30", StringFunction.substring(args));
     }
+
+    @Test
+    public void testAsciiNullAndEmptyReturnNull() {
+        List<Object> args = new ArrayList<>();
+        args.add(null);
+        Assertions.assertNull(StringFunction.ascii(args));
+
+        args.clear();
+        args.add("");
+        Assertions.assertNull(StringFunction.ascii(args));
+    }
+
+    @Test
+    public void testLeftAndRightNegativeCountReturnEmpty() {
+        List<Object> args = new ArrayList<>();
+        args.add("abc");
+        args.add(-1);
+        Assertions.assertEquals("", StringFunction.left(args));
+
+        args.clear();
+        args.add("abc");
+        args.add(-2);
+        Assertions.assertEquals("", StringFunction.right(args));
+    }
 }

seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/zeta/functions/SystemFunctionTest.java

@@ -0,0 +1,43 @@
+/*
+ * 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.zeta.functions;
+
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+
+public class SystemFunctionTest {
+
+    @Test
+    public void testCastAsDoesNotReturnNullWhenValueEqualsTypeName() {
+        Object result = SystemFunction.castAs(Arrays.asList("VARCHAR", "VARCHAR"));
+        Assertions.assertEquals("VARCHAR", result);
+    }
+
+    @Test
+    public void testCoalesceRespectsTargetType() {
+        SeaTunnelDataType<?> targetType = BasicType.INT_TYPE;
+        Object result = SystemFunction.coalesce(Arrays.asList(null, "123"), targetType);
+
+        Assertions.assertEquals(123, result);
+    }
+}