[NSE-979] fix data source (#993)

* fix data source Signed-off-by: Yuan Zhou <[email protected]> * fix Signed-off-by: Yuan Zhou <[email protected]> * Fix ut issue Co-authored-by: philo <[email protected]>
oap-project · Jun 27, 2022 · 84511b5 · 84511b5
1 parent b28ec12
commit 84511b5
Show file tree

Hide file tree

Showing 3 changed files with 3 additions and 3 deletions.
diff --git a/.../src/main/scala/com/intel/oap/spark/sql/execution/datasources/arrow/ArrowFileFormat.scala b/.../src/main/scala/com/intel/oap/spark/sql/execution/datasources/arrow/ArrowFileFormat.scala
@@ -131,7 +131,7 @@ class ArrowFileFormat extends FileFormat with DataSourceRegister with Serializab
       // todo predicate validation / pushdown
       val parquetFileFields = factory.inspect().getFields.asScala
       val caseInsensitiveFieldMap = mutable.Map[String, String]()
-      val requiredFields = if (sqlConf.caseSensitiveAnalysis) {
+      val requiredFields = if (caseSensitive) {
         new Schema(requiredSchema.map { field =>
           parquetFileFields.find(_.getName.equals(field.name))
             .getOrElse(ArrowUtils.toArrowField(field))

diff --git a/.../com/intel/oap/spark/sql/execution/datasources/v2/arrow/ArrowPartitionReaderFactory.scala b/.../com/intel/oap/spark/sql/execution/datasources/v2/arrow/ArrowPartitionReaderFactory.scala
@@ -48,6 +48,7 @@ case class ArrowPartitionReaderFactory(
 
   private val batchSize = sqlConf.parquetVectorizedReaderBatchSize
   private val enableFilterPushDown: Boolean = sqlConf.arrowFilterPushDown
+  private val caseSensitive: Boolean = sqlConf.caseSensitiveAnalysis
 
   override def supportColumnarReads(partition: InputPartition): Boolean = true
 
@@ -63,7 +64,7 @@ case class ArrowPartitionReaderFactory(
       partitionedFile.start, partitionedFile.length, options)
     val parquetFileFields = factory.inspect().getFields.asScala
     val caseInsensitiveFieldMap = mutable.Map[String, String]()
-    val requiredFields = if (sqlConf.caseSensitiveAnalysis) {
+    val requiredFields = if (caseSensitive) {
       new Schema(readDataSchema.map { field =>
         parquetFileFields.find(_.getName.equals(field.name))
           .getOrElse(ArrowUtils.toArrowField(field))

diff --git a/.../test/scala/com/intel/oap/spark/sql/execution/datasources/arrow/ArrowDataSourceTest.scala b/.../test/scala/com/intel/oap/spark/sql/execution/datasources/arrow/ArrowDataSourceTest.scala
@@ -306,7 +306,6 @@ class ArrowDataSourceTest extends QueryTest with SharedSparkSession {
               "id"
             }
             val df = spark.read
-              .schema(s"$selectColName long")
               .arrow(tempPath.getPath)
               .filter(s"$selectColName <= 2")
             checkAnswer(df, Row(0) :: Row(1) :: Row(2) :: Nil)