@Test
    public void testCascadeFunctionWrapper() 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);
        // step 1. Register custom function of REGEXP_REPLACE_FIRST
        tableEnv.createTemporarySystemFunction("REGEXP_REPLACE_FIRST",
                RegexpReplaceFirstFunction.class);
        // step 2. Generate test data and convert to DataStream
        List<Row> data = new ArrayList<>();
        data.add(Row.of("aabbccaabbcc"));
        TypeInformation<?>[] types = {
                BasicTypeInfo.STRING_TYPE_INFO};
        String[] names = {"f1"};
        RowTypeInfo typeInfo = new RowTypeInfo(types, names);
        List<CascadeFunction> functions = new ArrayList<>();
        functions.add(new org.apache.inlong.sort.protocol.transformation.function.RegexpReplaceFirstFunction(
                new FieldInfo("f1", new StringFormatInfo()),
                new StringConstantParam("aa"), new StringConstantParam("apache")));
        functions.add(new RegexpReplaceFunction(
                new FieldInfo("f1", new StringFormatInfo()),
                new StringConstantParam("bb"), new StringConstantParam("inlong")));
        functions.add(new RegexpReplaceFunction(
                new FieldInfo("f1", new StringFormatInfo()),
                new StringConstantParam("cc"), new StringConstantParam("etl")));
        CascadeFunctionWrapper wrapper = new CascadeFunctionWrapper(functions);
        DataStream<Row> ds = env.fromCollection(data).returns(typeInfo);
        // step 3. Convert from DataStream to Table and execute the REGEXP_REPLACE_FIRST function
        Table tempView = tableEnv.fromDataStream(ds).as("f1");
        tableEnv.createTemporaryView("temp_view", tempView);
        String sqlQuery = String.format("SELECT %s as a FROM temp_view", wrapper.format());
        Table outputTable = tableEnv.sqlQuery(sqlQuery);
        DataStream<Row> resultSet = tableEnv.toAppendStream(outputTable, Row.class);
        // step 4. Get function execution result and parse it
        List<String> r = new ArrayList<>();
        for (CloseableIterator<String> it = resultSet.map(s -> s.getField(0).toString()).executeAndCollect(); it
                .hasNext();) {
            String next = it.next();
            r.add(next);
        }
        // step 5. Whether the comparison results are as expected
        String expect = "apacheinlongetlaainlongetl";
        Assert.assertEquals(expect, r.get(0));
    }
Test for CascadeFunctionWrapper

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