Pull query AVRO and PROTOBUF Key Format are not supporting Struct type group aggregation

[Taymindis]

When trying to use GROUP BY <STRUCT_DATA> , it causing schema 409 version issue when select pull query

Let's have a simple test case below

CREATE STREAM _xtest2 (
  ID INT KEY,
  custid int,
  zone string,
  pid int
)
WITH (
  KAFKA_TOPIC='xtest2',
  PARTITIONS=2,
  replicas=1,
  FORMAT = 'AVRO'
);


CREATE TABLE xtest2 (
  ID INT PRIMARY KEY, 
  CUSTID INT,
  zone string,
  pid int
) WITH (
  kafka_topic = 'xtest2',
  partitions = 2,
  replicas = 1,
  format = 'AVRO'
);


insert into _xtest2(id, custid, zone, pid) values (1,1, 'pro_zone', 1001);
insert into _xtest2(id, custid, zone, pid) values (2,2, 'pro_zone', 1002);
insert into _xtest2(id, custid, zone, pid) values (3,3, 'pro_zone', 1003);
insert into _xtest2(id, custid, zone, pid) values (4,4, 'pro_zone', 1004);
insert into _xtest2(id, custid, zone, pid) values (5,5, 'pro_zone', 1005);
insert into _xtest2(id, custid, zone, pid) values (6,6, 'pro_zone', 1006);
insert into _xtest2(id, custid, zone, pid) values (7,7, 'pro_zone', 1007);
insert into _xtest2(id, custid, zone, pid) values (8,8, 'pro_zone', 1008);
insert into _xtest2(id, custid, zone, pid) values (9,9, 'pro_zone', 1009);
insert into _xtest2(id, custid, zone, pid) values (10,10, 'pro_zone', 1010);


CREATE TABLE xtest_by_cust WITH (
  kafka_topic = 'xtest_by_cust',
  partitions = 2,
  replicas = 1,
  format = 'AVRO'
) AS 
SELECT STRUCT(custid:= custid, ZONE:=zone) as x, collect_list(pid) as ids FROM xtest2 
group by STRUCT(custid:= custid, ZONE:=zone);

When run the pull query, it should return result, but it cause a error

select * from xtest_by_cust where x = struct(CUSTID:=1, zone:='pro_zone');


Error starting pull query: Error serializing message to topic: _confluent-ksql-kafka-didiquery_CTAS_XTEST_BY_CUST_4547-Aggregate-GroupBy-repartition. Failed to access Avro data from topic _confluent-ksql-kafka-didiquery_CTAS_XTEST_BY_CUST_4547-Aggregate-GroupBy-repartition : Schema being registered is incompatible with an earlier schema for subject "_confluent-ksql-kafka-didiquery_CTAS_XTEST_BY_CUST_4547-Aggregate-GroupBy-repartition-key", details: [{errorType:'NAME_MISMATCH', description:'The name of the schema has changed (path '/name')', additionalInfo:'expected: io.confluent.ksql.avro_schemas.Xtest2Key'}, {oldSchemaVersion: 1}, {oldSchema: '{"type":"record","name":"Xtest2Key","namespace":"io.confluent.ksql.avro_schemas","fields":[{"name":"CUSTID","type":["null","int"],"default":null},{"name":"ZONE","type":["null","string"],"default":null}],"connect.name":"io.confluent.ksql.avro_schemas.Xtest2Key"}'}, {validateFields: 'false', compatibility: 'BACKWARD'}]; error code: 409; error code: 409 Hint: You probably forgot to add VALUE_SCHEMA_ID when creating the source.

But if you do in condition with more than 1 input, the result will come out, but it is not what i want, i can feel the bug

select * from xtest_by_cust where x in (struct(CUSTID:=1, zone:='pro_zone'),struct(CUSTID:=1, zone:='pro_zone'));