risingwave_common/util/
memcmp_encoding.rs

1// Copyright 2023 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 std::ops::Deref;
16
17use bytes::{Buf, BufMut};
18use itertools::Itertools;
19use risingwave_common_estimate_size::EstimateSize;
20use serde::{Deserialize, Serialize};
21
22use super::iter_util::{ZipEqDebug, ZipEqFast};
23use crate::array::{ArrayImpl, DataChunk, VectorItemType};
24use crate::row::{OwnedRow, Row};
25use crate::types::{
26    DataType, Date, Datum, F32, F64, Int256, ScalarImpl, Serial, Time, Timestamp, Timestamptz,
27    ToDatumRef,
28};
29use crate::util::sort_util::{ColumnOrder, OrderType};
30
31// NULL > any non-NULL value by default
32const DEFAULT_NULL_TAG_NONE: u8 = 1;
33const DEFAULT_NULL_TAG_SOME: u8 = 0;
34
35pub(crate) fn serialize_datum(
36    datum: impl ToDatumRef,
37    order: OrderType,
38    serializer: &mut memcomparable::Serializer<impl BufMut>,
39) -> memcomparable::Result<()> {
40    serializer.set_reverse(order.is_descending());
41    let (null_tag_none, null_tag_some) = if order.nulls_are_largest() {
42        (1u8, 0u8) // None > Some
43    } else {
44        (0u8, 1u8) // None < Some
45    };
46    if let Some(scalar) = datum.to_datum_ref() {
47        null_tag_some.serialize(&mut *serializer)?;
48        scalar.serialize(serializer)?;
49    } else {
50        null_tag_none.serialize(serializer)?;
51    }
52    Ok(())
53}
54
55pub(crate) fn serialize_datum_in_composite(
56    datum: impl ToDatumRef,
57    serializer: &mut memcomparable::Serializer<impl BufMut>,
58) -> memcomparable::Result<()> {
59    // NOTE: No need to call `serializer.set_reverse` because we are inside a
60    // composite type value, we should follow the outside order, except for `NULL`s.
61    if let Some(scalar) = datum.to_datum_ref() {
62        DEFAULT_NULL_TAG_SOME.serialize(&mut *serializer)?;
63        scalar.serialize(serializer)?;
64    } else {
65        DEFAULT_NULL_TAG_NONE.serialize(serializer)?;
66    }
67    Ok(())
68}
69
70pub(crate) fn deserialize_datum(
71    ty: &DataType,
72    order: OrderType,
73    deserializer: &mut memcomparable::Deserializer<impl Buf>,
74) -> memcomparable::Result<Datum> {
75    deserializer.set_reverse(order.is_descending());
76    let null_tag = u8::deserialize(&mut *deserializer)?;
77    let (null_tag_none, null_tag_some) = if order.nulls_are_largest() {
78        (1u8, 0u8) // None > Some
79    } else {
80        (0u8, 1u8) // None < Some
81    };
82    if null_tag == null_tag_none {
83        Ok(None)
84    } else if null_tag == null_tag_some {
85        Ok(Some(ScalarImpl::deserialize(ty, deserializer)?))
86    } else {
87        Err(memcomparable::Error::InvalidTagEncoding(null_tag as _))
88    }
89}
90
91pub(crate) fn deserialize_datum_in_composite(
92    ty: &DataType,
93    deserializer: &mut memcomparable::Deserializer<impl Buf>,
94) -> memcomparable::Result<Datum> {
95    // NOTE: Similar to serialization, we should follow the outside order, except for `NULL`s.
96    let null_tag = u8::deserialize(&mut *deserializer)?;
97    if null_tag == DEFAULT_NULL_TAG_NONE {
98        Ok(None)
99    } else if null_tag == DEFAULT_NULL_TAG_SOME {
100        Ok(Some(ScalarImpl::deserialize(ty, deserializer)?))
101    } else {
102        Err(memcomparable::Error::InvalidTagEncoding(null_tag as _))
103    }
104}
105
106/// Deserialize the `data_size` of `input_data_type` in `memcmp_encoding`. This function will
107/// consume the offset of deserializer then return the length (without memcopy, only length
108/// calculation).
109pub(crate) fn calculate_encoded_size(
110    ty: &DataType,
111    order: OrderType,
112    encoded_data: &[u8],
113) -> memcomparable::Result<usize> {
114    let mut deserializer = memcomparable::Deserializer::new(encoded_data);
115    let (null_tag_none, null_tag_some) = if order.nulls_are_largest() {
116        (1u8, 0u8) // None > Some
117    } else {
118        (0u8, 1u8) // None < Some
119    };
120    deserializer.set_reverse(order.is_descending());
121    calculate_encoded_size_inner(ty, null_tag_none, null_tag_some, &mut deserializer)
122}
123
124fn calculate_encoded_size_inner(
125    ty: &DataType,
126    null_tag_none: u8,
127    null_tag_some: u8,
128    deserializer: &mut memcomparable::Deserializer<impl Buf>,
129) -> memcomparable::Result<usize> {
130    let base_position = deserializer.position();
131    let null_tag = u8::deserialize(&mut *deserializer)?;
132    if null_tag == null_tag_none {
133        // deserialize nothing more
134    } else if null_tag == null_tag_some {
135        use std::mem::size_of;
136        let len = match ty {
137            DataType::Int16 => size_of::<i16>(),
138            DataType::Int32 => size_of::<i32>(),
139            DataType::Int64 => size_of::<i64>(),
140            DataType::Serial => size_of::<Serial>(),
141            DataType::Float32 => size_of::<F32>(),
142            DataType::Float64 => size_of::<F64>(),
143            DataType::Date => size_of::<Date>(),
144            DataType::Time => size_of::<Time>(),
145            DataType::Timestamp => size_of::<Timestamp>(),
146            DataType::Timestamptz => size_of::<Timestamptz>(),
147            DataType::Boolean => size_of::<u8>(),
148            // Interval is serialized as (i32, i32, i64)
149            DataType::Interval => size_of::<(i32, i32, i64)>(),
150            DataType::Decimal => {
151                deserializer.deserialize_decimal()?;
152                0 // the len is not used since decimal is not a fixed length type
153            }
154            // these types are var-length and should only be determine at runtime.
155            // TODO: need some test for this case (e.g. e2e test)
156            DataType::List { .. } | DataType::Map(_) => deserializer.skip_bytes()?,
157            DataType::Vector(d) => d * size_of::<VectorItemType>(),
158            DataType::Struct(t) => t
159                .types()
160                .map(|field| {
161                    // use default null tags inside composite type
162                    calculate_encoded_size_inner(
163                        field,
164                        DEFAULT_NULL_TAG_NONE,
165                        DEFAULT_NULL_TAG_SOME,
166                        deserializer,
167                    )
168                })
169                .try_fold(0, |a, b| b.map(|b| a + b))?,
170            DataType::Jsonb => deserializer.skip_bytes()?,
171            DataType::Variant => deserializer.skip_bytes()?,
172            DataType::Varchar => deserializer.skip_bytes()?,
173            DataType::Bytea => deserializer.skip_bytes()?,
174            DataType::Int256 => Int256::MEMCMP_ENCODED_SIZE,
175        };
176
177        // consume offset of fixed_type
178        if deserializer.position() == base_position + 1 {
179            // fixed type
180            deserializer.advance(len);
181        }
182    } else {
183        return Err(memcomparable::Error::InvalidTagEncoding(null_tag as _));
184    }
185
186    Ok(deserializer.position() - base_position)
187}
188
189#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord, EstimateSize)]
190pub struct MemcmpEncoded(Box<[u8]>);
191
192impl MemcmpEncoded {
193    pub fn as_inner(&self) -> &[u8] {
194        &self.0
195    }
196
197    pub fn into_inner(self) -> Box<[u8]> {
198        self.0
199    }
200}
201
202impl AsRef<[u8]> for MemcmpEncoded {
203    fn as_ref(&self) -> &[u8] {
204        &self.0
205    }
206}
207
208impl Deref for MemcmpEncoded {
209    type Target = [u8];
210
211    fn deref(&self) -> &Self::Target {
212        &self.0
213    }
214}
215
216impl IntoIterator for MemcmpEncoded {
217    type IntoIter = std::vec::IntoIter<Self::Item>;
218    type Item = u8;
219
220    fn into_iter(self) -> Self::IntoIter {
221        self.0.into_vec().into_iter()
222    }
223}
224
225impl FromIterator<u8> for MemcmpEncoded {
226    fn from_iter<T: IntoIterator<Item = u8>>(iter: T) -> Self {
227        Self(iter.into_iter().collect())
228    }
229}
230
231impl From<Vec<u8>> for MemcmpEncoded {
232    fn from(v: Vec<u8>) -> Self {
233        Self(v.into_boxed_slice())
234    }
235}
236
237impl From<Box<[u8]>> for MemcmpEncoded {
238    fn from(v: Box<[u8]>) -> Self {
239        Self(v)
240    }
241}
242
243impl From<MemcmpEncoded> for Vec<u8> {
244    fn from(v: MemcmpEncoded) -> Self {
245        v.0.into()
246    }
247}
248
249impl From<MemcmpEncoded> for Box<[u8]> {
250    fn from(v: MemcmpEncoded) -> Self {
251        v.0
252    }
253}
254
255/// Encode a datum into memcomparable format.
256pub fn encode_value(
257    value: impl ToDatumRef,
258    order: OrderType,
259) -> memcomparable::Result<MemcmpEncoded> {
260    let mut serializer = memcomparable::Serializer::new(vec![]);
261    serialize_datum(value, order, &mut serializer)?;
262    Ok(serializer.into_inner().into())
263}
264
265/// Decode a datum from memcomparable format.
266pub fn decode_value(
267    ty: &DataType,
268    encoded_value: &[u8],
269    order: OrderType,
270) -> memcomparable::Result<Datum> {
271    let mut deserializer = memcomparable::Deserializer::new(encoded_value);
272    deserialize_datum(ty, order, &mut deserializer)
273}
274
275/// Encode an array into memcomparable format.
276pub fn encode_array(
277    array: &ArrayImpl,
278    order: OrderType,
279) -> memcomparable::Result<Vec<MemcmpEncoded>> {
280    let mut data = Vec::with_capacity(array.len());
281    for datum in array.iter() {
282        data.push(encode_value(datum, order)?);
283    }
284    Ok(data)
285}
286
287/// Encode a chunk into memcomparable format.
288pub fn encode_chunk(
289    chunk: &DataChunk,
290    column_orders: &[ColumnOrder],
291) -> memcomparable::Result<Vec<MemcmpEncoded>> {
292    let encoded_columns: Vec<_> = column_orders
293        .iter()
294        .map(|o| encode_array(chunk.column_at(o.column_index), o.order_type))
295        .try_collect()?;
296
297    let mut encoded_chunk = vec![vec![]; chunk.capacity()];
298    for encoded_column in encoded_columns {
299        for (encoded_row, data) in encoded_chunk.iter_mut().zip_eq_fast(encoded_column) {
300            encoded_row.extend(data);
301        }
302    }
303
304    Ok(encoded_chunk.into_iter().map(Into::into).collect())
305}
306
307/// Encode a row into memcomparable format.
308pub fn encode_row(
309    row: impl Row,
310    order_types: &[OrderType],
311) -> memcomparable::Result<MemcmpEncoded> {
312    let mut serializer = memcomparable::Serializer::new(vec![]);
313    row.iter()
314        .zip_eq_debug(order_types)
315        .try_for_each(|(datum, order)| serialize_datum(datum, *order, &mut serializer))?;
316    Ok(serializer.into_inner().into())
317}
318
319/// Decode a row from memcomparable format.
320pub fn decode_row(
321    encoded_row: &[u8],
322    data_types: &[DataType],
323    order_types: &[OrderType],
324) -> memcomparable::Result<OwnedRow> {
325    let mut deserializer = memcomparable::Deserializer::new(encoded_row);
326    let row_data = data_types
327        .iter()
328        .zip_eq_debug(order_types)
329        .map(|(dt, ot)| deserialize_datum(dt, *ot, &mut deserializer))
330        .try_collect()?;
331    Ok(OwnedRow::new(row_data))
332}
333
334#[cfg(test)]
335mod tests {
336    use std::ops::Neg;
337
338    use rand::rng as thread_rng;
339
340    use super::*;
341    use crate::array::{ListValue, StructValue};
342    use crate::row::RowExt;
343    use crate::types::FloatExt;
344
345    #[test]
346    fn test_memcomparable() {
347        fn encode_num(num: Option<i32>, order_type: OrderType) -> MemcmpEncoded {
348            encode_value(num.map(ScalarImpl::from), order_type).unwrap()
349        }
350
351        {
352            // default ascending
353            let order_type = OrderType::ascending();
354            let memcmp_minus_1 = encode_num(Some(-1), order_type);
355            let memcmp_3874 = encode_num(Some(3874), order_type);
356            let memcmp_45745 = encode_num(Some(45745), order_type);
357            let memcmp_i32_min = encode_num(Some(i32::MIN), order_type);
358            let memcmp_i32_max = encode_num(Some(i32::MAX), order_type);
359            let memcmp_none = encode_num(None, order_type);
360
361            assert!(memcmp_3874 < memcmp_45745);
362            assert!(memcmp_3874 < memcmp_i32_max);
363            assert!(memcmp_45745 < memcmp_i32_max);
364
365            assert!(memcmp_i32_min < memcmp_i32_max);
366            assert!(memcmp_i32_min < memcmp_3874);
367            assert!(memcmp_i32_min < memcmp_45745);
368
369            assert!(memcmp_minus_1 < memcmp_3874);
370            assert!(memcmp_minus_1 < memcmp_45745);
371            assert!(memcmp_minus_1 < memcmp_i32_max);
372            assert!(memcmp_minus_1 > memcmp_i32_min);
373
374            assert!(memcmp_none > memcmp_minus_1);
375            assert!(memcmp_none > memcmp_3874);
376            assert!(memcmp_none > memcmp_i32_min);
377            assert!(memcmp_none > memcmp_i32_max);
378        }
379        {
380            // default descending
381            let order_type = OrderType::descending();
382            let memcmp_minus_1 = encode_num(Some(-1), order_type);
383            let memcmp_3874 = encode_num(Some(3874), order_type);
384            let memcmp_none = encode_num(None, order_type);
385
386            assert!(memcmp_none < memcmp_minus_1);
387            assert!(memcmp_none < memcmp_3874);
388            assert!(memcmp_3874 < memcmp_minus_1);
389        }
390        {
391            // ASC NULLS FIRST (NULLS SMALLEST)
392            let order_type = OrderType::ascending_nulls_first();
393            let memcmp_minus_1 = encode_num(Some(-1), order_type);
394            let memcmp_3874 = encode_num(Some(3874), order_type);
395            let memcmp_none = encode_num(None, order_type);
396            assert!(memcmp_none < memcmp_minus_1);
397            assert!(memcmp_none < memcmp_3874);
398        }
399        {
400            // ASC NULLS LAST (NULLS LARGEST)
401            let order_type = OrderType::ascending_nulls_last();
402            let memcmp_minus_1 = encode_num(Some(-1), order_type);
403            let memcmp_3874 = encode_num(Some(3874), order_type);
404            let memcmp_none = encode_num(None, order_type);
405            assert!(memcmp_none > memcmp_minus_1);
406            assert!(memcmp_none > memcmp_3874);
407        }
408        {
409            // DESC NULLS FIRST (NULLS LARGEST)
410            let order_type = OrderType::descending_nulls_first();
411            let memcmp_minus_1 = encode_num(Some(-1), order_type);
412            let memcmp_3874 = encode_num(Some(3874), order_type);
413            let memcmp_none = encode_num(None, order_type);
414            assert!(memcmp_none < memcmp_minus_1);
415            assert!(memcmp_none < memcmp_3874);
416        }
417        {
418            // DESC NULLS LAST (NULLS SMALLEST)
419            let order_type = OrderType::descending_nulls_last();
420            let memcmp_minus_1 = encode_num(Some(-1), order_type);
421            let memcmp_3874 = encode_num(Some(3874), order_type);
422            let memcmp_none = encode_num(None, order_type);
423            assert!(memcmp_none > memcmp_minus_1);
424            assert!(memcmp_none > memcmp_3874);
425        }
426    }
427
428    #[test]
429    fn test_memcomparable_structs() {
430        // NOTE: `NULL`s inside composite type values are always the largest.
431
432        let struct_none = Datum::None;
433        let struct_1 = Datum::Some(
434            StructValue::new(vec![Some(ScalarImpl::from(1)), Some(ScalarImpl::from(2))]).into(),
435        );
436        let struct_2 = Datum::Some(
437            StructValue::new(vec![Some(ScalarImpl::from(1)), Some(ScalarImpl::from(3))]).into(),
438        );
439        let struct_3 = Datum::Some(StructValue::new(vec![Some(ScalarImpl::from(1)), None]).into());
440
441        {
442            // ASC NULLS FIRST (NULLS SMALLEST)
443            let order_type = OrderType::ascending_nulls_first();
444            let memcmp_struct_none = encode_value(&struct_none, order_type).unwrap();
445            let memcmp_struct_1 = encode_value(&struct_1, order_type).unwrap();
446            let memcmp_struct_2 = encode_value(&struct_2, order_type).unwrap();
447            let memcmp_struct_3 = encode_value(&struct_3, order_type).unwrap();
448            assert!(memcmp_struct_none < memcmp_struct_1);
449            assert!(memcmp_struct_1 < memcmp_struct_2);
450            assert!(memcmp_struct_2 < memcmp_struct_3);
451        }
452        {
453            // ASC NULLS LAST (NULLS LARGEST)
454            let order_type = OrderType::ascending_nulls_last();
455            let memcmp_struct_none = encode_value(&struct_none, order_type).unwrap();
456            let memcmp_struct_1 = encode_value(&struct_1, order_type).unwrap();
457            let memcmp_struct_2 = encode_value(&struct_2, order_type).unwrap();
458            let memcmp_struct_3 = encode_value(&struct_3, order_type).unwrap();
459            assert!(memcmp_struct_1 < memcmp_struct_2);
460            assert!(memcmp_struct_2 < memcmp_struct_3);
461            assert!(memcmp_struct_3 < memcmp_struct_none);
462        }
463        {
464            // DESC NULLS FIRST (NULLS LARGEST)
465            let order_type = OrderType::descending_nulls_first();
466            let memcmp_struct_none = encode_value(&struct_none, order_type).unwrap();
467            let memcmp_struct_1 = encode_value(&struct_1, order_type).unwrap();
468            let memcmp_struct_2 = encode_value(&struct_2, order_type).unwrap();
469            let memcmp_struct_3 = encode_value(&struct_3, order_type).unwrap();
470            assert!(memcmp_struct_none < memcmp_struct_3);
471            assert!(memcmp_struct_3 < memcmp_struct_2);
472            assert!(memcmp_struct_2 < memcmp_struct_1);
473        }
474        {
475            // DESC NULLS LAST (NULLS SMALLEST)
476            let order_type = OrderType::descending_nulls_last();
477            let memcmp_struct_none = encode_value(&struct_none, order_type).unwrap();
478            let memcmp_struct_1 = encode_value(&struct_1, order_type).unwrap();
479            let memcmp_struct_2 = encode_value(&struct_2, order_type).unwrap();
480            let memcmp_struct_3 = encode_value(&struct_3, order_type).unwrap();
481            assert!(memcmp_struct_3 < memcmp_struct_2);
482            assert!(memcmp_struct_2 < memcmp_struct_1);
483            assert!(memcmp_struct_1 < memcmp_struct_none);
484        }
485    }
486
487    #[test]
488    fn test_memcomparable_lists() {
489        // NOTE: `NULL`s inside composite type values are always the largest.
490
491        let list_none = Datum::None;
492        let list_1 = Datum::Some(ListValue::from_iter([1, 2]).into());
493        let list_2 = Datum::Some(ListValue::from_iter([1, 3]).into());
494        let list_3 = Datum::Some(ListValue::from_iter([Some(1), None]).into());
495
496        {
497            // ASC NULLS FIRST (NULLS SMALLEST)
498            let order_type = OrderType::ascending_nulls_first();
499            let memcmp_list_none = encode_value(&list_none, order_type).unwrap();
500            let memcmp_list_1 = encode_value(&list_1, order_type).unwrap();
501            let memcmp_list_2 = encode_value(&list_2, order_type).unwrap();
502            let memcmp_list_3 = encode_value(&list_3, order_type).unwrap();
503            assert!(memcmp_list_none < memcmp_list_1);
504            assert!(memcmp_list_1 < memcmp_list_2);
505            assert!(memcmp_list_2 < memcmp_list_3);
506        }
507        {
508            // ASC NULLS LAST (NULLS LARGEST)
509            let order_type = OrderType::ascending_nulls_last();
510            let memcmp_list_none = encode_value(&list_none, order_type).unwrap();
511            let memcmp_list_1 = encode_value(&list_1, order_type).unwrap();
512            let memcmp_list_2 = encode_value(&list_2, order_type).unwrap();
513            let memcmp_list_3 = encode_value(&list_3, order_type).unwrap();
514            assert!(memcmp_list_1 < memcmp_list_2);
515            assert!(memcmp_list_2 < memcmp_list_3);
516            assert!(memcmp_list_3 < memcmp_list_none);
517        }
518        {
519            // DESC NULLS FIRST (NULLS LARGEST)
520            let order_type = OrderType::descending_nulls_first();
521            let memcmp_list_none = encode_value(&list_none, order_type).unwrap();
522            let memcmp_list_1 = encode_value(&list_1, order_type).unwrap();
523            let memcmp_list_2 = encode_value(&list_2, order_type).unwrap();
524            let memcmp_list_3 = encode_value(&list_3, order_type).unwrap();
525            assert!(memcmp_list_none < memcmp_list_3);
526            assert!(memcmp_list_3 < memcmp_list_2);
527            assert!(memcmp_list_2 < memcmp_list_1);
528        }
529        {
530            // DESC NULLS LAST (NULLS SMALLEST)
531            let order_type = OrderType::descending_nulls_last();
532            let memcmp_list_none = encode_value(&list_none, order_type).unwrap();
533            let memcmp_list_1 = encode_value(&list_1, order_type).unwrap();
534            let memcmp_list_2 = encode_value(&list_2, order_type).unwrap();
535            let memcmp_list_3 = encode_value(&list_3, order_type).unwrap();
536            assert!(memcmp_list_3 < memcmp_list_2);
537            assert!(memcmp_list_2 < memcmp_list_1);
538            assert!(memcmp_list_1 < memcmp_list_none);
539        }
540    }
541
542    #[test]
543    fn test_issue_legacy_2057_ordered_float_memcomparable() {
544        use num_traits::*;
545        use rand::seq::SliceRandom;
546
547        fn serialize(f: F32) -> MemcmpEncoded {
548            encode_value(Some(ScalarImpl::from(f)), OrderType::default()).unwrap()
549        }
550
551        fn deserialize(data: MemcmpEncoded) -> F32 {
552            decode_value(&DataType::Float32, &data, OrderType::default())
553                .unwrap()
554                .unwrap()
555                .into_float32()
556        }
557
558        let floats = vec![
559            // -inf
560            F32::neg_infinity(),
561            // -1
562            F32::one().neg(),
563            // 0, -0 should be treated the same
564            F32::zero(),
565            F32::neg_zero(),
566            F32::zero(),
567            // 1
568            F32::one(),
569            // inf
570            F32::infinity(),
571            // nan, -nan should be treated the same
572            F32::nan(),
573            F32::nan().neg(),
574            F32::nan(),
575        ];
576        assert!(floats.is_sorted());
577
578        let mut floats_clone = floats.clone();
579        floats_clone.shuffle(&mut thread_rng());
580        floats_clone.sort();
581        assert_eq!(floats, floats_clone);
582
583        let memcomparables = floats.clone().into_iter().map(serialize).collect_vec();
584        assert!(memcomparables.is_sorted());
585
586        let decoded_floats = memcomparables.into_iter().map(deserialize).collect_vec();
587        assert!(decoded_floats.is_sorted());
588        assert_eq!(floats, decoded_floats);
589    }
590
591    #[test]
592    fn test_encode_row() {
593        let v10 = Some(ScalarImpl::Int32(42));
594        let v10_cloned = v10.clone();
595        let v11 = Some(ScalarImpl::Utf8("hello".into()));
596        let v11_cloned = v11.clone();
597        let v12 = Some(ScalarImpl::Float32(4.0.into()));
598        let v20 = Some(ScalarImpl::Int32(42));
599        let v21 = Some(ScalarImpl::Utf8("hell".into()));
600        let v22 = Some(ScalarImpl::Float32(3.0.into()));
601
602        let row1 = OwnedRow::new(vec![v10, v11, v12]);
603        let row2 = OwnedRow::new(vec![v20, v21, v22]);
604        let order_col_indices = vec![0, 1];
605        let order_types = vec![OrderType::ascending(), OrderType::descending()];
606
607        let encoded_row1 = encode_row(row1.project(&order_col_indices), &order_types).unwrap();
608        let encoded_v10 = encode_value(
609            v10_cloned.as_ref().map(|x| x.as_scalar_ref_impl()),
610            OrderType::ascending(),
611        )
612        .unwrap();
613        let encoded_v11 = encode_value(
614            v11_cloned.as_ref().map(|x| x.as_scalar_ref_impl()),
615            OrderType::descending(),
616        )
617        .unwrap();
618        let concated_encoded_row1 = encoded_v10.into_iter().chain(encoded_v11).collect();
619        assert_eq!(encoded_row1, concated_encoded_row1);
620
621        let encoded_row2 = encode_row(row2.project(&order_col_indices), &order_types).unwrap();
622        assert!(encoded_row1 < encoded_row2);
623    }
624
625    // See also `row_value_encode_decode()` in `src/common/src/row/owned_row.rs`
626    #[test]
627    fn test_decode_row() {
628        let encoded: Vec<u8> = vec![
629            0, 128, 0, 0, 42, 255, 127, 255, 255, 255, 255, 255, 255, 213, 1, 0, 193, 186, 163,
630            215, 255, 254, 153, 144, 144, 144, 144, 144, 255, 255, 249, 0, 1, 98, 97, 97, 97, 97,
631            114, 0, 0, 6,
632        ];
633
634        let order_types = vec![
635            OrderType::ascending(),
636            OrderType::descending(),
637            OrderType::ascending(),
638            OrderType::ascending(),
639            OrderType::descending(),
640            OrderType::ascending(),
641        ];
642        let data_types = vec![
643            DataType::Int32,
644            DataType::Int64,
645            DataType::Timestamp,
646            DataType::Float32,
647            DataType::Varchar,
648            DataType::Bytea,
649        ];
650
651        let result = decode_row(&encoded, &data_types, &order_types).unwrap();
652        // println!("{:?}", &result);
653
654        let expected = OwnedRow::new(vec![
655            Some(ScalarImpl::Int32(42)),
656            Some(ScalarImpl::Int64(42)),
657            None,
658            Some(ScalarImpl::Float32(23.33.into())),
659            Some(ScalarImpl::Utf8("fooooo".into())),
660            Some(ScalarImpl::Bytea("baaaar".as_bytes().into())),
661        ]);
662        assert_eq!(&result, &expected);
663    }
664
665    #[test]
666    fn test_encode_chunk() {
667        let v10 = Some(ScalarImpl::Int32(42));
668        let v11 = Some(ScalarImpl::Utf8("hello".into()));
669        let v12 = Some(ScalarImpl::Float32(4.0.into()));
670        let v20 = Some(ScalarImpl::Int32(42));
671        let v21 = Some(ScalarImpl::Utf8("hell".into()));
672        let v22 = Some(ScalarImpl::Float32(3.0.into()));
673
674        let row1 = OwnedRow::new(vec![v10, v11, v12]);
675        let row2 = OwnedRow::new(vec![v20, v21, v22]);
676        let chunk = DataChunk::from_rows(
677            &[row1.clone(), row2.clone()],
678            &[DataType::Int32, DataType::Varchar, DataType::Float32],
679        );
680        let order_col_indices = vec![0, 1];
681        let order_types = vec![OrderType::ascending(), OrderType::descending()];
682        let column_orders = order_col_indices
683            .iter()
684            .zip_eq_fast(&order_types)
685            .map(|(i, o)| ColumnOrder::new(*i, *o))
686            .collect_vec();
687
688        let encoded_row1 = encode_row(row1.project(&order_col_indices), &order_types).unwrap();
689        let encoded_row2 = encode_row(row2.project(&order_col_indices), &order_types).unwrap();
690        let encoded_chunk = encode_chunk(&chunk, &column_orders).unwrap();
691        assert_eq!(&encoded_chunk, &[encoded_row1, encoded_row2]);
692    }
693}
risingwave_common/util/memcmp_encoding.rs

risingwave_common/util/
memcmp_encoding.rs
