copr: migrate executor and aggr to be compatible with ChunkedVec

[skyzh]

What problem does this PR solve?

Problem Summary

This PR refactors co-processor framework internally. Now all data flowing in aggregator and executor uses the new BytesRef-like interface (or ChunkRef). This laid solid foundation for migrating to ChunkedVec.

Overview

In RFC: Using chunk format in coprocessor framework, we proposed a ChunkedVec struct to support Chunk-based computing. We plan to:

1. Migrate current co-processor framework to use interface compatible with ChunkedVec.

The detailed plan can be found at Full Chunk-based Computing PR Roadmap.

This PR follows #7979, #8019, #8070 . Now all internal stuff of copr uses NotChunkedVec instead of Vec<Option<T>>. As NotChunkedVec exposes nearly the same interface as our proposed ChunkedVec, we can directly replace this temporary struct with new ChunkedVec in next PR.

2. Introduce ChunkedVec.

After migrating co-processor to have interface compatible with proposed ChunkedVec, we can introduce ChunkedVec into TiKV. Then, we may further optimize performance by merging null parameters, etc.

What is changed and how it works?

Proposal: RFC: Using chunk format in coprocessor framework

Related Issue: #7724

What's Changed:

The biggest change in this PR is the introduction of NotChunkedVec. NotChunkedVec has nearly the same interface as the proposed ChunkedVec. All data flowing in copr now uses this new struct instead of Vec<Option<T>>.

tidb_query_codegen

• rpn_fn now generates code with NotChunkedVec.

tidb_query_datatype

• All internal logic now uses new interface.
• Some functions that take Json before is now refactored to take JsonRef.
• Add map_option_owned in collation, to support some aggregators.
• Implement AsMySQLBool on EvaluableRef types. (e.g. &T, BytesRef)
• Introduce NotChunkedVec.
• ChunkRef now implements a new function called phantom_data, which returns None of the corresponding EvaluableRef type. This is used to let compiler find dynamic dispatch function in aggregators. This will be explained later.
• Introduce UnsafeRefInto trait, that is used to cast reference into 'static ones.
• Add to_owned_value and from_owned_value for EvaluableRef.
• Remove AsMut for VectorValue.

NotChunkedVec

A vector type that internally stores Option<T>, but only allows access to Option<&T>. It can be constructed from Vec<Option<T>>, or &[Option<T>].

tidb_query_vec_aggr

• All data flowing in aggregators are now refactored to use impl ChunkRef. To make this process easier, the generic parameters are all set to 'static.
• We ensure there won't be invalid memory access as we don't store reference in any of these aggr structs.
• Corresponding traits now have naming scheme like update_balahbalah_unsafe.
• Add update, update_concreate, update_vector and update_repeat macros for safely calling these traits.
• As update_vector_unsafe now has a parameter with associative type T::ChunkedType, the compiler cannot find corresponding function by dynamic dispatching when AggrFunctionStateUpdatePartial is in a boxed type. In this way, we added a phantom_data of type Option<T> to this function, so as to make it compile. Users using update_vector! macro don't need to supply this phantom parameter, as this has been done in macro.

tidb_query_vec_executors

• Refactored to use ChunkRef internally.
• The most notable change is in top_n_executor, where an as_slice call is eliminated.

TODO

• In fact, the tidb_query_vec_aggr interface can be refactored without using unsafe primitives. This can be done later.
• In top_n_executor, we still can't eliminate the overhead of clone.
• We should remove as_slice calls from non-test code. This will be done later with a trait only available in tests.
• Use to_owned_value instead of .map(|x| x.as_ref()) for code not modified in this PR.

Check List

Tests

• Unit test

[skyzh]

@breeswish @TennyZhuang I've made the aggregation functions even safer by applying the following constraints.

Developers should use impl_state_update_partial to automatically implement unsafe parts

impl<T> super::AggrFunctionStateUpdatePartial<T> for AggrFnStateFirst<T>
where
    T: EvaluableRef<'static> + 'static,
    VectorValue: VectorValueExt<T::EvaluableType>,
{
    // ChunkedType has been implemented in AggrFunctionStateUpdatePartial<T1> for AggrFnStateFirst<T2>
    impl_state_update_partial! { T }
}

This macro will implement update_vector_unsafe, etc. which directly calls safe versions of these functions.

Developers should use generic lifetime in "safe" functions, so as to eliminate dangling pointer

    #[inline]
    fn update_vector<'a, TT, CC>(
        &mut self,
        ctx: &mut EvalContext,
        _phantom_data: Option<TT>,
        physical_values: CC,
        logical_rows: &[usize],
    ) -> Result<()>
    where
        TT: EvaluableRef<'a, EvaluableType = T::EvaluableType>,
        CC: ChunkRef<'a, TT>,
    {
        if let Some(physical_index) = logical_rows.first() {
            self.update(ctx, physical_values.get_option_ref(*physical_index))?;
        }
        Ok(())
    }

As we use a generic lifetime parameter, this code MUST be valid for all possible lifetime 'a. In this way, we can provide a "safe" interface to developers. If a developer stores a reference into the struct, there will be a compiler error.

The only exception is in impl_max_min, where we have to use an unsafe transmute to deal with types. As we ensure T and TT are reference to same type, except that they have different lifetime, this transmute can be proven safe.

    #[inline]
    fn update_concrete<'a, TT>(&mut self, _ctx: &mut EvalContext, value: Option<TT>) -> Result<()>
    where
        TT: EvaluableRef<'a, EvaluableType = T::EvaluableType> + Ord,
    {
        let extreme_ref = self
            .extremum_value
            .as_ref()
            .map(|x| TT::from_owned_value(unsafe { std::mem::transmute(x) }));
        if value.is_some() && (self.extremum_value.is_none() || extreme_ref.cmp(&value) == E::ORD) {
            self.extremum_value = value.map(|x| x.to_owned_value());
        }
        Ok(())
    }

[skyzh]

@zhongzc PTAL~ Thanks!

[zhongzc]

/bench +sysbench +tpcc +tpch

[sre-bot]

Benchmark Report

Run Sysbench Performance Test on VMs

@@                               Benchmark Diff                               @@
================================================================================
tidb: 2b0b34b88e43ad20f4e5ab1a0b5daf7ae6ff6046
--- tikv: 3a4a0c98f9efc2b409add8cb6ac9e8886bb5730c
+++ tikv: fa4dcea87ac91a0c1bdc5f7c506911b6b4100776
pd: 9f1a55c0e36dba3ec35447d16d58dfd9d6becb90
================================================================================
oltp_update_index:
    * QPS: 4872.17 ± 8.92% (std=311.74) delta: -1.12% (p=0.608)
    * Latency p50: 27.91 ± 3.06% (std=0.60) delta: -2.47%
    * Latency p99: 15.27 ± 0.00% (std=0.00) delta: 0.00%
            
oltp_insert:
    * QPS: 5087.93 ± 10.50% (std=333.46) delta: 3.33% (p=0.778)
    * Latency p50: 25.21 ± 10.89% (std=1.66) delta: -11.86%
    * Latency p99: 12.75 ± 0.00% (std=0.00) delta: 0.00%
            
oltp_read_write:
    * QPS: 9264.75 ± 0.68% (std=50.14) delta: 2.35% (p=0.726)
    * Latency p50: 283.03 ± 3.62% (std=6.11) delta: -1.62%
    * Latency p99: 168.63 ± 6.70% (std=8.08) delta: -18.96%
            
oltp_point_select:
    * QPS: 113577.59 ± 0.55% (std=417.93) delta: -0.24% (p=0.498)
    * Latency p50: 1.12 ± 0.44% (std=0.00) delta: 0.45%
    * Latency p99: 4.18 ± 0.00% (std=0.00) delta: -1.65%
            
oltp_update_non_index:
    * QPS: 6564.89 ± 9.87% (std=512.56) delta: 1.26% (p=0.845)
    * Latency p50: 19.61 ± 9.54% (std=1.52) delta: -1.16%
    * Latency p99: 12.98 ± 0.00% (std=0.00) delta: -0.00%
            

[skyzh]

Hi @zhongzc , it seems that tpcc has failed and tpch has been running 14 hrs without a result. Shall we trigger benchmark again?

[zhongzc]

/bench +tpcc +tpch

[sre-bot]

Benchmark Report

Run TPC-C Performance Test on VMs

@@                               Benchmark Diff                               @@
================================================================================
tidb: 563f2c92d76ddd0b07672f6cd1952bba77f7b1df
--- tikv: 11cbf057ba8dec3a0833a9948b5ae9ec54d6113d
+++ tikv: fa4dcea87ac91a0c1bdc5f7c506911b6b4100776
pd: 16492b5e2e983f50622b2bbc614b646977abf07f
================================================================================
Measured tpmC (NewOrders): 246.88 ± 36.53% (std=57.64), delta: 9.76% (p=0.946)

[codecov]

Codecov Report

Merging #8141 into master will decrease coverage by 0.11%.
The diff coverage is 86.48%.

@@            Coverage Diff             @@
##           master    #8141      +/-   ##
==========================================
- Coverage   84.77%   84.66%   -0.12%     
==========================================
  Files         553      553              
  Lines      137630   137702      +72     
==========================================
- Hits       116678   116580      -98     
- Misses      20952    21122     +170     

Impacted Files	Coverage Δ
components/raftstore/src/coprocessor/mod.rs	49.20% <ø> (ø)
components/test_raftstore/src/router.rs	55.55% <0.00%> (-5.06%)	⬇️
src/server/server.rs	86.82% <0.00%> (-1.37%)	⬇️
components/raftstore/src/store/msg.rs	20.52% <75.00%> (+2.47%)	⬆️
components/raftstore/src/coprocessor/dispatcher.rs	95.05% <77.77%> (-0.54%)	⬇️
components/raftstore/src/store/peer.rs	90.22% <85.71%> (-0.08%)	⬇️
components/raftstore/src/store/fsm/apply.rs	85.48% <87.50%> (+0.43%)	⬆️
components/cdc/src/observer.rs	88.69% <100.00%> (+1.73%)	⬆️
components/raftstore/src/router.rs	74.35% <100.00%> (+1.02%)	⬆️
components/raftstore/src/store/fsm/peer.rs	83.83% <100.00%> (+0.14%)	⬆️
... and 40 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 75edb25...8af3cc4. Read the comment docs.

[zhongzc]

LGTM

[ti-srebot]

@zhongzc,Thanks for your review.

[zhongzc]

/run-all-tests

[zhongzc]

/merge

[ti-srebot]

/run-all-tests