# KsqlDB Table-Table join on STRUCT Type Primary Key

**URL:** <https://forum.confluent.io/t/ksqldb-table-table-join-on-struct-type-primary-key/9964>\
**Category:** ksqlDB\
**Created:** [30 January 2024 08:26 UTC](https://forum.confluent.io/t/ksqldb-table-table-join-on-struct-type-primary-key/9964 "2024-01-30T08:26:31Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![ghosh.sayak](https://avatars.discourse-cdn.com/v4/letter/g/d26b3c/32.png) [@ghosh.sayak](https://forum.confluent.io/u/ghosh.sayak)\
**Post date:** [30 January 2024 08:26 UTC](https://forum.confluent.io/t/ksqldb-table-table-join-on-struct-type-primary-key/9964/1 "2024-01-30T08:26:31Z")

</div>

We have 2 compact topics (C\_USERS [key - AID] and A\_USERS [key - account\_name]) both having AVRO Key & Value Schema with “record” type, and we want to do left join, take some attributes from A\_USERS topic and put the data to target compacted topic (USERS\_ENRICHED [key - AID])

And I am following this thread as [reference](https://www.confluent.io/blog/ksqldb-0-15-reads-more-message-keys-supports-more-data-types/#new-key-formats).

_Please Note_: The “A\_USERS” topic has been produced by a Transactional Producer.

Please find the steps we performed:

> A) For Topic: **C\_USERS**

1. 

```auto
CREATE OR REPLACE STREAM DEV1_KSTREAM_C_USERS WITH (KAFKA_TOPIC='dev-1-C_USERS', FORMAT='AVRO', VALUE_SCHEMA_ID=xxx, KEY_SCHEMA_ID=xxx);

```

1. 

```auto
CREATE OR REPLACE STREAM DEV1_KSTREAM_C_USERS_REKEYED WITH (KAFKA_TOPIC='dev-1-C_USERS_REKEYED', VALUE_FORMAT='AVRO', VALUE_SCHEMA_ID=xxx, KEY_FORMAT='AVRO', KEY_SCHEMA_ID=xxx) AS SELECT STRUCT(AID := ROWKEY->AID) AS ROWKEY,...<column_names> FROM DEV1_KSTREAM_C_USERS PARTITION BY STRUCT(AID := ROWKEY->AID) EMIT CHANGES;

```

1. 

```auto
CREATE OR REPLACE TABLE DEV1_KTABLE_C_USERS WITH (KAFKA_TOPIC='dev-1-C_USERS_REKEYED', VALUE_FORMAT='AVRO', VALUE_SCHEMA_ID=xxx, KEY_FORMAT='AVRO', KEY_SCHEMA_ID=xxx);

```

> B) For Topic: **A\_USERS**

1. 

```auto
SET 'processing.guarantee' = 'exactly_once_v2';
SET 'consumer.isolation.level' = 'read_committed';
CREATE OR REPLACE STREAM DEV1_KSTREAM_A_USERS WITH (KAFKA_TOPIC='dev-1-A_USERS', FORMAT='AVRO', VALUE_SCHEMA_ID=xxx, KEY_SCHEMA_ID=xxx);

```

1. 

```auto
CREATE OR REPLACE STREAM DEV1_KSTREAM_A_USERS_REKEYED WITH (KAFKA_TOPIC='dev-1-A_USERS_REKEYED', VALUE_FORMAT='AVRO', VALUE_SCHEMA_ID=xxx, KEY_FORMAT='AVRO', KEY_SCHEMA_ID=xxx) AS SELECT STRUCT(`account_name` := ROWKEY->`account_name`) AS ROWKEY,..., <column_names> FROM DEV1_KSTREAM_A_USERS PARTITION BY STRUCT(`account_name` := ROWKEY->`account_name`) EMIT CHANGES;

```

1. 

```auto
CREATE OR REPLACE TABLE DEV1_KTABLE_A_USERS WITH (KAFKA_TOPIC='dev-1-A_USERS_REKEYED', VALUE_FORMAT='AVRO', VALUE_SCHEMA_ID=xxx, KEY_FORMAT='AVRO', KEY_SCHEMA_ID=xxx);

```

C) Now **Join 2 KTables**

```auto
CREATE OR REPLACE TABLE 
    DEV1_KDEV_C_USERS_EVENTS_TABLE_KSQL_C_USERS_A_USERS
    WITH (
      KAFKA_TOPIC='dev-1-USERS_ENRICHED',
       PARTITIONS=1,
    REPLICAS=3,
    VALUE_FORMAT='AVRO',
    VALUE_SCHEMA_FULL_NAME='c.z.kafka.USERS_ENRICHED',
	VALUE_SCHEMA_ID=xxx,
	KEY_FORMAT='AVRO',
	KEY_SCHEMA_FULL_NAME='c.z.kafka.USERS_ENRICHED_KEY',
	KEY_SCHEMA_ID=xxx
  )
AS
  SELECT
    KT1.ROWKEY AS ROWKEY,
    KT1.AID AS AID,
	KT1.EMAIL_ADDRESS AS EMAIL_ADDRESS,
	KT1.PROFILE_CODE AS PROFILE_CODE,
	KT1.IS_DISABLED AS IS_DISABLED,
    KT2.`given_name` AS FIRST_NAME,
	KT2.`sn` AS LAST_NAME
  FROM
     DEV1_KTABLE_C_USERS KT1
  LEFT OUTER JOIN
    DEV1_KTABLE_A_USERS KT2
  ON
    STRUCT(AID := KT1.ROWKEY->AID) = STRUCT(`account_name` := KT2.ROWKEY->`account_name`)
EMIT CHANGES;

```

When I am trying to execute the JOIN query, I am getting the below error:

> Could not determine output schema for query due to error: Invalid join condition: table-table joins require to join on the primary key of the right input table. Got STRUCT(AID:=KT1.ROWKEY-\>ADID) = STRUCT(account\_name:=KT2.ROWKEY-\>account\_name).

And, if I join with the below changes -

```auto
CREATE OR REPLACE TABLE DEV1_KDEV_C_USERS_EVENTS_TABLE_KSQL_C_USERS_A_USERS
  WITH (
    KAFKA_TOPIC='dev-1-USERS_ENRICHED',
    PARTITIONS=1,
    REPLICAS=3,
    VALUE_FORMAT='AVRO',
    VALUE_SCHEMA_FULL_NAME='c.z.kafka.USERS_ENRICHED',
	VALUE_SCHEMA_ID=xxx,
	KEY_FORMAT='AVRO',
	KEY_SCHEMA_FULL_NAME='c.z.kafka.USERS_ENRICHED_KEY',
	KEY_SCHEMA_ID=xxx
  )
AS
  SELECT
    KT1.ROWKEY AS ROWKEY,
    KT1.AID AS AID,
	KT1.EMAIL_ADDRESS AS EMAIL_ADDRESS,
	KT1.PROFILE_CODE AS PROFILE_CODE,
	KT1.IS_DISABLED AS IS_DISABLED,
    KT2.`given_name` AS FIRST_NAME,
	KT2.`sn` AS LAST_NAME
  FROM
     DEV1_KTABLE_C_USERS KT1
  LEFT OUTER JOIN
    DEV1_KTABLE_A_USERS KT2
  ON
    --STRUCT(AID := KT1.ROWKEY->AID) = STRUCT(`account_name` := KT2.ROWKEY->`account_name`)
	KT1.ROWKEY=KT2.ROWKEY
EMIT CHANGES;

```

And I am getting error:

> Could not determine output schema for query due to error: Invalid join condition: types don’t match. Got KT1.ROWKEY{STRUCT\<`AID` STRING\>} = KT2.ROWKEY{STRUCT\<`account_name` STRING\>}

Am I missing something here? Kindly help!

**PLEASE NOTE** : It’s working fine while I am doing the join on _Primary Keys with Primitive Datatype_

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/flex019/uploads/confluentcommunity/original/1X/c49438c90c9df282e9996fdf6971be890c71b65a.svg) [@system](https://forum.confluent.io/u/system)\
**Post date:** [29 February 2024 08:26 UTC](https://forum.confluent.io/t/ksqldb-table-table-join-on-struct-type-primary-key/9964/2 "2024-02-29T08:26:38Z")

</div>

This topic was automatically closed after 30 days. New replies are no longer allowed.