[VL] Delta support / Hudi support as Gluten components (#8282)

Author: zhztheplayer

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

@@ -25,6 +25,7 @@ import org.apache.gluten.execution.WriteFilesExecTransformer
 import org.apache.gluten.expression.WindowFunctionsBuilder
 import org.apache.gluten.extension.ValidationResult
 import org.apache.gluten.extension.columnar.transition.{Convention, ConventionFunc}
+import org.apache.gluten.substrait.rel.LocalFilesNode
 import org.apache.gluten.substrait.rel.LocalFilesNode.ReadFileFormat
 import org.apache.gluten.substrait.rel.LocalFilesNode.ReadFileFormat._
 
@@ -34,6 +35,7 @@ import org.apache.spark.sql.catalyst.catalog.BucketSpec
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
 import org.apache.spark.sql.catalyst.plans._
+import org.apache.spark.sql.connector.read.Scan
 import org.apache.spark.sql.execution.SparkPlan
 import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec
 import org.apache.spark.sql.execution.datasources.FileFormat
@@ -198,6 +200,27 @@ object CHBackendSettings extends BackendSettingsApi with Logging {
     }
   }
 
+  override def getSubstraitReadFileFormatV1(
+      fileFormat: FileFormat): LocalFilesNode.ReadFileFormat = {
+    fileFormat.getClass.getSimpleName match {
+      case "OrcFileFormat" => ReadFileFormat.OrcReadFormat
+      case "ParquetFileFormat" => ReadFileFormat.ParquetReadFormat
+      case "DeltaParquetFileFormat" => ReadFileFormat.ParquetReadFormat
+      case "DeltaMergeTreeFileFormat" => ReadFileFormat.MergeTreeReadFormat
+      case "CSVFileFormat" => ReadFileFormat.TextReadFormat
+      case _ => ReadFileFormat.UnknownFormat
+    }
+  }
+
+  override def getSubstraitReadFileFormatV2(scan: Scan): LocalFilesNode.ReadFileFormat = {
+    scan.getClass.getSimpleName match {
+      case "OrcScan" => ReadFileFormat.OrcReadFormat
+      case "ParquetScan" => ReadFileFormat.ParquetReadFormat
+      case "ClickHouseScan" => ReadFileFormat.MergeTreeReadFormat
+      case _ => ReadFileFormat.UnknownFormat
+    }
+  }
+
   override def supportWriteFilesExec(
       format: FileFormat,
       fields: Array[StructField],

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

@@ -93,7 +93,6 @@ object CHRuleApi {
 
     // Legacy: Post-transform rules.
     injector.injectPostTransform(_ => PruneNestedColumnsInHiveTableScan)
-    injector.injectPostTransform(c => intercept(RewriteTransformer.apply(c.session)))
     injector.injectPostTransform(_ => PushDownFilterToScan)
     injector.injectPostTransform(_ => PushDownInputFileExpression.PostOffload)
     injector.injectPostTransform(_ => EnsureLocalSortRequirements)

backends-velox/src-delta/main/resources/META-INF/gluten-components/org.apache.gluten.component.VeloxDeltaComponent

@@ -0,0 +1,56 @@
+/*
+ * 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.velox.VeloxBackend
+import org.apache.gluten.execution.OffloadDeltaScan
+import org.apache.gluten.extension.DeltaPostTransformRules
+import org.apache.gluten.extension.columnar.enumerated.RasOffload
+import org.apache.gluten.extension.columnar.heuristic.HeuristicTransform
+import org.apache.gluten.extension.columnar.validator.Validators
+import org.apache.gluten.extension.injector.Injector
+
+import org.apache.spark.sql.execution.FileSourceScanExec
+
+class VeloxDeltaComponent extends Component {
+  override def name(): String = "velox-delta"
+  override def buildInfo(): Component.BuildInfo =
+    Component.BuildInfo("VeloxDelta", "N/A", "N/A", "N/A")
+  override def dependencies(): Seq[Class[_ <: Component]] = classOf[VeloxBackend] :: Nil
+  override def injectRules(injector: Injector): Unit = {
+    val legacy = injector.gluten.legacy
+    val ras = injector.gluten.ras
+    legacy.injectTransform {
+      c =>
+        val offload = Seq(OffloadDeltaScan())
+        HeuristicTransform.Simple(Validators.newValidator(c.glutenConf, offload), offload)
+    }
+    ras.injectRasRule {
+      c =>
+        RasOffload.Rule(
+          RasOffload.from[FileSourceScanExec](OffloadDeltaScan()),
+          Validators.newValidator(c.glutenConf),
+          Nil)
+    }
+    DeltaPostTransformRules.rules.foreach {
+      r =>
+        legacy.injectPostTransform(_ => r)
+        ras.injectPostTransform(_ => r)
+    }
+  }
+}

backends-velox/src-delta/main/scala/org/apache/gluten/component/VeloxDeltaComponent.scala

@@ -0,0 +1,50 @@
+/*
+ * 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.velox.VeloxBackend
+import org.apache.gluten.execution.OffloadHudiScan
+import org.apache.gluten.extension.columnar.enumerated.RasOffload
+import org.apache.gluten.extension.columnar.heuristic.HeuristicTransform
+import org.apache.gluten.extension.columnar.validator.Validators
+import org.apache.gluten.extension.injector.Injector
+
+import org.apache.spark.sql.execution.FileSourceScanExec
+
+class VeloxHudiComponent extends Component {
+  override def name(): String = "velox-hudi"
+  override def buildInfo(): Component.BuildInfo =
+    Component.BuildInfo("VeloxHudi", "N/A", "N/A", "N/A")
+  override def dependencies(): Seq[Class[_ <: Component]] = classOf[VeloxBackend] :: Nil
+  override def injectRules(injector: Injector): Unit = {
+    val legacy = injector.gluten.legacy
+    val ras = injector.gluten.ras
+    legacy.injectTransform {
+      c =>
+        val offload = Seq(OffloadHudiScan())
+        HeuristicTransform.Simple(Validators.newValidator(c.glutenConf, offload), offload)
+    }
+    ras.injectRasRule {
+      c =>
+        RasOffload.Rule(
+          RasOffload.from[FileSourceScanExec](OffloadHudiScan()),
+          Validators.newValidator(c.glutenConf),
+          Nil)
+    }
+  }
+}

backends-velox/src-hudi/main/resources/META-INF/gluten-components/org.apache.gluten.component.VeloxHudiComponent

@@ -27,6 +27,7 @@ import org.apache.gluten.expression.WindowFunctionsBuilder
 import org.apache.gluten.extension.ValidationResult
 import org.apache.gluten.extension.columnar.transition.{Convention, ConventionFunc}
 import org.apache.gluten.sql.shims.SparkShimLoader
+import org.apache.gluten.substrait.rel.LocalFilesNode
 import org.apache.gluten.substrait.rel.LocalFilesNode.ReadFileFormat
 import org.apache.gluten.substrait.rel.LocalFilesNode.ReadFileFormat.{DwrfReadFormat, OrcReadFormat, ParquetReadFormat}
 import org.apache.gluten.utils._
@@ -36,6 +37,7 @@ import org.apache.spark.sql.catalyst.expressions.{Alias, CumeDist, DenseRank, De
 import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, ApproximatePercentile, Percentile}
 import org.apache.spark.sql.catalyst.plans.{JoinType, LeftOuter, RightOuter}
 import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, CharVarcharUtils}
+import org.apache.spark.sql.connector.read.Scan
 import org.apache.spark.sql.execution.{ColumnarCachedBatchSerializer, SparkPlan}
 import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec
 import org.apache.spark.sql.execution.columnar.InMemoryTableScanExec
@@ -173,7 +175,7 @@ object VeloxBackendSettings extends BackendSettingsApi {
             }
             validateTypes(typeValidator)
           }
-        case _ => Some(s"Unsupported file format for $format.")
+        case _ => Some(s"Unsupported file format $format.")
       }
     }
 
@@ -194,6 +196,26 @@ object VeloxBackendSettings extends BackendSettingsApi {
       .toSeq
   }
 
+  override def getSubstraitReadFileFormatV1(
+      fileFormat: FileFormat): LocalFilesNode.ReadFileFormat = {
+    fileFormat.getClass.getSimpleName match {
+      case "OrcFileFormat" => ReadFileFormat.OrcReadFormat
+      case "ParquetFileFormat" => ReadFileFormat.ParquetReadFormat
+      case "DwrfFileFormat" => ReadFileFormat.DwrfReadFormat
+      case "CSVFileFormat" => ReadFileFormat.TextReadFormat
+      case _ => ReadFileFormat.UnknownFormat
+    }
+  }
+
+  override def getSubstraitReadFileFormatV2(scan: Scan): LocalFilesNode.ReadFileFormat = {
+    scan.getClass.getSimpleName match {
+      case "OrcScan" => ReadFileFormat.OrcReadFormat
+      case "ParquetScan" => ReadFileFormat.ParquetReadFormat
+      case "DwrfScan" => ReadFileFormat.DwrfReadFormat
+      case _ => ReadFileFormat.UnknownFormat
+    }
+  }
+
   override def supportWriteFilesExec(
       format: FileFormat,
       fields: Array[StructField],

backends-velox/src-hudi/main/scala/org/apache/gluten/component/VeloxHudiComponent.scala

@@ -85,7 +85,6 @@ object VeloxRuleApi {
     injector.injectPostTransform(_ => UnionTransformerRule())
     injector.injectPostTransform(c => PartialProjectRule.apply(c.session))
     injector.injectPostTransform(_ => RemoveNativeWriteFilesSortAndProject())
-    injector.injectPostTransform(c => RewriteTransformer.apply(c.session))
     injector.injectPostTransform(_ => PushDownFilterToScan)
     injector.injectPostTransform(_ => PushDownInputFileExpression.PostOffload)
     injector.injectPostTransform(_ => EnsureLocalSortRequirements)
@@ -163,7 +162,6 @@ object VeloxRuleApi {
     injector.injectPostTransform(_ => UnionTransformerRule())
     injector.injectPostTransform(c => PartialProjectRule.apply(c.session))
     injector.injectPostTransform(_ => RemoveNativeWriteFilesSortAndProject())
-    injector.injectPostTransform(c => RewriteTransformer.apply(c.session))
     injector.injectPostTransform(_ => PushDownFilterToScan)
     injector.injectPostTransform(_ => PushDownInputFileExpression.PostOffload)
     injector.injectPostTransform(_ => EnsureLocalSortRequirements)

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

@@ -38,7 +38,7 @@
 /**
  * A trivial memory consumer implementation used by Gluten.
  *
- * @deprecated Use {@link TreeMemoryConsumers#shared()} instead.
+ * @deprecated Use {@link TreeMemoryConsumers} instead.
  */
 @Deprecated
 public class RegularMemoryConsumer extends MemoryConsumer

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

@@ -16,14 +16,15 @@
  */
 package org.apache.gluten.execution
 
-import org.apache.spark.sql.execution.FileSourceScanExec
+import org.apache.gluten.extension.columnar.offload.OffloadSingleNode
 
-class HudiScanTransformerProvider extends DataSourceScanTransformerRegister {
+import org.apache.spark.sql.execution.{FileSourceScanExec, SparkPlan}
 
-  override val scanClassName: String = "HoodieParquetFileFormat"
-
-  override def createDataSourceTransformer(
-      batchScan: FileSourceScanExec): FileSourceScanExecTransformerBase = {
-    HudiScanTransformer(batchScan)
+case class OffloadDeltaScan() extends OffloadSingleNode {
+  override def offload(plan: SparkPlan): SparkPlan = plan match {
+    case scan: FileSourceScanExec
+        if scan.relation.fileFormat.getClass.getName == "org.apache.spark.sql.delta.DeltaParquetFileFormat" =>
+      DeltaScanTransformer(scan)
+    case other => other
   }
 }

gluten-core/src/main/java/org/apache/gluten/memory/memtarget/spark/RegularMemoryConsumer.java

@@ -17,8 +17,7 @@
 package org.apache.gluten.extension
 
 import org.apache.gluten.execution.{DeltaFilterExecTransformer, DeltaProjectExecTransformer, DeltaScanTransformer, ProjectExecTransformer}
-import org.apache.gluten.extension.DeltaRewriteTransformerRules.{columnMappingRule, filterRule, projectRule, pushDownInputFileExprRule}
-import org.apache.gluten.extension.columnar.RewriteTransformerRules
+import org.apache.gluten.extension.columnar.transition.RemoveTransitions
 
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeReference, Expression, InputFileBlockLength, InputFileBlockStart, InputFileName}
@@ -30,12 +29,10 @@ import org.apache.spark.sql.execution.datasources.FileFormat
 
 import scala.collection.mutable.ListBuffer
 
-class DeltaRewriteTransformerRules extends RewriteTransformerRules {
-  override def rules: Seq[Rule[SparkPlan]] =
-    columnMappingRule :: filterRule :: projectRule :: pushDownInputFileExprRule :: Nil
-}
-
-object DeltaRewriteTransformerRules {
+object DeltaPostTransformRules {
+  def rules: Seq[Rule[SparkPlan]] =
+    RemoveTransitions :: columnMappingRule :: filterRule :: projectRule ::
+      pushDownInputFileExprRule :: Nil
 
   private val COLUMN_MAPPING_RULE_TAG: TreeNodeTag[String] =
     TreeNodeTag[String]("org.apache.gluten.delta.column.mapping")

gluten-delta/src-delta/main/resources/META-INF/services/org.apache.gluten.execution.DataSourceScanTransformerRegister

@@ -14,16 +14,22 @@
  * 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.FileSourceScanExec
+package org.apache.gluten.execution
 
-class DeltaScanTransformerProvider extends DataSourceScanTransformerRegister {
+import org.apache.gluten.extension.columnar.offload.OffloadSingleNode
 
-  override val scanClassName: String = "org.apache.spark.sql.delta.DeltaParquetFileFormat"
+import org.apache.spark.sql.execution.SparkPlan
 
-  override def createDataSourceTransformer(
-      batchScan: FileSourceScanExec): FileSourceScanExecTransformerBase = {
-    DeltaScanTransformer(batchScan)
+/** Since https://github.com/apache/incubator-gluten/pull/6049. */
+case class OffloadHudiScan() extends OffloadSingleNode {
+  override def offload(plan: SparkPlan): SparkPlan = {
+    plan match {
+      // Hudi has multiple file format definitions whose names end with "HoodieParquetFileFormat".
+      case scan: org.apache.spark.sql.execution.FileSourceScanExec
+          if scan.relation.fileFormat.getClass.getName.endsWith("HoodieParquetFileFormat") =>
+        HudiScanTransformer(scan)
+      case other => other
+    }
   }
 }

gluten-delta/src-delta/main/resources/META-INF/services/org.apache.gluten.extension.columnar.RewriteTransformerRules

@@ -19,11 +19,13 @@ package org.apache.gluten.backendsapi
 import org.apache.gluten.GlutenConfig
 import org.apache.gluten.extension.ValidationResult
 import org.apache.gluten.extension.columnar.transition.Convention
+import org.apache.gluten.substrait.rel.LocalFilesNode
 import org.apache.gluten.substrait.rel.LocalFilesNode.ReadFileFormat
 
 import org.apache.spark.sql.catalyst.catalog.BucketSpec
 import org.apache.spark.sql.catalyst.expressions.{Expression, NamedExpression}
 import org.apache.spark.sql.catalyst.plans._
+import org.apache.spark.sql.connector.read.Scan
 import org.apache.spark.sql.execution.command.CreateDataSourceTableAsSelectCommand
 import org.apache.spark.sql.execution.datasources.{FileFormat, InsertIntoHadoopFsRelationCommand}
 import org.apache.spark.sql.types.StructField
@@ -39,6 +41,10 @@ trait BackendSettingsApi {
       rootPaths: Seq[String],
       properties: Map[String, String]): ValidationResult = ValidationResult.succeeded
 
+  def getSubstraitReadFileFormatV1(fileFormat: FileFormat): LocalFilesNode.ReadFileFormat
+
+  def getSubstraitReadFileFormatV2(scan: Scan): LocalFilesNode.ReadFileFormat
+
   def supportWriteFilesExec(
       format: FileFormat,
       fields: Array[StructField],

gluten-hudi/src-hudi/main/scala/org/apache/gluten/execution/HudiScanTransformerProvider.scala gluten-delta/src-delta/main/scala/org/apache/gluten/execution/OffloadDeltaScan.scala

@@ -175,13 +175,8 @@ abstract class BatchScanExecTransformerBase(
   @transient protected lazy val filteredFlattenPartitions: Seq[InputPartition] =
     filteredPartitions.flatten
 
-  @transient override lazy val fileFormat: ReadFileFormat = scan.getClass.getSimpleName match {
-    case "OrcScan" => ReadFileFormat.OrcReadFormat
-    case "ParquetScan" => ReadFileFormat.ParquetReadFormat
-    case "DwrfScan" => ReadFileFormat.DwrfReadFormat
-    case "ClickHouseScan" => ReadFileFormat.MergeTreeReadFormat
-    case _ => ReadFileFormat.UnknownFormat
-  }
+  @transient override lazy val fileFormat: ReadFileFormat =
+    BackendsApiManager.getSettings.getSubstraitReadFileFormatV2(scan)
 
   override def simpleString(maxFields: Int): String = {
     val truncatedOutputString = truncatedString(output, "[", ", ", "]", maxFields)

gluten-delta/src-delta/main/scala/org/apache/gluten/extension/DeltaRewriteTransformerRules.scala gluten-delta/src-delta/main/scala/org/apache/gluten/extension/DeltaPostTransformRules.scala

@@ -190,15 +190,7 @@ abstract class FileSourceScanExecTransformerBase(
   }
 
   @transient override lazy val fileFormat: ReadFileFormat =
-    relation.fileFormat.getClass.getSimpleName match {
-      case "OrcFileFormat" => ReadFileFormat.OrcReadFormat
-      case "ParquetFileFormat" => ReadFileFormat.ParquetReadFormat
-      case "DeltaParquetFileFormat" => ReadFileFormat.ParquetReadFormat
-      case "DwrfFileFormat" => ReadFileFormat.DwrfReadFormat
-      case "DeltaMergeTreeFileFormat" => ReadFileFormat.MergeTreeReadFormat
-      case "CSVFileFormat" => ReadFileFormat.TextReadFormat
-      case _ => ReadFileFormat.UnknownFormat
-    }
+    BackendsApiManager.getSettings.getSubstraitReadFileFormatV1(relation.fileFormat)
 
   override def simpleString(maxFields: Int): String = {
     val metadataEntries = metadata.toSeq.sorted.map {

gluten-hudi/src-hudi/main/resources/META-INF/services/org.apache.gluten.execution.DataSourceScanTransformerRegister

@@ -21,76 +21,29 @@ import org.apache.gluten.sql.shims.SparkShimLoader
 import org.apache.spark.sql.execution.FileSourceScanExec
 import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
 
-import java.util.ServiceLoader
-import java.util.concurrent.ConcurrentHashMap
-
-import scala.collection.JavaConverters._
-
 object ScanTransformerFactory {
 
-  private val scanTransformerMap = new ConcurrentHashMap[String, Class[_]]()
-
   def createFileSourceScanTransformer(
       scanExec: FileSourceScanExec): FileSourceScanExecTransformerBase = {
-    val fileFormat = scanExec.relation.fileFormat
-    lookupDataSourceScanTransformer(fileFormat.getClass.getName) match {
-      case Some(clz) =>
-        clz
-          .getDeclaredConstructor()
-          .newInstance()
-          .asInstanceOf[DataSourceScanTransformerRegister]
-          .createDataSourceTransformer(scanExec)
-      case _ =>
-        FileSourceScanExecTransformer(
-          scanExec.relation,
-          scanExec.output,
-          scanExec.requiredSchema,
-          scanExec.partitionFilters,
-          scanExec.optionalBucketSet,
-          scanExec.optionalNumCoalescedBuckets,
-          scanExec.dataFilters,
-          scanExec.tableIdentifier,
-          scanExec.disableBucketedScan
-        )
-    }
+    FileSourceScanExecTransformer(
+      scanExec.relation,
+      scanExec.output,
+      scanExec.requiredSchema,
+      scanExec.partitionFilters,
+      scanExec.optionalBucketSet,
+      scanExec.optionalNumCoalescedBuckets,
+      scanExec.dataFilters,
+      scanExec.tableIdentifier,
+      scanExec.disableBucketedScan
+    )
   }
 
   def createBatchScanTransformer(batchScanExec: BatchScanExec): BatchScanExecTransformerBase = {
-    val scan = batchScanExec.scan
-    lookupDataSourceScanTransformer(scan.getClass.getName) match {
-      case Some(clz) =>
-        clz
-          .getDeclaredConstructor()
-          .newInstance()
-          .asInstanceOf[DataSourceScanTransformerRegister]
-          .createDataSourceV2Transformer(batchScanExec)
-      case _ =>
-        BatchScanExecTransformer(
-          batchScanExec.output,
-          batchScanExec.scan,
-          batchScanExec.runtimeFilters,
-          table = SparkShimLoader.getSparkShims.getBatchScanExecTable(batchScanExec)
-        )
-    }
-  }
-
-  private def lookupDataSourceScanTransformer(scanClassName: String): Option[Class[_]] = {
-    val clz = scanTransformerMap.computeIfAbsent(
-      scanClassName,
-      _ => {
-        val loader = Option(Thread.currentThread().getContextClassLoader)
-          .getOrElse(getClass.getClassLoader)
-        val serviceLoader = ServiceLoader.load(classOf[DataSourceScanTransformerRegister], loader)
-        serviceLoader.asScala
-          .filter(service => scanClassName.contains(service.scanClassName))
-          .toList match {
-          case head :: Nil =>
-            // there is exactly one registered alias
-            head.getClass
-          case _ => null
-        }
-      }
+    BatchScanExecTransformer(
+      batchScanExec.output,
+      batchScanExec.scan,
+      batchScanExec.runtimeFilters,
+      table = SparkShimLoader.getSparkShims.getBatchScanExecTable(batchScanExec)
     )
-    Option(clz)
   }
 }

gluten-delta/src-delta/main/scala/org/apache/gluten/execution/DeltaScanTransformerProvider.scala gluten-hudi/src-hudi/main/scala/org/apache/gluten/execution/OffloadHudiScan.scala

@@ -30,7 +30,6 @@
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-resources-plugin</artifactId>
-        <version>3.0.1</version>
       </plugin>
       <plugin>
         <groupId>net.alchim31.maven</groupId>

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

@@ -71,7 +71,6 @@
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-resources-plugin</artifactId>
-        <version>3.0.1</version>
       </plugin>
       <plugin>
         <groupId>net.alchim31.maven</groupId>

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

@@ -340,8 +340,8 @@
         <iceberg.version>1.5.0</iceberg.version>
         <delta.package.name>delta-spark</delta.package.name>
         <delta.version>3.2.0</delta.version>
-	    <delta.binary.version>32</delta.binary.version>
-	    <hudi.version>0.15.0</hudi.version>
+        <delta.binary.version>32</delta.binary.version>
+        <hudi.version>0.15.0</hudi.version>
         <fasterxml.version>2.15.1</fasterxml.version>
         <hadoop.version>3.3.4</hadoop.version>
         <antlr4.version>4.9.3</antlr4.version>
@@ -1200,11 +1200,6 @@
         <version>${fasterxml.version}</version>
         <scope>provided</scope>
       </dependency>
-      <dependency>
-        <groupId>org.apache.maven.plugins</groupId>
-        <artifactId>maven-source-plugin</artifactId>
-        <version>3.2.1</version>
-      </dependency>
     </dependencies>
   </dependencyManagement>