fix: Fall back to native_comet for encrypted Parquet scans (apache#2250)

Author: andygrove

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"

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._