diff --git a/crates/proof-of-sql/src/base/database/group_by_util.rs b/crates/proof-of-sql/src/base/database/group_by_util.rs
index 920337c23..17098bf08 100644
--- a/crates/proof-of-sql/src/base/database/group_by_util.rs
+++ b/crates/proof-of-sql/src/base/database/group_by_util.rs
@@ -1,9 +1,11 @@
 //! Contains the utility functions for the `GroupByExec` node.
 
 use crate::base::{
-    database::{filter_util::filter_column_by_index, Column, OwnedColumn},
+    database::{
+        filter_util::filter_column_by_index, order_by_util::compare_indexes_by_columns, Column,
+    },
     if_rayon,
-    scalar::{Scalar, ScalarExt},
+    scalar::Scalar,
 };
 use alloc::vec::Vec;
 use bumpalo::Bump;
@@ -355,53 +357,3 @@ where
             .min_by(super::super::scalar::ScalarExt::signed_cmp)
     }))
 }
-
-/// Compares the tuples `(group_by[0][i], group_by[1][i], ...)` and
-/// `(group_by[0][j], group_by[1][j], ...)` in lexicographic order.
-pub(crate) fn compare_indexes_by_columns<S: Scalar>(
-    group_by: &[Column<S>],
-    i: usize,
-    j: usize,
-) -> Ordering {
-    group_by
-        .iter()
-        .map(|col| match col {
-            Column::Boolean(col) => col[i].cmp(&col[j]),
-            Column::TinyInt(col) => col[i].cmp(&col[j]),
-            Column::SmallInt(col) => col[i].cmp(&col[j]),
-            Column::Int(col) => col[i].cmp(&col[j]),
-            Column::BigInt(col) | Column::TimestampTZ(_, _, col) => col[i].cmp(&col[j]),
-            Column::Int128(col) => col[i].cmp(&col[j]),
-            Column::Decimal75(_, _, col) => col[i].signed_cmp(&col[j]),
-            Column::Scalar(col) => col[i].cmp(&col[j]),
-            Column::VarChar((col, _)) => col[i].cmp(col[j]),
-        })
-        .find(|&ord| ord != Ordering::Equal)
-        .unwrap_or(Ordering::Equal)
-}
-
-/// Compares the tuples `(group_by[0][i], group_by[1][i], ...)` and
-/// `(group_by[0][j], group_by[1][j], ...)` in lexicographic order.
-///
-/// Identical in functionality to [`compare_indexes_by_columns`]
-pub(crate) fn compare_indexes_by_owned_columns<S: Scalar>(
-    group_by: &[&OwnedColumn<S>],
-    i: usize,
-    j: usize,
-) -> Ordering {
-    group_by
-        .iter()
-        .map(|col| match col {
-            OwnedColumn::Boolean(col) => col[i].cmp(&col[j]),
-            OwnedColumn::TinyInt(col) => col[i].cmp(&col[j]),
-            OwnedColumn::SmallInt(col) => col[i].cmp(&col[j]),
-            OwnedColumn::Int(col) => col[i].cmp(&col[j]),
-            OwnedColumn::BigInt(col) | OwnedColumn::TimestampTZ(_, _, col) => col[i].cmp(&col[j]),
-            OwnedColumn::Int128(col) => col[i].cmp(&col[j]),
-            OwnedColumn::Decimal75(_, _, col) => col[i].signed_cmp(&col[j]),
-            OwnedColumn::Scalar(col) => col[i].cmp(&col[j]),
-            OwnedColumn::VarChar(col) => col[i].cmp(&col[j]),
-        })
-        .find(|&ord| ord != Ordering::Equal)
-        .unwrap_or(Ordering::Equal)
-}
diff --git a/crates/proof-of-sql/src/base/database/group_by_util_test.rs b/crates/proof-of-sql/src/base/database/group_by_util_test.rs
index 02c58e422..217203c2c 100644
--- a/crates/proof-of-sql/src/base/database/group_by_util_test.rs
+++ b/crates/proof-of-sql/src/base/database/group_by_util_test.rs
@@ -1,12 +1,11 @@
 use crate::{
     base::{
-        database::{group_by_util::*, Column, OwnedColumn},
+        database::{group_by_util::*, Column},
         scalar::test_scalar::TestScalar,
     },
     proof_primitive::dory::DoryScalar,
 };
 use bumpalo::Bump;
-use core::cmp::Ordering;
 
 #[test]
 fn we_can_aggregate_empty_columns() {
@@ -217,184 +216,6 @@ fn we_can_aggregate_columns() {
     assert_eq!(aggregate_result.min_columns, expected_min_result);
 }
 
-#[test]
-fn we_can_compare_indexes_by_columns_with_no_columns() {
-    let columns: &[Column<TestScalar>; 0] = &[];
-    assert_eq!(compare_indexes_by_columns(columns, 0, 1), Ordering::Equal);
-    assert_eq!(compare_indexes_by_columns(columns, 1, 2), Ordering::Equal);
-    assert_eq!(compare_indexes_by_columns(columns, 3, 2), Ordering::Equal);
-}
-
-#[test]
-fn we_can_compare_indexes_by_columns_for_bigint_columns() {
-    let slice_a = &[55, 44, 66, 66, 66, 77, 66, 66, 66, 66];
-    let slice_b = &[22, 44, 11, 44, 33, 22, 22, 11, 22, 22];
-    let slice_c = &[11, 55, 11, 44, 77, 11, 22, 55, 11, 22];
-    let column_a = Column::BigInt::<DoryScalar>(slice_a);
-    let column_b = Column::BigInt::<DoryScalar>(slice_b);
-    let column_c = Column::BigInt::<DoryScalar>(slice_c);
-
-    let columns = &[column_a];
-    assert_eq!(compare_indexes_by_columns(columns, 0, 1), Ordering::Greater);
-    assert_eq!(compare_indexes_by_columns(columns, 1, 2), Ordering::Less);
-    assert_eq!(compare_indexes_by_columns(columns, 2, 3), Ordering::Equal);
-    assert_eq!(compare_indexes_by_columns(columns, 2, 1), Ordering::Greater);
-    assert_eq!(compare_indexes_by_columns(columns, 1, 0), Ordering::Less);
-    let columns = &[column_a, column_b];
-    assert_eq!(compare_indexes_by_columns(columns, 0, 1), Ordering::Greater);
-    assert_eq!(compare_indexes_by_columns(columns, 1, 2), Ordering::Less);
-    assert_eq!(compare_indexes_by_columns(columns, 2, 3), Ordering::Less);
-    assert_eq!(compare_indexes_by_columns(columns, 3, 4), Ordering::Greater);
-    assert_eq!(compare_indexes_by_columns(columns, 2, 7), Ordering::Equal);
-    let columns = &[column_a, column_b, column_c];
-    assert_eq!(compare_indexes_by_columns(columns, 0, 1), Ordering::Greater);
-    assert_eq!(compare_indexes_by_columns(columns, 1, 2), Ordering::Less);
-    assert_eq!(compare_indexes_by_columns(columns, 2, 3), Ordering::Less);
-    assert_eq!(compare_indexes_by_columns(columns, 3, 4), Ordering::Greater);
-    assert_eq!(compare_indexes_by_columns(columns, 2, 7), Ordering::Less);
-    assert_eq!(compare_indexes_by_columns(columns, 6, 9), Ordering::Equal);
-}
-#[test]
-fn we_can_compare_indexes_by_columns_for_mixed_columns() {
-    let slice_a = &["55", "44", "66", "66", "66", "77", "66", "66", "66", "66"];
-    let slice_b = &[22, 44, 11, 44, 33, 22, 22, 11, 22, 22];
-    let slice_c = &[11, 55, 11, 44, 77, 11, 22, 55, 11, 22];
-    let scals_a: Vec<TestScalar> = slice_a.iter().map(core::convert::Into::into).collect();
-    let column_a = Column::VarChar((slice_a, &scals_a));
-    let column_b = Column::Int128(slice_b);
-    let column_c = Column::BigInt(slice_c);
-
-    let columns = &[column_a];
-    assert_eq!(compare_indexes_by_columns(columns, 0, 1), Ordering::Greater);
-    assert_eq!(compare_indexes_by_columns(columns, 1, 2), Ordering::Less);
-    assert_eq!(compare_indexes_by_columns(columns, 2, 3), Ordering::Equal);
-    assert_eq!(compare_indexes_by_columns(columns, 2, 1), Ordering::Greater);
-    assert_eq!(compare_indexes_by_columns(columns, 1, 0), Ordering::Less);
-    let columns = &[column_a, column_b];
-    assert_eq!(compare_indexes_by_columns(columns, 0, 1), Ordering::Greater);
-    assert_eq!(compare_indexes_by_columns(columns, 1, 2), Ordering::Less);
-    assert_eq!(compare_indexes_by_columns(columns, 2, 3), Ordering::Less);
-    assert_eq!(compare_indexes_by_columns(columns, 3, 4), Ordering::Greater);
-    assert_eq!(compare_indexes_by_columns(columns, 2, 7), Ordering::Equal);
-    let columns = &[column_a, column_b, column_c];
-    assert_eq!(compare_indexes_by_columns(columns, 0, 1), Ordering::Greater);
-    assert_eq!(compare_indexes_by_columns(columns, 1, 2), Ordering::Less);
-    assert_eq!(compare_indexes_by_columns(columns, 2, 3), Ordering::Less);
-    assert_eq!(compare_indexes_by_columns(columns, 3, 4), Ordering::Greater);
-    assert_eq!(compare_indexes_by_columns(columns, 2, 7), Ordering::Less);
-    assert_eq!(compare_indexes_by_columns(columns, 6, 9), Ordering::Equal);
-}
-#[test]
-fn we_can_compare_indexes_by_owned_columns_for_mixed_columns() {
-    let slice_a = ["55", "44", "66", "66", "66", "77", "66", "66", "66", "66"]
-        .into_iter()
-        .map(Into::into)
-        .collect();
-    let slice_b = vec![22, 44, 11, 44, 33, 22, 22, 11, 22, 22];
-    let slice_c = vec![11, 55, 11, 44, 77, 11, 22, 55, 11, 22];
-    let column_a = OwnedColumn::<DoryScalar>::VarChar(slice_a);
-    let column_b = OwnedColumn::Int128(slice_b);
-    let column_c = OwnedColumn::BigInt(slice_c);
-
-    let columns = &[&column_a];
-    assert_eq!(
-        compare_indexes_by_owned_columns(columns, 0, 1),
-        Ordering::Greater
-    );
-    assert_eq!(
-        compare_indexes_by_owned_columns(columns, 1, 2),
-        Ordering::Less
-    );
-    assert_eq!(
-        compare_indexes_by_owned_columns(columns, 2, 3),
-        Ordering::Equal
-    );
-    assert_eq!(
-        compare_indexes_by_owned_columns(columns, 2, 1),
-        Ordering::Greater
-    );
-    assert_eq!(
-        compare_indexes_by_owned_columns(columns, 1, 0),
-        Ordering::Less
-    );
-    let columns = &[&column_a, &column_b];
-    assert_eq!(
-        compare_indexes_by_owned_columns(columns, 0, 1),
-        Ordering::Greater
-    );
-    assert_eq!(
-        compare_indexes_by_owned_columns(columns, 1, 2),
-        Ordering::Less
-    );
-    assert_eq!(
-        compare_indexes_by_owned_columns(columns, 2, 3),
-        Ordering::Less
-    );
-    assert_eq!(
-        compare_indexes_by_owned_columns(columns, 3, 4),
-        Ordering::Greater
-    );
-    assert_eq!(
-        compare_indexes_by_owned_columns(columns, 2, 7),
-        Ordering::Equal
-    );
-    let columns = &[&column_a, &column_b, &column_c];
-    assert_eq!(
-        compare_indexes_by_owned_columns(columns, 0, 1),
-        Ordering::Greater
-    );
-    assert_eq!(
-        compare_indexes_by_owned_columns(columns, 1, 2),
-        Ordering::Less
-    );
-    assert_eq!(
-        compare_indexes_by_owned_columns(columns, 2, 3),
-        Ordering::Less
-    );
-    assert_eq!(
-        compare_indexes_by_owned_columns(columns, 3, 4),
-        Ordering::Greater
-    );
-    assert_eq!(
-        compare_indexes_by_owned_columns(columns, 2, 7),
-        Ordering::Less
-    );
-    assert_eq!(
-        compare_indexes_by_owned_columns(columns, 6, 9),
-        Ordering::Equal
-    );
-}
-#[test]
-fn we_can_compare_indexes_by_columns_for_scalar_columns() {
-    let slice_a = &[55, 44, 66, 66, 66, 77, 66, 66, 66, 66];
-    let slice_b = &[22, 44, 11, 44, 33, 22, 22, 11, 22, 22];
-    let slice_c = &[11, 55, 11, 44, 77, 11, 22, 55, 11, 22];
-    let scals_a: Vec<TestScalar> = slice_a.iter().map(core::convert::Into::into).collect();
-    let column_a = Column::Scalar(&scals_a);
-    let column_b = Column::Int128(slice_b);
-    let column_c = Column::BigInt(slice_c);
-
-    let columns = &[column_a];
-    assert_eq!(compare_indexes_by_columns(columns, 0, 1), Ordering::Greater);
-    assert_eq!(compare_indexes_by_columns(columns, 1, 2), Ordering::Less);
-    assert_eq!(compare_indexes_by_columns(columns, 2, 3), Ordering::Equal);
-    assert_eq!(compare_indexes_by_columns(columns, 2, 1), Ordering::Greater);
-    assert_eq!(compare_indexes_by_columns(columns, 1, 0), Ordering::Less);
-    let columns = &[column_a, column_b];
-    assert_eq!(compare_indexes_by_columns(columns, 0, 1), Ordering::Greater);
-    assert_eq!(compare_indexes_by_columns(columns, 1, 2), Ordering::Less);
-    assert_eq!(compare_indexes_by_columns(columns, 2, 3), Ordering::Less);
-    assert_eq!(compare_indexes_by_columns(columns, 3, 4), Ordering::Greater);
-    assert_eq!(compare_indexes_by_columns(columns, 2, 7), Ordering::Equal);
-    let columns = &[column_a, column_b, column_c];
-    assert_eq!(compare_indexes_by_columns(columns, 0, 1), Ordering::Greater);
-    assert_eq!(compare_indexes_by_columns(columns, 1, 2), Ordering::Less);
-    assert_eq!(compare_indexes_by_columns(columns, 2, 3), Ordering::Less);
-    assert_eq!(compare_indexes_by_columns(columns, 3, 4), Ordering::Greater);
-    assert_eq!(compare_indexes_by_columns(columns, 2, 7), Ordering::Less);
-    assert_eq!(compare_indexes_by_columns(columns, 6, 9), Ordering::Equal);
-}
-
 // SUM slices
 #[test]
 fn we_can_sum_aggregate_slice_by_counts_for_empty_slice() {
diff --git a/crates/proof-of-sql/src/base/database/mod.rs b/crates/proof-of-sql/src/base/database/mod.rs
index 2f1fa847f..dd3730255 100644
--- a/crates/proof-of-sql/src/base/database/mod.rs
+++ b/crates/proof-of-sql/src/base/database/mod.rs
@@ -55,7 +55,6 @@ pub use table_ref::TableRef;
 pub mod arrow_schema_utility;
 
 mod owned_column;
-pub(crate) use owned_column::compare_indexes_by_owned_columns_with_direction;
 pub use owned_column::OwnedColumn;
 
 mod owned_column_error;
@@ -115,3 +114,7 @@ mod group_by_util_test;
 
 #[allow(dead_code)]
 pub(crate) mod union_util;
+
+pub(crate) mod order_by_util;
+#[cfg(test)]
+mod order_by_util_test;
diff --git a/crates/proof-of-sql/src/base/database/order_by_util.rs b/crates/proof-of-sql/src/base/database/order_by_util.rs
new file mode 100644
index 000000000..e2589d1db
--- /dev/null
+++ b/crates/proof-of-sql/src/base/database/order_by_util.rs
@@ -0,0 +1,81 @@
+//! Contains the utility functions for ordering.
+use crate::base::{
+    database::{Column, OwnedColumn},
+    scalar::{Scalar, ScalarExt},
+};
+use alloc::vec::Vec;
+use core::cmp::Ordering;
+use proof_of_sql_parser::intermediate_ast::OrderByDirection;
+
+/// Compares the tuples `(order_by[0][i], order_by[1][i], ...)` and
+/// `(order_by[0][j], order_by[1][j], ...)` in lexicographic order.
+pub(crate) fn compare_indexes_by_columns<S: Scalar>(
+    order_by: &[Column<S>],
+    i: usize,
+    j: usize,
+) -> Ordering {
+    order_by
+        .iter()
+        .map(|col| match col {
+            Column::Boolean(col) => col[i].cmp(&col[j]),
+            Column::TinyInt(col) => col[i].cmp(&col[j]),
+            Column::SmallInt(col) => col[i].cmp(&col[j]),
+            Column::Int(col) => col[i].cmp(&col[j]),
+            Column::BigInt(col) | Column::TimestampTZ(_, _, col) => col[i].cmp(&col[j]),
+            Column::Int128(col) => col[i].cmp(&col[j]),
+            Column::Decimal75(_, _, col) => col[i].signed_cmp(&col[j]),
+            Column::Scalar(col) => col[i].cmp(&col[j]),
+            Column::VarChar((col, _)) => col[i].cmp(col[j]),
+        })
+        .find(|&ord| ord != Ordering::Equal)
+        .unwrap_or(Ordering::Equal)
+}
+
+/// Compares the tuples `(order_by[0][i], order_by[1][i], ...)` and
+/// `(order_by[0][j], order_by[1][j], ...)` in lexicographic order.
+///
+/// Identical in functionality to [`compare_indexes_by_columns`]
+pub(crate) fn compare_indexes_by_owned_columns<S: Scalar>(
+    order_by: &[&OwnedColumn<S>],
+    i: usize,
+    j: usize,
+) -> Ordering {
+    let order_by_pairs = order_by
+        .iter()
+        .map(|&col| (col.clone(), OrderByDirection::Asc))
+        .collect::<Vec<_>>();
+    compare_indexes_by_owned_columns_with_direction(&order_by_pairs, i, j)
+}
+
+/// Compares the tuples `(left[0][i], left[1][i], ...)` and
+/// `(right[0][j], right[1][j], ...)` in lexicographic order.
+/// Note that direction flips the ordering.
+pub(crate) fn compare_indexes_by_owned_columns_with_direction<S: Scalar>(
+    order_by_pairs: &[(OwnedColumn<S>, OrderByDirection)],
+    i: usize,
+    j: usize,
+) -> Ordering {
+    order_by_pairs
+        .iter()
+        .map(|(col, direction)| {
+            let ordering = match col {
+                OwnedColumn::Boolean(col) => col[i].cmp(&col[j]),
+                OwnedColumn::TinyInt(col) => col[i].cmp(&col[j]),
+                OwnedColumn::SmallInt(col) => col[i].cmp(&col[j]),
+                OwnedColumn::Int(col) => col[i].cmp(&col[j]),
+                OwnedColumn::BigInt(col) | OwnedColumn::TimestampTZ(_, _, col) => {
+                    col[i].cmp(&col[j])
+                }
+                OwnedColumn::Int128(col) => col[i].cmp(&col[j]),
+                OwnedColumn::Decimal75(_, _, col) => col[i].signed_cmp(&col[j]),
+                OwnedColumn::Scalar(col) => col[i].cmp(&col[j]),
+                OwnedColumn::VarChar(col) => col[i].cmp(&col[j]),
+            };
+            match direction {
+                OrderByDirection::Asc => ordering,
+                OrderByDirection::Desc => ordering.reverse(),
+            }
+        })
+        .find(|&ord| ord != Ordering::Equal)
+        .unwrap_or(Ordering::Equal)
+}
diff --git a/crates/proof-of-sql/src/base/database/order_by_util_test.rs b/crates/proof-of-sql/src/base/database/order_by_util_test.rs
new file mode 100644
index 000000000..644706bad
--- /dev/null
+++ b/crates/proof-of-sql/src/base/database/order_by_util_test.rs
@@ -0,0 +1,235 @@
+use crate::{
+    base::{
+        database::{order_by_util::*, Column, OwnedColumn},
+        math::decimal::Precision,
+        scalar::test_scalar::TestScalar,
+    },
+    proof_primitive::dory::DoryScalar,
+};
+use core::cmp::Ordering;
+use proof_of_sql_parser::intermediate_ast::OrderByDirection;
+
+#[test]
+fn we_can_compare_indexes_by_columns_with_no_columns() {
+    let columns: &[Column<TestScalar>; 0] = &[];
+    assert_eq!(compare_indexes_by_columns(columns, 0, 1), Ordering::Equal);
+    assert_eq!(compare_indexes_by_columns(columns, 1, 2), Ordering::Equal);
+    assert_eq!(compare_indexes_by_columns(columns, 3, 2), Ordering::Equal);
+}
+
+#[test]
+fn we_can_compare_indexes_by_columns_for_bigint_columns() {
+    let slice_a = &[55, 44, 66, 66, 66, 77, 66, 66, 66, 66];
+    let slice_b = &[22, 44, 11, 44, 33, 22, 22, 11, 22, 22];
+    let slice_c = &[11, 55, 11, 44, 77, 11, 22, 55, 11, 22];
+    let column_a = Column::BigInt::<DoryScalar>(slice_a);
+    let column_b = Column::BigInt::<DoryScalar>(slice_b);
+    let column_c = Column::BigInt::<DoryScalar>(slice_c);
+
+    let columns = &[column_a];
+    assert_eq!(compare_indexes_by_columns(columns, 0, 1), Ordering::Greater);
+    assert_eq!(compare_indexes_by_columns(columns, 1, 2), Ordering::Less);
+    assert_eq!(compare_indexes_by_columns(columns, 2, 3), Ordering::Equal);
+    assert_eq!(compare_indexes_by_columns(columns, 2, 1), Ordering::Greater);
+    assert_eq!(compare_indexes_by_columns(columns, 1, 0), Ordering::Less);
+    let columns = &[column_a, column_b];
+    assert_eq!(compare_indexes_by_columns(columns, 0, 1), Ordering::Greater);
+    assert_eq!(compare_indexes_by_columns(columns, 1, 2), Ordering::Less);
+    assert_eq!(compare_indexes_by_columns(columns, 2, 3), Ordering::Less);
+    assert_eq!(compare_indexes_by_columns(columns, 3, 4), Ordering::Greater);
+    assert_eq!(compare_indexes_by_columns(columns, 2, 7), Ordering::Equal);
+    let columns = &[column_a, column_b, column_c];
+    assert_eq!(compare_indexes_by_columns(columns, 0, 1), Ordering::Greater);
+    assert_eq!(compare_indexes_by_columns(columns, 1, 2), Ordering::Less);
+    assert_eq!(compare_indexes_by_columns(columns, 2, 3), Ordering::Less);
+    assert_eq!(compare_indexes_by_columns(columns, 3, 4), Ordering::Greater);
+    assert_eq!(compare_indexes_by_columns(columns, 2, 7), Ordering::Less);
+    assert_eq!(compare_indexes_by_columns(columns, 6, 9), Ordering::Equal);
+}
+
+#[test]
+fn we_can_compare_indexes_by_columns_for_mixed_columns() {
+    let slice_a = &["55", "44", "66", "66", "66", "77", "66", "66", "66", "66"];
+    let slice_b = &[22, 44, 11, 44, 33, 22, 22, 11, 22, 22];
+    let slice_c = &[11, 55, 11, 44, 77, 11, 22, 55, 11, 22];
+    let scals_a: Vec<TestScalar> = slice_a.iter().map(core::convert::Into::into).collect();
+    let column_a = Column::VarChar((slice_a, &scals_a));
+    let column_b = Column::Int128(slice_b);
+    let column_c = Column::BigInt(slice_c);
+
+    let columns = &[column_a];
+    assert_eq!(compare_indexes_by_columns(columns, 0, 1), Ordering::Greater);
+    assert_eq!(compare_indexes_by_columns(columns, 1, 2), Ordering::Less);
+    assert_eq!(compare_indexes_by_columns(columns, 2, 3), Ordering::Equal);
+    assert_eq!(compare_indexes_by_columns(columns, 2, 1), Ordering::Greater);
+    assert_eq!(compare_indexes_by_columns(columns, 1, 0), Ordering::Less);
+    let columns = &[column_a, column_b];
+    assert_eq!(compare_indexes_by_columns(columns, 0, 1), Ordering::Greater);
+    assert_eq!(compare_indexes_by_columns(columns, 1, 2), Ordering::Less);
+    assert_eq!(compare_indexes_by_columns(columns, 2, 3), Ordering::Less);
+    assert_eq!(compare_indexes_by_columns(columns, 3, 4), Ordering::Greater);
+    assert_eq!(compare_indexes_by_columns(columns, 2, 7), Ordering::Equal);
+    let columns = &[column_a, column_b, column_c];
+    assert_eq!(compare_indexes_by_columns(columns, 0, 1), Ordering::Greater);
+    assert_eq!(compare_indexes_by_columns(columns, 1, 2), Ordering::Less);
+    assert_eq!(compare_indexes_by_columns(columns, 2, 3), Ordering::Less);
+    assert_eq!(compare_indexes_by_columns(columns, 3, 4), Ordering::Greater);
+    assert_eq!(compare_indexes_by_columns(columns, 2, 7), Ordering::Less);
+    assert_eq!(compare_indexes_by_columns(columns, 6, 9), Ordering::Equal);
+}
+
+#[test]
+fn we_can_compare_indexes_by_owned_columns_for_mixed_columns() {
+    let slice_a = ["55", "44", "66", "66", "66", "77", "66", "66", "66", "66"]
+        .into_iter()
+        .map(Into::into)
+        .collect();
+    let slice_b = vec![22, 44, 11, 44, 33, 22, 22, 11, 22, 22];
+    let slice_c = vec![11, 55, 11, 44, 77, 11, 22, 55, 11, 22];
+    let column_a = OwnedColumn::<DoryScalar>::VarChar(slice_a);
+    let column_b = OwnedColumn::Int128(slice_b);
+    let column_c = OwnedColumn::BigInt(slice_c);
+
+    let columns = &[&column_a];
+    assert_eq!(
+        compare_indexes_by_owned_columns(columns, 0, 1),
+        Ordering::Greater
+    );
+    assert_eq!(
+        compare_indexes_by_owned_columns(columns, 1, 2),
+        Ordering::Less
+    );
+    assert_eq!(
+        compare_indexes_by_owned_columns(columns, 2, 3),
+        Ordering::Equal
+    );
+    assert_eq!(
+        compare_indexes_by_owned_columns(columns, 2, 1),
+        Ordering::Greater
+    );
+    assert_eq!(
+        compare_indexes_by_owned_columns(columns, 1, 0),
+        Ordering::Less
+    );
+    let columns = &[&column_a, &column_b];
+    assert_eq!(
+        compare_indexes_by_owned_columns(columns, 0, 1),
+        Ordering::Greater
+    );
+    assert_eq!(
+        compare_indexes_by_owned_columns(columns, 1, 2),
+        Ordering::Less
+    );
+    assert_eq!(
+        compare_indexes_by_owned_columns(columns, 2, 3),
+        Ordering::Less
+    );
+    assert_eq!(
+        compare_indexes_by_owned_columns(columns, 3, 4),
+        Ordering::Greater
+    );
+    assert_eq!(
+        compare_indexes_by_owned_columns(columns, 2, 7),
+        Ordering::Equal
+    );
+    let columns = &[&column_a, &column_b, &column_c];
+    assert_eq!(
+        compare_indexes_by_owned_columns(columns, 0, 1),
+        Ordering::Greater
+    );
+    assert_eq!(
+        compare_indexes_by_owned_columns(columns, 1, 2),
+        Ordering::Less
+    );
+    assert_eq!(
+        compare_indexes_by_owned_columns(columns, 2, 3),
+        Ordering::Less
+    );
+    assert_eq!(
+        compare_indexes_by_owned_columns(columns, 3, 4),
+        Ordering::Greater
+    );
+    assert_eq!(
+        compare_indexes_by_owned_columns(columns, 2, 7),
+        Ordering::Less
+    );
+    assert_eq!(
+        compare_indexes_by_owned_columns(columns, 6, 9),
+        Ordering::Equal
+    );
+}
+
+#[test]
+fn we_can_compare_indexes_by_columns_for_scalar_columns() {
+    let slice_a = &[55, 44, 66, 66, 66, 77, 66, 66, 66, 66];
+    let slice_b = &[22, 44, 11, 44, 33, 22, 22, 11, 22, 22];
+    let slice_c = &[11, 55, 11, 44, 77, 11, 22, 55, 11, 22];
+    let scals_a: Vec<TestScalar> = slice_a.iter().map(core::convert::Into::into).collect();
+    let column_a = Column::Scalar(&scals_a);
+    let column_b = Column::Int128(slice_b);
+    let column_c = Column::BigInt(slice_c);
+
+    let columns = &[column_a];
+    assert_eq!(compare_indexes_by_columns(columns, 0, 1), Ordering::Greater);
+    assert_eq!(compare_indexes_by_columns(columns, 1, 2), Ordering::Less);
+    assert_eq!(compare_indexes_by_columns(columns, 2, 3), Ordering::Equal);
+    assert_eq!(compare_indexes_by_columns(columns, 2, 1), Ordering::Greater);
+    assert_eq!(compare_indexes_by_columns(columns, 1, 0), Ordering::Less);
+    let columns = &[column_a, column_b];
+    assert_eq!(compare_indexes_by_columns(columns, 0, 1), Ordering::Greater);
+    assert_eq!(compare_indexes_by_columns(columns, 1, 2), Ordering::Less);
+    assert_eq!(compare_indexes_by_columns(columns, 2, 3), Ordering::Less);
+    assert_eq!(compare_indexes_by_columns(columns, 3, 4), Ordering::Greater);
+    assert_eq!(compare_indexes_by_columns(columns, 2, 7), Ordering::Equal);
+    let columns = &[column_a, column_b, column_c];
+    assert_eq!(compare_indexes_by_columns(columns, 0, 1), Ordering::Greater);
+    assert_eq!(compare_indexes_by_columns(columns, 1, 2), Ordering::Less);
+    assert_eq!(compare_indexes_by_columns(columns, 2, 3), Ordering::Less);
+    assert_eq!(compare_indexes_by_columns(columns, 3, 4), Ordering::Greater);
+    assert_eq!(compare_indexes_by_columns(columns, 2, 7), Ordering::Less);
+    assert_eq!(compare_indexes_by_columns(columns, 6, 9), Ordering::Equal);
+}
+
+#[test]
+fn we_can_compare_columns_with_direction() {
+    let col1: OwnedColumn<TestScalar> = OwnedColumn::SmallInt(vec![1, 1, 2, 1, 1]);
+    let col2: OwnedColumn<TestScalar> = OwnedColumn::VarChar(
+        ["b", "b", "a", "b", "a"]
+            .iter()
+            .map(ToString::to_string)
+            .collect(),
+    );
+    let col3: OwnedColumn<TestScalar> = OwnedColumn::Decimal75(
+        Precision::new(70).unwrap(),
+        20,
+        [-3, 2, 2, -3, 2]
+            .iter()
+            .map(|&i| TestScalar::from(i))
+            .collect(),
+    );
+    let order_by_pairs = vec![
+        (col1, OrderByDirection::Asc),
+        (col2, OrderByDirection::Desc),
+        (col3, OrderByDirection::Asc),
+    ];
+    // Equal on col1 and col2, less on col3
+    assert_eq!(
+        compare_indexes_by_owned_columns_with_direction(&order_by_pairs, 0, 1),
+        Ordering::Less
+    );
+    // Less on col1
+    assert_eq!(
+        compare_indexes_by_owned_columns_with_direction(&order_by_pairs, 0, 2),
+        Ordering::Less
+    );
+    // Equal on all 3 columns
+    assert_eq!(
+        compare_indexes_by_owned_columns_with_direction(&order_by_pairs, 0, 3),
+        Ordering::Equal
+    );
+    // Equal on col1, greater on col2 reversed
+    assert_eq!(
+        compare_indexes_by_owned_columns_with_direction(&order_by_pairs, 1, 4),
+        Ordering::Less
+    );
+}
diff --git a/crates/proof-of-sql/src/base/database/owned_column.rs b/crates/proof-of-sql/src/base/database/owned_column.rs
index 26a5eaffb..1c2247da8 100644
--- a/crates/proof-of-sql/src/base/database/owned_column.rs
+++ b/crates/proof-of-sql/src/base/database/owned_column.rs
@@ -15,11 +15,7 @@ use alloc::{
     string::{String, ToString},
     vec::Vec,
 };
-use core::cmp::Ordering;
-use proof_of_sql_parser::{
-    intermediate_ast::OrderByDirection,
-    posql_time::{PoSQLTimeUnit, PoSQLTimeZone},
-};
+use proof_of_sql_parser::posql_time::{PoSQLTimeUnit, PoSQLTimeZone};
 use serde::{Deserialize, Serialize};
 
 #[derive(Debug, PartialEq, Clone, Eq, Serialize, Deserialize)]
@@ -345,45 +341,12 @@ impl<'a, S: Scalar> From<&Column<'a, S>> for OwnedColumn<S> {
     }
 }
 
-/// Compares the tuples `(order_by_pairs[0][i], order_by_pairs[1][i], ...)` and
-/// `(order_by_pairs[0][j], order_by_pairs[1][j], ...)` in lexicographic order.
-/// Note that direction flips the ordering.
-pub(crate) fn compare_indexes_by_owned_columns_with_direction<S: Scalar>(
-    order_by_pairs: &[(OwnedColumn<S>, OrderByDirection)],
-    i: usize,
-    j: usize,
-) -> Ordering {
-    order_by_pairs
-        .iter()
-        .map(|(col, direction)| {
-            let ordering = match col {
-                OwnedColumn::Boolean(col) => col[i].cmp(&col[j]),
-                OwnedColumn::TinyInt(col) => col[i].cmp(&col[j]),
-                OwnedColumn::SmallInt(col) => col[i].cmp(&col[j]),
-                OwnedColumn::Int(col) => col[i].cmp(&col[j]),
-                OwnedColumn::BigInt(col) | OwnedColumn::TimestampTZ(_, _, col) => {
-                    col[i].cmp(&col[j])
-                }
-                OwnedColumn::Int128(col) => col[i].cmp(&col[j]),
-                OwnedColumn::Decimal75(_, _, col) | OwnedColumn::Scalar(col) => col[i].cmp(&col[j]),
-                OwnedColumn::VarChar(col) => col[i].cmp(&col[j]),
-            };
-            match direction {
-                OrderByDirection::Asc => ordering,
-                OrderByDirection::Desc => ordering.reverse(),
-            }
-        })
-        .find(|&ord| ord != Ordering::Equal)
-        .unwrap_or(Ordering::Equal)
-}
-
 #[cfg(test)]
 mod test {
     use super::*;
     use crate::base::{math::decimal::Precision, scalar::test_scalar::TestScalar};
     use alloc::vec;
     use bumpalo::Bump;
-    use proof_of_sql_parser::intermediate_ast::OrderByDirection;
 
     #[test]
     fn we_can_slice_a_column() {
@@ -401,50 +364,6 @@ mod test {
         );
     }
 
-    #[test]
-    fn we_can_compare_columns() {
-        let col1: OwnedColumn<TestScalar> = OwnedColumn::SmallInt(vec![1, 1, 2, 1, 1]);
-        let col2: OwnedColumn<TestScalar> = OwnedColumn::VarChar(
-            ["b", "b", "a", "b", "a"]
-                .iter()
-                .map(ToString::to_string)
-                .collect(),
-        );
-        let col3: OwnedColumn<TestScalar> = OwnedColumn::Decimal75(
-            Precision::new(70).unwrap(),
-            20,
-            [1, 2, 2, 1, 2]
-                .iter()
-                .map(|&i| TestScalar::from(i))
-                .collect(),
-        );
-        let order_by_pairs = vec![
-            (col1, OrderByDirection::Asc),
-            (col2, OrderByDirection::Desc),
-            (col3, OrderByDirection::Asc),
-        ];
-        // Equal on col1 and col2, less on col3
-        assert_eq!(
-            compare_indexes_by_owned_columns_with_direction(&order_by_pairs, 0, 1),
-            Ordering::Less
-        );
-        // Less on col1
-        assert_eq!(
-            compare_indexes_by_owned_columns_with_direction(&order_by_pairs, 0, 2),
-            Ordering::Less
-        );
-        // Equal on all 3 columns
-        assert_eq!(
-            compare_indexes_by_owned_columns_with_direction(&order_by_pairs, 0, 3),
-            Ordering::Equal
-        );
-        // Equal on col1, greater on col2 reversed
-        assert_eq!(
-            compare_indexes_by_owned_columns_with_direction(&order_by_pairs, 1, 4),
-            Ordering::Less
-        );
-    }
-
     #[test]
     fn we_can_convert_columns_to_owned_columns_round_trip() {
         let alloc = Bump::new();
diff --git a/crates/proof-of-sql/src/sql/postprocessing/order_by_postprocessing.rs b/crates/proof-of-sql/src/sql/postprocessing/order_by_postprocessing.rs
index dd5917d63..6bda616c3 100644
--- a/crates/proof-of-sql/src/sql/postprocessing/order_by_postprocessing.rs
+++ b/crates/proof-of-sql/src/sql/postprocessing/order_by_postprocessing.rs
@@ -1,6 +1,8 @@
 use super::{PostprocessingError, PostprocessingResult, PostprocessingStep};
 use crate::base::{
-    database::{compare_indexes_by_owned_columns_with_direction, OwnedColumn, OwnedTable},
+    database::{
+        order_by_util::compare_indexes_by_owned_columns_with_direction, OwnedColumn, OwnedTable,
+    },
     if_rayon,
     math::permutation::Permutation,
     scalar::Scalar,
diff --git a/crates/proof-of-sql/src/sql/proof_plans/group_by_exec.rs b/crates/proof-of-sql/src/sql/proof_plans/group_by_exec.rs
index f2299575c..b1e95b83e 100644
--- a/crates/proof-of-sql/src/sql/proof_plans/group_by_exec.rs
+++ b/crates/proof-of-sql/src/sql/proof_plans/group_by_exec.rs
@@ -2,9 +2,8 @@ use super::{fold_columns, fold_vals};
 use crate::{
     base::{
         database::{
-            group_by_util::{
-                aggregate_columns, compare_indexes_by_owned_columns, AggregatedColumns,
-            },
+            group_by_util::{aggregate_columns, AggregatedColumns},
+            order_by_util::compare_indexes_by_owned_columns,
             Column, ColumnField, ColumnRef, ColumnType, OwnedTable, Table, TableRef,
         },
         map::{IndexMap, IndexSet},