risingwave_common/test_utils/
rand_stream_chunk.rs

1// Copyright 2025 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use rand::{Rng, SeedableRng};
16
17use crate::array::stream_chunk::Op;
18use crate::array::{ArrayBuilder, ArrayImpl, I64ArrayBuilder};
19use crate::bitmap::Bitmap;
20
21pub fn gen_legal_stream_chunk(
22    bitmap: &Bitmap,
23    chunk_size: usize,
24    append_only: bool,
25    seed: u64,
26) -> (Vec<Op>, ArrayImpl) {
27    let mut data_builder = I64ArrayBuilder::new(chunk_size);
28    let mut ops: Vec<Op> = vec![];
29    let mut cur_data: Vec<i64> = vec![];
30    let mut rng = rand::rngs::StdRng::seed_from_u64(seed);
31    for i in 0..chunk_size {
32        // SAFETY(value_at_unchecked): the idx is always in bound.
33        if unsafe { bitmap.is_set_unchecked(i) } {
34            let op = if append_only || cur_data.is_empty() || rng.random() {
35                Op::Insert
36            } else {
37                Op::Delete
38            };
39            ops.push(op);
40            if op == Op::Insert {
41                let value = rng.random::<i32>() as i64;
42                data_builder.append(Some(value));
43                cur_data.push(value);
44            } else {
45                let idx = rng.random_range(0..cur_data.len());
46                data_builder.append(Some(cur_data[idx]));
47                cur_data.remove(idx);
48            }
49        } else {
50            ops.push(Op::Insert);
51            data_builder.append(Some(1234567890));
52        }
53    }
54    (ops, data_builder.finish().into())
55}
risingwave_common/test_utils/rand_stream_chunk.rs

risingwave_common/test_utils/
rand_stream_chunk.rs
