confluentinc / confluentinc/ksql

N-way joins don't work when the join condition is an expression

Open
#9,237 2 comments 0 reactions 0 assignees View on GitHub
bug join-functions streaming-engine
Dominant language
Java
Stars
315
Forks
1k
Avg merge
1d 7h
Merged PRs (30d)
36

Description

**Describe the bug**
N-way joins don't work when the join condition is an expression rather than a ColumnReference.

Consider this example:
```
select * from tab1 inner join tab2 on tab1.id+1 = tab2.id inner join tab3 on tab1.id*2 = tab3.id emit changes;
Invalid Left Join Expression: Cannot find the select field in the available fields. field: `ID`, schema: [`TAB1_SUM` INTEGER, `TAB1_ROWTIME` BIGINT, `TAB1_ROWPARTITION` INTEGER, `TAB1_ROWOFFSET` BIGINT, `TAB1_ID` INTEGER, `TAB2_COUNT` BIGINT, `TAB2_ROWTIME` BIGINT, `TAB2_ROWPARTITION` INTEGER, `TAB2_ROWOFFSET` BIGINT, `TAB2_ID` INTEGER]. expression: (ID * 2), schema:`TAB1_ID` INTEGER KEY, `TAB1_SUM` INTEGER, `TAB1_ROWTIME` BIGINT, `TAB1_ROWPARTITION` INTEGER, `TAB1_ROWOFFSET` BIGINT, `TAB1_ID` INTEGER, `TAB2_COUNT` BIGINT, `TAB2_ROWTIME` BIGINT, `TAB2_ROWPARTITION` INTEGER, `TAB2_ROWOFFSET` BIGINT, `TAB2_ID` INTEGER
Statement: select * from tab1 inner join tab2 on tab1.id+1 = tab2.id inner join tab3 on tab1.id*2 = tab3.id emit changes;
ks
```

Notice that although the join conditions are on the keys of the tables, they are expressions hence they get translated to foreign key joins.

The problem is that after each binary join, we change the schema to include non-qualified identifiers by replacing `tab1.id` with `tab1_id`. When translating the next join of the N-way join, we try to parse the join expression and lookup the join condition arguments (`tab1.id`) in the output schema of the previous join and cannot find it.

**To Reproduce**
Steps to reproduce the behavior, include:
1. The version of KSQL.
Master
2. Sample source data.
```
CREATE STREAM s1 (id int key, col1 int, col2 int) WITH (kafka_topic='s1', format='json', partitions=1);
CREATE STREAM s2 (id int key, col1 int, col2 int) WITH (kafka_topic='s2', format='json', partitions=1);
create table tab1 as select id, sum(col1) as sum from s1 group by id;
create table tab2 as select id, count(col1) as count from s2 group by id;
```
4. Any SQL statements you ran
```
select * from tab1 inner join tab2 on tab1.id+1 = tab2.id inner join tab3 on tab1.id*2 = tab3.id emit changes;
```

**Expected behavior**
We should throw a more meaningful error message saying that complex expressions are not supported in N-way joins for now.

**Actual behaviour**
A clear and concise description of what actually happens, including:
```
Invalid Left Join Expression: Cannot find the select field in the available fields
```
**Additional context**
Add any other context about the problem here.

Contributor guide

Open the contributing guide

Research direction

Start with the provided N-way JOIN reproduction and trace how the join expression is translated after the first binary join, where the issue says the schema is changed. Review the existing handling for invalid left join expressions and add coverage for the SQL example; done means the unsupported complex expression produces a meaningful error instead of the reported field lookup failure.

Written by the indexing model from the issue text.

Assessment

Tech stack
java, sql
Domain
databases, stream-processing
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.