[INLONG-11243][SDK] Enhance the functionality of the substring functi…

…on in Transform SQL (apache#11277)

Co-authored-by: ZKpLo <[email protected]>

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

@@ -24,13 +24,14 @@
 
 import net.sf.jsqlparser.expression.Expression;
 import net.sf.jsqlparser.expression.Function;
+import net.sf.jsqlparser.expression.operators.relational.ExpressionList;
 
 import java.util.List;
 
 /**
- * SubstringFunction
- * description: substring(string FROM INT1 [ FOR INT2 ])--returns a substring of STRING starting from position INT1 with
- * length INT2 (to the end by default)
+ * SubstringFunction -> substring(string FROM INT1 [ FOR INT2 ])
+ * description:
+ * return a substring of STRING starting from position INT1 with length INT2 (to the end by default)
  */
 @TransformFunction(names = {"substring", "substr"})
 public class SubstringFunction implements ValueParser {
@@ -39,28 +40,21 @@ public class SubstringFunction implements ValueParser {
     private ValueParser startPositionParser;
     private ValueParser lengthParser;
 
-    /**
-     * Constructor
-     *
-     * @param expr
-     */
     public SubstringFunction(Function expr) {
-        List<Expression> expressions = expr.getParameters().getExpressions();
-        // Determine the number of arguments and build parser
+        ExpressionList parameters = expr.getParameters();
+        List<Expression> expressions;
+        if (parameters != null) {
+            expressions = parameters.getExpressions();
+        } else {
+            expressions = expr.getNamedParameters().getExpressions();
+        }
         stringParser = OperatorTools.buildParser(expressions.get(0));
         startPositionParser = OperatorTools.buildParser(expressions.get(1));
         if (expressions.size() == 3) {
             lengthParser = OperatorTools.buildParser(expressions.get(2));
         }
     }
 
-    /**
-     * parse
-     *
-     * @param sourceData
-     * @param rowIndex
-     * @return
-     */
     @Override
     public Object parse(SourceData sourceData, int rowIndex, Context context) {
         Object stringObj = stringParser.parse(sourceData, rowIndex, context);

inlong-sdk/transform-sdk/src/test/java/org/apache/inlong/sdk/transform/process/function/string/TestSubstringFunction.java

@@ -32,6 +32,11 @@ public class TestSubstringFunction extends AbstractFunctionStringTestBase {
 
     @Test
     public void testSubstringFunction() throws Exception {
+        String transformSql = null, data = null;
+        TransformConfig config = null;
+        TransformProcessor<String, String> processor = null;
+        List<String> output = null;
+
         String transformSql1 = "select substring(string2, numeric1) from source";
         TransformConfig config1 = new TransformConfig(transformSql1);
         TransformProcessor<String, String> processor1 = TransformProcessor
@@ -41,6 +46,7 @@ public void testSubstringFunction() throws Exception {
         List<String> output1 = processor1.transform("apple|banana|cloud|2|1|3", new HashMap<>());
         Assert.assertEquals(1, output1.size());
         Assert.assertEquals(output1.get(0), "result=anana");
+
         String transformSql2 = "select substring(string1, numeric1, numeric3) from source";
         TransformConfig config2 = new TransformConfig(transformSql2);
         TransformProcessor<String, String> processor2 = TransformProcessor
@@ -54,5 +60,38 @@ public void testSubstringFunction() throws Exception {
         List<String> output3 = processor2.transform("apple|banana|cloud|2|1|9", new HashMap<>());
         Assert.assertEquals(1, output3.size());
         Assert.assertEquals(output3.get(0), "result=pple");
+
+        transformSql = "select substring(string1 from numeric1) from source";
+        config = new TransformConfig(transformSql);
+        processor = TransformProcessor
+                .create(config, SourceDecoderFactory.createCsvDecoder(csvSource),
+                        SinkEncoderFactory.createKvEncoder(kvSink));
+        // case4: substring('hello world' from 7)
+        data = "hello world|||7|3|3";
+        output = processor.transform(data, new HashMap<>());
+        Assert.assertEquals(1, output.size());
+        Assert.assertEquals("result=world", output.get(0));
+
+        transformSql = "select substring(string1 from numeric1 for numeric2) from source";
+        config = new TransformConfig(transformSql);
+        processor = TransformProcessor
+                .create(config, SourceDecoderFactory.createCsvDecoder(csvSource),
+                        SinkEncoderFactory.createKvEncoder(kvSink));
+        // case5: substring('hello world' from 7 for 3)
+        data = "hello world|||7|3|3";
+        output = processor.transform(data, new HashMap<>());
+        Assert.assertEquals(1, output.size());
+        Assert.assertEquals("result=wor", output.get(0));
+
+        transformSql = "select substring(string1 from numericx for numericx) from source";
+        config = new TransformConfig(transformSql);
+        processor = TransformProcessor
+                .create(config, SourceDecoderFactory.createCsvDecoder(csvSource),
+                        SinkEncoderFactory.createKvEncoder(kvSink));
+        // case6: substring('hello world' from null for null)
+        data = "hello world|||||";
+        output = processor.transform(data, new HashMap<>());
+        Assert.assertEquals(1, output.size());
+        Assert.assertEquals("result=", output.get(0));
     }
 }