[Delta] fix UT

apache · Dec 14, 2023 · 77c25f3 · 77c25f3
1 parent e4015ab
commit 77c25f3
Show file tree

Hide file tree

Showing 4 changed files with 9 additions and 12 deletions.
diff --git a/gluten-core/src/main/scala/io/glutenproject/execution/ScanTransformerFactory.scala b/gluten-core/src/main/scala/io/glutenproject/execution/ScanTransformerFactory.scala
@@ -44,8 +44,8 @@ object ScanTransformerFactory {
     } else {
       ExpressionConverter.transformDynamicPruningExpr(scanExec.partitionFilters, reuseSubquery)
     }
-    val fileIndex = scanExec.relation.location
-    lookupDataSourceScanTransformer(fileIndex.getClass.getName) match {
+    val fileFormat = scanExec.relation.fileFormat
+    lookupDataSourceScanTransformer(fileFormat.getClass.getName) match {
       case Some(clz) =>
         clz
           .getDeclaredConstructor()

diff --git a/...ject/extension/DeltaScanTransformer.scala → ...ject/execution/DeltaScanTransformer.scala b/...ject/extension/DeltaScanTransformer.scala → ...ject/execution/DeltaScanTransformer.scala
@@ -14,9 +14,8 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.glutenproject.extension
+package io.glutenproject.execution
 
-import io.glutenproject.execution.FileSourceScanExecTransformer
 import io.glutenproject.substrait.rel.LocalFilesNode.ReadFileFormat
 
 import org.apache.spark.sql.catalyst.TableIdentifier

diff --git a/...ension/DeltaScanTransformerProvider.scala → ...cution/DeltaScanTransformerProvider.scala b/...ension/DeltaScanTransformerProvider.scala → ...cution/DeltaScanTransformerProvider.scala
@@ -14,16 +14,14 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.glutenproject.extension
-
-import io.glutenproject.execution.{DataSourceScanTransformerRegister, FileSourceScanExecTransformer}
+package io.glutenproject.execution
 
 import org.apache.spark.sql.catalyst.expressions.Expression
 import org.apache.spark.sql.execution.FileSourceScanExec
 
 class DeltaScanTransformerProvider extends DataSourceScanTransformerRegister {
 
-  override val scanClassName: String = "org.apache.spark.sql.delta.files.TahoeLogFileIndex"
+  override val scanClassName: String = "org.apache.spark.sql.delta.DeltaParquetFileFormat"
 
   override def createDataSourceTransformer(
       batchScan: FileSourceScanExec,

diff --git a/gluten-delta/src/main/scala/io/glutenproject/extension/DeltaRewriteTransformerRules.scala b/gluten-delta/src/main/scala/io/glutenproject/extension/DeltaRewriteTransformerRules.scala
@@ -16,7 +16,7 @@
  */
 package io.glutenproject.extension
 
-import io.glutenproject.execution.{FileSourceScanExecTransformer, ProjectExecTransformer}
+import io.glutenproject.execution.{DeltaScanTransformer, ProjectExecTransformer}
 import io.glutenproject.extension.DeltaRewriteTransformerRules.columnMappingRule
 import io.glutenproject.extension.columnar.TransformHints
 
@@ -53,7 +53,7 @@ object DeltaRewriteTransformerRules {
       // If it enables Delta Column Mapping(e.g. nameMapping and idMapping),
       // transform the metadata of Delta into Parquet's,
       // so that gluten can read Delta File using Parquet Reader.
-      case p: FileSourceScanExecTransformer
+      case p: DeltaScanTransformer
           if isDeltaColumnMappingFileFormat(p.relation.fileFormat) && notAppliedColumnMappingRule(
             p) =>
         transformColumnMappingPlan(p)
@@ -71,7 +71,7 @@ object DeltaRewriteTransformerRules {
    * transform the metadata of Delta into Parquet's, each plan should only be transformed once.
    */
   private def transformColumnMappingPlan(plan: SparkPlan): SparkPlan = plan match {
-    case plan: FileSourceScanExecTransformer =>
+    case plan: DeltaScanTransformer =>
       val fmt = plan.relation.fileFormat.asInstanceOf[DeltaParquetFileFormat]
       // a mapping between the table schemas name to parquet schemas.
       val columnNameMapping = mutable.Map.empty[String, String]
@@ -136,7 +136,7 @@ object DeltaRewriteTransformerRules {
       val newRequiredFields = plan.requiredSchema.map {
         e => StructField(columnNameMapping(e.name), e.dataType, e.nullable, e.metadata)
       }
-      val scanExecTransformer = new FileSourceScanExecTransformer(
+      val scanExecTransformer = new DeltaScanTransformer(
         newFsRelation,
         newOutput,
         StructType(newRequiredFields),