Support sort for Decimal data type

[yjshen]

Which issue does this PR close?

Closes #1137 .

Rationale for this change

We already have sort_decimal but fail to sort by decimal columns in DataFusion.

What changes are included in this PR?

Support decimal type for build_compare.

Are there any user-facing changes?

No.

[liukun4515]

LGTM

[liukun4515]

and link the pr to the issue #1137

[liukun4515]

@yjshen Can you provide sql in the datafusion side?
I find the sorting of decimal works on my side.

❯ \d food
+---------------+--------------+------------+-------------+-----------------+-------------+
| table_catalog | table_schema | table_name | column_name | data_type       | is_nullable |
+---------------+--------------+------------+-------------+-----------------+-------------+
| datafusion    | public       | food       | a           | Decimal(10, 5)  | NO          |
| datafusion    | public       | food       | b           | Decimal(20, 15) | NO          |
| datafusion    | public       | food       | c           | Boolean         | NO          |
+---------------+--------------+------------+-------------+-----------------+-------------+

❯ select * from food;
+---------+-------------------+-------+
| a       | b                 | c     |
+---------+-------------------+-------+
| 0.00001 | 0.000000000001000 | true  |
| 0.00002 | 0.000000000002000 | false |
| 0.00002 | 0.000000000002000 | false |
| 0.00003 | 0.000000000003000 | true  |
| 0.00003 | 0.000000000003000 | true  |
| 0.00003 | 0.000000000003000 | true  |
| 0.00004 | 0.000000000004000 | false |
| 0.00004 | 0.000000000004000 | false |
| 0.00004 | 0.000000000004000 | false |
| 0.00004 | 0.000000000004000 | false |
| 0.00005 | 0.000000000005000 | true  |
| 0.00005 | 0.000000000005000 | true  |
| 0.00005 | 0.000000000005000 | true  |
| 0.00005 | 0.000000000005000 | true  |
| 0.00005 | 0.000000000005000 | true  |
+---------+-------------------+-------+

sort data by desc.

❯ select * from food order by a desc;
+---------+-------------------+-------+
| a       | b                 | c     |
+---------+-------------------+-------+
| 0.00005 | 0.000000000005000 | true  |
| 0.00005 | 0.000000000005000 | true  |
| 0.00005 | 0.000000000005000 | true  |
| 0.00005 | 0.000000000005000 | true  |
| 0.00005 | 0.000000000005000 | true  |
| 0.00004 | 0.000000000004000 | false |
| 0.00004 | 0.000000000004000 | false |
| 0.00004 | 0.000000000004000 | false |
| 0.00004 | 0.000000000004000 | false |
| 0.00003 | 0.000000000003000 | true  |
| 0.00003 | 0.000000000003000 | true  |
| 0.00003 | 0.000000000003000 | true  |
| 0.00002 | 0.000000000002000 | false |
| 0.00002 | 0.000000000002000 | false |
| 0.00001 | 0.000000000001000 | true  |
+---------+-------------------+-------+

[liukun4515]

The test data is from aggregate_simple.csv in the datafusion.

[yjshen]

It's been used in two places in DataFusion:
InMemorySorter when we have combined all buffered batches into one single batch and call arrow::compute::lexsort_to_indices with more than one sort column. https://github.com/apache/arrow-datafusion/blob/master/datafusion/src/physical_plan/sorts/sort.rs#L542

SortPreservingMerge while we are combing several partial orders into one final order, and during constructing SortKeyCursor https://github.com/apache/arrow-datafusion/blob/master/datafusion/src/physical_plan/sorts/mod.rs#L188

[yjshen]

#[tokio::test]
async fn csv_query_with_decimal_by_sql() -> Result<()> {
    let mut ctx = SessionContext::new();
    register_simple_aggregate_csv_with_decimal_by_sql(&mut ctx).await;
    let sql = "SELECT * from aggregate_simple order by c1, c2";
    let actual = execute_to_batches(&ctx, sql).await;
...
}

thread 'sql::select::csv_query_with_decimal_by_sql' panicked at 'Executing physical plan for 'SELECT * from aggregate_simple order by c1, c2': SortExec
{ input: CoalescePartitionsExec { input: ProjectionExec { expr: [(Column { name: "c1", index: 0 }, "c1"), (Column { name:  ............ sicalSortExpr { expr: Column { name: "c2", index: 1 }, options: SortOptions { descending: false, nulls_first: false } }], metrics_set: CompositeMetricsSet 
{ mid: ExecutionPlanMetricsSet { inner: Mutex { data: MetricsSet { metrics: [] } } }, final_: ExecutionPlanMetricsSet { inner: Mutex { data: MetricsSet { metrics: [] } } } }, preserve_partitioning: false }: 
ArrowError(InvalidArgumentError("The data type type Decimal(10, 6) has no natural order"))', datafusion/tests/sql/mod.rs:547:49

The error message is too long, so I cut the inner info

[alamb]

What if the precision and scale of left and right are not the same? In that case I don't think calling cmp on the value is correct

If precision and scale are always the same, maybe we can add an assert_eq! to make that clear

[yjshen]

I think it's validated in the beginning of this function

pub fn build_compare(left: &dyn Array, right: &dyn Array) -> Result<DynComparator> {
    use DataType::*;
    use IntervalUnit::*;
    use TimeUnit::*;
    Ok(match (left.data_type(), right.data_type()) {
        (a, b) if a != b => {
            return Err(ArrowError::InvalidArgumentError(
                "Can't compare arrays of different types".to_string(),
            ));
        }

[yjshen]

Should I assert it again here?