fix: properly set key when partition by ROWKEY and join on non-ROWKEY

[agavra]

fixes #4053

Description

Previously, if we were to join on a non-rowkey field in the source schema, and then partition by the original rowkey, the second operation was ignored. This fixes the issue by making sure that the alias is correct. This is a temporary fix and will need to be changed when we change how we handle aliasing.

Testing done

• new QTT tests
• checked for unnecessary repartitions

Reviewer checklist

• Ensure docs are updated if necessary. (eg. if a user visible feature is being added or changed).
• Ensure relevant issues are linked (description should include text like "Fixes #")

[purplefox]

LGTM

[big-andy-coates]

If this is fixing #4053, then it should mean we're applying the PARTITION BY to the output of the JOIN, so something like this should be possible right:

-- Given:
CREATE STREAM L (A STRING, B STRING) WITH (kafka_topic='LEFT', value_format='JSON', KEY='A');
CREATE STREAM R (C STRING, D STRING) WITH (kafka_topic='RIGHT', value_format='JSON', KEY='C');

-- Then can partition by an alias used in the projection:
CREATE STREAM OUTPUT AS SELECT L.ROWKEY AS RK, R.ROWKEY FROM L JOIN R WITHIN 10 SECONDS ON L.A = R.C PARTITION BY RK

Can you add a test to add if one doesn't already exist please?

Otherwise, LGTM.

[agavra]

@big-andy-coates - what you are suggesting is not possible. The source schema is the join schema, not the projection schema; the join schema is the union of all of the sources' fields. In your example, RK is not in the source join schema, but in the projection schema that is applied after the partition by step.