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 feat(nexmark source): generate events in a single source #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

• Imbalanced kafka source actors' throughput when running nexmark benchmark #5214
• feat: add distinct agg support to replace 2-phase rewriting #7682
• feat: support Append-only Group Top-N #7376
• improve common sub-plan detection #7360
• optimization: StreamHashJoin can fill cache on write directly without read when the join key contains input_pk #7290
• remove the unnecessary exchange before blackhole sink #7377
• feat: remove the redundant exchange after append-only source executor #7621
• perf: optimize to_char #7924
• Use may_exist in Hash Join #7938
• compaction seems to cause more-than-expected cache invalidation and leads to poor performance #8052
• eviction strategy can be aggressive sometimes unexpectedly #8065
• perf: merge identical agg calls in a streaming query #8101
• Tracking: Process time temporal join #8123
• feat: reduce unnecessary join keys as the logical pk of the join #8322
• revert #7957 to validate the performance regression of q7 and try to improve if true #8510
• fix(expr): do not construct error for extracting time subfield #8538
• perf(streaming): add read_prefix_len_hint in distinct agg table #8541
• Optimize SerializedKeySerializer::append #8683
• perf(array): avoid double bounds check in ArrayIterator #8685
• Optimize lru_cache get-and-put pattern in HashAgg #8687
• perf(encoding): add estimate size for value encoding #8692
• perf(parser): do to_ascii_lowercase only once #8718
• perf(connector): use Vec<u8> instead of Bytes and &[u8] #8732
• save the cost of building a new chunk when sinking #8798
• investigate the performance of HashKey #8802
• perf(common): optimize BitmapIter #8848
• Investigate use of compact in aggregation executor #9150
• perf(streaming): batching input for each group in the StreamHashAgg #9300
• perf: investigate the performance of DataChunkBuilder #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:

• Tracking: Watermark #6042

The queries

• perf: nexmark q0 #8712
• perf: nexmark q1 #7353 can remove the exchange before blackhole sink.
• perf: nexmark q2
• perf: nexmark q3 #8157
• bug: nexmark q4 performance degrade #6619
• perf: nexmark q5 #7343
• perf: nexmark q7 become slower and slower #7244 requests interval join, which is covered by band join RFC: Band Join rfcs#32
• perf: nexmark q8 #8158
• perf: nexmark q9 #7341 requests append-only Group Top-N (N = 1).
• perf: nexmark q10
• perf: nexmark q14 #8503
• perf: nexmark q15 #7350
• discussion: nexmark q16 low performance #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.
• perf: nexmark q17 #7351
• perf: nexmark q18 #7284 requests append-only Group Top-N (N = 1).
• perf: nexmark q20 #8160
• perf: nexmark q21 #8161
• perf: nexmark q22 #8162

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

• perf: nexmark q101 #8163
• perf: nexmark q102 #8164
• perf: nexmark q103 #8165
• perf: nexmark q104 #8166
• perf: nexmark q105 #8167
• perf: nexmark q106 #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.