Minor refactor for readability. spotless:apply

Author: mbutrovich

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

@@ -152,47 +152,6 @@ private[spark] class CometExecRDD(
 
 object CometExecRDD {
 
-  /**
-   * Creates an RDD for standalone Iceberg scan (no parent native operators).
-   */
-  def apply(
-      sc: SparkContext,
-      commonData: Array[Byte],
-      perPartitionData: Array[Array[Byte]],
-      numOutputCols: Int,
-      nativeMetrics: CometMetricNode): CometExecRDD = {
-
-    // Standalone mode needs a placeholder plan for PlanDataInjector to fill in.
-    // PlanDataInjector correlates common/partition data by key (metadata_location for Iceberg).
-    val common = OperatorOuterClass.IcebergScanCommon.parseFrom(commonData)
-    val metadataLocation = common.getMetadataLocation
-
-    val placeholderCommon = OperatorOuterClass.IcebergScanCommon
-      .newBuilder()
-      .setMetadataLocation(metadataLocation)
-      .build()
-    val placeholderScan = OperatorOuterClass.IcebergScan
-      .newBuilder()
-      .setCommon(placeholderCommon)
-      .build()
-    val placeholderPlan = OperatorOuterClass.Operator
-      .newBuilder()
-      .setIcebergScan(placeholderScan)
-      .build()
-      .toByteArray
-
-    new CometExecRDD(
-      sc,
-      inputRDDs = Seq.empty,
-      commonByKey = Map(metadataLocation -> commonData),
-      perPartitionByKey = Map(metadataLocation -> perPartitionData),
-      serializedPlan = placeholderPlan,
-      defaultNumPartitions = perPartitionData.length,
-      numOutputCols = numOutputCols,
-      nativeMetrics = nativeMetrics,
-      subqueries = Seq.empty)
-  }
-
   /**
    * Creates an RDD for native execution with optional per-partition planning data.
    */

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

@@ -237,7 +237,8 @@ case class CometIcebergNativeScanExec(
     if (originalPlan == null) {
       Seq.empty
     } else {
-      // Trigger serializedPartitionData to ensure Iceberg planning has run and metrics are populated
+      // Trigger serializedPartitionData to ensure Iceberg planning has run and
+      // metrics are populated
       val _ = serializedPartitionData
 
       originalPlan.metrics
@@ -294,7 +295,17 @@ case class CometIcebergNativeScanExec(
   /** Executes using CometExecRDD - planning data is computed lazily on first access. */
   override def doExecuteColumnar(): RDD[ColumnarBatch] = {
     val nativeMetrics = CometMetricNode.fromCometPlan(this)
-    CometExecRDD(sparkContext, commonData, perPartitionData, output.length, nativeMetrics)
+    val serializedPlan = CometExec.serializeNativePlan(nativeOp)
+    CometExecRDD(
+      sparkContext,
+      inputRDDs = Seq.empty,
+      commonByKey = Map(metadataLocation -> commonData),
+      perPartitionByKey = Map(metadataLocation -> perPartitionData),
+      serializedPlan = serializedPlan,
+      numPartitions = perPartitionData.length,
+      numOutputCols = output.length,
+      nativeMetrics = nativeMetrics,
+      subqueries = Seq.empty)
   }
 
   /**