[Feature][Transform] support boolean type for sql transform

docs/en/transform-v2/sql-functions.md

@@ -917,11 +917,17 @@ CALL FROM_UNIXTIME(1672502400, 'yyyy-MM-dd HH:mm:ss','UTC+6')
 
 Converts a value to another data type.
 
-Supported data types: STRING | VARCHAR, INT | INTEGER, LONG | BIGINT, BYTE, FLOAT, DOUBLE, DECIMAL(p,s), TIMESTAMP, DATE, TIME, BYTES
+Supported data types: STRING | VARCHAR, INT | INTEGER, LONG | BIGINT, BYTE, FLOAT, DOUBLE, DECIMAL(p,s), TIMESTAMP, DATE, TIME, BYTES, BOOLEAN
 
 Example:
+* CAST(NAME AS INT)
+* CAST(FLAG AS BOOLEAN)
 
-CONVERT(NAME AS INT)
+NOTE:
+Converts a value to a BOOLEAN data type according to the following rules:
+1. If the value can be interpreted as a boolean string (`'true'` or `'false'`), it returns the corresponding boolean value.
+2. If the value can be interpreted as a numeric value (`1` or `0`), it returns `true` for `1` and `false` for `0`.
+3. If the value cannot be interpreted according to the above rules, it throws a `TransformException`.
 
 ### TRY_CAST
 
@@ -1016,7 +1022,9 @@ It is used to determine whether the condition is valid and return different valu
 
 Example:
 
-case when c_string in ('c_string') then 1 else 0 end
+case when c_string in ('c_string') then 1 else 0 end 
+
+case when c_string in ('c_string') then true else false end
 
 ### UUID
 

docs/zh/transform-v2/sql-functions.md

@@ -914,7 +914,15 @@ CALL FROM_UNIXTIME(1672502400, 'yyyy-MM-dd HH:mm:ss','UTC+6')
 
 示例:
 
-CONVERT(NAME AS INT)
+CAST(NAME AS INT)
+
+CAST(FLAG AS BOOLEAN)
+
+注意：将值转换为布尔数据类型时，遵循以下规则：
+
+1.  如果值可以被解释为布尔字符串（'true' 或 'false'），则返回相应的布尔值。
+2.  如果值可以被解释为数值（1 或 0），则对于 1 返回 true，对于 0 返回 false。
+3.  如果值无法根据以上规则进行解释，则抛出 TransformException 异常。
 
 ### TRY_CAST
 
@@ -1007,6 +1015,8 @@ from
 
 case when c_string in ('c_string') then 1 else 0 end
 
+case when c_string in ('c_string') then true else false end
+
 ### UUID
 
 ```UUID()```

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

@@ -270,6 +270,11 @@ public Object computeForValue(Expression expression, Object[] inputFields) {
                 columnName = columnName.substring(1, columnName.length() - 1);
                 index = inputRowType.indexOf(columnName, false);
             }
+            if (index == -1
+                    && ("true".equalsIgnoreCase(columnName)
+                            || "false".equalsIgnoreCase(columnName))) {
+                return Boolean.parseBoolean(columnName);
+            }
 
             if (index != -1) {
                 return inputFields[index];

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

@@ -81,6 +81,7 @@ public class ZetaSQLType {
     public static final String DATETIME = "DATETIME";
     public static final String DATE = "DATE";
     public static final String TIME = "TIME";
+    public static final String BOOLEAN = "BOOLEAN";
 
     private final SeaTunnelRowType inputRowType;
 
@@ -121,7 +122,11 @@ public SeaTunnelDataType<?> getExpressionType(Expression expression) {
                 columnName = columnName.substring(1, columnName.length() - 1);
                 index = inputRowType.indexOf(columnName, false);
             }
-
+            if (index == -1
+                    && ("true".equalsIgnoreCase(columnName)
+                            || "false".equalsIgnoreCase(columnName))) {
+                return BasicType.BOOLEAN_TYPE;
+            }
             if (index != -1) {
                 return inputRowType.getFieldType(index);
             } else {
@@ -352,6 +357,8 @@ private SeaTunnelDataType<?> getCastType(CastExpression castExpression) {
                 return LocalTimeType.LOCAL_DATE_TYPE;
             case TIME:
                 return LocalTimeType.LOCAL_TIME_TYPE;
+            case BOOLEAN:
+                return BasicType.BOOLEAN_TYPE;
             default:
                 throw new TransformException(
                         CommonErrorCodeDeprecated.UNSUPPORTED_OPERATION,

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

@@ -33,6 +33,7 @@
 import java.time.LocalTime;
 import java.time.ZoneId;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 
 public class SystemFunction {
@@ -177,6 +178,26 @@ public static Object castAs(List<Object> args) {
                 BigDecimal bigDecimal = new BigDecimal(v1.toString());
                 Integer scale = (Integer) args.get(3);
                 return bigDecimal.setScale(scale, RoundingMode.CEILING);
+            case "BOOLEAN":
+                if (v1 instanceof Number) {
+                    if (Arrays.asList(1, 0).contains(((Number) v1).intValue())) {
+                        return ((Number) v1).intValue() == 1;
+                    } else {
+                        throw new TransformException(
+                                CommonErrorCodeDeprecated.UNSUPPORTED_OPERATION,
+                                String.format("Unsupported CAST AS Boolean: %s", v1));
+                    }
+                } else if (v1 instanceof String) {
+                    if (Arrays.asList("TRUE", "FALSE").contains(v1.toString().toUpperCase())) {
+                        return Boolean.parseBoolean(v1.toString());
+                    } else {
+                        throw new TransformException(
+                                CommonErrorCodeDeprecated.UNSUPPORTED_OPERATION,
+                                String.format("Unsupported CAST AS Boolean: %s", v1));
+                    }
+                } else if (v1 instanceof Boolean) {
+                    return v1;
+                }
         }
         throw new TransformException(
                 CommonErrorCodeDeprecated.UNSUPPORTED_OPERATION,

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

@@ -29,6 +29,7 @@
 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.exception.TransformException;
 
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
@@ -315,4 +316,129 @@ public void tesCaseWhenClausesWithBooleanField() {
         Assertions.assertEquals(false, result.get(0).getField(1));
         Assertions.assertEquals(2, result.get(0).getField(2));
     }
+
+    @Test
+    public void tesCaseWhenBooleanClausesWithField() {
+        String tableName = "test";
+        String[] fields = new String[] {"id", "int", "string"};
+        CatalogTable table =
+                CatalogTableUtil.getCatalogTable(
+                        tableName,
+                        new SeaTunnelRowType(
+                                fields,
+                                new SeaTunnelDataType[] {
+                                    BasicType.INT_TYPE, BasicType.INT_TYPE, BasicType.STRING_TYPE
+                                }));
+        ReadonlyConfig config =
+                ReadonlyConfig.fromMap(
+                        Collections.singletonMap(
+                                "query",
+                                "select `id`, `int`, (case when `int` = 1 then true else false end) as bool_1 , `string`, (case when `string` = 'true' then true else false end) as bool_2 from dual"));
+        SQLTransform sqlTransform = new SQLTransform(config, table);
+        List<SeaTunnelRow> result =
+                sqlTransform.transformRow(new SeaTunnelRow(new Object[] {1, 1, "true"}));
+
+        Assertions.assertEquals(1, result.get(0).getField(0));
+        Assertions.assertEquals(1, result.get(0).getField(1));
+        Assertions.assertEquals(true, result.get(0).getField(2));
+        Assertions.assertEquals("true", result.get(0).getField(3));
+        Assertions.assertEquals(true, result.get(0).getField(4));
+
+        result = sqlTransform.transformRow(new SeaTunnelRow(new Object[] {1, 0, "false"}));
+        Assertions.assertEquals(1, result.get(0).getField(0));
+        Assertions.assertEquals(0, result.get(0).getField(1));
+        Assertions.assertEquals(false, result.get(0).getField(2));
+        Assertions.assertEquals("false", result.get(0).getField(3));
+        Assertions.assertEquals(false, result.get(0).getField(4));
+    }
+
+    @Test
+    public void tesCastBooleanClausesWithField() {
+        String tableName = "test";
+        String[] fields = new String[] {"id", "int", "string"};
+        CatalogTable table =
+                CatalogTableUtil.getCatalogTable(
+                        tableName,
+                        new SeaTunnelRowType(
+                                fields,
+                                new SeaTunnelDataType[] {
+                                    BasicType.INT_TYPE, BasicType.INT_TYPE, BasicType.STRING_TYPE
+                                }));
+        ReadonlyConfig config =
+                ReadonlyConfig.fromMap(
+                        Collections.singletonMap(
+                                "query",
+                                "select `id`, `int`, cast(`int` as boolean) as bool_1 , `string`, cast(`string` as boolean) as bool_2 from dual"));
+        SQLTransform sqlTransform = new SQLTransform(config, table);
+        List<SeaTunnelRow> result =
+                sqlTransform.transformRow(
+                        new SeaTunnelRow(new Object[] {Integer.valueOf(1), 1, "true"}));
+
+        Assertions.assertEquals(1, result.get(0).getField(0));
+        Assertions.assertEquals(1, result.get(0).getField(1));
+        Assertions.assertEquals(true, result.get(0).getField(2));
+        Assertions.assertEquals("true", result.get(0).getField(3));
+        Assertions.assertEquals(true, result.get(0).getField(4));
+
+        result =
+                sqlTransform.transformRow(
+                        new SeaTunnelRow(new Object[] {Integer.valueOf(1), 0, "false"}));
+        Assertions.assertEquals(1, result.get(0).getField(0));
+        Assertions.assertEquals(0, result.get(0).getField(1));
+        Assertions.assertEquals(false, result.get(0).getField(2));
+        Assertions.assertEquals("false", result.get(0).getField(3));
+        Assertions.assertEquals(false, result.get(0).getField(4));
+
+        Assertions.assertThrows(
+                TransformException.class,
+                () -> {
+                    try {
+                        sqlTransform.transformRow(
+                                new SeaTunnelRow(new Object[] {Integer.valueOf(1), 3, "false"}));
+                    } catch (Exception e) {
+                        Assertions.assertEquals(
+                                "ErrorCode:[COMMON-05], ErrorDescription:[Unsupported operation] - Unsupported CAST AS Boolean: 3",
+                                e.getMessage());
+                        throw e;
+                    }
+                });
+
+        Assertions.assertThrows(
+                TransformException.class,
+                () -> {
+                    try {
+                        sqlTransform.transformRow(
+                                new SeaTunnelRow(new Object[] {Integer.valueOf(1), 0, "false333"}));
+                    } catch (Exception e) {
+                        Assertions.assertEquals(
+                                "ErrorCode:[COMMON-05], ErrorDescription:[Unsupported operation] - Unsupported CAST AS Boolean: false333",
+                                e.getMessage());
+                        throw e;
+                    }
+                });
+    }
+
+    @Test
+    public void tesBooleanField() {
+        String tableName = "test";
+        String[] fields = new String[] {"id", "int", "string"};
+        CatalogTable table =
+                CatalogTableUtil.getCatalogTable(
+                        tableName,
+                        new SeaTunnelRowType(
+                                fields,
+                                new SeaTunnelDataType[] {
+                                    BasicType.INT_TYPE, BasicType.INT_TYPE, BasicType.STRING_TYPE
+                                }));
+        ReadonlyConfig config =
+                ReadonlyConfig.fromMap(
+                        Collections.singletonMap(
+                                "query", "select `id`, true as bool_1, false as bool_2 from dual"));
+        SQLTransform sqlTransform = new SQLTransform(config, table);
+        List<SeaTunnelRow> result =
+                sqlTransform.transformRow(new SeaTunnelRow(new Object[] {1, 1, "true"}));
+        Assertions.assertEquals(1, result.get(0).getField(0));
+        Assertions.assertEquals(true, result.get(0).getField(1));
+        Assertions.assertEquals(false, result.get(0).getField(2));
+    }
 }