Apache Flink - 启用连接排序

我注意到Apache Flink没有优化表连接的顺序。目前，它保留了用户指定的连接顺序（基本上，它按字面顺序进行查询）。我想Apache Calcite可以优化连接的顺序，但由于某些原因，这些规则在Apache Flink中没有使用。

例如，如果我们有两个表' R '和' S '

private val tableEnv: BatchTableEnvironment = TableEnvironment.getTableEnvironment(env)
private val fileNumber = 1
tableEnv.registerTableSource("R", getDataSourceR(fileNumber))
tableEnv.registerTableSource("S", getDataSourceS(fileNumber))
private val r = tableEnv.scan("R")
private val s = tableEnv.scan("S")
我们假设' S '是空的，我们希望以两种方式加入这些表：

val tableOne = r.as("x1, x2").join(r.as("x3, x4")).where("x2 === x3").select("x1, x4")

    .join(s.as("x5, x6")).where("x4 === x5 ").select("x1, x6")

val tableTwo = s.as("x1, x2").join(r.as("x3, x4")).where("x2 === x3").select("x1, x4")

      .join(r.as("x5, x6")).where("x4 === x5 ").select("x1, x6")

如果我们想要计算tableOne和tableTwo中的行数，则两种情况下的结果都将为零。问题是评估tableOne所需的时间比评估tableTwo要长得多。

有没有什么办法可以自动优化连接的执行顺序，甚至通过添加一些统计数据来启用可能的计划成本操作？如何添加这些统计数据？

可能有必要更改表环境CalciteConfig，但我不清楚如何做到这一点。