java - 为什么自定义 BaseRelation 没有获得下推过滤器(在 buildScan 中)？

Question

我正在使用数据源 API 实现我自己的数据源。

我使用了在网上找到的几个文档并得出了以下代码:

public class MyRelation extends BaseRelation implements TableScan, PrunedScan, PrunedFilteredScan {
  public RDD<Row> buildScan(String[] requiredColumns, Filter[] filters) {
    System.out.println(Arrays.toString(filters));
  }
}

我使用以下查询得到空过滤器:

SELECT field1,field2,field3 from table WHERE field2>4 AND field3=1000

我运行的:

SQLContext sqc = new SQLContext(sparkContext);
Dataset<Row> sqlResult = sqc.sql( query ).where("field2 > 4").filter("field2 > 15");

如您所见，我还尝试了 API 的 filter 和 where 函数，但没有任何效果。

我还尝试了空白的 unhandledFilters 函数:

public Filter[] unhandledFilters(Filter[] filters)
{
    System.out.println(Arrays.toString(filters));
    return filters;
}

我希望在 filters 数组中得到过滤器“field2>15”和其他过滤器，但它是空白的。

有什么想法我做错了什么以及可以做些什么来解决这个问题吗？

附言我确实实现了 TableScan 和 PrunedScan，并且调用了正确的方法，当我运行没有 where 子句且没有过滤器的查询时，仍然调用了带过滤器的函数(继承自 PrunedFilteredScan)。

Answer 1

您应该简单地从您的 MyRelation 中删除其他契约(Contract)，即 TableScan 和 PrunedScan，因为它们妨碍并优先于PrunedFilteredScan。

引用 PrunedFilteredScan Contract — Relations with Column Pruning and Filter Pushdown :

PrunedFilteredScan is the contract of BaseRelations with support for column pruning (i.e. eliminating unneeded columns) and filter pushdown (i.e. filtering using selected predicates only).

filter pushdown 部分正是您所追求的，Spark SQL 2.3 中的 PrunedFilteredScan 关系支持该部分。

提示:看看JDBCRelation作为 Apache Spark 2.3 官方发行版中 PrunedFilteredScan 的唯一实现。