[AURON #2253] Support insert-only Iceberg changelog native scan

Author: weimingdiit

spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronConvertStrategy.scala

@@ -71,7 +71,9 @@ object AuronConvertStrategy extends Logging {
           if (!exec.getTagValue(neverConvertReasonTag).isDefined) {
             exec.setTagValue(
               neverConvertReasonTag,
-              s"${exec.getClass.getSimpleName} is not supported yet.")
+              converted
+                .getTagValue(neverConvertReasonTag)
+                .getOrElse(s"${exec.getClass.getSimpleName} is not supported yet."))
           }
       }
       danglingChildren = newDangling :+ converted

spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronConverters.scala

@@ -65,6 +65,7 @@ import org.apache.spark.sql.execution.auron.plan.NativeSortBase
 import org.apache.spark.sql.execution.auron.plan.NativeUnionBase
 import org.apache.spark.sql.execution.auron.plan.Util
 import org.apache.spark.sql.execution.command.DataWritingCommandExec
+import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
 import org.apache.spark.sql.execution.exchange.BroadcastExchangeExec
 import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
 import org.apache.spark.sql.execution.joins._
@@ -415,6 +416,8 @@ object AuronConverters extends Logging {
                 } else {
                   s"Falling back exec: ${exec.getClass.getSimpleName}: ${e.getMessage}"
                 }
+              case _: BatchScanExec =>
+                s"${e.getMessage.replaceFirst("^assertion failed: ?", "")}"
               case _ =>
                 s"Falling back exec: ${exec.getClass.getSimpleName}: ${e.getMessage}"
             }

thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergConvertProvider.scala

@@ -44,7 +44,8 @@ class IcebergConvertProvider extends AuronConvertProvider with Logging {
 
   override def isSupported(exec: SparkPlan): Boolean = {
     exec match {
-      case e: BatchScanExec => IcebergScanSupport.plan(e).nonEmpty
+      case e: BatchScanExec if IcebergScanSupport.isIcebergScan(e.scan) =>
+        IcebergScanSupport.plan(e).nonEmpty || IcebergScanSupport.fallbackReason(e).nonEmpty
       case _ => false
     }
   }
@@ -56,7 +57,10 @@ class IcebergConvertProvider extends AuronConvertProvider with Logging {
           case Some(plan) =>
             AuronConverters.addRenameColumnsExec(NativeIcebergTableScanExec(e, plan))
           case None =>
-            exec
+            IcebergScanSupport.fallbackReason(e) match {
+              case Some(reason) => throw new AssertionError(reason)
+              case None => exec
+            }
         }
       case _ => exec
     }