[KYUUBI #6943][2/2] OrcScan and ParquetScan support DPP - #7476
Conversation
| <module>extensions/spark/kyuubi-spark-connector-hive</module> | ||
| </modules> | ||
| <properties> | ||
| <maven.compiler.release>17</maven.compiler.release> |
There was a problem hiding this 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.
|
Two of the CI checks failed, but it seems unrelated to this PR. |
|
@pan3793 Could you please take a look when you have time? Thanks! |
|
@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. |
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! |
|
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. |
LuciferYang
left a comment
LuciferYang
left a comment
There was a problem hiding this comment.
Read through the change with a focus on the wrapper design and cross-version behavior. The direction looks right to me: wrapping rather than subclassing ParquetScan is the correct call given it is a case class, and the columnarSupportMode() override is a neat way to avoid the plan-stage full-table listing. I verified the reflection sites resolve on 3.3.4 / 3.4.4 / 3.5.8 / 4.0.x / 4.1.x / 4.2 (the Scala 2.12-vs-2.13 Seq erasure difference lines up correctly), and that the columnarSupportMode boolean matches the corresponding reader factory on every version including under aggregate pushdown.
Nothing here blocks merge in my view. The one finding with an externally visible consequence is the first: converted KSHC Parquet/ORC tables stop being offloadable to Gluten/Comet, silently, and I think that tradeoff deserves a line in the PR description and release notes rather than a code change. The two test comments are the ones I would most like to see addressed in this PR, since the current cases stay green if planInputPartitions' sibling-scan branch is gutted.
The remaining comments are lower value: two are gaps against Spark's own builders that I confirmed are unreachable today (the filterAttributes one has no reachable failure path, and hive-serde tables cannot hold a VARIANT column since HMS rejects the type), and the rest are comment or alignment nits. Marked with severity so they are easy to trim.
Caveat on my own work: the Comet claim is based on a stale local checkout (2024-02), so please double-check whether current Comet still keys on isInstanceOf[ParquetScan]. On 4.1/4.2 I compared the ParquetScan constructor descriptors byte-for-byte rather than running getConstructor.
| case Some("PARQUET") | ||
| if sparkSession.sessionState.conf.getConf(READ_CONVERT_METASTORE_PARQUET) => | ||
| ParquetScanBuilder(sparkSession, fileIndex, schema, dataSchema, options) | ||
| new KyuubiParquetScanBuilder( |
There was a problem hiding this 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.
There was a problem hiding this 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.
| // 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) |
There was a problem hiding this 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.
There was a problem hiding this 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 theDynamicPruningExpressionnode. - Added a
columnarSupportMode()assertion per scan type:SUPPORTEDforKyuubiOrcScan/KyuubiParquetScan,UNSUPPORTEDforHiveScan.
| // 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()) |
There was a problem hiding this 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.
There was a problem hiding this comment.
Kept the catalog-snapshot direction. Added:
- A note in
KyuubiParquetScanBuilder.hadoopConf/KyuubiOrcScanBuilder.hadoopConfscaladoc. - A guard test in
HiveCatalogSuite, assertingfieldId.read.enabled/ignoreMissingSET mid-session do not reach the resultinghadoopConf.
| <phase>generate-sources</phase> | ||
| <configuration> | ||
| <sources> | ||
| <source>src/main/${kshc.columnar.source.dir}</source> |
There was a problem hiding this 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.
There was a problem hiding this comment.
Resolved by rebase. The PR is now based on latest master which drops Spark 3.3 / 3.4 support (#7631).
|
|
||
| private val isCaseSensitive = sparkSession.sessionState.conf.caseSensitiveAnalysis | ||
|
|
||
| override def filterAttributes(): Array[NamedReference] = { |
There was a problem hiding this 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.
There was a problem hiding this comment.
Fixed in both KyuubiParquetScan.filterAttributes and KyuubiOrcScan.filterAttributes.
| // 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) |
There was a problem hiding this 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.
There was a problem hiding this comment.
Resolved by rebase. The PR is now based on latest master which drops Spark 3.3 / 3.4 support (#7631).
| OrcUtils.supportColumnarReads( | ||
| s.dataType, | ||
| sqlConf.orcVectorizedReaderNestedColumnEnabled)) | ||
| if (supportsColumnar) Scan.ColumnarSupportMode.SUPPORTED |
There was a problem hiding this 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.
There was a problem hiding this comment.
Added to scaladoc of columnarSupportMode() in both scans.
| catalogTable: CatalogTable, | ||
| hiveTableCatalog: HiveTableCatalog) | ||
| extends FileScanBuilder(sparkSession, fileIndex, dataSchema) | ||
| with SupportsPushDownAggregates { |
There was a problem hiding this 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.
There was a problem hiding this comment.
Updated the class-level scaladoc on KyuubiParquetScanBuilder to explicitly note the SupportsPushDownVariantExtractions gap on Spark 4.1+, dropping the misleading "semantically equivalent" wording.
| val enumValueCls = Class.forName("scala.Enumeration$Value") | ||
| DynConstructors.builder() | ||
| .impl( | ||
| "org.apache.spark.sql.catalyst.util.RebaseDateTime$RebaseSpec", |
There was a problem hiding this 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].
There was a problem hiding this comment.
Resolved by rebase. The PR is now based on latest master which drops Spark 3.3 / 3.4 support (#7631).
|
|
||
| override def createReaderFactory(): PartitionReaderFactory = inner.createReaderFactory() | ||
|
|
||
| override def equals(obj: Any): Boolean = obj match { |
There was a problem hiding this 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.
There was a problem hiding this comment.
Rewrote the class-level scaladoc (both KyuubiParquetScan and KyuubiOrcScan) to match the implementation.
|
I opened #7631 to drop support for Spark 3.4 and 3.5, so we don't need to worry about the |
|
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 |
| partitionFilters: Seq[Expression], | ||
| dataFilters: Seq[Expression]): ParquetScan = { | ||
| if (variantExtractionCls != null) { | ||
| // Spark 4.1+ |
There was a problem hiding this comment.
move the comment and link the JIRA ticket .impl // SPARK-XXXXX (4.1.0) do sth
There was a problem hiding this comment.
Fixed, use impl( // SPARK-53880 / SPARK-54656 (4.1.0): adds trailing Array[VariantExtraction]
| classOf[Seq[Expression]], | ||
| classOf[Seq[Expression]], | ||
| emptyVariantExtractions.getClass) | ||
| .buildChecked() |
There was a problem hiding this comment.
it resolves constructors with DynConstructors.builder().buildChecked() on every call, including each sibling-scan planning. Cache the Ctor in a lazy val.
There was a problem hiding this comment.
Fixed, cache the Ctor in a lazy val.
| val pushDownInFilterThreshold = sqlConf.parquetFilterPushDownInFilterThreshold | ||
| val isCaseSensitive = sqlConf.caseSensitiveAnalysis | ||
| val parquetSchema = new SparkToParquetSchemaConverter(sqlConf).convert(readDataSchema) | ||
| val rebaseSpec = RebaseSpec(LegacyBehaviorPolicy.CORRECTED) |
There was a problem hiding this comment.
why should we hardcode CORRECTED here?
There was a problem hiding this comment.
Following the same convention as Spark's own ParquetScanBuilder.pushDataFilters — 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.
| case s: KyuubiOrcScan => s.hadoopConf | ||
| case other => fail(s"unexpected scan type: ${other.getClass.getName}") | ||
| } | ||
| assert(hadoopConf.get(fieldIdRead) != "true") |
There was a problem hiding this 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?
There was a problem hiding this 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.
|
|
||
| exec.scan match { | ||
| case _: KyuubiOrcScan | _: KyuubiParquetScan => | ||
| assert(exec.scan.columnarSupportMode() == Scan.ColumnarSupportMode.SUPPORTED) |
There was a problem hiding this 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.
There was a problem hiding this comment.
Added four tests in DynamicPartitionPruningSuite
- parquet/orc vectorizedReader off →
UNSUPPORTED. - parquet/orc all-partitions-pruned via DPP →
planInputPartitions().isEmptythenSUPPORTED.
3 participants
Why are the changes needed?
Part 2 of 2 to add KSHC support for dynamic partition pruning (DPP). See #6943.
HiveScanfor non-Parquet/ORC tables.ParquetScan/ORCScanfor Parquet/ORC tables.How was this patch tested?
1. UT & TPC-DS benchmark
2. ORC benchmark
DPP trigger was detected by matching
runtime partition filterin the driver logs.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
DPP trigger was detected by matching
runtime partition filterin the driver logs.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 BYclause does not totally order the output. The single multiset diff (q39) is sub-ULP floating-point rounding instddev-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.
6. Known tradeoff: native engine offload
KyuubiParquetScan/KyuubiOrcScanwrap Spark's built-inParquetScan/OrcScanrather than subclass them, so DPP can be layered on without touching Spark internals. Native engines like Gluten and Comet identify the file format by scan class name, so KSHC-converted Parquet/ORC tables silently fall back to the JVM read path. Before this PR, KSHC-converted tables went through vanillaParquetScan/OrcScanand were offloadable; after this PR they are not.Was this patch authored or co-authored using generative AI tooling?
Assisted-by: Claude Opus 4.7