perf(executor): implement dedup pk decoding for BatchQueryExecutor

[kwannoel]

What's changed and what's your intention?

• Summarize your change (mandatory)
  • Continued from perf(executor): decode row datums from pk #2957.
  • Adds dedup pk decoding to BatchQueryExecutor.
  • Changes pk_descs to pass by reference to DedupPkCellBasedTableIter. This avoids a clone of pk_descs each time we call the execute_inner method of BatchQueryExecutor.

Checklist

• I have written necessary docs and comments
• I have added necessary unit tests and integration tests
• All checks passed in ./risedev check (or alias, ./risedev c)

Refer to a related PR or issue link (optional)

related #588

[codecov]

Codecov Report

Merging #3060 (3f77a51) into main (c8c4acf) will increase coverage by 0.00%.
The diff coverage is 61.36%.

@@           Coverage Diff           @@
##             main    #3060   +/-   ##
=======================================
  Coverage   73.54%   73.54%           
=======================================
  Files         735      735           
  Lines      100309   100340   +31     
=======================================
+ Hits        73769    73793   +24     
- Misses      26540    26547    +7     

Flag	Coverage Δ
rust	73.54% <61.36%> (+<0.01%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
src/batch/src/executor/row_seq_scan.rs	23.27% <0.00%> (ø)
...ntend/src/optimizer/plan_node/stream_index_scan.rs	20.00% <0.00%> (-0.94%)	⬇️
src/stream/src/from_proto/batch_query.rs	32.05% <0.00%> (-2.20%)	⬇️
...frontend/src/optimizer/plan_node/batch_seq_scan.rs	95.76% <100.00%> (ø)
...ntend/src/optimizer/plan_node/stream_table_scan.rs	96.37% <100.00%> (+0.16%)	⬆️
src/storage/src/table/cell_based_table.rs	72.61% <100.00%> (ø)
src/storage/src/table/test_relational_table.rs	98.21% <100.00%> (ø)
src/stream/src/executor/batch_query.rs	91.42% <100.00%> (+1.31%)	⬆️
src/meta/src/hummock/mock_hummock_meta_client.rs	43.82% <0.00%> (-1.13%)	⬇️
src/storage/src/hummock/local_version_manager.rs	84.12% <0.00%> (-0.16%)	⬇️
... and 1 more

📣 Codecov can now indicate which changes are the most critical in Pull Requests. Learn more

[kwannoel]

rebased

[skyzh]

LGTM except the naming. By the way, I found that every time we need to access a table, our executors only get parts of its information. e.g., in this case, BatchExecutor only get column desc without ordered columns. We should have a single struct to describe a storage table in the future, maybe after the state table refactor.

related: #2212
cc @BowenXiao1999 @wcy-fdu

[skyzh]

The field name should be order_key instead? pk is different from order key in the streaming engine.

[BowenXiao1999]

Ineed. Maybe Meta/FE should embed this info into Stream Plan, so executor can just read it and build state table.

[kwannoel]

Updated: 2206ad5 Sorry realized it should be order_key rather than order_keys.

[kwannoel]

updated: https://github.com/singularity-data/risingwave/pull/3060/files/b4635acce4dbe7f03e41c6031d819e88acf5ffc1..598aadb3f6a233046c50c1ccfdcc0324a3fd4499

[skyzh]

StateTable will need:

• column desc (for encoding value and generate cell-based encoding)
• ordered column (for encoding key)
• distribution key (for calculating vnode)
  To read a table. Maybe we can make this a new StateTableDesc in proto, and pass it everywhere.