@Test
    public void testFullOuterJoinWithDistinctAndFilter() throws Exception {
        EnvironmentSettings settings = EnvironmentSettings
                .newInstance()
                .inStreamingMode()
                .build();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(10000);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
        Node inputNode = buildKafkaExtractNode();
        Node inputNode2 = buildKafkaExtractNode2();
        Node inputNode3 = buildKafkaExtractNode3();
        Node tansformNode = buildDistinctNode();
        Node outputNode = buildKafkaLoadNode();
        StreamInfo streamInfo = new StreamInfo("1",
                Arrays.asList(inputNode, inputNode2, inputNode3, tansformNode, outputNode),
                Arrays.asList(
                        buildFullOuterJoinRelation(Arrays.asList(inputNode, inputNode2, inputNode3),
                                Collections.singletonList(tansformNode)),
                        buildNodeRelation(Collections.singletonList(tansformNode),
                                Collections.singletonList(outputNode))));
        GroupInfo groupInfo = new GroupInfo("group_id", Collections.singletonList(streamInfo));
        FlinkSqlParser parser = FlinkSqlParser.getInstance(tableEnv, groupInfo);
        ParseResult result = parser.parse();
        Assert.assertTrue(result.tryExecute());
    }
Test full outer join with distinct and filter
The same as {@link #testFullOuterJoin()} ,
it's just the difference that the transform node is {@link DistinctNode} and have filters

@throws Exception The exception may throws when executing
@see #testFullOuterJoin()
inlong-sort/sort-core/src/test/java/org/apache/inlong/sort/parser/FullOuterJoinSqlParseTest.java