[INLONG-12056][SDK] TransformSDK supports the "not in" operator (#12057)

Author: luchunliang

inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/process/operator/InOperator.java

@@ -39,8 +39,10 @@ public class InOperator implements ExpressionOperator {
 
     private final ValueParser left;
     private final List<ValueParser> right;
+    private final boolean isNot;
 
     public InOperator(InExpression expr) {
+        this.isNot = expr.isNot();
         this.left = OperatorTools.buildParser(expr.getLeftExpression());
         ItemsList itemsList = expr.getRightItemsList();
         this.right = new ArrayList<>();
@@ -63,9 +65,17 @@ public InOperator(InExpression expr) {
      * @param rowIndex
      * @return
      */
-    @SuppressWarnings("rawtypes")
     @Override
     public boolean check(SourceData sourceData, int rowIndex, Context context) {
+        if (isNot) {
+            return !this.checkIn(sourceData, rowIndex, context);
+        } else {
+            return this.checkIn(sourceData, rowIndex, context);
+        }
+    }
+
+    @SuppressWarnings("rawtypes")
+    private boolean checkIn(SourceData sourceData, int rowIndex, Context context) {
         Comparable leftValue = (Comparable) this.left.parse(sourceData, rowIndex, context);
         for (ValueParser parser : right) {
             Comparable rightValue = (Comparable) parser.parse(sourceData, rowIndex, context);

inlong-sdk/transform-sdk/src/test/java/org/apache/inlong/sdk/transform/process/operator/TestInOperator.java

@@ -42,11 +42,11 @@ public void testAndOperator() throws Exception {
         Assert.assertEquals(output1.get(0), "result=1");
         // case2: "3.14159265358979323846|5|4|8"
         List<String> output2 = processor.transform("3.14159265358979323846|5|4|8");
-        Assert.assertEquals(1, output1.size());
+        Assert.assertEquals(1, output2.size());
         Assert.assertEquals(output2.get(0), "result=1");
         // case3: "3.14159265358979323846|3|4|8"
         List<String> output3 = processor.transform("3.14159265358979323846|3|4|8");
-        Assert.assertEquals(1, output1.size());
+        Assert.assertEquals(1, output3.size());
         Assert.assertEquals(output3.get(0), "result=0");
 
         transformSql = "select if(numeric3 IN (4,3.2),1,0) from source";
@@ -56,15 +56,34 @@ public void testAndOperator() throws Exception {
                 .create(config, SourceDecoderFactory.createCsvDecoder(csvSource),
                         SinkEncoderFactory.createKvEncoder(kvSink));
         List<String> output4 = processor.transform("3.14159265358979323846|4|4|8");
-        Assert.assertEquals(1, output1.size());
+        Assert.assertEquals(1, output4.size());
         Assert.assertEquals(output4.get(0), "result=1");
         // case5: "3.14159265358979323846|4|3.2|4"
         List<String> output5 = processor.transform("3.14159265358979323846|4|3.2|4");
-        Assert.assertEquals(1, output1.size());
+        Assert.assertEquals(1, output5.size());
         Assert.assertEquals(output5.get(0), "result=1");
         // case6: "3.14159265358979323846|4|3.2|8"
         List<String> output6 = processor.transform("3.14159265358979323846|4|4.2|8");
-        Assert.assertEquals(1, output1.size());
+        Assert.assertEquals(1, output6.size());
         Assert.assertEquals(output6.get(0), "result=0");
+
+        // not in
+        transformSql = "select if(numeric3 not IN (4,3.2),1,0) from source";
+        config = new TransformConfig(transformSql);
+        // case7: "3.14159265358979323846|4|4|8"
+        processor = TransformProcessor
+                .create(config, SourceDecoderFactory.createCsvDecoder(csvSource),
+                        SinkEncoderFactory.createKvEncoder(kvSink));
+        List<String> output7 = processor.transform("3.14159265358979323846|4|4|8");
+        Assert.assertEquals(1, output7.size());
+        Assert.assertEquals(output7.get(0), "result=0");
+        // case8: "3.14159265358979323846|4|3.2|4"
+        List<String> output8 = processor.transform("3.14159265358979323846|4|3.2|4");
+        Assert.assertEquals(1, output8.size());
+        Assert.assertEquals(output8.get(0), "result=0");
+        // case9: "3.14159265358979323846|4|3.2|8"
+        List<String> output9 = processor.transform("3.14159265358979323846|4|4.2|8");
+        Assert.assertEquals(1, output9.size());
+        Assert.assertEquals(output9.get(0), "result=1");
     }
 }