ScanTableSource 为什么不支持SupportsFilterPushDown的问题

[jy l]

Hi:
由于业务需要，我想在flink1.11.2中自定义一个ScanTableSource， 并想实现SupportsFilterPushDown特性。
但是在摸索的过程中发现，以下代码确似乎禁止使用SupportsFilterPushDown特性。

val unsupportedAbilities = List(
 classOf[SupportsFilterPushDown],
 classOf[SupportsLimitPushDown],
 classOf[SupportsPartitionPushDown],
 classOf[SupportsComputedColumnPushDown],
 classOf[SupportsWatermarkPushDown])
unsupportedAbilities.foreach { ability =>
 if (ability.isAssignableFrom(tableSource.getClass)) {
 throw new UnsupportedOperationException("Currently, a DynamicTableSource
with " +
 s"${ability.getSimpleName} ability is not supported.")
 }
}

SupportsFilterPushDown的介绍，Enables to push down filters into a {@link
ScanTableSource}.

这样的话，是不是有点矛盾呢？

我怎么能在ScanTableSource上实现push down filters？
以及实现了SupportsFilterPushDown的source有哪些？

望知道的大佬告知，感谢。

祝好！

[Leonard Xu]

Hi，
在1.11中，planner 并没有支持下表中的各种PushDown, 所以这里做了check，这是planner层面不支持的。在1.12里，planner层面已经支持了这些PushDown，所以这些check都没有了，用户可以自定义 connector 并实现各种PushDown,比如，1.12中已经支持了kafka source上的watermarkPushdown。

因此，有这类需求建议基于1.12开发。

祝好,
Leonard

>
> Hi:
> 由于业务需要，我想在flink1.11.2中自定义一个ScanTableSource， 并想实现SupportsFilterPushDown特性。
> 但是在摸索的过程中发现，以下代码确似乎禁止使用SupportsFilterPushDown特性。
>
> val unsupportedAbilities = List(
> classOf[SupportsFilterPushDown],
> classOf[SupportsLimitPushDown],
> classOf[SupportsPartitionPushDown],
> classOf[SupportsComputedColumnPushDown],
> classOf[SupportsWatermarkPushDown])
> unsupportedAbilities.foreach { ability =>
> if (ability.isAssignableFrom(tableSource.getClass)) {
> throw new UnsupportedOperationException("Currently, a DynamicTableSource
> with " +
> s"${ability.getSimpleName} ability is not supported.")
> }
> }
>
> SupportsFilterPushDown的介绍，Enables to push down filters into a {@link
> ScanTableSource}.
>
> 这样的话，是不是有点矛盾呢？
>
> 我怎么能在ScanTableSource上实现push down filters？
> 以及实现了SupportsFilterPushDown的source有哪些？
>
> 望知道的大佬告知，感谢。
>
> 祝好！