sum(distinct) support

[WinkerDu]

Which issue does this PR close?

Closes #2404 .

Rationale for this change

For now, DataFusion hasn't support sum(distinct) yet. Though optimizer SingleDistinctToGroupBy (#1315) supports single distinct usage, there are more SQL scenes have not been covered.
For example, runs the following unit test,

#[tokio::test]
async fn query_sum_distinct() -> Result<()> {
    let schema = Arc::new(Schema::new(vec![
        Field::new("c1", DataType::Int64, true),
        Field::new("c2", DataType::Int64, true),
    ]));

    let data = RecordBatch::try_new(
        schema.clone(),
        vec![
            Arc::new(Int64Array::from(vec![
                Some(0),
                Some(1),
                None,
                Some(3),
                Some(3),
            ])),
            Arc::new(Int64Array::from(vec![
                None,
                Some(1),
                Some(1),
                Some(2),
                Some(2),
            ])),
        ],
    )?;

    let table = MemTable::try_new(schema, vec![vec![data]])?;
    let ctx = SessionContext::new();
    ctx.register_table("test", Arc::new(table))?;

    let sql = "SELECT AVG(c1), SUM(DISTINCT c2) FROM test";
    let actual = execute_to_batches(&ctx, sql).await;
    Ok(())
}

then error raises like

NotImplemented(\"SUM(DISTINCT) aggregations are not available\") at Creating physical plan for 'SELECT AVG(c1), SUM(DISTINCT c2) FROM test': Projection: #AVG(test.c1), #SUM(DISTINCT test.c2)\n  Aggregate: groupBy=[[]], aggr=[[AVG(#test.c1), SUM(DISTINCT #test.c2)]]\n    TableScan: test projection=Some([0, 1])

What changes are included in this PR?

Introduces expressions::DistinctSum into DF

• Maintains a HashSet to record unique numeric list, update HashSet when new item input.
• Aggregate state stores in ScalarValue::List, which Built from HashSet.
• During Final aggregate, evaluates DistinctSum by computing sum from numeric stored in HashSet

Are there any user-facing changes?

No.

[WinkerDu]

cc @andygrove @alamb @yjshen
Please have a review, thank you.

[alamb]

LGTM -- thanks @WinkerDu

[alamb]

👍

[alamb]

This looks like a fine change in this PR -- though it is strange to me that we have to be doing these casts in sum.rs as it duplicates some non trivial amount of the logic in coercion -- maybe it would be possible to make this code cleaner / consolidate more of the coercion logic.

Again, no changes needed for this PR but I figured I would point it out while reading this

[WinkerDu]

Thanks @alamb
I'd look into this coercion logic in some follow up pr

[yjshen]

Coercion u64 to i64 seems irrational to me. Why do we need this kind of coercion in sum distinct?

[alamb]

I think @WinkerDu is just following the same pre-existing pattern with this PR

I agree that the pre-existing pattern doesn't make sense to me (it should have already been done by the time the distinct aggregate code is being executed)

[alamb]

👍

[alamb]

👍

[alamb]

👍

[Ted-Jiang]

LGTM

[WinkerDu]

cc @andygrove @tustvold @Dandandan
PTAL, thank you

[alamb]

Thanks @WinkerDu for the code and @yjshen @Ted-Jiang and @xudong963 for the reviews.

It will be nice to clean up some of the coercion logic -- @WinkerDu let me know if I should file a ticket to track that work

[WinkerDu]

@WinkerDu let me know if I should file a ticket to track that work

Thank you all

It would be good if you can help me filing the ticket, thank you @alamb

[alamb]

It would be good if you can help me filing the ticket, thank you @alamb

I filed #2447 though in so doing I realized I don't fully understand if there is any change needed or if the coercion is required.