use filter, don't assert at all if no parquet encryptions requried

Author: EmilyMatt

spark/src/main/scala/org/apache/spark/sql/comet/operators.scala

@@ -217,24 +217,14 @@ abstract class CometNativeExec extends CometExec {
         // TODO: support native metrics for all operators.
         val nativeMetrics = CometMetricNode.fromCometPlan(this)
 
+        // Go over all all the native scans, in order to see if they need encryption options.
         // For each relation in a CometNativeScan generate a hadoopConf,
         // for each file path in a relation associate with hadoopConf
         // This is done per native plan, so only count scans until a comet input is reached.
-        val cometNativeScans = mutable.ArrayBuffer.empty[CometNativeScanExec]
+        val encryptionOptions =
+          mutable.ArrayBuffer.empty[(Broadcast[SerializableConfiguration], Seq[String])]
         foreachUntilCometInput(this) {
-          case scan: CometNativeScanExec => cometNativeScans += scan
-          case _ => // no-op
-        }
-        assert(
-          cometNativeScans.size <= 1,
-          "We expect one native scan in a Comet plan since we will broadcast one hadoopConf.")
-        // If this assumption changes in the future, you can look at the commit history of #2447
-        // to see how there used to be a map of relations to broadcasted confs in case multiple
-        // relations in a single plan. The example that came up was UNION. See discussion at:
-        // https://github.com/apache/datafusion-comet/pull/2447#discussion_r2406118264
-        val (broadcastedHadoopConfForEncryption, encryptedFilePaths) =
-          cometNativeScans.headOption.fold(
-            (None: Option[Broadcast[SerializableConfiguration]], Seq.empty[String])) { scan =>
+          case scan: CometNativeScanExec =>
             // This creates a hadoopConf that brings in any SQLConf "spark.hadoop.*" configs and
             // per-relation configs since different tables might have different decryption
             // properties.
@@ -246,10 +236,25 @@ abstract class CometNativeExec extends CometExec {
               val broadcastedConf =
                 scan.relation.sparkSession.sparkContext
                   .broadcast(new SerializableConfiguration(hadoopConf))
-              (Some(broadcastedConf), scan.relation.inputFiles.toSeq)
-            } else {
-              (None, Seq.empty)
+
+              val optsTuple: (Broadcast[SerializableConfiguration], Seq[String]) =
+                (broadcastedConf, scan.relation.inputFiles.toSeq)
+              encryptionOptions += optsTuple
             }
+          case _ => // no-op
+        }
+        assert(
+          encryptionOptions.size <= 1,
+          "We expect one native scan that requires encryption reading in a Comet plan," +
+            " since we will broadcast one hadoopConf.")
+        // If this assumption changes in the future, you can look at the commit history of #2447
+        // to see how there used to be a map of relations to broadcasted confs in case multiple
+        // relations in a single plan. The example that came up was UNION. See discussion at:
+        // https://github.com/apache/datafusion-comet/pull/2447#discussion_r2406118264
+        val (broadcastedHadoopConfForEncryption, encryptedFilePaths) =
+          encryptionOptions.headOption match {
+            case Some((conf, paths)) => (Some(conf), paths)
+            case None => (None, Seq.empty)
           }
 
         def createCometExecIter(

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

@@ -398,12 +398,12 @@ class ParquetEncryptionITCase extends CometTestBase with SQLTestUtils {
         // If we add another comet aggregate after the union, we see the need for the
         // foreachUntilCometInput() in operator.scala
         // as we would error on multiple native scan execs despite no longer being in the same plan at all
-        val aggDf = unionDF.agg(functions.sum("id"))
+        val filterDf = unionDF.filter(functions.col("id") % 2 === 0)
 
         if (CometConf.COMET_ENABLED.get(conf)) {
-          checkSparkAnswerAndOperator(aggDf)
+          checkSparkAnswerAndOperator(filterDf)
         } else {
-          checkSparkAnswer(aggDf)
+          checkSparkAnswer(filterDf)
         }
       }
     }