[INLONG-10833][SDK] Transform SQL supports parsing of Mod methods and % expressions

inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/process/function/ModuloFunction.java

@@ -0,0 +1,54 @@
+/*
+ * 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.
+ */
+
+package org.apache.inlong.sdk.transform.process.function;
+
+import org.apache.inlong.sdk.transform.decode.SourceData;
+import org.apache.inlong.sdk.transform.process.Context;
+import org.apache.inlong.sdk.transform.process.operator.OperatorTools;
+import org.apache.inlong.sdk.transform.process.parser.ValueParser;
+
+import net.sf.jsqlparser.expression.Expression;
+import net.sf.jsqlparser.expression.Function;
+
+import java.math.BigDecimal;
+import java.util.List;
+
+/**
+ * ModuloFunction
+ * description: MOD(NUMERIC1, NUMERIC2) : Return the remainder of numeric1 divided by numeric2.
+ */
+public class ModuloFunction implements ValueParser {
+
+    private ValueParser dividendParser;
+    private ValueParser divisorParser;
+
+    public ModuloFunction(Function expr) {
+        List<Expression> expressions = expr.getParameters().getExpressions();
+        dividendParser = OperatorTools.buildParser(expressions.get(0));
+        divisorParser = OperatorTools.buildParser(expressions.get(1));
+    }
+
+    @Override
+    public Object parse(SourceData sourceData, int rowIndex, Context context) {
+        Object dividendObj = dividendParser.parse(sourceData, rowIndex, context);
+        Object divisorObj = divisorParser.parse(sourceData, rowIndex, context);
+        BigDecimal dividend = OperatorTools.parseBigDecimal(dividendObj);
+        BigDecimal divisor = OperatorTools.parseBigDecimal(divisorObj);
+        return dividend.remainder(divisor);
+    }
+}

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

@@ -32,6 +32,7 @@
 import org.apache.inlong.sdk.transform.process.function.Log10Function;
 import org.apache.inlong.sdk.transform.process.function.Log2Function;
 import org.apache.inlong.sdk.transform.process.function.LogFunction;
+import org.apache.inlong.sdk.transform.process.function.ModuloFunction;
 import org.apache.inlong.sdk.transform.process.function.NowFunction;
 import org.apache.inlong.sdk.transform.process.function.PowerFunction;
 import org.apache.inlong.sdk.transform.process.function.RoundFunction;
@@ -48,6 +49,7 @@
 import org.apache.inlong.sdk.transform.process.parser.DateParser;
 import org.apache.inlong.sdk.transform.process.parser.DivisionParser;
 import org.apache.inlong.sdk.transform.process.parser.LongParser;
+import org.apache.inlong.sdk.transform.process.parser.ModuloParser;
 import org.apache.inlong.sdk.transform.process.parser.MultiplicationParser;
 import org.apache.inlong.sdk.transform.process.parser.ParenthesisParser;
 import org.apache.inlong.sdk.transform.process.parser.StringParser;
@@ -65,6 +67,7 @@
 import net.sf.jsqlparser.expression.TimestampValue;
 import net.sf.jsqlparser.expression.operators.arithmetic.Addition;
 import net.sf.jsqlparser.expression.operators.arithmetic.Division;
+import net.sf.jsqlparser.expression.operators.arithmetic.Modulo;
 import net.sf.jsqlparser.expression.operators.arithmetic.Multiplication;
 import net.sf.jsqlparser.expression.operators.arithmetic.Subtraction;
 import net.sf.jsqlparser.expression.operators.conditional.AndExpression;
@@ -86,7 +89,7 @@
 
 /**
  * OperatorTools
- * 
+ *
  */
 public class OperatorTools {
 
@@ -134,6 +137,7 @@ public class OperatorTools {
         functionMap.put("from_unixtime", FromUnixTimeFunction::new);
         functionMap.put("unix_timestamp", UnixTimestampFunction::new);
         functionMap.put("to_timestamp", ToTimestampFunction::new);
+        functionMap.put("mod", ModuloFunction::new);
     }
 
     public static ExpressionOperator buildOperator(Expression expr) {
@@ -178,6 +182,8 @@ public static ValueParser buildParser(Expression expr) {
             return new MultiplicationParser((Multiplication) expr);
         } else if (expr instanceof Division) {
             return new DivisionParser((Division) expr);
+        } else if (expr instanceof Modulo) {
+            return new ModuloParser((Modulo) expr);
         } else if (expr instanceof DateValue) {
             return new DateParser((DateValue) expr);
         } else if (expr instanceof TimestampValue) {

inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/process/parser/ModuloParser.java

@@ -0,0 +1,51 @@
+/*
+ * 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.
+ */
+
+package org.apache.inlong.sdk.transform.process.parser;
+
+import org.apache.inlong.sdk.transform.decode.SourceData;
+import org.apache.inlong.sdk.transform.process.Context;
+import org.apache.inlong.sdk.transform.process.operator.OperatorTools;
+
+import net.sf.jsqlparser.expression.operators.arithmetic.Modulo;
+
+import java.math.BigDecimal;
+
+/**
+ * ModuloParser
+ * description: analyze the % expression
+ */
+public class ModuloParser implements ValueParser {
+
+    private ValueParser left;
+
+    private ValueParser right;
+
+    public ModuloParser(Modulo expr) {
+        this.left = OperatorTools.buildParser(expr.getLeftExpression());
+        this.right = OperatorTools.buildParser(expr.getRightExpression());
+    }
+
+    @Override
+    public Object parse(SourceData sourceData, int rowIndex, Context context) {
+        Object leftObj = this.left.parse(sourceData, rowIndex, context);
+        Object rightObj = this.right.parse(sourceData, rowIndex, context);
+        BigDecimal leftValue = OperatorTools.parseBigDecimal(leftObj);
+        BigDecimal rightValue = OperatorTools.parseBigDecimal(rightObj);
+        return leftValue.remainder(rightValue);
+    }
+}

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

@@ -55,6 +55,37 @@ public class TestTransformArithmeticFunctionsProcessor {
         kvSink = new KvSinkInfo("UTF-8", dstFields);
     }
 
+    @Test
+    public void testModuloFunction() throws Exception {
+        String transformFunctionSql = "select mod(numeric1,100) from source";
+        String transformExpressionSql = "select numeric1 % 100 from source";
+        TransformConfig functionConfig = new TransformConfig(transformFunctionSql);
+        TransformProcessor<String, String> functionProcessor = TransformProcessor
+                .create(functionConfig, SourceDecoderFactory.createCsvDecoder(csvSource),
+                        SinkEncoderFactory.createKvEncoder(kvSink));
+        TransformConfig expressionConfig = new TransformConfig(transformExpressionSql);
+        TransformProcessor<String, String> expressionProcessor = TransformProcessor
+                .create(expressionConfig, SourceDecoderFactory.createCsvDecoder(csvSource),
+                        SinkEncoderFactory.createKvEncoder(kvSink));
+        // case1: "3.1415926|4a|4|8"
+        String data = "3.1415926|4a|4|8";
+        List<String> output1 = functionProcessor.transform(data);
+        Assert.assertEquals(1, output1.size());
+        Assert.assertEquals("result=3.1415926", output1.get(0));
+        List<String> output2 = expressionProcessor.transform(data);
+        Assert.assertEquals(1, output1.size());
+        Assert.assertEquals("result=3.1415926", output2.get(0));
+
+        // case2: "-3.1415926|4a|4|8"
+        data = "-3.1415926|4a|4|8";
+        output1 = functionProcessor.transform(data);
+        Assert.assertEquals(1, output1.size());
+        Assert.assertEquals("result=-3.1415926", output1.get(0));
+        output2 = expressionProcessor.transform(data);
+        Assert.assertEquals(1, output1.size());
+        Assert.assertEquals("result=-3.1415926", output2.get(0));
+    }
+
     @Test
     public void testRoundFunction() throws Exception {
         String transformSql = "select round(numeric1) from source";