This is an automated email from the ASF dual-hosted git repository.
mbutrovich pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new da515282a fix: Fall back to `native_comet` for encrypted Parquet scans
(#2250)
da515282a is described below
commit da515282a0499baae6895dcb68a870fbe2566ad6
Author: Andy Grove <agr...@apache.org>
AuthorDate: Thu Aug 28 11:58:21 2025 -0600
fix: Fall back to `native_comet` for encrypted Parquet scans (#2250)
---
.../scala/org/apache/comet/rules/CometScanRule.scala | 17 ++++++++++++-----
.../spark/sql/comet/ParquetEncryptionITCase.scala | 6 ++++--
2 files changed, 16 insertions(+), 7 deletions(-)
diff --git a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
index 2969e7689..7cd2df81e 100644
--- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
+++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
@@ -131,11 +131,22 @@ case class CometScanRule(session: SparkSession) extends
Rule[SparkPlan] with Com
return withInfos(scanExec, fallbackReasons.toSet)
}
+ val encryptionEnabled: Boolean =
+ conf.getConfString("parquet.crypto.factory.class", "").nonEmpty &&
+ conf.getConfString("parquet.encryption.kms.client.class",
"").nonEmpty
+
var scanImpl = COMET_NATIVE_SCAN_IMPL.get()
// if scan is auto then pick the best available scan
if (scanImpl == SCAN_AUTO) {
- scanImpl = selectScan(scanExec, r.partitionSchema)
+ if (encryptionEnabled) {
+ logInfo(
+ s"Auto scan mode falling back to $SCAN_NATIVE_COMET because " +
+ s"$SCAN_NATIVE_ICEBERG_COMPAT does not support reading
encrypted Parquet files")
+ scanImpl = SCAN_NATIVE_COMET
+ } else {
+ scanImpl = selectScan(scanExec, r.partitionSchema)
+ }
}
if (scanImpl == SCAN_NATIVE_DATAFUSION && !COMET_EXEC_ENABLED.get()) {
@@ -182,10 +193,6 @@ case class CometScanRule(session: SparkSession) extends
Rule[SparkPlan] with Com
return withInfos(scanExec, fallbackReasons.toSet)
}
- val encryptionEnabled: Boolean =
- conf.getConfString("parquet.crypto.factory.class", "").nonEmpty &&
- conf.getConfString("parquet.encryption.kms.client.class",
"").nonEmpty
-
if (scanImpl != CometConf.SCAN_NATIVE_COMET && encryptionEnabled) {
fallbackReasons +=
"Full native scan disabled because encryption is not supported"
diff --git
a/spark/src/test/scala/org/apache/spark/sql/comet/ParquetEncryptionITCase.scala
b/spark/src/test/scala/org/apache/spark/sql/comet/ParquetEncryptionITCase.scala
index fd5ecef96..8d2c3db72 100644
---
a/spark/src/test/scala/org/apache/spark/sql/comet/ParquetEncryptionITCase.scala
+++
b/spark/src/test/scala/org/apache/spark/sql/comet/ParquetEncryptionITCase.scala
@@ -49,7 +49,8 @@ class ParquetEncryptionITCase extends CometTestBase with
SQLTestUtils {
private val key2 =
encoder.encodeToString("1234567890123451".getBytes(StandardCharsets.UTF_8))
test("SPARK-34990: Write and read an encrypted parquet") {
- assume(CometConf.COMET_NATIVE_SCAN_IMPL.get() ==
CometConf.SCAN_NATIVE_COMET)
+ assume(CometConf.COMET_NATIVE_SCAN_IMPL.get() !=
CometConf.SCAN_NATIVE_DATAFUSION)
+ assume(CometConf.COMET_NATIVE_SCAN_IMPL.get() !=
CometConf.SCAN_NATIVE_ICEBERG_COMPAT)
import testImplicits._
@@ -92,7 +93,8 @@ class ParquetEncryptionITCase extends CometTestBase with
SQLTestUtils {
}
test("SPARK-37117: Can't read files in Parquet encryption external key
material mode") {
- assume(CometConf.COMET_NATIVE_SCAN_IMPL.get() ==
CometConf.SCAN_NATIVE_COMET)
+ assume(CometConf.COMET_NATIVE_SCAN_IMPL.get() !=
CometConf.SCAN_NATIVE_DATAFUSION)
+ assume(CometConf.COMET_NATIVE_SCAN_IMPL.get() !=
CometConf.SCAN_NATIVE_ICEBERG_COMPAT)
import testImplicits._
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@datafusion.apache.org
For additional commands, e-mail: commits-h...@datafusion.apache.org
