Hi, all. It seems the format is not normal. So I add a google doc in
link[1]. This discussion is about the interfaces in FLIP-95: New Table
Source And Table Sink and propose to merge two interfaces
SupportsWatermarkPushDown and SupportsComputedColumnPushDown.
I am looking forward to any opinions and suggestions from the community.
Regards,
Shengkai
[1]
https://docs.google.com/document/d/1sIT8lFZ_MeNIh_GLE3hi7Y4pgzN90Ahw_LoBFni-GT4/edit#
Shengkai Fang <fskm...@gmail.com> 于2020年9月4日周五 下午2:58写道：
Hi, all. Currently, we use two seperated interfaces
SupportsComputedColumnPushDown and SupportsWatermarkPushDown in design.
The
interface SupportsWatermarkPushDown relies on
SupportsComputedColumnPushDown when watermark is defined on a computed
column. During the implementation, we find the method in
SupportsWatermarkPushDown uses an out-of-date interface WatermarkProvider
and the duplication of SupportsComputedColumnPushDown and
SupportsProjectionPushDown. Therefore, we decide to propose a new
interface of SupportsWatermarkPushDown to solve the problems we
mentioned.
*Problems of SupportsComputedColumnPushDown and
SupportsWatermarkPushDown*Problems
of SupportsWatermarkPushDown
SupportsWatermarkPushDown uses an inner interface named
WatermarkProvider to
register WatermarkGenerator into DynamicTableSource now. However, the
community uses org.apache.flink.api.common.eventtime.WatermarkStrategy to
create watermark generators in FLIP-126. WatermarkStrategy is a factory
of TimestampAssigner and WatermarkGeneartor and FlinkKafkaConsumer uses
the method assignTimestampsAndWatermarks(WatermarkStrategy) to generate
Kafka-partition-aware watermarks. As for the origin WatermarkProvider, it
is used to generate deprecated AssignerWithPeriodicWatermarks and
PunctuatedWatermarkAssignerProvider. Therefore, we think it's not
suitable to use the WatermarkProvider any more.
Problems of SupportsComputedColumnPushDown
There are two problems around when using SupportsComputedColumnPushDown
alone.
First, planner will transform the computed column and query such as
select
a+b to a LogicalProject. When it comes to the optimization phase, we have
no means to distinguish whether the Rexnode in the projection is from
computed columns or query. So SupportsComputedColumnPushDown in reality
will push not only the computed column but also the calculation in the
query.
Second, when a plan matches the rule
PushComputedColumnIntoTableSourceScanRule, we have to build a new
RowData to
place all fields we require. However, both two rules
PushComputedColumnIntoTableSourceScanRule and
PushProjectIntoTableSourceScanRule will do the same work that prune the
records that read from source. It seems that we have two duplicate rules
in
planner. But I think we should use the rule
PushProjectIntoTableSourceScanRule rather than
PushComputedColumnIntoTableSourceScanRule if we don't support watermark
push down. Compared to PushComputedColumnIntoTableSourceScanRule,
PushProjectIntoTableSourceScanRule is much lighter and we can read pruned
data from source rather than use a map function in flink.
Therefore, we think it's not a good idea to use two interfaces rather
than
one.
*New Proposal*
First of all, let us address some background when pushing watermarks into
table source scan. There are two structures that we need to consider. We
list two examples below for discussion.
structure 1:LogicalWatermarkAssigner(rowtime=[c], watermark=[-($2,
5000:INTERVAL SECOND)])+- LogicalTableScan(table=[[default_catalog,
default_database, MyTable]])structure
2:LogicalWatermarkAssigner(rowtime=[d], watermark=[-($3, 5000:INTERVAL
SECOND)])+- LogicalProject(a=[$0], b=[$1], c=[$2], d=[+($2, 5000:INTERVAL
SECOND)]) +- LogicalTableScan(table=[[default_catalog, default_database,
MyTable]])
As we can see, structure 2 is much more complicated than structure 1. For
structure 1, we can use the row from table scan to generate watermarks
directly. But for structure 2, we need to calculate the rowtime
expression
in LogicalProject and use the result of calculation to generate
watermarks. Considering that WatermarkStrategy has the ability to extract
timestamp from row, we have a proposal to push only WatermarkStrategy to
scan.
Push WatermarkStrategy to Scan
In this interface, we will only push WatermarkStrategy to
DynamicTableSource.
public interface SupportsWatermarkPushDown { void
applyWatermark(WatermarkStrategy<RowData> watermarkStrategy); }
The advantage of the new api is that it's very simple for the developers
of Connector. They only need to take WatermarkStrategy into consideration
and don't need to deal with other infos such as ComputedColumnConverter
in SupportsComputedColumnPushDown. But it also has one disadvantage that
it needs to calculate the rowtime expression again in LogicalProjection
because we don't build a new row in scan to store the calculated
timestamp.
However, we can replace the calculation of rowtime in LogicalProjection
with a reference to eliminate duplicate calculation, which will use the
StreamRecord's getter to read the timestamp that is calculated before.
But
this optimization still has one limitation that it relies on computed
columns are not defined on other computed columns. For nested computed
columns, we have no place to save the intermediate result.
But we still have a problem that when we push an udf into the source, we
need a context as powerful as FunctionContext to open the udf. But the
current WatermarkGeneratorSupplier.Context only supports method
getMetricGroup and misses methods getCachedFile, getJobParameter and g
etExternalResourceInfos, which means we can't convert the
WatermarkGeneratorSupplier.Context to FunctionContext safely. Considering
that the udf is only used to generate watermark, we suggest to throw
UnsupportedException when invoking the methods exist in FunctionContext
but don't exist in WatermarkGeneratorSupplier.Context. But we have to
admit that there are risks in doing so because we have no promise that
the
udf will not invoke these methods.
*Summary*
We have addressed the whole problem and solution in detail. As a
conclusion, I think the new interface avoids the problems we mentioned
before. It has a clear definition as its name tells and has nothing in
common with SupportProjectionPushDown. As for its disadvantage, I think
it's acceptable to calculate the rowtime column twice and we also have a
plan to improve its efficiency as a follow up if it brings some
performance
problems.
