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

Author: hawk9821

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

@@ -59,7 +59,6 @@
 import net.sf.jsqlparser.expression.operators.arithmetic.Multiplication;
 import net.sf.jsqlparser.expression.operators.arithmetic.Subtraction;
 import net.sf.jsqlparser.expression.operators.relational.ExpressionList;
-import net.sf.jsqlparser.parser.Node;
 import net.sf.jsqlparser.schema.Column;
 import net.sf.jsqlparser.statement.select.LateralView;
 
@@ -264,19 +263,18 @@ public Object computeForValue(Expression expression, Object[] inputFields) {
         if (expression instanceof Column) {
             Column columnExp = (Column) expression;
             String columnName = columnExp.getColumnName();
-            Node node = columnExp.getASTNode().jjtGetParent();
-            if (("true".equalsIgnoreCase(columnExp.getColumnName())
-                            || "false".equalsIgnoreCase(columnExp.getColumnName()))
-                    && ("CaseWhenExpression".equalsIgnoreCase(node.toString()))) {
-                return Boolean.parseBoolean(columnExp.getColumnName());
-            }
             int index = inputRowType.indexOf(columnName, false);
             if (index == -1
                     && columnName.startsWith(ZetaSQLEngine.ESCAPE_IDENTIFIER)
                     && columnName.endsWith(ZetaSQLEngine.ESCAPE_IDENTIFIER)) {
                 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

@@ -56,7 +56,6 @@
 import net.sf.jsqlparser.expression.operators.relational.IsNullExpression;
 import net.sf.jsqlparser.expression.operators.relational.LikeExpression;
 import net.sf.jsqlparser.expression.operators.relational.NotEqualsTo;
-import net.sf.jsqlparser.parser.Node;
 import net.sf.jsqlparser.schema.Column;
 
 import java.util.ArrayList;
@@ -115,12 +114,6 @@ public SeaTunnelDataType<?> getExpressionType(Expression expression) {
         }
         if (expression instanceof Column) {
             Column columnExp = (Column) expression;
-            Node node = columnExp.getASTNode().jjtGetParent();
-            if (("true".equalsIgnoreCase(columnExp.getColumnName())
-                            || "false".equalsIgnoreCase(columnExp.getColumnName()))
-                    && ("CaseWhenExpression".equalsIgnoreCase(node.toString()))) {
-                return BasicType.BOOLEAN_TYPE;
-            }
             String columnName = columnExp.getColumnName();
             int index = inputRowType.indexOf(columnName, false);
             if (index == -1
@@ -129,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 {