[GLUTEN-8356][CORE][VL][CH] Make Iceberg code implement component API

backends-clickhouse/pom.xml

@@ -39,6 +39,11 @@
         </dependency>
       </dependencies>
       <build>
+        <resources>
+          <resource>
+            <directory>${project.basedir}/src/main-iceberg/resource</directory>
+          </resource>
+        </resources>
         <plugins>
           <plugin>
             <groupId>org.codehaus.mojo</groupId>

backends-clickhouse/src/main-iceberg/resource/META-INF/services/org.apache.gluten.component.Component

@@ -0,0 +1 @@
+org.apache.gluten.component.CHIcebergComponent

backends-clickhouse/src/main-iceberg/scala/org/apache/gluten/component/CHIcebergComponent.scala

@@ -0,0 +1,32 @@
+/*
+ * 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.gluten.component
+
+import org.apache.gluten.backendsapi.clickhouse.CHBackend
+import org.apache.gluten.execution.OffloadIcebergScan
+import org.apache.gluten.extension.injector.Injector
+
+class CHIcebergComponent extends Component {
+  override def name(): String = "clickhouse-iceberg"
+  override def buildInfo(): Component.BuildInfo =
+    Component.BuildInfo("ClickHouseIceberg", "N/A", "N/A", "N/A")
+  override def dependencies(): Seq[Class[_ <: Component]] = classOf[CHBackend] :: Nil
+  override def injectRules(injector: Injector): Unit = {
+    OffloadIcebergScan.inject(injector)
+  }
+}

backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHRuleApi.scala

@@ -26,8 +26,7 @@ import org.apache.gluten.extension.columnar.heuristic.{ExpandFallbackPolicy, Heu
 import org.apache.gluten.extension.columnar.offload.{OffloadExchange, OffloadJoin, OffloadOthers}
 import org.apache.gluten.extension.columnar.rewrite._
 import org.apache.gluten.extension.columnar.transition.{InsertTransitions, RemoveTransitions}
-import org.apache.gluten.extension.columnar.validator.Validator
-import org.apache.gluten.extension.columnar.validator.Validators.ValidatorBuilderImplicits
+import org.apache.gluten.extension.columnar.validator.{Validator, Validators}
 import org.apache.gluten.extension.injector.{Injector, SparkInjector}
 import org.apache.gluten.extension.injector.GlutenInjector.{LegacyInjector, RasInjector}
 import org.apache.gluten.parser.{GlutenCacheFilesSqlParser, GlutenClickhouseSqlParser}
@@ -84,20 +83,13 @@ object CHRuleApi {
     // Legacy: The legacy transform rule.
     val offloads = Seq(OffloadOthers(), OffloadExchange(), OffloadJoin())
     val validatorBuilder: GlutenConfig => Validator = conf =>
-      Validator
-        .builder()
-        .fallbackByHint()
-        .fallbackIfScanOnlyWithFilterPushed(conf.enableScanOnly)
-        .fallbackComplexExpressions()
-        .fallbackByBackendSettings()
-        .fallbackByUserOptions()
-        .fallbackByTestInjects()
-        .fallbackByNativeValidation(offloads)
-        .build()
+      Validators.newValidator(conf, offloads)
     val rewrites =
       Seq(RewriteIn, RewriteMultiChildrenCount, RewriteJoin, PullOutPreProject, PullOutPostProject)
     injector.injectTransform(
-      c => intercept(HeuristicTransform.Single(validatorBuilder(c.glutenConf), rewrites, offloads)))
+      c =>
+        intercept(
+          HeuristicTransform.WithRewrites(validatorBuilder(c.glutenConf), rewrites, offloads)))
 
     // Legacy: Post-transform rules.
     injector.injectPostTransform(_ => PruneNestedColumnsInHiveTableScan)

backends-velox/pom.xml

@@ -61,6 +61,11 @@
         </dependency>
       </dependencies>
       <build>
+        <resources>
+          <resource>
+            <directory>${project.basedir}/src/main-iceberg/resource</directory>
+          </resource>
+        </resources>
         <plugins>
           <plugin>
             <groupId>org.codehaus.mojo</groupId>

backends-velox/src/main-iceberg/resource/META-INF/services/org.apache.gluten.component.Component

@@ -0,0 +1 @@
+org.apache.gluten.component.VeloxIcebergComponent

gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergTransformerProvider.scala → backends-velox/src/main-iceberg/scala/org/apache/gluten/component/VeloxIcebergComponent.scala

@@ -14,16 +14,18 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.gluten.execution
 
-import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
+package org.apache.gluten.component
+import org.apache.gluten.backendsapi.velox.VeloxBackend
+import org.apache.gluten.execution.OffloadIcebergScan
+import org.apache.gluten.extension.injector.Injector
 
-class IcebergTransformerProvider extends DataSourceScanTransformerRegister {
-
-  override val scanClassName: String = "org.apache.iceberg.spark.source.SparkBatchQueryScan"
-
-  override def createDataSourceV2Transformer(
-      batchScan: BatchScanExec): BatchScanExecTransformerBase = {
-    IcebergScanTransformer(batchScan)
+class VeloxIcebergComponent extends Component {
+  override def name(): String = "velox-iceberg"
+  override def buildInfo(): Component.BuildInfo =
+    Component.BuildInfo("VeloxIceberg", "N/A", "N/A", "N/A")
+  override def dependencies(): Seq[Class[_ <: Component]] = classOf[VeloxBackend] :: Nil
+  override def injectRules(injector: Injector): Unit = {
+    OffloadIcebergScan.inject(injector)
   }
 }

backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala

@@ -28,8 +28,7 @@ import org.apache.gluten.extension.columnar.heuristic.{ExpandFallbackPolicy, Heu
 import org.apache.gluten.extension.columnar.offload.{OffloadExchange, OffloadJoin, OffloadOthers}
 import org.apache.gluten.extension.columnar.rewrite._
 import org.apache.gluten.extension.columnar.transition.{InsertTransitions, RemoveTransitions}
-import org.apache.gluten.extension.columnar.validator.Validator
-import org.apache.gluten.extension.columnar.validator.Validators.ValidatorBuilderImplicits
+import org.apache.gluten.extension.columnar.validator.{Validator, Validators}
 import org.apache.gluten.extension.injector.{Injector, SparkInjector}
 import org.apache.gluten.extension.injector.GlutenInjector.{LegacyInjector, RasInjector}
 import org.apache.gluten.sql.shims.SparkShimLoader
@@ -76,20 +75,11 @@ object VeloxRuleApi {
     // Legacy: The legacy transform rule.
     val offloads = Seq(OffloadOthers(), OffloadExchange(), OffloadJoin())
     val validatorBuilder: GlutenConfig => Validator = conf =>
-      Validator
-        .builder()
-        .fallbackByHint()
-        .fallbackIfScanOnlyWithFilterPushed(conf.enableScanOnly)
-        .fallbackComplexExpressions()
-        .fallbackByBackendSettings()
-        .fallbackByUserOptions()
-        .fallbackByTestInjects()
-        .fallbackByNativeValidation(offloads)
-        .build()
+      Validators.newValidator(conf, offloads)
     val rewrites =
       Seq(RewriteIn, RewriteMultiChildrenCount, RewriteJoin, PullOutPreProject, PullOutPostProject)
     injector.injectTransform(
-      c => HeuristicTransform.Single(validatorBuilder(c.glutenConf), rewrites, offloads))
+      c => HeuristicTransform.WithRewrites(validatorBuilder(c.glutenConf), rewrites, offloads))
 
     // Legacy: Post-transform rules.
     injector.injectPostTransform(_ => UnionTransformerRule())
@@ -132,16 +122,7 @@ object VeloxRuleApi {
     injector.injectPreTransform(c => ArrowScanReplaceRule.apply(c.session))
 
     // Gluten RAS: The RAS rule.
-    val validatorBuilder: GlutenConfig => Validator = conf =>
-      Validator
-        .builder()
-        .fallbackByHint()
-        .fallbackIfScanOnlyWithFilterPushed(conf.enableScanOnly)
-        .fallbackComplexExpressions()
-        .fallbackByBackendSettings()
-        .fallbackByUserOptions()
-        .fallbackByTestInjects()
-        .build()
+    val validatorBuilder: GlutenConfig => Validator = conf => Validators.newValidator(conf)
     val rewrites =
       Seq(RewriteIn, RewriteMultiChildrenCount, RewriteJoin, PullOutPreProject, PullOutPostProject)
     injector.injectCoster(_ => LegacyCoster)

gluten-core/src/main/scala/org/apache/gluten/extension/columnar/heuristic/HeuristicTransform.scala

@@ -26,6 +26,7 @@ import org.apache.gluten.extension.injector.Injector
 import org.apache.gluten.extension.util.AdaptiveContext
 import org.apache.gluten.logging.LogLevelUtil
 
+import org.apache.spark.internal.Logging
 import org.apache.spark.sql.{SparkSession, SparkSessionExtensions}
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.execution.SparkPlan
@@ -64,7 +65,39 @@ object HeuristicTransform {
     new HeuristicTransform(all)
   }
 
-  case class Single(
+  /**
+   * A simple heuristic transform rule with a validator and some offload rules.
+   *
+   * Validator will be called before applying the offload rules.
+   */
+  case class Simple(validator: Validator, offloadRules: Seq[OffloadSingleNode])
+    extends Rule[SparkPlan]
+    with Logging {
+    override def apply(plan: SparkPlan): SparkPlan = {
+      offloadRules.foldLeft(plan) {
+        case (p, rule) =>
+          p.transformUp {
+            node =>
+              validator.validate(node) match {
+                case Validator.Passed =>
+                  rule.offload(node)
+                case Validator.Failed(reason) =>
+                  logDebug(s"Validation failed by reason: $reason on query plan: ${node.nodeName}")
+                  node
+              }
+          }
+      }
+    }
+  }
+
+  /**
+   * A heuristic transform rule with given rewrite rules. Fallback tags will be used in the
+   * procedure to determine which part of the plan is or is not eligible to be offloaded. The tags
+   * should also be correctly handled in the offload rules.
+   *
+   * TODO: Handle tags internally. Remove tag handling code in user offload rules.
+   */
+  case class WithRewrites(
       validator: Validator,
       rewriteRules: Seq[RewriteSingleNode],
       offloadRules: Seq[OffloadSingleNode])

gluten-core/src/main/scala/org/apache/gluten/extension/columnar/validator/Validator.scala

@@ -46,6 +46,7 @@ object Validator {
   def builder(): Builder = Builder()
 
   class Builder private {
+    import Builder._
     private val buffer: ListBuffer[Validator] = mutable.ListBuffer()
 
     /** Add a custom validator to pipeline. */
@@ -69,6 +70,14 @@ object Validator {
         p.validators.flatMap(flatten)
       case other => Seq(other)
     }
+  }
+
+  private object Builder {
+    def apply(): Builder = new Builder()
+
+    private object NoopValidator extends Validator {
+      override def validate(plan: SparkPlan): Validator.OutCome = pass()
+    }
 
     private class ValidatorPipeline(val validators: Seq[Validator]) extends Validator {
       assert(!validators.exists(_.isInstanceOf[ValidatorPipeline]))
@@ -85,14 +94,6 @@ object Validator {
         finalOut
       }
     }
-
-    private object NoopValidator extends Validator {
-      override def validate(plan: SparkPlan): Validator.OutCome = pass()
-    }
-  }
-
-  private object Builder {
-    def apply(): Builder = new Builder()
   }
 
   implicit class ValidatorImplicits(v: Validator) {

gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergScanTransformer.scala

@@ -46,6 +46,10 @@ case class IcebergScanTransformer(
     commonPartitionValues = commonPartitionValues
   ) {
 
+  protected[this] def supportsBatchScan(scan: Scan): Boolean = {
+    IcebergScanTransformer.supportsBatchScan(scan)
+  }
+
   override def filterExprs(): Seq[Expression] = pushdownFilters.getOrElse(Seq.empty)
 
   override lazy val getPartitionSchema: StructType =
@@ -94,4 +98,8 @@ object IcebergScanTransformer {
       commonPartitionValues = SparkShimLoader.getSparkShims.getCommonPartitionValues(batchScan)
     )
   }
+
+  def supportsBatchScan(scan: Scan): Boolean = {
+    scan.getClass.getName == "org.apache.iceberg.spark.source.SparkBatchQueryScan"
+  }
 }

gluten-iceberg/src/main/scala/org/apache/gluten/execution/OffloadIcebergScan.scala

@@ -0,0 +1,57 @@
+/*
+ * 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.gluten.execution
+
+import org.apache.gluten.extension.columnar.enumerated.RasOffload
+import org.apache.gluten.extension.columnar.heuristic.HeuristicTransform
+import org.apache.gluten.extension.columnar.offload.OffloadSingleNode
+import org.apache.gluten.extension.columnar.validator.Validators
+import org.apache.gluten.extension.injector.Injector
+
+import org.apache.spark.sql.execution.SparkPlan
+import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
+
+case class OffloadIcebergScan() extends OffloadSingleNode {
+  override def offload(plan: SparkPlan): SparkPlan = plan match {
+    case scan: BatchScanExec if IcebergScanTransformer.supportsBatchScan(scan.scan) =>
+      IcebergScanTransformer(scan)
+    case other => other
+  }
+}
+
+object OffloadIcebergScan {
+  def inject(injector: Injector): Unit = {
+    // Inject legacy rule.
+    injector.gluten.legacy.injectTransform {
+      c =>
+        val offload = Seq(OffloadIcebergScan())
+        HeuristicTransform.Simple(
+          Validators.newValidator(c.glutenConf, offload),
+          offload
+        )
+    }
+
+    // Inject RAS rule.
+    injector.gluten.ras.injectRasRule {
+      c =>
+        RasOffload.Rule(
+          RasOffload.from[BatchScanExec](OffloadIcebergScan()),
+          Validators.newValidator(c.glutenConf),
+          Nil)
+    }
+  }
+}

gluten-substrait/src/main/scala/org/apache/gluten/execution/BatchScanExecTransformer.scala

@@ -56,6 +56,10 @@ case class BatchScanExecTransformer(
     applyPartialClustering,
     replicatePartitions) {
 
+  protected[this] def supportsBatchScan(scan: Scan): Boolean = {
+    scan.isInstanceOf[FileScan]
+  }
+
   override def doCanonicalize(): BatchScanExecTransformer = {
     this.copy(
       output = output.map(QueryPlan.normalizeExpressions(_, output)),
@@ -134,8 +138,10 @@ abstract class BatchScanExecTransformerBase(
     }
   }
 
+  protected[this] def supportsBatchScan(scan: Scan): Boolean
+
   override def doValidateInternal(): ValidationResult = {
-    if (!ScanTransformerFactory.supportedBatchScan(scan)) {
+    if (!supportsBatchScan(scan)) {
       return ValidationResult.failed(s"Unsupported scan $scan")
     }
 

gluten-substrait/src/main/scala/org/apache/gluten/execution/ScanTransformerFactory.scala

@@ -18,9 +18,8 @@ package org.apache.gluten.execution
 
 import org.apache.gluten.sql.shims.SparkShimLoader
 
-import org.apache.spark.sql.connector.read.Scan
 import org.apache.spark.sql.execution.FileSourceScanExec
-import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec, FileScan}
+import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
 
 import java.util.ServiceLoader
 import java.util.concurrent.ConcurrentHashMap
@@ -75,11 +74,6 @@ object ScanTransformerFactory {
     }
   }
 
-  def supportedBatchScan(scan: Scan): Boolean = scan match {
-    case _: FileScan => true
-    case _ => lookupDataSourceScanTransformer(scan.getClass.getName).nonEmpty
-  }
-
   private def lookupDataSourceScanTransformer(scanClassName: String): Option[Class[_]] = {
     val clz = scanTransformerMap.computeIfAbsent(
       scanClassName,