Merge branch 'master' into INLONG-11010

# Conflicts:
#	inlong-sdk/transform-sdk/src/test/java/org/apache/inlong/sdk/transform/process/TestTransformArithmeticFunctionsProcessor.java
#	inlong-sdk/transform-sdk/src/test/java/org/apache/inlong/sdk/transform/process/TestTransformStringFunctionsProcessor.java

inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/process/function/AsinFunction.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.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.Function;
+
+import java.math.BigDecimal;
+
+/**
+ * AsinFunction
+ * description: asin(numeric)--returns the arc sine of numeric
+ */
+@TransformFunction(names = {"asin"})
+public class AsinFunction implements ValueParser {
+
+    private ValueParser numberParser;
+
+    public AsinFunction(Function expr) {
+        numberParser = OperatorTools.buildParser(expr.getParameters().getExpressions().get(0));
+    }
+
+    @Override
+    public Object parse(SourceData sourceData, int rowIndex, Context context) {
+        Object numberObj = numberParser.parse(sourceData, rowIndex, context);
+        if (numberObj == null) {
+            throw new NullPointerException("Parsed number object is null");
+        }
+        BigDecimal numberValue = OperatorTools.parseBigDecimal(numberObj);
+        return Math.asin(numberValue.doubleValue());
+    }
+}

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

@@ -0,0 +1,58 @@
+/*
+ * 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.Function;
+
+import java.math.BigDecimal;
+
+/**
+ * Atan2Function
+ * description: atan2(numeric)--returns the arc tangent of a coordinate (numeric1, numeric2).
+ */
+@TransformFunction(names = {"atan2"})
+public class Atan2Function implements ValueParser {
+
+    private ValueParser xParser;
+    private ValueParser yParser;
+
+    public Atan2Function(Function expr) {
+        xParser = OperatorTools.buildParser(expr.getParameters().getExpressions().get(0));
+        yParser = OperatorTools.buildParser(expr.getParameters().getExpressions().get(1));
+    }
+
+    @Override
+    public Object parse(SourceData sourceData, int rowIndex, Context context) {
+        Object xObj = xParser.parse(sourceData, rowIndex, context);
+        Object yObj = yParser.parse(sourceData, rowIndex, context);
+
+        if (xObj == null) {
+            throw new NullPointerException("Parsed number object on the x-axis is null");
+        }
+
+        BigDecimal xValue = OperatorTools.parseBigDecimal(xObj);
+        BigDecimal yValue = OperatorTools.parseBigDecimal(yObj);
+
+        return Math.atan2(xValue.doubleValue(), yValue.doubleValue());
+    }
+}

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

@@ -0,0 +1,48 @@
+/*
+ * 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.Function;
+
+import java.math.BigDecimal;
+
+/**
+ * AtanFunction
+ * description: atan(numeric)--returns the arc tangent of numeric
+ */
+@TransformFunction(names = {"atan"})
+public class AtanFunction implements ValueParser {
+
+    private ValueParser numberParser;
+
+    public AtanFunction(Function expr) {
+        numberParser = OperatorTools.buildParser(expr.getParameters().getExpressions().get(0));
+    }
+
+    @Override
+    public Object parse(SourceData sourceData, int rowIndex, Context context) {
+        Object numberObj = numberParser.parse(sourceData, rowIndex, context);
+        BigDecimal numberValue = OperatorTools.parseBigDecimal(numberObj);
+        return Math.atan(numberValue.doubleValue());
+    }
+}

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

@@ -0,0 +1,56 @@
+/*
+ * 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.util.List;
+
+/**
+ * ContainsFunction
+ * description: contains(left, right) - Returns a boolean.
+ * The value is True if right is found inside left, otherwise, returns False.
+ * Both left or right must be of STRING type.
+ */
+@TransformFunction(names = {"contains"})
+public class ContainsFunction implements ValueParser {
+
+    private ValueParser leftStrParser;
+    private ValueParser rightStrParser;
+
+    public ContainsFunction(Function expr) {
+        List<Expression> expressions = expr.getParameters().getExpressions();
+        leftStrParser = OperatorTools.buildParser(expressions.get(0));
+        rightStrParser = OperatorTools.buildParser(expressions.get(1));
+    }
+
+    @Override
+    public Object parse(SourceData sourceData, int rowIndex, Context context) {
+        Object leftStrObj = leftStrParser.parse(sourceData, rowIndex, context);
+        Object rightStrObj = rightStrParser.parse(sourceData, rowIndex, context);
+        String leftStr = OperatorTools.parseString(leftStrObj);
+        String rightStr = OperatorTools.parseString(rightStrObj);
+        return (leftStr == null || rightStr == null) ? null : leftStr.contains(rightStr);
+    }
+}

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

@@ -0,0 +1,48 @@
+/*
+ * 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.Function;
+
+import java.math.BigDecimal;
+
+/**
+ * CoshFunction
+ * description: cosh(numeric)--returns the hyperbolic cosine of numeric
+ */
+@TransformFunction(names = {"cosh"})
+public class CoshFunction implements ValueParser {
+
+    private ValueParser numberParser;
+
+    public CoshFunction(Function expr) {
+        numberParser = OperatorTools.buildParser(expr.getParameters().getExpressions().get(0));
+    }
+
+    @Override
+    public Object parse(SourceData sourceData, int rowIndex, Context context) {
+        Object numberObj = numberParser.parse(sourceData, rowIndex, context);
+        BigDecimal numberValue = OperatorTools.parseBigDecimal(numberObj);
+        return Math.cosh(numberValue.doubleValue());
+    }
+}

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

@@ -0,0 +1,55 @@
+/*
+ * 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.Function;
+
+import java.math.BigDecimal;
+
+/**
+ * CotFunction
+ * description: cot(numeric) -- returns the cotangent of the numeric (in radians)
+ */
+@TransformFunction(names = {"cot"})
+public class CotFunction implements ValueParser {
+
+    private final ValueParser valueParser;
+
+    public CotFunction(Function expr) {
+        this.valueParser = OperatorTools.buildParser(expr.getParameters().getExpressions().get(0));
+    }
+
+    @Override
+    public Object parse(SourceData sourceData, int rowIndex, Context context) {
+        Object valueObj = valueParser.parse(sourceData, rowIndex, context);
+
+        BigDecimal value = OperatorTools.parseBigDecimal(valueObj);
+
+        // Calculate tan(x) and take the inverse to find cot(x)
+        double tanValue = Math.tan(value.doubleValue());
+        if (tanValue == 0) {
+            throw new ArithmeticException("Cotangent undefined for this input, tan(x) is zero.");
+        }
+        return 1.0 / tanValue;
+    }
+}

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

@@ -0,0 +1,66 @@
+/*
+ * 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.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.List;
+
+/**
+ * FromBase64Function
+ * description: Returns the base64-decoded result from string; returns NULL if string is NULL
+ */
+@TransformFunction(names = {"from_base64"})
+public class FromBase64Function implements ValueParser {
+
+    private final ValueParser stringParser;
+
+    public FromBase64Function(Function expr) {
+        List<Expression> expressions = expr.getParameters().getExpressions();
+        stringParser = OperatorTools.buildParser(expressions.get(0));
+    }
+
+    @Override
+    public Object parse(SourceData sourceData, int rowIndex, Context context) {
+        Object stringObj = stringParser.parse(sourceData, rowIndex, context);
+        if (stringObj == null) {
+            return null;
+        }
+        String encodedString = OperatorTools.parseString(stringObj);
+
+        if (encodedString == null) {
+            return null;
+        }
+
+        try {
+            byte[] decodedBytes = Base64.getDecoder().decode(encodedString);
+            return new String(decodedBytes, StandardCharsets.UTF_8);
+        } catch (IllegalArgumentException e) {
+            // handle decoding exceptions and log exception information
+            throw new RuntimeException("Invalid Base64 input: " + encodedString, e);
+        }
+    }
+}