You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@hudi.apache.org by Jack Vanlightly <va...@apache.org> on 2024/08/19 10:24:56 UTC
CDC and MOR tables
Hi all,
I'm trying to understand the CDC read process that Flink uses in Hudi.
According to the Flink option, READ_CDC_FROM_CHANGELOG (
https://github.com/apache/hudi/blob/db5c2d97dc94122ebd63e6200858eabc4b119178/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java#L365),
when True, Flink should only read from CDC files, when false, it must infer
the deltas from the base and log files. But reading the code, the
HoodieCDCExtractor creates file splits for both CDC files and for the
inference cases (
https://github.com/apache/hudi/blob/db5c2d97dc94122ebd63e6200858eabc4b119178/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCExtractor.java#L234)
when READ_CDC_FROM_CHANGELOG =True.
This is confusing as it seems you would do one or the other but not both.
Why go through all the work of inferring deltas from base and log files
when perhaps a couple of commits ahead there is a compact instant that has
it all precomputed?
Thanks
Jack
Re: CDC and MOR tables
Posted by Danny Chan <da...@apache.org>.
yeah, the InstantRange would include exact the commits that need to be read.
Best,
Danny
Jack Vanlightly <ja...@gmail.com> 于2024年9月5日周四 01:27写道：
>
> Ok, so the InstantRange would be an exact match one, with only the instants
> of compactions?
>
> On Mon, Aug 26, 2024, 02:46 Danny Chan <da...@apache.org> wrote:
>
> > > So if it works as you say, then there would have to be some other
> > filtering
> > going on somewhere.
> >
> > It is set up as intention, because for mor tables, a commit that is
> > not a compaction (insert overwrite etc.) should also be read.
> >
> > > it calls getCdcInputSplits(metaClient, instantRange) which does not use
> > the
> > configured IncrementalQueryAnalyzer
> >
> > It utilities the InstantRange which got the filtered instants by
> > IncrementalQueryAnalyzer, maybe you can add a test case and fire a fix
> > for it.
> >
> > Jack Vanlightly <va...@apache.org> 于2024年8月21日周三 16:32写道：
> > >
> > > Thanks Danny. I think there may be a problem, as reading the code and
> > > having written a test to check it, it seems this is not happening.
> > >
> > > The CDCExtractor itself is fed an instant range and a boolean
> > > (consumeChangesFromCompaction ) for whether to consume from compaction
> > > instants. When consumeChangesFromCompaction is True is reads from
> > COMPACT *and
> > > *the other instant types like COMMIT etc (
> > >
> > https://github.com/apache/hudi/blob/db5c2d97dc94122ebd63e6200858eabc4b119178/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCExtractor.java#L234
> > ).
> > > So if it works as you say, then there would have to be some other
> > filtering
> > > going on somewhere.
> > >
> > > I went to look for the code that would further filter down the instants
> > in
> > > some way. In the Flink code, the IncrementalInputSplits class uses the
> > > CDCExtractor. The interesting thing in this class is that the inputSplits
> > > method (
> > >
> > https://github.com/apache/hudi/blob/4b5c22339c7e412828b61469d8a037a6d73d2aa2/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java#L132
> > )
> > > uses the IncrementalQueryAnalyzer which is configured to read only from
> > the
> > > CDC changelog when the Flink option is enabled (
> > >
> > https://github.com/apache/hudi/blob/4b5c22339c7e412828b61469d8a037a6d73d2aa2/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java#L144C10-L144C30
> > ).
> > > However, importantly, when cdc is enabled (
> > >
> > https://github.com/apache/hudi/blob/4b5c22339c7e412828b61469d8a037a6d73d2aa2/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java#L315
> > ),
> > > it calls getCdcInputSplits(metaClient, instantRange) which does not use
> > the
> > > configured IncrementalQueryAnalyzer to source the timeline. It just uses
> > > the HoodieTableMetaClient, so there is no additional timeline filtering,
> > > only what there is within HoodieCDCExtractor.
> > >
> > > I wrote a MOR test case in TestIncrementalSplits and it does indeed
> > return
> > > inference based splits (LOG_FILE) even when told to only read from CDC
> > > change log. This looks like a bug to me.
> > >
> > > Thanks
> > > Jack
> > >
> > >
> > > Thanks
> > > Jack
> > >
> > > On Tue, Aug 20, 2024 at 6:29 AM Danny Chan <da...@apache.org> wrote:
> > >
> > > > yeah, you are right, for mor table, when the cdc log is enabled, which
> > > > are generated during compaction, there are two choices for the reader:
> > > >
> > > > 1. read the changes from the change log, which got a hight TTL delay
> > > > because these logs are only generated during compaction;
> > > > 2. or it can infers the changes by itself, which got much shorter TTL
> > > > delay but more resource cost on the reader side.
> > > >
> > > > 1 and 2 are mutual exclusion, you can only enable either one of them.
> > > >
> > > > The reason we add some tricky logic in HoodieCDCExtractor is that when
> > > > we choose 1, the timeline needs to be filtered to only include
> > > > compaction commits;
> > > > while when 2 is enabled, the timeline needs to be filtered to exlucde
> > > > compaction commits.
> > > >
> > > > Best,
> > > > Danny
> > > >
> > > > Jack Vanlightly <va...@apache.org> 于2024年8月19日周一 18:25写道：
> > > > >
> > > > > Hi all,
> > > > >
> > > > > I'm trying to understand the CDC read process that Flink uses in
> > Hudi.
> > > > > According to the Flink option, READ_CDC_FROM_CHANGELOG (
> > > > >
> > > >
> > https://github.com/apache/hudi/blob/db5c2d97dc94122ebd63e6200858eabc4b119178/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java#L365
> > > > ),
> > > > > when True, Flink should only read from CDC files, when false, it must
> > > > infer
> > > > > the deltas from the base and log files. But reading the code, the
> > > > > HoodieCDCExtractor creates file splits for both CDC files and for the
> > > > > inference cases (
> > > > >
> > > >
> > https://github.com/apache/hudi/blob/db5c2d97dc94122ebd63e6200858eabc4b119178/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCExtractor.java#L234
> > > > )
> > > > > when READ_CDC_FROM_CHANGELOG =True.
> > > > >
> > > > > This is confusing as it seems you would do one or the other but not
> > both.
> > > > > Why go through all the work of inferring deltas from base and log
> > files
> > > > > when perhaps a couple of commits ahead there is a compact instant
> > that
> > > > has
> > > > > it all precomputed?
> > > > >
> > > > > Thanks
> > > > > Jack
> > > >
> >
Re: CDC and MOR tables
Posted by Jack Vanlightly <ja...@gmail.com>.
Ok, so the InstantRange would be an exact match one, with only the instants
of compactions?
On Mon, Aug 26, 2024, 02:46 Danny Chan <da...@apache.org> wrote:
> > So if it works as you say, then there would have to be some other
> filtering
> going on somewhere.
>
> It is set up as intention, because for mor tables, a commit that is
> not a compaction (insert overwrite etc.) should also be read.
>
> > it calls getCdcInputSplits(metaClient, instantRange) which does not use
> the
> configured IncrementalQueryAnalyzer
>
> It utilities the InstantRange which got the filtered instants by
> IncrementalQueryAnalyzer, maybe you can add a test case and fire a fix
> for it.
>
> Jack Vanlightly <va...@apache.org> 于2024年8月21日周三 16:32写道：
> >
> > Thanks Danny. I think there may be a problem, as reading the code and
> > having written a test to check it, it seems this is not happening.
> >
> > The CDCExtractor itself is fed an instant range and a boolean
> > (consumeChangesFromCompaction ) for whether to consume from compaction
> > instants. When consumeChangesFromCompaction is True is reads from
> COMPACT *and
> > *the other instant types like COMMIT etc (
> >
> https://github.com/apache/hudi/blob/db5c2d97dc94122ebd63e6200858eabc4b119178/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCExtractor.java#L234
> ).
> > So if it works as you say, then there would have to be some other
> filtering
> > going on somewhere.
> >
> > I went to look for the code that would further filter down the instants
> in
> > some way. In the Flink code, the IncrementalInputSplits class uses the
> > CDCExtractor. The interesting thing in this class is that the inputSplits
> > method (
> >
> https://github.com/apache/hudi/blob/4b5c22339c7e412828b61469d8a037a6d73d2aa2/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java#L132
> )
> > uses the IncrementalQueryAnalyzer which is configured to read only from
> the
> > CDC changelog when the Flink option is enabled (
> >
> https://github.com/apache/hudi/blob/4b5c22339c7e412828b61469d8a037a6d73d2aa2/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java#L144C10-L144C30
> ).
> > However, importantly, when cdc is enabled (
> >
> https://github.com/apache/hudi/blob/4b5c22339c7e412828b61469d8a037a6d73d2aa2/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java#L315
> ),
> > it calls getCdcInputSplits(metaClient, instantRange) which does not use
> the
> > configured IncrementalQueryAnalyzer to source the timeline. It just uses
> > the HoodieTableMetaClient, so there is no additional timeline filtering,
> > only what there is within HoodieCDCExtractor.
> >
> > I wrote a MOR test case in TestIncrementalSplits and it does indeed
> return
> > inference based splits (LOG_FILE) even when told to only read from CDC
> > change log. This looks like a bug to me.
> >
> > Thanks
> > Jack
> >
> >
> > Thanks
> > Jack
> >
> > On Tue, Aug 20, 2024 at 6:29 AM Danny Chan <da...@apache.org> wrote:
> >
> > > yeah, you are right, for mor table, when the cdc log is enabled, which
> > > are generated during compaction, there are two choices for the reader:
> > >
> > > 1. read the changes from the change log, which got a hight TTL delay
> > > because these logs are only generated during compaction;
> > > 2. or it can infers the changes by itself, which got much shorter TTL
> > > delay but more resource cost on the reader side.
> > >
> > > 1 and 2 are mutual exclusion, you can only enable either one of them.
> > >
> > > The reason we add some tricky logic in HoodieCDCExtractor is that when
> > > we choose 1, the timeline needs to be filtered to only include
> > > compaction commits;
> > > while when 2 is enabled, the timeline needs to be filtered to exlucde
> > > compaction commits.
> > >
> > > Best,
> > > Danny
> > >
> > > Jack Vanlightly <va...@apache.org> 于2024年8月19日周一 18:25写道：
> > > >
> > > > Hi all,
> > > >
> > > > I'm trying to understand the CDC read process that Flink uses in
> Hudi.
> > > > According to the Flink option, READ_CDC_FROM_CHANGELOG (
> > > >
> > >
> https://github.com/apache/hudi/blob/db5c2d97dc94122ebd63e6200858eabc4b119178/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java#L365
> > > ),
> > > > when True, Flink should only read from CDC files, when false, it must
> > > infer
> > > > the deltas from the base and log files. But reading the code, the
> > > > HoodieCDCExtractor creates file splits for both CDC files and for the
> > > > inference cases (
> > > >
> > >
> https://github.com/apache/hudi/blob/db5c2d97dc94122ebd63e6200858eabc4b119178/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCExtractor.java#L234
> > > )
> > > > when READ_CDC_FROM_CHANGELOG =True.
> > > >
> > > > This is confusing as it seems you would do one or the other but not
> both.
> > > > Why go through all the work of inferring deltas from base and log
> files
> > > > when perhaps a couple of commits ahead there is a compact instant
> that
> > > has
> > > > it all precomputed?
> > > >
> > > > Thanks
> > > > Jack
> > >
>
Re: CDC and MOR tables
Posted by Danny Chan <da...@apache.org>.
> So if it works as you say, then there would have to be some other filtering
going on somewhere.
It is set up as intention, because for mor tables, a commit that is
not a compaction (insert overwrite etc.) should also be read.
> it calls getCdcInputSplits(metaClient, instantRange) which does not use the
configured IncrementalQueryAnalyzer
It utilities the InstantRange which got the filtered instants by
IncrementalQueryAnalyzer, maybe you can add a test case and fire a fix
for it.
Jack Vanlightly <va...@apache.org> 于2024年8月21日周三 16:32写道：
>
> Thanks Danny. I think there may be a problem, as reading the code and
> having written a test to check it, it seems this is not happening.
>
> The CDCExtractor itself is fed an instant range and a boolean
> (consumeChangesFromCompaction ) for whether to consume from compaction
> instants. When consumeChangesFromCompaction is True is reads from COMPACT *and
> *the other instant types like COMMIT etc (
> https://github.com/apache/hudi/blob/db5c2d97dc94122ebd63e6200858eabc4b119178/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCExtractor.java#L234).
> So if it works as you say, then there would have to be some other filtering
> going on somewhere.
>
> I went to look for the code that would further filter down the instants in
> some way. In the Flink code, the IncrementalInputSplits class uses the
> CDCExtractor. The interesting thing in this class is that the inputSplits
> method (
> https://github.com/apache/hudi/blob/4b5c22339c7e412828b61469d8a037a6d73d2aa2/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java#L132)
> uses the IncrementalQueryAnalyzer which is configured to read only from the
> CDC changelog when the Flink option is enabled (
> https://github.com/apache/hudi/blob/4b5c22339c7e412828b61469d8a037a6d73d2aa2/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java#L144C10-L144C30).
> However, importantly, when cdc is enabled (
> https://github.com/apache/hudi/blob/4b5c22339c7e412828b61469d8a037a6d73d2aa2/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java#L315),
> it calls getCdcInputSplits(metaClient, instantRange) which does not use the
> configured IncrementalQueryAnalyzer to source the timeline. It just uses
> the HoodieTableMetaClient, so there is no additional timeline filtering,
> only what there is within HoodieCDCExtractor.
>
> I wrote a MOR test case in TestIncrementalSplits and it does indeed return
> inference based splits (LOG_FILE) even when told to only read from CDC
> change log. This looks like a bug to me.
>
> Thanks
> Jack
>
>
> Thanks
> Jack
>
> On Tue, Aug 20, 2024 at 6:29 AM Danny Chan <da...@apache.org> wrote:
>
> > yeah, you are right, for mor table, when the cdc log is enabled, which
> > are generated during compaction, there are two choices for the reader:
> >
> > 1. read the changes from the change log, which got a hight TTL delay
> > because these logs are only generated during compaction;
> > 2. or it can infers the changes by itself, which got much shorter TTL
> > delay but more resource cost on the reader side.
> >
> > 1 and 2 are mutual exclusion, you can only enable either one of them.
> >
> > The reason we add some tricky logic in HoodieCDCExtractor is that when
> > we choose 1, the timeline needs to be filtered to only include
> > compaction commits;
> > while when 2 is enabled, the timeline needs to be filtered to exlucde
> > compaction commits.
> >
> > Best,
> > Danny
> >
> > Jack Vanlightly <va...@apache.org> 于2024年8月19日周一 18:25写道：
> > >
> > > Hi all,
> > >
> > > I'm trying to understand the CDC read process that Flink uses in Hudi.
> > > According to the Flink option, READ_CDC_FROM_CHANGELOG (
> > >
> > https://github.com/apache/hudi/blob/db5c2d97dc94122ebd63e6200858eabc4b119178/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java#L365
> > ),
> > > when True, Flink should only read from CDC files, when false, it must
> > infer
> > > the deltas from the base and log files. But reading the code, the
> > > HoodieCDCExtractor creates file splits for both CDC files and for the
> > > inference cases (
> > >
> > https://github.com/apache/hudi/blob/db5c2d97dc94122ebd63e6200858eabc4b119178/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCExtractor.java#L234
> > )
> > > when READ_CDC_FROM_CHANGELOG =True.
> > >
> > > This is confusing as it seems you would do one or the other but not both.
> > > Why go through all the work of inferring deltas from base and log files
> > > when perhaps a couple of commits ahead there is a compact instant that
> > has
> > > it all precomputed?
> > >
> > > Thanks
> > > Jack
> >
Re: CDC and MOR tables
Posted by Jack Vanlightly <va...@apache.org>.
Thanks Danny. I think there may be a problem, as reading the code and
having written a test to check it, it seems this is not happening.
The CDCExtractor itself is fed an instant range and a boolean
(consumeChangesFromCompaction ) for whether to consume from compaction
instants. When consumeChangesFromCompaction is True is reads from COMPACT *and
*the other instant types like COMMIT etc (
https://github.com/apache/hudi/blob/db5c2d97dc94122ebd63e6200858eabc4b119178/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCExtractor.java#L234).
So if it works as you say, then there would have to be some other filtering
going on somewhere.
I went to look for the code that would further filter down the instants in
some way. In the Flink code, the IncrementalInputSplits class uses the
CDCExtractor. The interesting thing in this class is that the inputSplits
method (
https://github.com/apache/hudi/blob/4b5c22339c7e412828b61469d8a037a6d73d2aa2/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java#L132)
uses the IncrementalQueryAnalyzer which is configured to read only from the
CDC changelog when the Flink option is enabled (
https://github.com/apache/hudi/blob/4b5c22339c7e412828b61469d8a037a6d73d2aa2/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java#L144C10-L144C30).
However, importantly, when cdc is enabled (
https://github.com/apache/hudi/blob/4b5c22339c7e412828b61469d8a037a6d73d2aa2/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java#L315),
it calls getCdcInputSplits(metaClient, instantRange) which does not use the
configured IncrementalQueryAnalyzer to source the timeline. It just uses
the HoodieTableMetaClient, so there is no additional timeline filtering,
only what there is within HoodieCDCExtractor.
I wrote a MOR test case in TestIncrementalSplits and it does indeed return
inference based splits (LOG_FILE) even when told to only read from CDC
change log. This looks like a bug to me.
Thanks
Jack
Thanks
Jack
On Tue, Aug 20, 2024 at 6:29 AM Danny Chan <da...@apache.org> wrote:
> yeah, you are right, for mor table, when the cdc log is enabled, which
> are generated during compaction, there are two choices for the reader:
>
> 1. read the changes from the change log, which got a hight TTL delay
> because these logs are only generated during compaction;
> 2. or it can infers the changes by itself, which got much shorter TTL
> delay but more resource cost on the reader side.
>
> 1 and 2 are mutual exclusion, you can only enable either one of them.
>
> The reason we add some tricky logic in HoodieCDCExtractor is that when
> we choose 1, the timeline needs to be filtered to only include
> compaction commits;
> while when 2 is enabled, the timeline needs to be filtered to exlucde
> compaction commits.
>
> Best,
> Danny
>
> Jack Vanlightly <va...@apache.org> 于2024年8月19日周一 18:25写道：
> >
> > Hi all,
> >
> > I'm trying to understand the CDC read process that Flink uses in Hudi.
> > According to the Flink option, READ_CDC_FROM_CHANGELOG (
> >
> https://github.com/apache/hudi/blob/db5c2d97dc94122ebd63e6200858eabc4b119178/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java#L365
> ),
> > when True, Flink should only read from CDC files, when false, it must
> infer
> > the deltas from the base and log files. But reading the code, the
> > HoodieCDCExtractor creates file splits for both CDC files and for the
> > inference cases (
> >
> https://github.com/apache/hudi/blob/db5c2d97dc94122ebd63e6200858eabc4b119178/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCExtractor.java#L234
> )
> > when READ_CDC_FROM_CHANGELOG =True.
> >
> > This is confusing as it seems you would do one or the other but not both.
> > Why go through all the work of inferring deltas from base and log files
> > when perhaps a couple of commits ahead there is a compact instant that
> has
> > it all precomputed?
> >
> > Thanks
> > Jack
>
Re: CDC and MOR tables
Posted by Danny Chan <da...@apache.org>.
yeah, you are right, for mor table, when the cdc log is enabled, which
are generated during compaction, there are two choices for the reader:
1. read the changes from the change log, which got a hight TTL delay
because these logs are only generated during compaction;
2. or it can infers the changes by itself, which got much shorter TTL
delay but more resource cost on the reader side.
1 and 2 are mutual exclusion, you can only enable either one of them.
The reason we add some tricky logic in HoodieCDCExtractor is that when
we choose 1, the timeline needs to be filtered to only include
compaction commits;
while when 2 is enabled, the timeline needs to be filtered to exlucde
compaction commits.
Best,
Danny
Jack Vanlightly <va...@apache.org> 于2024年8月19日周一 18:25写道：
>
> Hi all,
>
> I'm trying to understand the CDC read process that Flink uses in Hudi.
> According to the Flink option, READ_CDC_FROM_CHANGELOG (
> https://github.com/apache/hudi/blob/db5c2d97dc94122ebd63e6200858eabc4b119178/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java#L365),
> when True, Flink should only read from CDC files, when false, it must infer
> the deltas from the base and log files. But reading the code, the
> HoodieCDCExtractor creates file splits for both CDC files and for the
> inference cases (
> https://github.com/apache/hudi/blob/db5c2d97dc94122ebd63e6200858eabc4b119178/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCExtractor.java#L234)
> when READ_CDC_FROM_CHANGELOG =True.
>
> This is confusing as it seems you would do one or the other but not both.
> Why go through all the work of inferring deltas from base and log files
> when perhaps a couple of commits ahead there is a compact instant that has
> it all precomputed?
>
> Thanks
> Jack