Flink SQL 优化 —— 谓词下推

public class JdbcDynamicTableSource
        implements ScanTableSource,
        LookupTableSource,
        SupportsFilterPushDown, // 此处新增实现 SupportsFilterPushDown 接口
        SupportsProjectionPushDown,
        SupportsLimitPushDown {
          
          
          // 省略其他原先代码
          
          /**
           * 1. 该方法的作用是将filters(过滤条件)应用到当前的DynamicTableSource上
           * 2. 该方法的返回值是一个Result对象，该对象包含两个List<ResolvedExpression>类型的属性，
           *    分别是acceptedFilters(接受的过滤条件)和remainingFilters(剩余的过滤条件)
           * 3. acceptedFilters是指那些被DynamicTableSource使用的filters，
           *    但是这些filters可能只是尽可能在原基础上被应用，
           *    因此，这些filters的信息可以帮助planner来调整当前plan的成本估算
           * 4. remainingFilters是指那些不能被DynamicTableSource完全消费的filters，
           *    这些filters决定了是否需要在runtime阶段继续应用filter操作
           * 5. acceptedFilters和remainingFilters不能是disjunctive lists，
           *    一个filter可以同时出现在这两个list中，但是所有的filters必须至少出现在其中一个list中
           */
          @Override
          public Result applyFilters(List<ResolvedExpression> filters) {
              this.filters = filters;
              return Result.of(this.filters, Lists.newArrayList());
          }
          
          public boolean equals(Object o) {
            
            if (this == o) {
              return true;
            } else if (!(o instanceof JdbcDynamicTableSource)) {
              return false;
            } else {
              JdbcDynamicTableSource that = (JdbcDynamicTableSource)o;
              return Objects.equals(this.options, that.options)
                && Objects.equals(this.readOptions, that.readOptions)
                && Objects.equals(this.lookupOptions, that.lookupOptions)
                && Objects.equals(this.physicalRowDataType, that.physicalRowDataType)
                && Objects.equals(this.dialectName, that.dialectName)
                && Objects.equals(this.limit, that.limit)
                // Jdbc Connector 重写了 equals 方法，需要加上新增的 filters 属性
                && Objects.equals(this.filters, that.filters);
              }
          }
          
          public int hashCode() {
            return Objects.hash(
              new Object[]{
                this.options,
                this.readOptions,
                this.lookupOptions,
                this.physicalRowDataType,
                this.dialectName,
                this.limit,
                // Jdbc Connector 重写了 hashCode 方法，需要加上新增的 filters 属性
                this.filters
              });
          }
}