Tracking: Nexmark queries optimization

[lmatz]

Validation:

1. Compare the executable plan of RW and other systems. Make sure we are running the same/similar plan(unless it is some non-trivial plan-level optimization).
2. Make sure when benchmarking, the test is executed with a unified nexmark source instead of 3 separate sources, details in https://github.com/risingwavelabs/risingwave/issues/6747. In short, otherwise join will miss much more than normal and it leads to an ill-structured workload.
3. Also make sure that the query uses create sink XXX with ( connector = 'blackhole' ) instead of create materialized view as the other system doesn't have a materialized view. Although some plans shown in the sub-issues are still in the form of materialized view.

Optimization Tasks

• [ ] #5214
• [x] #7682
• [x] #7376
• [x] #7360
• [x] #7290
• [x] #7377
• [x] #7621
• [x] #7924
• [x] #7938
• [x] #8052
• [x] #8065
• [x] #8101
• [x] #8123
• [x] #8322
• [ ] #8510
• [x] #8538
• [x] #8541
• [x] #8683
• [x] #8685
• [x] #8687
• [x] #8692
• [x] #8718
• [x] #8732
• [x] #8798
• [x] #8802
• [ ] #8848
• [ ] #9150
• [ ] #9300
• [ ] #9301

Watermark

We notice that at https://github.com/nexmark/nexmark/blob/master/nexmark-flink/src/main/resources/queries/ddl_gen.sql#L37, Nexmark's source table has specified watermark as FOR dateTime AS dateTime - INTERVAL '4' SECOND. As watermark helps clean the state table and thus potentially improve the performance of state access, we also track the progress of watermark here:

• [ ] #6042

The queries

• [ ] #8712
• [x] #7353 can remove the exchange before blackhole sink.
• [ ] perf: nexmark q2
• [ ] #8157
• [x] #6619
• [ ] #7343
• [ ] #7244 requests interval join, which is covered by band join https://github.com/risingwavelabs/rfcs/pull/32
• [ ] #8158
• [ ] #7341 requests append-only Group Top-N (N = 1).
• [ ] perf: nexmark q10
• [x] #8503
• [x] #7350
• [x] #7271 requests shuffle aggregation instead of 2-phase when encountering distinct aggregation with Group By. After this, all aggregation with Group By(non-simple) will no longer choose 2-phase aggregation.
• [x] #7351
• [x] #7284 requests append-only Group Top-N (N = 1).
• [ ] #8160
• [x] #8161
• [x] #8162

Queries made up by @BugenZhao to cover stream operators that are not covered by the standard Nexmark:

• [ ] #8163
• [x] #8164
• [x] #8165
• [x] #8166
• [x] #8167
• [x] #8263

[lmatz]

q15, q16, q17 have similar query patterns, and they may suffer from the same problem.

[lmatz]

Q6 requires AVG over. Flink would output error when running this query. Q11 requires session_start. Q12 requires proc_time. Q13 requires proc_time. Q14 requires count_char UDF.

By 4/4, Q19 is supported but not enabled on the performance dashboard.