Flink SQL 之 Table 查询与过滤原理及代码实现：从投影裁剪到谓词下推与短路求值

上一篇讲了 Table API 的整体原理和表达式体系。这篇深入 Table API 中最基础、最常用的两个操作——查询select 投影与过滤filter/where。几乎每一个 Flink SQL 作业都会用到 select 和 filter但很多同学只是简单地写select(...).filter(...)对背后的原理、优化机制、性能影响知之甚少。实际上select 和 filter 是性能优化的关键抓手——投影裁剪可以减少 50% 以上的数据读取量谓词下推可以将过滤推到数据源端执行条件重排和短路求值可以减少大量无效计算。这篇从本质定义、四大操作详解、执行顺序对比、六步内部执行流程、四大优化原理、表达式计算与短路求值、完整代码实现、六个常见坑、八条最佳实践把查询与过滤一次性讲透。一、查询与过滤本质定义查询与过滤的本质一句话概括select投影负责列的选择和计算filter过滤负责行的筛选两者配合从原始表中提取出需要的数据子集是关系代数中最基础的两个操作。下面这张图把查询与过滤的本质定义、四大核心操作、select 与 filter 执行顺序对比放在一起展示。从关系代数的角度看select 对应投影操作πfilter 对应选择操作σ。这两个操作是所有复杂查询的基础——聚合、连接、窗口等高级操作都建立在投影和过滤之上。1.1 执行顺序在逻辑执行计划中filter 通常先于 select 执行先过滤行再计算列这样可以减少需要计算的数据量。优化器会自动进行谓词下推Predicate Pushdown和投影裁剪Projection Pushdown将过滤和投影尽可能推到数据源端执行减少数据传输和计算量。即使你写的是table.select(...).filter(...)select 在前优化器也会自动将 filter 提到 select 之前执行。这是查询优化器的基本能力不需要手动调整顺序。1.2 无状态算子Projection 和 Filter 都是无状态算子不需要维护状态逐行处理Filter 算子对每一行计算条件表达式返回 true 的行向下游传递false 的行丢弃Projection 算子对每一行计算所有输出列的表达式生成新的行向下游传递因为无状态这两个算子的性能非常高也不会有状态膨胀的问题distinct 除外distinct 需要状态。二、四大操作详解2.1 select 投影Projection作用选择列、计算表达式、重命名列。方法select($(a), $(b))选择列select($(a).plus($(b)).as(sum))表达式计算 重命名select($(*))选择所有列不推荐原理生成 Projection 算子对每一行输入计算输出列。表达式在编译期解析为 RexNode运行时通过 CodeGen 生成专门的 Java 代码直接操作 BinaryRowData 的字段性能接近手写 Java 代码。优化投影裁剪Projection Pushdown将不需要的列裁剪掉推到数据源端。列式存储Parquet/ORC可以只读取需要的列大幅减少 IO 和反序列化开销。2.2 filter/where 过滤Filter作用根据条件表达式筛选行只保留满足条件的行。方法filter($(a).isGreater(10))where($(a).isGreater(10).and($(b).isEqual(x)))filter 和 where 完全等价where 是 SQL 风格的别名。原理生成 Filter 算子对每一行计算条件表达式返回 true 的行向下游传递false 的行丢弃。条件表达式同样通过 CodeGen 生成专门的 Java 代码。优化谓词下推Predicate Pushdown将过滤条件推到数据源端支持谓词下推的连接器JDBC/Hive/Parquet可以在数据源端执行过滤大幅减少数据传输。2.3 distinct 去重Deduplication作用去除重复行只保留唯一行。方法table.distinct()原理批处理基于排序或哈希去重一次性处理所有数据流处理基于状态去重维护一个已见行的集合状态每来一行检查是否已存在不存在则输出并加入状态存在则丢弃注意流处理 distinct 需要状态管理状态会持续增长需要设置状态 TTLtable.exec.state.ttl防止 OOM。distinct 是全局去重不是分组去重分组去重用 groupBy。2.4 limit/offset 分页Pagination作用限制返回行数实现分页查询。方法table.limit(10)取前 10 行table.orderBy($(a).desc()).fetch(10).offset(20)排序后分页原理批处理全局排序后取前 N 行需要将所有数据拉到一个节点排序全局排序开销大流处理limit 不直接支持无界流因为流是无限的无法确定前 N 行需要配合窗口或 ROW_NUMBER() 使用注意limit 是批处理操作流处理中使用需要谨慎。offset 会跳过前 N 行大数据量下 offset 较大时性能差需要扫描并丢弃前 N 行。三、select 与 filter 执行顺序对比对比维度select 投影filter 过滤关系代数投影操作 π选择操作 σ操作对象列垂直裁剪行水平筛选逻辑执行顺序通常在 filter 之后执行通常在 select 之前执行优化策略投影裁剪Projection Pushdown谓词下推Predicate Pushdown状态需求无状态逐行计算无状态逐行判断流处理支持完全支持无界流逐行投影完全支持无界流逐行过滤性能影响减少列数减少序列化/网络传输表达式计算增加 CPU过滤率越高下游数据量越小条件计算增加 CPU最佳实践只选择需要的列避免 select(*)过滤条件尽早执行利用谓词下推避免复杂 UDF核心结论filter 先于 select 执行是优化器的默认行为不需要手动调整。但在写代码时应该有意识地将过滤条件尽早写出来在聚合、连接之前让优化器有更多优化空间。四、查询与过滤内部执行流程理解了本质和操作分类下面深入内部执行流程看 select 和 filter 的方法调用到底是怎么变成执行代码的。下面这张图把六步执行流程、四大优化原理、表达式计算与短路求值放在一起展示。4.1 六步执行流程一条 select/filter 链式调用从输入到执行完整的流程分为六步方法调用用户调用 select/filter/distinct 等方法构建 Operation每个方法构建一个 QueryOperation 节点RelNode 转换Operation 翻译为 Calcite LogicalProject/LogicalFilter优化器优化谓词下推/投影裁剪/常量折叠/条件重排代码生成CodeGen 生成 Projection/Filter 算子的 Java 代码逐行执行算子逐行处理Filter 判断条件 → Projection 计算列4.2 关键理解select 和 filter 在逻辑执行计划中分别对应LogicalProject和LogicalFilter两个 RelNode 节点。优化器会对这两个节点进行多种优化。代码生成CodeGen是性能的关键Flink 会为 Projection 和 Filter 算子生成专门的 Java 代码而不是用解释执行。生成的代码直接操作二进制行数据BinaryRowData避免了对象创建和虚方法调用性能非常高。这也是 Flink Table API/SQL 性能优秀的关键原因之一。比如$(amount).times(lit(0.8)).as(discount)会被翻译为类似out.setDecimal(0, in.getDecimal(2).multiply(0.8))这样的直接字段操作代码没有任何抽象层开销。五、四大优化原理5.1 谓词下推Predicate Pushdown原理将 Filter 条件尽可能推到数据源端执行减少从数据源读取的数据量。效果支持谓词下推的连接器JDBC/Hive/Parquet可以在数据源端执行过滤只读取满足条件的数据大幅减少网络传输和计算量。示例SELECT * FROM orders WHERE amount 100谓词 amount 100 下推到 MySQLMySQL 只返回 amount 100 的行。限制不是所有谓词都能下推包含 UDF、复杂表达式的谓词可能无法下推。用explain()查看执行计划确认谓词是否下推。5.2 投影裁剪Projection Pushdown原理将 Projection列选择推到数据源端只读取需要的列不需要的列直接裁剪掉。效果列式存储Parquet/ORC可以只读取需要的列大幅减少 IO 和反序列化开销。行式存储JSON/CSV也可以在读取后立即裁剪不需要的列。示例SELECT user_id, amount FROM orders只读取 user_id 和 amount 两列其他列order_id/status/order_time直接裁剪。最佳实践永远不要用select(*)或SELECT *只选择需要的列。这是最简单也最有效的性能优化手段。5.3 常量折叠Constant Folding原理在优化阶段直接计算只包含常量的表达式不需要在运行时逐行计算。效果减少运行时计算量表达式结果在编译期就确定了。示例SELECT amount * 2, 1 2 FROM orders常量表达式 1 2 在优化期直接计算为 3运行时不需要计算。5.4 条件重排Condition Reordering原理优化器会根据条件的选择性和计算成本重新排列 AND 条件的顺序将高选择性过滤率高、低成本的条件放在前面利用短路求值减少计算量。效果高选择性条件先执行大部分行在第一个条件就被过滤掉不需要计算后续的复杂条件。注意优化器的条件重排是基于统计信息的如果没有统计信息如流处理优化器可能不会重排此时需要手动将高选择性条件放在前面。六、表达式计算与短路求值6.1 表达式编译RexNode → CodeGenTable API 中的 Expression如$(a).plus($(b))在编译期被解析为 Calcite 的 RexNode行表达式节点然后在代码生成阶段被翻译为专门的 Java 代码。生成的代码直接操作 BinaryRowData 的字段避免了表达式解释执行的开销性能接近手写 Java 代码。6.2 短路求值Short-circuit Evaluation原理AND 条件中如果前面的条件返回 false后面的条件不会执行因为整个表达式已经确定为 false。OR 条件中如果前面的条件返回 true后面的条件不会执行。效果减少不必要的条件计算尤其是包含 UDF 或复杂计算的条件。最佳实践将高选择性、低成本的条件放在 AND 的前面将低选择性、高成本的条件如 UDF放在后面。这样大部分行在第一个条件就被过滤掉不需要执行昂贵的 UDF。6.3 NULL 处理三值逻辑SQL 采用三值逻辑TRUE/FALSE/UNKNOWN任何与 NULL 的比较结果都是 UNKNOWNUNKNOWN 在 WHERE 中被视为 false行被过滤掉。常见坑$(a).isEqual(lit(null))永远返回 UNKNOWN不会匹配任何行判断 NULL 必须用$(a).isNull()。如果需要保留 NULL 行用WHERE a 10 OR a IS NULL。6.4 过滤率与性能过滤率Filter Selectivity 过滤后行数 / 原始行数。过滤率越低过滤掉的行越多下游数据量越小整体性能越好。Filter 是减少数据量最有效的算子一个高过滤率的 Filter 可以大幅减少下游聚合、连接、窗口等算子的计算量和状态大小。最佳实践尽早执行 Filter在聚合/连接之前利用谓词下推将 Filter 推到数据源端避免在 Filter 中使用复杂 UDFUDF 无法谓词下推且计算成本高用explain()确认 Filter 的位置和谓词下推情况七、完整代码实现下面是一个完整的查询与过滤使用示例包含创建环境、DDL 建表、select 投影、filter 过滤、组合查询、explain 检查、输出执行。下面这张图把完整代码实现、六个常见坑、八条最佳实践放在一起展示。importorg.apache.flink.streaming.api.environment.StreamExecutionEnvironment;importorg.apache.flink.table.api.Table;importorg.apache.flink.table.api.TableResult;importorg.apache.flink.table.api.bridge.java.StreamTableEnvironment;importjava.time.ZoneId;importstaticorg.apache.flink.table.api.Expressions.*;publicclassTableQueryExample{publicstaticvoidmain(String[]args)throwsException{// 1. 创建执行环境StreamExecutionEnvironmentenvStreamExecutionEnvironment.getExecutionEnvironment();env.enableCheckpointing(60000);StreamTableEnvironmenttableEnvStreamTableEnvironment.create(env);tableEnv.getConfig().setLocalTimeZone(ZoneId.of(Asia/Shanghai));// 2. DDL创建Kafka源表tableEnv.executeSql( CREATE TABLE orders ( order_id BIGINT, user_id BIGINT, amount DECIMAL(10,2), status STRING, city STRING, order_time TIMESTAMP(3), WATERMARK FOR order_time AS order_time - INTERVAL 5 SECOND ) WITH ( connector kafka, topic orders, properties.bootstrap.servers localhost:9092, format json ) );// 3. DDL创建MySQL结果表tableEnv.executeSql( CREATE TABLE high_value_orders ( order_id BIGINT, user_id BIGINT, discount_amount DECIMAL(10,2), city STRING, PRIMARY KEY (order_id) NOT ENFORCED ) WITH ( connector jdbc, url jdbc:mysql://localhost:3306/flink_db, table-name high_value_orders, username root, password 123456 ) );// 4. select投影只选择需要的列计算折扣金额TableprojectedtableEnv.from(orders).select($(order_id),$(user_id),$(amount),$(amount).times(lit(0.8)).as(discount_amount),$(status),$(city));// 5. filter过滤高价值订单金额500 且 已支付 且 一线城市// 注意高选择性条件放在前面利用短路求值减少计算Tablefilteredprojected.filter($(status).isEqual(lit(PAID))// 高选择性先过滤.and($(city).in(lit(北京),lit(上海),lit(广州),lit(深圳))).and($(amount).isGreater(lit(500)))// 数值比较成本低.and($(order_id).isNotNull())// NULL检查放在最后);// 6. 再次select只保留输出需要的列投影裁剪Tableresultfiltered.select($(order_id),$(user_id),$(discount_amount),$(city));// 7. 查看执行计划确认谓词下推和投影裁剪Stringplanresult.explain();System.out.println( 执行计划 );System.out.println(plan);// 8. 写入结果表并执行TableResultexecResultresult.executeInsert(high_value_orders);// 9. 等待作业完成execResult.await();}}代码关键点静态导入 Expressionsimport static org.apache.flink.table.api.Expressions.*这是 Table API 的第一步。select 投影只选择需要的列计算折扣金额并用as()重命名。避免select(*)。filter 过滤高选择性条件status ‘PAID’放在最前面利用短路求值减少计算。城市 IN 条件次之数值比较再次之NULL 检查放在最后。再次 select 投影裁剪过滤后再次 select只保留输出需要的列让优化器进行投影裁剪。explain 检查执行result.explain()查看执行计划确认谓词下推和投影裁剪是否生效。executeInsert 执行写入结果表并触发执行result.await()等待作业完成。八、六个常见坑8.1 坑一SELECT * 导致投影裁剪失效现象查询性能差数据源读取了大量不需要的列网络传输和反序列化开销大。根因用select($(*))或SELECT *选择所有列优化器无法进行投影裁剪必须读取所有列。解决方案永远只选择需要的列明确列出列名。这是最简单也最有效的性能优化手段尤其是列式存储Parquet/ORC可以只读取需要的列。8.2 坑二Filter 中使用 UDF 导致谓词无法下推现象查询性能差数据源读取了全量数据过滤在 Flink 端执行。根因Filter 条件中包含自定义函数UDF优化器无法将包含 UDF 的谓词下推到数据源端必须在 Flink 端读取全量数据后再过滤。解决方案尽量用内置函数替代 UDF内置函数可能支持下推将 UDF 条件和可下推条件分开可下推条件放在前面会被下推UDF 条件放在后面在 Flink 端执行用explain()确认谓词下推情况8.3 坑三NULL 比较用 isEqual 导致结果错误现象查询结果不符合预期NULL 行没有被正确匹配或过滤。根因SQL 采用三值逻辑任何与 NULL 的比较结果都是 UNKNOWN在 WHERE 中视为 false。$(a).isEqual(lit(null))永远返回 UNKNOWN不会匹配任何行。解决方案判断 NULL 必须用$(a).isNull()或$(a).isNotNull()。如果需要保留 NULL 行用$(a).isGreater(10).or($(a).isNull())。8.4 坑四Filter 条件顺序不当导致性能差现象查询性能差大量行执行了昂贵的 UDF 或复杂计算后才被过滤掉。根因AND 条件中低选择性、高成本的条件如 UDF放在前面高选择性、低成本的条件放在后面。虽然优化器可能会重排条件但流处理中没有统计信息时优化器不会重排。解决方案手动将高选择性、低成本的条件放在 AND 的前面如 status ‘PAID’、city IN (…)将低选择性、高成本的条件如 UDF、复杂计算放在后面。利用短路求值大部分行在第一个条件就被过滤掉。8.5 坑五流处理 distinct 状态无限增长现象作业运行一段时间后 Checkpoint 越来越大TaskManager OOM作业失败。根因流处理 distinct 基于状态去重维护一个已见行的集合。如果 key 基数大如用户 ID、订单 ID状态会持续增长不会自动清理。解决方案设置状态 TTLtable.exec.state.ttl过期的状态会被清理考虑用 groupBy 替代 distinctgroupBy 可以配合窗口限制状态如果是基于事件时间的去重用 ROW_NUMBER() 去重可以配合窗口限制状态8.6 坑六流处理 limit/offset 不支持或结果不确定现象流处理中使用 limit 报错或者结果不确定、不符合预期。根因limit 是批处理操作需要全局排序后取前 N 行。流处理是无界流无法确定前 N 行因为数据持续到来。直接在无界流上使用 limit 会报错或结果不确定。解决方案流处理中不要直接使用 limit需要配合窗口使用先窗口聚合再在窗口内取 TopN用 ROW_NUMBER() OVER (PARTITION BY … ORDER BY …) 实现 TopN流处理标准做法如果是批处理模式可以正常使用 limit/offset九、八条最佳实践 Checklist上线前逐条检查永远只选择需要的列不要用 select(*) 或 SELECT *明确列出需要的列名让优化器进行投影裁剪减少数据读取和传输。过滤条件尽早执行在聚合、连接、窗口之前执行 Filter减少下游数据量。高过滤率的 Filter 是减少数据量最有效的手段。利用谓词下推尽量用内置函数和简单比较条件避免在 Filter 中使用 UDF。用 explain() 确认谓词是否下推到数据源端。条件顺序手动优化AND 条件中将高选择性、低成本的条件放在前面低选择性、高成本的条件UDF放在后面利用短路求值减少计算。NULL 判断用 isNull/isNotNull不要用 isEqual(lit(null)) 判断 NULLSQL 三值逻辑下结果永远是 UNKNOWN。需要保留 NULL 行时用 OR isNull()。流处理 distinct 设置状态 TTL流处理 distinct 基于状态去重必须设置 table.exec.state.ttl 防止状态无限增长导致 OOM。流处理 TopN 用 ROW_NUMBER不要在无界流上直接用 limit用 ROW_NUMBER() OVER (PARTITION BY … ORDER BY …) 实现流处理 TopN。上线前 explain() 检查执行计划对复杂查询执行 explain() 检查执行计划确认谓词下推、投影裁剪、Filter 位置、算子顺序是否符合预期。十、总结与下一篇预告Table 查询与过滤原理及代码实现要点回顾第一本质select投影负责列的选择和计算filter过滤负责行的筛选两者配合从原始表中提取出需要的数据子集是关系代数中最基础的两个操作π投影和σ选择。第二四大操作select 投影列选择/表达式计算/重命名投影裁剪优化、filter 过滤行筛选/条件判断/短路求值谓词下推优化、distinct 去重批处理排序/哈希流处理基于状态需设置 TTL、limit 分页批处理全局排序取 N 行流处理需配合窗口或 ROW_NUMBER。第三执行顺序filter 先于 select 执行是优化器的默认行为先过滤行再计算列不需要手动调整。Projection 和 Filter 都是无状态算子逐行处理性能高。第四六步内部执行流程方法调用 → 构建 Operation → RelNode 转换LogicalProject/LogicalFilter→ 优化器优化 → 代码生成CodeGen 生成专门 Java 代码直接操作 BinaryRowData→ 逐行执行。第五四大优化原理谓词下推Filter 推到数据源端JDBC/Hive/Parquet 支持、投影裁剪只读取需要的列列式存储效果显著、常量折叠常量表达式编译期计算、条件重排高选择性低成本条件在前利用短路求值。第六表达式计算与短路求值表达式通过 RexNode → CodeGen 编译为直接操作 BinaryRowData 的 Java 代码AND 条件短路求值前 false 不执行后条件NULL 三值逻辑isEqual(null) 永远 UNKNOWN用 isNull 判断过滤率越低下游数据量越小Filter 是减少数据量最有效算子。第七完整代码实现创建 StreamEnv → 创建 StreamTableEnv → DDL 建表Kafka 源/MySQL 结果→ select 投影只选需要的列 计算折扣金额→ filter 过滤高选择性条件在前 城市 IN 金额 500 isNotNull→ 再次 select 投影裁剪 → explain 检查执行计划 → executeInsert 执行 → await 等待完成。第八六个常见坑SELECT * 导致投影裁剪失效、Filter 中 UDF 导致谓词无法下推、NULL 比较用 isEqual 导致结果错误、Filter 条件顺序不当导致性能差、流处理 distinct 状态无限增长、流处理 limit 不支持或结果不确定。第九八条最佳实践永远只选择需要的列、过滤条件尽早执行、利用谓词下推、条件顺序手动优化、NULL 判断用 isNull/isNotNull、流处理 distinct 设置状态 TTL、流处理 TopN 用 ROW_NUMBER、上线前 explain() 检查执行计划。查询与过滤是 Flink SQL 中最基础也最重要的操作。理解了投影裁剪、谓词下推、短路求值等优化原理就能写出高性能的查询代码避免常见的性能陷阱。一个高过滤率的 Filter 精准的投影裁剪可以将查询性能提升数倍甚至数十倍。