According to @David Anderson’s answer to this question: https://stackoverflow.com/questions/73413870/flink-sql-watermark-strategy-after-join-operation, interval joins should have watermarks. However I tried a very simple interval join:

create view test4 as(select *, current_watermark(test1.order_time) as curr, if(test1.order_time > test2.order_time, test1.order_time, test2.order_time) as mt from test1 inner join test2 on test1.product = test2.product where test1.order_time BETWEEN test2.order_time - INTERVAL '1' SECOND and test2.order_time);

and the

desc test4

showed that the view doesn't have a watermark defiined. Did I mess up somewhere?

I'm not sure, but perhaps the output of

desc

doesn't accurately reflect whether or not the query result includes watermarks. To know for sure, you can use the view as input to a query that requires watermarks -- e.g., try applying event time windowing to the view and see if the planner will allow it.

Yep I think it does 🙂 I am still a bit confused on how the watermark gets generated past a interval join. Seems to be a delay of the interval length + 1 second. Is that right @David Anderson?

I think what happens is this: • A watermark arrives and triggers a timer, which causes the join operator to produce results. • The operator somehow decides what timestamp to put into the resulting streamrecords. • The operator then forwards the watermark.

and that timer is 1 second?

I just looked at the code; I was wrong about how this works. The actual implementation is more complicated/clever. Haven't figured out (yet?) how the timestamping and watermarking works. (But no, it wouldn't be periodic timers -- it would be event-time timers based on the records' timestamps and the interval involved.)

can you point me to the code as well?

Look for

IntervalJoinOperator

.

On this topic -- how do I make sure that FlinkSQL emits window results on every record instead of only after window end?

No, that’s not supported.

So I can never get results of my "current window" in my streaming program, so to speak

For the old-style GROUP BY windows (as opposed to the newer table-valued window functions) there’s an undocumented notion of a WindowEmitStrategy that supports allowed lateness, early firing, and late firing. For example, if you set

table.exec.emit.early-fire.delay

to 0, then all windows will be triggered on every element.

This can be a SQL config?

I think so. Set it on the cluster running the query.

org.apache.flink.table.api.TableException: Session window doesn't support EMIT strategy currently.

Yes, I noticed that restriction in the code.

What will be the result of this sorted by?

That doesn't surprise me. Many of the SQL operators behave non-deterministically (or in undefined ways), when the "correct" behavior is weakly defined.