[KYUUBI #7129] Support PARQUET hive table pushdown filter

### Why are the changes needed?

Previously, the `HiveScan` class was used to read data. If it is determined to be PARQUET type, the `ParquetScan` from Spark datasourcev2 can be used. `ParquetScan` supports pushfilter down, but `HiveScan` does not yet support it.

The conversation can be controlled by setting `spark.sql.kyuubi.hive.connector.read.convertMetastoreParquet`. When enabled, the data source PARQUET reader is used to process PARQUET tables created by using the HiveQL syntax, instead of Hive SerDe.

close #7129

### How was this patch tested?

added unit test

### Was this patch authored or co-authored using generative AI tooling?

No

Closes #7130 from flaming-archer/master_parquet_filterdown.

Closes #7129

d7059dc [tian bao] Support PARQUET hive table pushdown filter

Authored-by: tian bao <[email protected]>
Signed-off-by: Cheng Pan <[email protected]>

Author: flaming-archer

extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/HiveTable.scala

@@ -33,11 +33,12 @@ import org.apache.spark.sql.connector.expressions.Transform
 import org.apache.spark.sql.connector.read.ScanBuilder
 import org.apache.spark.sql.connector.write.{LogicalWriteInfo, WriteBuilder}
 import org.apache.spark.sql.execution.datasources.v2.orc.OrcScanBuilder
+import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScanBuilder
 import org.apache.spark.sql.hive.kyuubi.connector.HiveBridgeHelper.{BucketSpecHelper, LogicalExpressions}
 import org.apache.spark.sql.types.StructType
 import org.apache.spark.sql.util.CaseInsensitiveStringMap
 
-import org.apache.kyuubi.spark.connector.hive.KyuubiHiveConnectorConf.READ_CONVERT_METASTORE_ORC
+import org.apache.kyuubi.spark.connector.hive.KyuubiHiveConnectorConf.{READ_CONVERT_METASTORE_ORC, READ_CONVERT_METASTORE_PARQUET}
 import org.apache.kyuubi.spark.connector.hive.read.{HiveCatalogFileIndex, HiveScanBuilder}
 import org.apache.kyuubi.spark.connector.hive.write.HiveWriteBuilder
 
@@ -97,6 +98,9 @@ case class HiveTable(
     convertedProvider match {
       case Some("ORC") if sparkSession.sessionState.conf.getConf(READ_CONVERT_METASTORE_ORC) =>
         OrcScanBuilder(sparkSession, fileIndex, schema, dataSchema, options)
+      case Some("PARQUET")
+          if sparkSession.sessionState.conf.getConf(READ_CONVERT_METASTORE_PARQUET) =>
+        ParquetScanBuilder(sparkSession, fileIndex, schema, dataSchema, options)
       case _ => HiveScanBuilder(sparkSession, fileIndex, dataSchema, catalogTable)
     }
   }

extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/KyuubiHiveConnectorConf.scala

@@ -49,4 +49,12 @@ object KyuubiHiveConnectorConf {
       .version("1.11.0")
       .booleanConf
       .createWithDefault(true)
+
+  val READ_CONVERT_METASTORE_PARQUET =
+    buildConf("spark.sql.kyuubi.hive.connector.read.convertMetastoreParquet")
+      .doc("When enabled, the data source PARQUET reader is used to process " +
+        "PARQUET tables created by using the HiveQL syntax, instead of Hive SerDe.")
+      .version("1.11.0")
+      .booleanConf
+      .createWithDefault(true)
 }

extensions/spark/kyuubi-spark-connector-hive/src/test/scala/org/apache/kyuubi/spark/connector/hive/HiveCatalogSuite.scala

@@ -32,12 +32,13 @@ import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
 import org.apache.spark.sql.connector.catalog.{Identifier, TableCatalog}
 import org.apache.spark.sql.connector.expressions.Transform
 import org.apache.spark.sql.execution.datasources.v2.orc.OrcScan
+import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan
 import org.apache.spark.sql.hive.kyuubi.connector.HiveBridgeHelper._
 import org.apache.spark.sql.types.{IntegerType, StringType, StructType}
 import org.apache.spark.sql.util.CaseInsensitiveStringMap
 
 import org.apache.kyuubi.spark.connector.hive.HiveTableCatalog.IdentifierHelper
-import org.apache.kyuubi.spark.connector.hive.KyuubiHiveConnectorConf.READ_CONVERT_METASTORE_ORC
+import org.apache.kyuubi.spark.connector.hive.KyuubiHiveConnectorConf.{READ_CONVERT_METASTORE_ORC, READ_CONVERT_METASTORE_PARQUET}
 import org.apache.kyuubi.spark.connector.hive.read.HiveScan
 
 class HiveCatalogSuite extends KyuubiHiveTest {
@@ -349,9 +350,30 @@ class HiveCatalogSuite extends KyuubiHiveTest {
     val parProps: util.Map[String, String] = new util.HashMap[String, String]()
     parProps.put(TableCatalog.PROP_PROVIDER, "parquet")
     val pt = catalog.createTable(parquet_table, schema, Array.empty[Transform], parProps)
-    val parScan = pt.asInstanceOf[HiveTable]
-      .newScanBuilder(CaseInsensitiveStringMap.empty()).build().asInstanceOf[HiveScan]
-    assert(parScan.isSplitable(new Path("empty")))
+
+    Seq("true", "false").foreach { value =>
+      withSparkSession(Map(READ_CONVERT_METASTORE_PARQUET.key -> value)) { _ =>
+        val scan = pt.asInstanceOf[HiveTable]
+          .newScanBuilder(CaseInsensitiveStringMap.empty()).build()
+
+        val parScan = value match {
+          case "true" =>
+            assert(
+              scan.isInstanceOf[ParquetScan],
+              s"Expected ParquetScan, got ${scan.getClass.getSimpleName}")
+            scan.asInstanceOf[ParquetScan]
+          case "false" =>
+            assert(
+              scan.isInstanceOf[HiveScan],
+              s"Expected HiveScan, got ${scan.getClass.getSimpleName}")
+            scan.asInstanceOf[HiveScan]
+          case _ =>
+            throw new IllegalArgumentException(
+              s"Unexpected value: '$value'. Only 'true' or 'false' are allowed.")
+        }
+        assert(parScan.isSplitable(new Path("empty")))
+      }
+    }
 
     val orc_table = Identifier.of(testNs, "orc_table")
     val orcProps: util.Map[String, String] = new util.HashMap[String, String]()

extensions/spark/kyuubi-spark-connector-hive/src/test/scala/org/apache/kyuubi/spark/connector/hive/HiveQuerySuite.scala

@@ -353,6 +353,99 @@ class HiveQuerySuite extends KyuubiHiveTest {
     }
   }
 
+  test("PARQUET filter pushdown") {
+    val table = "hive.default.parquet_filter_pushdown"
+    withTable(table) {
+      spark.sql(
+        s"""
+           | CREATE TABLE $table (
+           |  id INT,
+           |  data STRING,
+           |  value INT
+           |  ) PARTITIONED BY (dt STRING, region STRING)
+           |  STORED AS PARQUET
+           | """.stripMargin).collect()
+
+      // Insert test data with partitions
+      spark.sql(
+        s"""
+           | INSERT INTO $table PARTITION (dt='2024-01-01', region='east')
+           | VALUES (1, 'a', 100), (2, 'b', 200), (11, 'aa', 100), (22, 'b', 200)
+           |""".stripMargin)
+
+      spark.sql(
+        s"""
+           | INSERT INTO $table PARTITION (dt='2024-01-01', region='west')
+           | VALUES (3, 'c', 300), (4, 'd', 400), (33, 'cc', 300), (44, 'dd', 400)
+           |""".stripMargin)
+      spark.sql(
+        s"""
+           | INSERT INTO $table PARTITION (dt='2024-01-02', region='east')
+           | VALUES (5, 'e', 500), (6, 'f', 600), (55, 'ee', 500), (66, 'ff', 600)
+           | """.stripMargin)
+
+      // Test multiple partition filters
+      val df1 = spark.sql(
+        s"""
+           | SELECT * FROM $table
+           | WHERE dt = '2024-01-01' AND region = 'east' AND value > 1500
+           |""".stripMargin)
+      assert(df1.count() === 0)
+
+      // Test multiple partition filters
+      val df2 = spark.sql(
+        s"""
+           | SELECT * FROM $table
+           | WHERE dt = '2024-01-01' AND region = 'east' AND value > 150
+           |""".stripMargin)
+      assert(df2.count() === 2)
+      assert(df2.collect().map(_.getInt(0)).toSet === Set(2, 22))
+
+      // Test explain
+      val df3 = spark.sql(
+        s"""
+           | EXPLAIN SELECT count(*) as total_rows
+           | FROM $table
+           | WHERE dt = '2024-01-01' AND region = 'east' AND value > 1
+           |""".stripMargin)
+      assert(df3.count() === 1)
+      // contains like : PushedFilters: [IsNotNull(value), GreaterThan(value,1)]
+      assert(df3.collect().map(_.getString(0)).filter { s =>
+        s.contains("PushedFilters") && !s.contains("PushedFilters: []")
+      }.toSet.size == 1)
+
+      // Test aggregation pushdown partition filters
+      spark.conf.set("spark.sql.parquet.aggregatePushdown", true)
+
+      // Test aggregation pushdown partition filters
+      val df4 = spark.sql(
+        s"""
+           | SELECT count(*) as total_rows
+           | FROM $table
+           | WHERE dt = '2024-01-01' AND region = 'east'
+           | group by dt, region
+           | """.stripMargin)
+      assert(df4.count() === 1)
+      assert(df4.collect().map(_.getLong(0)).toSet === Set(4L))
+
+      val df5 = spark.sql(
+        s"""
+           | EXPLAIN SELECT count(*) as total_rows
+           | FROM $table
+           | WHERE dt = '2024-01-01' AND region = 'east'
+           | group by dt, region
+           | """.stripMargin)
+      assert(df5.count() === 1)
+      // contains like :  PushedAggregation: [COUNT(*)],
+      assert(df5.collect().map(_.getString(0)).filter { s =>
+        s.contains("PushedAggregation") && !s.contains("PushedAggregation: []")
+      }.toSet.size == 1)
+
+      spark.conf.set("spark.sql.parquet.aggregatePushdown", false)
+
+    }
+  }
+
   private def readPartitionedTable(format: String, hiveTable: Boolean): Unit = {
     withSparkSession() { spark =>
       val table = "hive.default.employee"