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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
