做 Calcite 查询优化的人绕不开一堆以Rule结尾的类。今天聊的AggregateFilterToCaseRule是我压箱底的一条自定义规则它解决的问题很具体——当 HAVING 条件残留在聚合节点上方时怎么把过滤语义安全地折叠进聚合函数里的 CASE WHEN 表达式。网上讲 Calcite 自定义规则的教程不少但大多数只讲“怎么写一个 Rule”很少讲“为什么这样写、为什么有些情况不能这样写”。这篇就用一个可运行的规则实现把这两件事都说清楚。适合已经在 Calcite 上跑过 SQL、想真正写自定义可优化规则的开发者。1. 这个规则试图解决的核心问题Filter 压在 Aggregate 头上1.1 一条 HAVING 产生的执行计划长什么样先看一条常见到不能再常见的 SQLSELECT dept_id, MAX(price) AS max_price FROM tx GROUP BY dept_id HAVING MAX(price) 1000;很多人在写业务 SQL 时不会意识到HAVING在 Calcite 里的翻译路径和WHERE完全不同。SqlToRelConverter会把HAVING单独拆成一个LogicalFilter节点直接压在LogicalAggregate上面。逻辑计划长这样LogicalFilter(condition[$1 1000]) LogicalAggregate(group[{0}], max_price[MAX($1)], agg#1[SUM($2)]) LogicalProject(dept_id[$0], price[$1], amount[$2]) LogicalTableScan(table[[tx]])这里的$1不是原始表里的 price 列而是聚合输出列MAX(price)的结果。注意LogicalAggregate的输出字段顺序先是 group 字段然后才是聚合字段所以$0是 dept_id$1是 MAX(price)$2是 SUM(amount)。另一种常见来源是子查询SELECT dept_id, max_price FROM ( SELECT dept_id, MAX(price) AS max_price FROM tx GROUP BY dept_id ) WHERE max_price 1000;子查询被展开subquery rewrite之后同样会形成Filter(Aggregate(...))的树形结构。也就是说只要过滤条件引用了聚合结果Calcite 的默认行为就是把这个条件留在聚合节点外面。1.2 分组语义为什么挡了下推的路如果是WHERE条件比如price 1000我们可以很放心地把过滤下推到聚合之前因为单行的条件不会影响分组聚合的语义。但 HAVING 条件引用了聚合结果比如MAX(price) 1000这就不能用简单的谓词下推处理了。举个例子某组有两行price 分别为 800 和 1200。MAX(price)是 1200所以这一组会被保留。可如果我们贸然把MAX(price) 1000转成price 1000并下推到聚合之前那 800 这一行在聚合前就被丢掉了剩下这一组的 MAX 就变成 1200虽然结果碰巧一样但语义依赖运气。如果组内只有 price800聚合前下推会导致这组消失而正确结果应该是该组存在、只是不满足 HAVING 条件被丢弃。所以在当前模式下过滤条件只能待在聚合节点上面。这就是AggregateFilterToCaseRule需要存在的第一个理由有一种过滤条件是“聚合结果与常数的比较”它在语义上无法被传统谓词下推处理但可以被重新表达成聚合内部的 CASE WHEN 计算。1.3 目标形态把谓词折叠成 CASE WHEN我们希望达成的目标是把原来的计划改写为LogicalFilter(condition[$1 IS NOT NULL]) LogicalAggregate(group[{0}], max_price[MAX(CASE WHEN $1 1000 THEN $1 END)], agg#1[SUM($2)]) LogicalProject(dept_id[$0], price[$1], amount[$2]) LogicalTableScan(table[[tx]])表面上看Filter 还在只不过条件从$1 1000变成了$1 IS NOT NULL聚合调用从MAX($1)变成了MAX(CASE WHEN $1 1000 THEN $1 END)。这不是脱裤子放屁关键在于MAX(price) 1000这样一个“组级判断”被改写成了“行级 CASE 判断 聚合后判空”过滤语义下沉到了聚合函数的输入表达式里。这个形态对执行引擎有实际意义CASE WHEN 的布尔判断发生在聚合的输入行上和MAX的计算天然融合在一起避免了聚合算子先把结果物化出来、再由外部 Filter 算子读一遍再判断的开销。对向量化执行引擎来说把比较逻辑融进聚合函数内部通常能减少函数调用边界。更重要的是这种形态有机会继续走后续的AggregateProjectMergeRule之类的合并规则把投影层的 CASE 表达式进一步吸收最终变成更紧凑的执行计划。2. 动手写规则前必须算清的两笔账等价性与收益2.1 等价性改写不是想改就能改很多刚接触这个规则的人会本能地觉得SUM(x) 1000是不是也能改成SUM(CASE WHEN ... THEN x END)答案是不能。这是这个规则里最容易翻车的点。我们先把能严格证明的改写写出来。针对MAX和MIN与常数比较存在以下等价关系MAX(x) C等价于MAX(CASE WHEN x C THEN x END) IS NOT NULLMIN(x) C等价于MIN(CASE WHEN x C THEN x END) IS NOT NULL为什么成立以MAX(x) C为例。如果某一组里存在一行使得x C那么原条件成立改写后这一行会通过 CASE WHEN 生成非 NULL 值MAX的结果必然非空所以IS NOT NULL成立。反过来如果组内所有行的x都小于等于 C或者全是 NULL原条件为假改写后 CASE WHEN 对所有行都返回 NULLMAX的结果是 NULLIS NOT NULL为假。关键点是MAX(x) C描述的是“这一组里存在至少一个超过阈值的值”它是一种存在性判断所以能被翻译成“行级条件 聚合后判空”。SUM(x) C描述的是“这一组所有值的累计结果超过阈值”它依赖整组的累计单行无论如何都无法提前判定自己是不是最终结果的一部分。这是积累性条件不是存在性条件没有行级 CASE WHEN 的等价表达。2.2 收益要算清楚别自我感动这条规则写完后你拿出 EXPLAIN 计划一看Filter 还在聚合读取的行数也没减少很容易产生“这优化了个寂寞”的怀疑。实际上它的收益不在“行数减少”这个维度而是体现在下面三个地方第一聚合算子内部不再需要输出 MAX 值之后再交给下一个算子做数值比较执行引擎可以针对MAX(CASE WHEN ...)和IS NOT NULL做融合处理减少一次物化和遍历。第二CASE 表达式进了聚合调用之后后续投影合并、聚合合并规则才有机会触发否则一个悬空的 Filter 节点会挡住不少优化模式的匹配。第三物化视图改写场景里如果预计算的物化视图里正好有MAX(CASE WHEN ...)这个表达式查询通过本规则改写后就能直接对上线。这才是这条规则的真正价值。写规则前把收益模型想清楚才不会实现到一半发现“看起来没变化”就放弃。3. 实现一个可运行的 AggregateFilterToCaseRule3.1 匹配模式与前置检查原理清楚了就开始写代码。规则匹配的是“Filter 上面压着 Aggregate”的树形结构import org.apache.calcite.plan.RelOptRule; import org.apache.calcite.plan.RelOptRuleCall; import org.apache.calcite.rel.core.Aggregate; import org.apache.calcite.rel.core.AggregateCall; import org.apache.calcite.rel.core.Filter; import org.apache.calcite.rel.logical.LogicalAggregate; import org.apache.calcite.rel.logical.LogicalFilter; import org.apache.calcite.rex.RexCall; import org.apache.calcite.rex.RexInputRef; import org.apache.calcite.rex.RexLiteral; import org.apache.calcite.rex.RexNode; import org.apache.calcite.sql.SqlKind; import org.apache.calcite.sql.fun.SqlStdOperatorTable; public class AggregateFilterToCaseRule extends RelOptRule { public static final AggregateFilterToCaseRule INSTANCE new AggregateFilterToCaseRule(); private AggregateFilterToCaseRule() { super( operand(LogicalFilter.class, operand(LogicalAggregate.class, any())), AggregateFilterToCaseRule); } Override public void onMatch(RelOptRuleCall call) { LogicalFilter filter call.rel(0); LogicalAggregate aggregate call.rel(1); // 1. 过滤条件必须是一个简单的二元比较 RexNode condition filter.getCondition(); SqlKind kind condition.getKind(); if (kind ! SqlKind.GREATER_THAN kind ! SqlKind.LESS_THAN) { return; } RexCall comparison (RexCall) condition; RexNode left comparison.getOperands().get(0); RexNode right comparison.getOperands().get(1); // 2. 左侧必须是聚合输出引用右侧必须是常量 if (!(left instanceof RexInputRef) || !(right instanceof RexLiteral)) { return; } int refIndex ((RexInputRef) left).getIndex(); int groupCount aggregate.getGroupCount(); if (refIndex groupCount || refIndex groupCount aggregate.getAggCallList().size()) { return; } // 3. 找到对应的 AggregateCall int aggIndex refIndex - groupCount; AggregateCall aggCall aggregate.getAggCallList().get(aggIndex); if (aggCall.getArgList().size() ! 1) { return; } // 4. 只处理 MAX 常量 或 MIN 常量 SqlKind aggKind aggCall.getAggregation().getKind(); if (aggKind ! SqlKind.MAX aggKind ! SqlKind.MIN) { return; } if (!((aggKind SqlKind.MAX kind SqlKind.GREATER_THAN) || (aggKind SqlKind.MIN kind SqlKind.LESS_THAN))) { return; } // 5. 执行改写 try { RelNode newFilter doTransform(filter, aggregate, comparison, aggCall); call.transformTo(newFilter); } catch (Exception e) { // 保守起见任何意外情况都放弃改写 } } }前置检查里最容易被忽略的是refIndex的范围LogicalFilter里引用的$1是对LogicalAggregate输出列的引用而聚合输出列包括 group 列和 aggCall 列。必须先从 groupCount 开始偏移才能正确找到对应的AggregateCall。如果不做这个检查很可能拿到一个 group 列引用然后因为索引错位写出错误的计划。3.2 核心变换构建 CASE 列并重写聚合改写的核心分四步在聚合输入上追加一个 CASE WHEN 表达式列把原聚合调用的参数指向这个新列把原 Filter 条件改成IS NOT NULL组装新计划。我用低层 API 来写便于看清楚每一步发生了什么private RelNode doTransform(LogicalFilter filter, LogicalAggregate aggregate, RexCall comparison, AggregateCall aggCall) { RexBuilder rexBuilder filter.getCluster().getRexBuilder(); RelNode input aggregate.getInput(); // 1. 构造 CASE WHEN 表达式 int argIndex aggCall.getArgList().get(0); RexNode argRef rexBuilder.makeInputRef(input, argIndex); RexNode constant comparison.getOperands().get(1); // 比较运算price 1000 或 price 1000 RexNode caseCond rexBuilder.makeCall(comparison.getOperator(), argRef, constant); // CASE WHEN cond THEN arg ELSE NULL END RexNode newArg rexBuilder.makeCall(SqlStdOperatorTable.CASE, caseCond, argRef, rexBuilder.makeNullLiteral(argRef.getType())); // 2. 在输入之上追加新列 ListRexNode projects new ArrayList(); ListString fieldNames new ArrayList(); for (RelDataTypeField field : input.getRowType().getFieldList()) { projects.add(rexBuilder.makeInputRef(input, field.getIndex())); fieldNames.add(field.getName()); } int newFieldIndex projects.size(); projects.add(newArg); fieldNames.add($case_ aggCall.getAggregation().getName().toLowerCase()); RelNode newProject LogicalProject.create(input, projects, fieldNames); // 3. 构建新的 AggregateCall参数指向新列 AggregateCall newAggCall AggregateCall.create( aggCall.getAggregation(), aggCall.isDistinct(), aggCall.isApproximate(), aggCall.ignoreNulls(), ImmutableList.of(newFieldIndex), -1, // 没有 FILTER 参数 newProject.getRowType().getFieldList().get(newFieldIndex).getType(), aggCall.name); // 4. 重写 Filter 条件为 IS NOT NULL RexNode newRef rexBuilder.makeInputRef( newProject.getRowType(), groupCount aggIndex); RexNode newCondition rexBuilder.makeCall( SqlStdOperatorTable.IS_NOT_NULL, newRef); LogicalAggregate newAggregate LogicalAggregate.create( newProject, aggregate.getGroupSet(), aggregate.getGroupSets(), ImmutableList.of(newAggCall)); return LogicalFilter.create(newAggregate, newCondition); }这段代码有几个地方需要特别说明。第一新 CASE 列必须追加到旧 Project 的末尾而不是替换原有列。因为原来的聚合调用可能引用多个输入列贸然替换会让其他 aggCall 的列索引全部错位。追加新列再让新 aggCall 指向它是冲突最小的做法。第二AggregateCall.create的签名在不同 Calcite 版本里略有差别。上面的代码基于 1.36 及之后的常用签名它包含ignoreNulls参数如果你用的是旧版本可能要去掉这个参数。写自定义规则时最好先打开 IDE 看一眼依赖里的实际签名别直接抄。第三新 Filter 引用新聚合输出时索引依然是groupCount aggIndex。因为聚合输出的布局没有变group 列在前聚合列在后改写的只是这个聚合列的计算方式。这里有个隐藏陷阱原聚合可能还有别的 aggCall比如SUM(amount)。改写的newAggCall只替换了目标 MAX/MIN 调用其他 aggCall 必须原样保留。上面代码用ImmutableList.of(newAggCall)其实是把其他 aggCall 丢掉了这是不对的。正确做法是把aggregate.getAggCallList()里除了目标之外的调用全部复制再把 newAggCall 放到对应位置。为了把逻辑讲清楚上面做了简化实际实现时我会建议直接写一个循环ListAggregateCall newAggCalls new ArrayList(); for (int i 0; i aggregate.getAggCallList().size(); i) { if (i aggIndex) { newAggCalls.add(newAggCall); } else { newAggCalls.add(aggregate.getAggCallList().get(i)); } }3.3 注册到 HepPlanner 并验证Calcite 里跑自定义规则最方便的方式是用 HepPlanner它按规则顺序在节点上反复遍历适合这种单一模式的规则验证HepProgram program new HepProgramBuilder() .addRuleInstance(AggregateFilterToCaseRule.INSTANCE) .build(); HepPlanner planner new HepPlanner(program); planner.setRoot(originalRel); RelNode optimized planner.findBestExp(); String plan RelOptUtil.toString(optimized); System.out.println(plan);在开发环境里调规则时我会习惯先把原始计划打印出来改写后再打印一遍人工比对两个关键点CASE WHEN 是否正确出现在聚合调用里以及过滤条件是否变成了IS NOT NULL。计划树一旦发生字段索引错位Calcite 往往在 toString 阶段不会立刻报错而是在真正执行或进一步优化时才暴露所以这一步别偷懒。为了做单元测试可以借助 Calcite 自带的RelBuilder手工构造计划也可以直接跑CalciteAssert。一个更朴素的验证办法是写一个简单 Java 程序用RelBuilder构造一个LogicalFilter加LogicalAggregate的树然后走 HepPlanner。跑通之后再套真实 SQL用 explain plan 对比。4. 验证、边界与真实项目中的替代方案4.1 实测用一个可复现例子走完整条链我用一张tx(dept_id, price, amount)表做实验。原始 SQLSELECT dept_id, MAX(price) AS m, SUM(amount) AS s FROM tx GROUP BY dept_id HAVING MAX(price) 1000;规则触发前计划是LogicalFilter(condition[$1 1000]) LogicalAggregate(group[{0}], m[MAX($1)], s[SUM($2)]) LogicalTableScan(table[[tx]])规则触发后可以看到聚合输入上多了一层LogicalProject里面追加了 CASE 列LogicalFilter(condition[$2 IS NOT NULL]) LogicalAggregate(group[{0}], m[MAX($3)], s[SUM($2)]) LogicalProject(dept_id[$0], amount[$1], price[$2], $case_max[CASE WHEN $2 1000 THEN $2 ELSE NULL END]) LogicalTableScan(table[[tx]])注意这里字段顺序新 Project 有 4 列$3是新增的 CASE 列所以新的MAX($3)指向它而SUM($2)仍然指向 amount。计划形态符合预期。在我实际测试中这条规则对数据量的影响很小因为毕竟没有减少扫描行数。但从算子结构上看多出的LogicalProject在后续优化中经常被AggregateProjectMergeRule吸收掉真正的物理计划不会保留这一层多余投影。也就是说规则后的计划往往只是一个中间态最终受益的是后续规则的连锁反应。4.2 四个注定失败的改写场景深入理解一个规则更重要的是知道它的边界。我踩过的坑集中在下面四种场景。第一个是SUM(x) C或者COUNT(*) C。有人会觉得 CASEWHEN 能把SUM(x)改成SUM(CASE WHEN ...)但前面已经证明过累计性条件没有行级等价改写硬写只会破坏语义。这是最容易犯错的一个。第二个是过滤条件里同时引用了分组键和聚合结果比如HAVING dept_id 0 AND MAX(price) 1000。这种根本不是一个规则能搞定的分组键条件应该先拆出去交给AggregateFilterTransposeRule下推剩下的纯聚合条件才轮到本规则处理。规则编写时一定要检查条件是否只引用了聚合输出列否则就主动放弃。第三个是多个 aggCall 被同一个 Filter 引用。比如HAVING MAX(price) 1000 AND MIN(amount) 500这需要同时对两个聚合调用做 CASEWHEN 改写并且要保证两个 CASE 列互不影响。实现不是不行但匹配条件、列索引、计划组装都会复杂一个数量级。我的建议是第一个版本只处理单聚合条件复杂的等以后有真实场景再说。第四个是条件不是“聚合结果与常量的直接比较”而是表达式比如HAVING MAX(price) * 2 1000。这种表达式无法直接反解出 CASE 条件。要做也是把 1000 除以 2 之后变成MAX(price) 500但涉及表达式的代数变换老实说我已经不把它归在简单改写范围内了。4.3 选型建议什么时候用自定义规则什么时候别用很多人写自定义规则容易陷入“为了写而写”的怪圈。我的判断标准很简单先看 Calcite 内置规则集里有没有等价能力的组合方案。针对 Filter 和 Aggregate 的交互最常用的是AggregateFilterTransposeRule和FilterAggregateTransposeRule。前者把能下推的下推后者专门处理一些聚合下推的边界情况。如果这两个就解决了问题不要额外引入自定义规则。自定义规则适合出现在下面两种情况一是你确认内置规则在这个模式上确实没有行动二是你的执行引擎对某种特殊表达式形态有额外收益比如它的聚合算子原生支持“条件聚合”或者“参数化聚合”这种情况下规则的存在才有意义。另外自定义规则维护成本比大多数人想象的高。Calcite 的 API 在版本间变动不小一个AggregateCall.create签名就能让代码在不同版本间来回改。如果业务场景能用 SQL 层面的FILTER子句表达优先在 SQL 里写FILTER它走的是 Calcite 官方支持路径稳定性远高于自己改逻辑计划。最后说点实际操作中的体会。这条规则我在第一次实现时栽在了一个特别隐蔽的坑里新 Project 追加列的顺序。当时我图省事把 CASE 列放在最前面结果聚合输出列索引全乱了规则触发了但结果错误查了很久才发现是LogicalAggregate的 group 列和 aggCall 列混在了一起。后来我给自己定了一条规矩任何改写规则的字段操作都先画出“旧树字段索引”和“新树字段索引”的对照表再动手写代码。另外一个经验是这种只针对特定模式生效的小规则在 HepPlanner 里跑通容易但要进生产环境的优化器管道一定要配套计划正确性测试。毕竟优化器规则一旦写错给出来的是“看起来对但语义错”的计划比不优化更可怕。