[CORE] Query planner: Simplify validator FallbackByNativeValidation

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

@@ -18,9 +18,9 @@ package org.apache.gluten.backendsapi.clickhouse
 
 import org.apache.gluten.GlutenBuildInfo._
 import org.apache.gluten.GlutenConfig
-import org.apache.gluten.backend.Component.BuildInfo
 import org.apache.gluten.backendsapi._
 import org.apache.gluten.columnarbatch.CHBatch
+import org.apache.gluten.component.Component.BuildInfo
 import org.apache.gluten.execution.WriteFilesExecTransformer
 import org.apache.gluten.expression.WindowFunctionsBuilder
 import org.apache.gluten.extension.ValidationResult

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

@@ -82,6 +82,7 @@ object CHRuleApi {
     injector.injectPreTransform(_ => WriteFilesWithBucketValue)
 
     // Legacy: The legacy transform rule.
+    val offloads = Seq(OffloadOthers(), OffloadExchange(), OffloadJoin())
     val validatorBuilder: GlutenConfig => Validator = conf =>
       Validator
         .builder()
@@ -91,11 +92,10 @@ object CHRuleApi {
         .fallbackByBackendSettings()
         .fallbackByUserOptions()
         .fallbackByTestInjects()
-        .fallbackByNativeValidation()
+        .fallbackByNativeValidation(offloads)
         .build()
     val rewrites =
       Seq(RewriteIn, RewriteMultiChildrenCount, RewriteJoin, PullOutPreProject, PullOutPostProject)
-    val offloads = Seq(OffloadOthers(), OffloadExchange(), OffloadJoin())
     injector.injectTransform(
       c => intercept(HeuristicTransform.Single(validatorBuilder(c.glutenConf), rewrites, offloads)))
 

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

@@ -656,7 +656,7 @@ class CHSparkPlanExecApi extends SparkPlanExecApi with Logging {
   }
 
   override def createColumnarWriteFilesExec(
-      child: SparkPlan,
+      child: WriteFilesExecTransformer,
       noop: SparkPlan,
       fileFormat: FileFormat,
       partitionColumns: Seq[Attribute],
@@ -666,6 +666,7 @@ class CHSparkPlanExecApi extends SparkPlanExecApi with Logging {
     CHColumnarWriteFilesExec(
       child,
       noop,
+      child,
       fileFormat,
       partitionColumns,
       bucketSpec,

backends-clickhouse/src/main/scala/org/apache/spark/sql/execution/CHColumnarWriteFilesExec.scala

@@ -16,6 +16,8 @@
  */
 package org.apache.spark.sql.execution
 
+import org.apache.gluten.execution.WriteFilesExecTransformer
+import org.apache.gluten.extension.ValidationResult
 import org.apache.gluten.memory.CHThreadGroup
 
 import org.apache.spark.{Partition, SparkException, TaskContext, TaskOutputFileAlreadyExistException}
@@ -149,17 +151,22 @@ class CHColumnarWriteFilesRDD(
 case class CHColumnarWriteFilesExec(
     override val left: SparkPlan,
     override val right: SparkPlan,
+    t: WriteFilesExecTransformer,
     fileFormat: FileFormat,
     partitionColumns: Seq[Attribute],
     bucketSpec: Option[BucketSpec],
     options: Map[String, String],
     staticPartitions: TablePartitionSpec
 ) extends ColumnarWriteFilesExec(left, right) {
 
+  override protected def doValidateInternal(): ValidationResult = {
+    t.doValidateInternal()
+  }
+
   override protected def withNewChildrenInternal(
       newLeft: SparkPlan,
       newRight: SparkPlan): SparkPlan =
-    copy(newLeft, newRight, fileFormat, partitionColumns, bucketSpec, options, staticPartitions)
+    copy(newLeft, newRight, t, fileFormat, partitionColumns, bucketSpec, options, staticPartitions)
 
   override def doExecuteWrite(writeFilesSpec: WriteFilesSpec): RDD[WriterCommitMessage] = {
     assert(child.supportsColumnar)

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

@@ -18,9 +18,9 @@ package org.apache.gluten.backendsapi.velox
 
 import org.apache.gluten.GlutenBuildInfo._
 import org.apache.gluten.GlutenConfig
-import org.apache.gluten.backend.Component.BuildInfo
 import org.apache.gluten.backendsapi._
 import org.apache.gluten.columnarbatch.VeloxBatch
+import org.apache.gluten.component.Component.BuildInfo
 import org.apache.gluten.exception.GlutenNotSupportException
 import org.apache.gluten.execution.WriteFilesExecTransformer
 import org.apache.gluten.expression.WindowFunctionsBuilder

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

@@ -74,6 +74,7 @@ object VeloxRuleApi {
     injector.injectPreTransform(c => ArrowScanReplaceRule.apply(c.session))
 
     // Legacy: The legacy transform rule.
+    val offloads = Seq(OffloadOthers(), OffloadExchange(), OffloadJoin())
     val validatorBuilder: GlutenConfig => Validator = conf =>
       Validator
         .builder()
@@ -83,11 +84,10 @@ object VeloxRuleApi {
         .fallbackByBackendSettings()
         .fallbackByUserOptions()
         .fallbackByTestInjects()
-        .fallbackByNativeValidation()
+        .fallbackByNativeValidation(offloads)
         .build()
     val rewrites =
       Seq(RewriteIn, RewriteMultiChildrenCount, RewriteJoin, PullOutPreProject, PullOutPostProject)
-    val offloads = Seq(OffloadOthers(), OffloadExchange(), OffloadJoin())
     injector.injectTransform(
       c => HeuristicTransform.Single(validatorBuilder(c.glutenConf), rewrites, offloads))
 

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

@@ -560,7 +560,7 @@ class VeloxSparkPlanExecApi extends SparkPlanExecApi {
     ShuffleUtil.genColumnarShuffleWriter(parameters)
   }
   override def createColumnarWriteFilesExec(
-      child: SparkPlan,
+      child: WriteFilesExecTransformer,
       noop: SparkPlan,
       fileFormat: FileFormat,
       partitionColumns: Seq[Attribute],
@@ -570,6 +570,7 @@ class VeloxSparkPlanExecApi extends SparkPlanExecApi {
     VeloxColumnarWriteFilesExec(
       child,
       noop,
+      child,
       fileFormat,
       partitionColumns,
       bucketSpec,

backends-velox/src/main/scala/org/apache/spark/sql/execution/VeloxColumnarWriteFilesExec.scala

@@ -18,6 +18,8 @@ package org.apache.spark.sql.execution
 
 import org.apache.gluten.backendsapi.BackendsApiManager
 import org.apache.gluten.columnarbatch.ColumnarBatches
+import org.apache.gluten.execution.WriteFilesExecTransformer
+import org.apache.gluten.extension.ValidationResult
 import org.apache.gluten.memory.arrow.alloc.ArrowBufferAllocators
 
 import org.apache.spark.{Partition, SparkException, TaskContext, TaskOutputFileAlreadyExistException}
@@ -250,13 +252,18 @@ class VeloxColumnarWriteFilesRDD(
 case class VeloxColumnarWriteFilesExec private (
     override val left: SparkPlan,
     override val right: SparkPlan,
+    t: WriteFilesExecTransformer,
     fileFormat: FileFormat,
     partitionColumns: Seq[Attribute],
     bucketSpec: Option[BucketSpec],
     options: Map[String, String],
     staticPartitions: TablePartitionSpec)
   extends ColumnarWriteFilesExec(left, right) {
 
+  override protected def doValidateInternal(): ValidationResult = {
+    t.doValidateInternal()
+  }
+
   override def doExecuteWrite(writeFilesSpec: WriteFilesSpec): RDD[WriterCommitMessage] = {
     assert(child.supportsColumnar)
 
@@ -276,5 +283,5 @@ case class VeloxColumnarWriteFilesExec private (
   override protected def withNewChildrenInternal(
       newLeft: SparkPlan,
       newRight: SparkPlan): SparkPlan =
-    copy(newLeft, newRight, fileFormat, partitionColumns, bucketSpec, options, staticPartitions)
+    copy(newLeft, newRight, t, fileFormat, partitionColumns, bucketSpec, options, staticPartitions)
 }

gluten-core/src/main/scala/org/apache/gluten/GlutenPlugin.scala

@@ -18,7 +18,7 @@ package org.apache.gluten
 
 import org.apache.gluten.GlutenBuildInfo._
 import org.apache.gluten.GlutenConfig._
-import org.apache.gluten.backend.Component
+import org.apache.gluten.component.Component
 import org.apache.gluten.events.GlutenBuildInfoEvent
 import org.apache.gluten.exception.GlutenException
 import org.apache.gluten.extension.GlutenSessionExtensions

gluten-core/src/main/scala/org/apache/gluten/backend/Backend.scala

@@ -16,6 +16,8 @@
  */
 package org.apache.gluten.backend
 
+import org.apache.gluten.component.Component
+
 trait Backend extends Component {
 
   /**

gluten-core/src/main/scala/org/apache/gluten/backend/Component.scala → gluten-core/src/main/scala/org/apache/gluten/component/Component.scala

@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.gluten.backend
+package org.apache.gluten.component
 
 import org.apache.gluten.extension.columnar.transition.ConventionFunc
 import org.apache.gluten.extension.injector.Injector
@@ -100,7 +100,7 @@ object Component {
     graph.sorted()
   }
 
-  private[backend] def sortedUnsafe(): Seq[Component] = {
+  private[component] def sortedUnsafe(): Seq[Component] = {
     graph.sorted()
   }
 

gluten-core/src/main/scala/org/apache/gluten/backend/package.scala → gluten-core/src/main/scala/org/apache/gluten/component/package.scala

@@ -16,17 +16,19 @@
  */
 package org.apache.gluten
 
+import org.apache.gluten.backend.Backend
+
 import org.apache.spark.internal.Logging
 
 import java.util.ServiceLoader
 import java.util.concurrent.atomic.AtomicBoolean
 
 import scala.collection.JavaConverters._
 
-package object backend extends Logging {
-  private[backend] val allComponentsLoaded: AtomicBoolean = new AtomicBoolean(false)
+package object component extends Logging {
+  private val allComponentsLoaded: AtomicBoolean = new AtomicBoolean(false)
 
-  private[backend] def ensureAllComponentsRegistered(): Unit = {
+  private[component] def ensureAllComponentsRegistered(): Unit = {
     if (!allComponentsLoaded.compareAndSet(false, true)) {
       return
     }

gluten-core/src/main/scala/org/apache/gluten/extension/GlutenSessionExtensions.scala

@@ -17,7 +17,7 @@
 package org.apache.gluten.extension
 
 import org.apache.gluten.GlutenConfig
-import org.apache.gluten.backend.Component
+import org.apache.gluten.component.Component
 import org.apache.gluten.extension.injector.Injector
 
 import org.apache.spark.internal.Logging

gluten-core/src/main/scala/org/apache/gluten/extension/columnar/enumerated/EnumeratedTransform.scala

@@ -16,7 +16,7 @@
  */
 package org.apache.gluten.extension.columnar.enumerated
 
-import org.apache.gluten.backend.Component
+import org.apache.gluten.component.Component
 import org.apache.gluten.exception.GlutenException
 import org.apache.gluten.extension.columnar.ColumnarRuleApplier.ColumnarRuleCall
 import org.apache.gluten.extension.columnar.enumerated.planner.GlutenOptimization

gluten-core/src/main/scala/org/apache/gluten/extension/columnar/enumerated/planner/plan/GlutenPlanModel.scala

@@ -17,7 +17,7 @@
 package org.apache.gluten.extension.columnar.enumerated.planner.plan
 
 import org.apache.gluten.execution.GlutenPlan
-import org.apache.gluten.extension.columnar.enumerated.planner.metadata.GlutenMetadata
+import org.apache.gluten.extension.columnar.enumerated.planner.metadata.{GlutenMetadata, LogicalLink}
 import org.apache.gluten.extension.columnar.enumerated.planner.property.{Conv, ConvDef}
 import org.apache.gluten.extension.columnar.transition.{Convention, ConventionReq}
 import org.apache.gluten.ras.{Metadata, PlanModel}
@@ -27,6 +27,7 @@ import org.apache.gluten.sql.shims.SparkShimLoader
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.execution.{ColumnarToRowExec, LeafExecNode, SparkPlan}
 import org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanExecBase
 import org.apache.spark.task.{SparkTaskUtil, TaskResources}
@@ -75,6 +76,16 @@ object GlutenPlanModel {
     final override val supportsRowBased: Boolean = {
       rowType() != Convention.RowType.None
     }
+
+    override def logicalLink: Option[LogicalPlan] = {
+      if (metadata.logicalLink() eq LogicalLink.notFound) {
+        return None
+      }
+      Some(metadata.logicalLink().plan)
+    }
+
+    override def setLogicalLink(logicalPlan: LogicalPlan): Unit =
+      throw new UnsupportedOperationException()
   }
 
   private object PlanModelImpl extends PlanModel[SparkPlan] {

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

@@ -16,7 +16,7 @@
  */
 package org.apache.gluten.extension.columnar.heuristic
 
-import org.apache.gluten.backend.Component
+import org.apache.gluten.component.Component
 import org.apache.gluten.exception.GlutenException
 import org.apache.gluten.extension.columnar.ColumnarRuleApplier.ColumnarRuleCall
 import org.apache.gluten.extension.columnar.offload.OffloadSingleNode

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

@@ -23,7 +23,6 @@ import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.execution.SparkPlan
 
 class LegacyOffload(rules: Seq[OffloadSingleNode]) extends Rule[SparkPlan] with LogLevelUtil {
-
   def apply(plan: SparkPlan): SparkPlan = {
     val out =
       rules.foldLeft(plan)((p, rule) => p.transformUp { case p => rule.offload(p) })

gluten-core/src/main/scala/org/apache/gluten/extension/columnar/transition/ConventionFunc.scala

@@ -16,7 +16,7 @@
  */
 package org.apache.gluten.extension.columnar.transition
 
-import org.apache.gluten.backend.Component
+import org.apache.gluten.component.Component
 import org.apache.gluten.extension.columnar.transition.ConventionReq.KnownChildConvention
 import org.apache.gluten.sql.shims.SparkShimLoader
 

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

@@ -50,7 +50,7 @@ object Validator {
 
     /** Add a custom validator to pipeline. */
     def add(validator: Validator): Builder = {
-      buffer += validator
+      buffer ++= flatten(validator)
       this
     }
 
@@ -64,7 +64,15 @@ object Validator {
       new ValidatorPipeline(buffer.toSeq)
     }
 
-    private class ValidatorPipeline(validators: Seq[Validator]) extends Validator {
+    private def flatten(validator: Validator): Seq[Validator] = validator match {
+      case p: ValidatorPipeline =>
+        p.validators.flatMap(flatten)
+      case other => Seq(other)
+    }
+
+    private class ValidatorPipeline(val validators: Seq[Validator]) extends Validator {
+      assert(!validators.exists(_.isInstanceOf[ValidatorPipeline]))
+
       override def validate(plan: SparkPlan): Validator.OutCome = {
         val init: Validator.OutCome = pass()
         val finalOut = validators.foldLeft(init) {
@@ -86,4 +94,10 @@ object Validator {
   private object Builder {
     def apply(): Builder = new Builder()
   }
+
+  implicit class ValidatorImplicits(v: Validator) {
+    def andThen(other: Validator): Validator = {
+      builder().add(v).add(other).build()
+    }
+  }
 }

gluten-core/src/test/scala/org/apache/gluten/backend/ComponentSuite.scala → gluten-core/src/test/scala/org/apache/gluten/component/ComponentSuite.scala

@@ -14,8 +14,9 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.gluten.backend
+package org.apache.gluten.component
 
+import org.apache.gluten.backend.Backend
 import org.apache.gluten.extension.injector.Injector
 
 import org.scalatest.BeforeAndAfterAll

gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/BackendsApiManager.scala

@@ -16,7 +16,7 @@
  */
 package org.apache.gluten.backendsapi
 
-import org.apache.gluten.backend.Component
+import org.apache.gluten.component.Component
 
 object BackendsApiManager {
   private lazy val backend: SubstraitBackend = initializeInternal()

gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala

@@ -365,7 +365,7 @@ trait SparkPlanExecApi {
 
   /** Create ColumnarWriteFilesExec */
   def createColumnarWriteFilesExec(
-      child: SparkPlan,
+      child: WriteFilesExecTransformer,
       noop: SparkPlan,
       fileFormat: FileFormat,
       partitionColumns: Seq[Attribute],

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

@@ -198,7 +198,6 @@ case class WholeStageTransformer(child: SparkPlan, materializeInput: Boolean = f
     val transformStageId: Int
 ) extends WholeStageTransformerGenerateTreeStringShim
   with UnaryTransformSupport {
-  assert(child.isInstanceOf[TransformSupport])
 
   def stageId: Int = transformStageId
 
@@ -353,6 +352,7 @@ case class WholeStageTransformer(child: SparkPlan, materializeInput: Boolean = f
   }
 
   override def doExecuteColumnar(): RDD[ColumnarBatch] = {
+    assert(child.isInstanceOf[TransformSupport])
     val pipelineTime: SQLMetric = longMetric("pipelineTime")
     // We should do transform first to make sure all subqueries are materialized
     val wsCtx = GlutenTimeMetric.withMillisTime {