risingwave_common/row/
mod.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 std::borrow::Cow;
16use std::fmt::Display;
17use std::hash::{BuildHasher, Hasher};
18use std::ops::RangeBounds;
19
20use bytes::{BufMut, Bytes, BytesMut};
21use itertools::Itertools;
22
23use self::empty::EMPTY;
24use crate::hash::HashCode;
25use crate::types::{DatumRef, ToDatumRef, ToOwnedDatum, ToText, hash_datum};
26use crate::util::row_serde::OrderedRowSerde;
27use crate::util::value_encoding;
28
29/// The trait for abstracting over a Row-like type.
30pub trait Row: Sized + std::fmt::Debug + PartialEq + Eq + Send + Sync {
31    /// Returns the [`DatumRef`] at the given `index`.
32    fn datum_at(&self, index: usize) -> DatumRef<'_>;
33
34    /// Returns the [`DatumRef`] at the given `index` without bounds checking.
35    ///
36    /// # Safety
37    /// Calling this method with an out-of-bounds index is undefined behavior.
38    unsafe fn datum_at_unchecked(&self, index: usize) -> DatumRef<'_>;
39
40    /// Returns the number of datum in the row.
41    fn len(&self) -> usize;
42
43    /// Returns `true` if the row contains no datum.
44    #[inline]
45    fn is_empty(&self) -> bool {
46        self.len() == 0
47    }
48
49    /// Returns an iterator over the datums in the row, in [`DatumRef`] form.
50    fn iter(&self) -> impl Iterator<Item = DatumRef<'_>>;
51
52    /// Converts the row into an [`OwnedRow`].
53    ///
54    /// Prefer `into_owned_row` if the row is already owned.
55    #[inline]
56    fn to_owned_row(&self) -> OwnedRow {
57        OwnedRow::new(self.iter().map(|d| d.to_owned_datum()).collect())
58    }
59
60    /// Consumes `self` and converts it into an [`OwnedRow`].
61    #[inline]
62    fn into_owned_row(self) -> OwnedRow {
63        self.to_owned_row()
64    }
65
66    /// Serializes the row with value encoding, into the given `buf`.
67    #[inline]
68    fn value_serialize_into(&self, mut buf: impl BufMut) {
69        for datum in self.iter() {
70            value_encoding::serialize_datum_into(datum, &mut buf);
71        }
72    }
73
74    /// Serializes the row with value encoding and returns the bytes.
75    #[inline]
76    fn value_serialize(&self) -> Vec<u8> {
77        let estimate_size = self
78            .iter()
79            .map(value_encoding::estimate_serialize_datum_size)
80            .sum();
81        let mut buf = Vec::with_capacity(estimate_size);
82        self.value_serialize_into(&mut buf);
83        buf
84    }
85
86    /// Serializes the row with value encoding and returns the bytes.
87    #[inline]
88    fn value_serialize_bytes(&self) -> Bytes {
89        let estimate_size = self
90            .iter()
91            .map(value_encoding::estimate_serialize_datum_size)
92            .sum();
93        let mut buf = BytesMut::with_capacity(estimate_size);
94        self.value_serialize_into(&mut buf);
95        buf.freeze()
96    }
97
98    fn value_estimate_size(&self) -> usize {
99        self.iter()
100            .map(value_encoding::estimate_serialize_datum_size)
101            .sum()
102    }
103
104    /// Serializes the row with memcomparable encoding, into the given `buf`. As each datum may have
105    /// different order type, a `serde` should be provided.
106    #[inline]
107    fn memcmp_serialize_into(&self, serde: &OrderedRowSerde, buf: impl BufMut) {
108        serde.serialize(self, buf);
109    }
110
111    /// Serializes the row with memcomparable encoding and return the bytes. As each datum may have
112    /// different order type, a `serde` should be provided.
113    #[inline]
114    fn memcmp_serialize(&self, serde: &OrderedRowSerde) -> Vec<u8> {
115        let mut buf = Vec::with_capacity(self.len()); // each datum is at least 1 byte
116        self.memcmp_serialize_into(serde, &mut buf);
117        buf
118    }
119
120    /// Hash the datums of this row into the given hasher.
121    ///
122    /// Implementors should delegate [`std::hash::Hash::hash`] to this method.
123    fn hash_datums_into<H: Hasher>(&self, state: &mut H) {
124        for datum in self.iter() {
125            hash_datum(datum, state);
126        }
127    }
128
129    /// Returns the hash code of the row.
130    fn hash<H: BuildHasher>(&self, hash_builder: H) -> HashCode<H> {
131        let mut hasher = hash_builder.build_hasher();
132        self.hash_datums_into(&mut hasher);
133        hasher.finish().into()
134    }
135
136    /// Determines whether the datums of this row are equal to those of another.
137    #[inline]
138    fn eq(this: &Self, other: impl Row) -> bool {
139        if this.len() != other.len() {
140            return false;
141        }
142        for i in (0..this.len()).rev() {
143            // compare from the end to the start, as it's more likely to have same prefix
144            // SAFETY: index is in bounds as we are iterating from 0 to len.
145            if unsafe { this.datum_at_unchecked(i) != other.datum_at_unchecked(i) } {
146                return false;
147            }
148        }
149        true
150    }
151}
152
153const fn assert_row<R: Row>(r: R) -> R {
154    r
155}
156
157/// An extension trait for [`Row`]s that provides a variety of convenient adapters.
158pub trait RowExt: Row {
159    /// Adapter for chaining two rows together.
160    fn chain<R: Row>(self, other: R) -> Chain<Self, R>
161    where
162        Self: Sized,
163    {
164        assert_row(Chain::new(self, other))
165    }
166
167    /// Adapter for projecting a row onto a subset of its columns with the given `indices`.
168    ///
169    /// # Panics
170    /// Panics if `indices` contains an out-of-bounds index.
171    fn project(self, indices: &[usize]) -> Project<'_, Self>
172    where
173        Self: Sized,
174    {
175        assert_row(Project::new(self, indices))
176    }
177
178    /// Adapter for slicing a row with the given `range`.
179    ///
180    /// # Panics
181    /// Panics if range is out of bounds.
182    fn slice(self, range: impl RangeBounds<usize>) -> Slice<Self>
183    where
184        Self: Sized,
185    {
186        assert_row(Slice::new(self, range))
187    }
188
189    /// Returns a displayable wrapper for the row.
190    fn display(&self) -> impl Display + '_ {
191        struct D<'a, T: Row>(&'a T);
192        impl<T: Row> Display for D<'_, T> {
193            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
194                write!(
195                    f,
196                    "[{}]",
197                    self.0.iter().format_with(", ", |datum, f| {
198                        match datum {
199                            None => f(&"NULL"),
200                            Some(scalar) => f(&scalar.text_display()),
201                        }
202                    })
203                )
204            }
205        }
206        D(self)
207    }
208
209    fn is_null_at(&self, index: usize) -> bool {
210        self.datum_at(index).is_none()
211    }
212}
213
214impl<R: Row> RowExt for R {}
215
216/// Forward the implementation of [`Row`] to the deref target.
217macro_rules! deref_forward_row {
218    () => {
219        fn datum_at(&self, index: usize) -> crate::types::DatumRef<'_> {
220            (**self).datum_at(index)
221        }
222
223        unsafe fn datum_at_unchecked(&self, index: usize) -> crate::types::DatumRef<'_> {
224            unsafe { (**self).datum_at_unchecked(index) }
225        }
226
227        fn len(&self) -> usize {
228            (**self).len()
229        }
230
231        fn is_empty(&self) -> bool {
232            (**self).is_empty()
233        }
234
235        fn iter(&self) -> impl Iterator<Item = crate::types::DatumRef<'_>> {
236            (**self).iter()
237        }
238
239        fn to_owned_row(&self) -> OwnedRow {
240            (**self).to_owned_row()
241        }
242
243        fn value_serialize_into(&self, buf: impl bytes::BufMut) {
244            (**self).value_serialize_into(buf)
245        }
246
247        fn value_serialize(&self) -> Vec<u8> {
248            (**self).value_serialize()
249        }
250
251        fn memcmp_serialize_into(
252            &self,
253            serde: &$crate::util::row_serde::OrderedRowSerde,
254            buf: impl bytes::BufMut,
255        ) {
256            (**self).memcmp_serialize_into(serde, buf)
257        }
258
259        fn memcmp_serialize(&self, serde: &$crate::util::row_serde::OrderedRowSerde) -> Vec<u8> {
260            (**self).memcmp_serialize(serde)
261        }
262
263        fn hash<H: std::hash::BuildHasher>(&self, hash_builder: H) -> $crate::hash::HashCode<H> {
264            (**self).hash(hash_builder)
265        }
266
267        fn hash_datums_into<H: std::hash::Hasher>(&self, state: &mut H) {
268            (**self).hash_datums_into(state)
269        }
270
271        fn eq(this: &Self, other: impl Row) -> bool {
272            Row::eq(&(**this), other)
273        }
274    };
275}
276
277impl<R: Row> Row for &R {
278    deref_forward_row!();
279}
280
281impl<R: Row + Clone> Row for Cow<'_, R> {
282    deref_forward_row!();
283
284    // Manually implemented in case `R` has a more efficient implementation.
285    fn into_owned_row(self) -> OwnedRow {
286        self.into_owned().into_owned_row()
287    }
288}
289
290impl<R: Row> Row for Box<R> {
291    deref_forward_row!();
292
293    // Manually implemented in case the `Cow` is `Owned` and `R` has a more efficient
294    // implementation.
295    fn into_owned_row(self) -> OwnedRow {
296        (*self).into_owned_row()
297    }
298}
299
300/// Implements [`Row`] for a slice of datums.
301macro_rules! impl_slice_row {
302    () => {
303        #[inline]
304        fn datum_at(&self, index: usize) -> DatumRef<'_> {
305            self[index].to_datum_ref()
306        }
307
308        #[inline]
309        unsafe fn datum_at_unchecked(&self, index: usize) -> DatumRef<'_> {
310            unsafe { self.get_unchecked(index).to_datum_ref() }
311        }
312
313        #[inline]
314        fn len(&self) -> usize {
315            AsRef::<[D]>::as_ref(self).len()
316        }
317
318        #[inline]
319        fn iter(&self) -> impl Iterator<Item = DatumRef<'_>> {
320            AsRef::<[D]>::as_ref(self)
321                .iter()
322                .map(ToDatumRef::to_datum_ref)
323        }
324    };
325}
326
327impl<D: ToDatumRef> Row for &[D] {
328    impl_slice_row!();
329}
330
331impl<D: ToDatumRef, const N: usize> Row for [D; N] {
332    impl_slice_row!();
333}
334
335impl<D: ToDatumRef + Default, const N: usize> Row for ArrayVec<[D; N]> {
336    impl_slice_row!();
337}
338
339/// Implements [`Row`] for an optional row.
340impl<R: Row> Row for Option<R> {
341    fn datum_at(&self, index: usize) -> DatumRef<'_> {
342        match self {
343            Some(row) => row.datum_at(index),
344            None => EMPTY.datum_at(index),
345        }
346    }
347
348    unsafe fn datum_at_unchecked(&self, index: usize) -> DatumRef<'_> {
349        unsafe {
350            match self {
351                Some(row) => row.datum_at_unchecked(index),
352                None => EMPTY.datum_at_unchecked(index),
353            }
354        }
355    }
356
357    fn len(&self) -> usize {
358        match self {
359            Some(row) => row.len(),
360            None => 0,
361        }
362    }
363
364    fn iter(&self) -> impl Iterator<Item = DatumRef<'_>> {
365        match self {
366            Some(row) => either::Either::Left(row.iter()),
367            None => either::Either::Right(EMPTY.iter()),
368        }
369    }
370
371    fn to_owned_row(&self) -> OwnedRow {
372        match self {
373            Some(row) => row.to_owned_row(),
374            None => OwnedRow::new(Vec::new()),
375        }
376    }
377
378    fn into_owned_row(self) -> OwnedRow {
379        match self {
380            Some(row) => row.into_owned_row(),
381            None => OwnedRow::new(Vec::new()),
382        }
383    }
384
385    fn value_serialize_into(&self, buf: impl BufMut) {
386        if let Some(row) = self {
387            row.value_serialize_into(buf);
388        }
389    }
390
391    fn memcmp_serialize_into(&self, serde: &OrderedRowSerde, buf: impl BufMut) {
392        if let Some(row) = self {
393            row.memcmp_serialize_into(serde, buf);
394        }
395    }
396}
397
398/// Implements [`Row`] for an [`either::Either`] of two different types of rows.
399impl<R1: Row, R2: Row> Row for either::Either<R1, R2> {
400    fn datum_at(&self, index: usize) -> DatumRef<'_> {
401        either::for_both!(self, row => row.datum_at(index))
402    }
403
404    unsafe fn datum_at_unchecked(&self, index: usize) -> DatumRef<'_> {
405        unsafe { either::for_both!(self, row => row.datum_at_unchecked(index)) }
406    }
407
408    fn len(&self) -> usize {
409        either::for_both!(self, row => row.len())
410    }
411
412    fn is_empty(&self) -> bool {
413        either::for_both!(self, row => row.is_empty())
414    }
415
416    fn iter(&self) -> impl Iterator<Item = DatumRef<'_>> {
417        self.as_ref().map_either(Row::iter, Row::iter)
418    }
419
420    fn to_owned_row(&self) -> OwnedRow {
421        either::for_both!(self, row => row.to_owned_row())
422    }
423
424    fn into_owned_row(self) -> OwnedRow {
425        either::for_both!(self, row => row.into_owned_row())
426    }
427
428    fn value_serialize_into(&self, buf: impl BufMut) {
429        either::for_both!(self, row => row.value_serialize_into(buf))
430    }
431
432    fn value_serialize(&self) -> Vec<u8> {
433        either::for_both!(self, row => row.value_serialize())
434    }
435
436    fn value_serialize_bytes(&self) -> Bytes {
437        either::for_both!(self, row => row.value_serialize_bytes())
438    }
439
440    fn memcmp_serialize_into(&self, serde: &OrderedRowSerde, buf: impl BufMut) {
441        either::for_both!(self, row => row.memcmp_serialize_into(serde, buf))
442    }
443
444    fn memcmp_serialize(&self, serde: &OrderedRowSerde) -> Vec<u8> {
445        either::for_both!(self, row => row.memcmp_serialize(serde))
446    }
447
448    fn hash_datums_into<H: Hasher>(&self, state: &mut H) {
449        either::for_both!(self, row => row.hash_datums_into(state))
450    }
451
452    fn hash<H: BuildHasher>(&self, hash_builder: H) -> HashCode<H> {
453        either::for_both!(self, row => row.hash(hash_builder))
454    }
455
456    fn eq(this: &Self, other: impl Row) -> bool {
457        either::for_both!(this, row => Row::eq(row, other))
458    }
459}
460
461mod chain;
462mod compacted_row;
463mod empty;
464mod once;
465mod ordered;
466mod owned_row;
467mod project;
468mod repeat_n;
469mod slice;
470pub use ::tinyvec::ArrayVec;
471pub use chain::Chain;
472pub use compacted_row::CompactedRow;
473pub use empty::{Empty, empty};
474pub use once::{Once, once};
475pub use owned_row::{OwnedRow, RowDeserializer};
476pub use project::Project;
477pub use repeat_n::{RepeatN, repeat_n};
478pub use slice::Slice;
risingwave_common/row/mod.rs

risingwave_common/row/
mod.rs
