You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by Yuval Itzchakov <yu...@gmail.com> on 2021/03/04 17:52:44 UTC
Re: LocalWatermarkAssigner causes predicate pushdown to be skipped
Bumping this up again, would appreciate any help if anyone is familiar with
the blink planner.
Thanks,
Yuval.
On Fri, Feb 26, 2021, 18:53 Yuval Itzchakov <yu...@gmail.com> wrote:
> Hi Jark,
> Would appreciate your help with this.
>
> On Wed, Feb 24, 2021 at 12:09 PM Roman Khachatryan <ro...@apache.org>
> wrote:
>
>> Hi Yuval,
>>
>> I'm not familiar with the Blink planner but probably Jark can help.
>>
>> Regards,
>> Roman
>>
>>
>> On Sun, Feb 21, 2021 at 6:52 PM Yuval Itzchakov <yu...@gmail.com>
>> wrote:
>>
>>> Update: When I don't set the watermark explicitly on the TableSchema,
>>> `applyWatermarkStrategy` never gets called on my ScanTableSource, which
>>> does make sense. But now the question is what should be done? This feels a
>>> bit unintuitive.
>>>
>>> On Sun, Feb 21, 2021 at 7:09 PM Yuval Itzchakov <yu...@gmail.com>
>>> wrote:
>>>
>>>> Hi,
>>>> Flink 1.12.1, Blink Planner, Scala 2.12
>>>>
>>>> I have the following logical plan:
>>>>
>>>> LogicalSink(table=[default_catalog.default_database.table],
>>>> fields=[bar, baz, hello_world, a, b])
>>>> +- LogicalProject(value=[$2],
>>>> bar=[CAST(CAST($0):TIMESTAMP(3)):TIMESTAMP(6)],
>>>> baz=[CAST(CAST($0):TIMESTAMP(3)):TIMESTAMP(6)],
>>>> hello_world=[null:VARCHAR(2147483647) CHARACTER SET "UTF-16LE"],
>>>> a=[null:VARCHAR(2147483647) CHARACTER SET "UTF-16LE"], b=[EMPTY_MAP()])
>>>> +- LogicalFilter(condition=[AND(=($4, _UTF-16LE'bar'), =($34,
>>>> _UTF-16LE'baz'))])
>>>> +- LogicalWatermarkAssigner(rowtime=[bar], watermark=[$0])
>>>> +- LogicalTableScan(table=[[default_catalog, default_database,
>>>> foo]])
>>>>
>>>> I have a custom source which creates a TableSchema based on an external
>>>> table. When I create the schema, I push the watermark definition to the
>>>> schema:
>>>>
>>>> [image: image.png]
>>>>
>>>> When the HepPlanner starts the optimization phase and reaches the
>>>> "PushFilterInotTableSourceScanRule", it matches on the LogicalFilter in the
>>>> definition. But then, since the RelOptRuleOperandChildPolicy is set to
>>>> "SOME", it attempts to do a full match on the child nodes. Since the rule
>>>> is defined as so:
>>>>
>>>> [image: image.png]
>>>>
>>>> The child filter fails since the immediate child of the filter is a
>>>> "LocalWatermarkAssigner", and not the "LogicalTableScan" which is the
>>>> grandchild:
>>>>
>>>> [image: image.png]
>>>>
>>>> Is this the desired behavior? Should I create the TableSchema without
>>>> the row time attribute and use "SupportsWatermarkPushdown" to generate the
>>>> watermark dynamically from the source record?
>>>>
>>>> --
>>>> Best Regards,
>>>> Yuval Itzchakov.
>>>>
>>>
>>>
>>> --
>>> Best Regards,
>>> Yuval Itzchakov.
>>>
>>
>
> --
> Best Regards,
> Yuval Itzchakov.
>
Re: LocalWatermarkAssigner causes predicate pushdown to be skipped
Posted by Shengkai Fang <fs...@gmail.com>.
Sorry for the typo...
I mean it will not take too much time.
Best,
Shengkai
Shengkai Fang <fs...@gmail.com> 于2021年3月9日周二 上午10:25写道：
> Hi, Yuval.
>
> I have opened a ticket about this[1]. But I don't think we have any
> solution to solve.
>
> Do you have time to help us to solve this? I think it will take too much
> time.
>
> [1] https://issues.apache.org/jira/browse/FLINK-21675
>
> Yuval Itzchakov <yu...@gmail.com> 于2021年3月8日周一 下午9:18写道：
>
>> Thank you Shenkai,
>> That does explain what I'm seeing.
>>
>> Jark / Shenkai - Is there any workaround to get Flink to work with push
>> watermarks and predicate pushdown until this is resolved?
>>
>> On Mon, Mar 8, 2021 at 4:54 AM Shengkai Fang <fs...@gmail.com> wrote:
>>
>>> Hi, Yuval, Jark, Timo.
>>>
>>> Currently the watermark push down happens in the logical rewrite phase
>>> but the filter push down happens in the local phase, which means the
>>> planner will first check the Filter push down and then check the watermark
>>> push down.
>>>
>>> I think we need a rule to transpose between the filter and watermark
>>> assigner or extend the filter push down rule to capture the structure that
>>> the watermark assigner is the parent of the table scan.
>>>
>>> Best,
>>> Shengkai
>>>
>>> Yuval Itzchakov <yu...@gmail.com> 于2021年3月8日周一 上午12:13写道：
>>>
>>>> Hi Jark,
>>>>
>>>> Even after implementing both, I don't see the watermark being pushed to
>>>> the tablesource in the logical plan and avoids predicate pushdown from
>>>> running.
>>>>
>>>> On Sun, Mar 7, 2021, 15:43 Jark Wu <im...@gmail.com> wrote:
>>>>
>>>>> Hi Yuval,
>>>>>
>>>>> That's correct you will always get a LogicalWatermarkAssigner if you
>>>>> assigned a watermark.
>>>>> If you implement SupportsWatermarkPushdown,
>>>>> the LogicalWatermarkAssigner will be pushed
>>>>> into TableSource, and then you can push Filter into source if source
>>>>> implement SupportsFilterPushdown.
>>>>>
>>>>> Best,
>>>>> Jark
>>>>>
>>>>> On Sat, 6 Mar 2021 at 01:16, Yuval Itzchakov <yu...@gmail.com>
>>>>> wrote:
>>>>>
>>>>>> Hi Timo,
>>>>>> After investigating this further, this is actually non related to
>>>>>> implementing SupportsWatermarkPushdown.
>>>>>>
>>>>>> Once I create a TableSchema for my custom source's RowData, and
>>>>>> assign it a watermark (see my example in the original mail), the plan will
>>>>>> always include a LogicalWatermarkAssigner. This assigner that is between
>>>>>> the LogicalTableScan and the LogicalFilter will then go on and fail the
>>>>>> HepPlanner from invoking the optimization since it requires
>>>>>> LogicalTableScan to be a direct child of LogicalFilter. Since I have
>>>>>> LogicalFilter -> LogicalWatermarkAssigner -> LogicalTableScan, this won't
>>>>>> work.
>>>>>>
>>>>>> On Fri, Mar 5, 2021 at 5:59 PM Timo Walther <tw...@apache.org>
>>>>>> wrote:
>>>>>>
>>>>>>> Hi Yuval,
>>>>>>>
>>>>>>> sorry that nobody replied earlier. Somehow your email fell through
>>>>>>> the
>>>>>>> cracks.
>>>>>>>
>>>>>>> If I understand you correctly, could would like to implement a table
>>>>>>> source that implements both `SupportsWatermarkPushDown` and
>>>>>>> `SupportsFilterPushDown`?
>>>>>>>
>>>>>>> The current behavior might be on purpose. Filters and Watermarks are
>>>>>>> not
>>>>>>> very compatible. Filtering would also mean that records (from which
>>>>>>> watermarks could be generated) are skipped. If the filter is very
>>>>>>> strict, we would not generate any new watermarks and the pipeline
>>>>>>> would
>>>>>>> stop making progress in time.
>>>>>>>
>>>>>>> Watermark push down is only necessary, if per-partition watermarks
>>>>>>> are
>>>>>>> required. Otherwise the watermarks are generated in a subsequent
>>>>>>> operator after the source. So you can still use rowtime without
>>>>>>> implementing `SupportsWatermarkPushDown` in your custom source.
>>>>>>>
>>>>>>> I will lookp in Shengkai who worked on this topic recently.
>>>>>>>
>>>>>>> Regards,
>>>>>>> Timo
>>>>>>>
>>>>>>>
>>>>>>> On 04.03.21 18:52, Yuval Itzchakov wrote:
>>>>>>> > Bumping this up again, would appreciate any help if anyone is
>>>>>>> familiar
>>>>>>> > with the blink planner.
>>>>>>> >
>>>>>>> > Thanks,
>>>>>>> > Yuval.
>>>>>>> >
>>>>>>> > On Fri, Feb 26, 2021, 18:53 Yuval Itzchakov <yuvalos@gmail.com
>>>>>>> > <ma...@gmail.com>> wrote:
>>>>>>> >
>>>>>>> > Hi Jark,
>>>>>>> > Would appreciate your help with this.
>>>>>>> >
>>>>>>> > On Wed, Feb 24, 2021 at 12:09 PM Roman Khachatryan <
>>>>>>> roman@apache.org
>>>>>>> > <ma...@apache.org>> wrote:
>>>>>>> >
>>>>>>> > Hi Yuval,
>>>>>>> >
>>>>>>> > I'm not familiar with the Blink planner but probably Jark
>>>>>>> can help.
>>>>>>> >
>>>>>>> > Regards,
>>>>>>> > Roman
>>>>>>> >
>>>>>>> >
>>>>>>> > On Sun, Feb 21, 2021 at 6:52 PM Yuval Itzchakov
>>>>>>> > <yuvalos@gmail.com <ma...@gmail.com>> wrote:
>>>>>>> >
>>>>>>> > Update: When I don't set the watermark explicitly on
>>>>>>> the
>>>>>>> > TableSchema, `applyWatermarkStrategy` never gets
>>>>>>> called on
>>>>>>> > my ScanTableSource, which does make sense. But now the
>>>>>>> > question is what should be done? This feels a bit
>>>>>>> unintuitive.
>>>>>>> >
>>>>>>> > On Sun, Feb 21, 2021 at 7:09 PM Yuval Itzchakov
>>>>>>> > <yuvalos@gmail.com <ma...@gmail.com>> wrote:
>>>>>>> >
>>>>>>> > Hi,
>>>>>>> > Flink 1.12.1, Blink Planner, Scala 2.12
>>>>>>> >
>>>>>>> > I have the following logical plan:
>>>>>>> >
>>>>>>> >
>>>>>>> LogicalSink(table=[default_catalog.default_database.table], fields=[bar,
>>>>>>> baz, hello_world, a, b])
>>>>>>> > +- LogicalProject(value=[$2],
>>>>>>> > bar=[CAST(CAST($0):TIMESTAMP(3)):TIMESTAMP(6)],
>>>>>>> > baz=[CAST(CAST($0):TIMESTAMP(3)):TIMESTAMP(6)],
>>>>>>> > hello_world=[null:VARCHAR(2147483647) CHARACTER SET
>>>>>>> > "UTF-16LE"], a=[null:VARCHAR(2147483647) CHARACTER
>>>>>>> SET
>>>>>>> > "UTF-16LE"], b=[EMPTY_MAP()])
>>>>>>> > +- LogicalFilter(condition=[AND(=($4,
>>>>>>> > _UTF-16LE'bar'), =($34, _UTF-16LE'baz'))])
>>>>>>> > +- LogicalWatermarkAssigner(rowtime=[bar],
>>>>>>> > watermark=[$0])
>>>>>>> > +-
>>>>>>> LogicalTableScan(table=[[default_catalog,
>>>>>>> > default_database, foo]])
>>>>>>> >
>>>>>>> > I have a custom source which creates a TableSchema
>>>>>>> based
>>>>>>> > on an external table. When I create the schema, I
>>>>>>> push
>>>>>>> > the watermark definition to the schema:
>>>>>>> >
>>>>>>> > image.png
>>>>>>> >
>>>>>>> > When the HepPlanner starts the optimization phase
>>>>>>> and
>>>>>>> > reaches the "PushFilterInotTableSourceScanRule", it
>>>>>>> > matches on the LogicalFilter in the definition. But
>>>>>>> > then, since the RelOptRuleOperandChildPolicy is
>>>>>>> set to
>>>>>>> > "SOME", it attempts to do a full match on the child
>>>>>>> > nodes. Since the rule is defined as so:
>>>>>>> >
>>>>>>> > image.png
>>>>>>> >
>>>>>>> > The child filter fails since the immediate child
>>>>>>> of the
>>>>>>> > filter is a "LocalWatermarkAssigner", and not the
>>>>>>> > "LogicalTableScan" which is the grandchild:
>>>>>>> >
>>>>>>> > image.png
>>>>>>> >
>>>>>>> > Is this the desired behavior? Should I create the
>>>>>>> > TableSchema without the row time attribute and use
>>>>>>> > "SupportsWatermarkPushdown" to generate the
>>>>>>> watermark
>>>>>>> > dynamically from the source record?
>>>>>>> >
>>>>>>> > --
>>>>>>> > Best Regards,
>>>>>>> > Yuval Itzchakov.
>>>>>>> >
>>>>>>> >
>>>>>>> >
>>>>>>> > --
>>>>>>> > Best Regards,
>>>>>>> > Yuval Itzchakov.
>>>>>>> >
>>>>>>> >
>>>>>>> >
>>>>>>> > --
>>>>>>> > Best Regards,
>>>>>>> > Yuval Itzchakov.
>>>>>>> >
>>>>>>>
>>>>>>>
>>>>>>
>>>>>> --
>>>>>> Best Regards,
>>>>>> Yuval Itzchakov.
>>>>>>
>>>>>
>>
>> --
>> Best Regards,
>> Yuval Itzchakov.
>>
>
Re: LocalWatermarkAssigner causes predicate pushdown to be skipped
Posted by Shengkai Fang <fs...@gmail.com>.
Hi, Yuval.
I have opened a ticket about this[1]. But I don't think we have any
solution to solve.
Do you have time to help us to solve this? I think it will take too much
time.
[1] https://issues.apache.org/jira/browse/FLINK-21675
Yuval Itzchakov <yu...@gmail.com> 于2021年3月8日周一 下午9:18写道：
> Thank you Shenkai,
> That does explain what I'm seeing.
>
> Jark / Shenkai - Is there any workaround to get Flink to work with push
> watermarks and predicate pushdown until this is resolved?
>
> On Mon, Mar 8, 2021 at 4:54 AM Shengkai Fang <fs...@gmail.com> wrote:
>
>> Hi, Yuval, Jark, Timo.
>>
>> Currently the watermark push down happens in the logical rewrite phase
>> but the filter push down happens in the local phase, which means the
>> planner will first check the Filter push down and then check the watermark
>> push down.
>>
>> I think we need a rule to transpose between the filter and watermark
>> assigner or extend the filter push down rule to capture the structure that
>> the watermark assigner is the parent of the table scan.
>>
>> Best,
>> Shengkai
>>
>> Yuval Itzchakov <yu...@gmail.com> 于2021年3月8日周一 上午12:13写道：
>>
>>> Hi Jark,
>>>
>>> Even after implementing both, I don't see the watermark being pushed to
>>> the tablesource in the logical plan and avoids predicate pushdown from
>>> running.
>>>
>>> On Sun, Mar 7, 2021, 15:43 Jark Wu <im...@gmail.com> wrote:
>>>
>>>> Hi Yuval,
>>>>
>>>> That's correct you will always get a LogicalWatermarkAssigner if you
>>>> assigned a watermark.
>>>> If you implement SupportsWatermarkPushdown,
>>>> the LogicalWatermarkAssigner will be pushed
>>>> into TableSource, and then you can push Filter into source if source
>>>> implement SupportsFilterPushdown.
>>>>
>>>> Best,
>>>> Jark
>>>>
>>>> On Sat, 6 Mar 2021 at 01:16, Yuval Itzchakov <yu...@gmail.com> wrote:
>>>>
>>>>> Hi Timo,
>>>>> After investigating this further, this is actually non related to
>>>>> implementing SupportsWatermarkPushdown.
>>>>>
>>>>> Once I create a TableSchema for my custom source's RowData, and assign
>>>>> it a watermark (see my example in the original mail), the plan will always
>>>>> include a LogicalWatermarkAssigner. This assigner that is between the
>>>>> LogicalTableScan and the LogicalFilter will then go on and fail the
>>>>> HepPlanner from invoking the optimization since it requires
>>>>> LogicalTableScan to be a direct child of LogicalFilter. Since I have
>>>>> LogicalFilter -> LogicalWatermarkAssigner -> LogicalTableScan, this won't
>>>>> work.
>>>>>
>>>>> On Fri, Mar 5, 2021 at 5:59 PM Timo Walther <tw...@apache.org>
>>>>> wrote:
>>>>>
>>>>>> Hi Yuval,
>>>>>>
>>>>>> sorry that nobody replied earlier. Somehow your email fell through
>>>>>> the
>>>>>> cracks.
>>>>>>
>>>>>> If I understand you correctly, could would like to implement a table
>>>>>> source that implements both `SupportsWatermarkPushDown` and
>>>>>> `SupportsFilterPushDown`?
>>>>>>
>>>>>> The current behavior might be on purpose. Filters and Watermarks are
>>>>>> not
>>>>>> very compatible. Filtering would also mean that records (from which
>>>>>> watermarks could be generated) are skipped. If the filter is very
>>>>>> strict, we would not generate any new watermarks and the pipeline
>>>>>> would
>>>>>> stop making progress in time.
>>>>>>
>>>>>> Watermark push down is only necessary, if per-partition watermarks
>>>>>> are
>>>>>> required. Otherwise the watermarks are generated in a subsequent
>>>>>> operator after the source. So you can still use rowtime without
>>>>>> implementing `SupportsWatermarkPushDown` in your custom source.
>>>>>>
>>>>>> I will lookp in Shengkai who worked on this topic recently.
>>>>>>
>>>>>> Regards,
>>>>>> Timo
>>>>>>
>>>>>>
>>>>>> On 04.03.21 18:52, Yuval Itzchakov wrote:
>>>>>> > Bumping this up again, would appreciate any help if anyone is
>>>>>> familiar
>>>>>> > with the blink planner.
>>>>>> >
>>>>>> > Thanks,
>>>>>> > Yuval.
>>>>>> >
>>>>>> > On Fri, Feb 26, 2021, 18:53 Yuval Itzchakov <yuvalos@gmail.com
>>>>>> > <ma...@gmail.com>> wrote:
>>>>>> >
>>>>>> > Hi Jark,
>>>>>> > Would appreciate your help with this.
>>>>>> >
>>>>>> > On Wed, Feb 24, 2021 at 12:09 PM Roman Khachatryan <
>>>>>> roman@apache.org
>>>>>> > <ma...@apache.org>> wrote:
>>>>>> >
>>>>>> > Hi Yuval,
>>>>>> >
>>>>>> > I'm not familiar with the Blink planner but probably Jark
>>>>>> can help.
>>>>>> >
>>>>>> > Regards,
>>>>>> > Roman
>>>>>> >
>>>>>> >
>>>>>> > On Sun, Feb 21, 2021 at 6:52 PM Yuval Itzchakov
>>>>>> > <yuvalos@gmail.com <ma...@gmail.com>> wrote:
>>>>>> >
>>>>>> > Update: When I don't set the watermark explicitly on the
>>>>>> > TableSchema, `applyWatermarkStrategy` never gets called
>>>>>> on
>>>>>> > my ScanTableSource, which does make sense. But now the
>>>>>> > question is what should be done? This feels a bit
>>>>>> unintuitive.
>>>>>> >
>>>>>> > On Sun, Feb 21, 2021 at 7:09 PM Yuval Itzchakov
>>>>>> > <yuvalos@gmail.com <ma...@gmail.com>> wrote:
>>>>>> >
>>>>>> > Hi,
>>>>>> > Flink 1.12.1, Blink Planner, Scala 2.12
>>>>>> >
>>>>>> > I have the following logical plan:
>>>>>> >
>>>>>> >
>>>>>> LogicalSink(table=[default_catalog.default_database.table], fields=[bar,
>>>>>> baz, hello_world, a, b])
>>>>>> > +- LogicalProject(value=[$2],
>>>>>> > bar=[CAST(CAST($0):TIMESTAMP(3)):TIMESTAMP(6)],
>>>>>> > baz=[CAST(CAST($0):TIMESTAMP(3)):TIMESTAMP(6)],
>>>>>> > hello_world=[null:VARCHAR(2147483647) CHARACTER SET
>>>>>> > "UTF-16LE"], a=[null:VARCHAR(2147483647) CHARACTER
>>>>>> SET
>>>>>> > "UTF-16LE"], b=[EMPTY_MAP()])
>>>>>> > +- LogicalFilter(condition=[AND(=($4,
>>>>>> > _UTF-16LE'bar'), =($34, _UTF-16LE'baz'))])
>>>>>> > +- LogicalWatermarkAssigner(rowtime=[bar],
>>>>>> > watermark=[$0])
>>>>>> > +-
>>>>>> LogicalTableScan(table=[[default_catalog,
>>>>>> > default_database, foo]])
>>>>>> >
>>>>>> > I have a custom source which creates a TableSchema
>>>>>> based
>>>>>> > on an external table. When I create the schema, I
>>>>>> push
>>>>>> > the watermark definition to the schema:
>>>>>> >
>>>>>> > image.png
>>>>>> >
>>>>>> > When the HepPlanner starts the optimization phase
>>>>>> and
>>>>>> > reaches the "PushFilterInotTableSourceScanRule", it
>>>>>> > matches on the LogicalFilter in the definition. But
>>>>>> > then, since the RelOptRuleOperandChildPolicy is set
>>>>>> to
>>>>>> > "SOME", it attempts to do a full match on the child
>>>>>> > nodes. Since the rule is defined as so:
>>>>>> >
>>>>>> > image.png
>>>>>> >
>>>>>> > The child filter fails since the immediate child of
>>>>>> the
>>>>>> > filter is a "LocalWatermarkAssigner", and not the
>>>>>> > "LogicalTableScan" which is the grandchild:
>>>>>> >
>>>>>> > image.png
>>>>>> >
>>>>>> > Is this the desired behavior? Should I create the
>>>>>> > TableSchema without the row time attribute and use
>>>>>> > "SupportsWatermarkPushdown" to generate the
>>>>>> watermark
>>>>>> > dynamically from the source record?
>>>>>> >
>>>>>> > --
>>>>>> > Best Regards,
>>>>>> > Yuval Itzchakov.
>>>>>> >
>>>>>> >
>>>>>> >
>>>>>> > --
>>>>>> > Best Regards,
>>>>>> > Yuval Itzchakov.
>>>>>> >
>>>>>> >
>>>>>> >
>>>>>> > --
>>>>>> > Best Regards,
>>>>>> > Yuval Itzchakov.
>>>>>> >
>>>>>>
>>>>>>
>>>>>
>>>>> --
>>>>> Best Regards,
>>>>> Yuval Itzchakov.
>>>>>
>>>>
>
> --
> Best Regards,
> Yuval Itzchakov.
>
Re: LocalWatermarkAssigner causes predicate pushdown to be skipped
Posted by Yuval Itzchakov <yu...@gmail.com>.
Thank you Shenkai,
That does explain what I'm seeing.
Jark / Shenkai - Is there any workaround to get Flink to work with push
watermarks and predicate pushdown until this is resolved?
On Mon, Mar 8, 2021 at 4:54 AM Shengkai Fang <fs...@gmail.com> wrote:
> Hi, Yuval, Jark, Timo.
>
> Currently the watermark push down happens in the logical rewrite phase but
> the filter push down happens in the local phase, which means the planner
> will first check the Filter push down and then check the watermark push
> down.
>
> I think we need a rule to transpose between the filter and watermark
> assigner or extend the filter push down rule to capture the structure that
> the watermark assigner is the parent of the table scan.
>
> Best,
> Shengkai
>
> Yuval Itzchakov <yu...@gmail.com> 于2021年3月8日周一 上午12:13写道：
>
>> Hi Jark,
>>
>> Even after implementing both, I don't see the watermark being pushed to
>> the tablesource in the logical plan and avoids predicate pushdown from
>> running.
>>
>> On Sun, Mar 7, 2021, 15:43 Jark Wu <im...@gmail.com> wrote:
>>
>>> Hi Yuval,
>>>
>>> That's correct you will always get a LogicalWatermarkAssigner if you
>>> assigned a watermark.
>>> If you implement SupportsWatermarkPushdown, the LogicalWatermarkAssigner
>>> will be pushed
>>> into TableSource, and then you can push Filter into source if source
>>> implement SupportsFilterPushdown.
>>>
>>> Best,
>>> Jark
>>>
>>> On Sat, 6 Mar 2021 at 01:16, Yuval Itzchakov <yu...@gmail.com> wrote:
>>>
>>>> Hi Timo,
>>>> After investigating this further, this is actually non related to
>>>> implementing SupportsWatermarkPushdown.
>>>>
>>>> Once I create a TableSchema for my custom source's RowData, and assign
>>>> it a watermark (see my example in the original mail), the plan will always
>>>> include a LogicalWatermarkAssigner. This assigner that is between the
>>>> LogicalTableScan and the LogicalFilter will then go on and fail the
>>>> HepPlanner from invoking the optimization since it requires
>>>> LogicalTableScan to be a direct child of LogicalFilter. Since I have
>>>> LogicalFilter -> LogicalWatermarkAssigner -> LogicalTableScan, this won't
>>>> work.
>>>>
>>>> On Fri, Mar 5, 2021 at 5:59 PM Timo Walther <tw...@apache.org> wrote:
>>>>
>>>>> Hi Yuval,
>>>>>
>>>>> sorry that nobody replied earlier. Somehow your email fell through the
>>>>> cracks.
>>>>>
>>>>> If I understand you correctly, could would like to implement a table
>>>>> source that implements both `SupportsWatermarkPushDown` and
>>>>> `SupportsFilterPushDown`?
>>>>>
>>>>> The current behavior might be on purpose. Filters and Watermarks are
>>>>> not
>>>>> very compatible. Filtering would also mean that records (from which
>>>>> watermarks could be generated) are skipped. If the filter is very
>>>>> strict, we would not generate any new watermarks and the pipeline
>>>>> would
>>>>> stop making progress in time.
>>>>>
>>>>> Watermark push down is only necessary, if per-partition watermarks are
>>>>> required. Otherwise the watermarks are generated in a subsequent
>>>>> operator after the source. So you can still use rowtime without
>>>>> implementing `SupportsWatermarkPushDown` in your custom source.
>>>>>
>>>>> I will lookp in Shengkai who worked on this topic recently.
>>>>>
>>>>> Regards,
>>>>> Timo
>>>>>
>>>>>
>>>>> On 04.03.21 18:52, Yuval Itzchakov wrote:
>>>>> > Bumping this up again, would appreciate any help if anyone is
>>>>> familiar
>>>>> > with the blink planner.
>>>>> >
>>>>> > Thanks,
>>>>> > Yuval.
>>>>> >
>>>>> > On Fri, Feb 26, 2021, 18:53 Yuval Itzchakov <yuvalos@gmail.com
>>>>> > <ma...@gmail.com>> wrote:
>>>>> >
>>>>> > Hi Jark,
>>>>> > Would appreciate your help with this.
>>>>> >
>>>>> > On Wed, Feb 24, 2021 at 12:09 PM Roman Khachatryan <
>>>>> roman@apache.org
>>>>> > <ma...@apache.org>> wrote:
>>>>> >
>>>>> > Hi Yuval,
>>>>> >
>>>>> > I'm not familiar with the Blink planner but probably Jark
>>>>> can help.
>>>>> >
>>>>> > Regards,
>>>>> > Roman
>>>>> >
>>>>> >
>>>>> > On Sun, Feb 21, 2021 at 6:52 PM Yuval Itzchakov
>>>>> > <yuvalos@gmail.com <ma...@gmail.com>> wrote:
>>>>> >
>>>>> > Update: When I don't set the watermark explicitly on the
>>>>> > TableSchema, `applyWatermarkStrategy` never gets called
>>>>> on
>>>>> > my ScanTableSource, which does make sense. But now the
>>>>> > question is what should be done? This feels a bit
>>>>> unintuitive.
>>>>> >
>>>>> > On Sun, Feb 21, 2021 at 7:09 PM Yuval Itzchakov
>>>>> > <yuvalos@gmail.com <ma...@gmail.com>> wrote:
>>>>> >
>>>>> > Hi,
>>>>> > Flink 1.12.1, Blink Planner, Scala 2.12
>>>>> >
>>>>> > I have the following logical plan:
>>>>> >
>>>>> >
>>>>> LogicalSink(table=[default_catalog.default_database.table], fields=[bar,
>>>>> baz, hello_world, a, b])
>>>>> > +- LogicalProject(value=[$2],
>>>>> > bar=[CAST(CAST($0):TIMESTAMP(3)):TIMESTAMP(6)],
>>>>> > baz=[CAST(CAST($0):TIMESTAMP(3)):TIMESTAMP(6)],
>>>>> > hello_world=[null:VARCHAR(2147483647) CHARACTER SET
>>>>> > "UTF-16LE"], a=[null:VARCHAR(2147483647) CHARACTER
>>>>> SET
>>>>> > "UTF-16LE"], b=[EMPTY_MAP()])
>>>>> > +- LogicalFilter(condition=[AND(=($4,
>>>>> > _UTF-16LE'bar'), =($34, _UTF-16LE'baz'))])
>>>>> > +- LogicalWatermarkAssigner(rowtime=[bar],
>>>>> > watermark=[$0])
>>>>> > +-
>>>>> LogicalTableScan(table=[[default_catalog,
>>>>> > default_database, foo]])
>>>>> >
>>>>> > I have a custom source which creates a TableSchema
>>>>> based
>>>>> > on an external table. When I create the schema, I
>>>>> push
>>>>> > the watermark definition to the schema:
>>>>> >
>>>>> > image.png
>>>>> >
>>>>> > When the HepPlanner starts the optimization phase and
>>>>> > reaches the "PushFilterInotTableSourceScanRule", it
>>>>> > matches on the LogicalFilter in the definition. But
>>>>> > then, since the RelOptRuleOperandChildPolicy is set
>>>>> to
>>>>> > "SOME", it attempts to do a full match on the child
>>>>> > nodes. Since the rule is defined as so:
>>>>> >
>>>>> > image.png
>>>>> >
>>>>> > The child filter fails since the immediate child of
>>>>> the
>>>>> > filter is a "LocalWatermarkAssigner", and not the
>>>>> > "LogicalTableScan" which is the grandchild:
>>>>> >
>>>>> > image.png
>>>>> >
>>>>> > Is this the desired behavior? Should I create the
>>>>> > TableSchema without the row time attribute and use
>>>>> > "SupportsWatermarkPushdown" to generate the watermark
>>>>> > dynamically from the source record?
>>>>> >
>>>>> > --
>>>>> > Best Regards,
>>>>> > Yuval Itzchakov.
>>>>> >
>>>>> >
>>>>> >
>>>>> > --
>>>>> > Best Regards,
>>>>> > Yuval Itzchakov.
>>>>> >
>>>>> >
>>>>> >
>>>>> > --
>>>>> > Best Regards,
>>>>> > Yuval Itzchakov.
>>>>> >
>>>>>
>>>>>
>>>>
>>>> --
>>>> Best Regards,
>>>> Yuval Itzchakov.
>>>>
>>>
--
Best Regards,
Yuval Itzchakov.
Re: LocalWatermarkAssigner causes predicate pushdown to be skipped
Posted by Shengkai Fang <fs...@gmail.com>.
Hi, Yuval, Jark, Timo.
Currently the watermark push down happens in the logical rewrite phase but
the filter push down happens in the local phase, which means the planner
will first check the Filter push down and then check the watermark push
down.
I think we need a rule to transpose between the filter and watermark
assigner or extend the filter push down rule to capture the structure that
the watermark assigner is the parent of the table scan.
Best,
Shengkai
Yuval Itzchakov <yu...@gmail.com> 于2021年3月8日周一 上午12:13写道：
> Hi Jark,
>
> Even after implementing both, I don't see the watermark being pushed to
> the tablesource in the logical plan and avoids predicate pushdown from
> running.
>
> On Sun, Mar 7, 2021, 15:43 Jark Wu <im...@gmail.com> wrote:
>
>> Hi Yuval,
>>
>> That's correct you will always get a LogicalWatermarkAssigner if you
>> assigned a watermark.
>> If you implement SupportsWatermarkPushdown, the LogicalWatermarkAssigner
>> will be pushed
>> into TableSource, and then you can push Filter into source if source
>> implement SupportsFilterPushdown.
>>
>> Best,
>> Jark
>>
>> On Sat, 6 Mar 2021 at 01:16, Yuval Itzchakov <yu...@gmail.com> wrote:
>>
>>> Hi Timo,
>>> After investigating this further, this is actually non related to
>>> implementing SupportsWatermarkPushdown.
>>>
>>> Once I create a TableSchema for my custom source's RowData, and assign
>>> it a watermark (see my example in the original mail), the plan will always
>>> include a LogicalWatermarkAssigner. This assigner that is between the
>>> LogicalTableScan and the LogicalFilter will then go on and fail the
>>> HepPlanner from invoking the optimization since it requires
>>> LogicalTableScan to be a direct child of LogicalFilter. Since I have
>>> LogicalFilter -> LogicalWatermarkAssigner -> LogicalTableScan, this won't
>>> work.
>>>
>>> On Fri, Mar 5, 2021 at 5:59 PM Timo Walther <tw...@apache.org> wrote:
>>>
>>>> Hi Yuval,
>>>>
>>>> sorry that nobody replied earlier. Somehow your email fell through the
>>>> cracks.
>>>>
>>>> If I understand you correctly, could would like to implement a table
>>>> source that implements both `SupportsWatermarkPushDown` and
>>>> `SupportsFilterPushDown`?
>>>>
>>>> The current behavior might be on purpose. Filters and Watermarks are
>>>> not
>>>> very compatible. Filtering would also mean that records (from which
>>>> watermarks could be generated) are skipped. If the filter is very
>>>> strict, we would not generate any new watermarks and the pipeline would
>>>> stop making progress in time.
>>>>
>>>> Watermark push down is only necessary, if per-partition watermarks are
>>>> required. Otherwise the watermarks are generated in a subsequent
>>>> operator after the source. So you can still use rowtime without
>>>> implementing `SupportsWatermarkPushDown` in your custom source.
>>>>
>>>> I will lookp in Shengkai who worked on this topic recently.
>>>>
>>>> Regards,
>>>> Timo
>>>>
>>>>
>>>> On 04.03.21 18:52, Yuval Itzchakov wrote:
>>>> > Bumping this up again, would appreciate any help if anyone is
>>>> familiar
>>>> > with the blink planner.
>>>> >
>>>> > Thanks,
>>>> > Yuval.
>>>> >
>>>> > On Fri, Feb 26, 2021, 18:53 Yuval Itzchakov <yuvalos@gmail.com
>>>> > <ma...@gmail.com>> wrote:
>>>> >
>>>> > Hi Jark,
>>>> > Would appreciate your help with this.
>>>> >
>>>> > On Wed, Feb 24, 2021 at 12:09 PM Roman Khachatryan <
>>>> roman@apache.org
>>>> > <ma...@apache.org>> wrote:
>>>> >
>>>> > Hi Yuval,
>>>> >
>>>> > I'm not familiar with the Blink planner but probably Jark can
>>>> help.
>>>> >
>>>> > Regards,
>>>> > Roman
>>>> >
>>>> >
>>>> > On Sun, Feb 21, 2021 at 6:52 PM Yuval Itzchakov
>>>> > <yuvalos@gmail.com <ma...@gmail.com>> wrote:
>>>> >
>>>> > Update: When I don't set the watermark explicitly on the
>>>> > TableSchema, `applyWatermarkStrategy` never gets called on
>>>> > my ScanTableSource, which does make sense. But now the
>>>> > question is what should be done? This feels a bit
>>>> unintuitive.
>>>> >
>>>> > On Sun, Feb 21, 2021 at 7:09 PM Yuval Itzchakov
>>>> > <yuvalos@gmail.com <ma...@gmail.com>> wrote:
>>>> >
>>>> > Hi,
>>>> > Flink 1.12.1, Blink Planner, Scala 2.12
>>>> >
>>>> > I have the following logical plan:
>>>> >
>>>> >
>>>> LogicalSink(table=[default_catalog.default_database.table], fields=[bar,
>>>> baz, hello_world, a, b])
>>>> > +- LogicalProject(value=[$2],
>>>> > bar=[CAST(CAST($0):TIMESTAMP(3)):TIMESTAMP(6)],
>>>> > baz=[CAST(CAST($0):TIMESTAMP(3)):TIMESTAMP(6)],
>>>> > hello_world=[null:VARCHAR(2147483647) CHARACTER SET
>>>> > "UTF-16LE"], a=[null:VARCHAR(2147483647) CHARACTER SET
>>>> > "UTF-16LE"], b=[EMPTY_MAP()])
>>>> > +- LogicalFilter(condition=[AND(=($4,
>>>> > _UTF-16LE'bar'), =($34, _UTF-16LE'baz'))])
>>>> > +- LogicalWatermarkAssigner(rowtime=[bar],
>>>> > watermark=[$0])
>>>> > +- LogicalTableScan(table=[[default_catalog,
>>>> > default_database, foo]])
>>>> >
>>>> > I have a custom source which creates a TableSchema
>>>> based
>>>> > on an external table. When I create the schema, I push
>>>> > the watermark definition to the schema:
>>>> >
>>>> > image.png
>>>> >
>>>> > When the HepPlanner starts the optimization phase and
>>>> > reaches the "PushFilterInotTableSourceScanRule", it
>>>> > matches on the LogicalFilter in the definition. But
>>>> > then, since the RelOptRuleOperandChildPolicy is set to
>>>> > "SOME", it attempts to do a full match on the child
>>>> > nodes. Since the rule is defined as so:
>>>> >
>>>> > image.png
>>>> >
>>>> > The child filter fails since the immediate child of
>>>> the
>>>> > filter is a "LocalWatermarkAssigner", and not the
>>>> > "LogicalTableScan" which is the grandchild:
>>>> >
>>>> > image.png
>>>> >
>>>> > Is this the desired behavior? Should I create the
>>>> > TableSchema without the row time attribute and use
>>>> > "SupportsWatermarkPushdown" to generate the watermark
>>>> > dynamically from the source record?
>>>> >
>>>> > --
>>>> > Best Regards,
>>>> > Yuval Itzchakov.
>>>> >
>>>> >
>>>> >
>>>> > --
>>>> > Best Regards,
>>>> > Yuval Itzchakov.
>>>> >
>>>> >
>>>> >
>>>> > --
>>>> > Best Regards,
>>>> > Yuval Itzchakov.
>>>> >
>>>>
>>>>
>>>
>>> --
>>> Best Regards,
>>> Yuval Itzchakov.
>>>
>>
Re: LocalWatermarkAssigner causes predicate pushdown to be skipped
Posted by Yuval Itzchakov <yu...@gmail.com>.
Hi Jark,
Even after implementing both, I don't see the watermark being pushed to the
tablesource in the logical plan and avoids predicate pushdown from running.
On Sun, Mar 7, 2021, 15:43 Jark Wu <im...@gmail.com> wrote:
> Hi Yuval,
>
> That's correct you will always get a LogicalWatermarkAssigner if you
> assigned a watermark.
> If you implement SupportsWatermarkPushdown, the LogicalWatermarkAssigner
> will be pushed
> into TableSource, and then you can push Filter into source if source
> implement SupportsFilterPushdown.
>
> Best,
> Jark
>
> On Sat, 6 Mar 2021 at 01:16, Yuval Itzchakov <yu...@gmail.com> wrote:
>
>> Hi Timo,
>> After investigating this further, this is actually non related to
>> implementing SupportsWatermarkPushdown.
>>
>> Once I create a TableSchema for my custom source's RowData, and assign it
>> a watermark (see my example in the original mail), the plan will always
>> include a LogicalWatermarkAssigner. This assigner that is between the
>> LogicalTableScan and the LogicalFilter will then go on and fail the
>> HepPlanner from invoking the optimization since it requires
>> LogicalTableScan to be a direct child of LogicalFilter. Since I have
>> LogicalFilter -> LogicalWatermarkAssigner -> LogicalTableScan, this won't
>> work.
>>
>> On Fri, Mar 5, 2021 at 5:59 PM Timo Walther <tw...@apache.org> wrote:
>>
>>> Hi Yuval,
>>>
>>> sorry that nobody replied earlier. Somehow your email fell through the
>>> cracks.
>>>
>>> If I understand you correctly, could would like to implement a table
>>> source that implements both `SupportsWatermarkPushDown` and
>>> `SupportsFilterPushDown`?
>>>
>>> The current behavior might be on purpose. Filters and Watermarks are not
>>> very compatible. Filtering would also mean that records (from which
>>> watermarks could be generated) are skipped. If the filter is very
>>> strict, we would not generate any new watermarks and the pipeline would
>>> stop making progress in time.
>>>
>>> Watermark push down is only necessary, if per-partition watermarks are
>>> required. Otherwise the watermarks are generated in a subsequent
>>> operator after the source. So you can still use rowtime without
>>> implementing `SupportsWatermarkPushDown` in your custom source.
>>>
>>> I will lookp in Shengkai who worked on this topic recently.
>>>
>>> Regards,
>>> Timo
>>>
>>>
>>> On 04.03.21 18:52, Yuval Itzchakov wrote:
>>> > Bumping this up again, would appreciate any help if anyone is familiar
>>> > with the blink planner.
>>> >
>>> > Thanks,
>>> > Yuval.
>>> >
>>> > On Fri, Feb 26, 2021, 18:53 Yuval Itzchakov <yuvalos@gmail.com
>>> > <ma...@gmail.com>> wrote:
>>> >
>>> > Hi Jark,
>>> > Would appreciate your help with this.
>>> >
>>> > On Wed, Feb 24, 2021 at 12:09 PM Roman Khachatryan <
>>> roman@apache.org
>>> > <ma...@apache.org>> wrote:
>>> >
>>> > Hi Yuval,
>>> >
>>> > I'm not familiar with the Blink planner but probably Jark can
>>> help.
>>> >
>>> > Regards,
>>> > Roman
>>> >
>>> >
>>> > On Sun, Feb 21, 2021 at 6:52 PM Yuval Itzchakov
>>> > <yuvalos@gmail.com <ma...@gmail.com>> wrote:
>>> >
>>> > Update: When I don't set the watermark explicitly on the
>>> > TableSchema, `applyWatermarkStrategy` never gets called on
>>> > my ScanTableSource, which does make sense. But now the
>>> > question is what should be done? This feels a bit
>>> unintuitive.
>>> >
>>> > On Sun, Feb 21, 2021 at 7:09 PM Yuval Itzchakov
>>> > <yuvalos@gmail.com <ma...@gmail.com>> wrote:
>>> >
>>> > Hi,
>>> > Flink 1.12.1, Blink Planner, Scala 2.12
>>> >
>>> > I have the following logical plan:
>>> >
>>> >
>>> LogicalSink(table=[default_catalog.default_database.table], fields=[bar,
>>> baz, hello_world, a, b])
>>> > +- LogicalProject(value=[$2],
>>> > bar=[CAST(CAST($0):TIMESTAMP(3)):TIMESTAMP(6)],
>>> > baz=[CAST(CAST($0):TIMESTAMP(3)):TIMESTAMP(6)],
>>> > hello_world=[null:VARCHAR(2147483647) CHARACTER SET
>>> > "UTF-16LE"], a=[null:VARCHAR(2147483647) CHARACTER SET
>>> > "UTF-16LE"], b=[EMPTY_MAP()])
>>> > +- LogicalFilter(condition=[AND(=($4,
>>> > _UTF-16LE'bar'), =($34, _UTF-16LE'baz'))])
>>> > +- LogicalWatermarkAssigner(rowtime=[bar],
>>> > watermark=[$0])
>>> > +- LogicalTableScan(table=[[default_catalog,
>>> > default_database, foo]])
>>> >
>>> > I have a custom source which creates a TableSchema
>>> based
>>> > on an external table. When I create the schema, I push
>>> > the watermark definition to the schema:
>>> >
>>> > image.png
>>> >
>>> > When the HepPlanner starts the optimization phase and
>>> > reaches the "PushFilterInotTableSourceScanRule", it
>>> > matches on the LogicalFilter in the definition. But
>>> > then, since the RelOptRuleOperandChildPolicy is set to
>>> > "SOME", it attempts to do a full match on the child
>>> > nodes. Since the rule is defined as so:
>>> >
>>> > image.png
>>> >
>>> > The child filter fails since the immediate child of the
>>> > filter is a "LocalWatermarkAssigner", and not the
>>> > "LogicalTableScan" which is the grandchild:
>>> >
>>> > image.png
>>> >
>>> > Is this the desired behavior? Should I create the
>>> > TableSchema without the row time attribute and use
>>> > "SupportsWatermarkPushdown" to generate the watermark
>>> > dynamically from the source record?
>>> >
>>> > --
>>> > Best Regards,
>>> > Yuval Itzchakov.
>>> >
>>> >
>>> >
>>> > --
>>> > Best Regards,
>>> > Yuval Itzchakov.
>>> >
>>> >
>>> >
>>> > --
>>> > Best Regards,
>>> > Yuval Itzchakov.
>>> >
>>>
>>>
>>
>> --
>> Best Regards,
>> Yuval Itzchakov.
>>
>
Re: LocalWatermarkAssigner causes predicate pushdown to be skipped
Posted by Jark Wu <im...@gmail.com>.
Hi Yuval,
That's correct you will always get a LogicalWatermarkAssigner if you
assigned a watermark.
If you implement SupportsWatermarkPushdown, the LogicalWatermarkAssigner
will be pushed
into TableSource, and then you can push Filter into source if source
implement SupportsFilterPushdown.
Best,
Jark
On Sat, 6 Mar 2021 at 01:16, Yuval Itzchakov <yu...@gmail.com> wrote:
> Hi Timo,
> After investigating this further, this is actually non related to
> implementing SupportsWatermarkPushdown.
>
> Once I create a TableSchema for my custom source's RowData, and assign it
> a watermark (see my example in the original mail), the plan will always
> include a LogicalWatermarkAssigner. This assigner that is between the
> LogicalTableScan and the LogicalFilter will then go on and fail the
> HepPlanner from invoking the optimization since it requires
> LogicalTableScan to be a direct child of LogicalFilter. Since I have
> LogicalFilter -> LogicalWatermarkAssigner -> LogicalTableScan, this won't
> work.
>
> On Fri, Mar 5, 2021 at 5:59 PM Timo Walther <tw...@apache.org> wrote:
>
>> Hi Yuval,
>>
>> sorry that nobody replied earlier. Somehow your email fell through the
>> cracks.
>>
>> If I understand you correctly, could would like to implement a table
>> source that implements both `SupportsWatermarkPushDown` and
>> `SupportsFilterPushDown`?
>>
>> The current behavior might be on purpose. Filters and Watermarks are not
>> very compatible. Filtering would also mean that records (from which
>> watermarks could be generated) are skipped. If the filter is very
>> strict, we would not generate any new watermarks and the pipeline would
>> stop making progress in time.
>>
>> Watermark push down is only necessary, if per-partition watermarks are
>> required. Otherwise the watermarks are generated in a subsequent
>> operator after the source. So you can still use rowtime without
>> implementing `SupportsWatermarkPushDown` in your custom source.
>>
>> I will lookp in Shengkai who worked on this topic recently.
>>
>> Regards,
>> Timo
>>
>>
>> On 04.03.21 18:52, Yuval Itzchakov wrote:
>> > Bumping this up again, would appreciate any help if anyone is familiar
>> > with the blink planner.
>> >
>> > Thanks,
>> > Yuval.
>> >
>> > On Fri, Feb 26, 2021, 18:53 Yuval Itzchakov <yuvalos@gmail.com
>> > <ma...@gmail.com>> wrote:
>> >
>> > Hi Jark,
>> > Would appreciate your help with this.
>> >
>> > On Wed, Feb 24, 2021 at 12:09 PM Roman Khachatryan <
>> roman@apache.org
>> > <ma...@apache.org>> wrote:
>> >
>> > Hi Yuval,
>> >
>> > I'm not familiar with the Blink planner but probably Jark can
>> help.
>> >
>> > Regards,
>> > Roman
>> >
>> >
>> > On Sun, Feb 21, 2021 at 6:52 PM Yuval Itzchakov
>> > <yuvalos@gmail.com <ma...@gmail.com>> wrote:
>> >
>> > Update: When I don't set the watermark explicitly on the
>> > TableSchema, `applyWatermarkStrategy` never gets called on
>> > my ScanTableSource, which does make sense. But now the
>> > question is what should be done? This feels a bit
>> unintuitive.
>> >
>> > On Sun, Feb 21, 2021 at 7:09 PM Yuval Itzchakov
>> > <yuvalos@gmail.com <ma...@gmail.com>> wrote:
>> >
>> > Hi,
>> > Flink 1.12.1, Blink Planner, Scala 2.12
>> >
>> > I have the following logical plan:
>> >
>> >
>> LogicalSink(table=[default_catalog.default_database.table], fields=[bar,
>> baz, hello_world, a, b])
>> > +- LogicalProject(value=[$2],
>> > bar=[CAST(CAST($0):TIMESTAMP(3)):TIMESTAMP(6)],
>> > baz=[CAST(CAST($0):TIMESTAMP(3)):TIMESTAMP(6)],
>> > hello_world=[null:VARCHAR(2147483647) CHARACTER SET
>> > "UTF-16LE"], a=[null:VARCHAR(2147483647) CHARACTER SET
>> > "UTF-16LE"], b=[EMPTY_MAP()])
>> > +- LogicalFilter(condition=[AND(=($4,
>> > _UTF-16LE'bar'), =($34, _UTF-16LE'baz'))])
>> > +- LogicalWatermarkAssigner(rowtime=[bar],
>> > watermark=[$0])
>> > +- LogicalTableScan(table=[[default_catalog,
>> > default_database, foo]])
>> >
>> > I have a custom source which creates a TableSchema based
>> > on an external table. When I create the schema, I push
>> > the watermark definition to the schema:
>> >
>> > image.png
>> >
>> > When the HepPlanner starts the optimization phase and
>> > reaches the "PushFilterInotTableSourceScanRule", it
>> > matches on the LogicalFilter in the definition. But
>> > then, since the RelOptRuleOperandChildPolicy is set to
>> > "SOME", it attempts to do a full match on the child
>> > nodes. Since the rule is defined as so:
>> >
>> > image.png
>> >
>> > The child filter fails since the immediate child of the
>> > filter is a "LocalWatermarkAssigner", and not the
>> > "LogicalTableScan" which is the grandchild:
>> >
>> > image.png
>> >
>> > Is this the desired behavior? Should I create the
>> > TableSchema without the row time attribute and use
>> > "SupportsWatermarkPushdown" to generate the watermark
>> > dynamically from the source record?
>> >
>> > --
>> > Best Regards,
>> > Yuval Itzchakov.
>> >
>> >
>> >
>> > --
>> > Best Regards,
>> > Yuval Itzchakov.
>> >
>> >
>> >
>> > --
>> > Best Regards,
>> > Yuval Itzchakov.
>> >
>>
>>
>
> --
> Best Regards,
> Yuval Itzchakov.
>
Re: LocalWatermarkAssigner causes predicate pushdown to be skipped
Posted by Yuval Itzchakov <yu...@gmail.com>.
Hi Timo,
After investigating this further, this is actually non related to
implementing SupportsWatermarkPushdown.
Once I create a TableSchema for my custom source's RowData, and assign it a
watermark (see my example in the original mail), the plan will always
include a LogicalWatermarkAssigner. This assigner that is between the
LogicalTableScan and the LogicalFilter will then go on and fail the
HepPlanner from invoking the optimization since it requires
LogicalTableScan to be a direct child of LogicalFilter. Since I have
LogicalFilter -> LogicalWatermarkAssigner -> LogicalTableScan, this won't
work.
On Fri, Mar 5, 2021 at 5:59 PM Timo Walther <tw...@apache.org> wrote:
> Hi Yuval,
>
> sorry that nobody replied earlier. Somehow your email fell through the
> cracks.
>
> If I understand you correctly, could would like to implement a table
> source that implements both `SupportsWatermarkPushDown` and
> `SupportsFilterPushDown`?
>
> The current behavior might be on purpose. Filters and Watermarks are not
> very compatible. Filtering would also mean that records (from which
> watermarks could be generated) are skipped. If the filter is very
> strict, we would not generate any new watermarks and the pipeline would
> stop making progress in time.
>
> Watermark push down is only necessary, if per-partition watermarks are
> required. Otherwise the watermarks are generated in a subsequent
> operator after the source. So you can still use rowtime without
> implementing `SupportsWatermarkPushDown` in your custom source.
>
> I will lookp in Shengkai who worked on this topic recently.
>
> Regards,
> Timo
>
>
> On 04.03.21 18:52, Yuval Itzchakov wrote:
> > Bumping this up again, would appreciate any help if anyone is familiar
> > with the blink planner.
> >
> > Thanks,
> > Yuval.
> >
> > On Fri, Feb 26, 2021, 18:53 Yuval Itzchakov <yuvalos@gmail.com
> > <ma...@gmail.com>> wrote:
> >
> > Hi Jark,
> > Would appreciate your help with this.
> >
> > On Wed, Feb 24, 2021 at 12:09 PM Roman Khachatryan <roman@apache.org
> > <ma...@apache.org>> wrote:
> >
> > Hi Yuval,
> >
> > I'm not familiar with the Blink planner but probably Jark can
> help.
> >
> > Regards,
> > Roman
> >
> >
> > On Sun, Feb 21, 2021 at 6:52 PM Yuval Itzchakov
> > <yuvalos@gmail.com <ma...@gmail.com>> wrote:
> >
> > Update: When I don't set the watermark explicitly on the
> > TableSchema, `applyWatermarkStrategy` never gets called on
> > my ScanTableSource, which does make sense. But now the
> > question is what should be done? This feels a bit
> unintuitive.
> >
> > On Sun, Feb 21, 2021 at 7:09 PM Yuval Itzchakov
> > <yuvalos@gmail.com <ma...@gmail.com>> wrote:
> >
> > Hi,
> > Flink 1.12.1, Blink Planner, Scala 2.12
> >
> > I have the following logical plan:
> >
> >
> LogicalSink(table=[default_catalog.default_database.table], fields=[bar,
> baz, hello_world, a, b])
> > +- LogicalProject(value=[$2],
> > bar=[CAST(CAST($0):TIMESTAMP(3)):TIMESTAMP(6)],
> > baz=[CAST(CAST($0):TIMESTAMP(3)):TIMESTAMP(6)],
> > hello_world=[null:VARCHAR(2147483647) CHARACTER SET
> > "UTF-16LE"], a=[null:VARCHAR(2147483647) CHARACTER SET
> > "UTF-16LE"], b=[EMPTY_MAP()])
> > +- LogicalFilter(condition=[AND(=($4,
> > _UTF-16LE'bar'), =($34, _UTF-16LE'baz'))])
> > +- LogicalWatermarkAssigner(rowtime=[bar],
> > watermark=[$0])
> > +- LogicalTableScan(table=[[default_catalog,
> > default_database, foo]])
> >
> > I have a custom source which creates a TableSchema based
> > on an external table. When I create the schema, I push
> > the watermark definition to the schema:
> >
> > image.png
> >
> > When the HepPlanner starts the optimization phase and
> > reaches the "PushFilterInotTableSourceScanRule", it
> > matches on the LogicalFilter in the definition. But
> > then, since the RelOptRuleOperandChildPolicy is set to
> > "SOME", it attempts to do a full match on the child
> > nodes. Since the rule is defined as so:
> >
> > image.png
> >
> > The child filter fails since the immediate child of the
> > filter is a "LocalWatermarkAssigner", and not the
> > "LogicalTableScan" which is the grandchild:
> >
> > image.png
> >
> > Is this the desired behavior? Should I create the
> > TableSchema without the row time attribute and use
> > "SupportsWatermarkPushdown" to generate the watermark
> > dynamically from the source record?
> >
> > --
> > Best Regards,
> > Yuval Itzchakov.
> >
> >
> >
> > --
> > Best Regards,
> > Yuval Itzchakov.
> >
> >
> >
> > --
> > Best Regards,
> > Yuval Itzchakov.
> >
>
>
--
Best Regards,
Yuval Itzchakov.
Re: LocalWatermarkAssigner causes predicate pushdown to be skipped
Posted by Timo Walther <tw...@apache.org>.
Hi Yuval,
sorry that nobody replied earlier. Somehow your email fell through the
cracks.
If I understand you correctly, could would like to implement a table
source that implements both `SupportsWatermarkPushDown` and
`SupportsFilterPushDown`?
The current behavior might be on purpose. Filters and Watermarks are not
very compatible. Filtering would also mean that records (from which
watermarks could be generated) are skipped. If the filter is very
strict, we would not generate any new watermarks and the pipeline would
stop making progress in time.
Watermark push down is only necessary, if per-partition watermarks are
required. Otherwise the watermarks are generated in a subsequent
operator after the source. So you can still use rowtime without
implementing `SupportsWatermarkPushDown` in your custom source.
I will lookp in Shengkai who worked on this topic recently.
Regards,
Timo
On 04.03.21 18:52, Yuval Itzchakov wrote:
> Bumping this up again, would appreciate any help if anyone is familiar
> with the blink planner.
>
> Thanks,
> Yuval.
>
> On Fri, Feb 26, 2021, 18:53 Yuval Itzchakov <yuvalos@gmail.com
> <ma...@gmail.com>> wrote:
>
> Hi Jark,
> Would appreciate your help with this.
>
> On Wed, Feb 24, 2021 at 12:09 PM Roman Khachatryan <roman@apache.org
> <ma...@apache.org>> wrote:
>
> Hi Yuval,
>
> I'm not familiar with the Blink planner but probably Jark can help.
>
> Regards,
> Roman
>
>
> On Sun, Feb 21, 2021 at 6:52 PM Yuval Itzchakov
> <yuvalos@gmail.com <ma...@gmail.com>> wrote:
>
> Update: When I don't set the watermark explicitly on the
> TableSchema, `applyWatermarkStrategy` never gets called on
> my ScanTableSource, which does make sense. But now the
> question is what should be done? This feels a bit unintuitive.
>
> On Sun, Feb 21, 2021 at 7:09 PM Yuval Itzchakov
> <yuvalos@gmail.com <ma...@gmail.com>> wrote:
>
> Hi,
> Flink 1.12.1, Blink Planner, Scala 2.12
>
> I have the following logical plan:
>
> LogicalSink(table=[default_catalog.default_database.table], fields=[bar, baz, hello_world, a, b])
> +- LogicalProject(value=[$2],
> bar=[CAST(CAST($0):TIMESTAMP(3)):TIMESTAMP(6)],
> baz=[CAST(CAST($0):TIMESTAMP(3)):TIMESTAMP(6)],
> hello_world=[null:VARCHAR(2147483647) CHARACTER SET
> "UTF-16LE"], a=[null:VARCHAR(2147483647) CHARACTER SET
> "UTF-16LE"], b=[EMPTY_MAP()])
> +- LogicalFilter(condition=[AND(=($4,
> _UTF-16LE'bar'), =($34, _UTF-16LE'baz'))])
> +- LogicalWatermarkAssigner(rowtime=[bar],
> watermark=[$0])
> +- LogicalTableScan(table=[[default_catalog,
> default_database, foo]])
>
> I have a custom source which creates a TableSchema based
> on an external table. When I create the schema, I push
> the watermark definition to the schema:
>
> image.png
>
> When the HepPlanner starts the optimization phase and
> reaches the "PushFilterInotTableSourceScanRule", it
> matches on the LogicalFilter in the definition. But
> then, since the RelOptRuleOperandChildPolicy is set to
> "SOME", it attempts to do a full match on the child
> nodes. Since the rule is defined as so:
>
> image.png
>
> The child filter fails since the immediate child of the
> filter is a "LocalWatermarkAssigner", and not the
> "LogicalTableScan" which is the grandchild:
>
> image.png
>
> Is this the desired behavior? Should I create the
> TableSchema without the row time attribute and use
> "SupportsWatermarkPushdown" to generate the watermark
> dynamically from the source record?
>
> --
> Best Regards,
> Yuval Itzchakov.
>
>
>
> --
> Best Regards,
> Yuval Itzchakov.
>
>
>
> --
> Best Regards,
> Yuval Itzchakov.
>