simplification

Author: cloud-fan

sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/Sum.java

@@ -18,7 +18,6 @@
 package org.apache.spark.sql.connector.expressions;
 
 import org.apache.spark.annotation.Evolving;
-import org.apache.spark.sql.types.DataType;
 
 /**
  * An aggregate function that returns the summation of all the values in a group.

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRDD.scala

@@ -54,9 +54,14 @@ object JDBCRDD extends Logging {
     val url = options.url
     val table = options.tableOrQuery
     val dialect = JdbcDialects.get(url)
+    getQueryOutputSchema(dialect.getSchemaQuery(table), options, dialect)
+  }
+
+  def getQueryOutputSchema(
+      query: String, options: JDBCOptions, dialect: JdbcDialect): StructType = {
     val conn: Connection = JdbcUtils.createConnectionFactory(options)()
     try {
-      val statement = conn.prepareStatement(dialect.getSchemaQuery(table))
+      val statement = conn.prepareStatement(query)
       try {
         statement.setQueryTimeout(options.queryTimeout)
         val rs = statement.executeQuery()

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala

@@ -84,7 +84,7 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
                 expr.collect {
                   // Do not push down duplicated aggregate expressions. For example,
                   // `SELECT max(a) + 1, max(a) + 2 FROM ...`, we should only push down one
-                  // `sum(a)` to the data source.
+                  // `max(a)` to the data source.
                   case agg: AggregateExpression
                       if !aggExprToOutputOrdinal.contains(agg.canonicalized) =>
                     aggExprToOutputOrdinal(agg.canonicalized) = ordinal

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCScanBuilder.scala

@@ -84,18 +84,10 @@ case class JDBCScanBuilder(
       "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))
+    val aggQuery = s"SELECT ${selectList.mkString(",")} FROM ${jdbcOptions.tableOrQuery} " +
+      s"WHERE 1=0 $groupByClause"
     try {
-      finalSchema = JDBCRDD.resolveTable(jdbcOptionsWithAggQuery)
+      finalSchema = JDBCRDD.getQueryOutputSchema(aggQuery, jdbcOptions, dialect)
       pushedAggregateList = selectList
       pushedGroupByCols = Some(groupByCols)
       true