@Test
    public void testRegexpReplace() throws Exception {
        // step 0. Initialize the execution environment
        EnvironmentSettings settings = EnvironmentSettings
                .newInstance()
                .inStreamingMode()
                .build();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(10000);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
        tableEnv.createTemporaryFunction("REGEXP_REPLACE", RegexpReplaceFunction.class);
        // step 1. Generate test data and convert to DataStream
        List<Row> data = new ArrayList<>();
        data.add(Row.of("111222333,111222333,111222333"));
        TypeInformation<?>[] types = {
                BasicTypeInfo.STRING_TYPE_INFO};
        String[] names = {"f1"};
        RowTypeInfo typeInfo = new RowTypeInfo(types, names);
        DataStream<Row> dataStream = env.fromCollection(data).returns(typeInfo);
        // step 2. Convert from DataStream to Table and execute the REGEXP_REPLACE function
        org.apache.inlong.sort.protocol.transformation.function.RegexpReplaceFunction regexpReplaceFunction =
                new org.apache.inlong.sort.protocol.transformation.function.RegexpReplaceFunction(
                        new FieldInfo("f1", new StringFormatInfo()),
                        new StringConstantParam("(\\d{3})\\d*(\\d{3})"),
                        new StringConstantParam("$1***$2"));
        Table tempView = tableEnv.fromDataStream(dataStream).as("f1");
        tableEnv.createTemporaryView("temp_view", tempView);
        String sqlQuery = String.format("SELECT %s as f1 FROM temp_view", regexpReplaceFunction.format());
        Table outputTable = tableEnv.sqlQuery(sqlQuery);
        // step 3. Get function execution result and parse it
        DataStream<Row> resultSet = tableEnv.toAppendStream(outputTable, Row.class);
        List<String> result = new ArrayList<>();
        for (CloseableIterator<String> it = resultSet.map(s -> s.getField(0).toString()).executeAndCollect(); it
                .hasNext();) {
            String next = it.next();
            result.add(next);
        }
        // step 4. Whether the comparison results are as expected
        String expect = "111***333,111***333,111***333";
        Assert.assertEquals(expect, result.get(0));
    }
Test RegexpReplace

@throws Exception The exception may throw when test RegexpReplace
inlong-sort/sort-core/src/test/java/org/apache/inlong/sort/function/RegexpReplaceFunctionTest.java