Re: [VOTE] FLIP-444: Native file copy support

[yue ma]

+1 (non-binding)

Piotr Nowojski  于2024年6月25日周二 16:59写道：

> Hi all,
>
> I would like to start a vote for the FLIP-444 [1]. The discussion thread is
> here [2].
>
> The vote will be open for at least 72.
>
> Best,
> Piotrek
>
> [1] https://cwiki.apache.org/confluence/x/rAn9EQ
> [2] https://lists.apache.org/thread/lkwmyjt2bnmvgx4qpp82rldwmtd4516c
>


-- 
Best,
Yue

[jira] [Created] (FLINK-35582) Marking ingestDB as the default recovery mode for rescaling

[Yue Ma (Jira)]

Yue Ma created FLINK-35582:
--

 Summary: Marking ingestDB as the default recovery mode for 
rescaling
 Key: FLINK-35582
 URL: https://issues.apache.org/jira/browse/FLINK-35582
 Project: Flink
  Issue Type: Sub-task
  Components: Runtime / State Backends
Affects Versions: 2.0.0
Reporter: Yue Ma
 Fix For: 2.0.0






--
This message was sent by Atlassian Jira
(v8.20.10#820010)

[jira] [Created] (FLINK-35580) Fix ingestDB recovery mode related bugs

[Yue Ma (Jira)]

Yue Ma created FLINK-35580:
--

 Summary: Fix ingestDB recovery mode related bugs
 Key: FLINK-35580
 URL: https://issues.apache.org/jira/browse/FLINK-35580
 Project: Flink
  Issue Type: Sub-task
  Components: Runtime / State Backends
Affects Versions: 2.0.0
Reporter: Yue Ma
 Fix For: 2.0.0






--
This message was sent by Atlassian Jira
(v8.20.10#820010)

[jira] [Created] (FLINK-35581) Remove comments from the code related to ingestDB

[Yue Ma (Jira)]

Yue Ma created FLINK-35581:
--

 Summary: Remove comments from the code related to ingestDB
 Key: FLINK-35581
 URL: https://issues.apache.org/jira/browse/FLINK-35581
 Project: Flink
  Issue Type: Sub-task
  Components: Runtime / State Backends
Affects Versions: 2.0.0
Reporter: Yue Ma
 Fix For: 2.0.0






--
This message was sent by Atlassian Jira
(v8.20.10#820010)

[jira] [Created] (FLINK-35579) Update the FrocksDB version in FLINK

[Yue Ma (Jira)]

Yue Ma created FLINK-35579:
--

 Summary: Update the FrocksDB version in FLINK
 Key: FLINK-35579
 URL: https://issues.apache.org/jira/browse/FLINK-35579
 Project: Flink
  Issue Type: Sub-task
  Components: Runtime / State Backends
Affects Versions: 2.0.0
Reporter: Yue Ma
 Fix For: 2.0.0






--
This message was sent by Atlassian Jira
(v8.20.10#820010)

[jira] [Created] (FLINK-35578) Release Frocksdb-8.10.0 official products

[Yue Ma (Jira)]

Yue Ma created FLINK-35578:
--

 Summary: Release Frocksdb-8.10.0 official products
 Key: FLINK-35578
 URL: https://issues.apache.org/jira/browse/FLINK-35578
 Project: Flink
  Issue Type: Sub-task
  Components: Runtime / State Backends
Affects Versions: 2.0.0
Reporter: Yue Ma
 Fix For: 2.0.0






--
This message was sent by Atlassian Jira
(v8.20.10#820010)

[jira] [Created] (FLINK-35577) Setup the CI environment for FRocksDB-8.10

[Yue Ma (Jira)]

Yue Ma created FLINK-35577:
--

 Summary: Setup the CI environment for FRocksDB-8.10
 Key: FLINK-35577
 URL: https://issues.apache.org/jira/browse/FLINK-35577
 Project: Flink
  Issue Type: Sub-task
  Components: Runtime / State Backends
Affects Versions: 2.0.0
Reporter: Yue Ma
 Fix For: 2.0.0






--
This message was sent by Atlassian Jira
(v8.20.10#820010)

[jira] [Created] (FLINK-35576) FRocksdb Cherry pick IngestDB requires commit

[Yue Ma (Jira)]

Yue Ma created FLINK-35576:
--

 Summary: FRocksdb Cherry pick IngestDB requires  commit
 Key: FLINK-35576
 URL: https://issues.apache.org/jira/browse/FLINK-35576
 Project: Flink
  Issue Type: Sub-task
  Components: Runtime / State Backends
Affects Versions: 2.0.0
Reporter: Yue Ma
 Fix For: 2.0.0


We support the API related to ingest DB in FRocksDb-8.10.0, but many of the 
fixes related to ingest DB were only integrated in the latest RocksDB version. 
So we need to add these fixed commit cherryclicks to FRocksDB.
Mainly include:
[https://github.com/facebook/rocksdb/pull/11646]
[https://github.com/facebook/rocksdb/pull/11868]
[https://github.com/facebook/rocksdb/pull/11811]
[https://github.com/facebook/rocksdb/pull/11381]
[https://github.com/facebook/rocksdb/pull/11379]
[https://github.com/facebook/rocksdb/pull/11378]



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

[jira] [Created] (FLINK-35575) FRocksDB supports disabling perf context during compilation

[Yue Ma (Jira)]

Yue Ma created FLINK-35575:
--

 Summary: FRocksDB supports disabling perf context during 
compilation
 Key: FLINK-35575
 URL: https://issues.apache.org/jira/browse/FLINK-35575
 Project: Flink
  Issue Type: Sub-task
  Components: Runtime / State Backends
Affects Versions: 2.0.0
Reporter: Yue Ma
 Fix For: 2.0.0


In FrocksDB 6 thread-local perf-context is disabled by reverting a specific 
commit (FLINK-19710). However, this creates conflicts and makes upgrading more 
difficult. We found that disabling *PERF_CONTEXT* can improve the performance 
of statebenchmark by about 5% and it doesn't create any conflicts. So we plan 
to supports disabling perf context during compilation in FRocksDB new version



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

[jira] [Created] (FLINK-35574) Set up base branch for FrocksDB-8.10

[Yue Ma (Jira)]

Yue Ma created FLINK-35574:
--

 Summary: Set up base branch for FrocksDB-8.10
 Key: FLINK-35574
 URL: https://issues.apache.org/jira/browse/FLINK-35574
 Project: Flink
  Issue Type: Sub-task
  Components: Runtime / State Backends
Affects Versions: 2.0.0
Reporter: Yue Ma
 Fix For: 2.0.0


As the first part of FLINK-35573, we need to prepare a base branch for 
FRocksDB-8.10.0 first. Mainly, it needs to be checked out from version 8.10.0 
of the Rocksdb community. Then check pick the commit which used by Flink from 
FRocksDB-6.20.3 to 8.10.0



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

[jira] [Created] (FLINK-35573) [FLIP-447] Upgrade FRocksDB from 6.20.3 to 8.10.0

[Yue Ma (Jira)]

Yue Ma created FLINK-35573:
--

 Summary: [FLIP-447] Upgrade FRocksDB from 6.20.3 to 8.10.0
 Key: FLINK-35573
 URL: https://issues.apache.org/jira/browse/FLINK-35573
 Project: Flink
  Issue Type: Improvement
  Components: Runtime / State Backends
Affects Versions: 2.0.0
Reporter: Yue Ma
 Fix For: 2.0.0


The FLIP: 
[https://cwiki.apache.org/confluence/display/FLINK/FLIP-447%3A+Upgrade+FRocksDB+from+6.20.3++to+8.10.0|https://cwiki.apache.org/confluence/display/FLINK/FLIP-447%3A+Upgrade+FRocksDB+from+6.20.3++to+8.10.0]
 

*_This FLIP proposes upgrading the version of FRocksDB in the Flink Project 
from 6.20.3 to 8.10.0._*

_RocksDBStateBackend is widely used by Flink users in large state scenarios.The 
last upgrade of FRocksDB was in version Flink-1.14, which mainly supported 
features such as support arm platform, deleteRange API, period compaction, etc. 
It has been a long time since then, and RocksDB has now been released to 
version 8.x. The main motivation for this upgrade is to leverage the features 
of higher versions of Rocksdb to make Flink RocksDBStateBackend more powerful. 
While RocksDB is also continuously optimizing and bug fixing, we hope to keep 
FRocksDB more or less in sync with RocksDB and upgrade it periodically._



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

[RESULT][VOTE] FLIP-447: Upgrade FRocksDB from 6.20.3 to 8.10.0

[yue ma]

Hi everyone,

Thanks for your review and the votes!

I am happy to announce that FLIP-447: Upgrade FRocksDB from 6.20.3 to
8.10.0 [1]. has been accepted.

The proposal has been accepted with 15 approving votes (9 binding) and there
is no disapproval:

- Zakelly Lan (binding)
- Yanfei Lei (binding)
- Rui Fan (binding)
- Yuan Mei (binding)
- Hangxiang Yu (binding)
- Stefan Richter (binding)
- Muhammet Orazov (non-binding)
- Yun Tang (binding)
- Gabor Somogyi (non-binding)
- Roc Marshal (non-binding)
- gongzhongqiang (non-binding)
- Roman Khachatryan (binding)
- Piotr Nowojski (binding)
- ConradJam (non-binding)
- zhourenxiang (non-binding)


Thanks to all involved.

[1]
https://cwiki.apache.org/confluence/display/FLINK/FLIP-447%3A+Upgrade+FRocksDB+from+6.20.3++to+8.10.0
[2] https://lists.apache.org/thread/r92qoxkt1kwtkbx9p45cpx4jto7s3l0d

-- 
Best,
Yue

[VOTE] FLIP-447: Upgrade FRocksDB from 6.20.3 to 8.10.0

[yue ma]

Hi everyone,

Thanks for all the feedback, I'd like to start a vote on the FLIP-447:
Upgrade FRocksDB from 6.20.3 to 8.10.0 [1]. The discussion thread is here
[2].

The vote will be open for at least 72 hours unless there is an objection or
insufficient votes.

[1]
https://cwiki.apache.org/confluence/display/FLINK/FLIP-447%3A+Upgrade+FRocksDB+from+6.20.3++to+8.10.0
[2] https://lists.apache.org/thread/lrxjfpjjwlq4sjzm1oolx58n1n8r48hw

-- 
Best,
Yue

Re: [DISCUSS] FLIP-447: Upgrade FRocksDB from 6.20.3 to 8.10.0

[yue ma]

Hi all,

Thank you all for your feedback. If there are no extra comments, I will
start voting in three days, thank you :)


yue ma  于2024年4月22日周一 14:09写道：

> Hi Flink devs,
>
> I would like to start a discussion on FLIP-447: Upgrade FRocksDB from
> 6.20.3 to 8.10.0
>
>
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-447%3A+Upgrade+FRocksDB+from+6.20.3++to+8.10.0
>
> This FLIP proposes upgrading the version of FRocksDB in the Flink Project
> from 6.20.3 to 8.10.0.
> The FLIP mainly introduces the main benefits of upgrading FRocksDB,
> including the use of IngestDB which can improve Rescaling performance by
> more than 10 times in certain scenarios, as well as other potential
> optimization points such as async_io, blob db, and tiered storage.The
> FLIP also presented test results based on RocksDB 8.10, including
> StateBenchmark and Nexmark tests.
> Overall, upgrading FRocksDB may result in a small regression of write
> performance( which is a very small part of the overall overhead), but it
> can bring many important performance benefits.
> So we hope to upgrade the version of FRocksDB through this FLIP.
>
> Looking forward to everyone's feedback and suggestions. Thank you!
> --
> Best regards,
> Yue
>


-- 
Best,
Yue

Re: [DISCUSS] FLIP-447: Upgrade FRocksDB from 6.20.3 to 8.10.0

[yue ma]

Hi lorenzo.affetti,

Thank you for your feedback. I'm sorry, I didn't fully understand your
question.
>From the results of the StateBenchmark, it appears that interfaces with
regression mainly include
（listAdd/listAddAll/mapUpdate/mapRemove/valueAdd/valueUpdate), all of which
call Rocksdb.put. So the conclusion is that it will only affect write
performance.
Of course, I think it is also important and meaningful to investigate the
reasons for write performance regression. I will create tickets to track
the progress. But investigating the cause will not block our upgrade
process this time. I believe that these two can be done in parallel.

Thanks again.

 于2024年4月24日周三 15:41写道：

> Hello yuema!
>
> Thank you for the proposal.
> In light of what is happening to state backends with FLIP-423 and others,
> it definitely makes sense to leverage the full power of latest FRocksDB.
>
> Small nit and question for you:
> do you have any idea how to justify the regression in write perf?
>
> Thank you again, +1 for this.
> On Apr 24, 2024 at 07:58 +0200, Gyula Fóra , wrote:
> > Thank you for driving this effort
> >
> > +1
> >
> > Cheers
> > Gyula
> >
> > On Wed, 24 Apr 2024 at 06:25, Yuan Mei  wrote:
> >
> > > Hey Yue,
> > >
> > > Thanks for all the great efforts significantly improving rescaling and
> > > upgrading rocksdb.
> > >
> > > +1 for this.
> > >
> > > Best
> > > Yuan
> > >
> > > On Wed, Apr 24, 2024 at 10:46 AM Zakelly Lan 
> > > wrote:
> > >
> > > > > Hi Yue,
> > > > >
> > > > > Thanks for this proposal!
> > > > >
> > > > > Given the great improvement we could have, the slight regression
> in write
> > > > > performance is a worthwhile trade-off, particularly as the
> mem-table
> > > > > operations contribute only a minor part to the overall overhead.
> So +1
> > > for
> > > > > this.
> > > > >
> > > > >
> > > > > Best,
> > > > > Zakelly
> > > > >
> > > > > On Tue, Apr 23, 2024 at 12:53 PM Yun Tang 
> wrote:
> > > > >
> > > > > > > Hi Yue,
> > > > > > >
> > > > > > > Thanks for driving this work.
> > > > > > >
> > > > > > > It has been three years since last major upgrade of FRocksDB.
> And it
> > > > > would
> > > > > > > be great improvement of Flink's state-backend with this
> upgrade. +1 for
> > > > > > > this work.
> > > > > > >
> > > > > > >
> > > > > > > Best
> > > > > > > Yun Tang
> > > > > > > 
> > > > > > > From: Yanfei Lei 
> > > > > > > Sent: Tuesday, April 23, 2024 12:50
> > > > > > > To: dev@flink.apache.org 
> > > > > > > Subject: Re: [DISCUSS] FLIP-447: Upgrade FRocksDB from 6.20.3
> to 8.10.0
> > > > > > >
> > > > > > > Hi Yue & Roman,
> > > > > > >
> > > > > > > Thanks for initiating this FLIP and all the efforts for the
> upgrade.
> > > > > > >
> > > > > > > 8.10.0 introduces some new features, making it possible for
> Flink to
> > > > > > > implement some new exciting features, and the upgrade also
> makes
> > > > > > > FRocksDB easier to maintain, +1 for upgrading.
> > > > > > >
> > > > > > > I read the FLIP and have a minor comment, it would be better
> to add
> > > > > > > some description about the environment/configuration of the
> nexmark's
> > > > > > > result.
> > > > > > >
> > > > > > > Roman Khachatryan  于2024年4月23日周二 12:07写道：
> > > > > > >
> > > > > > > > >
> > > > > > > > > Hi,
> > > > > > > > >
> > > > > > > > > Thanks for writing the proposal and preparing the upgrade.
> > > > > > > > >
> > > > > > > > > FRocksDB definitely needs to be kept in sync with the
> upstream and
> > > the
> > > > > > > new
> > > > > > > > > APIs are necessary for faster rescaling.
> > > > >

Re: [DISCUSS] FLIP-447: Upgrade FRocksDB from 6.20.3 to 8.10.0

[yue ma]

hi Yanfei,

Thanks for your feedback and reminders I have updated related information.
In fact, most of them use the default Configrations.

Yanfei Lei  于2024年4月23日周二 12:51写道：

> Hi Yue & Roman,
>
> Thanks for initiating this FLIP and all the efforts for the upgrade.
>
> 8.10.0 introduces some new features, making it possible for Flink to
> implement some new exciting features, and the upgrade also makes
> FRocksDB easier to maintain, +1 for upgrading.
>
> I read the FLIP and have a minor comment, it would be better to add
> some description about the environment/configuration of the nexmark's
> result.
>
> Roman Khachatryan  于2024年4月23日周二 12:07写道：
>
> >
> > Hi,
> >
> > Thanks for writing the proposal and preparing the upgrade.
> >
> > FRocksDB  definitely needs to be kept in sync with the upstream and the
> new
> > APIs are necessary for faster rescaling.
> > We're already using a similar version internally.
> >
> > I reviewed the FLIP and it looks good to me (disclaimer: I took part in
> > some steps of this effort).
> >
> >
> > Regards,
> > Roman
> >
> > On Mon, Apr 22, 2024, 08:11 yue ma  wrote:
> >
> > > Hi Flink devs,
> > >
> > > I would like to start a discussion on FLIP-447: Upgrade FRocksDB from
> > > 6.20.3 to 8.10.0
> > >
> > >
> > >
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-447%3A+Upgrade+FRocksDB+from+6.20.3++to+8.10.0
> > >
> > > This FLIP proposes upgrading the version of FRocksDB in the Flink
> Project
> > > from 6.20.3 to 8.10.0.
> > > The FLIP mainly introduces the main benefits of upgrading FRocksDB,
> > > including the use of IngestDB which can improve Rescaling performance
> by
> > > more than 10 times in certain scenarios, as well as other potential
> > > optimization points such as async_io, blob db, and tiered storage.The
> > > FLIP also presented test results based on RocksDB 8.10, including
> > > StateBenchmark and Nexmark tests.
> > > Overall, upgrading FRocksDB may result in a small regression of write
> > > performance( which is a very small part of the overall overhead), but
> it
> > > can bring many important performance benefits.
> > > So we hope to upgrade the version of FRocksDB through this FLIP.
> > >
> > > Looking forward to everyone's feedback and suggestions. Thank you!
> > > --
> > > Best regards,
> > > Yue
> > >
>
>
>
> --
> Best,
> Yanfei
>


-- 
Best,
Yue

Re: [DISCUSS] FLIP-447: Upgrade FRocksDB from 6.20.3 to 8.10.0

[yue ma]

Hi , Zakelly Lan

Thank you very much for your feedback and effort, especially for the help
in State Benchmark testing !

Zakelly Lan  于2024年4月24日周三 10:46写道：

> Hi Yue,
>
> Thanks for this proposal!
>
> Given the great improvement we could have, the slight regression in write
> performance is a worthwhile trade-off, particularly as the mem-table
> operations contribute only a minor part to the overall overhead. So +1 for
> this.
>
>
> Best,
> Zakelly
>
> On Tue, Apr 23, 2024 at 12:53 PM Yun Tang  wrote:
>
> > Hi Yue,
> >
> > Thanks for driving this work.
> >
> > It has been three years since last major upgrade of FRocksDB. And it
> would
> > be great improvement of Flink's state-backend with this upgrade. +1 for
> > this work.
> >
> >
> > Best
> > Yun Tang
> > 
> > From: Yanfei Lei 
> > Sent: Tuesday, April 23, 2024 12:50
> > To: dev@flink.apache.org 
> > Subject: Re: [DISCUSS] FLIP-447: Upgrade FRocksDB from 6.20.3 to 8.10.0
> >
> > Hi Yue & Roman,
> >
> > Thanks for initiating this FLIP and all the efforts for the upgrade.
> >
> > 8.10.0 introduces some new features, making it possible for Flink to
> > implement some new exciting features, and the upgrade also makes
> > FRocksDB easier to maintain, +1 for upgrading.
> >
> > I read the FLIP and have a minor comment, it would be better to add
> > some description about the environment/configuration of the nexmark's
> > result.
> >
> > Roman Khachatryan  于2024年4月23日周二 12:07写道：
> >
> > >
> > > Hi,
> > >
> > > Thanks for writing the proposal and preparing the upgrade.
> > >
> > > FRocksDB  definitely needs to be kept in sync with the upstream and the
> > new
> > > APIs are necessary for faster rescaling.
> > > We're already using a similar version internally.
> > >
> > > I reviewed the FLIP and it looks good to me (disclaimer: I took part in
> > > some steps of this effort).
> > >
> > >
> > > Regards,
> > > Roman
> > >
> > > On Mon, Apr 22, 2024, 08:11 yue ma  wrote:
> > >
> > > > Hi Flink devs,
> > > >
> > > > I would like to start a discussion on FLIP-447: Upgrade FRocksDB from
> > > > 6.20.3 to 8.10.0
> > > >
> > > >
> > > >
> >
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-447%3A+Upgrade+FRocksDB+from+6.20.3++to+8.10.0
> > > >
> > > > This FLIP proposes upgrading the version of FRocksDB in the Flink
> > Project
> > > > from 6.20.3 to 8.10.0.
> > > > The FLIP mainly introduces the main benefits of upgrading FRocksDB,
> > > > including the use of IngestDB which can improve Rescaling performance
> > by
> > > > more than 10 times in certain scenarios, as well as other potential
> > > > optimization points such as async_io, blob db, and tiered storage.The
> > > > FLIP also presented test results based on RocksDB 8.10, including
> > > > StateBenchmark and Nexmark tests.
> > > > Overall, upgrading FRocksDB may result in a small regression of write
> > > > performance( which is a very small part of the overall overhead), but
> > it
> > > > can bring many important performance benefits.
> > > > So we hope to upgrade the version of FRocksDB through this FLIP.
> > > >
> > > > Looking forward to everyone's feedback and suggestions. Thank you!
> > > > --
> > > > Best regards,
> > > > Yue
> > > >
> >
> >
> >
> > --
> > Best,
> > Yanfei
> >
>


-- 
Best,
Yue

Re: [DISCUSS] FLIP-447: Upgrade FRocksDB from 6.20.3 to 8.10.0

[yue ma]

hi Roman,

Thank you very much for your feedback and effort, especially for your help
in releasing Frocksdb products and testing the performance.

Roman Khachatryan  于2024年4月23日周二 12:07写道：

> Hi,
>
> Thanks for writing the proposal and preparing the upgrade.
>
> FRocksDB  definitely needs to be kept in sync with the upstream and the new
> APIs are necessary for faster rescaling.
> We're already using a similar version internally.
>
> I reviewed the FLIP and it looks good to me (disclaimer: I took part in
> some steps of this effort).
>
>
> Regards,
> Roman
>
> On Mon, Apr 22, 2024, 08:11 yue ma  wrote:
>
> > Hi Flink devs,
> >
> > I would like to start a discussion on FLIP-447: Upgrade FRocksDB from
> > 6.20.3 to 8.10.0
> >
> >
> >
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-447%3A+Upgrade+FRocksDB+from+6.20.3++to+8.10.0
> >
> > This FLIP proposes upgrading the version of FRocksDB in the Flink Project
> > from 6.20.3 to 8.10.0.
> > The FLIP mainly introduces the main benefits of upgrading FRocksDB,
> > including the use of IngestDB which can improve Rescaling performance by
> > more than 10 times in certain scenarios, as well as other potential
> > optimization points such as async_io, blob db, and tiered storage.The
> > FLIP also presented test results based on RocksDB 8.10, including
> > StateBenchmark and Nexmark tests.
> > Overall, upgrading FRocksDB may result in a small regression of write
> > performance( which is a very small part of the overall overhead), but it
> > can bring many important performance benefits.
> > So we hope to upgrade the version of FRocksDB through this FLIP.
> >
> > Looking forward to everyone's feedback and suggestions. Thank you!
> > --
> > Best regards,
> > Yue
> >
>


-- 
Best,
Yue

[DISCUSS] FLIP-447: Upgrade FRocksDB from 6.20.3 to 8.10.0

[yue ma]

Hi Flink devs,

I would like to start a discussion on FLIP-447: Upgrade FRocksDB from
6.20.3 to 8.10.0

https://cwiki.apache.org/confluence/display/FLINK/FLIP-447%3A+Upgrade+FRocksDB+from+6.20.3++to+8.10.0

This FLIP proposes upgrading the version of FRocksDB in the Flink Project
from 6.20.3 to 8.10.0.
The FLIP mainly introduces the main benefits of upgrading FRocksDB,
including the use of IngestDB which can improve Rescaling performance by
more than 10 times in certain scenarios, as well as other potential
optimization points such as async_io, blob db, and tiered storage.The
FLIP also presented test results based on RocksDB 8.10, including
StateBenchmark and Nexmark tests.
Overall, upgrading FRocksDB may result in a small regression of write
performance( which is a very small part of the overall overhead), but it
can bring many important performance benefits.
So we hope to upgrade the version of FRocksDB through this FLIP.

Looking forward to everyone's feedback and suggestions. Thank you!
-- 
Best regards,
Yue

Re: [ANNOUNCE] New Apache Flink Committer - Zakelly Lan

[yue ma]

Congratulations Zakelly!

Best,
Yue

Re: [VOTE] FLIP-433: State Access on DataStream API V2

[yue ma]

+1 (non-binding)

weijie guo  于2024年3月20日周三 20:35写道：

> Hi everyone,
>
>
> Thanks for all the feedback about the FLIP-433: State Access on
> DataStream API V2 [1]. The discussion thread is here [2].
>
>
> The vote will be open for at least 72 hours unless there is an
> objection or insufficient votes.
>
>
> Best regards,
>
> Weijie
>
>
> [1]
>
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-433%3A+State+Access+on+DataStream+API+V2
>
> [2] https://lists.apache.org/thread/8ghzqkvt99p4k7hjqxzwhqny7zb7xrwo
>


-- 
Best,
Yue

Re: [VOTE] FLIP-428: Fault Tolerance/Rescale Integration for Disaggregated State

[yue ma]

+1(non-binding)

Jinzhong Li  于2024年3月27日周三 19:31写道：

> Hi devs,
>
>
> I'd like to start a vote on the FLIP-428: Fault Tolerance/Rescale
> Integration for Disaggregated State [1]. The discussion thread is here [2].
>
>
> The vote will be open for at least 72 hours unless there is an objection or
> insufficient votes.
>
> [1] https://cwiki.apache.org/confluence/x/UYp3EQ
>
> [2] https://lists.apache.org/thread/vr8f91p715ct4lop6b3nr0fh4z5p312b
>
> Best,
>
> Jinzhong
>


-- 
Best,
Yue

Re: [VOTE] FLIP-427: Disaggregated state Store

[yue ma]

+1(non-binding)

Hangxiang Yu  于2024年3月27日周三 18:37写道：

> Hi devs,
>
> Thanks all for your valuable feedback about FLIP-427: Disaggregated state
> Store [1].
> I'd like to start a vote on it.  The discussion thread is here [2].
>
> The vote will be open for at least 72 hours unless there is an objection or
> insufficient votes.
>
> [1] https://cwiki.apache.org/confluence/x/T4p3EQ
> [2] https://lists.apache.org/thread/vktfzqvb7t4rltg7fdlsyd9sfdmrc4ft
>
>
> Best,
> Hangxiang
>


-- 
Best,
Yue

Re: [VOTE] FLIP-426: Grouping Remote State Access

[yue ma]

+1 (non-binding)

Jinzhong Li  于2024年3月27日周三 18:57写道：

> Hi devs,
>
> I'd like to start a vote on the FLIP-426: Grouping Remote State Access [1].
> The discussion thread is here [2].
>
> The vote will be open for at least 72 hours unless there is an objection or
> insufficient votes.
>
>
> [1] https://cwiki.apache.org/confluence/x/TYp3EQ
>
> [2] https://lists.apache.org/thread/bt931focfl9971cwq194trmf3pkdsxrf
>
>
> Best,
>
> Jinzhong
>


-- 
Best,
Yue

Re: [VOTE] FLIP-423: Disaggregated State Storage and Management (Umbrella FLIP)

[yue ma]

+1 (non-binding)


Best,
Yue

Re: [VOTE] FLIP-425: Asynchronous Execution Model

[yue ma]

+1 (non-binding)

Yanfei Lei  于2024年3月27日周三 18:28写道：

> Hi everyone,
>
> Thanks for all the feedback about the FLIP-425: Asynchronous Execution
> Model [1]. The discussion thread is here [2].
>
> The vote will be open for at least 72 hours unless there is an
> objection or insufficient votes.
>
> [1] https://cwiki.apache.org/confluence/x/S4p3EQ
> [2] https://lists.apache.org/thread/wxn1j848fnfkqsnrs947wh1wmj8n8z0h
>
> Best regards,
> Yanfei
>


-- 
Best,
Yue

Re: [DISCUSS] FLIP-428: Fault Tolerance/Rescale Integration for Disaggregated State

[yue ma]

Hi jinzhong,
Thanks for you reply. I still have some doubts about the first question. Is
there such a case
When you made a snapshot during the synchronization phase, you recorded the
current and manifest 8, but before asynchronous phase, the manifest reached
the size threshold and then the CURRENT FILE pointed to the new manifest 9,
and then uploaded the incorrect CURRENT file ?

Jinzhong Li  于2024年3月20日周三 20:13写道：

> Hi Yue,
>
> Thanks for your feedback!
>
> > 1. If we choose Option-3 for ForSt , how would we handle Manifest File
> > ? Should we take a snapshot of the Manifest during the synchronization
> phase?
>
> IIUC, the GetLiveFiles() API in Option-3 can also catch the fileInfo of
> Manifest files, and this api also return the manifest file size, which
> means this api could take snapshot for Manifest FileInfo (filename +
> fileSize) during the synchronization phase.
> You could refer to the rocksdb source code[1] to verify this.
>
>
>  > However, many distributed storage systems do not support the
> > ability of Fast Duplicate (such as HDFS). But ForSt has the ability to
> > directly read and write remote files. Can we not copy or Fast duplicate
> > these files, but instand of directly reuse and. reference these remote
> > files? I think this can reduce file download time and may be more useful
> > for most users who use HDFS (do not support Fast Duplicate)?
>
> Firstly, as far as I know, most remote file systems support the
> FastDuplicate, eg. S3 copyObject/Azure Blob Storage copyBlob/OSS
> copyObject, and the HDFS indeed does not support FastDuplicate.
>
> Actually，we have considered the design which reuses remote files. And that
> is what we want to implement in the coming future, where both checkpoints
> and restores can reuse existing files residing on the remote state storage.
> However, this design conflicts with the current file management system in
> Flink.  At present, remote state files are managed by the ForStDB
> (TaskManager side), while checkpoint files are managed by the JobManager,
> which is a major hindrance to file reuse. For example, issues could arise
> if a TM reuses a checkpoint file that is subsequently deleted by the JM.
> Therefore, as mentioned in FLIP-423[2], our roadmap is to first integrate
> checkpoint/restore mechanisms with existing framework  at milestone-1.
> Then, at milestone-2, we plan to introduce TM State Ownership and Faster
> Checkpointing mechanisms, which will allow both checkpointing and restoring
> to directly reuse remote files, thus achieving faster checkpointing and
> restoring.
>
> [1]
>
> https://github.com/facebook/rocksdb/blob/6ddfa5f06140c8d0726b561e16dc6894138bcfa0/db/db_filesnapshot.cc#L77
> [2]
>
> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=293046855#FLIP423:DisaggregatedStateStorageandManagement(UmbrellaFLIP)-RoadMap+LaunchingPlan
>
> Best,
> Jinzhong
>
>
>
>
>
>
>
> On Wed, Mar 20, 2024 at 4:01 PM yue ma  wrote:
>
> > Hi Jinzhong
> >
> > Thank you for initiating this FLIP.
> >
> > I have just some minor question:
> >
> > 1. If we choice Option-3 for ForSt , how would we handle Manifest File
> > ? Should we take snapshot of the Manifest during the synchronization
> phase?
> > Otherwise, may the Manifest and MetaInfo information be inconsistent
> during
> > recovery?
> > 2. For the Restore Operation , we need Fast Duplicate  Checkpoint Files
> to
> > Working Dir . However, many distributed storage systems do not support
> the
> > ability of Fast Duplicate (such as HDFS). But ForSt has the ability to
> > directly read and write remote files. Can we not copy or Fast duplicate
> > these files, but instand of directly reuse and. reference these remote
> > files? I think this can reduce file download time and may be more useful
> > for most users who use HDFS (do not support Fast Duplicate)?
> >
> > --
> > Best,
> > Yue
> >
>


-- 
Best,
Yue

Re: [DISCUSS] FLIP-426: Grouping Remote State Access

[yue ma]

hi Jinzhong
Thanks for your reply
The reason why I mentioned this point is because according to the official
Rocksdb documentation
https://rocksdb.org/blog/2022/10/07/asynchronous-io-in-rocksdb.html. if we
turn on async_io and use multiGet, it can improve the performance of point
look upc  by 100%.  Moreover, especially in Flink SQL tasks, there are many
ways to access state through mini batch, so I believe this feature also
greatly optimizes the synchronous access method and is worth doing.
If we first support batch access for asynchronous models, I think it would
be okay. My point is, should we consider whether it can be easily extended
if we support synchronous models in the future

Jinzhong Li  于2024年3月19日周二 20:59写道：

> Hi Yue,
>
> Thanks for your feedback!
>
> > 1. Does Grouping Remote State Access only support asynchronous
> interfaces?
> >--If it is: IIUC, MultiGet can also greatly improve performance for
> > synchronous access modes. Do we need to support it ？
>
> Yes. If we want to support MultiGet on existing synchronous access mode, we
> have to introduce a grouping component akin to the AEC described in
> FLIP-425[1].
> I think such a change would introduce additional complexity to the current
> synchronous model, and the extent of performance gains remains uncertain.
> Therefore, I recommend only asynchronous interfaces support "Grouping
> Remote State Access", which is designed to efficiently minimize latency in
> accessing remote state storage.
>
> > 2. Can a simple example be added to FLip on how to use Batch to access
> > states and obtain the results of states on the API?
>
> Sure. I have added a code example in the Flip[2]. Note that the multiget in
> this Flip is an internal interface, not a user-facing interface.
>
> > 3. I also agree with XiaoRui's viewpoint. Is there a corresponding Config
> > to control the  state access batch strategy?
>
> Yes, we would offer some configurable options that allow users to adjust
> the behavior of batching and grouping state access (eg. batching size,
> etc.).
>
> [1]
>
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-425%3A+Asynchronous+Execution+Model
> [2]
>
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-426%3A+Grouping+Remote+State+Access#FLIP426:GroupingRemoteStateAccess-CodeExampleonHowtoAccessStateUsingBatch
>
> Best,
> Jinzhong Li
>
>
> On Tue, Mar 19, 2024 at 5:52 PM yue ma  wrote:
>
> > Hi Jinzhong,
> >
> > Thanks for the FLIP.  I have the following questions:
> >
> > 1. Does Grouping Remote State Access only support asynchronous
> interfaces?
> > --If it is: IIUC, MultiGet can also greatly improve performance for
> > synchronous access modes. Do we need to support it ？
> > --If not, how can we distinguish between using Grouping State Access
> in
> > asynchronous and synchronous modes?
> > 2.  Can a simple example be added to FLip on how to use Batch to access
> > states and obtain the results of states on the API?
> > 3. I also agree with XiaoRui's viewpoint. Is there a corresponding Config
> > to control the  state access batch strategy?
> >
> > --
> > Best,
> > Yue
> >
>


-- 
Best,
Yue

Re: [DISCUSS] FLIP-428: Fault Tolerance/Rescale Integration for Disaggregated State

[yue ma]

Hi Jinzhong

Thank you for initiating this FLIP.

I have just some minor question:

1. If we choice Option-3 for ForSt , how would we handle Manifest File
? Should we take snapshot of the Manifest during the synchronization phase?
Otherwise, may the Manifest and MetaInfo information be inconsistent during
recovery?
2. For the Restore Operation , we need Fast Duplicate  Checkpoint Files to
Working Dir . However, many distributed storage systems do not support the
ability of Fast Duplicate (such as HDFS). But ForSt has the ability to
directly read and write remote files. Can we not copy or Fast duplicate
these files, but instand of directly reuse and. reference these remote
files? I think this can reduce file download time and may be more useful
for most users who use HDFS (do not support Fast Duplicate)?

-- 
Best,
Yue

Re: [DISCUSS] FLIP-427: Disaggregated State Store

[yue ma]

Hi Hangxiang,

Thanks for bringing this discussion.
I have a few questions about the Proposal you mentioned in the FLIP.

The current conclusion is to use proposal 2, which is okay for me. My point
is whether we should retain the potential of proposal 1 in the design.
There are the following reasons:
1. No JNI overhead, just like the Performance Part mentioned in Flip
2. RocksDB currently also provides an interface for Env, and there are also
some implementations, such as HDFS-ENV, which seem to be easily scalable.
3. The RocksDB community continues to support LSM for different storage
media, such as  Tiered Storage

  And some optimizations have been made for this scenario, such as Per
Key Placement Comparison
.
 *Secondary cache
*,
similar to the Hybrid Block Cache mentioned in Flip-423
 If we use proposal1, we can easily reuse these optimizations .It is even
possible to discuss and review the solution together in the Rocksdb
community.
 In fact, we have already implemented some production practices using
Proposal1 internally. We have integrated HybridEnv, Tiered Storage, and
Secondary Cache on RocksDB and optimized the performance of Checkpoint and
State Restore. It seems work well for us.

-- 
Best,
Yue

Re: [DISCUSS] FLIP-426: Grouping Remote State Access

[yue ma]

Hi Jinzhong,

Thanks for the FLIP.  I have the following questions:

1. Does Grouping Remote State Access only support asynchronous interfaces?
--If it is: IIUC, MultiGet can also greatly improve performance for
synchronous access modes. Do we need to support it ？
--If not, how can we distinguish between using Grouping State Access in
asynchronous and synchronous modes?
2.  Can a simple example be added to FLip on how to use Batch to access
states and obtain the results of states on the API?
3. I also agree with XiaoRui's viewpoint. Is there a corresponding Config
to control the  state access batch strategy?

-- 
Best,
Yue

Re: [DISCUSS] FLIP-424: Asynchronous State APIs

[yue ma]

Hi Zakelly,

Thanks for your proposal. The FLIP looks good  to me +1! I'd like to ask
some minor questions
I found that there is also a definition of class `FutureUtils` under `org.
apache. flink. util. concurrent` which seems to offer more interfaces. My
question is:
1. Is it possible for all `FutureUtils` in Flink to reuse the same util
class?
2. It seems that there is no concept of retry, timeout, or delay in your
async state api design . Do we need to provide such capabilities like
`orTimeout` 、`completeDelayed`?

Jing Ge  于2024年3月13日周三 20:00写道：

> indeed! I missed that part. Thanks for the hint!
>
> Best regards,
> Jing
>
> On Wed, Mar 13, 2024 at 6:02 AM Zakelly Lan  wrote:
>
> > Hi Jing,
> >
> > The deprecation and removal of original APIs is beyond the scope of
> > current FLIP, but I do add/highlight such information under
> "Compatibility,
> > Deprecation, and Migration Plan" section.
> >
> >
> > Best,
> > Zakelly
> >
> > On Wed, Mar 13, 2024 at 9:18 AM Yunfeng Zhou <
> flink.zhouyunf...@gmail.com>
> > wrote:
> >
> >> Hi Zakelly,
> >>
> >> Thanks for your responses. I agree with it that we can keep the design
> >> as it is for now and see if others have any better ideas for these
> >> questions.
> >>
> >> Best,
> >> Yunfeng
> >>
> >> On Tue, Mar 12, 2024 at 5:23 PM Zakelly Lan 
> >> wrote:
> >> >
> >> > Hi Xuannan,
> >> >
> >> > Thanks for your comments, I modified the FLIP accordingly.
> >> >
> >> > Hi Yunfeng,
> >> >
> >> > Thanks for sharing your opinions!
> >> >
> >> >> Could you provide some hint on use cases where users need to mix sync
> >> >> and async state operations in spite of the performance regression?
> >> >> This information might help address our concerns on design. If the
> >> >> mixed usage is simply something not recommended, I would prefer to
> >> >> prohibit such usage from API.
> >> >
> >> > In fact, there is no scenario where users MUST use the sync APIs, but
> >> it is much easier to use for those who are not familiar with
> asynchronous
> >> programming. If they want to migrate their job from Flink 1.x to 2.0
> >> leveraging some benefits from asynchronous APIs, they may try the mixed
> >> usage. It is not user-friendly to directly throw exceptions at runtime,
> I
> >> think our better approach is to warn users and recommend avoiding this.
> I
> >> added an example in this FLIP.
> >> >
> >> > Well, I do not insist on allowing mixed usage of APIs if others reach
> >> an agreement that we won't support that . I think the most important is
> to
> >> keep the API easy to use and understand, thus I propose a unified state
> >> declaration and explicit meaning in method name. WDYT?
> >> >
> >> >> Sorry I missed the new sink API. I do still think that it would be
> >> >> better to make the package name more informative, and ".v2." does not
> >> >> contain information for new Flink users who did not know the v1 of
> >> >> state API. Unlike internal implementation and performance
> >> >> optimization, API will hardly be compromised for now and updated in
> >> >> future, so I still suggest we improve the package name now if
> >> >> possible. But given the existing practice of sink v2 and
> >> >> AbstractStreamOperatorV2, the current package name would be
> acceptable
> >> >> to me if other reviewers of this FLIP agrees on it.
> >> >
> >> > Actually, I don't like 'v2' either. So if there is another good name,
> >> I'd be happy to apply. This is a compromise to the current situation.
> Maybe
> >> we could refine this after the retirement of original state APIs.
> >> >
> >> >
> >> > Thanks & Best,
> >> > Zakelly
> >> >
> >> >
> >> > On Tue, Mar 12, 2024 at 1:42 PM Yunfeng Zhou <
> >> flink.zhouyunf...@gmail.com> wrote:
> >> >>
> >> >> Hi Zakelly,
> >> >>
> >> >> Thanks for the quick response!
> >> >>
> >> >> > Actually splitting APIs into two sets ... warn them in runtime.
> >> >>
> >> >> Could you provide some hint on use cases where users need to mix sync
> >> >> and async state operations in spite of the performance regression?
> >> >> This information might help address our concerns on design. If the
> >> >> mixed usage is simply something not recommended, I would prefer to
> >> >> prohibit such usage from API.
> >> >>
> >> >> > In fact ... .sink2`.
> >> >>
> >> >> Sorry I missed the new sink API. I do still think that it would be
> >> >> better to make the package name more informative, and ".v2." does not
> >> >> contain information for new Flink users who did not know the v1 of
> >> >> state API. Unlike internal implementation and performance
> >> >> optimization, API will hardly be compromised for now and updated in
> >> >> future, so I still suggest we improve the package name now if
> >> >> possible. But given the existing practice of sink v2 and
> >> >> AbstractStreamOperatorV2, the current package name would be
> acceptable
> >> >> to me if other reviewers of this FLIP agrees on it.
> >> >>
> >> >> Best,
> >> >> Yunfeng
> >> >>
> >> >> On Mon, Mar 11, 2024 at 5:27 PM 

[jira] [Created] (FLINK-34210) DefaultExecutionGraphBuilder#isCheckpointingEnabled may return Wrong Value when checkpoint disabled

[Yue Ma (Jira)]

Yue Ma created FLINK-34210:
--

 Summary: DefaultExecutionGraphBuilder#isCheckpointingEnabled may 
return Wrong Value when checkpoint disabled
 Key: FLINK-34210
 URL: https://issues.apache.org/jira/browse/FLINK-34210
 Project: Flink
  Issue Type: Bug
  Components: Runtime / Checkpointing
Affects Versions: 1.18.1, 1.19.0
Reporter: Yue Ma


The *DefaultExecutionGraphBuilder* will call 
_isCheckpointingEnabled(JobGraph jobGraph)_
to determine whether the job has enabled Checkpoint and whether to initialize 
CheckpointCoordinator related components such as CheckpointCoordinator, 
CheckpointIDCounter , etc.

The problem is that the logic for determining isCheckpointingEnable here is 
inaccurate, as *jobGraph. getCheckpointingSettings()* will not be NULL when 
checkpoint is not enabled, but with 
CheckpointCoordinatorConfiguration.DISABLED_CHECKPOINT_INTERVAL Interval

The method to fix this problem is also quite clear. We need to directly reuse 
the result of jobGraph.isCheckpointingEnable() here

 



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

Re: [DISCUSS] FLIP-414: Support Retry Mechanism in RocksDBStateDataTransfer

[yue ma]

Thanks for driving this effort, xiangyu!
The proposal overall LGTM.
I just have a small question. There are other places in Flink that interact
with external storage. Should we consider adding a general retry mechanism
to them?

xiangyu feng  于2024年1月8日周一 11:31写道：

> Hi devs,
>
> I'm opening this thread to discuss FLIP-414: Support Retry Mechanism in
> RocksDBStateDataTransfer[1].
>
> Currently, there is no retry mechanism for downloading and uploading
> RocksDB state files. Any jittering of remote filesystem might lead to a
> checkpoint failure. By supporting retry mechanism in
> `RocksDBStateDataTransfer`, we can significantly reduce the failure rate of
> checkpoint during asynchronous phrase.
>
> To make this retry mechanism configurable, we have introduced two options
> in this FLIP: `state.backend.rocksdb.checkpoint.transfer.retry.times` and `
> state.backend.rocksdb.checkpoint.transfer.retry.interval`. The default
> behavior remains to be no retry will be performed in order to be consistent
> with the original behavior.
>
> Looking forward to your feedback, thanks.
>
> [1]
>
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-414%3A+Support+Retry+Mechanism+in+RocksDBStateDataTransfer
>
> Best regards,
> Xiangyu Feng
>


-- 
Best,
Yue

[jira] [Created] (FLINK-33946) RocksDb sets setAvoidFlushDuringShutdown to true to speed up Task Cancel

[Yue Ma (Jira)]

Yue Ma created FLINK-33946:
--

 Summary: RocksDb sets setAvoidFlushDuringShutdown to true to speed 
up Task Cancel
 Key: FLINK-33946
 URL: https://issues.apache.org/jira/browse/FLINK-33946
 Project: Flink
  Issue Type: Improvement
  Components: Runtime / State Backends
Affects Versions: 1.19.0
Reporter: Yue Ma
 Fix For: 1.19.0


When a Job fails, the task needs to be canceled and re-deployed. 
RocksDBStatebackend will call RocksDB.close when disposing.


{code:java}
if (!shutting_down_.load(std::memory_order_acquire) &&
has_unpersisted_data_.load(std::memory_order_relaxed) &&
!mutable_db_options_.avoid_flush_during_shutdown) {
  if (immutable_db_options_.atomic_flush) {
autovector cfds;
SelectColumnFamiliesForAtomicFlush();
mutex_.Unlock();
Status s =
AtomicFlushMemTables(cfds, FlushOptions(), FlushReason::kShutDown);
s.PermitUncheckedError();  //**TODO: What to do on error?
mutex_.Lock();
  } else {
for (auto cfd : *versions_->GetColumnFamilySet()) {
  if (!cfd->IsDropped() && cfd->initialized() && !cfd->mem()->IsEmpty()) {
cfd->Ref();
mutex_.Unlock();
Status s = FlushMemTable(cfd, FlushOptions(), FlushReason::kShutDown);
s.PermitUncheckedError();  //**TODO: What to do on error?
mutex_.Lock();
cfd->UnrefAndTryDelete();
  }
}
  } {code}


By default (avoid_flush_during_shutdown=false) RocksDb requires FlushMemtable 
when Close. When the disk pressure is high or the Memtable is large, this 
process will be more time-consuming, which will cause the Task to get stuck in 
the Canceling stage and affect the speed of job Failover.
In fact, it is completely unnecessary to Flush memtable when Flink Task is 
Close, because the data can be replayed from Checkpoint. So we can set 
avoid_flush_during_shutdown to true to speed up Task Failover



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

[jira] [Created] (FLINK-33819) Suppor setting CompressType in RocksDBStateBackend

[Yue Ma (Jira)]

Yue Ma created FLINK-33819:
--

 Summary: Suppor setting CompressType in RocksDBStateBackend
 Key: FLINK-33819
 URL: https://issues.apache.org/jira/browse/FLINK-33819
 Project: Flink
  Issue Type: Improvement
  Components: Runtime / State Backends
Affects Versions: 1.18.0
Reporter: Yue Ma
 Fix For: 1.19.0
 Attachments: image-2023-12-14-11-32-32-968.png

Currently, RocksDBStateBackend does not support setting the compression level, 
and Snappy is used for compression by default. But we have some scenarios where 
compression will use a lot of CPU resources. Turning off compression can 
significantly reduce CPU overhead. So we may need to support a parameter for 
users to set the CompressType of Rocksdb.

  
!https://internal-api-drive-stream.larkoffice.com/space/api/box/stream/download/preview/ALADbWTMGoD6WexSFGecz2Olnrb/?preview_type=16!



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

[jira] [Created] (FLINK-33126) Fix EventTimeAllWindowCheckpointingITCase jobName typo

[Yue Ma (Jira)]

Yue Ma created FLINK-33126:
--

 Summary: Fix EventTimeAllWindowCheckpointingITCase jobName typo
 Key: FLINK-33126
 URL: https://issues.apache.org/jira/browse/FLINK-33126
 Project: Flink
  Issue Type: Improvement
  Components: Tests
Affects Versions: 1.17.1
Reporter: Yue Ma


Fix EventTimeAllWindowCheckpointingITCase jobName Typo 



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

[jira] [Created] (FLINK-32833) Rocksdb CacheIndexAndFilterBlocks must be true when using shared memory

[Yue Ma (Jira)]

Yue Ma created FLINK-32833:
--

 Summary: Rocksdb CacheIndexAndFilterBlocks must be true when using 
shared memory
 Key: FLINK-32833
 URL: https://issues.apache.org/jira/browse/FLINK-32833
 Project: Flink
  Issue Type: Improvement
  Components: Runtime / State Backends
Affects Versions: 1.17.1
Reporter: Yue Ma


Currently in RocksDBResourceContainer#getColumnOptions, if sharedResources is 
used, blockBasedTableConfig will add the following configuration by default.

blockBasedTableConfig.setBlockCache(blockCache);
blockBasedTableConfig.setCacheIndexAndFilterBlocks(true);

blockBasedTableConfig.setCacheIndexAndFilterBlocksWithHighPriority(true);
blockBasedTableConfig.setPinL0FilterAndIndexBlocksInCache(true);

In my understanding, these configurations can help flink better manage the 
memory of rocksdb and save some memory overhead in some scenarios. But this may 
not be the best practice, mainly for the following reasons:
1. After CacheIndexAndFilterBlocks is set to true, it may cause index and 
filter miss when reading, resulting in performance degradation.
2. These parameters may not be bound together with whether shared memory is 
used, or some configurations should be supported separately to decide whether 
to enable these features



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

Re: [ANNOUNCE] New Apache Flink Committer - Weihua Hu

[yue ma]

Congratulations, Weihua!

[jira] [Created] (FLINK-31238) Use IngestDB to speed up Rocksdb rescaling recovery

[Yue Ma (Jira)]

Yue Ma created FLINK-31238:
--

 Summary: Use IngestDB to speed up Rocksdb rescaling recovery 
 Key: FLINK-31238
 URL: https://issues.apache.org/jira/browse/FLINK-31238
 Project: Flink
  Issue Type: Improvement
  Components: Runtime / Checkpointing
Affects Versions: 1.16.1
Reporter: Yue Ma
 Fix For: 1.16.2
 Attachments: image-2023-02-27-16-41-18-552.png

There have been many discussions and optimizations in the community about 
optimizing rocksdb scaling and recovery.

https://issues.apache.org/jira/browse/FLINK-17971

https://issues.apache.org/jira/browse/FLINK-8845

https://issues.apache.org/jira/browse/FLINK-21321

We hope to discuss some of our explorations under this ticket

The process of scaling and recovering in rocksdb simply requires two steps
 # Insert the valid keyGroup data of the new task.
 # Delete the invalid data in the old stateHandle.

The current method for data writing is to specify the main Db first and then 
insert data using writeBatch.In addition, the method of deleteRange is 
currently used to speed up the ClipDB. But in our production environment, we 
found that the speed of rescaling is still very slow, especially when the state 
of a single Task is large. 

 

We hope that the previous sst file can be reused directly when restoring state, 
instead of retraversing the data. So we made some attempts to optimize it in 
our internal version of flink and frocksdb.

 

We added two APIs *ClipDb* and *IngestDb* in frocksdb. 
 * ClipDB is used to clip the data of a DB. Different from db.DeteleRange and 
db.Delete, DeleteValue and RangeTombstone will not be generated for parts 
beyond the key range. We will iterate over the FileMetaData of db. Process each 
sst file. There are three situations here. 
If all the keys of a file are required, we will keep the sst file and do 
nothing 
If all the keys of the sst file exceed the specified range, we will delete the 
file directly. 
If we only need some part of the sst file, we will rewrite the required keys to 
generate a new sst file。
All sst file changes will be placed in a VersionEdit, and the current versions 
will LogAndApply this edit to ensure that these changes can take effect
 * IngestDb is used to directly ingest all sst files of one DB into another DB. 
But it is necessary to strictly ensure that the keys of the two DBs do not 
overlap, which is easy to do in the Flink scenario. The hard link method will 
be used in the process of ingesting files, so it will be very fast. At the same 
time, the file number of the main DB will be incremented sequentially, and the 
SequenceNumber of the main DB will be updated to the larger SequenceNumber of 
the two DBs.

When IngestDb and ClipDb are supported, the state restoration logic is as 
follows
 * Open the first StateHandle as the main DB and pause the compaction.
 * Clip the main DB according to the KeyGroup range of the Task with ClipDB
 * Open other StateHandles in sequence as Tmp DB, and perform ClipDb  according 
to the KeyGroup range
 * Ingest all tmpDb into the main Db after tmpDb cliped
 * Open the Compaction process of the main DB
!image-2023-02-27-16-41-18-552.png!

We have done some benchmark tests on the internal Flink version, and the test 
results show that compared with the writeBatch method, the expansion and 
recovery speed of IngestDb can be increased by 5 to 10 times As follows 

 
 * parallelism changes from 4 to 2

|*TaskStateSize*|*Write_Batch*|*SST_File_Writer*|*Ingest_DB*|
|500M|Iteration 1: 8.018 s/op
Iteration 2: 9.551 s/op
Iteration 3: 7.486 s/op|Iteration 1: 6.041 s/op
Iteration 2: 5.934 s/op
Iteration 3: 6.707 s/o|{color:#FF}Iteration 1: 3.922 s/op{color}
{color:#FF}Iteration 2: 3.208 s/op{color}
{color:#FF}Iteration 3: 3.096 s/op{color}|
|1G|Iteration 1: 19.686 s/op
Iteration 2: 19.402 s/op
Iteration 3: 21.146 s/op|Iteration 1: 17.538 s/op
Iteration 2: 16.933 s/op
Iteration 3: 15.486 s/op|{color:#FF}Iteration 1: 6.207 s/op{color}
{color:#FF}Iteration 2: 7.164 s/op{color}
{color:#FF}Iteration 3: 6.397 s/op{color}|
|5G|Iteration 1: 244.795 s/op
Iteration 2: 243.141 s/op
Iteration 3: 253.542 s/op|Iteration 1: 78.058 s/op
Iteration 2: 85.635 s/op
Iteration 3: 76.568 s/op|{color:#FF}Iteration 1: 23.397 s/op{color}
{color:#FF}Iteration 2: 21.387 s/op{color}
{color:#FF}Iteration 3: 22.858 s/op{color}|
 * parallelism changes from 4 to 8

|*TaskStateSize*|*Write_Batch*|*SST_File_Writer*|*Ingest_DB*|
|500M|Iteration 1: 3.477 s/op
Iteration 2: 3.515 s/op
Iteration 3: 3.433 s/op|Iteration 1: 3.453 s/op
Iteration 2: 3.300 s/op
Iteration 3: 3.313 s/op|{color:#FF}Iteration 1: 0.941 s/op{color}
{color:#FF}Iteration 2: 0.963 s/op{color}
{color:#FF}Iteration 3: 1.102 s/op{color}|
|1G|IIteration 1: 7.571 s/op
Iteration 2: 7.352 s/op
Iteration 3: 7.568 s/op|Iteration 1: 5.032 s/op
Iteration 2: 4.689

Re: [VOTE] FLIP-228: Support Within between events in CEP Pattern

[yue ma]

Thanks for Nicholas driving this.
+1

Nicholas  于2022年6月13日周一 10:16写道：

> Hi everyone，
>
>
>
>
> Thanks for feedback for FLIP-228: Support Within between events in CEP
> Pattern[1] on the discussion thread[2]. I'd like to start a VOTE thread for
> FLIP-228.
>
> The vote will be open for at least 72 hours unless there is an objection
> or not enough votes.
>
>
>
>
> Regards,
>
> Nicholas Jiang
>
>
>
>
> [1]
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-228%3A+Support+Within+between+events+in+CEP+Pattern
>
> [2] https://lists.apache.org/thread/p60ctx213f8921rgklk5f0b6xfrs8ksz

Re: [DISCUSS] FLIP-228: Support Within between events in CEP Pattern

[yue ma]

hi Nicholas,

Thanks for bringing this discussion, we also think it's a useful feature.
Some fine-grained timeout pattern matching  can be implemented in CEP which
makes Flink CEP more powerful

Nicholas  于2022年5月5日周四 14:28写道：

> Hi everyone,
>
>
>
>
> Pattern#withIn interface in CEP defines the maximum time interval in which
> a matching pattern has to be completed in order to be considered valid,
> which interval corresponds to the maximum time gap between first and the
> last event. The interval representing the maximum time gap between events
> is required to define in the scenario like purchasing good within a maximum
> of 5 minutes after browsing.
>
>
>
>
> I would like to start a discussion about FLIP-228[1], in which within
> between events is proposed in Pattern to support the definition of the
> maximum time interval in which a completed partial matching pattern is
> considered valid, which interval represents the maximum time gap between
> events for partial matching Pattern.
>
>
>
>
> Hence we propose the Pattern#partialWithin interface to define the maximum
> time interval in which a completed partial matching pattern is considered
> valid. Please take a look at the FLIP page [1] to get more details. Any
> feedback about the FLIP-228 would be appreciated!
>
>
>
>
> [1]
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-228%3A+Support+Within+between+events+in+CEP+Pattern
>
>
>
>
> Best regards,
>
> Nicholas Jiang

Re: [DISCUSS] FLIP-228: Support Within between events in CEP Pattern

[yue ma]

hi Nicholas ,


Nicholas  于2022年5月5日周四 14:28写道：

> Hi everyone,
>
>
>
>
> Pattern#withIn interface in CEP defines the maximum time interval in which
> a matching pattern has to be completed in order to be considered valid,
> which interval corresponds to the maximum time gap between first and the
> last event. The interval representing the maximum time gap between events
> is required to define in the scenario like purchasing good within a maximum
> of 5 minutes after browsing.
>
>
>
>
> I would like to start a discussion about FLIP-228[1], in which within
> between events is proposed in Pattern to support the definition of the
> maximum time interval in which a completed partial matching pattern is
> considered valid, which interval represents the maximum time gap between
> events for partial matching Pattern.
>
>
>
>
> Hence we propose the Pattern#partialWithin interface to define the maximum
> time interval in which a completed partial matching pattern is considered
> valid. Please take a look at the FLIP page [1] to get more details. Any
> feedback about the FLIP-228 would be appreciated!
>
>
>
>
> [1]
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-228%3A+Support+Within+between+events+in+CEP+Pattern
>
>
>
>
> Best regards,
>
> Nicholas Jiang

[jira] [Created] (FLINK-27218) Serializer in OperatorState has not been updated when new Serializers are NOT incompatible

[Yue Ma (Jira)]

Yue Ma created FLINK-27218:
--

 Summary: Serializer in OperatorState has not been updated when new 
Serializers are NOT incompatible
 Key: FLINK-27218
 URL: https://issues.apache.org/jira/browse/FLINK-27218
 Project: Flink
  Issue Type: Bug
  Components: Runtime / State Backends
Affects Versions: 1.15.1
Reporter: Yue Ma
 Attachments: image-2022-04-13-14-50-10-921.png

Now OperatorState can only be constructed via DefaultOperatorStateBackend. But 
when *BroadcastState* or *PartitionableListState* Serializer changes, it seems 
to have the following problems.

As an example, we can see how PartitionableListState is initialized.

First, we will construct a restored PartitionableListState based on the 
information in the snapshot during the restoreOperation.

Then we will update the StateMetaInfo in partitionableListState as the 
following code

 
{code:java}
TypeSerializerSchemaCompatibility stateCompatibility =
                
restoredPartitionableListStateMetaInfo.updatePartitionStateSerializer(newPartitionStateSerializer);

partitionableListState.setStateMetaInfo(restoredPartitionableListStateMetaInfo);{code}
 


The main problem is that there is also an internalListCopySerializer in 
PartitionableListState that is built using the previous Serializer and it has 
not been updated. 
But internalListCopySerializer will be used later when making the snopshot. 
Therefore, when we update the StateMetaInfo, the internalListCopySerializer 
also needs to be updated.

This problem also exists in BroadcastState



--
This message was sent by Atlassian Jira
(v8.20.1#820001)

[jira] [Created] (FLINK-26941) Support Pattern end with notFollowedBy with window

[Yue Ma (Jira)]

Yue Ma created FLINK-26941:
--

 Summary: Support Pattern end with notFollowedBy with window
 Key: FLINK-26941
 URL: https://issues.apache.org/jira/browse/FLINK-26941
 Project: Flink
  Issue Type: Improvement
  Components: Library / CEP
Affects Versions: 1.14.4
Reporter: Yue Ma
 Fix For: 1.14.4


Currently a pattern sequence cannot end in notFollowedBy() in Flink CEP. But in 
fact, this requirement exists in many scenarios. As mentioned in the following 
tickets：

https://issues.apache.org/jira/browse/FLINK-16010
https://issues.apache.org/jira/browse/FLINK-9431

Unfortunately, these tickets are not active for a long time.But we still think 
this is an important feature for Flink CEP, so we would like to share our 
implementation.
if we want to find the users who created an order but didn't pay in 10 minutes. 
We could code like this:
Pattern.begin('create').notFollowedBy('pay_order').withIn(10min)
If we receive the create event but don't receive the pay event within 10 
minutes, then the match will be successful.

The idea of implementation is basically the same as the design of FLINK-16010.
A Pending State is introduced to represent the state of waiting for a timeout, 
and there is a take edge between the Pending node and the Stop node.
When advanceTime, if it is found that the pending node has timed out, then 
extract the timeout sequence and output it normally as successed matched 
sequence.



--
This message was sent by Atlassian Jira
(v8.20.1#820001)

Re: [DISCUSS] FLIP-200: Support Multiple Rule and Dynamic Rule Changing (Flink CEP)

[yue ma]

Glad to see the Community's progress in Flink CEP. After reading this Flip,
I have few questions, would you please take a look  ?

1. About Pattern Updating. If we use PatternProcessoerDiscoverer to update
the rules, will it increase the load of JM? For example, if the user wants
the updated rule to take effect immediately,, which means that we need to
set a shorter check interval  or there is another scenario when users
rarely update the pattern, will the PatternProcessoerDiscoverer be in most
of the time Do useless checks ? Will a lazy update mode could be used,
which the pattern only be updated when triggered by the user, and do
nothing at other times ?

2.   I still have some confusion about how Key Generating Opertator and
CepOperator (Pattern Matching & Processing Operator) work together. If
there are N PatternProcessors, will the Key Generating Opertator generate N
keyedStreams, and then N CepOperator would process each Key separately ? Or
every CepOperator Task would process all patterns, if so, does the key type
in each PatternProcessor need to be the same ?

3. Maybe need to pay attention to it when implementing it .If some Pattern
has been removed or updateed  ,will the partially matched results in
StateBackend would be clean up or We rely on state ttl to clean up these
expired states.

4. Will the PatternProcessorManager keep all the active PatternProcessor in
memory ? We have also Support Multiple Rule and Dynamic Rule Changing .
But we are facing such a problem, some users’ usage scenarios are that they
want to have their own pattern for each user_id, which means that there
could be thousands of patterns, which would make the performance of Pattern
Matching very poor. We are also trying to solve this problem.

Yunfeng Zhou  于2021年12月10日周五 19:16写道：

> Hi all，
>
> I'm opening this thread to propose the design to support multiple rule &
> dynamic rule changing in the Flink-CEP project, as described in FLIP-200
> [1]
> .
>
> Currently Flink CEP only supports having a single pattern inside a
> CepOperator and does not support changing the pattern dynamically. In order
> to reduce resource consumption and to experience shorter downtime during
> pattern updates, there is a growing need in the production environment that
> expects CEP to support having multiple patterns in one operator and to
> support dynamically changing them. Therefore I propose to add certain
> infrastructure as described in FLIP-200 to support these functionalities.
>
> Please feel free to reply to this email thread. Looking forward to your
> feedback!
>
> [1]
> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=195730308
>
> Best regards,
>
> Yunfeng
>

[jira] [Created] (FLINK-24597) RocksdbStateBackend getKeysAndNamespaces would return duplicate data when using MapState

[Yue Ma (Jira)]

Yue Ma created FLINK-24597:
--

 Summary: RocksdbStateBackend getKeysAndNamespaces would return 
duplicate data when using MapState 
 Key: FLINK-24597
 URL: https://issues.apache.org/jira/browse/FLINK-24597
 Project: Flink
  Issue Type: Bug
  Components: API / State Processor, Runtime / State Backends
Affects Versions: 1.13.3, 1.12.4, 1.14.0
Reporter: Yue Ma
 Attachments: image-2021-10-20-14-23-20-333.png

For example, in RocksdbStateBackend , if we worked in VoidNamespace , and And 
use the ValueState like below .
{code:java}
// insert record
for (int i = 0; i < 3; ++i) {
keyedStateBackend.setCurrentKey(i);
testValueState.update(String.valueOf(i));
}
{code}
Then we get all the keysAndNamespace according the method 
RocksDBKeyedStateBackend#getKeysAndNamespaces().The result of the traversal is
 <1,VoidNamespace>,<2,VoidNamespace>,<3,VoidNamespace> ,which is as expected.

Thus，if we use MapState , and update the MapState with different user key, the 
getKeysAndNamespaces would return duplicate data with same keyAndNamespace.
{code:java}
// insert record
for (int i = 0; i < 3; ++i) {
keyedStateBackend.setCurrentKey(i);
mapState.put("userKeyA_" + i, "userValue");
mapState.put("userKeyB_" + i, "userValue");
}
{code}
The result of the traversal is
 
<1,VoidNamespace>,<1,VoidNamespace>,<2,VoidNamespace>,<2,VoidNamespace>,<3,VoidNamespace>,<3,VoidNamespace>.

By reading the code, I found that the main reason for this problem is in the 
implementation of _RocksStateKeysAndNamespaceIterator_.
In the _hasNext_ method, when a new keyAndNamespace is created, there is no 
comparison with the previousKeyAndNamespace. So we can refer to 
RocksStateKeysIterator to implement the same logic should solve this problem.
  
  



--
This message was sent by Atlassian Jira
(v8.3.4#803005)

[jira] [Created] (FLINK-23890) CepOperator may create a large number of timers and cause performance problems

[Yue Ma (Jira)]

Yue Ma created FLINK-23890:
--

 Summary: CepOperator may create a large number of timers and cause 
performance problems
 Key: FLINK-23890
 URL: https://issues.apache.org/jira/browse/FLINK-23890
 Project: Flink
  Issue Type: Improvement
  Components: Library / CEP
Affects Versions: 1.12.1
Reporter: Yue Ma
 Attachments: image-2021-08-20-13-59-05-977.png

 There are two situations in the CepOperator that may register the time when 
dealing with EventTime. 
when the processElement will buffer the data first, and then register a timer 
with a timestamp of watermark+1.
{code:java}
if (timestamp > timerService.currentWatermark()) {
 // we have an event with a valid timestamp, so
 // we buffer it until we receive the proper watermark.
 saveRegisterWatermarkTimer();
 bufferEvent(value, timestamp);
}{code}
The other is when the EventTimer is triggered, if sortedTimestamps or 
partialMatches are not empty, a timer will also be registered.
{code:java}
if (!sortedTimestamps.isEmpty() || !partialMatches.isEmpty()) {
 saveRegisterWatermarkTimer();
}{code}
 
The problem is, if the partialMatches corresponding to each of my keys are not 
empty. Then every time the watermark advances, the timers of all keys will be 
triggered, and then a new EventTimer is re-registered under each key. When the 
number of task keys is very large, this operation greatly affects performance.
!https://code.byted.org/inf/flink/uploads/91aee639553df07fa376cf2865e91fd2/image.png!
I think it is unnecessary to register EventTimer frequently like this and can 
we make the following changes?

When an event comes, the timestamp of the EventTimer we registered is equal to 
the EventTime of this event instead of watermark + 1.
When a new ComputionState with window is created (like *withIn* pattern ),  we 
use the timeout of this window to create EventTimer (EventTime + WindowTime). 

After making such an attempt in our test environment, the number of registered 
timers has been greatly reduced, and the performance has been greatly improved.

!https://code.byted.org/inf/flink/uploads/24b85492c6a34a35c4445a4fd46c8363/image.png!

 
 



--
This message was sent by Atlassian Jira
(v8.3.4#803005)

[jira] [Created] (FLINK-22888) Matches results may be wrong when using notNext as the last part of the pattern with Window

[Yue Ma (Jira)]

Yue Ma created FLINK-22888:
--

 Summary: Matches results may be wrong when using notNext as the 
last part of the pattern with Window
 Key: FLINK-22888
 URL: https://issues.apache.org/jira/browse/FLINK-22888
 Project: Flink
  Issue Type: Bug
  Components: Library / CEP
Affects Versions: 1.9.0
Reporter: Yue Ma


the pattern is like 
Pattern.begin("start").where(records == "a")

            .notNext("notNext").where(records == "b")

            .withIn(5milliseconds).

 

If there is only one event *"a"* in 5 milliseconds. I think this *“a”* should 
be output as the correct result of the match next time in advanceTime.

But in the actual operation of CEP. This “a” will be treated as matching 
timeout data
{code:java}
// code placeholder
@Test
public void testNoNextWithWindow() throws Exception {
   StreamExecutionEnvironment env = 
StreamExecutionEnvironment.getExecutionEnvironment();
   env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

   // (Event, timestamp)
   DataStream input = env.fromElements(
  Tuple2.of(new Event(1, "start", 1.0), 5L),

  // last element for high final watermark
  Tuple2.of(new Event(5, "final", 5.0), 100L)
   ).assignTimestampsAndWatermarks(new 
AssignerWithPunctuatedWatermarks>() {

  @Override
  public long extractTimestamp(Tuple2 element, long 
previousTimestamp) {
 return element.f1;
  }

  @Override
  public Watermark checkAndGetNextWatermark(Tuple2 
lastElement, long extractedTimestamp) {
 return new Watermark(lastElement.f1 - 5);
  }

   }).map(new MapFunction, Event>() {

  @Override
  public Event map(Tuple2 value) throws Exception {
 return value.f0;
  }
   });

   Pattern pattern = Pattern.begin("start").where(new 
SimpleCondition() {
  @Override
  public boolean filter(Event value) throws Exception {
 return value.getName().equals("start");
  }
   }).notNext("middle").where(new SimpleCondition() {
  @Override
  public boolean filter(Event value) throws Exception {
 return value.getName().equals("middle");
  }
   }).within(Time.milliseconds(5L));

   DataStream result = CEP.pattern(input, pattern).select(
  new PatternSelectFunction() {
 @Override
 public String select(Map> pattern) {
StringBuilder builder = new StringBuilder();
builder.append(pattern.get("start").get(0).getId());
return builder.toString();
 }
  }
   );

   List resultList = new ArrayList<>();

   DataStreamUtils.collect(result).forEachRemaining(resultList::add);

   resultList.sort(String::compareTo);

   assertEquals(Arrays.asList("1"), resultList);
}
{code}
 



--
This message was sent by Atlassian Jira
(v8.3.4#803005)