Revert "chore: Remove dead code paths for deprecated native_comet scan (#3396)"

This reverts commit 5a9d066.

Author: mbutrovich

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

@@ -38,6 +38,7 @@ import org.apache.spark.sql.execution.datasources.HadoopFsRelation
 import org.apache.spark.sql.execution.datasources.parquet.ParquetUtils
 import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
 import org.apache.spark.sql.execution.datasources.v2.csv.CSVScan
+import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
 
@@ -47,7 +48,7 @@ import org.apache.comet.CometSparkSessionExtensions.{isCometLoaded, withInfo, wi
 import org.apache.comet.DataTypeSupport.isComplexType
 import org.apache.comet.iceberg.{CometIcebergNativeScanMetadata, IcebergReflection}
 import org.apache.comet.objectstore.NativeConfig
-import org.apache.comet.parquet.{Native, SupportsComet}
+import org.apache.comet.parquet.{CometParquetScan, Native, SupportsComet}
 import org.apache.comet.parquet.CometParquetUtils.{encryptionEnabled, isEncryptionConfigSupported}
 import org.apache.comet.serde.operator.CometNativeScan
 import org.apache.comet.shims.{CometTypeShim, ShimFileFormat, ShimSubqueryBroadcast}
@@ -60,6 +61,8 @@ case class CometScanRule(session: SparkSession)
     with CometTypeShim
     with ShimSubqueryBroadcast {
 
+  import CometScanRule._
+
   private lazy val showTransformations = CometConf.COMET_EXPLAIN_TRANSFORMATIONS.get()
 
   override def apply(plan: SparkPlan): SparkPlan = {
@@ -173,6 +176,8 @@ case class CometScanRule(session: SparkSession)
             nativeDataFusionScan(session, scanExec, r, hadoopConf).getOrElse(scanExec)
           case SCAN_NATIVE_ICEBERG_COMPAT =>
             nativeIcebergCompatScan(session, scanExec, r, hadoopConf).getOrElse(scanExec)
+          case SCAN_NATIVE_COMET =>
+            nativeCometScan(session, scanExec, r, hadoopConf).getOrElse(scanExec)
         }
 
       case _ =>
@@ -226,9 +231,47 @@ case class CometScanRule(session: SparkSession)
     Some(CometScanExec(scanExec, session, SCAN_NATIVE_ICEBERG_COMPAT))
   }
 
+  private def nativeCometScan(
+      session: SparkSession,
+      scanExec: FileSourceScanExec,
+      r: HadoopFsRelation,
+      hadoopConf: Configuration): Option[SparkPlan] = {
+    if (!isSchemaSupported(scanExec, SCAN_NATIVE_COMET, r)) {
+      return None
+    }
+    Some(CometScanExec(scanExec, session, SCAN_NATIVE_COMET))
+  }
+
   private def transformV2Scan(scanExec: BatchScanExec): SparkPlan = {
 
     scanExec.scan match {
+      case scan: ParquetScan if COMET_NATIVE_SCAN_IMPL.get() == SCAN_NATIVE_COMET =>
+        val fallbackReasons = new ListBuffer[String]()
+        val schemaSupported =
+          CometBatchScanExec.isSchemaSupported(scan.readDataSchema, fallbackReasons)
+        if (!schemaSupported) {
+          fallbackReasons += s"Schema ${scan.readDataSchema} is not supported"
+        }
+
+        val partitionSchemaSupported =
+          CometBatchScanExec.isSchemaSupported(scan.readPartitionSchema, fallbackReasons)
+        if (!partitionSchemaSupported) {
+          fallbackReasons += s"Partition schema ${scan.readPartitionSchema} is not supported"
+        }
+
+        if (scan.pushedAggregate.nonEmpty) {
+          fallbackReasons += "Comet does not support pushed aggregate"
+        }
+
+        if (schemaSupported && partitionSchemaSupported && scan.pushedAggregate.isEmpty) {
+          val cometScan = CometParquetScan(session, scanExec.scan.asInstanceOf[ParquetScan])
+          CometBatchScanExec(
+            scanExec.copy(scan = cometScan),
+            runtimeFilters = scanExec.runtimeFilters)
+        } else {
+          withInfos(scanExec, fallbackReasons.toSet)
+        }
+
       case scan: CSVScan if COMET_CSV_V2_NATIVE_ENABLED.get() =>
         val fallbackReasons = new ListBuffer[String]()
         val schemaSupported =
@@ -643,6 +686,48 @@ case class CometScanRule(session: SparkSession)
     }
   }
 
+  private def selectScan(
+      scanExec: FileSourceScanExec,
+      partitionSchema: StructType,
+      hadoopConf: Configuration): String = {
+
+    val fallbackReasons = new ListBuffer[String]()
+
+    // native_iceberg_compat only supports local filesystem and S3
+    if (scanExec.relation.inputFiles
+        .forall(path => path.startsWith("file://") || path.startsWith("s3a://"))) {
+
+      val filePath = scanExec.relation.inputFiles.headOption
+      if (filePath.exists(_.startsWith("s3a://"))) {
+        validateObjectStoreConfig(filePath.get, hadoopConf, fallbackReasons)
+      }
+    } else {
+      fallbackReasons += s"$SCAN_NATIVE_ICEBERG_COMPAT only supports local filesystem and S3"
+    }
+
+    val typeChecker = CometScanTypeChecker(SCAN_NATIVE_ICEBERG_COMPAT)
+    val schemaSupported =
+      typeChecker.isSchemaSupported(scanExec.requiredSchema, fallbackReasons)
+    val partitionSchemaSupported =
+      typeChecker.isSchemaSupported(partitionSchema, fallbackReasons)
+
+    val cometExecEnabled = COMET_EXEC_ENABLED.get()
+    if (!cometExecEnabled) {
+      fallbackReasons += s"$SCAN_NATIVE_ICEBERG_COMPAT requires ${COMET_EXEC_ENABLED.key}=true"
+    }
+
+    if (cometExecEnabled && schemaSupported && partitionSchemaSupported &&
+      fallbackReasons.isEmpty) {
+      logInfo(s"Auto scan mode selecting $SCAN_NATIVE_ICEBERG_COMPAT")
+      SCAN_NATIVE_ICEBERG_COMPAT
+    } else {
+      logInfo(
+        s"Auto scan mode falling back to $SCAN_NATIVE_COMET due to " +
+          s"${fallbackReasons.mkString(", ")}")
+      SCAN_NATIVE_COMET
+    }
+  }
+
   private def isDynamicPruningFilter(e: Expression): Boolean =
     e.exists(_.isInstanceOf[PlanExpression[_]])
 
@@ -685,12 +770,16 @@ case class CometScanTypeChecker(scanImpl: String) extends DataTypeSupport with C
       name: String,
       fallbackReasons: ListBuffer[String]): Boolean = {
     dt match {
-      case ShortType if CometConf.COMET_PARQUET_UNSIGNED_SMALL_INT_CHECK.get() =>
+      case ShortType
+          if scanImpl != CometConf.SCAN_NATIVE_COMET &&
+            CometConf.COMET_PARQUET_UNSIGNED_SMALL_INT_CHECK.get() =>
         fallbackReasons += s"$scanImpl scan may not handle unsigned UINT_8 correctly for $dt. " +
           s"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 _: StructType | _: ArrayType | _: MapType if scanImpl == CometConf.SCAN_NATIVE_COMET =>
+        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

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

@@ -155,6 +155,7 @@ case class EliminateRedundantTransitions(session: SparkSession) extends Rule[Spa
    * with such scans because the buffers may be modified after C2R reads them.
    *
    * This includes:
+   *   - CometScanExec with native_comet scan implementation (V1 path) - uses BatchReader
    *   - CometScanExec with native_iceberg_compat and partition columns - uses
    *     ConstantColumnReader
    *   - CometBatchScanExec with CometParquetScan (V2 Parquet path) - uses BatchReader
@@ -166,8 +167,9 @@ case class EliminateRedundantTransitions(session: SparkSession) extends Rule[Spa
       case _ =>
         op.exists {
           case scan: CometScanExec =>
-            scan.scanImpl == CometConf.SCAN_NATIVE_ICEBERG_COMPAT &&
-            scan.relation.partitionSchema.nonEmpty
+            scan.scanImpl == CometConf.SCAN_NATIVE_COMET ||
+            (scan.scanImpl == CometConf.SCAN_NATIVE_ICEBERG_COMPAT &&
+              scan.relation.partitionSchema.nonEmpty)
           case scan: CometBatchScanExec => scan.scan.isInstanceOf[CometParquetScan]
           case _ => false
         }