title: KSQL: table key column is not the column used in the join criteria
body: <p>I'd like to join two KSQL tables with one to many relationship on a column that is different from the key.<br>
  How can I do that?<br>
  I'll give an example, a context, a solution I started working on, and the problems I encountered.</p>
  
  <h1>Example</h1>
  
  <p>A single team can have many players, and the join should be on <code>teams.name = players.team</code>:</p>
  
  <p><code>teams</code>:</p>
  
  <pre><code>| name     | founded | arena          | championships |
  |----------|---------|----------------|---------------|
  | eagles   | 2020    | at&amp;t arena     | 0             |
  | vultures | 2019    | verizon center | 1             |
  </code></pre>
  
  <p><code>players</code>:</p>
  
  <pre><code>| name     | team     | avg_points | born |
  |----------|----------|------------|------|
  | alice    | vultures | 30         | 2013 |
  | bob      | vultures | 25         | 2015 |
  | charlie  | eagles   | 20         | 2014 |
  </code></pre>
  
  <p>Desired result - <code>players_teams_enriched</code>:</p>
  
  <pre><code>| player   | team     | avg_points | born | team_founded | arena          | team_championships |
  |----------|----------|------------|------|--------------|----------------|--------------------|
  | alice    | vultures | 30         | 2013 | 2019         | verizon center | 1                  |
  | bob      | vultures | 25         | 2015 | 2019         | verizon center | 1                  |
  | charlie  | eagles   | 20         | 2014 | 2020         | at&amp;t arena     | 0                  |
  </code></pre>
  
  <h1>Context</h1>
  
  <h3>Set up the infrastructure</h3>
  
  <p>The following assumes you have confluent platform 5.4.0 and mysql db locally. Briefly:</p>
  
  <pre class="lang-sh prettyprint-override"><code># Run confluent platform in docker:
  git clone git@github.com:confluentinc/examples
  # add mysql driver and connector before starting docker compose, then:
  cd examples/cp-all-in-one &amp;&amp; docker-compose up -d
  # connect as a client:
  docker run -it --rm --name ksql-cli-1 --network cp-all-in-one_default confluentinc/cp-ksql-cli:5.4.0 http://ksql-server:8088
  
  # Run mysql in docker:
  # create the docker network if it doesn't exist yet
  docker network inspect cp-all-in-one_default &amp;&gt;/dev/null || docker network create --driver bridge cp-all-in-one_default
  # create the container
  docker run --name mysql --network cp-all-in-one_default -p 3306:3306 -e MYSQL_ROOT_PASSWORD=pass -e MYSQL_DATABASE=example -e MYSQL_USER=demouser -e MYSQL_PASSWORD=demopass -d mysql
  # connect as a client
  docker run -it --network cp-all-in-one_default --name mysql-client --rm mysql mysql -hmysql -udemouser -pdemopass example
  </code></pre>
  
  <h3>Reproducing</h3>
  
  <p>The <code>teams</code> and <code>players</code> table already exist in my relational database (e.g mysql):</p>
  
  <pre class="lang-sql prettyprint-override"><code>CREATE TABLE teams(
      name VARCHAR(128) NOT NULL PRIMARY KEY,
      founded INT NOT NULL,
      arena VARCHAR(128) NOT NULL,
      championships INT NOT NULL,
      lastupdated TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL ON UPDATE CURRENT_TIMESTAMP
  );
  
  CREATE TABLE players(
      name VARCHAR(128) NOT NULL PRIMARY KEY,
      team VARCHAR(128) NOT NULL,
      avg_points DOUBLE NOT NULL,
      born INT NOT NULL,
      lastupdated TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL ON UPDATE CURRENT_TIMESTAMP
  );
  
  INSERT INTO teams   (name, founded, arena, championships) VALUES ('eagles',   2020, 'at&amp;t arena',     0);
  INSERT INTO teams   (name, founded, arena, championships) VALUES ('vultures', 2019, 'verizon center', 1);
  INSERT INTO players (name, team, avg_points, born)        VALUES ('alice', 'vultures', 30, 2013);
  INSERT INTO players (name, team, avg_points, born)        VALUES ('bob', 'vultures', 25, 2015);
  INSERT INTO players (name, team, avg_points, born)        VALUES ('charlie', 'eagles', 20, 2014);
  </code></pre>
  
  <p>Then I create two connectors using KSQL CLI and create two KSQL tables (all the commands from now on will be in KSQL CLI):</p>
  
  <pre class="lang-sql prettyprint-override"><code>SET 'auto.offset.reset'='earliest';
  
  CREATE SOURCE CONNECTOR teams WITH(
      "connector.class"='io.confluent.connect.jdbc.JdbcSourceConnector',
      "connection.url"='jdbc:mysql://mysql:3306/example?user=demouser&amp;password=demopass',
      "mode"='timestamp',
      "table.whitelist"='teams',
      "timestamp.column.name"='lastupdated',
      "topic.prefix"='jdbc_',
      "transforms"='valueToKey,extractFieldFromKey',
      "transforms.valueToKey.type"='org.apache.kafka.connect.transforms.ValueToKey',
      "transforms.valueToKey.fields"='name',
      "transforms.extractFieldFromKey.type"='org.apache.kafka.connect.transforms.ExtractField$Key',
      "transforms.extractFieldFromKey.field"='name',
      "value.converter"='org.apache.kafka.connect.json.JsonConverter',
      "key.converter.schemas.enable"='false',
      "value.converter.schemas.enable"='false'
  );
  -- PRINT jdbc_teams FROM BEGINNING;
  
  CREATE SOURCE CONNECTOR players WITH(
      "connector.class"='io.confluent.connect.jdbc.JdbcSourceConnector',
      "connection.url"='jdbc:mysql://mysql:3306/example?user=demouser&amp;password=demopass',
      "mode"='timestamp',
      "table.whitelist"='players',
      "timestamp.column.name"='lastupdated',
      "topic.prefix"='jdbc_',
      "transforms"='valueToKey,extractFieldFromKey',
      "transforms.valueToKey.type"='org.apache.kafka.connect.transforms.ValueToKey',
      "transforms.valueToKey.fields"='name',
      "transforms.extractFieldFromKey.type"='org.apache.kafka.connect.transforms.ExtractField$Key',
      "transforms.extractFieldFromKey.field"='name',
      "value.converter"='org.apache.kafka.connect.json.JsonConverter',
      "key.converter.schemas.enable"='false',
      "value.converter.schemas.enable"='false'
  );
  -- PRINT jdbc_players FROM BEGINNING;
  
  CREATE TABLE teams(
      name VARCHAR,
      founded INT,
      arena VARCHAR,
      championships INT,
      lastupdated BIGINT
  ) WITH (
      KAFKA_TOPIC = 'jdbc_teams',
      VALUE_FORMAT='JSON',
      TIMESTAMP='lastupdated',
      KEY = 'name'
  );
  -- SELECT * FROM teams EMIT CHANGES;
  
  CREATE TABLE players(
      name VARCHAR,
      team VARCHAR,
      avg_points DOUBLE,
      born INT,
      lastupdated BIGINT
  ) WITH (
      KAFKA_TOPIC = 'jdbc_players',
      VALUE_FORMAT='JSON',
      TIMESTAMP='lastupdated',
      KEY = 'name'
  );
  -- SELECT * FROM players EMIT CHANGES;
  </code></pre>
  
  <p>Now I'd like to join the tables. In KSQL CLI:</p>
  
  <pre class="lang-sql prettyprint-override"><code>SELECT
      players.name as player,
      teams.name as team,
      avg_points,
      born,
      teams.founded as team_founded,
      arena,
      teams.championships as team_championships
  FROM
      teams INNER JOIN players ON teams.name = players.team
  EMIT CHANGES;
  </code></pre>
  
  <p>This results in an error:</p>
  
  <blockquote>
    <p>Source table (PLAYERS) key column (PLAYERS.NAME) is not the column used in the join criteria (PLAYERS.TEAM). Only the table's key column or 'ROWKEY' is supported in the join criteria.</p>
  </blockquote>
  
  <h1>Possible solution: collect_list()</h1>
  
  <p>One solution may be to "massage" the players table and key it by the team name instead of the player name,
  using <code>collect_list(COLUMN)</code> and then (after joining with the teams table) <code>EXPLODE(COLUMN) AS NEW_NAME</code>:</p>
  
  <pre class="lang-sql prettyprint-override"><code>SELECT
      team,
      collect_list(name) as players,
      collect_list(avg_points) as avg_points,
      collect_list(born) as born_dates
  FROM players
  GROUP BY team
  EMIT CHANGES;
  </code></pre>
  
  <p>Returns good results:</p>
  
  <pre><code>+---------+-------------+-------------+-------------+
  |TEAM     |PLAYERS      |AVG_POINTS   |BORN_DATES   |
  +---------+-------------+-------------+-------------+
  |vultures |[alice]      |[30.0]       |[2013]       |
  |vultures |[alice, bob] |[30.0, 25.0] |[2013, 2015] |
  |eagles   |[charlie]    |[20.0]       |[2014]       |
  </code></pre>
  
  <p>So we can create a table from the query above and then join on teams:</p>
  
  <pre class="lang-sql prettyprint-override"><code>CREATE TABLE teamplayers AS
      SELECT
          team,
          collect_list(name) as players,
          collect_list(avg_points) as avg_points,
          collect_list(born) as born_dates
      FROM players
      GROUP BY team;
  
  CREATE TABLE enriched_team_players AS
      SELECT
          team,
          players,
          avg_points,
          born_dates,
          founded,
          arena,
          championships
      FROM teamplayers INNER JOIN teams ON teamplayers.team = teams.name;
  
  SELECT * FROM enriched_team_players EMIT CHANGES;
  </code></pre>
  
  <p>The result is really close to the desired result, and all there's left to do (unless i'm missing something) is to explode the lists.
  Here is the result:</p>
  
  <pre><code>+--------------+---------+---------+-------------+-------------+-------------+--------+---------------+--------------+
  |ROWTIME       |ROWKEY   |TEAM     |PLAYERS      |AVG_POINTS   |BORN_DATES   |FOUNDED |ARENA          |CHAMPIONSHIPS |
  +--------------+---------+---------+-------------+-------------+-------------+--------+---------------+--------------+
  |1582017691000 |vultures |vultures |[alice, bob] |[30.0, 25.0] |[2013, 2015] |2019    |verizon center |1             |
  |1582017691000 |eagles   |eagles   |[charlie]    |[20.0]       |[2014]       |2020    |at&amp;t arena     |0             |
  </code></pre>
  
  <p>Now I'm trying to explode the results with the following query:</p>
  
  <pre class="lang-sql prettyprint-override"><code>CREATE TABLE desired_result AS
      SELECT
          EXPLODE(players) as player,
          team,
          EXPLODE(avg_points) as avg_points,
          EXPLODE(born_dates) as born,
          founded as team_founded,
          arena,
          championships as team_championships
      FROM enriched_team_players;
  </code></pre>
  
  <p>and I get this in the output:</p>
  
  <blockquote>
    <p>source</p>
  </blockquote>
  
  <h1>How to proceed?</h1>
  
  <p>Where is this <code>source</code> printed from?  </p>
  
  <p>When I <code>list tables;</code> I don't see any table called <code>desired_result</code>, and when I <code>list queries;</code> I don't see a new query that creates this table.<br>
  Why don't I get the results for <code>EXPLODE</code> as described in <a href="https://docs.confluent.io/current/ksql/docs/developer-guide/table-functions.html#table-functions" rel="nofollow noreferrer">table functions</a>?</p>
  
  <p>Is there another (perhaps simpler) way to get the desired result considering the schema that was given to me in the relational DB (which I don't control)?<br>
  The current way uses <a href="https://docs.confluent.io/current/ksql/docs/developer-guide/syntax-reference.html#aggregate-functions" rel="nofollow noreferrer">collect_list</a>, which (for the version 5.4.0 of ksql) supports up tp 1000 entries per row.<br>
  In the example above, no team contains more than 1000 players, however, in my real use case, in ~5% of the cases some rows are linked to roughly 10,000 entities.<br>
  Is the 1000 limit configurable?<br>
  Is there a solution that does not have this limitation?</p>