diff --git a/crates/polars-core/src/chunked_array/object/mod.rs b/crates/polars-core/src/chunked_array/object/mod.rs
index c834a61fa990..94a6c203d19a 100644
--- a/crates/polars-core/src/chunked_array/object/mod.rs
+++ b/crates/polars-core/src/chunked_array/object/mod.rs
@@ -33,6 +33,14 @@ pub trait PolarsObjectSafe: Any + Debug + Send + Sync + Display {
     fn as_any(&self) -> &dyn Any;
 
     fn to_boxed(&self) -> Box<dyn PolarsObjectSafe>;
+
+    fn equal(&self, other: &dyn PolarsObjectSafe) -> bool;
+}
+
+impl PartialEq for &dyn PolarsObjectSafe {
+    fn eq(&self, other: &Self) -> bool {
+        self.equal(*other)
+    }
 }
 
 /// Values need to implement this so that they can be stored into a Series and DataFrame
@@ -55,6 +63,13 @@ impl<T: PolarsObject> PolarsObjectSafe for T {
     fn to_boxed(&self) -> Box<dyn PolarsObjectSafe> {
         Box::new(self.clone())
     }
+
+    fn equal(&self, other: &dyn PolarsObjectSafe) -> bool {
+        let Some(other) = other.as_any().downcast_ref::<T>() else {
+            return false;
+        };
+        self == other
+    }
 }
 
 pub type ObjectValueIter<'a, T> = std::slice::Iter<'a, T>;
diff --git a/crates/polars-core/src/chunked_array/ops/sort/arg_sort_multiple.rs b/crates/polars-core/src/chunked_array/ops/sort/arg_sort_multiple.rs
index 44a67af75294..4deffaab0165 100644
--- a/crates/polars-core/src/chunked_array/ops/sort/arg_sort_multiple.rs
+++ b/crates/polars-core/src/chunked_array/ops/sort/arg_sort_multiple.rs
@@ -3,7 +3,6 @@ use polars_row::{convert_columns, RowsEncoded, SortField};
 use polars_utils::iter::EnumerateIdxTrait;
 
 use super::*;
-#[cfg(feature = "dtype-struct")]
 use crate::utils::_split_offsets;
 
 pub(crate) fn args_validate<T: PolarsDataType>(
@@ -88,8 +87,7 @@ pub fn _get_rows_encoded_compat_array(by: &Series) -> PolarsResult<ArrayRef> {
     Ok(out)
 }
 
-#[cfg(feature = "dtype-struct")]
-pub(crate) fn encode_rows_vertical(by: &[Series]) -> PolarsResult<BinaryOffsetChunked> {
+pub(crate) fn encode_rows_vertical_par_default(by: &[Series]) -> PolarsResult<BinaryOffsetChunked> {
     let n_threads = POOL.current_num_threads();
     let len = by[0].len();
     let splits = _split_offsets(len, n_threads);
@@ -108,6 +106,12 @@ pub(crate) fn encode_rows_vertical(by: &[Series]) -> PolarsResult<BinaryOffsetCh
     Ok(BinaryOffsetChunked::from_chunk_iter("", chunks?))
 }
 
+pub(crate) fn encode_rows_default(by: &[Series]) -> PolarsResult<BinaryOffsetChunked> {
+    let descending = vec![false; by.len()];
+    let rows = _get_rows_encoded(by, &descending, false)?;
+    Ok(BinaryOffsetChunked::with_chunk("", rows.into_array()))
+}
+
 pub fn _get_rows_encoded(
     by: &[Series],
     descending: &[bool],
diff --git a/crates/polars-core/src/chunked_array/ops/sort/mod.rs b/crates/polars-core/src/chunked_array/ops/sort/mod.rs
index 1610ea6b2fb4..2611227031f7 100644
--- a/crates/polars-core/src/chunked_array/ops/sort/mod.rs
+++ b/crates/polars-core/src/chunked_array/ops/sort/mod.rs
@@ -15,9 +15,7 @@ use rayon::prelude::*;
 pub use slice::*;
 
 use crate::prelude::compare_inner::TotalOrdInner;
-#[cfg(feature = "dtype-struct")]
-use crate::prelude::sort::arg_sort_multiple::_get_rows_encoded_ca;
-use crate::prelude::sort::arg_sort_multiple::{arg_sort_multiple_impl, args_validate};
+use crate::prelude::sort::arg_sort_multiple::*;
 use crate::prelude::*;
 use crate::series::IsSorted;
 use crate::utils::NoNull;
diff --git a/crates/polars-core/src/datatypes/any_value.rs b/crates/polars-core/src/datatypes/any_value.rs
index 523b7a9939d4..251db06b0044 100644
--- a/crates/polars-core/src/datatypes/any_value.rs
+++ b/crates/polars-core/src/datatypes/any_value.rs
@@ -946,6 +946,8 @@ impl AnyValue<'_> {
                 // 1.2 at scale 1, and 1.20 at scale 2, are not equal.
                 *v_l == *v_r && *scale_l == *scale_r
             },
+            #[cfg(feature = "object")]
+            (Object(l), Object(r)) => l == r,
             _ => false,
         }
     }
diff --git a/crates/polars-core/src/datatypes/dtype.rs b/crates/polars-core/src/datatypes/dtype.rs
index db1f52497ff6..3f7b58a6379a 100644
--- a/crates/polars-core/src/datatypes/dtype.rs
+++ b/crates/polars-core/src/datatypes/dtype.rs
@@ -241,6 +241,21 @@ impl DataType {
         matches!(self, DataType::Binary)
     }
 
+    pub fn is_object(&self) -> bool {
+        #[cfg(feature = "object")]
+        {
+            matches!(self, DataType::Object(_, _))
+        }
+        #[cfg(not(feature = "object"))]
+        {
+            false
+        }
+    }
+
+    pub fn is_null(&self) -> bool {
+        matches!(self, DataType::Null)
+    }
+
     pub fn contains_views(&self) -> bool {
         use DataType::*;
         match self {
diff --git a/crates/polars-core/src/frame/group_by/hashing.rs b/crates/polars-core/src/frame/group_by/hashing.rs
index b3e85e5dacb5..418471abc388 100644
--- a/crates/polars-core/src/frame/group_by/hashing.rs
+++ b/crates/polars-core/src/frame/group_by/hashing.rs
@@ -1,19 +1,16 @@
-use std::hash::{BuildHasher, Hash};
+use std::hash::{BuildHasher, Hash, Hasher};
 
-use hashbrown::hash_map::{Entry, RawEntryMut};
-use hashbrown::HashMap;
+use hashbrown::hash_map::RawEntryMut;
 use polars_utils::hashing::{hash_to_partition, DirtyHash};
 use polars_utils::idx_vec::IdxVec;
-use polars_utils::iter::EnumerateIdxTrait;
 use polars_utils::sync::SyncPtr;
 use polars_utils::total_ord::{ToTotalOrd, TotalHash};
 use polars_utils::unitvec;
 use rayon::prelude::*;
 
 use crate::hashing::*;
-use crate::prelude::compare_inner::TotalEqInner;
 use crate::prelude::*;
-use crate::utils::{flatten, split_df};
+use crate::utils::flatten;
 use crate::POOL;
 
 fn get_init_size() -> usize {
@@ -76,89 +73,33 @@ fn finish_group_order(mut out: Vec<Vec<IdxItem>>, sorted: bool) -> GroupsProxy {
     }
 }
 
-// The inner vecs should be sorted by [`IdxSize`]
-// the group_by multiple keys variants suffice
-// this requirements as they use an [`IdxMap`] strategy
-fn finish_group_order_vecs(
-    mut vecs: Vec<(Vec<IdxSize>, Vec<IdxVec>)>,
-    sorted: bool,
-) -> GroupsProxy {
-    if sorted {
-        if vecs.len() == 1 {
-            let (first, all) = vecs.pop().unwrap();
-            return GroupsProxy::Idx(GroupsIdx::new(first, all, true));
-        }
-
-        let cap = vecs.iter().map(|v| v.0.len()).sum::<usize>();
-        let offsets = vecs
-            .iter()
-            .scan(0_usize, |acc, v| {
-                let out = *acc;
-                *acc += v.0.len();
-                Some(out)
-            })
-            .collect::<Vec<_>>();
-
-        // we write (first, all) tuple because of sorting
-        let mut items = Vec::with_capacity(cap);
-        let items_ptr = unsafe { SyncPtr::new(items.as_mut_ptr()) };
-
-        POOL.install(|| {
-            vecs.into_par_iter()
-                .zip(offsets)
-                .for_each(|((first, all), offset)| {
-                    // pre-sort every array not needed as items are already sorted
-                    // this is due to using an index hashmap
-
-                    unsafe {
-                        let mut items_ptr: *mut (IdxSize, IdxVec) = items_ptr.get();
-                        items_ptr = items_ptr.add(offset);
-
-                        // give the compiler some info
-                        // maybe it may elide some loop counters
-                        assert_eq!(first.len(), all.len());
-                        for (i, (first, all)) in first.into_iter().zip(all).enumerate() {
-                            std::ptr::write(items_ptr.add(i), (first, all))
-                        }
-                    }
-                });
-        });
-        unsafe {
-            items.set_len(cap);
-        }
-        // sort again
-        items.sort_unstable_by_key(|g| g.0);
-
-        let mut idx = GroupsIdx::from_iter(items);
-        idx.sorted = true;
-        GroupsProxy::Idx(idx)
-    } else {
-        // this materialization is parallel in the from impl.
-        GroupsProxy::Idx(GroupsIdx::from(vecs))
-    }
-}
-
 pub(crate) fn group_by<T>(a: impl Iterator<Item = T>, sorted: bool) -> GroupsProxy
 where
-    T: TotalHash + TotalEq + ToTotalOrd,
-    <T as ToTotalOrd>::TotalOrdItem: Hash + Eq,
+    T: TotalHash + TotalEq,
 {
     let init_size = get_init_size();
-    let mut hash_tbl: PlHashMap<T::TotalOrdItem, (IdxSize, IdxVec)> =
-        PlHashMap::with_capacity(init_size);
+    let mut hash_tbl: PlHashMap<T, (IdxSize, IdxVec)> = PlHashMap::with_capacity(init_size);
+    let hasher = hash_tbl.hasher().clone();
     let mut cnt = 0;
     a.for_each(|k| {
-        let k = k.to_total_ord();
         let idx = cnt;
         cnt += 1;
-        let entry = hash_tbl.entry(k);
+
+        let mut state = hasher.build_hasher();
+        k.tot_hash(&mut state);
+        let h = state.finish();
+        let entry = hash_tbl.raw_entry_mut().from_hash(h, |k_| k.tot_eq(k_));
 
         match entry {
-            Entry::Vacant(entry) => {
+            RawEntryMut::Vacant(entry) => {
                 let tuples = unitvec![idx];
-                entry.insert((idx, tuples));
+                entry.insert_with_hasher(h, k, (idx, tuples), |k| {
+                    let mut state = hasher.build_hasher();
+                    k.tot_hash(&mut state);
+                    state.finish()
+                });
             },
-            Entry::Occupied(mut entry) => {
+            RawEntryMut::Occupied(mut entry) => {
                 let v = entry.get_mut();
                 v.1.push(idx);
             },
@@ -318,206 +259,3 @@ where
     });
     finish_group_order(out, sorted)
 }
-
-#[inline]
-pub(crate) unsafe fn compare_keys<'a>(
-    keys_cmp: &'a [Box<dyn TotalEqInner + 'a>],
-    idx_a: usize,
-    idx_b: usize,
-) -> bool {
-    for cmp in keys_cmp {
-        if !cmp.eq_element_unchecked(idx_a, idx_b) {
-            return false;
-        }
-    }
-    true
-}
-
-// Differs in the because this one uses the TotalEqInner trait objects
-// is faster when multiple chunks. Not yet used in join.
-pub(crate) fn populate_multiple_key_hashmap2<'a, V, H, F, G>(
-    hash_tbl: &mut HashMap<IdxHash, V, H>,
-    // row index
-    idx: IdxSize,
-    // hash
-    original_h: u64,
-    // keys of the hash table (will not be inserted, the indexes will be used)
-    // the keys are needed for the equality check
-    keys_cmp: &'a [Box<dyn TotalEqInner + 'a>],
-    // value to insert
-    vacant_fn: G,
-    // function that gets a mutable ref to the occupied value in the hash table
-    occupied_fn: F,
-) where
-    G: Fn() -> V,
-    F: Fn(&mut V),
-    H: BuildHasher,
-{
-    let entry = hash_tbl
-        .raw_entry_mut()
-        // uses the idx to probe rows in the original DataFrame with keys
-        // to check equality to find an entry
-        // this does not invalidate the hashmap as this equality function is not used
-        // during rehashing/resize (then the keys are already known to be unique).
-        // Only during insertion and probing an equality function is needed
-        .from_hash(original_h, |idx_hash| {
-            // first check the hash values before we incur
-            // cache misses
-            original_h == idx_hash.hash && {
-                let key_idx = idx_hash.idx;
-                // SAFETY:
-                // indices in a group_by operation are always in bounds.
-                unsafe { compare_keys(keys_cmp, key_idx as usize, idx as usize) }
-            }
-        });
-    match entry {
-        RawEntryMut::Vacant(entry) => {
-            entry.insert_hashed_nocheck(original_h, IdxHash::new(idx, original_h), vacant_fn());
-        },
-        RawEntryMut::Occupied(mut entry) => {
-            let (_k, v) = entry.get_key_value_mut();
-            occupied_fn(v);
-        },
-    }
-}
-
-pub(crate) fn group_by_threaded_multiple_keys_flat(
-    mut keys: DataFrame,
-    n_partitions: usize,
-    sorted: bool,
-) -> PolarsResult<GroupsProxy> {
-    let dfs = split_df(&mut keys, n_partitions).unwrap();
-    let (hashes, _random_state) = _df_rows_to_hashes_threaded_vertical(&dfs, None)?;
-
-    let init_size = get_init_size();
-
-    // trait object to compare inner types.
-    let keys_cmp = keys
-        .iter()
-        .map(|s| s.into_total_eq_inner())
-        .collect::<Vec<_>>();
-
-    // We will create a hashtable in every thread.
-    // We use the hash to partition the keys to the matching hashtable.
-    // Every thread traverses all keys/hashes and ignores the ones that doesn't fall in that partition.
-    let v = POOL.install(|| {
-        (0..n_partitions)
-            .into_par_iter()
-            .map(|thread_no| {
-                let hashes = &hashes;
-
-                // IndexMap, the indexes are stored in flat vectors
-                // this ensures that order remains and iteration is fast
-                let mut hash_tbl: HashMap<IdxHash, IdxSize, IdBuildHasher> =
-                    HashMap::with_capacity_and_hasher(init_size, Default::default());
-                let mut first_vals = Vec::with_capacity(init_size);
-                let mut all_vals = Vec::with_capacity(init_size);
-
-                // put the buffers behind a pointer so we can access them from as the bchk doesn't allow
-                // 2 mutable borrows (this is safe as we don't alias)
-                // even if the vecs reallocate, we have a pointer to the stack vec, and thus always
-                // access the proper data.
-                let all_buf_ptr = &mut all_vals as *mut Vec<IdxVec> as *const Vec<IdxVec>;
-                let first_buf_ptr = &mut first_vals as *mut Vec<IdxSize> as *const Vec<IdxSize>;
-
-                let mut offset = 0;
-                for hashes in hashes {
-                    let len = hashes.len() as IdxSize;
-
-                    let mut idx = 0;
-                    for hashes_chunk in hashes.data_views() {
-                        for &h in hashes_chunk {
-                            // partition hashes by thread no.
-                            // So only a part of the hashes go to this hashmap
-                            if thread_no == hash_to_partition(h, n_partitions) {
-                                let row_idx = idx + offset;
-                                populate_multiple_key_hashmap2(
-                                    &mut hash_tbl,
-                                    row_idx,
-                                    h,
-                                    &keys_cmp,
-                                    || unsafe {
-                                        let first_vals = &mut *(first_buf_ptr as *mut Vec<IdxSize>);
-                                        let all_vals = &mut *(all_buf_ptr as *mut Vec<IdxVec>);
-                                        let offset_idx = first_vals.len() as IdxSize;
-
-                                        let tuples = unitvec![row_idx];
-                                        all_vals.push(tuples);
-                                        first_vals.push(row_idx);
-                                        offset_idx
-                                    },
-                                    |v| unsafe {
-                                        let all_vals = &mut *(all_buf_ptr as *mut Vec<IdxVec>);
-                                        let offset_idx = *v;
-                                        let buf = all_vals.get_unchecked_mut(offset_idx as usize);
-                                        buf.push(row_idx)
-                                    },
-                                );
-                            }
-                            idx += 1;
-                        }
-                    }
-
-                    offset += len;
-                }
-                (first_vals, all_vals)
-            })
-            .collect::<Vec<_>>()
-    });
-    Ok(finish_group_order_vecs(v, sorted))
-}
-
-pub(crate) fn group_by_multiple_keys(keys: DataFrame, sorted: bool) -> PolarsResult<GroupsProxy> {
-    let mut hashes = Vec::with_capacity(keys.height());
-    let _ = series_to_hashes(keys.get_columns(), None, &mut hashes)?;
-
-    let init_size = get_init_size();
-
-    // trait object to compare inner types.
-    let keys_cmp = keys
-        .iter()
-        .map(|s| s.into_total_eq_inner())
-        .collect::<Vec<_>>();
-
-    // IndexMap, the indexes are stored in flat vectors
-    // this ensures that order remains and iteration is fast
-    let mut hash_tbl: HashMap<IdxHash, IdxSize, IdBuildHasher> =
-        HashMap::with_capacity_and_hasher(init_size, Default::default());
-    let mut first_vals = Vec::with_capacity(init_size);
-    let mut all_vals = Vec::with_capacity(init_size);
-
-    // put the buffers behind a pointer so we can access them from as the bchk doesn't allow
-    // 2 mutable borrows (this is safe as we don't alias)
-    // even if the vecs reallocate, we have a pointer to the stack vec, and thus always
-    // access the proper data.
-    let all_buf_ptr = &mut all_vals as *mut Vec<IdxVec> as *const Vec<IdxVec>;
-    let first_buf_ptr = &mut first_vals as *mut Vec<IdxSize> as *const Vec<IdxSize>;
-
-    for (row_idx, h) in hashes.into_iter().enumerate_idx() {
-        populate_multiple_key_hashmap2(
-            &mut hash_tbl,
-            row_idx,
-            h,
-            &keys_cmp,
-            || unsafe {
-                let first_vals = &mut *(first_buf_ptr as *mut Vec<IdxSize>);
-                let all_vals = &mut *(all_buf_ptr as *mut Vec<IdxVec>);
-                let offset_idx = first_vals.len() as IdxSize;
-
-                let tuples = unitvec![row_idx];
-                all_vals.push(tuples);
-                first_vals.push(row_idx);
-                offset_idx
-            },
-            |v| unsafe {
-                let all_vals = &mut *(all_buf_ptr as *mut Vec<IdxVec>);
-                let offset_idx = *v;
-                let buf = all_vals.get_unchecked_mut(offset_idx as usize);
-                buf.push(row_idx)
-            },
-        );
-    }
-
-    let v = vec![(first_vals, all_vals)];
-    Ok(finish_group_order_vecs(v, sorted))
-}
diff --git a/crates/polars-core/src/frame/group_by/mod.rs b/crates/polars-core/src/frame/group_by/mod.rs
index 75df1e198c50..bc7e90406cf7 100644
--- a/crates/polars-core/src/frame/group_by/mod.rs
+++ b/crates/polars-core/src/frame/group_by/mod.rs
@@ -22,37 +22,9 @@ mod proxy;
 pub use into_groups::*;
 pub use proxy::*;
 
-#[cfg(feature = "dtype-struct")]
-use crate::prelude::sort::arg_sort_multiple::encode_rows_vertical;
-
-// This will remove the sorted flag on signed integers
-fn prepare_dataframe_unsorted(by: &[Series]) -> DataFrame {
-    let columns = by
-        .iter()
-        .map(|s| match s.dtype() {
-            #[cfg(feature = "dtype-categorical")]
-            DataType::Categorical(_, _) | DataType::Enum(_, _) => {
-                s.cast(&DataType::UInt32).unwrap()
-            },
-            _ => {
-                if s.dtype().to_physical().is_numeric() {
-                    let s = s.to_physical_repr();
-
-                    if s.dtype().is_float() {
-                        s.into_owned().into_series()
-                    } else if s.bit_repr_is_large() {
-                        s.bit_repr_large().into_series()
-                    } else {
-                        s.bit_repr_small().into_series()
-                    }
-                } else {
-                    s.clone()
-                }
-            },
-        })
-        .collect();
-    unsafe { DataFrame::new_no_checks(columns) }
-}
+use crate::prelude::sort::arg_sort_multiple::{
+    encode_rows_default, encode_rows_vertical_par_default,
+};
 
 impl DataFrame {
     pub fn group_by_with_series(
@@ -82,25 +54,42 @@ impl DataFrame {
             }
         };
 
-        let n_partitions = _set_partition_size();
-
         let groups = if by.len() == 1 {
             let series = &by[0];
             series.group_tuples(multithreaded, sorted)
-        } else {
-            #[cfg(feature = "dtype-struct")]
+        } else if by.iter().any(|s| s.dtype().is_object()) {
+            #[cfg(feature = "object")]
             {
-                if by.iter().any(|s| matches!(s.dtype(), DataType::Struct(_))) {
-                    let rows = encode_rows_vertical(&by)?;
-                    let groups = rows.group_tuples(multithreaded, sorted)?;
-                    return Ok(GroupBy::new(self, by, groups, None));
-                }
+                let mut df = DataFrame::new(by.clone()).unwrap();
+                let n = df.height();
+                let rows = df.to_av_rows();
+                let iter = (0..n).map(|i| rows.get(i));
+                Ok(group_by(iter, sorted))
+            }
+            #[cfg(not(feature = "object"))]
+            {
+                unreachable!()
             }
-            let keys_df = prepare_dataframe_unsorted(&by);
-            if multithreaded {
-                group_by_threaded_multiple_keys_flat(keys_df, n_partitions, sorted)
+        } else {
+            // Skip null dtype.
+            let by = by
+                .iter()
+                .filter(|s| !s.dtype().is_null())
+                .cloned()
+                .collect::<Vec<_>>();
+            if by.is_empty() {
+                Ok(GroupsProxy::Slice {
+                    groups: vec![[0, self.height() as IdxSize]],
+                    rolling: false,
+                })
             } else {
-                group_by_multiple_keys(keys_df, sorted)
+                let rows = if multithreaded {
+                    encode_rows_vertical_par_default(&by)
+                } else {
+                    encode_rows_default(&by)
+                }?
+                .into_series();
+                rows.group_tuples(multithreaded, sorted)
             }
         };
         Ok(GroupBy::new(self, by, groups?, None))
diff --git a/crates/polars-core/src/frame/mod.rs b/crates/polars-core/src/frame/mod.rs
index 0d8a696d190a..83c8292918f4 100644
--- a/crates/polars-core/src/frame/mod.rs
+++ b/crates/polars-core/src/frame/mod.rs
@@ -17,7 +17,7 @@ pub mod explode;
 mod from;
 #[cfg(feature = "algorithm_group_by")]
 pub mod group_by;
-#[cfg(feature = "rows")]
+#[cfg(any(feature = "rows", feature = "object"))]
 pub mod row;
 mod top_k;
 mod upstream_traits;
diff --git a/crates/polars-core/src/frame/row/mod.rs b/crates/polars-core/src/frame/row/mod.rs
index 05701260416f..e9cf92ffad13 100644
--- a/crates/polars-core/src/frame/row/mod.rs
+++ b/crates/polars-core/src/frame/row/mod.rs
@@ -4,16 +4,89 @@ mod transpose;
 
 use std::borrow::Borrow;
 use std::fmt::Debug;
+#[cfg(feature = "object")]
+use std::hash::{Hash, Hasher};
 use std::hint::unreachable_unchecked;
 
 use arrow::bitmap::Bitmap;
 pub use av_buffer::*;
+#[cfg(feature = "object")]
+use polars_utils::total_ord::TotalHash;
 use rayon::prelude::*;
 
 use crate::prelude::*;
 use crate::utils::{dtypes_to_schema, dtypes_to_supertype, try_get_supertype};
 use crate::POOL;
 
+#[cfg(feature = "object")]
+pub(crate) struct AnyValueRows<'a> {
+    vals: Vec<AnyValue<'a>>,
+    width: usize,
+}
+
+#[cfg(feature = "object")]
+pub(crate) struct AnyValueRow<'a>(&'a [AnyValue<'a>]);
+
+#[cfg(feature = "object")]
+impl<'a> AnyValueRows<'a> {
+    pub(crate) fn get(&'a self, i: usize) -> AnyValueRow<'a> {
+        let start = i * self.width;
+        let end = (i + 1) * self.width;
+        AnyValueRow(&self.vals[start..end])
+    }
+}
+
+#[cfg(feature = "object")]
+impl TotalEq for AnyValueRow<'_> {
+    fn tot_eq(&self, other: &Self) -> bool {
+        let lhs = self.0;
+        let rhs = other.0;
+
+        // Should only be used in that context.
+        debug_assert_eq!(lhs.len(), rhs.len());
+        lhs.iter().zip(rhs.iter()).all(|(l, r)| l == r)
+    }
+}
+
+#[cfg(feature = "object")]
+impl TotalHash for AnyValueRow<'_> {
+    fn tot_hash<H>(&self, state: &mut H)
+    where
+        H: Hasher,
+    {
+        self.0.iter().for_each(|av| av.hash(state))
+    }
+}
+
+impl DataFrame {
+    #[cfg(feature = "object")]
+    #[allow(clippy::wrong_self_convention)]
+    // Create indexable rows in a single allocation.
+    pub(crate) fn to_av_rows(&mut self) -> AnyValueRows<'_> {
+        self.as_single_chunk_par();
+        let width = self.width();
+        let size = width * self.height();
+        let mut buf = vec![AnyValue::Null; size];
+        for (col_i, s) in self.columns.iter().enumerate() {
+            match s.dtype() {
+                #[cfg(feature = "object")]
+                DataType::Object(_, _) => {
+                    for row_i in 0..s.len() {
+                        let av = s.get(row_i).unwrap();
+                        buf[row_i * width + col_i] = av
+                    }
+                },
+                _ => {
+                    for (row_i, av) in s.iter().enumerate() {
+                        buf[row_i * width + col_i] = av
+                    }
+                },
+            }
+        }
+        AnyValueRows { vals: buf, width }
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Default)]
 pub struct Row<'a>(pub Vec<AnyValue<'a>>);
 
diff --git a/crates/polars-core/src/series/implementations/struct_.rs b/crates/polars-core/src/series/implementations/struct_.rs
index 219b88f36f7d..861599cb91b6 100644
--- a/crates/polars-core/src/series/implementations/struct_.rs
+++ b/crates/polars-core/src/series/implementations/struct_.rs
@@ -272,9 +272,8 @@ impl SeriesTrait for SeriesWrap<StructChunked> {
         if self.len() == 1 {
             return Ok(IdxCa::new_vec(self.name(), vec![0 as IdxSize]));
         }
-        // TODO! try row encoding
         let main_thread = POOL.current_thread_index().is_none();
-        let groups = self.group_tuples(main_thread, false)?;
+        let groups = self.group_tuples(main_thread, true)?;
         let first = groups.take_group_firsts();
         Ok(IdxCa::from_vec(self.name(), first))
     }
diff --git a/crates/polars-utils/src/total_ord.rs b/crates/polars-utils/src/total_ord.rs
index 5b9af065779f..f5d7b22a95aa 100644
--- a/crates/polars-utils/src/total_ord.rs
+++ b/crates/polars-utils/src/total_ord.rs
@@ -90,46 +90,46 @@ pub struct TotalOrdWrap<T>(pub T);
 unsafe impl<T> TransparentWrapper<T> for TotalOrdWrap<T> {}
 
 impl<T: TotalOrd> PartialOrd for TotalOrdWrap<T> {
-    #[inline]
+    #[inline(always)]
     fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
         Some(self.cmp(other))
     }
 
-    #[inline]
+    #[inline(always)]
     fn lt(&self, other: &Self) -> bool {
         self.0.tot_lt(&other.0)
     }
 
-    #[inline]
+    #[inline(always)]
     fn le(&self, other: &Self) -> bool {
         self.0.tot_le(&other.0)
     }
 
-    #[inline]
+    #[inline(always)]
     fn gt(&self, other: &Self) -> bool {
         self.0.tot_gt(&other.0)
     }
 
-    #[inline]
+    #[inline(always)]
     fn ge(&self, other: &Self) -> bool {
         self.0.tot_ge(&other.0)
     }
 }
 
 impl<T: TotalOrd> Ord for TotalOrdWrap<T> {
-    #[inline]
+    #[inline(always)]
     fn cmp(&self, other: &Self) -> Ordering {
         self.0.tot_cmp(&other.0)
     }
 }
 
 impl<T: TotalEq> PartialEq for TotalOrdWrap<T> {
-    #[inline]
+    #[inline(always)]
     fn eq(&self, other: &Self) -> bool {
         self.0.tot_eq(&other.0)
     }
 
-    #[inline]
+    #[inline(always)]
     #[allow(clippy::partialeq_ne_impl)]
     fn ne(&self, other: &Self) -> bool {
         self.0.tot_ne(&other.0)
@@ -139,7 +139,7 @@ impl<T: TotalEq> PartialEq for TotalOrdWrap<T> {
 impl<T: TotalEq> Eq for TotalOrdWrap<T> {}
 
 impl<T: TotalHash> Hash for TotalOrdWrap<T> {
-    #[inline]
+    #[inline(always)]
     fn hash<H: Hasher>(&self, state: &mut H) {
         self.0.tot_hash(state);
     }
@@ -158,33 +158,33 @@ impl<T: IsNull> IsNull for TotalOrdWrap<T> {
     const HAS_NULLS: bool = T::HAS_NULLS;
     type Inner = T::Inner;
 
-    #[inline]
+    #[inline(always)]
     fn is_null(&self) -> bool {
         self.0.is_null()
     }
 
-    #[inline]
+    #[inline(always)]
     fn unwrap_inner(self) -> Self::Inner {
         self.0.unwrap_inner()
     }
 }
 
 impl DirtyHash for f32 {
-    #[inline]
+    #[inline(always)]
     fn dirty_hash(&self) -> u64 {
         canonical_f32(*self).to_bits().dirty_hash()
     }
 }
 
 impl DirtyHash for f64 {
-    #[inline]
+    #[inline(always)]
     fn dirty_hash(&self) -> u64 {
         canonical_f64(*self).to_bits().dirty_hash()
     }
 }
 
 impl<T: DirtyHash> DirtyHash for TotalOrdWrap<T> {
-    #[inline]
+    #[inline(always)]
     fn dirty_hash(&self) -> u64 {
         self.0.dirty_hash()
     }
@@ -193,46 +193,46 @@ impl<T: DirtyHash> DirtyHash for TotalOrdWrap<T> {
 macro_rules! impl_trivial_total {
     ($T: ty) => {
         impl TotalEq for $T {
-            #[inline]
+            #[inline(always)]
             fn tot_eq(&self, other: &Self) -> bool {
                 self == other
             }
 
-            #[inline]
+            #[inline(always)]
             fn tot_ne(&self, other: &Self) -> bool {
                 self != other
             }
         }
 
         impl TotalOrd for $T {
-            #[inline]
+            #[inline(always)]
             fn tot_cmp(&self, other: &Self) -> Ordering {
                 self.cmp(other)
             }
 
-            #[inline]
+            #[inline(always)]
             fn tot_lt(&self, other: &Self) -> bool {
                 self < other
             }
 
-            #[inline]
+            #[inline(always)]
             fn tot_gt(&self, other: &Self) -> bool {
                 self > other
             }
 
-            #[inline]
+            #[inline(always)]
             fn tot_le(&self, other: &Self) -> bool {
                 self <= other
             }
 
-            #[inline]
+            #[inline(always)]
             fn tot_ge(&self, other: &Self) -> bool {
                 self >= other
             }
         }
 
         impl TotalHash for $T {
-            #[inline]
+            #[inline(always)]
             fn tot_hash<H>(&self, state: &mut H)
             where
                 H: Hasher,
@@ -277,7 +277,7 @@ macro_rules! impl_float_eq_ord {
         }
 
         impl TotalOrd for $T {
-            #[inline]
+            #[inline(always)]
             fn tot_cmp(&self, other: &Self) -> Ordering {
                 if self.tot_lt(other) {
                     Ordering::Less
@@ -288,22 +288,22 @@ macro_rules! impl_float_eq_ord {
                 }
             }
 
-            #[inline]
+            #[inline(always)]
             fn tot_lt(&self, other: &Self) -> bool {
                 !self.tot_ge(other)
             }
 
-            #[inline]
+            #[inline(always)]
             fn tot_gt(&self, other: &Self) -> bool {
                 other.tot_lt(self)
             }
 
-            #[inline]
+            #[inline(always)]
             fn tot_le(&self, other: &Self) -> bool {
                 other.tot_ge(self)
             }
 
-            #[inline]
+            #[inline(always)]
             fn tot_ge(&self, other: &Self) -> bool {
                 // We consider all NaNs equal, and NaN is the largest possible
                 // value. Thus if self is NaN we always return true. Otherwise
@@ -320,7 +320,7 @@ impl_float_eq_ord!(f32);
 impl_float_eq_ord!(f64);
 
 impl TotalHash for f32 {
-    #[inline]
+    #[inline(always)]
     fn tot_hash<H>(&self, state: &mut H)
     where
         H: Hasher,
@@ -330,7 +330,7 @@ impl TotalHash for f32 {
 }
 
 impl TotalHash for f64 {
-    #[inline]
+    #[inline(always)]
     fn tot_hash<H>(&self, state: &mut H)
     where
         H: Hasher,
@@ -341,7 +341,7 @@ impl TotalHash for f64 {
 
 // Blanket implementations.
 impl<T: TotalEq> TotalEq for Option<T> {
-    #[inline]
+    #[inline(always)]
     fn tot_eq(&self, other: &Self) -> bool {
         match (self, other) {
             (None, None) => true,
@@ -350,7 +350,7 @@ impl<T: TotalEq> TotalEq for Option<T> {
         }
     }
 
-    #[inline]
+    #[inline(always)]
     fn tot_ne(&self, other: &Self) -> bool {
         match (self, other) {
             (None, None) => false,
@@ -361,7 +361,7 @@ impl<T: TotalEq> TotalEq for Option<T> {
 }
 
 impl<T: TotalOrd> TotalOrd for Option<T> {
-    #[inline]
+    #[inline(always)]
     fn tot_cmp(&self, other: &Self) -> Ordering {
         match (self, other) {
             (None, None) => Ordering::Equal,
@@ -371,7 +371,7 @@ impl<T: TotalOrd> TotalOrd for Option<T> {
         }
     }
 
-    #[inline]
+    #[inline(always)]
     fn tot_lt(&self, other: &Self) -> bool {
         match (self, other) {
             (None, Some(_)) => true,
@@ -380,12 +380,12 @@ impl<T: TotalOrd> TotalOrd for Option<T> {
         }
     }
 
-    #[inline]
+    #[inline(always)]
     fn tot_gt(&self, other: &Self) -> bool {
         other.tot_lt(self)
     }
 
-    #[inline]
+    #[inline(always)]
     fn tot_le(&self, other: &Self) -> bool {
         match (self, other) {
             (Some(_), None) => false,
@@ -394,13 +394,14 @@ impl<T: TotalOrd> TotalOrd for Option<T> {
         }
     }
 
-    #[inline]
+    #[inline(always)]
     fn tot_ge(&self, other: &Self) -> bool {
         other.tot_le(self)
     }
 }
 
 impl<T: TotalHash> TotalHash for Option<T> {
+    #[inline]
     fn tot_hash<H>(&self, state: &mut H)
     where
         H: Hasher,
@@ -413,19 +414,19 @@ impl<T: TotalHash> TotalHash for Option<T> {
 }
 
 impl<T: TotalEq + ?Sized> TotalEq for &T {
-    #[inline]
+    #[inline(always)]
     fn tot_eq(&self, other: &Self) -> bool {
         (*self).tot_eq(*other)
     }
 
-    #[inline]
+    #[inline(always)]
     fn tot_ne(&self, other: &Self) -> bool {
         (*self).tot_ne(*other)
     }
 }
 
 impl<T: TotalHash + ?Sized> TotalHash for &T {
-    #[inline]
+    #[inline(always)]
     fn tot_hash<H>(&self, state: &mut H)
     where
         H: Hasher,
@@ -435,12 +436,14 @@ impl<T: TotalHash + ?Sized> TotalHash for &T {
 }
 
 impl<T: TotalEq, U: TotalEq> TotalEq for (T, U) {
+    #[inline]
     fn tot_eq(&self, other: &Self) -> bool {
         self.0.tot_eq(&other.0) && self.1.tot_eq(&other.1)
     }
 }
 
 impl<T: TotalOrd, U: TotalOrd> TotalOrd for (T, U) {
+    #[inline]
     fn tot_cmp(&self, other: &Self) -> Ordering {
         self.0
             .tot_cmp(&other.0)
@@ -449,7 +452,7 @@ impl<T: TotalOrd, U: TotalOrd> TotalOrd for (T, U) {
 }
 
 impl<'a> TotalHash for BytesHash<'a> {
-    #[inline]
+    #[inline(always)]
     fn tot_hash<H>(&self, state: &mut H)
     where
         H: Hasher,
@@ -459,7 +462,7 @@ impl<'a> TotalHash for BytesHash<'a> {
 }
 
 impl<'a> TotalEq for BytesHash<'a> {
-    #[inline]
+    #[inline(always)]
     fn tot_eq(&self, other: &Self) -> bool {
         self == other
     }