[INLONG-11132][SDK] Transform SQL support parsing SIMILAR TO

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

@@ -0,0 +1,107 @@
+/*
+ * 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.common.util.StringUtil;
+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 lombok.extern.slf4j.Slf4j;
+import net.sf.jsqlparser.expression.operators.relational.SimilarToExpression;
+
+import java.util.regex.Pattern;
+/**
+ * SimilarToParser
+ */
+@Slf4j
+@TransformParser(values = SimilarToExpression.class)
+public class SimilarToParser implements ValueParser {
+
+    private final ValueParser destParser;
+    private final ValueParser patternParser;
+    private final String escapeChar;
+    private final boolean isNot;
+    private static final String REGEX_SPECIAL_CHAR = "[]()|^-+*?{}$\\.";
+
+    public SimilarToParser(SimilarToExpression expr) {
+        destParser = OperatorTools.buildParser(expr.getLeftExpression());
+        patternParser = OperatorTools.buildParser(expr.getRightExpression());
+        escapeChar = StringUtil.isEmpty(expr.getEscape()) ? "\\" : expr.getEscape();
+        isNot = expr.isNot();
+    }
+
+    @Override
+    public Object parse(SourceData sourceData, int rowIndex, Context context) {
+        Object destObj = destParser.parse(sourceData, rowIndex, context);
+        Object patternObj = patternParser.parse(sourceData, rowIndex, context);
+        if (destObj == null || patternObj == null) {
+            return null;
+        }
+        String destStr = destObj.toString();
+        String pattern = patternObj.toString();
+        try {
+            final String regex = buildSimilarToRegex(pattern, escapeChar.charAt(0));
+            boolean isMatch = Pattern.matches(regex.toLowerCase(), destStr.toLowerCase());
+            if (isNot) {
+                return !isMatch;
+            }
+            return isMatch;
+        } catch (Exception e) {
+            log.error(e.getMessage(), e);
+            return null;
+        }
+    }
+
+    private String buildSimilarToRegex(String pattern, char escapeChar) {
+        int len = pattern.length();
+        StringBuilder regexPattern = new StringBuilder(len + len);
+        for (int i = 0; i < len; i++) {
+            char c = pattern.charAt(i);
+            if (REGEX_SPECIAL_CHAR.indexOf(c) >= 0) {
+                regexPattern.append('\\');
+            }
+            if (c == escapeChar) {
+                if (i == (pattern.length() - 1)) {
+                    regexPattern.append(c);
+                    continue;
+                }
+                char nextChar = pattern.charAt(i + 1);
+                if (nextChar == '_' || nextChar == '%' || nextChar == escapeChar) {
+                    regexPattern.append(nextChar);
+                    i++;
+                } else {
+                    throw new RuntimeException("Illegal pattern string");
+                }
+            } else if (c == '_') {
+                regexPattern.append('.');
+            } else if (c == '%') {
+                regexPattern.append(".*");
+            } else if (c == '[') {
+                regexPattern.append('[');
+                while (i < len && pattern.charAt(i) != ']') {
+                    i++;
+                    regexPattern.append(pattern.charAt(i));
+                }
+            } else {
+                regexPattern.append(c);
+            }
+        }
+        return regexPattern.toString();
+    }
+}

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

@@ -0,0 +1,165 @@
+/*
+ * 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.SourceDecoderFactory;
+import org.apache.inlong.sdk.transform.encode.SinkEncoderFactory;
+import org.apache.inlong.sdk.transform.pojo.CsvSourceInfo;
+import org.apache.inlong.sdk.transform.pojo.FieldInfo;
+import org.apache.inlong.sdk.transform.pojo.KvSinkInfo;
+import org.apache.inlong.sdk.transform.pojo.TransformConfig;
+import org.apache.inlong.sdk.transform.process.TransformProcessor;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+
+public class TestSimilarToParser extends AbstractParserTestBase {
+
+    private static final List<FieldInfo> srcFields = new ArrayList<>();
+    private static final List<FieldInfo> dstFields = new ArrayList<>();
+    private static final CsvSourceInfo csvSource;
+    private static final KvSinkInfo kvSink;
+
+    static {
+        for (int i = 1; i < 3; i++) {
+            FieldInfo field = new FieldInfo();
+            field.setName("string" + i);
+            srcFields.add(field);
+        }
+        FieldInfo field = new FieldInfo();
+        field.setName("result");
+        dstFields.add(field);
+        csvSource = new CsvSourceInfo("UTF-8", '|', '\\', srcFields);
+        kvSink = new KvSinkInfo("UTF-8", dstFields);
+    }
+
+    @Test
+    public void testSimilarToFunction() throws Exception {
+        String transformSql = null, data = null;
+        TransformConfig config = null;
+        TransformProcessor<String, String> processor = null;
+        List<String> output = null;
+
+        transformSql = "select string1 similar to string2 from source";
+        config = new TransformConfig(transformSql);
+        processor = TransformProcessor
+                .create(config, SourceDecoderFactory.createCsvDecoder(csvSource),
+                        SinkEncoderFactory.createKvEncoder(kvSink));
+        // case1: apple similar to %App%
+        output = processor.transform("apple|%App%", new HashMap<>());
+        Assert.assertEquals(1, output.size());
+        Assert.assertEquals("result=true", output.get(0));
+
+        // case2: apple similar to /%App%
+        // The reason why '\' is not used as an escape string here is that when processing CSV data,
+        // the quote parameter defaults to the '\' character
+        transformSql = "select string1 similar to string2 ESCAPE '/' from source";
+        config = new TransformConfig(transformSql);
+        processor = TransformProcessor
+                .create(config, SourceDecoderFactory.createCsvDecoder(csvSource),
+                        SinkEncoderFactory.createKvEncoder(kvSink));
+
+        output = processor.transform("apple|/%App%", new HashMap<>());
+        Assert.assertEquals(1, output.size());
+        Assert.assertEquals("result=false", output.get(0));
+
+        // case3: %apple similar to /%App% ESCAPE '/'
+        output = processor.transform("%apple|/%App%", new HashMap<>());
+        Assert.assertEquals(1, output.size());
+        Assert.assertEquals("result=true", output.get(0));
+
+        // case4: %apple similar to /%Apple_ ESCAPE '/'
+        output = processor.transform("%apple|/%Apple_", new HashMap<>());
+        Assert.assertEquals(1, output.size());
+        Assert.assertEquals("result=false", output.get(0));
+
+        // case5: %apple similar to /%Appl_ ESCAPE '/'
+        output = processor.transform("%apple|/%Appl_", new HashMap<>());
+        Assert.assertEquals(1, output.size());
+        Assert.assertEquals("result=true", output.get(0));
+
+        // case6: %ap_ple similar to /%Ap%_e ESCAPE '/'
+        output = processor.transform("%ap_ple|/%Ap%_e", new HashMap<>());
+        Assert.assertEquals(1, output.size());
+        Assert.assertEquals("result=true", output.get(0));
+
+        // case7: %ap_ple/ similar to /%Ap%_e/ ESCAPE '/'
+        output = processor.transform("%ap_ple/|/%Ap%_e/", new HashMap<>());
+        Assert.assertEquals(1, output.size());
+        Assert.assertEquals("result=true", output.get(0));
+    }
+    @Test
+    public void testNotSimilarToFunction() throws Exception {
+        String transformSql = null, data = null;
+        TransformConfig config = null;
+        TransformProcessor<String, String> processor = null;
+        List<String> output = null;
+
+        transformSql = "select string1 not similar to string2 from source";
+        config = new TransformConfig(transformSql);
+        processor = TransformProcessor
+                .create(config, SourceDecoderFactory.createCsvDecoder(csvSource),
+                        SinkEncoderFactory.createKvEncoder(kvSink));
+        // case1: apple not similar to %App%
+        output = processor.transform("apple|%App%", new HashMap<>());
+        Assert.assertEquals(1, output.size());
+        Assert.assertEquals("result=false", output.get(0));
+
+        // case2: apple not similar to /%App%
+        // The reason why '\' is not used as an escape string here is that when processing CSV data,
+        // the quote parameter defaults to the '\' character
+        transformSql = "select string1 not similar to string2 ESCAPE '/' from source";
+        config = new TransformConfig(transformSql);
+        processor = TransformProcessor
+                .create(config, SourceDecoderFactory.createCsvDecoder(csvSource),
+                        SinkEncoderFactory.createKvEncoder(kvSink));
+
+        output = processor.transform("apple|/%App%", new HashMap<>());
+        Assert.assertEquals(1, output.size());
+        Assert.assertEquals("result=true", output.get(0));
+
+        // case3: %apple not similar to /%App% ESCAPE '/'
+        output = processor.transform("%apple|/%App%", new HashMap<>());
+        Assert.assertEquals(1, output.size());
+        Assert.assertEquals("result=false", output.get(0));
+
+        // case4: %apple not similar to /%Apple_ ESCAPE '/'
+        output = processor.transform("%apple|/%Apple_", new HashMap<>());
+        Assert.assertEquals(1, output.size());
+        Assert.assertEquals("result=true", output.get(0));
+
+        // case5: %apple not similar to /%Appl_ ESCAPE '/'
+        output = processor.transform("%apple|/%Appl_", new HashMap<>());
+        Assert.assertEquals(1, output.size());
+        Assert.assertEquals("result=false", output.get(0));
+
+        // case6: %ap_ple not similar to /%Ap%_e ESCAPE '/'
+        output = processor.transform("%ap_ple|/%Ap%_e", new HashMap<>());
+        Assert.assertEquals(1, output.size());
+        Assert.assertEquals("result=false", output.get(0));
+
+        // case7: %ap_ple/ not similar to /%Ap%_e/ ESCAPE '/'
+        output = processor.transform("%ap_ple/|/%Ap%_e/", new HashMap<>());
+        Assert.assertEquals(1, output.size());
+        Assert.assertEquals("result=false", output.get(0));
+    }
+}