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

Reply via email to