fix: fallback to Spark scan if encryption is enabled (native_datafusion/native_iceberg_compat) (#1785)

* fix: fallback to Spark scan if encryption is enabled (native_datafusion/native_iceberg_compat)

Author: parthchandra

spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala

@@ -132,6 +132,16 @@ case class CometScanRule(session: SparkSession) extends Rule[SparkPlan] {
           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"
+          return withInfos(scanExec, fallbackReasons.toSet)
+        }
+
         val typeChecker = CometScanTypeChecker(scanImpl)
         val schemaSupported =
           typeChecker.isSchemaSupported(scanExec.requiredSchema, fallbackReasons)

spark/src/test/scala/org/apache/spark/sql/comet/ParquetEncryptionITCase.scala

@@ -49,8 +49,7 @@ 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") {
-    // https://github.com/apache/datafusion-comet/issues/1488
-    assume(CometConf.COMET_NATIVE_SCAN_IMPL.get() != CometConf.SCAN_NATIVE_ICEBERG_COMPAT)
+    assume(CometConf.COMET_NATIVE_SCAN_IMPL.get() == CometConf.SCAN_NATIVE_COMET)
 
     import testImplicits._
 
@@ -93,8 +92,7 @@ class ParquetEncryptionITCase extends CometTestBase with SQLTestUtils {
   }
 
   test("SPARK-37117: Can't read files in Parquet encryption external key material mode") {
-    // https://github.com/apache/datafusion-comet/issues/1488
-    assume(CometConf.COMET_NATIVE_SCAN_IMPL.get() != CometConf.SCAN_NATIVE_ICEBERG_COMPAT)
+    assume(CometConf.COMET_NATIVE_SCAN_IMPL.get() == CometConf.SCAN_NATIVE_COMET)
 
     import testImplicits._