Apache Flink - 启用连接排序答案

【问题标题】：Apache Flink - enable join orderingApache Flink - 启用连接排序
【发布时间】：2019-05-05 04:46:39
【问题描述】：

我注意到 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 花费更长的时间。

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

在link 的文档中写道，也许有必要更改 Table 环境 CalciteConfig，但我不清楚该怎么做。

请帮忙。

【问题讨论】：

标签： apache-flink sql-execution-plan flink-sql

【解决方案1】：

由于 Flink 不能很好地处理统计信息，因此未启用 Join reordering。在没有准确估计基数的情况下重新排序连接基本上是赌博。因此，连接重新排序被禁用，表按照用户提供的顺序连接。这给出了确定性和可控的行为。

但是，您可以通过在创建TableEnvironment 时将TableConfig 和CalciteConfig 传递给优化器，即TableEnvironment.getTableEnvironment(env, yourTableConfig)。在CalciteConfig 中，您可以将优化规则添加到不同的优化阶段。您可能希望将JoinCommuteRule 和JoinAssociateRule 添加到逻辑优化阶段。您可能还必须深入研究代码以检查如何将统计信息传递给优化器。

【讨论】：