[jira] [Created] (KAFKA-9916) Materialize Table-Table Join Result to Avoid Performing Same Join Twice

["Bruno Cadonna (Jira)" <ji...@apache.org>]

Bruno Cadonna created KAFKA-9916:
------------------------------------

             Summary: Materialize Table-Table Join Result to Avoid Performing Same Join Twice
                 Key: KAFKA-9916
                 URL: https://issues.apache.org/jira/browse/KAFKA-9916
             Project: Kafka
          Issue Type: Improvement
          Components: streams
    Affects Versions: 2.5.0
            Reporter: Bruno Cadonna


If a table-table join processor performs a join and the join needs to forward downstream the old join result (e.g. due to an aggregation operation downstream), it performs the same join (i.e. calls the {{ValueJoiner}}) twice.

Given a left value {{L1}}, a right value {{R1}}, and a new right value {{R2}} with the same keys and input into the join operation in this order, the join processor at some point will join {{L1}} with {{R1}}. When the new right value {{R2}} triggers the join, it will join {{L1}} with {{R2}} and again {{L1}} with {{R1}}.

We could avoid calling the {{ValueJoiner}} twice by materializing the join result. We would trade a call to the {{ValueJoiner}} with a lookup into a state store. Depending on the logic in the {{ValueJoiner}} this may or may not improve the performance. However, calling the {{ValueJoiner}} once will only access the input values of the {{ValueJoiner}} once, which avoids the need to copy the input values each time the {{ValueJoiner}} is called. For example, consider the following {{ValueJoiner}}:

{code:java}
private ComplexValue eventFeesJoin(ComplexValue leftValue, Long rightValue) {
        leftValue.setSomeValue(rightValue);
        return leftValue;
}
{code}

With this {{ValueJoiner}}, {{setSomeValue(rightValue)}} will be called twice when {{R2}} trigger the join, the first time with {{R2}} and the second time with {{R1}}. That means, {{R2}} will be overwritten by {{R1}}, which is probably not what the users want. To get the correct result,  the {{ValueJoiner}} should be implemented as follows:
  
{code:java}
private ComplexValue eventFeesJoin(ComplexValue leftValue, Long rightValue) {
        ComplexValue copy = copy(leftValue);
        copy.setSomeValue(rightValue);
        return copy;
}
{code}

Copying values during joins could be avoided if the join result were materialized. 



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