feat(contrib): SPI surface additions for contribs that need core helpers

Format-agnostic surface additions PR2's Delta port (and any future
file-format contrib) needs without duplicating ~600 lines of
CometScanRule's existing schema-check / encryption-gate / marker-dispatch
helpers. No specific format is named anywhere in core; contribs register
their own tags through the SPI.

1. CometScanRule.isSchemaSupported widened from private to private[comet]
   so contrib extensions under org.apache.comet.contrib.* can reuse the
   same schema-check + fallback-reason emission rather than duplicating
   the 25-line body. Zero behaviour change for in-tree callers.

2. CometOperatorSerdeExtension.matchOperator new default-Some-or-None
   method for predicate-based serde dispatch. The class-keyed `serdes`
   map can't disambiguate a marker pattern like
   `CometScanExec(scanImpl="<contrib-specific-tag>")` (the class is
   shared with core's generic CometScanExec). Contribs using such
   markers override matchOperator. Backwards compatible: existing
   contribs that only populate `serdes` see no change. `serdes` now
   defaults to Map.empty so contribs that ONLY use matchOperator don't
   need to override both.

3. CometExecRule three-step dispatch: allExecs (core class map) ->
   mergedSerdes (contrib class map) -> matchOperator iteration (contrib
   predicate). First Some wins; multiple extensions' matchOperator
   results are tried in registration order.

4. CometOperatorSerdeExtension.nativeParquetScanImpls new
   default-Set.empty method. Contribs that use the CometScanExec marker
   pattern AND go through Comet's tuned ParquetSource declare their
   scanImpl tag(s) here. CometScanExec.supportedDataFilters consults the
   merged set (via CometExtensionRegistry.nativeParquetScanImpls) to
   decide whether to apply native-parquet filter exclusions. Core no
   longer needs to hard-code any contrib's tag name.

5. CometExtensionRegistry.nativeParquetScanImpls publishes the merged
   tag set at load() time. Populated/reset alongside mergedSerdesCache
   under the same monitor.

Contributor guide updated with the matchOperator + nativeParquetScanImpls
patterns and explicit guidance that contribs define their own scanImpl
strings in their own code -- core's CometConf only carries
SCAN_NATIVE_DATAFUSION / SCAN_NATIVE_ICEBERG_COMPAT for core's own
variants.

Verified
- cargo check (default features): green.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: schenksj

docs/source/contributor-guide/contrib-extensions.md

@@ -277,9 +277,10 @@ trait CometOperatorSerdeExtension {
 }
 ```
 
-Contribs that need a custom physical operator (e.g., a contrib-specific scan exec
-carrying contrib-private state) define their own `SparkPlan` subclass and register a
-serde keyed on the new class:
+Two dispatch shapes are supported:
+
+**Class-keyed** — the contrib defines its own `SparkPlan` subclass (typical for
+operator-style contribs):
 
 ```scala
 case class CometMyFormatScanExec(...) extends CometNativeExec { /* ... */ }
@@ -296,8 +297,42 @@ The merged map across all extensions is computed once at registry load time;
 contribs are logged as a warning at load — the convention is **one contrib defines a
 class, that contrib owns its serde**.
 
-Avoid relying on the legacy `scanImpl: String` tag pattern on a generic `CometScanExec`
-— the SPI dispatches by class, not by tag.
+**Predicate-keyed (marker-class with scanImpl tag)** — required when the contrib uses
+core's `CometScanExec` as a marker disambiguated by a `scanImpl` string. `CometScanExec`
+is a Scala case class shared with core, so two contribs marking different tag values
+on the same class would otherwise collide. Override `matchOperator` instead of (or in
+addition to) populating `serdes`, and declare your tag(s) via `nativeParquetScanImpls`
+if your scan goes through Comet's tuned ParquetSource:
+
+```scala
+class MyFormatSerdeExtension extends CometOperatorSerdeExtension {
+  override def name: String = "myformat"
+
+  // Your contrib's scanImpl marker. Pick a stable string; no central registry of these
+  // exists in core, but conventionally contribs use snake-case like "native_<name>_compat".
+  private val MyScanImpl = "native_myformat_compat"
+
+  override def matchOperator(op: SparkPlan): Option[CometOperatorSerde[_]] = op match {
+    case s: CometScanExec if s.scanImpl == MyScanImpl => Some(CometMyFormatScan)
+    case _ => None
+  }
+
+  // Tell core's CometScanExec.supportedDataFilters to apply DataFusion-style filter
+  // exclusions to this tag. Required when your scan goes through Comet's tuned
+  // ParquetSource (the same path SCAN_NATIVE_DATAFUSION uses).
+  override def nativeParquetScanImpls: Set[String] = Set(MyScanImpl)
+}
+```
+
+`CometExecRule` checks `matchOperator` only after the class-keyed `serdes` map misses,
+so the two patterns coexist. Multiple registered extensions' `matchOperator` calls are
+tried in registration order; the first `Some` wins.
+
+Core's CometConf defines `SCAN_NATIVE_DATAFUSION` / `SCAN_NATIVE_ICEBERG_COMPAT` for
+core's own scan variants. Contribs are expected to define their own scanImpl strings
+inside their own code (not in `CometConf`); registering via `nativeParquetScanImpls`
+is the SPI hook that lets `CometScanExec.supportedDataFilters` apply the right filter
+treatment without core needing to know the contrib's tag name.
 
 ##### `CometOperatorSerde[T <: SparkPlan]` contract
 

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

@@ -357,9 +357,20 @@ case class CometExecRule(session: SparkSession)
           // that aren't in `allExecs`, so this merge never overrides a core mapping in
           // practice; duplicate-class detection at load() time logs a warning if it
           // does happen.
+          // Three-step dispatch:
+          //   1. core's built-in class-keyed map (allExecs)
+          //   2. contrib serde-extensions' class-keyed map (mergedSerdes)
+          //   3. contrib serde-extensions' predicate-based matchOperator hook
+          //      (for marker-class patterns where one shared SparkPlan class --
+          //      e.g. CometScanExec -- is disambiguated by a runtime tag)
           val handler = allExecs
             .get(op.getClass)
             .orElse(CometExtensionRegistry.mergedSerdes.get(op.getClass))
+            .orElse {
+              CometExtensionRegistry.serdeExtensions.iterator
+                .flatMap(_.matchOperator(op))
+                .nextOption()
+            }
             .map(_.asInstanceOf[CometOperatorSerde[SparkPlan]])
           handler match {
             case Some(handler) =>

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

@@ -780,7 +780,9 @@ case class CometScanRule(session: SparkSession)
       case _ => false
     }
 
-  private def isSchemaSupported(
+  // private[comet] so contrib extensions (under org.apache.comet.contrib.*) can reuse
+  // the same schema check + fallback-reason emission rather than duplicating the body.
+  private[comet] def isSchemaSupported(
       scanExec: FileSourceScanExec,
       scanImpl: String,
       r: HadoopFsRelation): Boolean = {

spark/src/main/scala/org/apache/comet/spi/CometExtensionRegistry.scala

@@ -71,12 +71,14 @@ object CometExtensionRegistry extends Logging {
     val newScanExts = loadOne[CometScanRuleExtension]("CometScanRuleExtension")
     val newSerdeExts = loadOne[CometOperatorSerdeExtension]("CometOperatorSerdeExtension")
     val newMerged = newSerdeExts.flatMap(_.serdes).toMap
+    val newNativeParquetTags = newSerdeExts.flatMap(_.nativeParquetScanImpls).toSet
     // Publish the @volatile fields BEFORE flipping `loaded` so other threads either see
     // the empty defaults (and may re-enter -- benign, blocked by the monitor) or the
     // fully-populated state (and may skip -- also benign).
     scanExts = newScanExts
     serdeExts = newSerdeExts
     mergedSerdesCache = newMerged
+    nativeParquetScanImplsCache = newNativeParquetTags
     loaded.set(true)
     if (newScanExts.nonEmpty || newSerdeExts.nonEmpty) {
       logInfo(
@@ -114,6 +116,16 @@ object CometExtensionRegistry extends Logging {
     : Map[Class[_ <: org.apache.spark.sql.execution.SparkPlan],
       org.apache.comet.serde.CometOperatorSerde[_]] = Map.empty
 
+  /**
+   * Union of every registered extension's `nativeParquetScanImpls`. Consumed by
+   * `CometScanExec.supportedDataFilters` to decide whether the marker scan's filter set
+   * should get the same native-parquet exclusions as `SCAN_NATIVE_DATAFUSION`. Computed
+   * once at `load()` time; empty until `load()` has run.
+   */
+  def nativeParquetScanImpls: Set[String] = nativeParquetScanImplsCache
+
+  @volatile private var nativeParquetScanImplsCache: Set[String] = Set.empty
+
   /**
    * Log a warning when two registered contribs claim the same `Class[_ <: SparkPlan]` for serde
    * dispatch. The convention documented in `contrib-extensions.md` is that each contrib defines
@@ -162,6 +174,7 @@ object CometExtensionRegistry extends Logging {
     scanExts = Seq.empty
     serdeExts = Seq.empty
     mergedSerdesCache = Map.empty
+    nativeParquetScanImplsCache = Set.empty
   }
 
   private def loadOne[T](label: String)(implicit ct: scala.reflect.ClassTag[T]): Seq[T] = {

spark/src/main/scala/org/apache/comet/spi/CometOperatorSerdeExtension.scala

@@ -49,5 +49,44 @@ trait CometOperatorSerdeExtension {
    * Convention: each contrib's mapping should reference only classes the contrib itself defines,
    * so two contribs never claim ownership of the same operator class.
    */
-  def serdes: Map[Class[_ <: SparkPlan], CometOperatorSerde[_]]
+  def serdes: Map[Class[_ <: SparkPlan], CometOperatorSerde[_]] = Map.empty
+
+  /**
+   * Predicate-based dispatch hook for contribs whose serde key cannot be expressed as a unique
+   * `SparkPlan` class. The canonical case is the `CometScanExec` marker-with-`scanImpl`-tag
+   * pattern: a contrib's `CometScanRuleExtension.transformV1` returns
+   * `CometScanExec(scanExec, session, "native_delta_compat")`, but `CometScanExec` is a case
+   * class shared with core, so a class-keyed map can't disambiguate by the tag. The contrib
+   * overrides this method to inspect the plan and return its serde:
+   *
+   * {{{
+   *   override def matchOperator(op: SparkPlan): Option[CometOperatorSerde[_]] = op match {
+   *     case s: CometScanExec if s.scanImpl == CometConf.SCAN_NATIVE_DELTA_COMPAT =>
+   *       Some(CometDeltaNativeScan)
+   *     case _ => None
+   *   }
+   * }}}
+   *
+   * `CometExecRule` consults `matchOperator` only after the class-keyed `serdes` map misses, so
+   * contribs with a unique exec class never need to implement this. Multiple registered
+   * extensions' `matchOperator` returns are tried in registration order; the first `Some` wins.
+   */
+  def matchOperator(op: SparkPlan): Option[CometOperatorSerde[_]] = None
+
+  /**
+   * Declares which `scanImpl` string tags this contrib produces from
+   * `CometScanRuleExtension.transformV1` when using the `CometScanExec(marker, scanImpl=X)`
+   * pattern. Tags listed here get `CometScanExec.supportedDataFilters`'s native-parquet
+   * filter exclusions (drop dynamic pruning + IsNull/IsNotNull on ArrayType columns), the
+   * same treatment `SCAN_NATIVE_DATAFUSION` receives.
+   *
+   * Override only if your contrib uses the marker-class pattern AND your native side goes
+   * through Comet's tuned `ParquetSource`. Contribs that define their own `SparkPlan`
+   * subclass (rather than reusing `CometScanExec`) don't need this; they control filter
+   * selection themselves.
+   *
+   * Example: a Delta contrib that uses `CometScanExec(..., scanImpl="native_delta_compat")`
+   * would override this to `Set("native_delta_compat")`.
+   */
+  def nativeParquetScanImpls: Set[String] = Set.empty
 }

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

@@ -159,7 +159,13 @@ case class CometScanExec(
    * on array columns (see [[isNullCheckOnArrayColumn]]).
    */
   lazy val supportedDataFilters: Seq[Expression] = {
-    if (scanImpl == CometConf.SCAN_NATIVE_DATAFUSION) {
+    // Contribs that use the CometScanExec marker pattern with their own scanImpl
+    // string can declare that their scan goes through Comet's tuned ParquetSource
+    // (and therefore wants DataFusion-style filter exclusions) by registering the
+    // tag via `CometOperatorSerdeExtension.nativeParquetScanImpls`. Core doesn't
+    // need to know any contrib's marker name; the registry is the source of truth.
+    if (scanImpl == CometConf.SCAN_NATIVE_DATAFUSION ||
+      CometScanExec.contribNativeParquetScanImpls.contains(scanImpl)) {
       dataFilters
         .filterNot(isDynamicPruningFilter)
         .filterNot(isNullCheckOnArrayColumn)
@@ -534,6 +540,17 @@ case class CometScanExec(
 
 object CometScanExec {
 
+  /**
+   * Set of contrib-registered scanImpl tags whose CometScanExec should use Comet's
+   * native-parquet filter exclusion semantics (drop dynamic pruning + IsNull/IsNotNull on
+   * ArrayType columns). Populated lazily from
+   * `CometExtensionRegistry.serdeExtensions.flatMap(_.nativeParquetScanImpls)`. Each access
+   * re-reads the volatile field on `CometExtensionRegistry`; the cost is one HashSet
+   * lookup per CometScanExec construction, which is dwarfed by Spark's own per-plan work.
+   */
+  private[comet] def contribNativeParquetScanImpls: Set[String] =
+    org.apache.comet.spi.CometExtensionRegistry.nativeParquetScanImpls
+
   def apply(
       scanExec: FileSourceScanExec,
       session: SparkSession,