viirya commented on a change in pull request #33579:
URL: https://github.com/apache/spark/pull/33579#discussion_r679422484
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCScanBuilder.scala
##########
@@ -51,56 +57,53 @@ case class JDBCScanBuilder(
override def pushedFilters(): Array[Filter] = pushedFilter
- private var pushedAggregations = Option.empty[Aggregation]
-
- private var pushedAggregateColumn: Array[String] = Array()
+ private var pushedAggregateList: Array[String] = Array()
- private def getStructFieldForCol(col: FieldReference): StructField =
- schema.fields(schema.fieldNames.toList.indexOf(col.fieldNames.head))
+ private var pushedGroupByCols: Option[Array[String]] = None
override def pushAggregation(aggregation: Aggregation): Boolean = {
if (!jdbcOptions.pushDownAggregate) return false
val dialect = JdbcDialects.get(jdbcOptions.url)
val compiledAgg =
JDBCRDD.compileAggregates(aggregation.aggregateExpressions, dialect)
+ if (compiledAgg.isEmpty) return false
- var outputSchema = new StructType()
- aggregation.groupByColumns.foreach { col =>
- val structField = getStructFieldForCol(col)
- outputSchema = outputSchema.add(structField)
- pushedAggregateColumn = pushedAggregateColumn :+
dialect.quoteIdentifier(structField.name)
+ val groupByCols = aggregation.groupByColumns.map { col =>
+ if (col.fieldNames.length != 1) return false
+ dialect.quoteIdentifier(col.fieldNames.head)
}
// The column names here are already quoted and can be used to build sql
string directly.
// e.g. "DEPT","NAME",MAX("SALARY"),MIN("BONUS") =>
// SELECT "DEPT","NAME",MAX("SALARY"),MIN("BONUS") FROM "test"."employee"
// GROUP BY "DEPT", "NAME"
- pushedAggregateColumn = pushedAggregateColumn ++ compiledAgg
-
- aggregation.aggregateExpressions.foreach {
- case max: Max =>
- val structField = getStructFieldForCol(max.column)
- outputSchema = outputSchema.add(structField.copy("max(" +
structField.name + ")"))
- case min: Min =>
- val structField = getStructFieldForCol(min.column)
- outputSchema = outputSchema.add(structField.copy("min(" +
structField.name + ")"))
- case count: Count =>
- val distinct = if (count.isDistinct) "DISTINCT " else ""
- val structField = getStructFieldForCol(count.column)
- outputSchema =
- outputSchema.add(StructField(s"count($distinct" + structField.name +
")", LongType))
- case _: CountStar =>
- outputSchema = outputSchema.add(StructField("count(*)", LongType))
- case sum: Sum =>
- val distinct = if (sum.isDistinct) "DISTINCT " else ""
- val structField = getStructFieldForCol(sum.column)
- outputSchema =
- outputSchema.add(StructField(s"sum($distinct" + structField.name +
")", sum.dataType))
- case _ => return false
+ val selectList = groupByCols ++ compiledAgg.get
+ val groupByClause = if (groupByCols.isEmpty) {
+ ""
+ } else {
+ "GROUP BY " + groupByCols.mkString(",")
+ }
+
+ val aggQuery = s"SELECT ${selectList.mkString(",")} FROM " +
+ s"${jdbcOptions.tableOrQuery} $groupByClause"
+ val jdbcOptionsWithAggQuery = new JDBCOptions(
+ jdbcOptions.parameters
+ - JDBCOptions.JDBC_TABLE_NAME
+ - JDBCOptions.JDBC_PARTITION_COLUMN
+ - JDBCOptions.JDBC_NUM_PARTITIONS
+ - JDBCOptions.JDBC_LOWER_BOUND
+ - JDBCOptions.JDBC_UPPER_BOUND +
+ (JDBCOptions.JDBC_QUERY_STRING -> aggQuery))
+ try {
+ finalSchema = JDBCRDD.resolveTable(jdbcOptionsWithAggQuery)
Review comment:
Oh, this is a good change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
