[Improve][Sql] Support boolean fields in case-when condition clauses (#9073)

Author: hailin0

Diff for: docs/en/transform-v2/sql-functions.md

@@ -991,7 +991,11 @@ select
   case
     when c_tinyint <> 117 then 1
     else 0
-  end as c_number_0
+  end as c_number_0,
+  case
+    when c_boolean then 1
+    else 0
+  end as c_boolean_0
 from
   dual
 ```

Diff for: seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLFilter.java

@@ -39,6 +39,7 @@
 import net.sf.jsqlparser.expression.operators.relational.MinorThanEquals;
 import net.sf.jsqlparser.expression.operators.relational.NotEqualsTo;
 import net.sf.jsqlparser.expression.operators.relational.ParenthesedExpressionList;
+import net.sf.jsqlparser.schema.Column;
 
 import java.time.LocalDate;
 import java.time.LocalDateTime;
@@ -114,6 +115,9 @@ public boolean executeFilter(Expression whereExpr, Object[] inputFields) {
         if (whereExpr instanceof Parenthesis) {
             return parenthesisExpr((Parenthesis) whereExpr, inputFields);
         }
+        if (whereExpr instanceof Column) {
+            return (boolean) zetaSQLFunction.computeForValue(whereExpr, inputFields);
+        }
         throw new TransformException(
                 CommonErrorCodeDeprecated.UNSUPPORTED_OPERATION,
                 String.format("Unsupported SQL Expression: %s ", whereExpr));

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

@@ -282,4 +282,37 @@ public void testEscapeIdentifier() {
                 BasicType.STRING_TYPE, tableSchema.getColumns().get(1).getDataType());
         Assertions.assertEquals("a", result.get(0).getField(1));
     }
+
+    @Test
+    public void tesCaseWhenClausesWithBooleanField() {
+        String tableName = "test";
+        String[] fields = new String[] {"id", "bool"};
+        CatalogTable table =
+                CatalogTableUtil.getCatalogTable(
+                        tableName,
+                        new SeaTunnelRowType(
+                                fields,
+                                new SeaTunnelDataType[] {
+                                    BasicType.INT_TYPE, BasicType.BOOLEAN_TYPE
+                                }));
+        ReadonlyConfig config =
+                ReadonlyConfig.fromMap(
+                        Collections.singletonMap(
+                                "query",
+                                "select `id`, `bool`, case when bool then 1 else 2 end as bool_1 from dual"));
+        SQLTransform sqlTransform = new SQLTransform(config, table);
+        List<SeaTunnelRow> result =
+                sqlTransform.transformRow(
+                        new SeaTunnelRow(new Object[] {Integer.valueOf(1), true}));
+        Assertions.assertEquals(1, result.get(0).getField(0));
+        Assertions.assertEquals(true, result.get(0).getField(1));
+        Assertions.assertEquals(1, result.get(0).getField(2));
+
+        result =
+                sqlTransform.transformRow(
+                        new SeaTunnelRow(new Object[] {Integer.valueOf(1), false}));
+        Assertions.assertEquals(1, result.get(0).getField(0));
+        Assertions.assertEquals(false, result.get(0).getField(1));
+        Assertions.assertEquals(2, result.get(0).getField(2));
+    }
 }