[SPARK-26551][SQL] Fix schema pruning error when selecting one complex field and having is not null predicate on another one

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaPruning.scala

@@ -40,11 +40,19 @@ private[sql] object ParquetSchemaPruning extends Rule[LogicalPlan] {
       plan
     }
 
+  // `PhysicalOperation` pattern returns relation operator's outputs if there is no
+  // projects on it. In this case, we don't need to do schema pruning.
+  private def directOutput(projects: Seq[NamedExpression], plan: LogicalPlan): Boolean = {
+    projects.length == plan.output.length && projects.zip(plan.output).forall {
+      case (l, r) => l.name == r.name && l.dataType.sameType(r.dataType)
+    }
+  }
+
   private def apply0(plan: LogicalPlan): LogicalPlan =
     plan transformDown {
       case op @ PhysicalOperation(projects, filters,
           l @ LogicalRelation(hadoopFsRelation: HadoopFsRelation, _, _, _))
-        if canPruneRelation(hadoopFsRelation) =>
+        if canPruneRelation(hadoopFsRelation) && !directOutput(projects, l) =>
         val (normalizedProjects, normalizedFilters) =
           normalizeAttributeRefNames(l, projects, filters)
         val requestedRootFields = identifyRootFields(normalizedProjects, normalizedFilters)
@@ -119,7 +127,12 @@ private[sql] object ParquetSchemaPruning extends Rule[LogicalPlan] {
       .distinct.partition(_.contentAccessed)
 
     optRootFields.filter { opt =>
-      !rootFields.exists(_.field.name == opt.field.name)
+      !rootFields.exists { root =>
+        val rootFieldType = StructType(Array(root.field))
+        val optFieldType = StructType(Array(opt.field))
+        val merged = optFieldType.merge(rootFieldType)
+        root.field.name == opt.field.name && merged.sameType(optFieldType)
+      }
     } ++ rootFields
   }
 

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaPruningSuite.scala

@@ -25,6 +25,7 @@ import org.apache.spark.sql.{DataFrame, QueryTest, Row}
 import org.apache.spark.sql.catalyst.SchemaPruningTest
 import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
 import org.apache.spark.sql.execution.FileSourceScanExec
+import org.apache.spark.sql.functions._
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.test.SharedSQLContext
 import org.apache.spark.sql.types.StructType
@@ -217,6 +218,41 @@ class ParquetSchemaPruningSuite
       Row("Y.") :: Nil)
   }
 
+  testSchemaPruning("select one complex field and having is null predicate on another " +
+      "complex field") {
+    val query = sql("select * from contacts")
+      .where("name.middle is not null")
+      .select(
+        "id",
+        "name.first",
+        "name.middle",
+        "name.last"
+      )
+      .where("last = 'Jones'")
+      .select(count("id")).toDF()
+    checkScan(query,
+      "struct<id:int,name:struct<middle:string,last:string>>")
+    checkAnswer(query, Row(0) :: Nil)
+  }
+
+  testSchemaPruning("select one deep nested complex field and having is null predicate on " +
+      "another deep nested complex field") {
+    val query = sql("select * from contacts")
+      .where("employer.company.address is not null")
+      .selectExpr(
+        "id",
+        "name.first",
+        "name.middle",
+        "name.last",
+        "employer.id as employer_id"
+      )
+      .where("employer_id = 0")
+      .select(count("id")).toDF()
+    checkScan(query,
+      "struct<id:int,employer:struct<id:int,company:struct<address:string>>>")
+    checkAnswer(query, Row(1) :: Nil)
+  }
+
   private def testSchemaPruning(testName: String)(testThunk: => Unit) {
     test(s"Spark vectorized reader - without partition data column - $testName") {
       withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "true") {