feat: support concat for strings

[comphead]

Which issue does this PR close?

Closes #2552 .

Rationale for this change

What changes are included in this PR?

How are these changes tested?

[comphead]

Depends on #2586

[codecov-commenter]

Codecov Report

❌ Patch coverage is 83.33333% with 1 line in your changes missing coverage. Please review.
✅ Project coverage is 59.22%. Comparing base (f09f8af) to head (164dbfe).
⚠️ Report is 649 commits behind head on main.

Files with missing lines	Patch %	Lines
...rc/main/scala/org/apache/comet/serde/strings.scala	80.00%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #2604      +/-   ##
============================================
+ Coverage     56.12%   59.22%   +3.10%     
- Complexity      976     1448     +472     
============================================
  Files           119      147      +28     
  Lines         11743    13762    +2019     
  Branches       2251     2365     +114     
============================================
+ Hits           6591     8151    +1560     
- Misses         4012     4387     +375     
- Partials       1140     1224      +84     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[comphead]

@andygrove please take another look.
concat works with strings, and for other datatypes it is being fixed in DataFusion apache/datafusion#18020

[andygrove]

I think we need type checks so that we fall back to Spark for unsupported argument types?

Perhaps something like this?

object CometConcat extends CometScalarFunction[Concat]("concat") {
  override def getSupportLevel(expr: Concat): SupportLevel = {
    if (expr.children.forall(_.dataType == DataTypes.StringType)) {
      Compatible()
    } else {
      Incompatible(Some("Only string arguments are supported"))
    }
  }
}

[andygrove]

I know that this PR is just to support string inputs in Comet concat, but the fuzz tester should ideally test for all types that Spark supports

[andygrove]

Why is this hash aggregate now supported in Comet? I don't see concat used in the query.

https://github.com/apache/datafusion-benchmarks/blob/main/tpcds/queries-spark/q5.sql

[andygrove]

Could you enable these tests and use the recently added checkSparkAnswerAndFallbackReason method to make sure we are correctly falling back to Spark?

[comphead]

@andygrove PTAL

[andygrove]

LGTM. Thanks @comphead!

[andygrove]

In case anyone is wondering why more operators are now running natively, even though the TPC-DS queries do not contain concat ... Spark adds concats in aggregates and Comet was previously falling back to Spark:

:- HashAggregate [COMET: concat is not supported, Unsupported result expressions found in: Vector(MakeDecimal(sum(UnscaledValue(sales_price#3596))#3818L,17,2) AS sales#3606, MakeDecimal(sum(UnscaledValue(return_amt#3598))#3820L,17,2) AS returns#3608, (MakeDecimal(sum(UnscaledValue(profit#3597))#3819L,17,2) - MakeDecimal(sum(UnscaledValue(net_loss#3599))#3821L,17,2)) AS profit#3584, store channel AS channel#3840, concat(store, s_store_id#3856) AS id#3841)]

[rluvaton]

In case anyone is wondering why more operators are now running natively, even though the TPC-DS queries do not contain concat ... Spark adds concats in aggregates and Comet was previously falling back to Spark:

:- HashAggregate [COMET: concat is not supported, Unsupported result expressions found in: Vector(MakeDecimal(sum(UnscaledValue(sales_price#3596))#3818L,17,2) AS sales#3606, MakeDecimal(sum(UnscaledValue(return_amt#3598))#3820L,17,2) AS returns#3608, (MakeDecimal(sum(UnscaledValue(profit#3597))#3819L,17,2) - MakeDecimal(sum(UnscaledValue(net_loss#3599))#3821L,17,2)) AS profit#3584, store channel AS channel#3840, concat(store, s_store_id#3856) AS id#3841)]

concat exists in the query, no?:

'store' || s_store_id as id

https://github.com/apache/datafusion-benchmarks/blob/c472383a2e0570c85766cd8ec946614dd4fda542/tpcds/queries-spark/q5.sql#L105

but spark just move it to aggregate avoid copying more data perhaps?

[comphead]

:) tadaam @rluvaton thats' true, I totally forgot about || 🤦