Hi Benoît, Before discussing all the options you listed, I'd like understand more about your requirements.
The part I don't fully understand is, both your fact (Event) and dimension (DimensionAtJoinTimeX) tables are coming from the same table, Event or EventRawInput in your case. So it will result that both your fact and dimension tables are changing with time. My understanding is, when your DimensionAtJoinTimeX table emit the results, you don't want to change the result again. You want the fact table only join whatever data currently the dimension table have? I'm asking because your dimension table was calculated with a window aggregation, but your join logic seems doesn't care about the time attribute (LEFT JOIN DimensionAtJoinTime1 d1 ON e.uid = d1.uid). It's possible that when a record with uid=x comes from Event table, but the dimension table doesn't have any data around uid=x yet due to the window aggregation. In this case, you don't want them to join? Best, Kurt On Fri, Jan 3, 2020 at 1:11 AM Benoît Paris < benoit.pa...@centraliens-lille.org> wrote: > Hello all! > > I'm trying to design a stream pipeline, and have trouble controlling when > a JOIN is triggering an update: > > Setup: > > - The Event table; "probe side", "query side", the result of earlier > stream processing > - The DimensionAtJoinTimeX tables; of updating nature, "build side", > the results of earlier stream processing > > Joining them: > > SELECT * > FROM Event e > LEFT JOIN DimensionAtJoinTime1 d1 > ON e.uid = d1.uid > LEFT JOIN DimensionAtJoinTime2 d2 > ON e.uid = d2.uid > > The DimensionAtJoinTimeX Tables being the result of earlier stream > processing, possibly from the same Event table: > > SELECT uid, > hop_start(...), > sum(...) > FROM Event e > GROUP BY uid, > hop(...) > > The Event Table being: > > SELECT ... > FROM EventRawInput i > WHERE i.some_field = 'some_value' > > Requirements: > > - I need the JOINs to only be executed once, only when a new line is > appended to the probe / query / Event table. > - I also need the full pipeline to be defined in SQL. > - I very strongly prefer the Blink planner (mainly for Deduplication, > TopN and LAST_VALUE features). > > Problem exploration so far: > > - Option 1, "FOR SYSTEM_TIME AS OF" [1]: I need to have the solution > in SQL: it doesn't work out. But I might explore the following: insert > DimensionAtJoinTimeX into a special Sink, and use it in a > LookupableTableSource (I'm at a loss on how to do that, though. Do I need > an external kv store?). > - Option 2, "FOR SYSTEM_TIME AS OF" [1], used in SQL: Is there a > version of "FOR SYSTEM_TIME AS OF" readily usable in SQL? I might have > missed something in the documentation. > - Option 3, "LATERAL TABLE table_function" [2], on the Legacy planner: > It does not work with two tables [3], and I don't get to have the Blink > planner features. > - Option 4, "LATERAL TABLE table_function" [2], on the Blink planner: > It does not work with the "probe side" being the results of earlier stream > processing [4]. > - Option 5, let a regular JOIN materialize the updates, and somehow > find how to filter the ones coming from the build sides (I'm at a loss on > how to do that). > - Option 6, "TVR": I read this paper [5], which mentions "Time-Varying > Relation"s; Speculating here: could there be a way, to say that the build > side is not a TVR. Aka declare the stream as being somehow "static", while > still being updated (but I guess we're back to "FOR SYSTEM_TIME AS OF"). > - Option 7: Is there some features being developed, or hints, or > workarounds to control the JOIN updates that I have not considered so far? > - Remark 1: I believe that FLINK-15112 and FLINK-14200 are of the same > bug nature, even though they occur in different situations on different > planners (same Exception Stack Trace on files that have the same historical > parent before the Blink fork). FLINK-15112 has a workaround, but > FLINK-14200 does not. The existence of that workaround IMHO signals that > there is a simple fix for both bugs. I have tried to find it in Flink for a > few days, but no success so far. If you guys have pointers helping me > provide a fix, I'll gladly listen. So far I have progressed to: It revolves > around Calcite-based Flink streaming rules transforming a temporal table > function correlate into a Join on 2*Scan, and crashes when it encounters > something that is not a table that can be readily scanned. Also, there are > shenanigans on trying to find the right schema in the Catalog. But I am > blocked now, and not accustomed to the Flink internal code (would like to > though, if Alibaba/Ververica are recruiting remote workers, wink wink, > nudge nudge). > > All opinions very much welcomed on all Options and Remarks! > > Cheers, and a happy new year to all, > Benoît > > [1] > https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/table/streaming/temporal_tables.html#temporal-table > > [2] > https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/table/streaming/joins.html#processing-time-temporal-joins > > [3] https://issues.apache.org/jira/browse/FLINK-15112 > > [4] https://issues.apache.org/jira/browse/FLINK-14200 > > [5] https://arxiv.org/pdf/1905.12133.pdf >
