[SPARK-37802][SQL] Composite field name should work with Aggregate push down

[huaxingao]

What changes were proposed in this pull request?

Currently, composite filed name such as dept id doesn't work with aggregate push down

sql("SELECT COUNT(`dept id`) FROM h2.test.dept")

org.apache.spark.sql.catalyst.parser.ParseException: 
extraneous input 'id' expecting <EOF>(line 1, pos 5)

== SQL ==
dept id
-----^^^

	at org.apache.spark.sql.catalyst.parser.ParseException.withCommand(ParseDriver.scala:271)
	at org.apache.spark.sql.catalyst.parser.AbstractSqlParser.parse(ParseDriver.scala:132)
	at org.apache.spark.sql.catalyst.parser.AbstractSqlParser.parseMultipartIdentifier(ParseDriver.scala:63)
	at org.apache.spark.sql.connector.expressions.LogicalExpressions$.parseReference(expressions.scala:39)
	at org.apache.spark.sql.connector.expressions.FieldReference$.apply(expressions.scala:365)
	at org.apache.spark.sql.execution.datasources.DataSourceStrategy$.translateAggregate(DataSourceStrategy.scala:717)
	at org.apache.spark.sql.execution.datasources.v2.PushDownUtils$.$anonfun$pushAggregates$1(PushDownUtils.scala:125)
	at scala.collection.immutable.List.flatMap(List.scala:366)
	at org.apache.spark.sql.execution.datasources.v2.PushDownUtils$.pushAggregates(PushDownUtils.scala:125)

This PR fixes the problem.

Why are the changes needed?

bug fixing

Does this PR introduce any user-facing change?

No

How was this patch tested?

New test

[huaxingao]

@cloud-fan Could you please take a look? Thanks!

[Dintion]

and Chinese filed name has same problem

== SQL ==
缺陷编号
^^^
at org.apache.spark.sql.catalyst.parser.ParseException.withCommand(ParseDriver.scala:265)
at org.apache.spark.sql.catalyst.parser.AbstractSqlParser.parse(ParseDriver.scala:126)

[huaxingao]

Should be good now @Dintion

[cloud-fan]

I think we need to fix the caller side. We shouldn't call FieldReference.apply(String) which parses the given string. We should call FieldReference(Seq(col_name)).

[cloud-fan]

We know it's a top-level column and it's a waste to parse it again. The column name may contain backtick as well and we need to escape it.

A simpler solution is to skip the parsing: FieldReference(Seq(name)). We can even create an util method for it: FieldReference.column(name)

[beliefer]

@huaxingao Could you wait #35101 merged and update with FieldReference.column(name) ?

[Dintion]

@huaxingao I think the code at org.apache.spark.sql.execution.datasources.v2.PushDownUtils#pushAggregates#columnAsString

 def columnAsString(e: Expression): Option[FieldReference] = e match {
      case PushableColumnWithoutNestedColumn(name) =>
        Some(FieldReference(name).asInstanceOf[FieldReference])
      case _ => None
    }```

 also exist same problem

[beliefer]

Thank you for your work

[cloud-fan]

thanks, merging to master!

[huaxingao]

Thank you all!