feat(row): support remaining ordered dtypes

docs/specs/row-encoding.md

@@ -284,12 +284,14 @@ width is the smallest decimal value type for the decimal precision:
 | `5..=9` | `i32` |
 | `10..=18` | `i64` |
 | `19..=38` | `i128` |
+| `39..=76` | `i256` |
 
 The storage integer is encoded with the signed integer encoding described above. Decimal
 columns have one precision and scale, so ordering the scaled integer storage values matches
 ordering the decimal values in that column.
 
-`Decimal256` is not supported by row encoding.
+The signed integer transform is identical at every width, including the 32-byte `i256`
+representation used by Decimal256.
 
 ## UTF-8 and Binary
 
@@ -395,6 +397,29 @@ For a null fixed-size list, the body is canonicalized:
 
 A fixed-size list has fixed row width only when its element type has fixed row width.
 
+## Variable-Size List
+
+A variable-size list is ordered lexicographically by its elements. Null and empty lists use
+the variable-width sentinels. A non-empty list is encoded as:
+
+```text
+varlen_non_empty_sentinel || escaped_elements || list_terminator
+```
+
+Each byte of each recursively encoded element is escaped as `0x01 || byte`. The terminator is
+`0x00` for ascending fields and `0x02` for descending fields. This makes a shorter list that
+is an element-wise prefix sort before the longer list in ascending order and after it in
+descending order, without allowing a following column to affect that comparison.
+
+Element encodings use the same `RowSortField` as the list. Consequently, nested null placement
+remains independent of sort direction, consistent with structs and fixed-size lists.
+
+## Map
+
+A map is encoded as its ordered list of non-null `{key, value}` entry structs. Entry order is
+significant regardless of the dtype's `keys_sorted` producer assertion. Map comparison is
+therefore lexicographic first by entry, then by key and value within each entry.
+
 ## Nested Values
 
 Nested structs and fixed-size lists apply the same rules recursively. Each nullable parent
@@ -407,17 +432,16 @@ The current row encoder rejects types for which it does not define byte-sort sem
 
 | Type | Reason |
 | --- | --- |
-| Variable-size `List` | No row encoding order is defined. |
 | `Variant` | No row encoding order is defined. |
-| `Union` | No row encoding order is defined. |
-| `Extension` | No row encoding order is defined. |
-| `Decimal256` | Encoding is not implemented. |
+| `Union` | Values with different active variants have no defined order. |
+| `Extension` | The extension API does not declare whether logical ordering matches storage ordering. |
 
 The absence of these encodings is intentional. Adding one requires defining both the logical
 ordering and the exact byte representation that preserves that ordering.
 
-Temporal extensions could be added later by normalizing them to storage arrays at the
-row-encoder boundary, once the supported temporal ordering contract is made explicit.
+Extensions can be added once the extension API exposes an explicit contract declaring that
+logical ordering is identical to storage ordering. The row encoder must not infer that contract
+from the storage dtype alone.
 
 ## Size and Output Layout
 

fuzz/src/row.rs

@@ -28,12 +28,13 @@ use vortex_array::arrays::bool::BoolArrayExt;
 use vortex_array::arrays::fixed_size_list::FixedSizeListArrayExt;
 use vortex_array::arrays::fixed_size_list::FixedSizeListArraySlotsExt;
 use vortex_array::arrays::listview::ListViewArrayExt;
+use vortex_array::arrays::map::MapArraySlotsExt;
 use vortex_array::arrays::struct_::StructArrayExt;
-use vortex_array::dtype::BigCast;
 use vortex_array::dtype::DType;
-use vortex_array::dtype::DecimalType;
 use vortex_array::dtype::PType;
+use vortex_array::dtype::ToI256;
 use vortex_array::dtype::half::f16;
+use vortex_array::dtype::i256;
 use vortex_array::match_each_decimal_value_type;
 use vortex_array::match_each_integer_ptype;
 use vortex_error::VortexExpect;
@@ -201,12 +202,10 @@ fn dtype_row_encodable(dtype: &DType) -> bool {
         DType::Null | DType::Bool(_) | DType::Primitive(..) | DType::Utf8(_) | DType::Binary(_) => {
             true
         }
-        DType::Decimal(dt, _) => !matches!(
-            DecimalType::smallest_decimal_value_type(dt),
-            DecimalType::I256
-        ),
+        DType::Decimal(..) => true,
         DType::Struct(fields, _) => fields.fields().all(|f| dtype_row_encodable(&f)),
-        DType::FixedSizeList(elem, ..) => dtype_row_encodable(elem),
+        DType::List(elem, _) | DType::FixedSizeList(elem, ..) => dtype_row_encodable(elem),
+        DType::Map(map_dtype, _) => dtype_row_encodable(&map_dtype.entries_dtype()),
         _ => false,
     }
 }
@@ -230,8 +229,8 @@ fn collect_row_bytes(array: &ListViewArray, ctx: &mut ExecutionCtx) -> Vec<Vec<u
 enum RowKey {
     Null,
     Bool(bool),
-    /// Every integer ptype and decimal unscaled value (Decimal256 is unsupported upstream).
-    Int(i128),
+    /// Every integer ptype and decimal unscaled value.
+    Int(i256),
     /// IEEE-754 total-order bit key (sign-flipped bits), matching the encoded byte order.
     Float(u64),
     Bytes(Vec<u8>),
@@ -350,7 +349,12 @@ fn row_keys(array: &ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<Vec<RowKey
                 _ => match_each_integer_ptype!(a.ptype(), |P| {
                     a.as_slice::<P>()
                         .iter()
-                        .map(|v| RowKey::Int((*v).into()))
+                        .map(|v| {
+                            RowKey::Int(
+                                v.to_i256()
+                                    .vortex_expect("integer ptype must convert to i256"),
+                            )
+                        })
                         .collect::<Vec<_>>()
                 }),
             };
@@ -363,9 +367,11 @@ fn row_keys(array: &ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<Vec<RowKey
                 Ok((0..len)
                     .map(|i| {
                         if mask.value(i) {
-                            RowKey::Int(<i128 as BigCast>::from(buf[i]).vortex_expect(
-                                "valid decimal values fit i128 (Decimal256 dtypes are filtered)",
-                            ))
+                            RowKey::Int(
+                                buf[i]
+                                    .to_i256()
+                                    .vortex_expect("decimal value must convert to i256"),
+                            )
                         } else {
                             RowKey::Null
                         }
@@ -403,6 +409,19 @@ fn row_keys(array: &ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<Vec<RowKey
                 .collect();
             with_mask(keys, &a.into_array(), ctx)
         }
+        Canonical::List(a) => {
+            let mask = a.as_ref().validity()?.execute_mask(len, ctx)?;
+            let mut keys = Vec::with_capacity(len);
+            for i in 0..len {
+                if mask.value(i) {
+                    keys.push(RowKey::Composite(row_keys(&a.list_elements_at(i)?, ctx)?));
+                } else {
+                    keys.push(RowKey::Null);
+                }
+            }
+            Ok(keys)
+        }
+        Canonical::Map(a) => row_keys(a.entries(), ctx),
         c => unreachable!(
             "unsupported dtypes are rejected before oracle construction: {:?}",
             c.dtype()

vortex-row/Cargo.toml

@@ -18,6 +18,7 @@ version = { workspace = true }
 workspace = true
 
 [dependencies]
+num-traits = { workspace = true }
 smallvec = { workspace = true }
 vortex-array = { workspace = true }
 vortex-buffer = { workspace = true }

vortex-row/benches/row_encode.rs

@@ -3,8 +3,10 @@
 
 #![expect(
     clippy::unwrap_used,
+    clippy::expect_used,
     clippy::clone_on_ref_ptr,
     clippy::cloned_ref_to_slice_refs,
+    clippy::chunks_exact_to_as_chunks,
     clippy::redundant_clone
 )]
 
@@ -15,9 +17,13 @@
 use std::sync::Arc;
 use std::sync::LazyLock;
 
+use arrow_array::Array;
 use arrow_array::Int64Array;
 use arrow_array::StringArray;
 use arrow_array::StructArray as ArrowStructArray;
+use arrow_array::builder::Int64Builder;
+use arrow_array::builder::ListBuilder as ArrowListBuilder;
+use arrow_array::builder::StringBuilder;
 use arrow_row::RowConverter;
 use arrow_row::SortField as ArrowSortField;
 use arrow_schema::DataType;
@@ -34,6 +40,9 @@ use vortex_array::array_session;
 use vortex_array::arrays::PrimitiveArray;
 use vortex_array::arrays::StructArray;
 use vortex_array::arrays::VarBinViewArray;
+use vortex_array::arrays::listview::ListViewArray;
+use vortex_array::arrays::listview::ListViewArraySlotsExt;
+use vortex_array::validity::Validity;
 use vortex_row::RowEncoder;
 use vortex_session::VortexSession;
 
@@ -42,11 +51,16 @@ static GLOBAL: MiMalloc = MiMalloc;
 
 // Sized so the slowest scenario (struct_mixed) stays within the CodSpeed budget.
 const N: usize = 1_000;
+const LIST_LEN: usize = 8;
 
 static SESSION: LazyLock<VortexSession> = LazyLock::new(array_session);
 
 fn main() {
     LazyLock::force(&SESSION);
+    if std::env::args().any(|arg| arg == "--output-sizes") {
+        print_list_output_sizes();
+        return;
+    }
     divan::main();
 }
 
@@ -70,6 +84,82 @@ fn gen_words(n: usize, mean_len: usize, seed: u64) -> Vec<String> {
         .collect()
 }
 
+fn make_vortex_list(elements: vortex_array::ArrayRef) -> vortex_array::ArrayRef {
+    let offsets = PrimitiveArray::from_iter(
+        (0..N).map(|i| u32::try_from(i * LIST_LEN).expect("benchmark offset must fit u32")),
+    )
+    .into_array();
+    let list_len = u32::try_from(LIST_LEN).expect("benchmark list length must fit u32");
+    let sizes = PrimitiveArray::from_iter(std::iter::repeat_n(list_len, N)).into_array();
+    ListViewArray::new(elements, offsets, sizes, Validity::NonNullable).into_array()
+}
+
+fn arrow_output_bytes(rows: &arrow_row::Rows) -> u64 {
+    u64::try_from(rows.lengths().sum::<usize>()).expect("encoded output size must fit u64")
+}
+
+fn vortex_output_bytes(rows: &ListViewArray) -> u64 {
+    u64::try_from(rows.elements().len()).expect("encoded output size must fit u64")
+}
+
+fn list_i64_inputs() -> (arrow_array::ArrayRef, vortex_array::ArrayRef) {
+    let values = gen_i64(N * LIST_LEN, 11);
+    let mut builder = ArrowListBuilder::new(Int64Builder::with_capacity(values.len()));
+    for list in values.chunks_exact(LIST_LEN) {
+        builder.values().append_slice(list);
+        builder.append(true);
+    }
+    let arrow = Arc::new(builder.finish()) as arrow_array::ArrayRef;
+    let vortex = make_vortex_list(PrimitiveArray::from_iter(values).into_array());
+    (arrow, vortex)
+}
+
+fn list_utf8_inputs() -> (arrow_array::ArrayRef, vortex_array::ArrayRef) {
+    let values = gen_words(N * LIST_LEN, 16, 13);
+    let mut builder = ArrowListBuilder::new(StringBuilder::with_capacity(
+        values.len(),
+        values.iter().map(String::len).sum(),
+    ));
+    for list in values.chunks_exact(LIST_LEN) {
+        for value in list {
+            builder.values().append_value(value);
+        }
+        builder.append(true);
+    }
+    let arrow = Arc::new(builder.finish()) as arrow_array::ArrayRef;
+    let elements = VarBinViewArray::from_iter_str(values.iter().map(String::as_str)).into_array();
+    let vortex = make_vortex_list(elements);
+    (arrow, vortex)
+}
+
+fn list_output_sizes(arrow: &arrow_array::ArrayRef, vortex: &vortex_array::ArrayRef) -> (u64, u64) {
+    let converter = RowConverter::new(vec![ArrowSortField::new(arrow.data_type().clone())])
+        .expect("benchmark dtype must be supported by arrow-row");
+    let arrow_bytes = arrow_output_bytes(
+        &converter
+            .convert_columns(&[Arc::clone(arrow)])
+            .expect("arrow-row benchmark encode must succeed"),
+    );
+    let mut ctx = SESSION.create_execution_ctx();
+    let vortex_bytes = vortex_output_bytes(
+        &RowEncoder::default()
+            .encode(&[vortex.clone()], &mut ctx)
+            .expect("Vortex benchmark encode must succeed"),
+    );
+    (arrow_bytes, vortex_bytes)
+}
+
+fn print_list_output_sizes() {
+    println!("case,arrow_row_bytes,vortex_bytes");
+    for (name, (arrow, vortex)) in [
+        ("list_i64", list_i64_inputs()),
+        ("list_utf8", list_utf8_inputs()),
+    ] {
+        let (arrow_bytes, vortex_bytes) = list_output_sizes(&arrow, &vortex);
+        println!("{name},{arrow_bytes},{vortex_bytes}");
+    }
+}
+
 // ---------- primitive_i64 ----------
 
 #[divan::bench]
@@ -180,3 +270,51 @@ fn struct_mixed_vortex(bencher: divan::Bencher) {
         .with_inputs(|| SESSION.create_execution_ctx())
         .bench_local_values(|mut ctx| encoder.encode(&[struct_arr.clone()], &mut ctx).unwrap())
 }
+
+// ---------- list_i64 ----------
+
+#[divan::bench]
+fn list_i64_arrow_row(bencher: divan::Bencher) {
+    let (arr, _) = list_i64_inputs();
+    let conv = RowConverter::new(vec![ArrowSortField::new(arr.data_type().clone())]).unwrap();
+    let output_bytes = arrow_output_bytes(&conv.convert_columns(&[arr.clone()]).unwrap());
+    bencher
+        .counter(BytesCount::new(output_bytes))
+        .bench_local(|| conv.convert_columns(&[arr.clone()]).unwrap())
+}
+
+#[divan::bench]
+fn list_i64_vortex(bencher: divan::Bencher) {
+    let (_, list) = list_i64_inputs();
+    let encoder = RowEncoder::default();
+    let mut ctx = SESSION.create_execution_ctx();
+    let output_bytes = vortex_output_bytes(&encoder.encode(&[list.clone()], &mut ctx).unwrap());
+    bencher
+        .counter(BytesCount::new(output_bytes))
+        .with_inputs(|| SESSION.create_execution_ctx())
+        .bench_local_values(|mut ctx| encoder.encode(&[list.clone()], &mut ctx).unwrap())
+}
+
+// ---------- list_utf8 ----------
+
+#[divan::bench]
+fn list_utf8_arrow_row(bencher: divan::Bencher) {
+    let (arr, _) = list_utf8_inputs();
+    let conv = RowConverter::new(vec![ArrowSortField::new(arr.data_type().clone())]).unwrap();
+    let output_bytes = arrow_output_bytes(&conv.convert_columns(&[arr.clone()]).unwrap());
+    bencher
+        .counter(BytesCount::new(output_bytes))
+        .bench_local(|| conv.convert_columns(&[arr.clone()]).unwrap())
+}
+
+#[divan::bench]
+fn list_utf8_vortex(bencher: divan::Bencher) {
+    let (_, list) = list_utf8_inputs();
+    let encoder = RowEncoder::default();
+    let mut ctx = SESSION.create_execution_ctx();
+    let output_bytes = vortex_output_bytes(&encoder.encode(&[list.clone()], &mut ctx).unwrap());
+    bencher
+        .counter(BytesCount::new(output_bytes))
+        .with_inputs(|| SESSION.create_execution_ctx())
+        .bench_local_values(|mut ctx| encoder.encode(&[list.clone()], &mut ctx).unwrap())
+}