[
https://issues.apache.org/jira/browse/KAFKA-7502?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16804449#comment-16804449
]
ASF GitHub Bot commented on KAFKA-7502:
---------------------------------------
dongjinleekr commented on pull request #6519: KAFKA-7502: Cleanup KTable
materialization logic in a single place
URL: https://github.com/apache/kafka/pull/6519
This PR is a follow-up of #6174 and #6453, which cleans up
`KTableImpl#doTransformValues` method.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Cleanup KTable materialization logic in a single place
> ------------------------------------------------------
>
> Key: KAFKA-7502
> URL: https://issues.apache.org/jira/browse/KAFKA-7502
> Project: Kafka
> Issue Type: Improvement
> Components: streams
> Reporter: Guozhang Wang
> Assignee: Lee Dongjin
> Priority: Major
>
> Today since we pre-create all the `KTableXXX` operator along with the logical
> node, we are effectively duplicating the logic to determine whether the
> resulted KTable should be materialized. More specifically, the
> materialization principle today is that:
> 1) If users specified Materialized in the DSL and it contains a queryable
> name. We always materialize.
> 2) If users specified Materialized in the DSL but not contains a queryable
> name, or if users do not specify a Materialized object at all, Streams may
> choose to materialize or not. But in any cases, even if the KTable is
> materialized it will not be queryable since there's no queryable name (i.e.
> only storeName is not null, but queryableName is null):
> 2.a) If the resulted KTable is from an aggregation, we always materialize
> since it is needed for storing the aggregation (i.e. we use the
> MaterializedInternal constructor with nameProvider != null).
> 2.b) If the resulted KTable is from a source topic, we delay the
> materialization until the downstream operator requires this KTable to be
> materialized or send-old-values (see `KTableSourceNode` and `KTableSource`).
> 2.c) If the resulted KTable if from a join, we always materialize if users
> creates a Materialized object even without a queryable name. However this can
> be optimized similar to 2.b) but is orthogonal to this ticket (see
> `KTableImpl#buildJoin` where we always use constructor with nameProvider !=
> null).
> 2.d) If the resulted KTable is from a stateless operation like filter /
> mapValues, we never materialize.
> ------------
> Now, in all of these cases, we have logical node like "KTableKTableJoinNode",
> as well as physical node like `ProcessorNode`. Ideally we should always
> create the logical Plan (i.e. the StreamsGraph), and then optimize it if
> necessary, and then generate the physical plan (i.e. the Topology), however
> today we create some physical nodes beforehand, and the above logic is hence
> duplicated in the creation of both physical nodes and logical nodes. For
> example, in `KTableKTableJoinNode` we check if Materialized is null for
> adding a state store, and in `KTableImpl#doJoin` we check if materialized is
> specified (case 2.c) above).
> Another example is in TableProcessorNode which is used for 2.d) above, in
> which it includes the logic whereas its caller, `KTableImpl#doFilter` for
> example, also contains the logic when deciding to pass `queryableName`
> parameter to `KTableProcessorSupplier`.
> This is bug-vulnerable since we may update the logic in one class but forgot
> to update the other class.
> --------------
> What we want to have is a cleaner code path similar to what we have for 2.b),
> such that when creating the logical nodes we keep track of whether 1)
> materialized is specified, and 2) queryable name is provided. And during
> optimization phase, we may change the inner physical ProcessorBuilder's
> parameters like queryable name etc, and then when it is time to generate the
> physical node, we can just blindly take the parameters and go for it.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
