Hi, Jeyhun.
Thanks for the FLIP. Totally +1 for it.
I have a question about the part `Additional option to disable this
optimization`. Is this option a source configuration or a table
configuration?
Besides that, there is a little mistake if I do not understand wrongly.
Should `Check if upstream_any is pre-partitioned data source AND contains
the same partition keys as the source` be changed as `Check if upstream_any
is pre-partitioned data source AND contains the same partition keys as
downstream_any` ?
Best,
Hang
Jeyhun Karimov <je.kari...@gmail.com> 于2024年3月13日周三 21:11写道：
> Hi Jane,
>
> Thanks for your comments.
>
>
> 1. Concerning the `sourcePartitions()` method, the partition information
> > returned during the optimization phase may not be the same as the
> partition
> > information during runtime execution. For long-running jobs, partitions
> may
> > be continuously created. Is this FLIP equipped to handle scenarios?
>
>
> - Good point. This scenario is definitely supported.
> Once a new partition is added, or in general, new splits are
> discovered,
> PartitionAwareSplitAssigner::addSplits(Collection<FileSourceSplit>
> newSplits)
> method will be called. Inside that method, we are able to detect if a split
> belongs to existing partitions or there is a new partition.
> Once a new partition is detected, we add it to our existing mapping. Our
> mapping looks like Map<Integer, Set<Integer>> subtaskToPartitionAssignment,
> where
> it maps each source subtaskID to zero or more partitions.
>
> 2. Regarding the `RemoveRedundantShuffleRule` optimization rule, I
> > understand that it is also necessary to verify whether the hash key
> within
> > the Exchange node is consistent with the partition key defined in the
> table
> > source that implements `SupportsPartitioning`.
>
>
> - Yes, I overlooked that point, fixed. Actually, the rule is much
> complicated. I tried to simplify it in the FLIP. Good point.
>
>
> 3. Could you elaborate on the desired physical plan and integration with
> > `CompiledPlan` to enhance the overall functionality?
>
>
> - For compiled plan, PartitioningSpec will be used, with a json tag
> "Partitioning". As a result, in the compiled plan, the source operator will
> have
> "abilities" : [ { "type" : "Partitioning" } ] as part of the compiled plan.
> More about the implementation details below:
>
> --------------------------------
> PartitioningSpec class
> --------------------------------
> @JsonTypeName("Partitioning")
> public final class PartitioningSpec extends SourceAbilitySpecBase {
> // some code here
> @Override
> public void apply(DynamicTableSource tableSource, SourceAbilityContext
> context) {
> if (tableSource instanceof SupportsPartitioning) {
> ((SupportsPartitioning<?>) tableSource).applyPartitionedRead();
> } else {
> throw new TableException(
> String.format(
> "%s does not support SupportsPartitioning.",
> tableSource.getClass().getName()));
> }
> }
> // some code here
> }
>
> --------------------------------
> SourceAbilitySpec class
> --------------------------------
> @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include =
> JsonTypeInfo.As.PROPERTY, property = "type")
> @JsonSubTypes({
> @JsonSubTypes.Type(value = FilterPushDownSpec.class),
> @JsonSubTypes.Type(value = LimitPushDownSpec.class),
> @JsonSubTypes.Type(value = PartitionPushDownSpec.class),
> @JsonSubTypes.Type(value = ProjectPushDownSpec.class),
> @JsonSubTypes.Type(value = ReadingMetadataSpec.class),
> @JsonSubTypes.Type(value = WatermarkPushDownSpec.class),
> @JsonSubTypes.Type(value = SourceWatermarkSpec.class),
> @JsonSubTypes.Type(value = AggregatePushDownSpec.class),
> + @JsonSubTypes.Type(value = PartitioningSpec.class) //
> new added
>
>
>
> Please let me know if that answers your questions or if you have other
> comments.
>
> Regards,
> Jeyhun
>
>
> On Tue, Mar 12, 2024 at 8:56 AM Jane Chan <qingyue....@gmail.com> wrote:
>
> > Hi Jeyhun,
> >
> > Thank you for leading the discussion. I'm generally +1 with this
> proposal,
> > along with some questions. Please see my comments below.
> >
> > 1. Concerning the `sourcePartitions()` method, the partition information
> > returned during the optimization phase may not be the same as the
> partition
> > information during runtime execution. For long-running jobs, partitions
> may
> > be continuously created. Is this FLIP equipped to handle scenarios?
> >
> > 2. Regarding the `RemoveRedundantShuffleRule` optimization rule, I
> > understand that it is also necessary to verify whether the hash key
> within
> > the Exchange node is consistent with the partition key defined in the
> table
> > source that implements `SupportsPartitioning`.
> >
> > 3. Could you elaborate on the desired physical plan and integration with
> > `CompiledPlan` to enhance the overall functionality?
> >
> > Best,
> > Jane
> >
> > On Tue, Mar 12, 2024 at 11:11 AM Jim Hughes <jhug...@confluent.io.invalid
> >
> > wrote:
> >
> > > Hi Jeyhun,
> > >
> > > I like the idea! Given FLIP-376[1], I wonder if it'd make sense to
> > > generalize FLIP-434 to be about "pre-divided" data to cover "buckets"
> and
> > > "partitions" (and maybe even situations where a data source is
> > partitioned
> > > and bucketed).
> > >
> > > Separate from that, the page mentions TPC-H Q1 as an example. For a
> > join,
> > > any two tables joined on the same bucket key should provide a concrete
> > > example of a join. Systems like Kafka Streams/ksqlDB call this
> > > "co-partitioning"; for those systems, it is a requirement placed on the
> > > input sources. For Flink, with FLIP-434, the proposed planner rule
> > > could remove the shuffle.
> > >
> > > Definitely a fun idea; I look forward to hearing more!
> > >
> > > Cheers,
> > >
> > > Jim
> > >
> > >
> > > 1.
> > >
> > >
> >
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-376%3A+Add+DISTRIBUTED+BY+clause
> > > 2.
> > >
> > >
> >
> https://docs.ksqldb.io/en/latest/developer-guide/joins/partition-data/#co-partitioning-requirements
> > >
> > > On Sun, Mar 10, 2024 at 3:38 PM Jeyhun Karimov <je.kari...@gmail.com>
> > > wrote:
> > >
> > > > Hi devs,
> > > >
> > > > I’d like to start a discussion on FLIP-434: Support optimizations for
> > > > pre-partitioned data sources [1].
> > > >
> > > > The FLIP introduces taking advantage of pre-partitioned data sources
> > for
> > > > SQL/Table API (it is already supported as experimental feature in
> > > > DataStream API [2]).
> > > >
> > > >
> > > > Please find more details in the FLIP wiki document [1].
> > > > Looking forward to your feedback.
> > > >
> > > > Regards,
> > > > Jeyhun
> > > >
> > > > [1]
> > > >
> > > >
> > >
> >
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-434%3A+Support+optimizations+for+pre-partitioned+data+sources
> > > > [2]
> > > >
> > > >
> > >
> >
> https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/experimental/
> > > >
> > >
> >
>
