You are viewing a plain text version of this content. The canonical link for it is here.

Posted to notifications@kyuubi.apache.org by "maomaodev (via GitHub)" <gi...@apache.org> on 2026/05/26 04:04:04 UTC

[PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

maomaodev opened a new pull request, #7476:
URL: https://github.com/apache/kyuubi/pull/7476

   <!--
   Thanks for sending a pull request!
   
   Here are some tips for you:
     1. If this is your first time, please read our contributor guidelines: https://kyuubi.readthedocs.io/en/master/contributing/code/index.html
     2. If the PR is related to an issue in https://github.com/apache/kyuubi/issues, add '[KYUUBI #XXXX]' in your PR title, e.g., '[KYUUBI #XXXX] Your PR title ...'.
     3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][KYUUBI #XXXX] Your PR title ...'.
   -->
   
   ### Why are the changes needed?
   <!--
   Please clarify why the changes are needed. For instance,
     1. If you propose a new API, clarify the use case for a new API.
     2. If you fix a bug, you can clarify why it is a bug, and what versions are affected.
   -->
   Part 2 of 2 to add KSHC support for dynamic partition pruning (DPP). See https://github.com/apache/kyuubi/issues/6943.
   - [x] Add DPP support in `HiveScan` for non-Parquet/ORC tables.
   - [x] Add DPP support in `ParquetScan` / `ORCScan` for Parquet/ORC tables.
   
   ### How was this patch tested?
   <!--
   If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
   If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
   If tests were not added, please describe why they were not added and/or why it was difficult to add.
   -->
   #### 1. UT & TPC-DS benchmark
   - Unit tests
   - Manual test: TPC-DS benchmark (10 GB dataset, ORC and Parquet separately). Spark configuration used for the benchmark(Spark 3.5.7, Kyuubi 1.12.0-SNAPSHOT):
   ```
   spark.driver.cores            1
   spark.driver.memory           4g
   spark.executor.cores          1
   spark.executor.instances      10
   spark.executor.memory         4g
   spark.master                  yarn
   spark.shuffle.service.enabled true
   spark.yarn.appMasterEnv.JAVA_HOME /usr/local/jdk-17
   spark.executorEnv.JAVA_HOME       /usr/local/jdk-17
   ```
   #### 2. ORC benchmark
   
   - **Overall performance (sum of 99)**
   
   | Dimension         | Vanilla Spark | KSHC Before |    KSHC Now |
   | ----------------- | ------------: | ----------: | ----------: |
   | Total time        |     2481.85 s |   3353.43 s |   2197.26 s |
   | vs. Vanilla Spark |             — |     +35.12% |     −11.47% |
   | vs. KSHC Before   |             — |           — | **−34.48%** |
   
   - **DPP hit subset (73/99)**
   
   DPP trigger was detected by matching `runtime partition filter` in the driver logs.
   
   ```
   3,4,5,6,7,8,10,11,12,13,14,15,17,18,19,20,23,25,26,27,29,30,31,32,33,
   35,36,38,40,42,43,45,46,47,48,49,50,51,52,53,54,55,56,57,58,60,61,63,
   64,65,66,67,68,69,70,71,72,74,75,77,78,79,80,81,83,85,86,87,89,91,92,
   97,98
   ```
   
   | Dimension         | Vanilla Spark | KSHC Before |    KSHC Now |
   | ----------------- | ------------: | ----------: | ----------: |
   | Subset total time |     1823.48 s |   2642.11 s |   1484.39 s |
   | vs. Vanilla Spark |             — |     +44.89% |     −18.60% |
   | vs. KSHC Before   |             — |           — | **−43.82%** |
   
   On the DPP-hit subset, KSHC Now provides a 43.82% speedup over KSHC Before, noticeably larger than the overall 34.48%, indicating the performance benefit mainly comes from queries where DPP is triggered.
   
   #### 3. Parquet benchmark
   
   - **Overall performance (sum of 99)**
   
   | Dimension         | Vanilla Spark | KSHC Before |    KSHC Now |
   | ----------------- | ------------: | ----------: | ----------: |
   | Total time        |     2325.13 s |   3363.57 s |   2152.18 s |
   | vs. Vanilla Spark |             — |     +44.66% |      −7.44% |
   | vs. KSHC Before   |             — |           — | **−36.02%** |
   
   - **DPP hit subset (73/99)**
   
   DPP trigger was detected by matching `runtime partition filter` in the driver logs.
   
   ```
   3,4,5,6,7,8,10,11,12,13,14,15,17,18,19,20,23,25,26,27,29,30,31,32,33,
   35,36,38,40,42,43,45,46,47,48,49,50,51,52,53,54,55,56,57,58,60,61,63,
   64,65,66,67,68,69,70,71,72,74,75,77,78,79,80,81,83,85,86,87,89,91,92,
   97,98
   ```
   
   | Dimension         | Vanilla Spark | KSHC Before |    KSHC Now |
   | ----------------- | ------------: | ----------: | ----------: |
   | Subset total time |     1619.13 s |   2487.20 s |   1369.55 s |
   | vs. Vanilla Spark |             — |     +53.61% |     −15.41% |
   | vs. KSHC Before   |             — |           — | **−44.94%** |
   
   On the DPP-hit subset, KSHC Now provides a 44.94% speedup over KSHC Before, noticeably larger than the overall 36.02%, indicating the performance benefit mainly comes from queries where DPP is triggered.
   
   #### 4. Result correctness
   
   Compared each of the 99 result files between KSHC Now and Vanilla Spark for both ORC and Parquet. ORC: 94/99 byte-identical and 98/99 row-multiset-identical; Parquet: identical figures. The 4 row-order-only diffs (q31/q65/q71/q79) come from queries whose `ORDER BY` clause does not totally order the output. The single multiset diff (q39) is sub-ULP floating-point rounding in `stddev`-style aggregates and is also present between KSHC Before and Vanilla Spark, so it is unrelated to this PR. No correctness regression introduced.
   
   
   #### 5. Spark 4.0.1 benchmark
   
   The same TPC-DS benchmark was also run against Spark 4.0.1 with KSHC. Results align with the Spark 3.5.7 numbers: KSHC matches or outperforms the native Hive path on DPP-eligible queries, and produces identical result sets. Full Spark 4.0.1 benchmark result are omitted here to keep the report compact, they can be shared on request.
   
   ### Was this patch authored or co-authored using generative AI tooling?
   <!--
   If generative AI tooling has assisted in authoring or reviewing this patch, please include
   an 'Assisted-by: ' trailer that identifies the agent and model.
   For example: 'Assisted-by: Claude:claude-opus-4-7' or 'Assisted-by: Claude Opus 4.7'.
   If no, write 'No'.
   Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
   -->
   Partially assisted by Claude Code (Claude Opus 4.7) for unit test, code style fixes, and analysis of TPC-DS benchmark results. Core design and implementation are human-authored.


-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "pan3793 (via GitHub)" <gi...@apache.org>.

pan3793 commented on PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#issuecomment-5261538977

   @maomaodev, can you rebase on master and see what happens for the new supported 4.1 and 4.2? Currently, we deliver one binary-compatible KSHC jar for all supported Spark versions, but the proposed approach breaks that. This needs further discussion, and one option is to drop support for Spark 3.3 and 3.4, as they have been EOL and marked as deprecated by Kyuubi


-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "pan3793 (via GitHub)" <gi...@apache.org>.

pan3793 commented on PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#issuecomment-5289124658

   I opened https://github.com/apache/kyuubi/pull/7631 to drop support for Spark 3.4 and 3.5, so we don't need to worry about the `scala-spark-3.5-plus` / `scala-spark-pre-3.5` issue


-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.

LuciferYang commented on PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#issuecomment-5279601666

   > cc @LuciferYang, do you have time to take a look? Since you tried to make similar changes on the Spark side
   
   ok, will feedback later.


-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "pan3793 (via GitHub)" <gi...@apache.org>.

pan3793 commented on PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#issuecomment-5396385767

   **Merge Summary:**
   - merged into master https://github.com/apache/kyuubi/commit/0e26b7030ad75824f80e95e05da1129fd50f361d
   
   *Posted by `merge_kyuubi_pr.py`*


-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "maomaodev (via GitHub)" <gi...@apache.org>.

maomaodev commented on code in PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#discussion_r3788109016


##########
extensions/spark/kyuubi-spark-connector-hive/src/test/scala/org/apache/kyuubi/spark/connector/hive/DynamicPartitionPruningSuite.scala:
##########
@@ -86,19 +85,32 @@ class DynamicPartitionPruningSuite extends KyuubiHiveTest {
                | WHERE d.tag = 'target'
                |""".stripMargin
 
+          val df = spark.sql(sql)
           checkAnswer(
-            spark.sql(sql),
+            df,
             Seq(
               Row(3, "c", "2026-05-01"),
               Row(4, "d", "2026-05-01")))
 
           // DPP being actually applied is observable as a `DynamicPruningExpression`
           // injected into `BatchScanExec.runtimeFilters`.
-          val exec = findBatchScanExec(spark, sql, fact.split('.').last)
+          val exec = findBatchScanExec(df.queryExecution.executedPlan, fact.split('.').last)

Review Comment:
   Fixed in `DynamicPartitionPruningSuite`:
   * Added a strict `plannedPartitions(on)` < `plannedPartitions(off)` assertion, so the test now proves DPP actually prunes partitions instead of only observing the `DynamicPruningExpression` node.
   * Added a `columnarSupportMode()` assertion per scan type:` SUPPORTED` for `KyuubiOrcScan` / `KyuubiParquetScan`, `UNSUPPORTED` for `HiveScan`.



-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "maomaodev (via GitHub)" <gi...@apache.org>.

maomaodev commented on code in PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#discussion_r3301154375


##########
pom.xml:
##########
@@ -2069,6 +2069,7 @@
                 <module>extensions/spark/kyuubi-spark-connector-hive</module>
             </modules>
             <properties>
+                <maven.compiler.release>17</maven.compiler.release>

Review Comment:
   The existing spark-4.0 profile is missing this property, while spark-4.1 already has it. Without it, scalac reports `Class java.lang.Record not found` once a module references a JDK-17-only Spark 4.0 API (e.g. Aggregation's Record type from SPARK-45919).
   Inlining this one-line fix here to unblock CI, happy to split it out into a follow-up PR if preferred.
   
   <img width="1324" height="370" alt="Image" src="https://github.com/user-attachments/assets/35bb2f4c-ae01-4d31-8bb6-1b23a3eea17b" />



-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "maomaodev (via GitHub)" <gi...@apache.org>.

maomaodev commented on code in PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#discussion_r3805355056


##########
extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/spark/sql/hive/kyuubi/connector/HiveBridgeHelper.scala:
##########
@@ -119,4 +124,36 @@ object HiveBridgeHelper {
   implicit class NamespaceHelper(namespace: Array[String]) {
     def quoted: String = namespace.map(quoteIfNeeded).mkString(".")
   }
+
+  def orcConvertibleFilters(
+      schema: StructType,
+      caseSensitive: Boolean,
+      dataFilters: Seq[Filter]): Seq[Filter] = {
+    val dataTypeMap = OrcFilters.getSearchableTypeMap(schema, caseSensitive)
+    OrcFilters.convertibleFilters(dataTypeMap, dataFilters)
+  }
+
+  def parquetConvertibleFilters(
+      readDataSchema: StructType,
+      dataFilters: Seq[Filter]): Seq[Filter] = {
+    val sqlConf = SQLConf.get
+    val pushDownDate = sqlConf.parquetFilterPushDownDate
+    val pushDownTimestamp = sqlConf.parquetFilterPushDownTimestamp
+    val pushDownDecimal = sqlConf.parquetFilterPushDownDecimal
+    val pushDownStringPredicate = sqlConf.parquetFilterPushDownStringPredicate
+    val pushDownInFilterThreshold = sqlConf.parquetFilterPushDownInFilterThreshold
+    val isCaseSensitive = sqlConf.caseSensitiveAnalysis
+    val parquetSchema = new SparkToParquetSchemaConverter(sqlConf).convert(readDataSchema)
+    val rebaseSpec = RebaseSpec(LegacyBehaviorPolicy.CORRECTED)

Review Comment:
   Following the same convention as Spark's own [ParquetScanBuilder.pushDataFilters](https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScanBuilder.scala#L77-L79) — this ParquetFilters instance is only used by `convertibleFilters(...)` to decide push-down eligibility, which does not consume `rebaseSpec` at all. Spark's own code has an inline comment explaining exactly this, I'll port the same comment here to make the intent explicit.



-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "maomaodev (via GitHub)" <gi...@apache.org>.

maomaodev commented on code in PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#discussion_r3788145452


##########
extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/read/KyuubiParquetScanBuilder.scala:
##########
@@ -0,0 +1,120 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.spark.connector.hive.read
+
+import scala.collection.JavaConverters._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.catalog.CatalogTable
+import org.apache.spark.sql.connector.expressions.aggregate.Aggregation
+import org.apache.spark.sql.connector.read.SupportsPushDownAggregates
+import org.apache.spark.sql.execution.datasources.{AggregatePushDownUtils, PartitioningAwareFileIndex}
+import org.apache.spark.sql.execution.datasources.v2.FileScanBuilder
+import org.apache.spark.sql.hive.kyuubi.connector.HiveBridgeHelper
+import org.apache.spark.sql.sources.Filter
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+import org.apache.kyuubi.spark.connector.hive.HiveTableCatalog
+
+/**
+ * A ScanBuilder that mirrors Spark's built-in [[ParquetScanBuilder]] but builds
+ * [[KyuubiParquetScan]] instances, which additionally implement
+ * `SupportsRuntimeFiltering` so that Dynamic Partition Pruning works when
+ * a Hive Parquet table goes through Spark's vectorized Parquet reader path.
+ *
+ * Kept semantically equivalent to [[ParquetScanBuilder]] - filter, aggregate and
+ * column pushdown behaviour is identical.
+ */
+class KyuubiParquetScanBuilder(
+    sparkSession: SparkSession,
+    fileIndex: PartitioningAwareFileIndex,
+    schema: StructType,
+    dataSchema: StructType,
+    options: CaseInsensitiveStringMap,
+    catalogTable: CatalogTable,
+    hiveTableCatalog: HiveTableCatalog)
+  extends FileScanBuilder(sparkSession, fileIndex, dataSchema)
+  with SupportsPushDownAggregates {

Review Comment:
   Updated the class-level scaladoc on `KyuubiParquetScanBuilder` to explicitly note the `SupportsPushDownVariantExtractions` gap on Spark 4.1+, dropping the misleading "semantically equivalent" wording.



-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "maomaodev (via GitHub)" <gi...@apache.org>.

maomaodev commented on code in PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#discussion_r3788122021


##########
extensions/spark/kyuubi-spark-connector-hive/pom.xml:
##########
@@ -140,6 +144,25 @@
 
     <build>
         <plugins>
+            <plugin>
+                <groupId>org.codehaus.mojo</groupId>
+                <artifactId>build-helper-maven-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>add-spark-version-source</id>
+                        <goals>
+                            <goal>add-source</goal>
+                        </goals>
+                        <phase>generate-sources</phase>
+                        <configuration>
+                            <sources>
+                                <source>src/main/${kshc.columnar.source.dir}</source>

Review Comment:
   Resolved by rebase. The PR is now based on latest master which drops Spark 3.3 / 3.4 support (#7631).



-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "pan3793 (via GitHub)" <gi...@apache.org>.

pan3793 commented on PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#issuecomment-5396384318

   thanks, merged to master


-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "pan3793 (via GitHub)" <gi...@apache.org>.

pan3793 commented on code in PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#discussion_r3793550560


##########
extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/read/KyuubiParquetScan.scala:
##########
@@ -0,0 +1,261 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.spark.connector.hive.read
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.catalog.CatalogTable
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.connector.expressions.NamedReference
+import org.apache.spark.sql.connector.expressions.aggregate.Aggregation
+import org.apache.spark.sql.connector.read.{InputPartition, PartitionReaderFactory, Scan, SupportsRuntimeFiltering}
+import org.apache.spark.sql.execution.WholeStageCodegenExec
+import org.apache.spark.sql.execution.datasources.PartitioningAwareFileIndex
+import org.apache.spark.sql.execution.datasources.parquet.ParquetUtils
+import org.apache.spark.sql.execution.datasources.v2.FileScan
+import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan
+import org.apache.spark.sql.sources.Filter
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+import org.apache.kyuubi.util.reflect.{DynClasses, DynConstructors}
+
+/**
+ * A DPP-aware wrapper around Spark's built-in [[ParquetScan]] that adds
+ * [[SupportsRuntimeFiltering]] so Dynamic Partition Pruning can push runtime
+ * IN predicates down to the Hive partitioned scan.
+ *
+ * Implementation notes:
+ * 1. Only DPP-specific methods ([[filter]] / [[filterAttributes]] /
+ *    [[planInputPartitions]]) contain custom logic, all other methods
+ *    delegate to the wrapped [[ParquetScan]].
+ * 2. [[equals]] uses a `KyuubiParquetScan` type pattern before delegating to
+ *    `inner.equals`, so a plain [[ParquetScan]] never compares equal and is
+ *    never reused in its place during exchange/subquery reuse. [[hashCode]]
+ *    is a constant, matching [[FileScan]]'s default.
+ * 3. Native engines (Gluten / Comet) identify scans by class name, so this
+ *    wrapper is not recognized and falls back to JVM reads.
+ */
+class KyuubiParquetScan(
+    val sparkSession: SparkSession,
+    val hadoopConf: Configuration,
+    val fileIndex: PartitioningAwareFileIndex,
+    val dataSchema: StructType,
+    val readDataSchema: StructType,
+    val readPartitionSchema: StructType,
+    val pushedFilters: Array[Filter],
+    val options: CaseInsensitiveStringMap,
+    val pushedAggregate: Option[Aggregation],
+    val partitionFilters: Seq[Expression],
+    val dataFilters: Seq[Expression],
+    val catalogTable: CatalogTable)
+  extends FileScan
+  with SupportsRuntimeFiltering {
+
+  private[hive] val inner: ParquetScan = KyuubiParquetScan.newParquetScan(
+    sparkSession,
+    hadoopConf,
+    fileIndex,
+    dataSchema,
+    readDataSchema,
+    readPartitionSchema,
+    pushedFilters,
+    options,
+    pushedAggregate,
+    partitionFilters,
+    dataFilters)
+
+  private var runtimeFilters: Seq[Expression] = Seq.empty
+
+  private val isCaseSensitive = sparkSession.sessionState.conf.caseSensitiveAnalysis
+
+  /**
+   * The default [[Scan.ColumnarSupportMode.PARTITION_DEFINED]] (SPARK-44505)
+   * would drive `DataSourceV2ScanExecBase.supportsColumnar` to materialise
+   * `inputPartitions` during planning (via `FileScan.partitions` ->
+   * `HiveCatalogFileIndex.listFiles`), triggering a full-table HDFS listing
+   * before runtime filters arrive via [[SupportsRuntimeFiltering.filter]] and
+   * cancelling DPP's end-to-end win.
+   *
+   * We instead decide from sqlConf + schema, matching
+   * `ParquetPartitionReaderFactory.supportColumnarReads` in all non-empty
+   * cases. When DPP prunes every partition, Spark's default would return
+   * `UNSUPPORTED` on the empty list, we still return `SUPPORTED`, adding a
+   * harmless `ColumnarToRow` on an empty RDD.
+   */
+  override def columnarSupportMode(): Scan.ColumnarSupportMode = {
+    val sqlConf = sparkSession.sessionState.conf
+    val schema = StructType(readDataSchema.fields ++ readPartitionSchema.fields)
+    val supportsColumnar = ParquetUtils.isBatchReadSupportedForSchema(sqlConf, schema) &&
+      sqlConf.wholeStageEnabled &&
+      !WholeStageCodegenExec.isTooManyFields(sqlConf, schema)
+    if (supportsColumnar) Scan.ColumnarSupportMode.SUPPORTED
+    else Scan.ColumnarSupportMode.UNSUPPORTED
+  }
+
+  override def filterAttributes(): Array[NamedReference] = {
+    // Under aggregate pushdown, the scan outputs aggregate columns only, so
+    // partition columns may be absent from its output, runtime filtering on
+    // them is also meaningless once results are aggregated.
+    if (pushedAggregate.nonEmpty) Array.empty[NamedReference]
+    else HiveRuntimeFilterSupport.filterAttributes(readPartitionSchema.fieldNames.toSeq)
+  }
+
+  override def filter(filters: Array[Filter]): Unit = {
+    runtimeFilters = HiveRuntimeFilterSupport.toCatalystPartitionFilters(
+      filters,
+      fileIndex.partitionSchema,
+      isCaseSensitive)
+    if (runtimeFilters.nonEmpty) {
+      logInfo(s"Received ${runtimeFilters.length} runtime partition filter(s) for " +
+        s"${catalogTable.identifier}")
+      logDebug(s"Runtime partition filter(s) for ${catalogTable.identifier}: " +
+        s"${runtimeFilters.mkString(", ")}")
+    }
+  }
+
+  override def planInputPartitions(): Array[InputPartition] = {
+    if (runtimeFilters.isEmpty) {
+      inner.planInputPartitions()
+    } else {
+      // Delegate planning to a sibling ParquetScan carrying the merged
+      // partitionFilters ++ runtimeFilters so DPP predicates take effect.
+      val sibling = KyuubiParquetScan.newParquetScan(
+        sparkSession,
+        hadoopConf,
+        fileIndex,
+        dataSchema,
+        readDataSchema,
+        readPartitionSchema,
+        pushedFilters,
+        options,
+        pushedAggregate,
+        partitionFilters ++ runtimeFilters,
+        dataFilters)
+      sibling.planInputPartitions()
+    }
+  }
+
+  override def isSplitable(path: Path): Boolean = inner.isSplitable(path)
+
+  override def readSchema(): StructType = inner.readSchema()
+
+  override def getMetaData(): Map[String, String] = inner.getMetaData()
+
+  override def createReaderFactory(): PartitionReaderFactory = inner.createReaderFactory()
+
+  override def equals(obj: Any): Boolean = obj match {
+    case that: KyuubiParquetScan => this.inner.equals(that.inner)
+    case _ => false
+  }
+
+  override def hashCode(): Int = getClass.hashCode()
+}
+
+object KyuubiParquetScan {
+
+  // Element type of `Array[VariantExtraction]` in Spark 4.1+, null when absent.
+  private lazy val variantExtractionCls: Class[_] = DynClasses.builder()
+    .impl("org.apache.spark.sql.connector.read.VariantExtraction")
+    .orNull()
+    .build()
+
+  private lazy val emptyVariantExtractions: AnyRef =
+    if (variantExtractionCls == null) null
+    else java.lang.reflect.Array.newInstance(variantExtractionCls, 0).asInstanceOf[AnyRef]
+
+  // scalastyle:off parameter.number
+  private[hive] def newParquetScan(
+      sparkSession: SparkSession,
+      hadoopConf: Configuration,
+      fileIndex: PartitioningAwareFileIndex,
+      dataSchema: StructType,
+      readDataSchema: StructType,
+      readPartitionSchema: StructType,
+      pushedFilters: Array[Filter],
+      options: CaseInsensitiveStringMap,
+      pushedAggregate: Option[Aggregation],
+      partitionFilters: Seq[Expression],
+      dataFilters: Seq[Expression]): ParquetScan = {
+    if (variantExtractionCls != null) {
+      // Spark 4.1+
+      DynConstructors.builder()
+        .impl(
+          classOf[ParquetScan],
+          classOf[SparkSession],
+          classOf[Configuration],
+          classOf[PartitioningAwareFileIndex],
+          classOf[StructType],
+          classOf[StructType],
+          classOf[StructType],
+          classOf[Array[Filter]],
+          classOf[CaseInsensitiveStringMap],
+          classOf[Option[Aggregation]],
+          classOf[Seq[Expression]],
+          classOf[Seq[Expression]],
+          emptyVariantExtractions.getClass)
+        .buildChecked()

Review Comment:
   it resolves constructors with `DynConstructors.builder().buildChecked()` on every call, including each sibling-scan planning. Cache the Ctor in a lazy val.



-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "maomaodev (via GitHub)" <gi...@apache.org>.

maomaodev commented on code in PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#discussion_r3811124434


##########
extensions/spark/kyuubi-spark-connector-hive/src/test/scala/org/apache/kyuubi/spark/connector/hive/DynamicPartitionPruningSuite.scala:
##########
@@ -86,19 +90,51 @@ class DynamicPartitionPruningSuite extends KyuubiHiveTest {
                | WHERE d.tag = 'target'
                |""".stripMargin
 
+          val df = spark.sql(sql)
           checkAnswer(
-            spark.sql(sql),
+            df,
             Seq(
               Row(3, "c", "2026-05-01"),
               Row(4, "d", "2026-05-01")))
 
           // DPP being actually applied is observable as a `DynamicPruningExpression`
           // injected into `BatchScanExec.runtimeFilters`.
-          val exec = findBatchScanExec(spark, sql, fact.split('.').last)
+          val exec = findBatchScanExec(df.queryExecution.executedPlan, fact.split('.').last)
           val hasDpp = exec.runtimeFilters.exists(_.isInstanceOf[DynamicPruningExpression])
           assert(hasDpp == enabled)
+
+          val planned = exec.scan.toBatch.planInputPartitions().length
+          plannedPartitions(enabled) = planned
+
+          exec.scan match {
+            case _: KyuubiOrcScan | _: KyuubiParquetScan =>
+              assert(exec.scan.columnarSupportMode() == Scan.ColumnarSupportMode.SUPPORTED)

Review Comment:
   Added four tests in `DynamicPartitionPruningSuite`
   * parquet/orc vectorizedReader off → `UNSUPPORTED`.
   * parquet/orc all-partitions-pruned via DPP → `planInputPartitions().isEmpty` then `SUPPORTED`.



-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "maomaodev (via GitHub)" <gi...@apache.org>.

maomaodev commented on code in PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#discussion_r3788127620


##########
extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/read/KyuubiParquetScan.scala:
##########
@@ -0,0 +1,229 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.spark.connector.hive.read
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.catalog.CatalogTable
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.connector.expressions.NamedReference
+import org.apache.spark.sql.connector.expressions.aggregate.Aggregation
+import org.apache.spark.sql.connector.read.{InputPartition, PartitionReaderFactory, SupportsRuntimeFiltering}
+import org.apache.spark.sql.execution.datasources.PartitioningAwareFileIndex
+import org.apache.spark.sql.execution.datasources.v2.FileScan
+import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan
+import org.apache.spark.sql.sources.Filter
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+import org.apache.kyuubi.util.reflect.{DynClasses, DynConstructors}
+
+/**
+ * A DPP-aware wrapper around Spark's built-in [[ParquetScan]] that adds
+ * [[SupportsRuntimeFiltering]] so Dynamic Partition Pruning can push runtime
+ * IN predicates down to the Hive partitioned scan.
+ *
+ * Implementation notes:
+ * 1. Only DPP-specific methods ([[filter]] / [[filterAttributes]] /
+ *    [[planInputPartitions]]) contain custom logic, all other methods
+ *    delegate to the wrapped [[ParquetScan]].
+ * 2. [[equals]] / [[hashCode]] are overridden to key on `getClass`, so a
+ *    `KyuubiParquetScan` is never reused in place of a plain [[ParquetScan]]
+ *    during exchange/subquery reuse.
+ */
+class KyuubiParquetScan(
+    val sparkSession: SparkSession,
+    val hadoopConf: Configuration,
+    val fileIndex: PartitioningAwareFileIndex,
+    val dataSchema: StructType,
+    val readDataSchema: StructType,
+    val readPartitionSchema: StructType,
+    val pushedFilters: Array[Filter],
+    val options: CaseInsensitiveStringMap,
+    val pushedAggregate: Option[Aggregation],
+    val partitionFilters: Seq[Expression],
+    val dataFilters: Seq[Expression],
+    val catalogTable: CatalogTable)
+  extends FileScan
+  with SupportsRuntimeFiltering
+  with KyuubiParquetColumnarMixin {
+
+  private[hive] val inner: ParquetScan = KyuubiParquetScan.newParquetScan(
+    sparkSession,
+    hadoopConf,
+    fileIndex,
+    dataSchema,
+    readDataSchema,
+    readPartitionSchema,
+    pushedFilters,
+    options,
+    pushedAggregate,
+    partitionFilters,
+    dataFilters)
+
+  private var runtimeFilters: Seq[Expression] = Seq.empty
+
+  private val isCaseSensitive = sparkSession.sessionState.conf.caseSensitiveAnalysis
+
+  override def filterAttributes(): Array[NamedReference] = {

Review Comment:
   Fixed in both `KyuubiParquetScan.filterAttributes` and `KyuubiOrcScan.filterAttributes`.



-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "pan3793 (via GitHub)" <gi...@apache.org>.

pan3793 commented on PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#issuecomment-5278827689

   cc @LuciferYang, do you have time to take a look? Since you tried to make similar changes on the Spark side


-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "maomaodev (via GitHub)" <gi...@apache.org>.

maomaodev commented on code in PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#discussion_r3793840609


##########
extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/read/KyuubiParquetScan.scala:
##########
@@ -0,0 +1,261 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.spark.connector.hive.read
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.catalog.CatalogTable
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.connector.expressions.NamedReference
+import org.apache.spark.sql.connector.expressions.aggregate.Aggregation
+import org.apache.spark.sql.connector.read.{InputPartition, PartitionReaderFactory, Scan, SupportsRuntimeFiltering}
+import org.apache.spark.sql.execution.WholeStageCodegenExec
+import org.apache.spark.sql.execution.datasources.PartitioningAwareFileIndex
+import org.apache.spark.sql.execution.datasources.parquet.ParquetUtils
+import org.apache.spark.sql.execution.datasources.v2.FileScan
+import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan
+import org.apache.spark.sql.sources.Filter
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+import org.apache.kyuubi.util.reflect.{DynClasses, DynConstructors}
+
+/**
+ * A DPP-aware wrapper around Spark's built-in [[ParquetScan]] that adds
+ * [[SupportsRuntimeFiltering]] so Dynamic Partition Pruning can push runtime
+ * IN predicates down to the Hive partitioned scan.
+ *
+ * Implementation notes:
+ * 1. Only DPP-specific methods ([[filter]] / [[filterAttributes]] /
+ *    [[planInputPartitions]]) contain custom logic, all other methods
+ *    delegate to the wrapped [[ParquetScan]].
+ * 2. [[equals]] uses a `KyuubiParquetScan` type pattern before delegating to
+ *    `inner.equals`, so a plain [[ParquetScan]] never compares equal and is
+ *    never reused in its place during exchange/subquery reuse. [[hashCode]]
+ *    is a constant, matching [[FileScan]]'s default.
+ * 3. Native engines (Gluten / Comet) identify scans by class name, so this
+ *    wrapper is not recognized and falls back to JVM reads.
+ */
+class KyuubiParquetScan(
+    val sparkSession: SparkSession,
+    val hadoopConf: Configuration,
+    val fileIndex: PartitioningAwareFileIndex,
+    val dataSchema: StructType,
+    val readDataSchema: StructType,
+    val readPartitionSchema: StructType,
+    val pushedFilters: Array[Filter],
+    val options: CaseInsensitiveStringMap,
+    val pushedAggregate: Option[Aggregation],
+    val partitionFilters: Seq[Expression],
+    val dataFilters: Seq[Expression],
+    val catalogTable: CatalogTable)
+  extends FileScan
+  with SupportsRuntimeFiltering {
+
+  private[hive] val inner: ParquetScan = KyuubiParquetScan.newParquetScan(
+    sparkSession,
+    hadoopConf,
+    fileIndex,
+    dataSchema,
+    readDataSchema,
+    readPartitionSchema,
+    pushedFilters,
+    options,
+    pushedAggregate,
+    partitionFilters,
+    dataFilters)
+
+  private var runtimeFilters: Seq[Expression] = Seq.empty
+
+  private val isCaseSensitive = sparkSession.sessionState.conf.caseSensitiveAnalysis
+
+  /**
+   * The default [[Scan.ColumnarSupportMode.PARTITION_DEFINED]] (SPARK-44505)
+   * would drive `DataSourceV2ScanExecBase.supportsColumnar` to materialise
+   * `inputPartitions` during planning (via `FileScan.partitions` ->
+   * `HiveCatalogFileIndex.listFiles`), triggering a full-table HDFS listing
+   * before runtime filters arrive via [[SupportsRuntimeFiltering.filter]] and
+   * cancelling DPP's end-to-end win.
+   *
+   * We instead decide from sqlConf + schema, matching
+   * `ParquetPartitionReaderFactory.supportColumnarReads` in all non-empty
+   * cases. When DPP prunes every partition, Spark's default would return
+   * `UNSUPPORTED` on the empty list, we still return `SUPPORTED`, adding a
+   * harmless `ColumnarToRow` on an empty RDD.
+   */
+  override def columnarSupportMode(): Scan.ColumnarSupportMode = {
+    val sqlConf = sparkSession.sessionState.conf
+    val schema = StructType(readDataSchema.fields ++ readPartitionSchema.fields)
+    val supportsColumnar = ParquetUtils.isBatchReadSupportedForSchema(sqlConf, schema) &&
+      sqlConf.wholeStageEnabled &&
+      !WholeStageCodegenExec.isTooManyFields(sqlConf, schema)
+    if (supportsColumnar) Scan.ColumnarSupportMode.SUPPORTED
+    else Scan.ColumnarSupportMode.UNSUPPORTED
+  }
+
+  override def filterAttributes(): Array[NamedReference] = {
+    // Under aggregate pushdown, the scan outputs aggregate columns only, so
+    // partition columns may be absent from its output, runtime filtering on
+    // them is also meaningless once results are aggregated.
+    if (pushedAggregate.nonEmpty) Array.empty[NamedReference]
+    else HiveRuntimeFilterSupport.filterAttributes(readPartitionSchema.fieldNames.toSeq)
+  }
+
+  override def filter(filters: Array[Filter]): Unit = {
+    runtimeFilters = HiveRuntimeFilterSupport.toCatalystPartitionFilters(
+      filters,
+      fileIndex.partitionSchema,
+      isCaseSensitive)
+    if (runtimeFilters.nonEmpty) {
+      logInfo(s"Received ${runtimeFilters.length} runtime partition filter(s) for " +
+        s"${catalogTable.identifier}")
+      logDebug(s"Runtime partition filter(s) for ${catalogTable.identifier}: " +
+        s"${runtimeFilters.mkString(", ")}")
+    }
+  }
+
+  override def planInputPartitions(): Array[InputPartition] = {
+    if (runtimeFilters.isEmpty) {
+      inner.planInputPartitions()
+    } else {
+      // Delegate planning to a sibling ParquetScan carrying the merged
+      // partitionFilters ++ runtimeFilters so DPP predicates take effect.
+      val sibling = KyuubiParquetScan.newParquetScan(
+        sparkSession,
+        hadoopConf,
+        fileIndex,
+        dataSchema,
+        readDataSchema,
+        readPartitionSchema,
+        pushedFilters,
+        options,
+        pushedAggregate,
+        partitionFilters ++ runtimeFilters,
+        dataFilters)
+      sibling.planInputPartitions()
+    }
+  }
+
+  override def isSplitable(path: Path): Boolean = inner.isSplitable(path)
+
+  override def readSchema(): StructType = inner.readSchema()
+
+  override def getMetaData(): Map[String, String] = inner.getMetaData()
+
+  override def createReaderFactory(): PartitionReaderFactory = inner.createReaderFactory()
+
+  override def equals(obj: Any): Boolean = obj match {
+    case that: KyuubiParquetScan => this.inner.equals(that.inner)
+    case _ => false
+  }
+
+  override def hashCode(): Int = getClass.hashCode()
+}
+
+object KyuubiParquetScan {
+
+  // Element type of `Array[VariantExtraction]` in Spark 4.1+, null when absent.
+  private lazy val variantExtractionCls: Class[_] = DynClasses.builder()
+    .impl("org.apache.spark.sql.connector.read.VariantExtraction")
+    .orNull()
+    .build()
+
+  private lazy val emptyVariantExtractions: AnyRef =
+    if (variantExtractionCls == null) null
+    else java.lang.reflect.Array.newInstance(variantExtractionCls, 0).asInstanceOf[AnyRef]
+
+  // scalastyle:off parameter.number
+  private[hive] def newParquetScan(
+      sparkSession: SparkSession,
+      hadoopConf: Configuration,
+      fileIndex: PartitioningAwareFileIndex,
+      dataSchema: StructType,
+      readDataSchema: StructType,
+      readPartitionSchema: StructType,
+      pushedFilters: Array[Filter],
+      options: CaseInsensitiveStringMap,
+      pushedAggregate: Option[Aggregation],
+      partitionFilters: Seq[Expression],
+      dataFilters: Seq[Expression]): ParquetScan = {
+    if (variantExtractionCls != null) {
+      // Spark 4.1+
+      DynConstructors.builder()
+        .impl(
+          classOf[ParquetScan],
+          classOf[SparkSession],
+          classOf[Configuration],
+          classOf[PartitioningAwareFileIndex],
+          classOf[StructType],
+          classOf[StructType],
+          classOf[StructType],
+          classOf[Array[Filter]],
+          classOf[CaseInsensitiveStringMap],
+          classOf[Option[Aggregation]],
+          classOf[Seq[Expression]],
+          classOf[Seq[Expression]],
+          emptyVariantExtractions.getClass)
+        .buildChecked()

Review Comment:
   Fixed, cache the Ctor in a lazy val.



-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "maomaodev (via GitHub)" <gi...@apache.org>.

maomaodev commented on code in PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#discussion_r3788098091


##########
extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/HiveTable.scala:
##########
@@ -109,10 +107,24 @@ case class HiveTable(
   override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder = {
     convertedProvider match {
       case Some("ORC") if sparkSession.sessionState.conf.getConf(READ_CONVERT_METASTORE_ORC) =>
-        OrcScanBuilder(sparkSession, fileIndex, schema, dataSchema, options)
+        new KyuubiOrcScanBuilder(
+          sparkSession,
+          fileIndex,
+          schema,
+          dataSchema,
+          options,
+          catalogTable,
+          hiveTableCatalog)
       case Some("PARQUET")
           if sparkSession.sessionState.conf.getConf(READ_CONVERT_METASTORE_PARQUET) =>
-        ParquetScanBuilder(sparkSession, fileIndex, schema, dataSchema, options)
+        new KyuubiParquetScanBuilder(

Review Comment:
   Documented the tradeoff both in the source (class-level scaladoc of `KyuubiParquetScan` / `KyuubiOrcScan`, point 3) and added a paragraph to the PR description: KSHC-converted Parquet/ORC tables fall back to JVM reads under Gluten/Comet. A dedicated escape-hatch config can follow up if a real user hits this.



-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "maomaodev (via GitHub)" <gi...@apache.org>.

maomaodev commented on code in PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#discussion_r3788125964


##########
extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/spark/sql/hive/kyuubi/connector/HiveBridgeHelper.scala:
##########
@@ -119,4 +126,74 @@ object HiveBridgeHelper {
   implicit class NamespaceHelper(namespace: Array[String]) {
     def quoted: String = namespace.map(quoteIfNeeded).mkString(".")
   }
+
+  def orcConvertibleFilters(
+      schema: StructType,
+      caseSensitive: Boolean,
+      dataFilters: Seq[Filter]): Seq[Filter] = {
+    val dataTypeMap = OrcFilters.getSearchableTypeMap(schema, caseSensitive)
+    OrcFilters.convertibleFilters(dataTypeMap, dataFilters)
+  }
+
+  def parquetConvertibleFilters(
+      readDataSchema: StructType,
+      dataFilters: Seq[Filter]): Seq[Filter] = {
+    val sqlConf = SQLConf.get
+    val pushDownDate = sqlConf.parquetFilterPushDownDate
+    val pushDownTimestamp = sqlConf.parquetFilterPushDownTimestamp
+    val pushDownDecimal = sqlConf.parquetFilterPushDownDecimal
+    // sqlConf.parquetFilterPushDownStringPredicate is added in 3.4+, so we use
+    // spark.sql.parquet.filterPushdown.string.startsWith to remain compatible with Spark 3.3
+    val pushDownStringPredicate =
+      sqlConf.getConf(SQLConf.PARQUET_FILTER_PUSHDOWN_STRING_STARTSWITH_ENABLED)
+    val pushDownInFilterThreshold = sqlConf.parquetFilterPushDownInFilterThreshold
+    val isCaseSensitive = sqlConf.caseSensitiveAnalysis
+    val parquetSchema = new SparkToParquetSchemaConverter(sqlConf).convert(readDataSchema)
+    val rebaseSpec = rebaseSpecCorrected
+    val parquetFilters = new ParquetFilters(
+      parquetSchema,
+      pushDownDate,
+      pushDownTimestamp,
+      pushDownDecimal,
+      pushDownStringPredicate,
+      pushDownInFilterThreshold,
+      isCaseSensitive,
+      rebaseSpec)
+    parquetFilters.convertibleFilters(dataFilters)
+  }
+
+  /**
+   * `RebaseSpec(LegacyBehaviorPolicy.CORRECTED, None)` constructed via reflection so
+   * the same code compiles against Spark 3.3 / 3.4 (where `LegacyBehaviorPolicy` is
+   * nested in `SQLConf`) and Spark 3.5+ (SPARK-44538 promoted it to a top-level object
+   * under `org.apache.spark.sql.internal`). Cached because the value is immutable.
+   */
+  private lazy val rebaseSpecCorrected: RebaseSpec = {
+    val policyCls = DynClasses.builder()
+      .impl("org.apache.spark.sql.internal.LegacyBehaviorPolicy$") // SPARK-44538: Spark 3.5+
+      .impl("org.apache.spark.sql.internal.SQLConf$LegacyBehaviorPolicy$") // Spark 3.3 / 3.4
+      .buildChecked()
+
+    val policyModule = DynFields.builder()
+      .impl(policyCls, "MODULE$")
+      .buildStaticChecked[AnyRef]()
+      .get()
+
+    val correctedPolicy = DynMethods.builder("CORRECTED")
+      .impl(policyCls)
+      .buildChecked()
+      .invoke[AnyRef](policyModule)
+
+    // `RebaseSpec.mode` is typed as `LegacyBehaviorPolicy.Value`, which after
+    // erasure becomes `scala.Enumeration$Value` in the constructor signature.
+    // Use the erased type explicitly so `DynConstructors` can match exactly.
+    val enumValueCls = Class.forName("scala.Enumeration$Value")
+    DynConstructors.builder()
+      .impl(
+        "org.apache.spark.sql.catalyst.util.RebaseDateTime$RebaseSpec",

Review Comment:
   Resolved by rebase. The PR is now based on latest master which drops Spark 3.3 / 3.4 support (#7631).



-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "maomaodev (via GitHub)" <gi...@apache.org>.

maomaodev commented on PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#issuecomment-5278753803

> @maomaodev, can you rebase on master and see what happens for the new supported 4.1 and 4.2? Currently, we deliver one binary-compatible KSHC jar for all supported Spark versions, but the proposed approach breaks that. This needs further discussion, and one option is to drop support for Spark 3.3 and 3.4, as they have been EOL and marked as deprecated by Kyuubi

I've rebased on master and made KSHC compile and work against Spark 4.1 and 4.2 as well. Since the previous test environment had already been torn down, I re-ran the full TPC-DS 10 GB benchmark on a fresh cluster covering 4 combinations (Spark 4.1 / 4.2 × ORC / Parquet), and the results are consistent with what we previously observed on Spark 3.5 / 4.0. Happy to share the detailed report / raw logs / per-SQL Excel breakdown if that helps the review.

As you pointed out, binary compatibility with Spark 3.3 / 3.4 is still an open problem on my side. Do you have any suggestions? Thanks!

--
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "maomaodev (via GitHub)" <gi...@apache.org>.

maomaodev commented on code in PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#discussion_r3793843085


##########
extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/read/KyuubiParquetScan.scala:
##########
@@ -0,0 +1,261 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.spark.connector.hive.read
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.catalog.CatalogTable
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.connector.expressions.NamedReference
+import org.apache.spark.sql.connector.expressions.aggregate.Aggregation
+import org.apache.spark.sql.connector.read.{InputPartition, PartitionReaderFactory, Scan, SupportsRuntimeFiltering}
+import org.apache.spark.sql.execution.WholeStageCodegenExec
+import org.apache.spark.sql.execution.datasources.PartitioningAwareFileIndex
+import org.apache.spark.sql.execution.datasources.parquet.ParquetUtils
+import org.apache.spark.sql.execution.datasources.v2.FileScan
+import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan
+import org.apache.spark.sql.sources.Filter
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+import org.apache.kyuubi.util.reflect.{DynClasses, DynConstructors}
+
+/**
+ * A DPP-aware wrapper around Spark's built-in [[ParquetScan]] that adds
+ * [[SupportsRuntimeFiltering]] so Dynamic Partition Pruning can push runtime
+ * IN predicates down to the Hive partitioned scan.
+ *
+ * Implementation notes:
+ * 1. Only DPP-specific methods ([[filter]] / [[filterAttributes]] /
+ *    [[planInputPartitions]]) contain custom logic, all other methods
+ *    delegate to the wrapped [[ParquetScan]].
+ * 2. [[equals]] uses a `KyuubiParquetScan` type pattern before delegating to
+ *    `inner.equals`, so a plain [[ParquetScan]] never compares equal and is
+ *    never reused in its place during exchange/subquery reuse. [[hashCode]]
+ *    is a constant, matching [[FileScan]]'s default.
+ * 3. Native engines (Gluten / Comet) identify scans by class name, so this
+ *    wrapper is not recognized and falls back to JVM reads.
+ */
+class KyuubiParquetScan(
+    val sparkSession: SparkSession,
+    val hadoopConf: Configuration,
+    val fileIndex: PartitioningAwareFileIndex,
+    val dataSchema: StructType,
+    val readDataSchema: StructType,
+    val readPartitionSchema: StructType,
+    val pushedFilters: Array[Filter],
+    val options: CaseInsensitiveStringMap,
+    val pushedAggregate: Option[Aggregation],
+    val partitionFilters: Seq[Expression],
+    val dataFilters: Seq[Expression],
+    val catalogTable: CatalogTable)
+  extends FileScan
+  with SupportsRuntimeFiltering {
+
+  private[hive] val inner: ParquetScan = KyuubiParquetScan.newParquetScan(
+    sparkSession,
+    hadoopConf,
+    fileIndex,
+    dataSchema,
+    readDataSchema,
+    readPartitionSchema,
+    pushedFilters,
+    options,
+    pushedAggregate,
+    partitionFilters,
+    dataFilters)
+
+  private var runtimeFilters: Seq[Expression] = Seq.empty
+
+  private val isCaseSensitive = sparkSession.sessionState.conf.caseSensitiveAnalysis
+
+  /**
+   * The default [[Scan.ColumnarSupportMode.PARTITION_DEFINED]] (SPARK-44505)
+   * would drive `DataSourceV2ScanExecBase.supportsColumnar` to materialise
+   * `inputPartitions` during planning (via `FileScan.partitions` ->
+   * `HiveCatalogFileIndex.listFiles`), triggering a full-table HDFS listing
+   * before runtime filters arrive via [[SupportsRuntimeFiltering.filter]] and
+   * cancelling DPP's end-to-end win.
+   *
+   * We instead decide from sqlConf + schema, matching
+   * `ParquetPartitionReaderFactory.supportColumnarReads` in all non-empty
+   * cases. When DPP prunes every partition, Spark's default would return
+   * `UNSUPPORTED` on the empty list, we still return `SUPPORTED`, adding a
+   * harmless `ColumnarToRow` on an empty RDD.
+   */
+  override def columnarSupportMode(): Scan.ColumnarSupportMode = {
+    val sqlConf = sparkSession.sessionState.conf
+    val schema = StructType(readDataSchema.fields ++ readPartitionSchema.fields)
+    val supportsColumnar = ParquetUtils.isBatchReadSupportedForSchema(sqlConf, schema) &&
+      sqlConf.wholeStageEnabled &&
+      !WholeStageCodegenExec.isTooManyFields(sqlConf, schema)
+    if (supportsColumnar) Scan.ColumnarSupportMode.SUPPORTED
+    else Scan.ColumnarSupportMode.UNSUPPORTED
+  }
+
+  override def filterAttributes(): Array[NamedReference] = {
+    // Under aggregate pushdown, the scan outputs aggregate columns only, so
+    // partition columns may be absent from its output, runtime filtering on
+    // them is also meaningless once results are aggregated.
+    if (pushedAggregate.nonEmpty) Array.empty[NamedReference]
+    else HiveRuntimeFilterSupport.filterAttributes(readPartitionSchema.fieldNames.toSeq)
+  }
+
+  override def filter(filters: Array[Filter]): Unit = {
+    runtimeFilters = HiveRuntimeFilterSupport.toCatalystPartitionFilters(
+      filters,
+      fileIndex.partitionSchema,
+      isCaseSensitive)
+    if (runtimeFilters.nonEmpty) {
+      logInfo(s"Received ${runtimeFilters.length} runtime partition filter(s) for " +
+        s"${catalogTable.identifier}")
+      logDebug(s"Runtime partition filter(s) for ${catalogTable.identifier}: " +
+        s"${runtimeFilters.mkString(", ")}")
+    }
+  }
+
+  override def planInputPartitions(): Array[InputPartition] = {
+    if (runtimeFilters.isEmpty) {
+      inner.planInputPartitions()
+    } else {
+      // Delegate planning to a sibling ParquetScan carrying the merged
+      // partitionFilters ++ runtimeFilters so DPP predicates take effect.
+      val sibling = KyuubiParquetScan.newParquetScan(
+        sparkSession,
+        hadoopConf,
+        fileIndex,
+        dataSchema,
+        readDataSchema,
+        readPartitionSchema,
+        pushedFilters,
+        options,
+        pushedAggregate,
+        partitionFilters ++ runtimeFilters,
+        dataFilters)
+      sibling.planInputPartitions()
+    }
+  }
+
+  override def isSplitable(path: Path): Boolean = inner.isSplitable(path)
+
+  override def readSchema(): StructType = inner.readSchema()
+
+  override def getMetaData(): Map[String, String] = inner.getMetaData()
+
+  override def createReaderFactory(): PartitionReaderFactory = inner.createReaderFactory()
+
+  override def equals(obj: Any): Boolean = obj match {
+    case that: KyuubiParquetScan => this.inner.equals(that.inner)
+    case _ => false
+  }
+
+  override def hashCode(): Int = getClass.hashCode()
+}
+
+object KyuubiParquetScan {
+
+  // Element type of `Array[VariantExtraction]` in Spark 4.1+, null when absent.
+  private lazy val variantExtractionCls: Class[_] = DynClasses.builder()
+    .impl("org.apache.spark.sql.connector.read.VariantExtraction")
+    .orNull()
+    .build()
+
+  private lazy val emptyVariantExtractions: AnyRef =
+    if (variantExtractionCls == null) null
+    else java.lang.reflect.Array.newInstance(variantExtractionCls, 0).asInstanceOf[AnyRef]
+
+  // scalastyle:off parameter.number
+  private[hive] def newParquetScan(
+      sparkSession: SparkSession,
+      hadoopConf: Configuration,
+      fileIndex: PartitioningAwareFileIndex,
+      dataSchema: StructType,
+      readDataSchema: StructType,
+      readPartitionSchema: StructType,
+      pushedFilters: Array[Filter],
+      options: CaseInsensitiveStringMap,
+      pushedAggregate: Option[Aggregation],
+      partitionFilters: Seq[Expression],
+      dataFilters: Seq[Expression]): ParquetScan = {
+    if (variantExtractionCls != null) {
+      // Spark 4.1+

Review Comment:
   Fixed, use `impl( // SPARK-53880 / SPARK-54656 (4.1.0): adds trailing Array[VariantExtraction]`



-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "maomaodev (via GitHub)" <gi...@apache.org>.

maomaodev commented on code in PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#discussion_r3811112375


##########
extensions/spark/kyuubi-spark-connector-hive/src/test/scala/org/apache/kyuubi/spark/connector/hive/HiveCatalogSuite.scala:
##########
@@ -554,4 +553,40 @@ class HiveCatalogSuite extends KyuubiHiveTest {
       }
     }
   }
+
+  test("KyuubiParquetScan and KyuubiOrcScan hadoopConf snapshots at catalog first use") {
+    // Unlike Spark's built-in ScanBuilders rebuilt per query planning from
+    // the current sqlConf, KSHC clones a `lazy val` snapshot taken at the
+    // catalog's first use, so confs SET after that snapshot cannot reach
+    // readers that consult the Hadoop `Configuration` directly.
+    val fieldIdRead = SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key
+    val fieldIdIgnoreMissing = SQLConf.IGNORE_MISSING_PARQUET_FIELD_ID.key
+
+    Seq("orc", "parquet").foreach { provider =>
+      withSparkSession() { spark =>
+        // `catalog` is built in `beforeEach`, its `hadoopConf` snapshot is
+        // already frozen by the time this test runs, so any conf we SET below
+        // is "mid-session" from its perspective.
+        val props = new util.HashMap[String, String]()
+        props.put(TableCatalog.PROP_PROVIDER, provider)
+
+        try {
+          val table = catalog.createTable(testIdent, schema, Array.empty[Transform], props)
+          spark.sessionState.conf.setConfString(fieldIdRead, "true")
+          spark.sessionState.conf.setConfString(fieldIdIgnoreMissing, "true")
+
+          val hadoopConf = table.asInstanceOf[HiveTable]
+            .newScanBuilder(CaseInsensitiveStringMap.empty()).build() match {
+            case s: KyuubiParquetScan => s.hadoopConf
+            case s: KyuubiOrcScan => s.hadoopConf
+            case other => fail(s"unexpected scan type: ${other.getClass.getName}")
+          }
+          assert(hadoopConf.get(fieldIdRead) != "true")

Review Comment:
   Not deliberate, added `HiveTableCatalog#newScanHadoopConf()` (non-cached, same steps incl. token signature), both ScanBuilders clone from it, lazy val kept for the catalog side, and flipped the test to a positive assertion on the lowercased overlay key.



-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "pan3793 (via GitHub)" <gi...@apache.org>.

pan3793 closed pull request #7476: [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP
URL: https://github.com/apache/kyuubi/pull/7476


-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "maomaodev (via GitHub)" <gi...@apache.org>.

maomaodev commented on PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#issuecomment-4552447586

   @pan3793  Could you please take a look when you have time? Thanks!


-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "maomaodev (via GitHub)" <gi...@apache.org>.

maomaodev commented on code in PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#discussion_r3788132012


##########
extensions/spark/kyuubi-spark-connector-hive/src/main/scala-spark-3.5-plus/org/apache/kyuubi/spark/connector/hive/read/KyuubiOrcColumnarMixin.scala:
##########
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.spark.connector.hive.read
+
+import org.apache.spark.sql.connector.read.Scan
+import org.apache.spark.sql.execution.WholeStageCodegenExec
+import org.apache.spark.sql.execution.datasources.orc.OrcUtils
+import org.apache.spark.sql.types.StructType
+
+/**
+ * Hard prerequisite for [[KyuubiOrcScan]]'s DPP support on Spark 3.5+.
+ * [[org.apache.spark.sql.connector.read.Scan.ColumnarSupportMode]] is
+ * introduced in Spark 3.5.0 (SPARK-44505). A no-op trait lives under
+ * `src/main/scala-spark-pre-3.5/` for Spark 3.3 / 3.4.
+ *
+ * With the default [[Scan.ColumnarSupportMode.PARTITION_DEFINED]],
+ * `DataSourceV2ScanExecBase.supportsColumnar` materialises `inputPartitions`
+ * during planning to probe each partition's reader factory, which through
+ * `FileScan.partitions` -> `HiveCatalogFileIndex.listFiles` triggers a
+ * full-table HDFS listing before runtime filters arrive via
+ * [[org.apache.spark.sql.connector.read.SupportsRuntimeFiltering.filter]].
+ * DPP still prunes the data scan correctly, but its end-to-end win is
+ * cancelled by this wasted pre-DPP listing. Vanilla Spark sidesteps the
+ * issue because Hive Parquet/ORC tables fall back to V1 `FileSourceScanExec`
+ * and never consult [[Scan.columnarSupportMode]].
+ */
+trait KyuubiOrcColumnarMixin { this: KyuubiOrcScan =>
+
+  /**
+   * The decision returned here is semantically identical to Spark's
+   * `OrcPartitionReaderFactory.supportColumnarReads`, so advertising it at
+   * scan-level is safe.
+   */
+  override def columnarSupportMode(): Scan.ColumnarSupportMode = {
+    val sqlConf = sparkSession.sessionState.conf
+    val schema = StructType(readDataSchema.fields ++ readPartitionSchema.fields)
+    val supportsColumnar = sqlConf.orcVectorizedReaderEnabled &&
+      sqlConf.wholeStageEnabled &&
+      !WholeStageCodegenExec.isTooManyFields(sqlConf, schema) &&
+      schema.forall(s =>
+        OrcUtils.supportColumnarReads(
+          s.dataType,
+          sqlConf.orcVectorizedReaderNestedColumnEnabled))
+    if (supportsColumnar) Scan.ColumnarSupportMode.SUPPORTED

Review Comment:
   Added to scaladoc of `columnarSupportMode()` in both scans.



-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "pan3793 (via GitHub)" <gi...@apache.org>.

pan3793 commented on code in PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#discussion_r3793531020


##########
extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/read/KyuubiParquetScan.scala:
##########
@@ -0,0 +1,261 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.spark.connector.hive.read
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.catalog.CatalogTable
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.connector.expressions.NamedReference
+import org.apache.spark.sql.connector.expressions.aggregate.Aggregation
+import org.apache.spark.sql.connector.read.{InputPartition, PartitionReaderFactory, Scan, SupportsRuntimeFiltering}
+import org.apache.spark.sql.execution.WholeStageCodegenExec
+import org.apache.spark.sql.execution.datasources.PartitioningAwareFileIndex
+import org.apache.spark.sql.execution.datasources.parquet.ParquetUtils
+import org.apache.spark.sql.execution.datasources.v2.FileScan
+import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan
+import org.apache.spark.sql.sources.Filter
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+import org.apache.kyuubi.util.reflect.{DynClasses, DynConstructors}
+
+/**
+ * A DPP-aware wrapper around Spark's built-in [[ParquetScan]] that adds
+ * [[SupportsRuntimeFiltering]] so Dynamic Partition Pruning can push runtime
+ * IN predicates down to the Hive partitioned scan.
+ *
+ * Implementation notes:
+ * 1. Only DPP-specific methods ([[filter]] / [[filterAttributes]] /
+ *    [[planInputPartitions]]) contain custom logic, all other methods
+ *    delegate to the wrapped [[ParquetScan]].
+ * 2. [[equals]] uses a `KyuubiParquetScan` type pattern before delegating to
+ *    `inner.equals`, so a plain [[ParquetScan]] never compares equal and is
+ *    never reused in its place during exchange/subquery reuse. [[hashCode]]
+ *    is a constant, matching [[FileScan]]'s default.
+ * 3. Native engines (Gluten / Comet) identify scans by class name, so this
+ *    wrapper is not recognized and falls back to JVM reads.
+ */
+class KyuubiParquetScan(
+    val sparkSession: SparkSession,
+    val hadoopConf: Configuration,
+    val fileIndex: PartitioningAwareFileIndex,
+    val dataSchema: StructType,
+    val readDataSchema: StructType,
+    val readPartitionSchema: StructType,
+    val pushedFilters: Array[Filter],
+    val options: CaseInsensitiveStringMap,
+    val pushedAggregate: Option[Aggregation],
+    val partitionFilters: Seq[Expression],
+    val dataFilters: Seq[Expression],
+    val catalogTable: CatalogTable)
+  extends FileScan
+  with SupportsRuntimeFiltering {
+
+  private[hive] val inner: ParquetScan = KyuubiParquetScan.newParquetScan(
+    sparkSession,
+    hadoopConf,
+    fileIndex,
+    dataSchema,
+    readDataSchema,
+    readPartitionSchema,
+    pushedFilters,
+    options,
+    pushedAggregate,
+    partitionFilters,
+    dataFilters)
+
+  private var runtimeFilters: Seq[Expression] = Seq.empty
+
+  private val isCaseSensitive = sparkSession.sessionState.conf.caseSensitiveAnalysis
+
+  /**
+   * The default [[Scan.ColumnarSupportMode.PARTITION_DEFINED]] (SPARK-44505)
+   * would drive `DataSourceV2ScanExecBase.supportsColumnar` to materialise
+   * `inputPartitions` during planning (via `FileScan.partitions` ->
+   * `HiveCatalogFileIndex.listFiles`), triggering a full-table HDFS listing
+   * before runtime filters arrive via [[SupportsRuntimeFiltering.filter]] and
+   * cancelling DPP's end-to-end win.
+   *
+   * We instead decide from sqlConf + schema, matching
+   * `ParquetPartitionReaderFactory.supportColumnarReads` in all non-empty
+   * cases. When DPP prunes every partition, Spark's default would return
+   * `UNSUPPORTED` on the empty list, we still return `SUPPORTED`, adding a
+   * harmless `ColumnarToRow` on an empty RDD.
+   */
+  override def columnarSupportMode(): Scan.ColumnarSupportMode = {
+    val sqlConf = sparkSession.sessionState.conf
+    val schema = StructType(readDataSchema.fields ++ readPartitionSchema.fields)
+    val supportsColumnar = ParquetUtils.isBatchReadSupportedForSchema(sqlConf, schema) &&
+      sqlConf.wholeStageEnabled &&
+      !WholeStageCodegenExec.isTooManyFields(sqlConf, schema)
+    if (supportsColumnar) Scan.ColumnarSupportMode.SUPPORTED
+    else Scan.ColumnarSupportMode.UNSUPPORTED
+  }
+
+  override def filterAttributes(): Array[NamedReference] = {
+    // Under aggregate pushdown, the scan outputs aggregate columns only, so
+    // partition columns may be absent from its output, runtime filtering on
+    // them is also meaningless once results are aggregated.
+    if (pushedAggregate.nonEmpty) Array.empty[NamedReference]
+    else HiveRuntimeFilterSupport.filterAttributes(readPartitionSchema.fieldNames.toSeq)
+  }
+
+  override def filter(filters: Array[Filter]): Unit = {
+    runtimeFilters = HiveRuntimeFilterSupport.toCatalystPartitionFilters(
+      filters,
+      fileIndex.partitionSchema,
+      isCaseSensitive)
+    if (runtimeFilters.nonEmpty) {
+      logInfo(s"Received ${runtimeFilters.length} runtime partition filter(s) for " +
+        s"${catalogTable.identifier}")
+      logDebug(s"Runtime partition filter(s) for ${catalogTable.identifier}: " +
+        s"${runtimeFilters.mkString(", ")}")
+    }
+  }
+
+  override def planInputPartitions(): Array[InputPartition] = {
+    if (runtimeFilters.isEmpty) {
+      inner.planInputPartitions()
+    } else {
+      // Delegate planning to a sibling ParquetScan carrying the merged
+      // partitionFilters ++ runtimeFilters so DPP predicates take effect.
+      val sibling = KyuubiParquetScan.newParquetScan(
+        sparkSession,
+        hadoopConf,
+        fileIndex,
+        dataSchema,
+        readDataSchema,
+        readPartitionSchema,
+        pushedFilters,
+        options,
+        pushedAggregate,
+        partitionFilters ++ runtimeFilters,
+        dataFilters)
+      sibling.planInputPartitions()
+    }
+  }
+
+  override def isSplitable(path: Path): Boolean = inner.isSplitable(path)
+
+  override def readSchema(): StructType = inner.readSchema()
+
+  override def getMetaData(): Map[String, String] = inner.getMetaData()
+
+  override def createReaderFactory(): PartitionReaderFactory = inner.createReaderFactory()
+
+  override def equals(obj: Any): Boolean = obj match {
+    case that: KyuubiParquetScan => this.inner.equals(that.inner)
+    case _ => false
+  }
+
+  override def hashCode(): Int = getClass.hashCode()
+}
+
+object KyuubiParquetScan {
+
+  // Element type of `Array[VariantExtraction]` in Spark 4.1+, null when absent.
+  private lazy val variantExtractionCls: Class[_] = DynClasses.builder()
+    .impl("org.apache.spark.sql.connector.read.VariantExtraction")
+    .orNull()
+    .build()
+
+  private lazy val emptyVariantExtractions: AnyRef =
+    if (variantExtractionCls == null) null
+    else java.lang.reflect.Array.newInstance(variantExtractionCls, 0).asInstanceOf[AnyRef]
+
+  // scalastyle:off parameter.number
+  private[hive] def newParquetScan(
+      sparkSession: SparkSession,
+      hadoopConf: Configuration,
+      fileIndex: PartitioningAwareFileIndex,
+      dataSchema: StructType,
+      readDataSchema: StructType,
+      readPartitionSchema: StructType,
+      pushedFilters: Array[Filter],
+      options: CaseInsensitiveStringMap,
+      pushedAggregate: Option[Aggregation],
+      partitionFilters: Seq[Expression],
+      dataFilters: Seq[Expression]): ParquetScan = {
+    if (variantExtractionCls != null) {
+      // Spark 4.1+

Review Comment:
   move the comment and link the JIRA ticket `.impl // SPARK-XXXXX (4.1.0) do sth`



-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.

LuciferYang commented on code in PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#discussion_r3809717861


##########
extensions/spark/kyuubi-spark-connector-hive/src/test/scala/org/apache/kyuubi/spark/connector/hive/HiveCatalogSuite.scala:
##########
@@ -554,4 +553,40 @@ class HiveCatalogSuite extends KyuubiHiveTest {
       }
     }
   }
+
+  test("KyuubiParquetScan and KyuubiOrcScan hadoopConf snapshots at catalog first use") {
+    // Unlike Spark's built-in ScanBuilders rebuilt per query planning from
+    // the current sqlConf, KSHC clones a `lazy val` snapshot taken at the
+    // catalog's first use, so confs SET after that snapshot cannot reach
+    // readers that consult the Hadoop `Configuration` directly.
+    val fieldIdRead = SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key
+    val fieldIdIgnoreMissing = SQLConf.IGNORE_MISSING_PARQUET_FIELD_ID.key
+
+    Seq("orc", "parquet").foreach { provider =>
+      withSparkSession() { spark =>
+        // `catalog` is built in `beforeEach`, its `hadoopConf` snapshot is
+        // already frozen by the time this test runs, so any conf we SET below
+        // is "mid-session" from its perspective.
+        val props = new util.HashMap[String, String]()
+        props.put(TableCatalog.PROP_PROVIDER, provider)
+
+        try {
+          val table = catalog.createTable(testIdent, schema, Array.empty[Transform], props)
+          spark.sessionState.conf.setConfString(fieldIdRead, "true")
+          spark.sessionState.conf.setConfString(fieldIdIgnoreMissing, "true")
+
+          val hadoopConf = table.asInstanceOf[HiveTable]
+            .newScanBuilder(CaseInsensitiveStringMap.empty()).build() match {
+            case s: KyuubiParquetScan => s.hadoopConf
+            case s: KyuubiOrcScan => s.hadoopConf
+            case other => fail(s"unexpected scan type: ${other.getClass.getName}")
+          }
+          assert(hadoopConf.get(fieldIdRead) != "true")

Review Comment:
   `assert(hadoopConf.get(fieldIdRead) != "true")` is trivially true: `Configuration.get` returns `null` for an absent key, and `null != "true"` always holds, so the assertion cannot tell "never arrived" from "arrived with a different value". `== null` is the assertion you actually mean, and it documents the snapshot semantics on the spot.
   
   It would also pass with an empty `hadoopConf`, so a positive assertion on a catalog option would close that — but it has to use the lowercased key (`javax.jdo.option.connectionurl`), since `catalogOptions` is a `CaseInsensitiveStringMap` and the overlay writes lowercase; the camelCase spelling also arrives from the SparkConf via `getAllConfs`, so asserting that one proves nothing about the overlay.
   
   The test also cements "a conf SET after the snapshot cannot reach the reader" as a contract, which does not hold for Spark's own builder on the same path. Honoring per-catalog settings does not require freezing session confs: add a non-cached `newScanHadoopConf()` on `HiveTableCatalog` (the same steps as today's `hadoopConf`, token signature included), have both ScanBuilders clone from it, and leave the catalog's own `lazy val` alone. Could you confirm the freeze is deliberate, or switch to the overlay and flip this assertion with it?



##########
extensions/spark/kyuubi-spark-connector-hive/src/test/scala/org/apache/kyuubi/spark/connector/hive/DynamicPartitionPruningSuite.scala:
##########
@@ -86,19 +90,51 @@ class DynamicPartitionPruningSuite extends KyuubiHiveTest {
                | WHERE d.tag = 'target'
                |""".stripMargin
 
+          val df = spark.sql(sql)
           checkAnswer(
-            spark.sql(sql),
+            df,
             Seq(
               Row(3, "c", "2026-05-01"),
               Row(4, "d", "2026-05-01")))
 
           // DPP being actually applied is observable as a `DynamicPruningExpression`
           // injected into `BatchScanExec.runtimeFilters`.
-          val exec = findBatchScanExec(spark, sql, fact.split('.').last)
+          val exec = findBatchScanExec(df.queryExecution.executedPlan, fact.split('.').last)
           val hasDpp = exec.runtimeFilters.exists(_.isInstanceOf[DynamicPruningExpression])
           assert(hasDpp == enabled)
+
+          val planned = exec.scan.toBatch.planInputPartitions().length
+          plannedPartitions(enabled) = planned
+
+          exec.scan match {
+            case _: KyuubiOrcScan | _: KyuubiParquetScan =>
+              assert(exec.scan.columnarSupportMode() == Scan.ColumnarSupportMode.SUPPORTED)

Review Comment:
   For the two Kyuubi scans the `columnarSupportMode()` assertions only cover the `SUPPORTED` side (the `HiveScan` one asserts `UNSUPPORTED`, but that is a constant override and pins no logic). Two things are untested: the `UNSUPPORTED` branch when the vectorized reader is off, and the deliberate divergence spelled out in the scaladoc — with every partition pruned by DPP, Spark's default evaluates to false while this still returns `SUPPORTED`. CI would catch neither logic rewritten to ignore the conf nor a future "cleanup" of that divergence.
   
   Both are worth adding, with two things to watch: the `UNSUPPORTED` case needs `spark.sql.parquet.enableVectorizedReader` and `spark.sql.orc.enableVectorizedReader` turned off separately, or `wholeStage`, the one input both branches read; and the all-pruned case has to assert `planInputPartitions().isEmpty` first, since `columnarSupportMode()` never consults the partition list and the assertion would otherwise just repeat the existing one.



-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "maomaodev (via GitHub)" <gi...@apache.org>.

maomaodev commented on PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#issuecomment-5261594504

   > @maomaodev, can you rebase on master and see what happens for the new supported 4.1 and 4.2? Currently, we deliver one binary-compatible KSHC jar for all supported Spark versions, but the proposed approach breaks that. This needs further discussion, and one option is to drop support for Spark 3.3 and 3.4, as they have been EOL and marked as deprecated by Kyuubi
   
   Ok, I'll rebase on master and test the behavior with Spark 4.1 and 4.2 as soon as possible.


-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "maomaodev (via GitHub)" <gi...@apache.org>.

maomaodev commented on code in PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#discussion_r3788116241


##########
extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/read/KyuubiParquetScanBuilder.scala:
##########
@@ -0,0 +1,120 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.spark.connector.hive.read
+
+import scala.collection.JavaConverters._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.catalog.CatalogTable
+import org.apache.spark.sql.connector.expressions.aggregate.Aggregation
+import org.apache.spark.sql.connector.read.SupportsPushDownAggregates
+import org.apache.spark.sql.execution.datasources.{AggregatePushDownUtils, PartitioningAwareFileIndex}
+import org.apache.spark.sql.execution.datasources.v2.FileScanBuilder
+import org.apache.spark.sql.hive.kyuubi.connector.HiveBridgeHelper
+import org.apache.spark.sql.sources.Filter
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+import org.apache.kyuubi.spark.connector.hive.HiveTableCatalog
+
+/**
+ * A ScanBuilder that mirrors Spark's built-in [[ParquetScanBuilder]] but builds
+ * [[KyuubiParquetScan]] instances, which additionally implement
+ * `SupportsRuntimeFiltering` so that Dynamic Partition Pruning works when
+ * a Hive Parquet table goes through Spark's vectorized Parquet reader path.
+ *
+ * Kept semantically equivalent to [[ParquetScanBuilder]] - filter, aggregate and
+ * column pushdown behaviour is identical.
+ */
+class KyuubiParquetScanBuilder(
+    sparkSession: SparkSession,
+    fileIndex: PartitioningAwareFileIndex,
+    schema: StructType,
+    dataSchema: StructType,
+    options: CaseInsensitiveStringMap,
+    catalogTable: CatalogTable,
+    hiveTableCatalog: HiveTableCatalog)
+  extends FileScanBuilder(sparkSession, fileIndex, dataSchema)
+  with SupportsPushDownAggregates {
+
+  lazy val hadoopConf: Configuration = {
+    // Start from the catalog-level hadoopConf so that per-catalog Hadoop
+    // configurations are honored. Clone it to avoid polluting the shared
+    // instance held by HiveTableCatalog.
+    val conf = new Configuration(hiveTableCatalog.hadoopConfiguration())

Review Comment:
   Kept the catalog-snapshot direction. Added:
   
   - A note in `KyuubiParquetScanBuilder.hadoopConf` / `KyuubiOrcScanBuilder.hadoopConf` scaladoc.
   - A guard test in `HiveCatalogSuite`, asserting `fieldId.read.enabled` / `ignoreMissing` SET mid-session do **not** reach the resulting `hadoopConf`.



-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "maomaodev (via GitHub)" <gi...@apache.org>.

maomaodev commented on code in PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#discussion_r3788123793


##########
extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/spark/sql/hive/kyuubi/connector/HiveBridgeHelper.scala:
##########
@@ -119,4 +126,74 @@ object HiveBridgeHelper {
   implicit class NamespaceHelper(namespace: Array[String]) {
     def quoted: String = namespace.map(quoteIfNeeded).mkString(".")
   }
+
+  def orcConvertibleFilters(
+      schema: StructType,
+      caseSensitive: Boolean,
+      dataFilters: Seq[Filter]): Seq[Filter] = {
+    val dataTypeMap = OrcFilters.getSearchableTypeMap(schema, caseSensitive)
+    OrcFilters.convertibleFilters(dataTypeMap, dataFilters)
+  }
+
+  def parquetConvertibleFilters(
+      readDataSchema: StructType,
+      dataFilters: Seq[Filter]): Seq[Filter] = {
+    val sqlConf = SQLConf.get
+    val pushDownDate = sqlConf.parquetFilterPushDownDate
+    val pushDownTimestamp = sqlConf.parquetFilterPushDownTimestamp
+    val pushDownDecimal = sqlConf.parquetFilterPushDownDecimal
+    // sqlConf.parquetFilterPushDownStringPredicate is added in 3.4+, so we use
+    // spark.sql.parquet.filterPushdown.string.startsWith to remain compatible with Spark 3.3
+    val pushDownStringPredicate =
+      sqlConf.getConf(SQLConf.PARQUET_FILTER_PUSHDOWN_STRING_STARTSWITH_ENABLED)

Review Comment:
   Resolved by rebase. The PR is now based on latest master which drops Spark 3.3 / 3.4 support (#7631).



-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "maomaodev (via GitHub)" <gi...@apache.org>.

maomaodev commented on PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#issuecomment-5367241283

   Thanks @LuciferYang  and @pan3793  for the review and guidance. Addressed both comments in the latest commit — please take another look when convenient.


-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "maomaodev (via GitHub)" <gi...@apache.org>.

maomaodev commented on PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#issuecomment-4541123315

   Two of the CI checks failed, but it seems unrelated to this PR.


-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "maomaodev (via GitHub)" <gi...@apache.org>.

maomaodev commented on PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#issuecomment-5299710953

   > I'll rebase on master
   
   Thanks a lot @LuciferYang  for the incredibly detailed review! The comments spanned ecosystem impact, correctness, test coverage, and maintainability — each one hit a real issue and made the patch materially better. 
   
   I've pushed a new revision that addresses every code-level comment; happy to iterate on any of them. The revision is rebased onto the latest `master`, since #7631 (thanks @pan3793 ) dropped Spark 3.3 / 3.4 support, a few earlier comments that targeted the pre-3.5 compat shims resolve naturally with the rebase.


-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "pan3793 (via GitHub)" <gi...@apache.org>.

pan3793 commented on code in PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#discussion_r3805047913


##########
extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/spark/sql/hive/kyuubi/connector/HiveBridgeHelper.scala:
##########
@@ -119,4 +124,36 @@ object HiveBridgeHelper {
   implicit class NamespaceHelper(namespace: Array[String]) {
     def quoted: String = namespace.map(quoteIfNeeded).mkString(".")
   }
+
+  def orcConvertibleFilters(
+      schema: StructType,
+      caseSensitive: Boolean,
+      dataFilters: Seq[Filter]): Seq[Filter] = {
+    val dataTypeMap = OrcFilters.getSearchableTypeMap(schema, caseSensitive)
+    OrcFilters.convertibleFilters(dataTypeMap, dataFilters)
+  }
+
+  def parquetConvertibleFilters(
+      readDataSchema: StructType,
+      dataFilters: Seq[Filter]): Seq[Filter] = {
+    val sqlConf = SQLConf.get
+    val pushDownDate = sqlConf.parquetFilterPushDownDate
+    val pushDownTimestamp = sqlConf.parquetFilterPushDownTimestamp
+    val pushDownDecimal = sqlConf.parquetFilterPushDownDecimal
+    val pushDownStringPredicate = sqlConf.parquetFilterPushDownStringPredicate
+    val pushDownInFilterThreshold = sqlConf.parquetFilterPushDownInFilterThreshold
+    val isCaseSensitive = sqlConf.caseSensitiveAnalysis
+    val parquetSchema = new SparkToParquetSchemaConverter(sqlConf).convert(readDataSchema)
+    val rebaseSpec = RebaseSpec(LegacyBehaviorPolicy.CORRECTED)

Review Comment:
   why should we hardcode CORRECTED here?



-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "maomaodev (via GitHub)" <gi...@apache.org>.

maomaodev commented on code in PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#discussion_r3788146658


##########
extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/read/KyuubiParquetScan.scala:
##########
@@ -0,0 +1,229 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.spark.connector.hive.read
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.catalog.CatalogTable
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.connector.expressions.NamedReference
+import org.apache.spark.sql.connector.expressions.aggregate.Aggregation
+import org.apache.spark.sql.connector.read.{InputPartition, PartitionReaderFactory, SupportsRuntimeFiltering}
+import org.apache.spark.sql.execution.datasources.PartitioningAwareFileIndex
+import org.apache.spark.sql.execution.datasources.v2.FileScan
+import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan
+import org.apache.spark.sql.sources.Filter
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+import org.apache.kyuubi.util.reflect.{DynClasses, DynConstructors}
+
+/**
+ * A DPP-aware wrapper around Spark's built-in [[ParquetScan]] that adds
+ * [[SupportsRuntimeFiltering]] so Dynamic Partition Pruning can push runtime
+ * IN predicates down to the Hive partitioned scan.
+ *
+ * Implementation notes:
+ * 1. Only DPP-specific methods ([[filter]] / [[filterAttributes]] /
+ *    [[planInputPartitions]]) contain custom logic, all other methods
+ *    delegate to the wrapped [[ParquetScan]].
+ * 2. [[equals]] / [[hashCode]] are overridden to key on `getClass`, so a
+ *    `KyuubiParquetScan` is never reused in place of a plain [[ParquetScan]]
+ *    during exchange/subquery reuse.
+ */
+class KyuubiParquetScan(
+    val sparkSession: SparkSession,
+    val hadoopConf: Configuration,
+    val fileIndex: PartitioningAwareFileIndex,
+    val dataSchema: StructType,
+    val readDataSchema: StructType,
+    val readPartitionSchema: StructType,
+    val pushedFilters: Array[Filter],
+    val options: CaseInsensitiveStringMap,
+    val pushedAggregate: Option[Aggregation],
+    val partitionFilters: Seq[Expression],
+    val dataFilters: Seq[Expression],
+    val catalogTable: CatalogTable)
+  extends FileScan
+  with SupportsRuntimeFiltering
+  with KyuubiParquetColumnarMixin {
+
+  private[hive] val inner: ParquetScan = KyuubiParquetScan.newParquetScan(
+    sparkSession,
+    hadoopConf,
+    fileIndex,
+    dataSchema,
+    readDataSchema,
+    readPartitionSchema,
+    pushedFilters,
+    options,
+    pushedAggregate,
+    partitionFilters,
+    dataFilters)
+
+  private var runtimeFilters: Seq[Expression] = Seq.empty
+
+  private val isCaseSensitive = sparkSession.sessionState.conf.caseSensitiveAnalysis
+
+  override def filterAttributes(): Array[NamedReference] = {
+    HiveRuntimeFilterSupport.filterAttributes(readPartitionSchema.fieldNames.toSeq)
+  }
+
+  override def filter(filters: Array[Filter]): Unit = {
+    runtimeFilters = HiveRuntimeFilterSupport.toCatalystPartitionFilters(
+      filters,
+      fileIndex.partitionSchema,
+      isCaseSensitive)
+    if (runtimeFilters.nonEmpty) {
+      logInfo(s"Received ${runtimeFilters.length} runtime partition filter(s) for " +
+        s"${catalogTable.identifier}")
+      logDebug(s"Runtime partition filter(s) for ${catalogTable.identifier}: " +
+        s"${runtimeFilters.mkString(", ")}")
+    }
+  }
+
+  override def planInputPartitions(): Array[InputPartition] = {
+    if (runtimeFilters.isEmpty) {
+      inner.planInputPartitions()
+    } else {
+      // Delegate planning to a sibling ParquetScan carrying the merged
+      // partitionFilters ++ runtimeFilters so DPP predicates take effect.
+      val sibling = KyuubiParquetScan.newParquetScan(
+        sparkSession,
+        hadoopConf,
+        fileIndex,
+        dataSchema,
+        readDataSchema,
+        readPartitionSchema,
+        pushedFilters,
+        options,
+        pushedAggregate,
+        partitionFilters ++ runtimeFilters,
+        dataFilters)
+      sibling.planInputPartitions()
+    }
+  }
+
+  override def isSplitable(path: Path): Boolean = inner.isSplitable(path)
+
+  override def readSchema(): StructType = inner.readSchema()
+
+  override def getMetaData(): Map[String, String] = inner.getMetaData()
+
+  override def createReaderFactory(): PartitionReaderFactory = inner.createReaderFactory()
+
+  override def equals(obj: Any): Boolean = obj match {

Review Comment:
   Rewrote the class-level scaladoc (both `KyuubiParquetScan` and `KyuubiOrcScan`) to match the implementation.



-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org

Re: [PR] [KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP [kyuubi]

Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.

LuciferYang commented on code in PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#discussion_r3780802525


##########
extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/read/KyuubiParquetScanBuilder.scala:
##########
@@ -0,0 +1,120 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.spark.connector.hive.read
+
+import scala.collection.JavaConverters._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.catalog.CatalogTable
+import org.apache.spark.sql.connector.expressions.aggregate.Aggregation
+import org.apache.spark.sql.connector.read.SupportsPushDownAggregates
+import org.apache.spark.sql.execution.datasources.{AggregatePushDownUtils, PartitioningAwareFileIndex}
+import org.apache.spark.sql.execution.datasources.v2.FileScanBuilder
+import org.apache.spark.sql.hive.kyuubi.connector.HiveBridgeHelper
+import org.apache.spark.sql.sources.Filter
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+import org.apache.kyuubi.spark.connector.hive.HiveTableCatalog
+
+/**
+ * A ScanBuilder that mirrors Spark's built-in [[ParquetScanBuilder]] but builds
+ * [[KyuubiParquetScan]] instances, which additionally implement
+ * `SupportsRuntimeFiltering` so that Dynamic Partition Pruning works when
+ * a Hive Parquet table goes through Spark's vectorized Parquet reader path.
+ *
+ * Kept semantically equivalent to [[ParquetScanBuilder]] - filter, aggregate and
+ * column pushdown behaviour is identical.
+ */
+class KyuubiParquetScanBuilder(
+    sparkSession: SparkSession,
+    fileIndex: PartitioningAwareFileIndex,
+    schema: StructType,
+    dataSchema: StructType,
+    options: CaseInsensitiveStringMap,
+    catalogTable: CatalogTable,
+    hiveTableCatalog: HiveTableCatalog)
+  extends FileScanBuilder(sparkSession, fileIndex, dataSchema)
+  with SupportsPushDownAggregates {
+
+  lazy val hadoopConf: Configuration = {
+    // Start from the catalog-level hadoopConf so that per-catalog Hadoop
+    // configurations are honored. Clone it to avoid polluting the shared
+    // instance held by HiveTableCatalog.
+    val conf = new Configuration(hiveTableCatalog.hadoopConfiguration())

Review Comment:
   **MEDIUM · behavior compatibility**
   
   The new builders' `hadoopConf` starts from a clone of `hiveTableCatalog.hadoopConfiguration()`, which is a `lazy val` snapshot taken from `newHadoopConf()` at first catalog use; Spark's builders use `newHadoopConfWithOptions`, rebuilt per ScanBuilder (per query planning) with the current `sqlConf.getAllConfs`. So a session conf set after that snapshot only takes effect if `createReaderFactory` re-writes it. `spark.sql.parquet.fieldId.read.enabled` and `...ignoreMissing` are not in that list, and `ParquetReadSupport.init` reads them straight off the Hadoop conf, so enabling field-id reads mid-session has no effect under KSHC.
   
   Sourcing the Hadoop conf from the catalog is the right direction and matches `HiveScan`, so I am not asking for a revert. A sentence in the `hadoopConf` comment saying it is a catalog snapshot that session confs may not reach, plus a test for those two field-id confs, would cover it.



##########
extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/read/KyuubiParquetScanBuilder.scala:
##########
@@ -0,0 +1,120 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.spark.connector.hive.read
+
+import scala.collection.JavaConverters._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.catalog.CatalogTable
+import org.apache.spark.sql.connector.expressions.aggregate.Aggregation
+import org.apache.spark.sql.connector.read.SupportsPushDownAggregates
+import org.apache.spark.sql.execution.datasources.{AggregatePushDownUtils, PartitioningAwareFileIndex}
+import org.apache.spark.sql.execution.datasources.v2.FileScanBuilder
+import org.apache.spark.sql.hive.kyuubi.connector.HiveBridgeHelper
+import org.apache.spark.sql.sources.Filter
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+import org.apache.kyuubi.spark.connector.hive.HiveTableCatalog
+
+/**
+ * A ScanBuilder that mirrors Spark's built-in [[ParquetScanBuilder]] but builds
+ * [[KyuubiParquetScan]] instances, which additionally implement
+ * `SupportsRuntimeFiltering` so that Dynamic Partition Pruning works when
+ * a Hive Parquet table goes through Spark's vectorized Parquet reader path.
+ *
+ * Kept semantically equivalent to [[ParquetScanBuilder]] - filter, aggregate and
+ * column pushdown behaviour is identical.
+ */
+class KyuubiParquetScanBuilder(
+    sparkSession: SparkSession,
+    fileIndex: PartitioningAwareFileIndex,
+    schema: StructType,
+    dataSchema: StructType,
+    options: CaseInsensitiveStringMap,
+    catalogTable: CatalogTable,
+    hiveTableCatalog: HiveTableCatalog)
+  extends FileScanBuilder(sparkSession, fileIndex, dataSchema)
+  with SupportsPushDownAggregates {

Review Comment:
   **LOW · behavior compatibility**
   
   On 4.1 `ParquetScanBuilder` also mixes in `SupportsPushDownVariantExtractions`; `KyuubiParquetScanBuilder` mixes in only `SupportsPushDownAggregates`, and `newParquetScan` hardcodes an empty array for that parameter. It is unreachable today: HMS's `validateColumnType` does not know `variant`, and hive-serde data columns have no placeholder fallback (only partition columns and views do), so such a table cannot be created.
   
   Raising it because `KyuubiParquetScanBuilder`'s comment claims it is "semantically equivalent to `ParquetScanBuilder`", which it is not here; noting the gap in the comment would be more accurate. Also worth knowing that the parity framing does not hold for variant anyway: vanilla Spark reads Hive Parquet tables through a V1 `LogicalRelation`, where the V1 `PushVariantIntoScan` rule does the pushdown.



##########
extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/HiveTable.scala:
##########
@@ -109,10 +107,24 @@ case class HiveTable(
   override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder = {
     convertedProvider match {
       case Some("ORC") if sparkSession.sessionState.conf.getConf(READ_CONVERT_METASTORE_ORC) =>
-        OrcScanBuilder(sparkSession, fileIndex, schema, dataSchema, options)
+        new KyuubiOrcScanBuilder(
+          sparkSession,
+          fileIndex,
+          schema,
+          dataSchema,
+          options,
+          catalogTable,
+          hiveTableCatalog)
       case Some("PARQUET")
           if sparkSession.sessionState.conf.getConf(READ_CONVERT_METASTORE_PARQUET) =>
-        ParquetScanBuilder(sparkSession, fileIndex, schema, dataSchema, options)
+        new KyuubiParquetScanBuilder(

Review Comment:
   **HIGH · ecosystem compatibility**
   
   `KyuubiParquetScan` / `KyuubiOrcScan` extend `FileScan` directly rather than subclassing `ParquetScan` / `OrcScan`, so native engines that identify the file format by scan type stop recognizing them. Gluten's `getSubstraitReadFileFormatV2` is a `getClass.getSimpleName` string match; it now falls through to `UnknownFormat`, validation fails, and the plan silently keeps the vanilla `BatchScanExec`. Comet's `isInstanceOf[ParquetScan]` behaves the same way. This path did work before: Gluten's only structural gate is `scan.isInstanceOf[FileScan]` and it reads the FileIndex generically, so a converted KSHC table was genuinely offloadable and no longer is.
   
   Gluten's check is a string match, so subclassing would not save it either, and `ParquetScan` being a case class makes wrapping the right call in my view. So the ask is not to change the implementation but to record the tradeoff: note in the PR description and release notes that KSHC Parquet/ORC tables fall back to JVM reads under Gluten/Comet. If you want an escape hatch it would have to be a new config, since turning off `convertMetastoreParquet` today routes to `HiveScanBuilder` rather than to the plain scan.



##########
extensions/spark/kyuubi-spark-connector-hive/src/main/scala-spark-3.5-plus/org/apache/kyuubi/spark/connector/hive/read/KyuubiOrcColumnarMixin.scala:
##########
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.spark.connector.hive.read
+
+import org.apache.spark.sql.connector.read.Scan
+import org.apache.spark.sql.execution.WholeStageCodegenExec
+import org.apache.spark.sql.execution.datasources.orc.OrcUtils
+import org.apache.spark.sql.types.StructType
+
+/**
+ * Hard prerequisite for [[KyuubiOrcScan]]'s DPP support on Spark 3.5+.
+ * [[org.apache.spark.sql.connector.read.Scan.ColumnarSupportMode]] is
+ * introduced in Spark 3.5.0 (SPARK-44505). A no-op trait lives under
+ * `src/main/scala-spark-pre-3.5/` for Spark 3.3 / 3.4.
+ *
+ * With the default [[Scan.ColumnarSupportMode.PARTITION_DEFINED]],
+ * `DataSourceV2ScanExecBase.supportsColumnar` materialises `inputPartitions`
+ * during planning to probe each partition's reader factory, which through
+ * `FileScan.partitions` -> `HiveCatalogFileIndex.listFiles` triggers a
+ * full-table HDFS listing before runtime filters arrive via
+ * [[org.apache.spark.sql.connector.read.SupportsRuntimeFiltering.filter]].
+ * DPP still prunes the data scan correctly, but its end-to-end win is
+ * cancelled by this wasted pre-DPP listing. Vanilla Spark sidesteps the
+ * issue because Hive Parquet/ORC tables fall back to V1 `FileSourceScanExec`
+ * and never consult [[Scan.columnarSupportMode]].
+ */
+trait KyuubiOrcColumnarMixin { this: KyuubiOrcScan =>
+
+  /**
+   * The decision returned here is semantically identical to Spark's
+   * `OrcPartitionReaderFactory.supportColumnarReads`, so advertising it at
+   * scan-level is safe.
+   */
+  override def columnarSupportMode(): Scan.ColumnarSupportMode = {
+    val sqlConf = sparkSession.sessionState.conf
+    val schema = StructType(readDataSchema.fields ++ readPartitionSchema.fields)
+    val supportsColumnar = sqlConf.orcVectorizedReaderEnabled &&
+      sqlConf.wholeStageEnabled &&
+      !WholeStageCodegenExec.isTooManyFields(sqlConf, schema) &&
+      schema.forall(s =>
+        OrcUtils.supportColumnarReads(
+          s.dataType,
+          sqlConf.orcVectorizedReaderNestedColumnEnabled))
+    if (supportsColumnar) Scan.ColumnarSupportMode.SUPPORTED

Review Comment:
   **LOW · behavior compatibility**
   
   `columnarSupportMode` and Spark's default `PARTITION_DEFINED` disagree on one boundary: Spark ends in `inputPartitions.exists(...)`, which is false for an empty list and takes the row path, while this looks only at conf and schema and still returns SUPPORTED. DPP pruning every partition hits exactly that case: one extra `ColumnarToRow` in the plan, and since the closure never runs on an empty RDD, results and stability are unaffected.
   
   Outside that boundary the boolean matches the corresponding reader factory on 3.5 through master, aggregate pushdown included. So no code change needed; a sentence in the scaladoc noting the difference would stop someone reasoning from "semantically identical" later. The dropped `require("Cannot mix row-based and columnar input partitions")` costs nothing either: both factories ignore the `partition` argument, so mixing was never possible.



##########
extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/read/KyuubiParquetScan.scala:
##########
@@ -0,0 +1,229 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.spark.connector.hive.read
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.catalog.CatalogTable
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.connector.expressions.NamedReference
+import org.apache.spark.sql.connector.expressions.aggregate.Aggregation
+import org.apache.spark.sql.connector.read.{InputPartition, PartitionReaderFactory, SupportsRuntimeFiltering}
+import org.apache.spark.sql.execution.datasources.PartitioningAwareFileIndex
+import org.apache.spark.sql.execution.datasources.v2.FileScan
+import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan
+import org.apache.spark.sql.sources.Filter
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+import org.apache.kyuubi.util.reflect.{DynClasses, DynConstructors}
+
+/**
+ * A DPP-aware wrapper around Spark's built-in [[ParquetScan]] that adds
+ * [[SupportsRuntimeFiltering]] so Dynamic Partition Pruning can push runtime
+ * IN predicates down to the Hive partitioned scan.
+ *
+ * Implementation notes:
+ * 1. Only DPP-specific methods ([[filter]] / [[filterAttributes]] /
+ *    [[planInputPartitions]]) contain custom logic, all other methods
+ *    delegate to the wrapped [[ParquetScan]].
+ * 2. [[equals]] / [[hashCode]] are overridden to key on `getClass`, so a
+ *    `KyuubiParquetScan` is never reused in place of a plain [[ParquetScan]]
+ *    during exchange/subquery reuse.
+ */
+class KyuubiParquetScan(
+    val sparkSession: SparkSession,
+    val hadoopConf: Configuration,
+    val fileIndex: PartitioningAwareFileIndex,
+    val dataSchema: StructType,
+    val readDataSchema: StructType,
+    val readPartitionSchema: StructType,
+    val pushedFilters: Array[Filter],
+    val options: CaseInsensitiveStringMap,
+    val pushedAggregate: Option[Aggregation],
+    val partitionFilters: Seq[Expression],
+    val dataFilters: Seq[Expression],
+    val catalogTable: CatalogTable)
+  extends FileScan
+  with SupportsRuntimeFiltering
+  with KyuubiParquetColumnarMixin {
+
+  private[hive] val inner: ParquetScan = KyuubiParquetScan.newParquetScan(
+    sparkSession,
+    hadoopConf,
+    fileIndex,
+    dataSchema,
+    readDataSchema,
+    readPartitionSchema,
+    pushedFilters,
+    options,
+    pushedAggregate,
+    partitionFilters,
+    dataFilters)
+
+  private var runtimeFilters: Seq[Expression] = Seq.empty
+
+  private val isCaseSensitive = sparkSession.sessionState.conf.caseSensitiveAnalysis
+
+  override def filterAttributes(): Array[NamedReference] = {
+    HiveRuntimeFilterSupport.filterAttributes(readPartitionSchema.fieldNames.toSeq)
+  }
+
+  override def filter(filters: Array[Filter]): Unit = {
+    runtimeFilters = HiveRuntimeFilterSupport.toCatalystPartitionFilters(
+      filters,
+      fileIndex.partitionSchema,
+      isCaseSensitive)
+    if (runtimeFilters.nonEmpty) {
+      logInfo(s"Received ${runtimeFilters.length} runtime partition filter(s) for " +
+        s"${catalogTable.identifier}")
+      logDebug(s"Runtime partition filter(s) for ${catalogTable.identifier}: " +
+        s"${runtimeFilters.mkString(", ")}")
+    }
+  }
+
+  override def planInputPartitions(): Array[InputPartition] = {
+    if (runtimeFilters.isEmpty) {
+      inner.planInputPartitions()
+    } else {
+      // Delegate planning to a sibling ParquetScan carrying the merged
+      // partitionFilters ++ runtimeFilters so DPP predicates take effect.
+      val sibling = KyuubiParquetScan.newParquetScan(
+        sparkSession,
+        hadoopConf,
+        fileIndex,
+        dataSchema,
+        readDataSchema,
+        readPartitionSchema,
+        pushedFilters,
+        options,
+        pushedAggregate,
+        partitionFilters ++ runtimeFilters,
+        dataFilters)
+      sibling.planInputPartitions()
+    }
+  }
+
+  override def isSplitable(path: Path): Boolean = inner.isSplitable(path)
+
+  override def readSchema(): StructType = inner.readSchema()
+
+  override def getMetaData(): Map[String, String] = inner.getMetaData()
+
+  override def createReaderFactory(): PartitionReaderFactory = inner.createReaderFactory()
+
+  override def equals(obj: Any): Boolean = obj match {

Review Comment:
   **LOW · maintainability**
   
   The class-level comment says `equals` / `hashCode` key on `getClass`, but `equals` is just `case that: KyuubiParquetScan => this.inner.equals(that.inner)`: what separates it from a plain `ParquetScan` is that case's type pattern. `getClass` appears only in `hashCode`, which returns a constant just like Spark's `FileScan` and therefore does not distinguish anything. Delegating to `inner.equals` is the right semantics, so adjusting the comment to match the implementation is enough.
   
   Separately, `KyuubiOrcScan` and `KyuubiParquetScan` have near-identical `filter` / `filterAttributes` / `equals` / `hashCode` plus five delegating methods; a shared trait would remove one copy and make it harder to change the DPP logic on only one side.



##########
extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/spark/sql/hive/kyuubi/connector/HiveBridgeHelper.scala:
##########
@@ -119,4 +126,74 @@ object HiveBridgeHelper {
   implicit class NamespaceHelper(namespace: Array[String]) {
     def quoted: String = namespace.map(quoteIfNeeded).mkString(".")
   }
+
+  def orcConvertibleFilters(
+      schema: StructType,
+      caseSensitive: Boolean,
+      dataFilters: Seq[Filter]): Seq[Filter] = {
+    val dataTypeMap = OrcFilters.getSearchableTypeMap(schema, caseSensitive)
+    OrcFilters.convertibleFilters(dataTypeMap, dataFilters)
+  }
+
+  def parquetConvertibleFilters(
+      readDataSchema: StructType,
+      dataFilters: Seq[Filter]): Seq[Filter] = {
+    val sqlConf = SQLConf.get
+    val pushDownDate = sqlConf.parquetFilterPushDownDate
+    val pushDownTimestamp = sqlConf.parquetFilterPushDownTimestamp
+    val pushDownDecimal = sqlConf.parquetFilterPushDownDecimal
+    // sqlConf.parquetFilterPushDownStringPredicate is added in 3.4+, so we use
+    // spark.sql.parquet.filterPushdown.string.startsWith to remain compatible with Spark 3.3
+    val pushDownStringPredicate =
+      sqlConf.getConf(SQLConf.PARQUET_FILTER_PUSHDOWN_STRING_STARTSWITH_ENABLED)
+    val pushDownInFilterThreshold = sqlConf.parquetFilterPushDownInFilterThreshold
+    val isCaseSensitive = sqlConf.caseSensitiveAnalysis
+    val parquetSchema = new SparkToParquetSchemaConverter(sqlConf).convert(readDataSchema)
+    val rebaseSpec = rebaseSpecCorrected
+    val parquetFilters = new ParquetFilters(
+      parquetSchema,
+      pushDownDate,
+      pushDownTimestamp,
+      pushDownDecimal,
+      pushDownStringPredicate,
+      pushDownInFilterThreshold,
+      isCaseSensitive,
+      rebaseSpec)
+    parquetFilters.convertibleFilters(dataFilters)
+  }
+
+  /**
+   * `RebaseSpec(LegacyBehaviorPolicy.CORRECTED, None)` constructed via reflection so
+   * the same code compiles against Spark 3.3 / 3.4 (where `LegacyBehaviorPolicy` is
+   * nested in `SQLConf`) and Spark 3.5+ (SPARK-44538 promoted it to a top-level object
+   * under `org.apache.spark.sql.internal`). Cached because the value is immutable.
+   */
+  private lazy val rebaseSpecCorrected: RebaseSpec = {
+    val policyCls = DynClasses.builder()
+      .impl("org.apache.spark.sql.internal.LegacyBehaviorPolicy$") // SPARK-44538: Spark 3.5+
+      .impl("org.apache.spark.sql.internal.SQLConf$LegacyBehaviorPolicy$") // Spark 3.3 / 3.4
+      .buildChecked()
+
+    val policyModule = DynFields.builder()
+      .impl(policyCls, "MODULE$")
+      .buildStaticChecked[AnyRef]()
+      .get()
+
+    val correctedPolicy = DynMethods.builder("CORRECTED")
+      .impl(policyCls)
+      .buildChecked()
+      .invoke[AnyRef](policyModule)
+
+    // `RebaseSpec.mode` is typed as `LegacyBehaviorPolicy.Value`, which after
+    // erasure becomes `scala.Enumeration$Value` in the constructor signature.
+    // Use the erased type explicitly so `DynConstructors` can match exactly.
+    val enumValueCls = Class.forName("scala.Enumeration$Value")
+    DynConstructors.builder()
+      .impl(
+        "org.apache.spark.sql.catalyst.util.RebaseDateTime$RebaseSpec",

Review Comment:
   **LOW · maintainability**
   
   `rebaseSpecCorrected` loads `RebaseSpec` reflectively by string name, but the file imports it at the top and declares it as the return type, so it is already a compile-time dependency resolved through the connector's own loader; loading it by name again buys no extra version coverage. The only thing that actually moved across versions is where `LegacyBehaviorPolicy` lives.
   
   `.impl(classOf[RebaseSpec], enumValueCls, classOf[Option[_]])` gets a compile-time check instead: `DynConstructors` has that `Class` overload and resolves via `getConstructor`, and the descriptor matches on 3.3 through 4.2 (verified). If you want to drop the last `Class.forName` too, `enumValueCls` can be `classOf[scala.Enumeration#Value]`.



##########
extensions/spark/kyuubi-spark-connector-hive/src/test/scala/org/apache/kyuubi/spark/connector/hive/DynamicPartitionPruningSuite.scala:
##########
@@ -86,19 +85,32 @@ class DynamicPartitionPruningSuite extends KyuubiHiveTest {
                | WHERE d.tag = 'target'
                |""".stripMargin
 
+          val df = spark.sql(sql)
           checkAnswer(
-            spark.sql(sql),
+            df,
             Seq(
               Row(3, "c", "2026-05-01"),
               Row(4, "d", "2026-05-01")))
 
           // DPP being actually applied is observable as a `DynamicPruningExpression`
           // injected into `BatchScanExec.runtimeFilters`.
-          val exec = findBatchScanExec(spark, sql, fact.split('.').last)
+          val exec = findBatchScanExec(df.queryExecution.executedPlan, fact.split('.').last)

Review Comment:
   **MEDIUM · tests**
   
   Both new cases assert that `runtimeFilters` contains a `DynamicPruningExpression`, which the `filterAttributes` + `SupportsRuntimeFiltering` mix-in alone produces; it says nothing about the sibling-scan branch in `planInputPartitions`. Replace `planInputPartitions` with an unconditional `inner.planInputPartitions()`, gutting the core of this change, and both cases still pass, because `checkAnswer` also passes without pruning. The `columnarSupportMode()` override, the source of the 34–44% win, has no assertion at all.
   
   Two additions would cover it: one asserting the number of partitions/files actually read (`exec.inputRDD.partitions.length` or the scan metrics), and one on `columnarSupportMode()`'s return value.



##########
extensions/spark/kyuubi-spark-connector-hive/pom.xml:
##########
@@ -140,6 +144,25 @@
 
     <build>
         <plugins>
+            <plugin>
+                <groupId>org.codehaus.mojo</groupId>
+                <artifactId>build-helper-maven-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>add-spark-version-source</id>
+                        <goals>
+                            <goal>add-source</goal>
+                        </goals>
+                        <phase>generate-sources</phase>
+                        <configuration>
+                            <sources>
+                                <source>src/main/${kshc.columnar.source.dir}</source>

Review Comment:
   **MEDIUM · maintainability**
   
   The new `src/main/scala-spark-3.5-plus` / `scala-spark-pre-3.5` directories are outside scalastyle's scope: the root pom hardcodes `sourceDirectory` to `src/main/scala`. Spotless is fine, its includes cover `src/main/scala-*/**/*.scala`, so formatting still applies.
   
   This is pre-existing: `kyuubi-spark-sql-engine`'s `src/main/scala-2.13` has the same gap, so it is not this PR's doing and not a merge blocker. It is worth a line here only because this is the first version-specific directory carrying real logic; overriding scalastyle's `sourceDirectory` in the module pom (or leaving a TODO pointing at an issue to fix it globally) would save trouble later.



##########
extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/read/KyuubiParquetScan.scala:
##########
@@ -0,0 +1,229 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.spark.connector.hive.read
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.catalog.CatalogTable
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.connector.expressions.NamedReference
+import org.apache.spark.sql.connector.expressions.aggregate.Aggregation
+import org.apache.spark.sql.connector.read.{InputPartition, PartitionReaderFactory, SupportsRuntimeFiltering}
+import org.apache.spark.sql.execution.datasources.PartitioningAwareFileIndex
+import org.apache.spark.sql.execution.datasources.v2.FileScan
+import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan
+import org.apache.spark.sql.sources.Filter
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+import org.apache.kyuubi.util.reflect.{DynClasses, DynConstructors}
+
+/**
+ * A DPP-aware wrapper around Spark's built-in [[ParquetScan]] that adds
+ * [[SupportsRuntimeFiltering]] so Dynamic Partition Pruning can push runtime
+ * IN predicates down to the Hive partitioned scan.
+ *
+ * Implementation notes:
+ * 1. Only DPP-specific methods ([[filter]] / [[filterAttributes]] /
+ *    [[planInputPartitions]]) contain custom logic, all other methods
+ *    delegate to the wrapped [[ParquetScan]].
+ * 2. [[equals]] / [[hashCode]] are overridden to key on `getClass`, so a
+ *    `KyuubiParquetScan` is never reused in place of a plain [[ParquetScan]]
+ *    during exchange/subquery reuse.
+ */
+class KyuubiParquetScan(
+    val sparkSession: SparkSession,
+    val hadoopConf: Configuration,
+    val fileIndex: PartitioningAwareFileIndex,
+    val dataSchema: StructType,
+    val readDataSchema: StructType,
+    val readPartitionSchema: StructType,
+    val pushedFilters: Array[Filter],
+    val options: CaseInsensitiveStringMap,
+    val pushedAggregate: Option[Aggregation],
+    val partitionFilters: Seq[Expression],
+    val dataFilters: Seq[Expression],
+    val catalogTable: CatalogTable)
+  extends FileScan
+  with SupportsRuntimeFiltering
+  with KyuubiParquetColumnarMixin {
+
+  private[hive] val inner: ParquetScan = KyuubiParquetScan.newParquetScan(
+    sparkSession,
+    hadoopConf,
+    fileIndex,
+    dataSchema,
+    readDataSchema,
+    readPartitionSchema,
+    pushedFilters,
+    options,
+    pushedAggregate,
+    partitionFilters,
+    dataFilters)
+
+  private var runtimeFilters: Seq[Expression] = Seq.empty
+
+  private val isCaseSensitive = sparkSession.sessionState.conf.caseSensitiveAnalysis
+
+  override def filterAttributes(): Array[NamedReference] = {

Review Comment:
   **LOW · robustness**
   
   `filterAttributes` returns every `readPartitionSchema` field name unconditionally, but the new ScanBuilders mix in `SupportsPushDownAggregates`, and the aggregate-pushdown path skips column pruning: `buildScanWithPushedAggregate` builds and replaces the holder before `pruneColumns` ever sees the builder, so `readPartitionSchema()` still holds all partition columns while `readSchema()` (delegated to `inner`) holds only the aggregate columns. The names reported by `filterAttributes` are then absent from the scan's output.
   
   I could not find a reachable failure: with no GROUP BY the residual `Aggregate`'s aliasMap skips aliases containing aggregates, so `PartitionPruning`'s lineage never reaches the scan; with GROUP BY every partition column is in `finalSchema` and `resolveRefs` resolves. So it is not urgent, but `if (pushedAggregate.nonEmpty) Array.empty else …` would make it robust, and runtime filtering on partition columns is meaningless under aggregate pushdown anyway.



##########
extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/spark/sql/hive/kyuubi/connector/HiveBridgeHelper.scala:
##########
@@ -119,4 +126,74 @@ object HiveBridgeHelper {
   implicit class NamespaceHelper(namespace: Array[String]) {
     def quoted: String = namespace.map(quoteIfNeeded).mkString(".")
   }
+
+  def orcConvertibleFilters(
+      schema: StructType,
+      caseSensitive: Boolean,
+      dataFilters: Seq[Filter]): Seq[Filter] = {
+    val dataTypeMap = OrcFilters.getSearchableTypeMap(schema, caseSensitive)
+    OrcFilters.convertibleFilters(dataTypeMap, dataFilters)
+  }
+
+  def parquetConvertibleFilters(
+      readDataSchema: StructType,
+      dataFilters: Seq[Filter]): Seq[Filter] = {
+    val sqlConf = SQLConf.get
+    val pushDownDate = sqlConf.parquetFilterPushDownDate
+    val pushDownTimestamp = sqlConf.parquetFilterPushDownTimestamp
+    val pushDownDecimal = sqlConf.parquetFilterPushDownDecimal
+    // sqlConf.parquetFilterPushDownStringPredicate is added in 3.4+, so we use
+    // spark.sql.parquet.filterPushdown.string.startsWith to remain compatible with Spark 3.3
+    val pushDownStringPredicate =
+      sqlConf.getConf(SQLConf.PARQUET_FILTER_PUSHDOWN_STRING_STARTSWITH_ENABLED)

Review Comment:
   **LOW · consistency**
   
   `parquetConvertibleFilters` reads the old key `PARQUET_FILTER_PUSHDOWN_STRING_STARTSWITH_ENABLED`, while the 3.4+ key `spark.sql.parquet.filterPushdown.stringPredicate` is defined with it as `fallbackConf`, and the fallback is one-directional: with only the new key set, the old one still returns its own default of `true`. Spark's `ParquetScanBuilder` reads the new key. The consequence is milder than it looks: `ParquetPartitionReaderFactory` rebuilds `ParquetFilters` from the new key at read time and all three string branches are gated on that flag, so nothing reaches Parquet; the divergence shows up only as extra entries under `PushedFilters:` in EXPLAIN.
   
   Matching Spark needs no version branch: `sqlConf.getConfString("spark.sql.parquet.filterPushdown.stringPredicate", sqlConf.getConf(PARQUET_FILTER_PUSHDOWN_STRING_STARTSWITH_ENABLED).toString).toBoolean` walks the `FallbackConfigEntry` on 3.4+ and falls back to the supplied default on 3.3 where the key is unregistered. Referring to `sqlConf.parquetFilterPushDownStringPredicate` directly would not compile under `-Pspark-3.3`.



-- 
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: notifications-unsubscribe@kyuubi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org