arrow 24.0.0

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

//! A comparable row-oriented representation of a collection of [`Array`]

use crate::array::{
    as_boolean_array, as_generic_binary_array, as_largestring_array, as_string_array,
    Array, ArrayRef, Decimal128Array, Decimal256Array,
};
use crate::compute::SortOptions;
use crate::datatypes::*;
use crate::error::{ArrowError, Result};
use crate::row::interner::{Interned, OrderPreservingInterner};
use crate::util::decimal::{Decimal128, Decimal256};
use crate::{downcast_dictionary_array, downcast_primitive_array};

mod fixed;
mod interner;
mod variable;

/// Converts [`ArrayRef`] columns into a row-oriented format that are [normalized for sorting].
///
/// In particular, a byte-wise comparison of the rows, e.g. [`memcmp`], is sufficient
/// to establish the ordering of two rows, allowing for extremely fast comparisons,
/// and permitting the use of [non-comparison sorts] such as [radix sort]
///
/// Comparing [`Rows`] generated by different [`RowConverter`] is not guaranteed to
/// yield a meaningful ordering
///
/// # Format
///
/// The encoding of the row format should not be considered stable, but is documented here
/// for reference.
///
/// ## Unsigned Integer Encoding
///
/// A null integer is encoded as a `0_u8`, followed by a zero-ed number of bytes corresponding
/// to the integer's length
///
/// A valid integer is encoded as `1_u8`, followed by the big-endian representation of the
/// integer
///
/// ## Signed Integer Encoding
///
/// Signed integers have their most significant sign bit flipped, and are then encoded in the
/// same manner as an unsigned integer
///
/// ## Float Encoding
///
/// Floats are converted from IEEE 754 representation to a signed integer representation
/// by flipping all bar the sign bit if they are negative.
///
/// They are then encoded in the same manner as a signed integer
///
/// ## Variable Length Bytes Encoding
///
/// A null is encoded as a `0_u8`
///
/// An empty byte array is encoded as `1_u8`
///
/// A non-null, non-empty byte array is encoded as `2_u8` followed by the byte array
/// encoded using a block based scheme described below.
///
/// The byte array is broken up into 32-byte blocks, each block is written in turn
/// to the output, followed by `0xFF_u8`. The final block is padded to 32-bytes
/// with `0_u8` and written to the output, followed by the un-padded length in bytes
/// of this final block as a `u8`
///
/// This is loosely inspired by [COBS] encoding, and chosen over more traditional
/// [byte stuffing] as it is more amenable to vectorisation, in particular AVX-256.
///
/// ## Dictionary Encoding
///
/// [`RowConverter`] needs to support converting dictionary encoded arrays with unsorted, and
/// potentially distinct dictionaries. One simple mechanism to avoid this would be to reverse
/// the dictionary encoding, and encode the array values directly, however, this would lose
/// the benefits of dictionary encoding to reduce memory and CPU consumption.
///
/// As such the [`RowConverter`] maintains an order-preserving dictionary encoding for each
/// dictionary encoded column. As this is a variable-length encoding, new dictionary values
/// can be added whilst preserving the sort order.
///
/// A null dictionary value is encoded as `0_u8`.
///
/// A non-null dictionary value is encoded as `1_u8` followed by a null-terminated byte array
/// key determined by the order-preserving dictionary encoding
///
/// # Ordering
///
/// ## Float Ordering
///
/// Floats are totally ordered in accordance to the `totalOrder` predicate as defined
/// in the IEEE 754 (2008 revision) floating point standard.
///
/// The ordering established by this does not always agree with the
/// [`PartialOrd`] and [`PartialEq`] implementations of `f32`. For example,
/// they consider negative and positive zero equal, while this does not
///
/// ## Null Ordering
///
/// The encoding described above will order nulls first, this can be inverted by representing
/// nulls as `0xFF_u8` instead of `0_u8`
///
/// ## Reverse Column Ordering
///
/// The order of a given column can be reversed by negating the encoded bytes of non-null values
///
/// ## Reconstruction
///
/// Given a schema it would theoretically be possible to reconstruct the columnar data from
/// the row format, however, this is currently not implemented. It is recommended that the row
/// format is instead used to obtain a sorted list of row indices, which can then be used
/// with [`take`](crate::compute::take) to obtain a sorted [`Array`]
///
/// [non-comparison sorts]:[https://en.wikipedia.org/wiki/Sorting_algorithm#Non-comparison_sorts]
/// [radix sort]:[https://en.wikipedia.org/wiki/Radix_sort]
/// [normalized for sorting]:[https://citeseerx.ist.psu.edu/viewdoc/download?doi=10.1.1.83.1080&rep=rep1&type=pdf]
/// [`memcmp`]:[https://www.man7.org/linux/man-pages/man3/memcmp.3.html]
/// [COBS]:[https://en.wikipedia.org/wiki/Consistent_Overhead_Byte_Stuffing]
/// [byte stuffing]:[https://en.wikipedia.org/wiki/High-Level_Data_Link_Control#Asynchronous_framing]
#[derive(Debug)]
pub struct RowConverter {
    fields: Vec<SortField>,
    /// interning state for column `i`, if column`i` is a dictionary
    interners: Vec<Option<Box<OrderPreservingInterner>>>,
}

/// Configure the data type and sort order for a given column
#[derive(Debug, Clone)]
pub struct SortField {
    /// Sort options
    options: SortOptions,
    /// Data type
    data_type: DataType,
}

impl SortField {
    /// Create a new column with the given data type
    pub fn new(data_type: DataType) -> Self {
        Self::new_with_options(data_type, Default::default())
    }

    /// Create a new column with the given data type and [`SortOptions`]
    pub fn new_with_options(data_type: DataType, options: SortOptions) -> Self {
        Self { options, data_type }
    }
}

impl RowConverter {
    /// Create a new [`RowConverter`] with the provided schema
    pub fn new(fields: Vec<SortField>) -> Self {
        let interners = (0..fields.len()).map(|_| None).collect();
        Self { fields, interners }
    }

    /// Convert [`ArrayRef`] columns into [`Rows`]
    ///
    /// See [`Row`] for information on when [`Row`] can be compared
    ///
    /// # Panics
    ///
    /// Panics if the schema of `columns` does not match that provided to [`RowConverter::new`]
    pub fn convert_columns(&mut self, columns: &[ArrayRef]) -> Result<Rows> {
        if columns.len() != self.fields.len() {
            return Err(ArrowError::InvalidArgumentError(format!(
                "Incorrect number of arrays provided to RowConverter, expected {} got {}",
                self.fields.len(),
                columns.len()
            )));
        }

        let dictionaries = columns
            .iter()
            .zip(&mut self.interners)
            .zip(&self.fields)
            .map(|((column, interner), field)| {
                if !column.data_type().equals_datatype(&field.data_type) {
                    return Err(ArrowError::InvalidArgumentError(format!(
                        "RowConverter column schema mismatch, expected {} got {}",
                        field.data_type,
                        column.data_type()
                    )));
                }

                let values = downcast_dictionary_array! {
                    column => column.values(),
                    _ => return Ok(None)
                };

                let interner = interner.get_or_insert_with(Default::default);

                let mapping: Vec<_> = compute_dictionary_mapping(interner, values)?
                    .into_iter()
                    .map(|maybe_interned| {
                        maybe_interned.map(|interned| interner.normalized_key(interned))
                    })
                    .collect();

                Ok(Some(mapping))
            })
            .collect::<Result<Vec<_>>>()?;

        let mut rows = new_empty_rows(columns, &dictionaries)?;

        for ((column, field), dictionary) in
            columns.iter().zip(&self.fields).zip(dictionaries)
        {
            // We encode a column at a time to minimise dispatch overheads
            encode_column(&mut rows, column, field.options, dictionary.as_deref())
        }

        if cfg!(debug_assertions) {
            assert_eq!(*rows.offsets.last().unwrap(), rows.buffer.len());
            rows.offsets
                .windows(2)
                .for_each(|w| assert!(w[0] < w[1], "offsets should be monotonic"));
        }

        Ok(rows)
    }
}

/// A row-oriented representation of arrow data, that is normalized for comparison
///
/// See [`RowConverter`]
#[derive(Debug)]
pub struct Rows {
    /// Underlying row bytes
    buffer: Box<[u8]>,
    /// Row `i` has data `&buffer[offsets[i]..offsets[i+1]]`
    offsets: Box<[usize]>,
}

impl Rows {
    pub fn row(&self, row: usize) -> Row<'_> {
        let end = self.offsets[row + 1];
        let start = self.offsets[row];
        Row(&self.buffer[start..end])
    }

    pub fn num_rows(&self) -> usize {
        self.offsets.len() - 1
    }
}

/// A comparable representation of a row
///
/// Two [`Row`] can be compared if they both belong to [`Rows`] returned by calls to
/// [`RowConverter::convert_columns`] on the same [`RowConverter`]
///
/// Otherwise any ordering established by comparing the [`Row`] is arbitrary
#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct Row<'a>(&'a [u8]);

impl<'a> AsRef<[u8]> for Row<'a> {
    fn as_ref(&self) -> &[u8] {
        self.0
    }
}

/// Computes the dictionary mapping for the given dictionary values
fn compute_dictionary_mapping(
    interner: &mut OrderPreservingInterner,
    values: &ArrayRef,
) -> Result<Vec<Option<Interned>>> {
    use fixed::FixedLengthEncoding;
    Ok(downcast_primitive_array! {
        values => interner
            .intern(values.iter().map(|x| x.map(|x| x.encode()))),
        DataType::Binary => {
            let iter = as_generic_binary_array::<i64>(values).iter();
            interner.intern(iter)
        }
        DataType::LargeBinary => {
            let iter = as_generic_binary_array::<i64>(values).iter();
            interner.intern(iter)
        }
        DataType::Utf8 => {
            let iter = as_string_array(values).iter().map(|x| x.map(|x| x.as_bytes()));
            interner.intern(iter)
        }
        DataType::LargeUtf8 => {
            let iter = as_largestring_array(values).iter().map(|x| x.map(|x| x.as_bytes()));
            interner.intern(iter)
        }
        t => return Err(ArrowError::NotYetImplemented(format!("dictionary value {} is not supported", t))),
    })
}

/// Computes the length of each encoded [`Rows`] and returns an empty [`Rows`]
fn new_empty_rows(
    cols: &[ArrayRef],
    dictionaries: &[Option<Vec<Option<&[u8]>>>],
) -> Result<Rows> {
    use fixed::FixedLengthEncoding;

    let num_rows = cols.first().map(|x| x.len()).unwrap_or(0);
    let mut lengths = vec![0; num_rows];

    for (array, dict) in cols.iter().zip(dictionaries) {
        downcast_primitive_array! {
            array => lengths.iter_mut().for_each(|x| *x += fixed::encoded_len(array)),
            DataType::Null => lengths.iter_mut().for_each(|x| *x += 1),
            DataType::Boolean => lengths.iter_mut().for_each(|x| *x += bool::ENCODED_LEN),
            DataType::Decimal128(_, _) => lengths.iter_mut().for_each(|x| *x += Decimal128::ENCODED_LEN),
            DataType::Decimal256(_, _) => lengths.iter_mut().for_each(|x| *x += Decimal256::ENCODED_LEN),
            DataType::Binary => as_generic_binary_array::<i32>(array)
                .iter()
                .zip(lengths.iter_mut())
                .for_each(|(slice, length)| *length += variable::encoded_len(slice)),
            DataType::LargeBinary => as_generic_binary_array::<i64>(array)
                .iter()
                .zip(lengths.iter_mut())
                .for_each(|(slice, length)| *length += variable::encoded_len(slice)),
            DataType::Utf8 => as_string_array(array)
                .iter()
                .zip(lengths.iter_mut())
                .for_each(|(slice, length)| {
                    *length += variable::encoded_len(slice.map(|x| x.as_bytes()))
                }),
            DataType::LargeUtf8 => as_largestring_array(array)
                .iter()
                .zip(lengths.iter_mut())
                .for_each(|(slice, length)| {
                    *length += variable::encoded_len(slice.map(|x| x.as_bytes()))
                }),
            DataType::Dictionary(_, _) => downcast_dictionary_array! {
                array => {
                    let dict = dict.as_ref().unwrap();
                    for (v, length) in array.keys().iter().zip(lengths.iter_mut()) {
                        match v.and_then(|v| dict[v as usize]) {
                            Some(k) => *length += k.len() + 1,
                            None => *length += 1,
                        }
                    }
                }
                _ => unreachable!(),
            }
            t => return Err(ArrowError::NotYetImplemented(format!("not yet implemented: {}", t)))
        }
    }

    let mut offsets = Vec::with_capacity(num_rows + 1);
    offsets.push(0);

    // We initialize the offsets shifted down by one row index.
    //
    // As the rows are appended to the offsets will be incremented to match
    //
    // For example, consider the case of 3 rows of length 3, 4, and 6 respectively.
    // The offsets would be initialized to `0, 0, 3, 7`
    //
    // Writing the first row entirely would yield `0, 3, 3, 7`
    // The second, `0, 3, 7, 7`
    // The third, `0, 3, 7, 13`
    //
    // This would be the final offsets for reading
    //
    // In this way offsets tracks the position during writing whilst eventually serving
    // as identifying the offsets of the written rows
    let mut cur_offset = 0_usize;
    for l in lengths {
        offsets.push(cur_offset);
        cur_offset = cur_offset.checked_add(l).expect("overflow");
    }

    let buffer = vec![0_u8; cur_offset];

    Ok(Rows {
        buffer: buffer.into(),
        offsets: offsets.into(),
    })
}

/// Encodes a column to the provided [`Rows`] incrementing the offsets as it progresses
fn encode_column(
    out: &mut Rows,
    column: &ArrayRef,
    opts: SortOptions,
    dictionary: Option<&[Option<&[u8]>]>,
) {
    downcast_primitive_array! {
        column => fixed::encode(out, column, opts),
        DataType::Null => {
            fixed::encode(out, std::iter::repeat(None::<bool>).take(column.len()), opts)
        }
        DataType::Boolean => fixed::encode(out, as_boolean_array(column), opts),
        DataType::Decimal128(_, _) => fixed::encode(
            out,
            column.as_any().downcast_ref::<Decimal128Array>().unwrap(),
            opts,
        ),
        DataType::Decimal256(_, _) => fixed::encode(
            out,
            column.as_any().downcast_ref::<Decimal256Array>().unwrap(),
            opts,
        ),
        DataType::Binary => {
            variable::encode(out, as_generic_binary_array::<i32>(column).iter(), opts)
        }
        DataType::LargeBinary => {
            variable::encode(out, as_generic_binary_array::<i64>(column).iter(), opts)
        }
        DataType::Utf8 => variable::encode(
            out,
            as_string_array(column).iter().map(|x| x.map(|x| x.as_bytes())),
            opts,
        ),
        DataType::LargeUtf8 => variable::encode(
            out,
            as_largestring_array(column)
                .iter()
                .map(|x| x.map(|x| x.as_bytes())),
            opts,
        ),
        DataType::Dictionary(_, _) => downcast_dictionary_array! {
            column => {
                let dict = dictionary.unwrap();
                for (offset, k) in out.offsets.iter_mut().skip(1).zip(column.keys()) {
                    match k.and_then(|k| dict[k as usize]) {
                        Some(v) => {
                            let end_offset = *offset + 1 + v.len();
                            out.buffer[*offset] = 1;
                            out.buffer[*offset+1..end_offset].copy_from_slice(v);
                            if opts.descending {
                                out.buffer[*offset..end_offset].iter_mut().for_each(|v| *v = !*v)
                            }
                            *offset = end_offset;
                        }
                        None => {
                            if !opts.nulls_first {
                                out.buffer[*offset] = 0xFF;
                            }
                            *offset += 1;
                        }
                    }
                }
            },
            _ => unreachable!()
        }
        t => unimplemented!("not yet implemented: {}", t)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::array::{
        BinaryArray, BooleanArray, DictionaryArray, Float32Array, GenericStringArray,
        Int16Array, Int32Array, OffsetSizeTrait, PrimitiveArray,
        PrimitiveDictionaryBuilder, StringArray,
    };
    use crate::compute::{LexicographicalComparator, SortColumn};
    use crate::util::display::array_value_to_string;
    use rand::distributions::uniform::SampleUniform;
    use rand::distributions::{Distribution, Standard};
    use rand::{thread_rng, Rng};
    use std::sync::Arc;

    #[test]
    fn test_fixed_width() {
        let cols = [
            Arc::new(Int16Array::from_iter([
                Some(1),
                Some(2),
                None,
                Some(-5),
                Some(2),
                Some(2),
                Some(0),
            ])) as ArrayRef,
            Arc::new(Float32Array::from_iter([
                Some(1.3),
                Some(2.5),
                None,
                Some(4.),
                Some(0.1),
                Some(-4.),
                Some(-0.),
            ])) as ArrayRef,
        ];

        let mut converter = RowConverter::new(vec![
            SortField::new(DataType::Int16),
            SortField::new(DataType::Float32),
        ]);
        let rows = converter.convert_columns(&cols).unwrap();

        assert_eq!(rows.offsets.as_ref(), &[0, 8, 16, 24, 32, 40, 48, 56]);
        assert_eq!(
            rows.buffer.as_ref(),
            &[
                1, 128, 1, //
                1, 191, 166, 102, 102, //
                1, 128, 2, //
                1, 192, 32, 0, 0, //
                0, 0, 0, //
                0, 0, 0, 0, 0, //
                1, 127, 251, //
                1, 192, 128, 0, 0, //
                1, 128, 2, //
                1, 189, 204, 204, 205, //
                1, 128, 2, //
                1, 63, 127, 255, 255, //
                1, 128, 0, //
                1, 127, 255, 255, 255 //
            ]
        );

        assert!(rows.row(3) < rows.row(6));
        assert!(rows.row(0) < rows.row(1));
        assert!(rows.row(3) < rows.row(0));
        assert!(rows.row(4) < rows.row(1));
        assert!(rows.row(5) < rows.row(4))
    }

    #[test]
    fn test_bool() {
        let mut converter = RowConverter::new(vec![SortField::new(DataType::Boolean)]);

        let col = Arc::new(BooleanArray::from_iter([None, Some(false), Some(true)]));
        let rows = converter.convert_columns(&[col]).unwrap();
        assert!(rows.row(2) > rows.row(1));
        assert!(rows.row(2) > rows.row(0));
        assert!(rows.row(1) > rows.row(0));

        let mut converter = RowConverter::new(vec![SortField::new_with_options(
            DataType::Boolean,
            SortOptions {
                descending: true,
                nulls_first: false,
            },
        )]);

        let col = Arc::new(BooleanArray::from_iter([None, Some(false), Some(true)]));
        let rows = converter.convert_columns(&[col]).unwrap();
        assert!(rows.row(2) < rows.row(1));
        assert!(rows.row(2) < rows.row(0));
        assert!(rows.row(1) < rows.row(0));
    }

    #[test]
    fn test_variable_width() {
        let col = Arc::new(StringArray::from_iter([
            Some("hello"),
            Some("he"),
            None,
            Some("foo"),
            Some(""),
        ]));

        let mut converter = RowConverter::new(vec![SortField::new(DataType::Utf8)]);
        let rows = converter.convert_columns(&[col]).unwrap();

        assert!(rows.row(1) < rows.row(0));
        assert!(rows.row(2) < rows.row(4));
        assert!(rows.row(3) < rows.row(0));
        assert!(rows.row(3) < rows.row(1));

        let col = Arc::new(BinaryArray::from_iter([
            None,
            Some(vec![0_u8; 0]),
            Some(vec![0_u8; 6]),
            Some(vec![0_u8; variable::BLOCK_SIZE]),
            Some(vec![0_u8; variable::BLOCK_SIZE + 1]),
            Some(vec![1_u8; 6]),
            Some(vec![1_u8; variable::BLOCK_SIZE]),
            Some(vec![1_u8; variable::BLOCK_SIZE + 1]),
            Some(vec![0xFF_u8; 6]),
            Some(vec![0xFF_u8; variable::BLOCK_SIZE]),
            Some(vec![0xFF_u8; variable::BLOCK_SIZE + 1]),
        ])) as ArrayRef;

        let mut converter = RowConverter::new(vec![SortField::new(DataType::Binary)]);
        let rows = converter.convert_columns(&[Arc::clone(&col)]).unwrap();

        for i in 0..rows.num_rows() {
            for j in i + 1..rows.num_rows() {
                assert!(
                    rows.row(i) < rows.row(j),
                    "{} < {} - {:?} < {:?}",
                    i,
                    j,
                    rows.row(i),
                    rows.row(j)
                );
            }
        }

        let mut converter = RowConverter::new(vec![SortField::new_with_options(
            DataType::Binary,
            SortOptions {
                descending: true,
                nulls_first: false,
            },
        )]);
        let rows = converter.convert_columns(&[col]).unwrap();

        for i in 0..rows.num_rows() {
            for j in i + 1..rows.num_rows() {
                assert!(
                    rows.row(i) > rows.row(j),
                    "{} > {} - {:?} > {:?}",
                    i,
                    j,
                    rows.row(i),
                    rows.row(j)
                );
            }
        }
    }

    #[test]
    fn test_string_dictionary() {
        let a = Arc::new(DictionaryArray::<Int32Type>::from_iter([
            Some("foo"),
            Some("hello"),
            Some("he"),
            None,
            Some("hello"),
            Some(""),
            Some("hello"),
            Some("hello"),
        ])) as ArrayRef;

        let mut converter =
            RowConverter::new(vec![SortField::new(a.data_type().clone())]);
        let rows_a = converter.convert_columns(&[Arc::clone(&a)]).unwrap();

        assert!(rows_a.row(3) < rows_a.row(5));
        assert!(rows_a.row(2) < rows_a.row(1));
        assert!(rows_a.row(0) < rows_a.row(1));
        assert!(rows_a.row(3) < rows_a.row(0));

        assert_eq!(rows_a.row(1), rows_a.row(4));
        assert_eq!(rows_a.row(1), rows_a.row(6));
        assert_eq!(rows_a.row(1), rows_a.row(7));

        let b = Arc::new(DictionaryArray::<Int32Type>::from_iter([
            Some("hello"),
            None,
            Some("cupcakes"),
        ]));

        let rows_b = converter.convert_columns(&[b]).unwrap();
        assert_eq!(rows_a.row(1), rows_b.row(0));
        assert_eq!(rows_a.row(3), rows_b.row(1));
        assert!(rows_b.row(2) < rows_a.row(0));

        let mut converter = RowConverter::new(vec![SortField::new_with_options(
            a.data_type().clone(),
            SortOptions {
                descending: true,
                nulls_first: false,
            },
        )]);

        let rows_c = converter.convert_columns(&[a]).unwrap();
        assert!(rows_c.row(3) > rows_c.row(5));
        assert!(rows_c.row(2) > rows_c.row(1));
        assert!(rows_c.row(0) > rows_c.row(1));
        assert!(rows_c.row(3) > rows_c.row(0));
    }

    #[test]
    fn test_primitive_dictionary() {
        let mut builder = PrimitiveDictionaryBuilder::<Int32Type, Int32Type>::new();
        builder.append(2).unwrap();
        builder.append(3).unwrap();
        builder.append(0).unwrap();
        builder.append_null();
        builder.append(5).unwrap();
        builder.append(3).unwrap();
        builder.append(-1).unwrap();

        let a = builder.finish();

        let mut converter =
            RowConverter::new(vec![SortField::new(a.data_type().clone())]);
        let rows = converter.convert_columns(&[Arc::new(a)]).unwrap();
        assert!(rows.row(0) < rows.row(1));
        assert!(rows.row(2) < rows.row(0));
        assert!(rows.row(3) < rows.row(2));
        assert!(rows.row(6) < rows.row(2));
        assert!(rows.row(3) < rows.row(6));
    }

    #[test]
    fn test_dictionary_nulls() {
        let values =
            Int32Array::from_iter([Some(1), Some(-1), None, Some(4), None]).into_data();
        let keys =
            Int32Array::from_iter([Some(0), Some(0), Some(1), Some(2), Some(4), None])
                .into_data();

        let data_type =
            DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Int32));
        let data = keys
            .into_builder()
            .data_type(data_type.clone())
            .child_data(vec![values])
            .build()
            .unwrap();

        let mut converter = RowConverter::new(vec![SortField::new(data_type)]);
        let rows = converter
            .convert_columns(&[Arc::new(DictionaryArray::<Int32Type>::from(data))])
            .unwrap();

        assert_eq!(rows.row(0), rows.row(1));
        assert_eq!(rows.row(3), rows.row(4));
        assert_eq!(rows.row(4), rows.row(5));
        assert!(rows.row(3) < rows.row(0));
    }

    fn generate_primitive_array<K>(len: usize, valid_percent: f64) -> PrimitiveArray<K>
    where
        K: ArrowPrimitiveType,
        Standard: Distribution<K::Native>,
    {
        let mut rng = thread_rng();
        (0..len)
            .map(|_| rng.gen_bool(valid_percent).then(|| rng.gen()))
            .collect()
    }

    fn generate_strings<O: OffsetSizeTrait>(
        len: usize,
        valid_percent: f64,
    ) -> GenericStringArray<O> {
        let mut rng = thread_rng();
        (0..len)
            .map(|_| {
                rng.gen_bool(valid_percent).then(|| {
                    let len = rng.gen_range(0..100);
                    let bytes = (0..len).map(|_| rng.gen_range(0..128)).collect();
                    String::from_utf8(bytes).unwrap()
                })
            })
            .collect()
    }

    fn generate_dictionary<K>(
        values: ArrayRef,
        len: usize,
        valid_percent: f64,
    ) -> DictionaryArray<K>
    where
        K: ArrowDictionaryKeyType,
        K::Native: SampleUniform,
    {
        let mut rng = thread_rng();
        let min_key = K::Native::from_usize(0).unwrap();
        let max_key = K::Native::from_usize(values.len()).unwrap();
        let keys: PrimitiveArray<K> = (0..len)
            .map(|_| {
                rng.gen_bool(valid_percent)
                    .then(|| rng.gen_range(min_key..max_key))
            })
            .collect();

        let data_type = DataType::Dictionary(
            Box::new(K::DATA_TYPE),
            Box::new(values.data_type().clone()),
        );

        let data = keys
            .into_data()
            .into_builder()
            .data_type(data_type)
            .add_child_data(values.data().clone())
            .build()
            .unwrap();

        DictionaryArray::from(data)
    }

    fn generate_column(len: usize) -> ArrayRef {
        let mut rng = thread_rng();
        match rng.gen_range(0..9) {
            0 => Arc::new(generate_primitive_array::<Int32Type>(len, 0.8)),
            1 => Arc::new(generate_primitive_array::<UInt32Type>(len, 0.8)),
            2 => Arc::new(generate_primitive_array::<Int64Type>(len, 0.8)),
            3 => Arc::new(generate_primitive_array::<UInt64Type>(len, 0.8)),
            4 => Arc::new(generate_primitive_array::<Float32Type>(len, 0.8)),
            5 => Arc::new(generate_primitive_array::<Float64Type>(len, 0.8)),
            6 => Arc::new(generate_strings::<i32>(len, 0.8)),
            7 => Arc::new(generate_dictionary::<Int64Type>(
                // Cannot test dictionaries containing null values because of #2687
                Arc::new(generate_strings::<i32>(rng.gen_range(1..len), 1.0)),
                len,
                0.8,
            )),
            8 => Arc::new(generate_dictionary::<Int64Type>(
                // Cannot test dictionaries containing null values because of #2687
                Arc::new(generate_primitive_array::<Int64Type>(
                    rng.gen_range(1..len),
                    1.0,
                )),
                len,
                0.8,
            )),
            _ => unreachable!(),
        }
    }

    fn print_row(cols: &[SortColumn], row: usize) -> String {
        let t: Vec<_> = cols
            .iter()
            .map(|x| array_value_to_string(&x.values, row).unwrap())
            .collect();
        t.join(",")
    }

    fn print_col_types(cols: &[SortColumn]) -> String {
        let t: Vec<_> = cols
            .iter()
            .map(|x| x.values.data_type().to_string())
            .collect();
        t.join(",")
    }

    #[test]
    #[cfg_attr(miri, ignore)]
    fn fuzz_test() {
        for _ in 0..100 {
            let mut rng = thread_rng();
            let num_columns = rng.gen_range(1..5);
            let len = rng.gen_range(5..100);
            let arrays: Vec<_> = (0..num_columns).map(|_| generate_column(len)).collect();

            let options: Vec<_> = (0..num_columns)
                .map(|_| SortOptions {
                    descending: rng.gen_bool(0.5),
                    nulls_first: rng.gen_bool(0.5),
                })
                .collect();

            let sort_columns: Vec<_> = options
                .iter()
                .zip(&arrays)
                .map(|(o, c)| SortColumn {
                    values: Arc::clone(c),
                    options: Some(*o),
                })
                .collect();

            let comparator = LexicographicalComparator::try_new(&sort_columns).unwrap();

            let columns = options
                .into_iter()
                .zip(&arrays)
                .map(|(o, a)| SortField::new_with_options(a.data_type().clone(), o))
                .collect();

            let mut converter = RowConverter::new(columns);
            let rows = converter.convert_columns(&arrays).unwrap();

            for i in 0..len {
                for j in 0..len {
                    let row_i = rows.row(i);
                    let row_j = rows.row(j);
                    let row_cmp = row_i.cmp(&row_j);
                    let lex_cmp = comparator.compare(&i, &j);
                    assert_eq!(
                        row_cmp,
                        lex_cmp,
                        "({:?} vs {:?}) vs ({:?} vs {:?}) for types {}",
                        print_row(&sort_columns, i),
                        print_row(&sort_columns, j),
                        row_i,
                        row_j,
                        print_col_types(&sort_columns)
                    );
                }
            }
        }
    }
}