Optimize code and add test

Author: aiwenmo

docs/content.zh/docs/core-concept/transform.md

@@ -162,7 +162,7 @@ Flink CDC uses [Calcite](https://calcite.apache.org/) to parse expressions and [
 | TO_TIMESTAMP(string1[, string2]) | toTimestamp(string1[, string2]) | Converts date time string string1 with format string2 (by default: 'yyyy-MM-dd HH:mm:ss') to a timestamp, without time zone. |
 | FROM_UNIXTIME(numeric[, string]) | fromUnixtime(NUMERIC[, STRING]) | Returns a representation of the numeric argument as a value in string format (default is ‘yyyy-MM-dd HH:mm:ss’). numeric is an internal timestamp value representing seconds since ‘1970-01-01 00:00:00’ UTC, such as produced by the UNIX_TIMESTAMP() function. The return value is expressed in the session time zone (specified in TableConfig). E.g., FROM_UNIXTIME(44) returns ‘1970-01-01 00:00:44’ if in UTC time zone, but returns ‘1970-01-01 09:00:44’ if in ‘Asia/Tokyo’ time zone. |
 | UNIX_TIMESTAMP() | unixTimestamp() | Gets current Unix timestamp in seconds. This function is not deterministic which means the value would be recalculated for each record. |
-| UNIX_TIMESTAMP(string1[, string2]) | unixTimestamp(STRING1[, STRING2]) | Converts a date time string string1 with format string2 (by default: yyyy-MM-dd HH:mm:ss if not specified) to Unix timestamp (in seconds), using the specified timezone in table config.If a time zone is specified in the date time string and parsed by UTC+X format such as “yyyy-MM-dd HH:mm:ss.SSS X”, this function will use the specified timezone in the date time string instead of the timezone in table config. If the date time string can not be parsed, the default value Long.MIN_VALUE(-9223372036854775808) will be returned.|
+| UNIX_TIMESTAMP(string1[, string2]) | unixTimestamp(STRING1[, STRING2]) | Converts a date time string string1 with format string2 (by default: yyyy-MM-dd HH:mm:ss if not specified) to Unix timestamp (in seconds), using the specified timezone in table config.<br/>If a time zone is specified in the date time string and parsed by UTC+X format such as “yyyy-MM-dd HH:mm:ss.SSS X”, this function will use the specified timezone in the date time string instead of the timezone in table config. If the date time string can not be parsed, the default value Long.MIN_VALUE(-9223372036854775808) will be returned.|
 
 ## Conditional Functions
 

docs/content/docs/core-concept/transform.md

@@ -162,7 +162,7 @@ Flink CDC uses [Calcite](https://calcite.apache.org/) to parse expressions and [
 | TO_TIMESTAMP(string1[, string2]) | toTimestamp(string1[, string2]) | Converts date time string string1 with format string2 (by default: 'yyyy-MM-dd HH:mm:ss') to a timestamp, without time zone. |
 | FROM_UNIXTIME(numeric[, string]) | fromUnixtime(NUMERIC[, STRING]) | Returns a representation of the numeric argument as a value in string format (default is ‘yyyy-MM-dd HH:mm:ss’). numeric is an internal timestamp value representing seconds since ‘1970-01-01 00:00:00’ UTC, such as produced by the UNIX_TIMESTAMP() function. The return value is expressed in the session time zone (specified in TableConfig). E.g., FROM_UNIXTIME(44) returns ‘1970-01-01 00:00:44’ if in UTC time zone, but returns ‘1970-01-01 09:00:44’ if in ‘Asia/Tokyo’ time zone. |
 | UNIX_TIMESTAMP() | unixTimestamp() | Gets current Unix timestamp in seconds. This function is not deterministic which means the value would be recalculated for each record. |
-| UNIX_TIMESTAMP(string1[, string2]) | unixTimestamp(STRING1[, STRING2]) | Converts a date time string string1 with format string2 (by default: yyyy-MM-dd HH:mm:ss if not specified) to Unix timestamp (in seconds), using the specified timezone in table config.If a time zone is specified in the date time string and parsed by UTC+X format such as “yyyy-MM-dd HH:mm:ss.SSS X”, this function will use the specified timezone in the date time string instead of the timezone in table config. If the date time string can not be parsed, the default value Long.MIN_VALUE(-9223372036854775808) will be returned.|
+| UNIX_TIMESTAMP(string1[, string2]) | unixTimestamp(STRING1[, STRING2]) | Converts a date time string string1 with format string2 (by default: yyyy-MM-dd HH:mm:ss if not specified) to Unix timestamp (in seconds), using the specified timezone in table config.<br/>If a time zone is specified in the date time string and parsed by UTC+X format such as “yyyy-MM-dd HH:mm:ss.SSS X”, this function will use the specified timezone in the date time string instead of the timezone in table config. If the date time string can not be parsed, the default value Long.MIN_VALUE(-9223372036854775808) will be returned.|
 
 ## Conditional Functions
 

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/DateTimeUtils.java

@@ -146,18 +146,18 @@ private static int ymdToJulian(int year, int month, int day) {
      * Convert unix timestamp (seconds since '1970-01-01 00:00:00' UTC) to datetime string in the
      * "yyyy-MM-dd HH:mm:ss" format.
      */
-    public static String formatUnixTimestamp(long unixtime, TimeZone tz) {
-        return formatUnixTimestamp(unixtime, TIMESTAMP_FORMAT_STRING, tz);
+    public static String formatUnixTimestamp(long unixTime, TimeZone timeZone) {
+        return formatUnixTimestamp(unixTime, TIMESTAMP_FORMAT_STRING, timeZone);
     }
 
     /**
      * Convert unix timestamp (seconds since '1970-01-01 00:00:00' UTC) to datetime string in the
      * given format.
      */
-    public static String formatUnixTimestamp(long unixtime, String format, TimeZone tz) {
+    public static String formatUnixTimestamp(long unixTime, String format, TimeZone timeZone) {
         SimpleDateFormat formatter = FORMATTER_CACHE.get(format);
-        formatter.setTimeZone(tz);
-        Date date = new Date(unixtime * 1000);
+        formatter.setTimeZone(timeZone);
+        Date date = new Date(unixTime * 1000);
         try {
             return formatter.format(date);
         } catch (Exception e) {
@@ -166,25 +166,20 @@ public static String formatUnixTimestamp(long unixtime, String format, TimeZone
         }
     }
 
-    /** Returns the value of the timestamp to seconds since '1970-01-01 00:00:00' UTC. */
-    public static long unixTimestamp(long ts) {
-        return ts / 1000;
-    }
-
     /**
      * Returns the value of the argument as an unsigned integer in seconds since '1970-01-01
      * 00:00:00' UTC.
      */
-    public static long unixTimestamp(String dateStr, TimeZone tz) {
-        return unixTimestamp(dateStr, TIMESTAMP_FORMAT_STRING, tz);
+    public static long unixTimestamp(String dateStr, TimeZone timeZone) {
+        return unixTimestamp(dateStr, TIMESTAMP_FORMAT_STRING, timeZone);
     }
 
     /**
      * Returns the value of the argument as an unsigned integer in seconds since '1970-01-01
      * 00:00:00' UTC.
      */
-    public static long unixTimestamp(String dateStr, String format, TimeZone tz) {
-        long ts = internalParseTimestampMillis(dateStr, format, tz);
+    public static long unixTimestamp(String dateStr, String format, TimeZone timeZone) {
+        long ts = internalParseTimestampMillis(dateStr, format, timeZone);
         if (ts == Long.MIN_VALUE) {
             return Long.MIN_VALUE;
         } else {

flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/SystemFunctionUtils.java

@@ -89,7 +89,7 @@ public static String fromUnixtime(long seconds, String format, String timezone)
     }
 
     public static long unixTimestamp(long epochTime, String timezone) {
-        return DateTimeUtils.unixTimestamp(epochTime);
+        return epochTime / 1000;
     }
 
     public static long unixTimestamp(String dateTimeStr, long epochTime, String timezone) {