This page is meant as a template for writing a KIP. To create a KIP choose Tools->Copy on this page and modify with your content and replace the heading with the next KIP number and a description of your issue. Replace anything in italics with your own description.

Status

Current state: [One of "Under Discussion", "Accepted", "Rejected"]

Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]

JIRA: Unable to render Jira issues macro, execution error.

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

A self-join is a join whose left and right-side arguments are the same entity (a stream reading from the same topic). Although self-joins are currently supported in Streams, their implementation is inefficient as they are implemented like regular joins where a state store is created for both left and right join arguments. Since both of these arguments represent the same entity, we don't need to create two state stores (as they will contain the exact same data) but only one. This optimization is only suitable for inner joins whose join condition is on the primary key. We do not consider foreign-key joins as we would need to create a state store for both arguments in order to be able to do efficient lookups. Hence, we will handle foreign-key self joins as regular inner foreign-key joins. Moreover, we do not consider outer joins since we are focusing on primary key joins and there will always be at least one join result, the current record joining with itself. 

One should consider this work as a logical plan optimization rule. If the graph contains a join whose arguments are the same entity, then we will rewrite it by applying the aforementioned self-join optimization that will create only one state store. 

This optimization only applies to Stream-Stream inner joins. A Table-Table inner self-join would not yield interesting results since a row would always join with itself. Moreover, the optimization does not apply to N-way self-joins as this notion doesn't exist. Joins are binary operators and once a self-join has been applied, the result contains records whose columns are the concatenation of the columns of the left and right join arguments. A subsequent join will be applied on the result of the previous join and not on the inputs of the previous join which means it won't be a self-join anymore. 

Public Interfaces

No public interfaces will be impacted but the config TOPOLOGY_OPTIMIZATION_CONFIG will be used to enable or disable this optimization along with the other ones that it already controls. This introduces backwards compatibility issues. However, this is the case with any optimization happening under the umbrella of this config and the decision has been that users who have this config turned on should be prepared to build new topologies when upgrading. Another way to go about this is to introduce a config per optimization. 

Proposed Changes

The changes required to implement this proposal are:

• Identify whether the arguments to a join are the same entity

The cases we are going to support are :

stream1 = builder.stream(INPUT_TOPIC);
stream1.join(stream1);

and

stream1 = builder.stream(INPUT_TOPIC);
stream2 = builder.stream(INPUT_TOPIC);
stream1.join(stream2);

The algorithm to identify whether two streams represent the same entity will first check if the StreamSourceNode nodes that are the join arguments are reading from the same topic(s) and then, will do a traversal of the graph and will check if the path from the StreamSourceNode node to the root does not have nodes that can change the semantics of the stream. Initially, most (all) nodes will be disallowed to be on the path. Later, we can look into adding some smarts into the algorithm to identify whether a node actually changes the data in the stream or not.  If both conditions hold, then the join can be implemented as a self join. 

• Implement the self-join operator 

The self-join operator will maintain one state store. For every new record, the record will first get added to the state store and then the operator will perform a lookup into the state store to do the actual join. This means that every record will at least join with itself.

• Add a rule to the optimizer that will rewrite an inner join to a self-join. The graph (logical plan) created from the DSL excerpts above are the same (after the optimization mergeDuplicateSourceNodes) and look as follows:

The graph gets translated into the following topology (physical plan)

The self-join rewriting will take the above graph, and will translate it into the following topology instead:

Compatibility, Deprecation, and Migration Plan

• Existing users with the optimization config turned on, will see their topologies failing. They will need to restart their applications to create the new topologies. This will not affect users that don't use optimizations.

Rejected Alternatives

Add to the DSL the operator selfJoin. We did not go with this approach as we prefer to push the complexity of the optimization to streams instead of to the user.

Pros:

1. This will make backwards-compatibility a non-issue as a user that upgrades from an older version does not have access to this DSL. If they want to use a self-join, they have to manually make the change in their code.
2. The code for a join can be chained like: builder.stream("topic").selfJoin().map(...)... If we don't have the operator, the code for a self-join needs to be broken into two parts: create the stream and get a reference, use the reference in a join like: stream.join(stream)
3. Implementation is straight-forward as we don't need to implement an optimization rule to do the rewriting

Cons:

1. The self-join is a physical plan optimization, it is not a different operator. We expose physical plan information to the user. 
2. We put the burden of creating an optimal topology on the user as they need to know about the self-join operator to use it. If they don't, then their topologies will be inefficient. That's the beauty of having it as a rewriting, users will take advantage of it without even realizing it.
3. Topologies may not be created by one user only and/or topologies may involve multiple operators that do all sorts of stuff. At some point in the code, the user has two variables that they join. They may not know that these refer to the same stream to realize that they can do a self-join instead of a regular join.

Another option that combines the benefits of both, is to implement both the DSL operator and the optimization rule. The downside of this approach is it could get confusing to the users as there would be multiple ways to achieve the same thing.