Hi all, Are non primary key joins supported by Flink sql where pkey and join key have M:N relationship? If it is supported, are there any cases which needs to be considered? Also, if I have a Kafka topic partitioned by pkey, to work without any issues should it only have single partition?

You can join on any key you want. You need to consider that for regular joins, your state size can grow indefinitely because a streaming join might mean that an update or delete needs to happen downstream. It’s better to evaluate if you can use things like a temporal join

How will it guarantee result accuracy? Consider a case where a topic has 2 partitions and join key is spread across these two partitions (because topic is partitioned on pkey). Now because of some reason, data from partition 1 is being read slow as compared to partition 2. Consider these messages t1 -> update[pkey1, jkey1, val1] and this is in partition 1 t2 -> update[pkey2, jkey1, val2] and this is in partition 2 now t2 message is read first and it updates the state and produced val2, and now when t1 message is being read after it will update it back to val1 while the correct result for the output was the latest value val2. Will this be overcome by event time processing?

First of all, it depends on the type of join you want. It all depends on your requirements

We are providing our users a wrapper over Flink sql to run any sql queries and are currently in the process of migrating from in house system earlier built using datastream api to this new system. In that system we used to have join key in the keyBy and I don’t have full understanding of how Flink sql keys data internally. The query in question is inner join on non pkey attributes.

Why not use the Table API?

The new system uses the Table API to build tables and descriptors from the avro schema and simply executes the queries (i.e flink sql/table api). We do create watermark based on the input message timestamp for the source table if users want event time processing.

So the only thing you’re interested in is understand how the Table API/SQL works under the hood?

yes, that will be able to help me understand if there are any special cases or things that should be considered.

The thing is, that I don’t think it’s feasible to know all the edge cases, especially since they heavily depend on the incoming data, formats, type of joins etc. If a join isn’t supported in your setup, Flink will throw and error. If there’s no error thrown yet it doesn’t work as expected, I think it’s safe to assume it’s a bug

thanks a lot. From our side, we will keep an eye out and if we discover any discrepancy in the output, we will raise bug on the official github. Have a nice and lovely day 🙂