remove CometScanExec.scala, CometBatchScanExec.scala, CometScanRule.scala, CometExecRule.scala, and related stuff

Author: mbutrovich

spark/src/main/scala/org/apache/comet/CometConf.scala

@@ -93,17 +93,6 @@ object CometConf extends ShimCometConf {
     .booleanConf
     .createWithEnvVarOrDefault("ENABLE_COMET", true)
 
-  val COMET_USE_PLANNER: ConfigEntry[Boolean] = conf("spark.comet.planner.enabled")
-    .category(CATEGORY_EXEC)
-    .doc(
-      "When true, Comet registers the single-rule CometPlanner in place of the legacy " +
-        "CometScanRule + CometExecRule pair. Default true. Flip to false to run the legacy " +
-        "path as a rollback. The choice is evaluated once at session extension injection, so " +
-        "changes after session creation do not switch rules. The selected rules assert on this " +
-        "flag at entry to surface configuration drift early.")
-    .booleanConf
-    .createWithDefault(true)
-
   val COMET_NATIVE_SCAN_ENABLED: ConfigEntry[Boolean] = conf("spark.comet.scan.enabled")
     .category(CATEGORY_TESTING)
     .doc("Whether to enable native scans. Intended for use in Comet's own test suites to " +

spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala

@@ -33,24 +33,21 @@ import org.apache.spark.sql.internal.SQLConf
 
 import org.apache.comet.CometConf._
 import org.apache.comet.planner.CometPlanner
-import org.apache.comet.rules.{CometExecRule, CometPlanAdaptiveDynamicPruningFilters, CometReuseSubquery, CometScanRule, CometShuffleRule, CometSpark34AqeDppFallbackRule, EliminateRedundantTransitions}
+import org.apache.comet.rules.{CometPlanAdaptiveDynamicPruningFilters, CometReuseSubquery, EliminateRedundantTransitions}
 import org.apache.comet.shims.ShimCometSparkSessionExtensions
 
 /**
  * CometDriverPlugin will register an instance of this class with Spark.
  *
- * Comet rules are injected into Spark's rule pipeline at several extension points. The execution
- * order differs between AQE and non-AQE paths:
+ * Comet rules are injected into Spark's rule pipeline at several extension points:
  *
  * Non-AQE (QueryExecution.preparations):
  * {{{
  *   1. PlanDynamicPruningFilters    -- Spark creates non-AQE DPP (SubqueryBroadcastExec)
  *   2. PlanSubqueries               -- Spark creates SubqueryExec for scalar subqueries
  *   3. EnsureRequirements            -- Spark inserts shuffles/sorts
  *   4. ApplyColumnarRulesAndInsertTransitions:
- *      a. preColumnarTransitions:   CometScanRule, CometExecRule
- *         - CometExecRule.convertSubqueryBroadcasts converts SubqueryBroadcastExec to
- *           CometSubqueryBroadcastExec for exchange reuse with Comet broadcasts
+ *      a. preColumnarTransitions:   CometPlanner
  *      b. insertTransitions:        ColumnarToRow/RowToColumnar added
  *      c. postColumnarTransitions:  EliminateRedundantTransitions
  *   5. ReuseExchangeAndSubquery     -- Spark deduplicates subqueries (sees Comet nodes)
@@ -60,28 +57,27 @@ import org.apache.comet.shims.ShimCometSparkSessionExtensions
  * {{{
  *   Initial plan:
  *     PlanAdaptiveSubqueries:       creates SubqueryAdaptiveBroadcastExec (SAB) for AQE DPP
- *     queryStagePreparationRules:   CometScanRule, CometExecRule
- *       - CometExecRule.convertSubqueryBroadcasts wraps SABs in
- *         CometSubqueryAdaptiveBroadcastExec to prevent Spark's
- *         PlanAdaptiveDynamicPruningFilters from replacing DPP with Literal.TrueLiteral
+ *     queryStagePreparationRules:   CometPlanner
+ *       - prePass `Spark34DppFallbackPrePass` is a no-op on 3.5+
  *
  *   Per stage (optimizeQueryStage + postStageCreationRules):
  *     1. queryStageOptimizerRules:
- *        a. PlanAdaptiveDynamicPruningFilters (Spark) -- skips wrapped SABs
+ *        a. PlanAdaptiveDynamicPruningFilters (Spark) -- skips planner-wrapped SABs
  *        b. ReuseAdaptiveSubquery (Spark)
  *        c. CometPlanAdaptiveDynamicPruningFilters   -- converts wrapped SABs to
  *           CometSubqueryBroadcastExec with BroadcastQueryStageExec for broadcast reuse
  *        d. CometReuseSubquery                       -- deduplicates converted subqueries
  *     2. postStageCreationRules -> ApplyColumnarRulesAndInsertTransitions:
- *        a. preColumnarTransitions: CometScanRule, CometExecRule (no-ops, already converted)
+ *        a. preColumnarTransitions: CometPlanner (no-op, already converted)
  *        b. insertTransitions
  *        c. postColumnarTransitions: EliminateRedundantTransitions
  * }}}
  *
- * On Spark 3.4, injectQueryStageOptimizerRule is unavailable. CometExecRule does not wrap SABs,
- * and CometPlanAdaptiveDynamicPruningFilters/CometReuseSubquery are not registered. AQE DPP scans
- * fall back to Spark so that Spark's PlanAdaptiveDynamicPruningFilters handles them natively
- * (with DPP).
+ * On Spark 3.4, `injectQueryStageOptimizerRule` is unavailable.
+ * `CometPlanAdaptiveDynamicPruningFilters` and `CometReuseSubquery` are not registered. The
+ * planner's `Spark34DppFallbackPrePass` arranges for Spark's own
+ * `PlanAdaptiveDynamicPruningFilters` to succeed by tagging specific nodes with
+ * `CometTags.SKIP_COMET`, which Phase 2 honors as `Fallback`.
  */
 class CometSparkSessionExtensions
     extends (SparkSessionExtensions => Unit)
@@ -90,33 +86,17 @@ class CometSparkSessionExtensions
   override def apply(extensions: SparkSessionExtensions): Unit = {
     extensions.injectColumnar { session => CometScanColumnar(session) }
     extensions.injectColumnar { session => CometExecColumnar(session) }
-    // Pre-3.5 only: tag AQE DPP regions so the conversion rules below leave them Spark-native.
-    // Registered before CometScanRule/CometExecRule so tags are in place when conversion runs.
-    // No-op on Spark 3.5+; see CometSpark34AqeDppFallbackRule's class docstring.
-    injectPreSpark35QueryStagePrepRuleShim(extensions, CometSpark34AqeDppFallbackRule)
-    if (CometConf.COMET_USE_PLANNER.get()) {
-      extensions.injectQueryStagePrepRule { session => CometPlanner(session) }
-      // Covers the legacy `exec=off + shuffle=on` mode that CometPlanner intentionally
-      // leaves untouched. See CometShuffleRule for the removal plan.
-      extensions.injectQueryStagePrepRule { session => CometShuffleRule(session) }
-    } else {
-      extensions.injectQueryStagePrepRule { session => CometScanRule(session) }
-      extensions.injectQueryStagePrepRule { session => CometExecRule(session) }
-    }
+    extensions.injectQueryStagePrepRule { session => CometPlanner(session) }
     injectQueryStageOptimizerRuleShim(extensions, CometPlanAdaptiveDynamicPruningFilters)
     injectQueryStageOptimizerRuleShim(extensions, CometReuseSubquery)
   }
 
   case class CometScanColumnar(session: SparkSession) extends ColumnarRule {
-    override def preColumnarTransitions: Rule[SparkPlan] =
-      if (CometConf.COMET_USE_PLANNER.get()) CometPlanner(session)
-      else CometScanRule(session)
+    override def preColumnarTransitions: Rule[SparkPlan] = CometPlanner(session)
   }
 
   case class CometExecColumnar(session: SparkSession) extends ColumnarRule {
-    override def preColumnarTransitions: Rule[SparkPlan] =
-      if (CometConf.COMET_USE_PLANNER.get()) CometShuffleRule(session)
-      else CometExecRule(session)
+    override def preColumnarTransitions: Rule[SparkPlan] = CometPlanner(session)
 
     override def postColumnarTransitions: Rule[SparkPlan] =
       EliminateRedundantTransitions(session)
@@ -181,7 +161,9 @@ object CometSparkSessionExtensions extends Logging {
   }
 
   def isCometScan(op: SparkPlan): Boolean = {
-    op.isInstanceOf[CometBatchScanExec] || op.isInstanceOf[CometScanExec]
+    op.isInstanceOf[CometNativeScanExec] ||
+    op.isInstanceOf[CometIcebergNativeScanExec] ||
+    op.isInstanceOf[CometCsvNativeScanExec]
   }
 
   def isSpark35Plus: Boolean = {

spark/src/main/scala/org/apache/comet/planner/CometPlanner.scala

@@ -32,7 +32,6 @@ import org.apache.comet.CometConf
 import org.apache.comet.CometSparkSessionExtensions.isCometLoaded
 import org.apache.comet.planner.phases.{NormalizePrePass, Phase1LikelyComet, Phase2Decision, Phase3Emit, Spark34DppFallbackPrePass, SubqueryBroadcastRewrite}
 import org.apache.comet.planner.tags.CometTags
-import org.apache.comet.rules.RewriteJoin
 
 /**
  * Single-pass compiler from Spark physical plans to Comet-accelerated plans. Replaces the
@@ -63,12 +62,6 @@ case class CometPlanner(session: SparkSession) extends Rule[SparkPlan] with Logg
       logDebug("CometPlanner skip: Comet extension not loaded")
       return plan
     }
-    assert(
-      CometConf.COMET_USE_PLANNER.get(conf),
-      s"CometPlanner ran while ${CometConf.COMET_USE_PLANNER.key}=false. The legacy " +
-        "CometScanRule + CometExecRule should be the sole rules on this path. Either " +
-        "COMET_USE_PLANNER was flipped after session creation or CometPlanner was registered " +
-        "by mistake.")
 
     // Comet exec globally disabled OR root already converted (AQE re-entry): skip phase 1/2/3
     // but still run convertBlocks. AQE re-planning can graft a previously-emitted CometNativeExec
@@ -345,10 +338,6 @@ case class CometPlanner(session: SparkSession) extends Rule[SparkPlan] with Logg
    */
   private def checkPostEmitInvariants(plan: SparkPlan): Unit = {
     plan.foreach { node =>
-      assert(
-        !node.getClass.getName.contains("CometBatchScanExec"),
-        "CometBatchScanExec found in emitted plan. CometPlanner should emit " +
-          s"CometIcebergNativeScanExec / CometCsvNativeScanExec directly. node=$node")
       assert(
         !node.getClass.getName.endsWith(".CometSinkPlaceHolder") &&
           !node.getClass.getName.endsWith(".CometScanWrapper"),

spark/src/main/scala/org/apache/comet/planner/FileFormats.scala

@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.comet.planner
+
+import org.apache.spark.sql.execution.datasources.FileFormat
+import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
+
+/**
+ * File-format predicates used by the V1 scan gate. Kept as a standalone object so the "strict
+ * class equality" check is unit-testable. Lives outside `V1ScanGate` because that class scopes a
+ * per-scan classification call; this is a pure type check.
+ */
+object FileFormats {
+
+  /**
+   * True iff `fileFormat` is exactly Spark's built-in `ParquetFileFormat`. Subclasses (Delta
+   * Lake's variant, etc.) are rejected — they may extend Parquet but rely on extra read-time
+   * machinery the native scan doesn't implement.
+   */
+  def isParquetFileFormat(fileFormat: FileFormat): Boolean =
+    fileFormat.getClass.equals(classOf[ParquetFileFormat])
+}

spark/src/main/scala/org/apache/comet/planner/OperatorRegistry.scala

@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.comet.planner
+
+import org.apache.spark.sql.comet._
+import org.apache.spark.sql.comet.execution.shuffle.CometShuffleExchangeExec
+import org.apache.spark.sql.execution._
+import org.apache.spark.sql.execution.aggregate.{HashAggregateExec, ObjectHashAggregateExec}
+import org.apache.spark.sql.execution.command.DataWritingCommandExec
+import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec, ShuffledHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.execution.window.WindowExec
+
+import org.apache.comet.serde.CometOperatorSerde
+import org.apache.comet.serde.operator._
+
+/**
+ * Map from Spark operator class to its CometOperatorSerde. The planner's Phase 1/2/3 use this to
+ * look up the serde for a given node and drive prediction, decision, and emission via the serde's
+ * generic interface.
+ *
+ * Two categories preserved from the legacy split for documentation only; the planner treats them
+ * uniformly via `allExecs`:
+ *
+ *   - `nativeExecs`: serdes that produce a fully native CometNativeExec at emit time.
+ *   - `sinks`: serdes whose native plan is a `Scan` operator (JVM-orchestrated wrappers like
+ *     `CometCollectLimitExec`, `CometUnionExec`).
+ */
+object OperatorRegistry {
+
+  val nativeExecs: Map[Class[_ <: SparkPlan], CometOperatorSerde[_]] =
+    Map(
+      classOf[ProjectExec] -> CometProjectExec,
+      classOf[FilterExec] -> CometFilterExec,
+      classOf[LocalLimitExec] -> CometLocalLimitExec,
+      classOf[GlobalLimitExec] -> CometGlobalLimitExec,
+      classOf[ExpandExec] -> CometExpandExec,
+      classOf[GenerateExec] -> CometExplodeExec,
+      classOf[HashAggregateExec] -> CometHashAggregateExec,
+      classOf[ObjectHashAggregateExec] -> CometObjectHashAggregateExec,
+      classOf[BroadcastHashJoinExec] -> CometBroadcastHashJoinExec,
+      classOf[ShuffledHashJoinExec] -> CometHashJoinExec,
+      classOf[SortMergeJoinExec] -> CometSortMergeJoinExec,
+      classOf[SortExec] -> CometSortExec,
+      classOf[LocalTableScanExec] -> CometLocalTableScanExec,
+      classOf[WindowExec] -> CometWindowExec,
+      classOf[DataWritingCommandExec] -> CometDataWritingCommand)
+
+  val sinks: Map[Class[_ <: SparkPlan], CometOperatorSerde[_]] =
+    Map(
+      classOf[CoalesceExec] -> CometCoalesceExec,
+      classOf[CollectLimitExec] -> CometCollectLimitExec,
+      classOf[TakeOrderedAndProjectExec] -> CometTakeOrderedAndProjectExec,
+      classOf[UnionExec] -> CometUnionExec)
+
+  val allExecs: Map[Class[_ <: SparkPlan], CometOperatorSerde[_]] = nativeExecs ++ sinks
+}

…org/apache/comet/rules/RewriteJoin.scala …g/apache/comet/planner/RewriteJoin.scalaspark/src/main/scala/org/apache/comet/rules/RewriteJoin.scala renamed to spark/src/main/scala/org/apache/comet/planner/RewriteJoin.scala spark/src/main/scala/org/apache/comet/rules/RewriteJoin.scala renamed to spark/src/main/scala/org/apache/comet/planner/RewriteJoin.scala

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.apache.comet.rules
+package org.apache.comet.planner
 
 import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, BuildSide, JoinSelectionHelper}
 import org.apache.spark.sql.catalyst.plans.LeftSemi

spark/src/main/scala/org/apache/comet/planner/gates/CometScanTypeChecker.scala

@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.comet.planner.gates
+
+import scala.collection.mutable.ListBuffer
+
+import org.apache.spark.sql.types.{DataType, ShortType, StructType}
+
+import org.apache.comet.{CometConf, DataTypeSupport}
+import org.apache.comet.serde.QueryPlanSerde.isStringCollationType
+import org.apache.comet.shims.CometTypeShim
+
+/**
+ * Schema-level support check used by both V1 and V2 native scan gates. Moved out of the deleted
+ * `CometScanRule` companion. Rejects types known to be unsupported by the native scan path
+ * (unsigned small ints when the safety check is on, collated strings, variant-shredded structs,
+ * empty structs) and delegates everything else to `DataTypeSupport`.
+ */
+case class CometScanTypeChecker() extends DataTypeSupport with CometTypeShim {
+
+  override def isTypeSupported(
+      dt: DataType,
+      name: String,
+      fallbackReasons: ListBuffer[String]): Boolean = {
+    dt match {
+      case ShortType if CometConf.COMET_PARQUET_UNSIGNED_SMALL_INT_CHECK.get() =>
+        fallbackReasons += "Native Parquet scan may not handle unsigned UINT_8 correctly for " +
+          s"$dt. Set ${CometConf.COMET_PARQUET_UNSIGNED_SMALL_INT_CHECK.key}=false to allow " +
+          "native execution if your data does not contain unsigned small integers. " +
+          CometConf.COMPAT_GUIDE
+        false
+      case dt if isStringCollationType(dt) =>
+        // we don't need specific support for collation in scans, but this is a convenient
+        // place to force the whole query to fall back to Spark for now
+        false
+      case s: StructType if isVariantStruct(s) =>
+        // Spark 4.0's PushVariantIntoScan rewrites a VariantType column into a struct of typed
+        // fields plus per-field VariantMetadata, expecting the scan to honor Parquet variant
+        // shredding semantics. Comet's native scan does not, so fall back to Spark.
+        fallbackReasons +=
+          s"Unsupported $name of type VariantType (shredded; not supported by native scan)"
+        false
+      case s: StructType if s.fields.isEmpty =>
+        false
+      case _ =>
+        super.isTypeSupported(dt, name, fallbackReasons)
+    }
+  }
+}

spark/src/main/scala/org/apache/comet/planner/gates/V1ScanGate.scala

@@ -26,15 +26,14 @@ import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.catalyst.expressions.{Expression, GenericInternalRow}
 import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
 import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns.getExistenceDefaultValues
-import org.apache.spark.sql.comet.CometScanExec
 import org.apache.spark.sql.execution.{FileSourceScanExec, InSubqueryExec, SubqueryAdaptiveBroadcastExec}
 import org.apache.spark.sql.execution.datasources.HadoopFsRelation
 import org.apache.spark.sql.internal.SQLConf
 
 import org.apache.comet.CometConf._
 import org.apache.comet.CometSparkSessionExtensions.isSpark35Plus
 import org.apache.comet.parquet.CometParquetUtils.{encryptionEnabled, isEncryptionConfigSupported}
-import org.apache.comet.rules.CometScanTypeChecker
+import org.apache.comet.planner.FileFormats
 import org.apache.comet.serde.operator.CometNativeScan
 import org.apache.comet.shims.ShimFileFormat
 
@@ -82,7 +81,7 @@ object V1ScanGate extends Logging {
         return reject(s"Unsupported relation $other")
     }
 
-    if (!CometScanExec.isFileFormatSupported(r.fileFormat)) {
+    if (!FileFormats.isParquetFileFormat(r.fileFormat)) {
       return reject(s"Unsupported file format ${r.fileFormat}")
     }