Apache Calcite中的自定义特征

Apache Calcite带有约定和排序规则（排序顺序）属性。 我们探索如何在Apache Calcite中定义和实施自定义物理属性（特征）。

概述

物理属性是优化过程的重要组成部分，可让您探索更多替代方案。

Apache Calcite带有约定和排序规则（排序顺序）属性。 许多查询引擎需要自定义属性。 例如，我们在日常实践中经常看到的分布式和异构引擎需要仔细计划机器和设备之间的数据移动，这需要自定义属性来描述数据位置。

在此博客文章中，我们将探讨如何使用Apache Calcite基于成本的优化器来定义，注册和实施自定义属性（也称为特征）。

物理性质

我们通过查看常见物理属性的示例（排序顺序）开始我们的旅程。

查询优化器与关系运算符（如“扫描”，“项目”，“过滤器”和“联接”）一起使用。 在优化过程中，操作员可能需要其输入才能满足特定条件。 为了检查条件是否满足，操作员可以公开物理属性-与操作员关联的纯值。 操作员可以比较其输入的期望属性和实际属性，并通过在输入之上注入特殊的强制执行器来强制执行期望的属性。

考虑联接运算符t1 JOIN t2 ON t1.a =t2.b。 如果两个输入都分别在其连接属性t1.a和t2.b上排序，则可以使用合并连接。 我们可以为每个运算符定义排序规则属性，以描述产生的行的排序顺序：

Join[t1.a=t2.b]
  Input[t1]      [SORTED by a]
  Input[t2]      [NOT SORTED]

合并联接运算符可以在其输入上对t1.a和t2.b进行排序。 由于第一个输入已经在t1.a上排序，因此它保持不变。 第二个输入未排序，因此注入了强制执行器Sort运算符，从而使合并联接成为可能：

MergeJoin[t1.a=t2.b]  
  Input[t1]           [SORTED by t1.a]
  Sort[t2.a]          [SORTED by t2.b]
    Input[t2]         [NOT SORTED]

Apache Calcite API

在Apache Calcite中，属性由RelTrait和RelTraitDef类定义。 RelTrait是该属性的具体值。 RelTraitDef是一个属性定义，它描述属性名称，该属性的预期Java类，该属性的默认值以及如何执行该属性。 属性定义通过RelOptPlanner.addRelTraitDef方法注册到计划程序中。 计划者将确保每个操作员对于每个已注册的属性定义都具有特定的值，无论是否为默认值。

节点的所有属性都组织在一个不变的数据结构RelTraitSet中。 此类具有使用复制语义添加和更新属性的便捷方法。 您可以使用RelOptNode.getTraitSet方法访问具体运算符的属性。

要在计划过程中对操作员实施特定属性，您应该在规则内执行以下操作：

• 使用RelOptNode.getTraitSet方法获取节点的当前属性。
• 使用更新的属性创建RelTraitSet的新实例。
• 通过调用RelOptRule.convert方法来增强属性。

最后，在调用计划程序之前，您可以定义优化关系树的根运算符的所需属性。 优化之后，计划者将返回满足这些属性的运算符，或者引发异常。

在内部，Apache Calcite通过在目标运算符的顶部添加一个具有所需特征的特殊AbstractConverter运算符来强制执行属性。

AbstractConverter [SORTED by a]
  Input[t2]       [NOT SORTED]

要将AbstractConverter转换为实际的执行器节点（例如Sort），应将内置的ExpandConversionRule规则添加到优化程序中。 此规则将尝试将AbstractConverter扩展为一系列强制执行器，以满足与我们已经讨论过的特征定义相关的所需特征。 在我们的示例中，我们只有一个不满足的属性，因此转换器将扩展为单个Sort运算符。

Sort[t2.a]        [SORTED by a]
  Input[t2]       [NOT SORTED]

自定义属性

当我们了解属性的用途以及要使用的Apache Calcite API时，我们将定义，注册和实施我们的自定义属性。

考虑我们有一个分布式数据库，其中每个关系运算符都可以通过以下两种方式之一在节点之间分布：

• 已分区—关系在节点之间分区。 每个元组（行）都驻留在一个节点上。 一个示例是典型的分布式数据结构。
• SINGLETON —关系位于单个节点上。 一个示例是将最终结果传递给用户应用程序的游标。

在我们的示例中，我们希望确保top运算符始终具有SINGLETON分布，以模拟结果向单个节点的传递。

Enforcer

首先，我们定义强制运算符。 为了确保SINGLETON分布，我们需要从所有节点移动到单个节点。 在分布式数据库中，数据移动运算符通常称为Exchange。 Apache Calcite中对自定义运算符的最低要求是定义构造函数和copy方法。

public class ExchangeRel extends SingleRel {
    public RedistributeRel(
        RelOptCluster cluster,
        RelTraitSet traits,
        RelNode input
    ) {
        super(cluster, traits, input);
    }

    @Override
    public RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) {
        return new ExchangeRel(getCluster(), traitSet, inputs.get(0));
    }
}

Trait

接下来，我们定义自定义特征和特征定义。 我们的实现必须遵守以下规则：

• 特征必须在方法getTraitDef中引用公共特征定义实例。
• 该特征必须重写satisfies方法，以定义当前特征是否满足目标特征。 如果没有，将使用执行器。
• 特征定义必须在getTraitClass方法中声明特征的预期Java类。
• 特征定义必须在getDefault方法中声明特征的默认值。
• 特征定义必须实现方法convert，如果当前特征不满足所需特征，Apache Calcite将调用该方法来创建执行程序。 如果特征之间没有有效的转换，则应返回null。

下面是我们特质的源代码。 我们定义两个具体的值，PARTITIONED和SINGLETON。 我们还定义了特殊值ANY，我们将其用作默认值。 我们说PARTITIONED和SINGLETON都满足任何条件，但PARTITIONED和SINGLETON彼此不满足。

public class Distribution implements RelTrait {

    public static final Distribution ANY = new Distribution(Type.ANY);
    public static final Distribution PARTITIONED = new Distribution(Type.PARTITIONED);
    public static final Distribution SINGLETON = new Distribution(Type.SINGLETON);

    private final Type type;

    private Distribution(Type type) {
        this.type = type;
    }

    @Override
    public RelTraitDef getTraitDef() {
        return DistributionTraitDef.INSTANCE;
    }

    @Override
    public boolean satisfies(RelTrait toTrait) {
        Distribution toTrait0 = (Distribution) toTrait;

        if (toTrait0.type == Type.ANY) {
            return true;
        }

        return this.type.equals(toTrait0.type);
    }

    enum Type {
        ANY,
        PARTITIONED,
        SINGLETON
    }
}

我们的特征定义定义了convert函数，如果当前属性不满足目标属性，则该函数将注入ExchangeRel强制程序。

public class DistributionTraitDef extends RelTraitDef<Distribution> {

    public static DistributionTraitDef INSTANCE = new DistributionTraitDef();

    private DistributionTraitDef() {
        // No-op.
    }

    @Override
    public Class<Distribution> getTraitClass() {
        return Distribution.class;
    }

    @Override
    public String getSimpleName() {
        return "DISTRIBUTION";
    }

    @Override
    public RelNode convert(
        RelOptPlanner planner,
        RelNode rel,
        Distribution toTrait,
        boolean allowInfiniteCostConverters
    ) {
        Distribution fromTrait = rel.getTraitSet().getTrait(DistributionTraitDef.INSTANCE);

        if (fromTrait.satisfies(toTrait)) {
            return rel;
        }

        return new ExchangeRel(
            rel.getCluster(),
            rel.getTraitSet().plus(toTrait),
            rel
        );
    }

    @Override
    public boolean canConvert(
        RelOptPlanner planner,
        Distribution fromTrait,
        Distribution toTrait
    ) {
        return true;
    }

    @Override
    public Distribution getDefault() {
        return Distribution.ANY;
    }
}

在生产实现中，您可能会有更多的分发类型，专用的分发列以及不同的交换类型。 您可以参考Apache Flink作为真实分发特征的示例。

结合在一起

首先，我们创建一个具有两个表的模式-一个表具有PARTITIONED分布，另一个表具有SINGLETON分布。 我们使用自定义表和架构实现，类似于上一篇博文中使用的实现。

// Table with PARTITIONED distribution.
Table table1 = Table.newBuilder("table1", Distribution.PARTITIONED)
  .addField("field", SqlTypeName.DECIMAL).build();

// Table with SINGLETON distribution.
Table table2 = Table.newBuilder("table2", Distribution.SINGLETON)
  .addField("field", SqlTypeName.DECIMAL).build();

Schema schema = Schema.newBuilder("schema").addTable(table1).addTable(table2).build();

然后，我们创建一个计划器实例，并在其中注册我们的自定义特征定义。

VolcanoPlanner planner = new VolcanoPlanner();

planner.addRelTraitDef(ConventionTraitDef.INSTANCE);
planner.addRelTraitDef(DistributionTraitDef.INSTANCE);

最后，我们为每个表创建一个表扫描运算符，并执行SINGLETON分布。 请注意，我们在优化程序中使用了前面提到的ExpandConversionRule。 否则，强制执行将无法进行。

// Use the built-in rule that will expand abstract converters.
RuleSet rules = RuleSets.ofList(AbstractConverter.ExpandConversionRule.INSTANCE);

// Prepare the desired traits with the SINGLETON distribution.
RelTraitSet desiredTraits = node.getTraitSet().plus(Distribution.SINGLETON);

// Use the planner to enforce the desired traits
RelNode optimizedNode = Programs.of(rules).run(
    planner,
    node,
    desiredTraits,
    Collections.emptyList(),
    Collections.emptyList()
);

现在，我们从示例项目中运行TraitTest，以查看实际效果。 对于PARTITIONED表，计划者已添加ExchangeRel以强制执行SINGLETON分发。

BEFORE:
2:LogicalTableScan(table=[[schema, partitioned]])

AFTER:
7:ExchangeRel
  2:LogicalTableScan(table=[[schema, partitioned]])

但是具有SINGLETON分布的表保持不变，因为它已经具有所需的分布。

BEFORE:
0:LogicalTableScan(table=[[schema, singleton]])

AFTER:
0:LogicalTableScan(table=[[schema, singleton]])

总结

物理属性是查询优化中的重要概念，可让您探索更多替代计划。

在此博客文章中，我们演示了如何在Apache Calcite中定义自定义物理属性。 我们创建了一个自定义的RelTraitDef和RelTrait类，并在计划程序中注册了它们，并使用该自定义运算符强制实施了所需的属性值。 在以后的文章中，我们将深入探讨Apache Calcite的各个组件。 敬请关注！

但是，我们省略了一个关键问题-如何在运算符之间传播属性？ 事实证明，Apache Calcite不能很好地做到这一点，您将不得不在几个非理想的解决方案之间做出艰难的抉择。