[Feature][Zeta] add from_unixtime function

release-note.md

@@ -19,7 +19,7 @@
 ### Formats
 - [Canal]Support read canal format message #3950
 - [Debezium]Support debezium canal format message #3981
- 
+
 ### Connector-V2
 
 - [Json-format] [Canal-Json] Fix json deserialize NPE (#4195)
@@ -76,6 +76,7 @@
 - [Zeta] Fix cpu load problem (#4828)
 - [zeta] Fix the deadlock issue with JDBC driver loading (#4878)
 - [zeta] dynamically replace the value of the variable at runtime (#4950)
+- [Zeta] Add from_unixtime function (#5462)
 
 ### E2E
 

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

@@ -155,6 +155,7 @@ public class ZetaSQLFunction {
     public static final String SECOND = "SECOND";
     public static final String WEEK = "WEEK";
     public static final String YEAR = "YEAR";
+    public static final String FROM_UNIXTIME = "FROM_UNIXTIME";
 
     // -------------------------system functions----------------------------
     public static final String COALESCE = "COALESCE";
@@ -377,6 +378,8 @@ public Object executeFunctionExpr(String functionName, List<Object> args) {
                 return DateTimeFunction.dayOfWeek(args);
             case DAY_OF_YEAR:
                 return DateTimeFunction.dayOfYear(args);
+            case FROM_UNIXTIME:
+                return DateTimeFunction.fromUnixTime(args);
             case EXTRACT:
                 return DateTimeFunction.extract(args);
             case FORMATDATETIME:

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

@@ -236,6 +236,8 @@ private SeaTunnelDataType<?> getFunctionType(Function function) {
             case ZetaSQLFunction.SECOND:
             case ZetaSQLFunction.WEEK:
             case ZetaSQLFunction.YEAR:
+            case ZetaSQLFunction.FROM_UNIXTIME:
+                return BasicType.STRING_TYPE;
             case ZetaSQLFunction.SIGN:
                 return BasicType.INT_TYPE;
             case ZetaSQLFunction.BIT_LENGTH:

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

@@ -23,10 +23,12 @@
 
 import java.text.DateFormatSymbols;
 import java.time.Duration;
+import java.time.Instant;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.time.LocalTime;
 import java.time.Period;
+import java.time.ZoneId;
 import java.time.format.DateTimeFormatter;
 import java.time.temporal.Temporal;
 import java.time.temporal.TemporalAccessor;
@@ -541,4 +543,16 @@ public static Integer year(List<Object> args) {
         LocalDate localDate = convertToLocalDate(datetime);
         return localDate.getYear();
     }
+
+    public static String fromUnixTime(List<Object> args) {
+        Long unixTime = (Long) args.get(0);
+        if (unixTime == null) {
+            return null;
+        }
+        String format = (String) args.get(1);
+        DateTimeFormatter df = DateTimeFormatter.ofPattern(format);
+        LocalDateTime datetime =
+                Instant.ofEpochSecond(unixTime).atZone(ZoneId.systemDefault()).toLocalDateTime();
+        return df.format(datetime);
+    }
 }

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

@@ -0,0 +1,58 @@
+/*
+ * 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;
+
+import org.apache.seatunnel.api.table.type.BasicType;
+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.apache.seatunnel.transform.sql.SQLEngine;
+import org.apache.seatunnel.transform.sql.SQLEngineFactory;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class DateTimeFunctionTest {
+
+    @Test
+    public void testFromUnixtimeFunction() {
+
+        SQLEngine sqlEngine = SQLEngineFactory.getSQLEngine(SQLEngineFactory.EngineType.ZETA);
+
+        SeaTunnelRowType rowType =
+                new SeaTunnelRowType(
+                        new String[] {"unixtime"}, new SeaTunnelDataType[] {BasicType.LONG_TYPE});
+
+        // transform by `from_unixtime` function
+        sqlEngine.init(
+                "test",
+                null,
+                rowType,
+                "select from_unixtime(unixtime,'yyyy-MM-dd HH:mm:ss') as ts from test");
+
+        // 1672502400 means `2023-01-01 00:00:00` in unix time
+        Long unixTime = 1672502400L;
+        SeaTunnelRow inputRow = new SeaTunnelRow(new Long[] {unixTime});
+
+        // transform by sql
+        SeaTunnelRow outRow = sqlEngine.transformBySQL(inputRow);
+        Object field = outRow.getField(0);
+
+        Assertions.assertEquals("2023-01-01 00:00:00", field.toString());
+    }
+}