StreamEdges are not unique in self-union, which blocks propagation of watermarks

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        Table inputTable =
                tEnv.fromValues(
                        DataTypes.ROW(
                                DataTypes.FIELD("weight", DataTypes.DOUBLE()),
                                DataTypes.FIELD("f0", DataTypes.STRING()),
                                DataTypes.FIELD("f1", DataTypes.DOUBLE()),
                                DataTypes.FIELD("f2", DataTypes.DOUBLE()),
                                DataTypes.FIELD("f3", DataTypes.DOUBLE()),
                                DataTypes.FIELD("f4", DataTypes.INT()),
                                DataTypes.FIELD("label", DataTypes.STRING())),
                        Row.of(1., "a", 1., 1., 1., 2, "l1"),
                        Row.of(1., "a", 1., 1., 1., 2, "l1"));

        DataStream<Row> input = tEnv.toDataStream(inputTable);