[Feature][Transform-V2][SQL] Support case when clause for SQL Transform plugin (#5013)

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

@@ -891,3 +891,19 @@ Returns NULL if 'a' is equal to 'b', otherwise 'a'.
 Example:
 
 NULLIF(A, B)
+
+## Conditional Functions
+
+### CASE
+
+```sql
+CASE value WHEN compare_value THEN result [WHEN compare_value THEN result ...] [ELSE result] END
+CASE WHEN condition THEN result [WHEN condition THEN result ...] [ELSE result] END
+```
+
+Converts a value to another.
+
+Example:
+
+```CASE state WHEN 1 THEN 'on' ELSE 'off' END```
+```CASE WHEN state>0 THEN 1 ELSE 0 END```

seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/java/org/apache/seatunnel/e2e/transform/TestSQLIT.java

@@ -52,5 +52,17 @@ public void testSQLTransform(TestContainer container) throws IOException, Interr
         Container.ExecResult sqlAllColumns =
                 container.executeJob("/sql_transform/sql_all_columns.conf");
         Assertions.assertEquals(0, sqlAllColumns.getExitCode());
+
+        // region case when
+        Container.ExecResult caseFieldWhenCondition =
+                container.executeJob("/sql_transform/case_field_when_condition.conf");
+        Assertions.assertEquals(0, caseFieldWhenCondition.getExitCode());
+        Container.ExecResult caseFieldWhenValue =
+                container.executeJob("/sql_transform/case_field_when_value.conf");
+        Assertions.assertEquals(0, caseFieldWhenValue.getExitCode());
+        Container.ExecResult caseWhenCondition =
+                container.executeJob("/sql_transform/case_when_condition.conf");
+        Assertions.assertEquals(0, caseWhenCondition.getExitCode());
+        // endregion
     }
 }

seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/sql_transform/case_field_when_condition.conf

@@ -0,0 +1,83 @@
+#
+# 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.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config
+######
+
+env {
+  execution.parallelism = 1
+  job.mode = "BATCH"
+  checkpoint.interval = 10000
+}
+
+source {
+  FakeSource {
+    result_table_name = "fake"
+    schema = {
+      fields {
+        state = "int"
+        name = "string"
+        price = "double"
+      }
+    }
+    rows = [
+      {fields = [-6, "javalover123", 134.22], kind = INSERT}
+    ]
+  }
+}
+
+transform {
+  Sql {
+    source_table_name = "fake"
+    result_table_name = "fake1"
+    query = "select case state when state>0 then 1 else 0 end show_tag, name, price from fake"
+  }
+}
+
+sink {
+  Console {
+    source_table_name = "fake1"
+  }
+  Assert {
+    source_table_name = "fake1"
+    rules = {
+      field_rules = [
+        {
+          field_name = "show_tag"
+          field_type = "int"
+          field_value = [
+            {equals_to = 0}
+          ]
+        },
+        {
+          field_name = "name"
+          field_type = "string"
+          field_value = [
+            {equals_to = "javalover123"}
+          ]
+        },
+        {
+          field_name = "price"
+          field_type = "double"
+          field_value = [
+            {equals_to = 134.22}
+          ]
+        }
+      ]
+    }
+  }
+}

seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/sql_transform/case_field_when_value.conf

@@ -0,0 +1,83 @@
+#
+# 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.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config
+######
+
+env {
+  execution.parallelism = 1
+  job.mode = "BATCH"
+  checkpoint.interval = 10000
+}
+
+source {
+  FakeSource {
+    result_table_name = "fake"
+    schema = {
+      fields {
+        state = "bigint"
+        name = "string"
+        price = "double"
+      }
+    }
+    rows = [
+      {fields = [6, "javalover123", 134.22], kind = INSERT}
+    ]
+  }
+}
+
+transform {
+  Sql {
+    source_table_name = "fake"
+    result_table_name = "fake1"
+    query = "select case state when 6 then 1 else 0 end show_tag, name, price from fake"
+  }
+}
+
+sink {
+  Console {
+    source_table_name = "fake1"
+  }
+  Assert {
+    source_table_name = "fake1"
+    rules = {
+      field_rules = [
+        {
+          field_name = "show_tag"
+          field_type = "int"
+          field_value = [
+            {equals_to = 1}
+          ]
+        },
+        {
+          field_name = "name"
+          field_type = "string"
+          field_value = [
+            {equals_to = "javalover123"}
+          ]
+        },
+        {
+          field_name = "price"
+          field_type = "double"
+          field_value = [
+            {equals_to = 134.22}
+          ]
+        }
+      ]
+    }
+  }
+}

seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/sql_transform/case_when_condition.conf

@@ -0,0 +1,83 @@
+#
+# 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.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config
+######
+
+env {
+  execution.parallelism = 1
+  job.mode = "BATCH"
+  checkpoint.interval = 10000
+}
+
+source {
+  FakeSource {
+    result_table_name = "fake"
+    schema = {
+      fields {
+        state = "int"
+        name = "string"
+        price = "double"
+      }
+    }
+    rows = [
+      {fields = [6, "javalover123", 134.22], kind = INSERT}
+    ]
+  }
+}
+
+transform {
+  Sql {
+    source_table_name = "fake"
+    result_table_name = "fake1"
+    query = "select case when state>0 then 1 else 0 end show_tag, name, price from fake"
+  }
+}
+
+sink {
+  Console {
+    source_table_name = "fake1"
+  }
+  Assert {
+    source_table_name = "fake1"
+    rules = {
+      field_rules = [
+        {
+          field_name = "show_tag"
+          field_type = "int"
+          field_value = [
+            {equals_to = 1}
+          ]
+        },
+        {
+          field_name = "name"
+          field_type = "string"
+          field_value = [
+            {equals_to = "javalover123"}
+          ]
+        },
+        {
+          field_name = "price"
+          field_type = "double"
+          field_value = [
+            {equals_to = 134.22}
+          ]
+        }
+      ]
+    }
+  }
+}

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

@@ -185,7 +185,7 @@ private Pair<Object, Object> executeComparisonOperator(
         return Pair.of(leftVal, rightVal);
     }
 
-    private boolean equalsToExpr(Pair<Object, Object> pair) {
+    boolean equalsToExpr(Pair<Object, Object> pair) {
         Object leftVal = pair.getLeft();
         Object rightVal = pair.getRight();
         if (leftVal == null || rightVal == null) {