Hi Zoi, This is exactly the issue I had to deal with when implementing the initial join operator in SQL api.
The Tuple2<..> is a must for the dataflow case as it "encodes" the pairing of two inputs while preserving the sides. But of course this is the wrong algebra for the relational case. Prposal 1 seems like a better option IMO. Best Kaustubh On Mon, Jun 22, 2026 at 7:45 PM Zoi Kaoudi via dev <[email protected]> wrote: > Dear all, > > this is a long email, please bare with me. > > We have a fundamental and recurring issue with the output type of the > current Join operator. The Join operator as is now takes as input two > generic datatypes (potentially different) and outputs a Tuple2 of these > input datatypes. See here: > > > https://github.com/apache/wayang/blob/cb4bb0fd0245240124174b562e4ad1abbc9cd0c0/wayang-commons/wayang-basic/src/main/java/org/apache/wayang/basic/operators/JoinOperator.java#L37 > > This is fine for a general dataflow type of engine. It also enables > different types of joins, like joining specific java object (Employee with > Employee) with a custom join key. However, when we are handling relational > data and have JDBC platforms for execution this creates problems. > > *Problem 1* > When the inputs are Records, the join output is a Tuple2<Record, Record> > rather than a flat Record. Any relational operator we chain after the join > therefore cannot consume the output directly; the user has to insert a > flatten step (a MapOperator) to turn the Tuple2 into a single Record. This > is very cumbersome of course. > > *Problem 2* > Even if we "hide" a flatten map operator step somehow and we insert it > automatically, this step cannot run inside a database. So even when the > whole pipeline could execute on a JDBC platform, the flatten MapOperator > forces the optimizer to move data out of the database into the JVM just to > reshape it. This is ok as long as every in-database pipeline ends by > streaming its results back to Java anyway. However, if we want to have > fully in-database pipelines, this becomes problematic: For example, with an > in-database sink that writes the join result back into a table (CREATE > TABLE AS SELECT) and expects a Record as input, connecting the join to the > sink fails at plan construction because Tuple2 != Record, even though the > SQL engine would naturally produce a flat row. > > *Proposed solution 1* > We introduce a new join operator for relational data that takes two > Records and outputs a single, flat Record. On JDBC this maps directly to a > SQL join statement which already yields flat rows; on Java/Spark it > concatenates the two Records instead of wrapping them in a Tuple2. The > current Tuple2-based Join should stay for the general dataflow case, and > the Record-based one is used for the relational case. The same idea should > be extended to the other operators that share this wrapping behaviour (e.g. > the cartesian product). > Pros: flat types end to end and in-database pipelines can stay fully in > the database. > Cons: a second join operator to maintain and the relational/SQL path has > to target it. > > *Proposed solution 2* > We keep the current Join and add a new dedicated Wayang flatten operator > (eg FlattenOperator) that converts Tuple2<Record, Record> into a flat > Record, with a platform-specific implementations: on JDBC it essentially > adds an empty string (the SQL already produces flat rows, so it contributes > nothing to the query), while on Java/Spark it performs the actual > concatenation. Ideally the optimizer should insert it automatically > wherever the join will be executed. > Pros: reuses the existing, fully supported operators, with a single > mechanism. > Cons: we keep the Tuple2 in the logical plan and since the type check > happens before optimization, the automatic insertion may need to happen > somewhere else. > > I'd appreciate any thoughts on this and also any experiences you had and > how you solved them or how you would like to have them solved. > Also feel free to propose some other solution. > > Best > -- > Zoi >
