Stream-stream joins

[Piotr Nowojski <pi...@data-artisans.com>]

Hi,

In Flink we are now discussing how to implement “enrichment joins” (stream to stream) in SQL. I have create a document with our proposal how to implement it [1] (relevant to this email is “Syntax / SQL / querying” section). I was working on top of the Julian’s document from 2016 [2].

In [2] proposal was to use the following syntax:

SELECT
  SUM(o.amount * r.rate) AS amount
FROM 
  Orders AS o
  LATERAL JOIN (TABLE Rates(o.rowtime)) AS r
ON r.currency = o.currency;

Because we in Flink we would like to have this feature as soon as possible, I proposed in [1] SQL 1999 complaint syntax (as far as I know) that’s also currently supported by Calcite:

SELECT
  SUM(o.amount * r.rate) AS amount
FROM 
  Orders AS o,
  LATERAL TABLE (Rates(o.rowtime)) AS r
WHERE r.currency = o.currency;

However I have also found “temporal tables” from SQL 2011 "FOR SYSTEM_TIME AS OF” syntax and proposal to expand SQL standard to handle stream-stream join case [3]:

 SELECT  *
    FROM Orders AS o
    JOIN LATERAL ProductPrices FOR SYSTEM_TIME AS OF O.orderTime AS P
      ON o.productId = p.productId

The question that I have is what’s the current official status from calcite regarding this feature? Is [3] the way to go in the foreseeable future? If so, is anyone working on it at the moment? 

Piotr Nowojski

[1] https://docs.google.com/document/d/1KaAkPZjWFeu-ffrC9FhYuxE6CIKsatHTTxyrxSBR8Sk/edit?ts=5b45fea6#heading=h.avxnnym51mbe <https://docs.google.com/document/d/1KaAkPZjWFeu-ffrC9FhYuxE6CIKsatHTTxyrxSBR8Sk/edit?ts=5b45fea6#heading=h.avxnnym51mbe>
[2] https://docs.google.com/document/d/1KaAkPZjWFeu-ffrC9FhYuxE6CIKsatHTTxyrxSBR8Sk/edit?ts=5b45fea6# <https://docs.google.com/document/d/1KaAkPZjWFeu-ffrC9FhYuxE6CIKsatHTTxyrxSBR8Sk/edit?ts=5b45fea6#>
[3] https://issues.apache.org/jira/browse/CALCITE-1917

Re: Stream-stream joins

[Julian Hyde <jh...@apache.org>]

Just to clarify the subject. We’re not talking about stream-to-stream joins but stream joining to a temporal table. It’s not an equi-join: when an order occurs, we look for the applicable rate for that order. If I am mistaken in this, please let me know.

All 3 syntaxes look reasonable in principle. (3 is definitely not implemented; I thought 1 is supported, especially after I fixed some bugs in LATERAL recently (see https://issues.apache.org/jira/browse/CALCITE-531 <https://issues.apache.org/jira/browse/CALCITE-531>), but if not, go with 2.)

I think you should go with whichever syntax works, and we can roll out other syntaxes in future.

The bigger question is what these queries look like in relational algebra. Let’s go with syntax 2 for now, and once we get the algebra working, then we can make other syntaxes map to it.

This is a hard problem, and I’ve not thought it through, so we will need to iterate. I have a feeling that Rates might be a table macro rather than a table function.

Julian

> On Jul 11, 2018, at 6:55 AM, Piotr Nowojski <pi...@data-artisans.com> wrote:
> 
> Hi,
> 
> In Flink we are now discussing how to implement “enrichment joins” (stream to stream) in SQL. I have create a document with our proposal how to implement it [1] (relevant to this email is “Syntax / SQL / querying” section). I was working on top of the Julian’s document from 2016 [2].
> 
> In [2] proposal was to use the following syntax:
> 
> SELECT
>  SUM(o.amount * r.rate) AS amount
> FROM 
>  Orders AS o
>  LATERAL JOIN (TABLE Rates(o.rowtime)) AS r
> ON r.currency = o.currency;
> 
> Because we in Flink we would like to have this feature as soon as possible, I proposed in [1] SQL 1999 complaint syntax (as far as I know) that’s also currently supported by Calcite:
> 
> SELECT
>  SUM(o.amount * r.rate) AS amount
> FROM 
>  Orders AS o,
>  LATERAL TABLE (Rates(o.rowtime)) AS r
> WHERE r.currency = o.currency;
> 
> However I have also found “temporal tables” from SQL 2011 "FOR SYSTEM_TIME AS OF” syntax and proposal to expand SQL standard to handle stream-stream join case [3]:
> 
> SELECT  *
>    FROM Orders AS o
>    JOIN LATERAL ProductPrices FOR SYSTEM_TIME AS OF O.orderTime AS P
>      ON o.productId = p.productId
> 
> The question that I have is what’s the current official status from calcite regarding this feature? Is [3] the way to go in the foreseeable future? If so, is anyone working on it at the moment? 
> 
> Piotr Nowojski
> 
> [1] https://docs.google.com/document/d/1KaAkPZjWFeu-ffrC9FhYuxE6CIKsatHTTxyrxSBR8Sk/edit?ts=5b45fea6#heading=h.avxnnym51mbe <https://docs.google.com/document/d/1KaAkPZjWFeu-ffrC9FhYuxE6CIKsatHTTxyrxSBR8Sk/edit?ts=5b45fea6#heading=h.avxnnym51mbe>
> [2] https://docs.google.com/document/d/1KaAkPZjWFeu-ffrC9FhYuxE6CIKsatHTTxyrxSBR8Sk/edit?ts=5b45fea6# <https://docs.google.com/document/d/1KaAkPZjWFeu-ffrC9FhYuxE6CIKsatHTTxyrxSBR8Sk/edit?ts=5b45fea6#>
> [3] https://issues.apache.org/jira/browse/CALCITE-1917