Convert nested fields to UnresolvedAttribute's to avoid schema pruning mismatches

Author: Kimahriman

core/src/main/scala/org/apache/spark/sql/delta/stats/DeltaScan.scala

@@ -20,7 +20,9 @@ import org.apache.spark.sql.delta.actions.AddFile
 import org.apache.spark.sql.delta.stats.DeltaDataSkippingType.DeltaDataSkippingType
 import com.fasterxml.jackson.databind.annotation.JsonDeserialize
 
+import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute
 import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.types.StructType
 
 /**
  * DataSize describes following attributes for data that consists of a list of input files
@@ -82,4 +84,54 @@ case class DeltaScan(
     val scanDurationMs: Long,
     val dataSkippingType: DeltaDataSkippingType) {
   def allFilters: ExpressionSet = partitionFilters ++ dataFilters ++ unusedFilters
+
+  /**
+   * Compare a set of filters to the filters for this DeltaScan. Because these filters could
+   * be post optimization, nested fields may have different schemas due to schema pruning. To
+   * get around this, we convert any nested field to an UnresolvedAttribute for the comparison.
+   *
+   * @param other ExpressionSet to compare the filters against
+   * @return Whether the expressions match with nested schemas ignored
+   */
+  def filtersMatch(other: ExpressionSet): Boolean = DeltaScan.filtersMatch(allFilters, other)
+}
+
+object DeltaScan {
+  private def constructSchema(source: StructType, ordinals: Seq[Int]): StructType = {
+    val extractedField = source.fields(ordinals.head)
+    val nestedType = if (ordinals.tail.nonEmpty) {
+      constructSchema(extractedField.dataType.asInstanceOf[StructType], ordinals.tail)
+    } else {
+      extractedField.dataType
+    }
+    StructType(Seq(extractedField.copy(dataType = nestedType)))
+  }
+
+  private def pruneExpression(expr: Expression): Expression = expr transform {
+    case NestedFieldExtraction(nameParts) =>
+      new UnresolvedAttribute(nameParts)
+  }
+
+  private[delta] def filtersMatch(source: ExpressionSet, target: ExpressionSet): Boolean = {
+    val prunedSource = source.map(pruneExpression _)
+    val prunedTarget = target.map(pruneExpression _)
+    prunedSource == prunedTarget
+  }
+}
+
+object NestedFieldExtraction {
+  def unapply(e: Expression): Option[Seq[String]] = e match {
+    case GetStructField(child, ordinal, _) =>
+      val nested = child match {
+        case NestedFieldExtraction(nameParts) => Some(nameParts)
+        case _ => None
+      }
+      val childSchema = child.dataType.asInstanceOf[StructType]
+      nested.map { nameParts =>
+        nameParts :+ childSchema.fields(ordinal).name
+      }
+    case a: AttributeReference if a.dataType.isInstanceOf[StructType] =>
+      Some(Seq(a.name))
+    case _ => None
+  }
 }

core/src/main/scala/org/apache/spark/sql/delta/stats/PrepareDeltaScan.scala

@@ -320,7 +320,7 @@ case class PreparedDeltaFileIndex(
       partitionFilters: Seq[Expression],
       dataFilters: Seq[Expression]): Seq[AddFile] = {
     val actualFilters = ExpressionSet(partitionFilters ++ dataFilters)
-    if (preparedScan.allFilters == actualFilters) {
+    if (preparedScan.filtersMatch(actualFilters)) {
       preparedScan.files.distinct
     } else {
       logInfo(

core/src/test/scala/org/apache/spark/sql/delta/stats/DataSkippingDeltaTests.scala

@@ -31,13 +31,41 @@ import org.scalatest.GivenWhenThen
 
 // scalastyle:off import.ordering.noEmptyLine
 import org.apache.spark.sql._
-import org.apache.spark.sql.catalyst.expressions.{Expression, Literal, PredicateHelper}
+import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.functions.{col, lit}
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.test.SharedSparkSession
 import org.apache.spark.sql.types._
 import org.apache.spark.util.Utils
 
+
+class DeltaScanSuite extends QueryTest
+    with SharedSparkSession
+    with PredicateHelper {
+
+  test("Comparing filters should ignore nested schema differences") {
+    // Simulate outer.inner.b = "abc" filter
+    val schema = StructType.fromDDL("inner STRUCT<a: STRING, b: STRING>, c LONG")
+    val prunedSchema = StructType.fromDDL("inner STRUCT<b: STRING>")
+
+    val originalAttr = AttributeReference("outer", schema, true)(ExprId(1))
+    val prunedAttr = AttributeReference("outer", prunedSchema, true)(ExprId(1))
+
+    val originalExprs = Seq(
+      GetStructField(GetStructField(originalAttr, 0), 1),
+      IsNotNull(originalAttr),
+      IsNotNull(GetStructField(originalAttr, 0))
+    )
+    val prunedExprs = Seq(
+      GetStructField(GetStructField(prunedAttr, 0), 0),
+      IsNotNull(prunedAttr),
+      IsNotNull(GetStructField(prunedAttr, 0))
+    )
+
+    assert(DeltaScan.filtersMatch(ExpressionSet(originalExprs), ExpressionSet(prunedExprs)))
+  }
+}
+
 trait DataSkippingDeltaTestsBase extends QueryTest
     with SharedSparkSession    with DeltaSQLCommandTest
     with PredicateHelper