Repository: apache/iceberg (language: python)

Issue:
Spec: Add section on `null_value_counts`
As mentioned in the community sync, there is some odd behavior on the null-count:

```sql
CREATE TABLE ...
TBLPROPERTIES (
    "write.format.default"="orc"
)
          SELECT named_struct('a', 1, 'b', 2) AS some_struct    -- null_value_counts={1: 0, 2: 0, 3: 0}
UNION ALL SELECT named_struct('a', 1, 'b', null) AS some_struct -- null_value_counts={1: 0, 2: 0, 3: 0}
UNION ALL SELECT null AS some_struct                            -- null_value_counts={1: 1, 2: 0, 3: 0}

CREATE TABLE ...
TBLPROPERTIES (
    "write.format.default"="parquet"
)
          SELECT named_struct('a', 1, 'b', 2) AS some_struct    -- null_value_counts={2: 0, 3: 0}
UNION ALL SELECT named_struct('a', 1, 'b', null) AS some_struct -- null_value_counts={2: 0, 3: 1}
UNION ALL SELECT null AS some_struct                            -- null_value_counts={2: 1, 3: 1}
```

The goal of this PR is to agree on how to count the nulls for nested objects.
Cannot delete column with nested field filter
### Query engine

Spark

### Question

I have a table with some nested fields, and tried to delete rows on spark sql.
But the operation was field with error like:
`java.lang.NullPointerException: Cannot filter by nested column`


For simplify I wrote scala test code
```scala
class NestedDeleteTest extends AnyFunSuite with Matchers {

  test("delete with nested field") {
    val catalogName = "hadoop"
    val dbName      = "test_db"
    val tableName   = "test_table"
    val tableId = s"$catalogName.$dbName.$tableName"

    val warehouse = s"${System.getProperty("user.dir")}/warehouse"

    val spark = SparkSession.builder
      .master("local[1]")
      .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
      .config(s"spark.sql.catalog.$catalogName", "org.apache.iceberg.spark.SparkCatalog")
      .config(s"spark.sql.catalog.$catalogName.type", "hadoop")
      .config(s"spark.sql.catalog.$catalogName.warehouse", warehouse)
      .getOrCreate()

    val rows = List(
      """{"nested":{"id":1},"timestamp":1}""",
      """{"nested":{"id":2},"timestamp":1}""",
      """{"nested":{"id":1},"timestamp":2}"""
    )

    val df = spark.read.json(spark.createDataset(spark.sparkContext.parallelize(rows))(Encoders.STRING))

    df.writeTo(tableId).createOrReplace()

    val before = spark.table(tableId)
    before.show()
    before.count() shouldBe 3

    val sql = s"DELETE FROM $tableId WHERE nested.id=1 AND timestamp=2"
    spark.sql(sql)

    val after = spark.table(tableId)
    after.show()
    after.count() shouldBe 2
  }
}
```

Then, the problem is reproduced.

stacktrace:
```
java.lang.NullPointerException: Cannot filter by nested column: 5: id: optional long
	at org.apache.iceberg.relocated.com.google.common.base.Preconditions.checkNotNull(Preconditions.java:994)
	at org.apache.iceberg.expressions.StrictMetricsEvaluator$MetricsEvalVisitor.eq(StrictMetricsEvaluator.java:305)
	at org.apache.iceberg.expressions.StrictMetricsEvaluator$MetricsEvalVisitor.eq(StrictMetricsEvaluator.java:83)
	at org.apache.iceberg.expressions.ExpressionVisitors$BoundExpressionVisitor.predicate(ExpressionVisitors.java:162)
	at org.apache.iceberg.expressions.ExpressionVisitors.visitEvaluator(ExpressionVisitors.java:390)
	at org.apache.iceberg.expressions.ExpressionVisitors.visitEvaluator(ExpressionVisitors.java:405)
	at org.apache.iceberg.expressions.StrictMetricsEvaluator$MetricsEvalVisitor.eval(StrictMetricsEvaluator.java:101)
	at org.apache.iceberg.expressions.StrictMetricsEvaluator$MetricsEvalVisitor.access$100(StrictMetricsEvaluator.java:83)
	at org.apache.iceberg.expressions.StrictMetricsEvaluator.eval(StrictMetricsEvaluator.java:77)
	at org.apache.iceberg.spark.source.SparkTable.lambda$canDeleteUsingMetadata$2(SparkTable.java:300)
	at org.apache.iceberg.relocated.com.google.common.collect.Iterators.all(Iterators.java:710)
	at org.apache.iceberg.relocated.com.google.common.collect.Iterables.all(Iterables.java:645)
	at org.apache.iceberg.spark.source.SparkTable.canDeleteUsingMetadata(SparkTable.java:289)
	at org.apache.iceberg.spark.source.SparkTable.canDeleteWhere(SparkTable.java:269)
	at org.apache.spark.sql.execution.datasources.v2.OptimizeMetadataOnlyDeleteFromTable$$anonfun$apply$1.applyOrElse(OptimizeMetadataOnlyDeleteFromTable.scala:53)
	at org.apache.spark.sql.execution.datasources.v2.OptimizeMetadataOnlyDeleteFromTable$$anonfun$apply$1.applyOrElse(OptimizeMetadataOnlyDeleteFromTable.scala:44)
	at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDownWithPruning$1(TreeNode.scala:481)
	at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:82)
	at org.apache.spark.sql.catalyst.trees.TreeNode.transformDownWithPruning(TreeNode.scala:481)
	at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDownWithPruning(LogicalPlan.scala:30)
	at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning(AnalysisHelper.scala:267)
	at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning$(AnalysisHelper.scala:263)
	at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:30)
	at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:30)
	at org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:457)
	at org.apache.spark.sql.catalyst.trees.TreeNode.transform(TreeNode.scala:425)
	at org.apache.spark.sql.execution.datasources.v2.OptimizeMetadataOnlyDeleteFromTable$.apply(OptimizeMetadataOnlyDeleteFromTable.scala:44)
	at org.apache.spark.sql.execution.datasources.v2.OptimizeMetadataOnlyDeleteFromTable$.apply(OptimizeMetadataOnlyDeleteFromTable.scala:40)
	at org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$execute$2(RuleExecutor.scala:211)
	at scala.collection.LinearSeqOptimized.foldLeft(LinearSeqOptimized.scala:126)
	at scala.collection.LinearSeqOptimized.foldLeft$(LinearSeqOptimized.scala:122)
	at scala.collection.immutable.List.foldLeft(List.scala:91)
	at org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$execute$1(RuleExecutor.scala:208)
	at org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$execute$1$adapted(RuleExecutor.scala:200)
	at scala.collection.immutable.List.foreach(List.scala:431)
	at org.apache.spark.sql.catalyst.rules.RuleExecutor.execute(RuleExecutor.scala:200)
	at org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$executeAndTrack$1(RuleExecutor.scala:179)
	at org.apache.spark.sql.catalyst.QueryPlanningTracker$.withTracker(QueryPlanningTracker.scala:88)
	at org.apache.spark.sql.catalyst.rules.RuleExecutor.executeAndTrack(RuleExecutor.scala:179)
	at org.apache.spark.sql.execution.QueryExecution.$anonfun$optimizedPlan$1(QueryExecution.scala:125)
	at org.apache.spark.sql.catalyst.QueryPlanningTracker.measurePhase(QueryPlanningTracker.scala:111)
	at org.apache.spark.sql.execution.QueryExecution.$anonfun$executePhase$1(QueryExecution.scala:183)
	at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:775)
	at org.apache.spark.sql.execution.QueryExecution.executePhase(QueryExecution.scala:183)
	at org.apache.spark.sql.execution.QueryExecution.optimizedPlan$lzycompute(QueryExecution.scala:121)
	at org.apache.spark.sql.execution.QueryExecution.optimizedPlan(QueryExecution.scala:117)
	at org.apache.spark.sql.execution.QueryExecution.assertOptimized(QueryExecution.scala:135)
	at org.apache.spark.sql.execution.QueryExecution.executedPlan$lzycompute(QueryExecution.scala:153)
	at org.apache.spark.sql.execution.QueryExecution.executedPlan(QueryExecution.scala:150)
	at org.apache.spark.sql.execution.QueryExecution.simpleString(QueryExecution.scala:201)
	at org.apache.spark.sql.execution.QueryExecution.org$apache$spark$sql$execution$QueryExecution$$explainString(QueryExecution.scala:246)
	at org.apache.spark.sql.execution.QueryExecution.explainString(QueryExecution.scala:215)
	at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$5(SQLExecution.scala:98)
	at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:163)
	at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:90)
	at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:775)
	at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:64)
	at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.applyOrElse(QueryExecution.scala:97)
	at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.applyOrElse(QueryExecution.scala:93)
	at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDownWithPruning$1(TreeNode.scala:481)
	at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:82)
	at org.apache.spark.sql.catalyst.trees.TreeNode.transformDownWithPruning(TreeNode.scala:481)
	at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDownWithPruning(LogicalPlan.scala:30)
	at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning(AnalysisHelper.scala:267)
	at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning$(AnalysisHelper.scala:263)
	at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:30)
	at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:30)
	at org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:457)
	at org.apache.spark.sql.execution.QueryExecution.eagerlyExecuteCommands(QueryExecution.scala:93)
	at org.apache.spark.sql.execution.QueryExecution.commandExecuted$lzycompute(QueryExecution.scala:80)
	at org.apache.spark.sql.execution.QueryExecution.commandExecuted(QueryExecution.scala:78)
	at org.apache.spark.sql.Dataset.<init>(Dataset.scala:219)
	at org.apache.spark.sql.Dataset$.$anonfun$ofRows$2(Dataset.scala:99)
	at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:775)
	at org.apache.spark.sql.Dataset$.ofRows(Dataset.scala:96)
	at org.apache.spark.sql.SparkSession.$anonfun$sql$1(SparkSession.scala:618)
	at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:775)
	at org.apache.spark.sql.SparkSession.sql(SparkSession.scala:613)
```

Is it normal operation?
I am confused because I found test case for delete with nested column
https://github.com/apache/iceberg/blob/39a2c12b843f6f16e3389e738531e24a32f5bf39/spark/v3.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java#L383-L397


Discussion / hints:
> I do think we should reconsider a patch to rescan footers and rewrite iceberg metadata since I think we will basically have tables which have previously written iceberg metrics with null counts that do not follow this spec

+1 on this there could be engines using these stats 
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.
This pull request has been closed due to lack of activity. This is not a judgement on the merit of the PR in any way. It is just a way of keeping the PR queue manageable. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.
I debugged this test locally, and found that it didn't run `canDeleteWhere`, I think this is because the expression in the test cannot be pushed down, after I changed the expression to `complex.c1 = 2`, it threw the same exception. I think exception is unexpected, since this test case already shows that deleting via filter on nested fields is possible (without running `canDeleteUsingMetadata`). But this exception is thrown because `StrictMetricsEvaluator ` cannot handle nested fields. 
I think we can make `StrictMetricsEvaluator` handle nested fileds properly to solve this, I think we can remove the nested fields check in `StrictMetricsEvaluator` directly, it seems safe to me, @rdblue can you please confirm this?
This issue has been automatically marked as stale because it has been open for 180 days with no activity. It will be closed in next 14 days if no further activity occurs. To permanently prevent this issue from being considered stale, add the label 'not-stale', but commenting on the issue is preferred when possible.
This issue has been closed because it has not received any activity in the last 14 days since being marked as 'stale'

Fix patch:
diff --git a/api/src/main/java/org/apache/iceberg/expressions/StrictMetricsEvaluator.java b/api/src/main/java/org/apache/iceberg/expressions/StrictMetricsEvaluator.java
index 4aee75c447d3..1a5a884f651a 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/StrictMetricsEvaluator.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/StrictMetricsEvaluator.java
@@ -29,9 +29,7 @@
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.expressions.ExpressionVisitors.BoundExpressionVisitor;
-import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.types.Conversions;
-import org.apache.iceberg.types.Types;
 import org.apache.iceberg.types.Types.StructType;
 import org.apache.iceberg.util.NaNUtil;
 
@@ -51,7 +49,6 @@
  * checks for NaN is necessary in order to not include files that may contain rows that don't match.
  */
 public class StrictMetricsEvaluator {
-  private final Schema schema;
   private final StructType struct;
   private final Expression expr;
 
@@ -60,7 +57,6 @@ public StrictMetricsEvaluator(Schema schema, Expression unbound) {
   }
 
   public StrictMetricsEvaluator(Schema schema, Expression unbound, boolean caseSensitive) {
-    this.schema = schema;
     this.struct = schema.asStruct();
     this.expr = Binder.bind(struct, rewriteNot(unbound), caseSensitive);
   }
@@ -144,8 +140,9 @@ public <T> Boolean isNull(BoundReference<T> ref) {
       // no need to check whether the field is required because binding evaluates that case
       // if the column has any non-null values, the expression does not match
       int id = ref.fieldId();
-      Preconditions.checkNotNull(
-          struct.field(id), "Cannot filter by nested column: %s", schema.findField(id));
+      if (isNestedColumn(id)) {
+        return ROWS_MIGHT_NOT_MATCH;
+      }
 
       if (containsNullsOnly(id)) {
         return ROWS_MUST_MATCH;
@@ -159,8 +156,9 @@ public <T> Boolean notNull(BoundReference<T> ref) {
       // no need to check whether the field is required because binding evaluates that case
       // if the column has any null values, the expression does not match
       int id = ref.fieldId();
-      Preconditions.checkNotNull(
-          struct.field(id), "Cannot filter by nested column: %s", schema.findField(id));
+      if (isNestedColumn(id)) {
+        return ROWS_MIGHT_NOT_MATCH;
+      }
 
       if (nullCounts != null && nullCounts.containsKey(id) && nullCounts.get(id) == 0) {
         return ROWS_MUST_MATCH;
@@ -199,15 +197,16 @@ public <T> Boolean notNaN(BoundReference<T> ref) {
     public <T> Boolean lt(BoundReference<T> ref, Literal<T> lit) {
       // Rows must match when: <----------Min----Max---X------->
       Integer id = ref.fieldId();
-      Types.NestedField field = struct.field(id);
-      Preconditions.checkNotNull(field, "Cannot filter by nested column: %s", schema.findField(id));
+      if (isNestedColumn(id)) {
+        return ROWS_MIGHT_NOT_MATCH;
+      }
 
       if (canContainNulls(id) || canContainNaNs(id)) {
         return ROWS_MIGHT_NOT_MATCH;
       }
 
       if (upperBounds != null && upperBounds.containsKey(id)) {
-        T upper = Conversions.fromByteBuffer(field.type(), upperBounds.get(id));
+        T upper = Conversions.fromByteBuffer(ref.type(), upperBounds.get(id));
 
         int cmp = lit.comparator().compare(upper, lit.value());
         if (cmp < 0) {
@@ -222,15 +221,16 @@ public <T> Boolean lt(BoundReference<T> ref, Literal<T> lit) {
     public <T> Boolean ltEq(BoundReference<T> ref, Literal<T> lit) {
       // Rows must match when: <----------Min----Max---X------->
       Integer id = ref.fieldId();
-      Types.NestedField field = struct.field(id);
-      Preconditions.checkNotNull(field, "Cannot filter by nested column: %s", schema.findField(id));
+      if (isNestedColumn(id)) {
+        return ROWS_MIGHT_NOT_MATCH;
+      }
 
       if (canContainNulls(id) || canContainNaNs(id)) {
         return ROWS_MIGHT_NOT_MATCH;
       }
 
       if (upperBounds != null && upperBounds.containsKey(id)) {
-        T upper = Conversions.fromByteBuffer(field.type(), upperBounds.get(id));
+        T upper = Conversions.fromByteBuffer(ref.type(), upperBounds.get(id));
 
         int cmp = lit.comparator().compare(upper, lit.value());
         if (cmp <= 0) {
@@ -245,15 +245,16 @@ public <T> Boolean ltEq(BoundReference<T> ref, Literal<T> lit) {
     public <T> Boolean gt(BoundReference<T> ref, Literal<T> lit) {
       // Rows must match when: <-------X---Min----Max---------->
       Integer id = ref.fieldId();
-      Types.NestedField field = struct.field(id);
-      Preconditions.checkNotNull(field, "Cannot filter by nested column: %s", schema.findField(id));
+      if (isNestedColumn(id)) {
+        return ROWS_MIGHT_NOT_MATCH;
+      }
 
       if (canContainNulls(id) || canContainNaNs(id)) {
         return ROWS_MIGHT_NOT_MATCH;
       }
 
       if (lowerBounds != null && lowerBounds.containsKey(id)) {
-        T lower = Conversions.fromByteBuffer(field.type(), lowerBounds.get(id));
+        T lower = Conversions.fromByteBuffer(ref.type(), lowerBounds.get(id));
 
         if (NaNUtil.isNaN(lower)) {
           // NaN indicates unreliable bounds. See the StrictMetricsEvaluator docs for more.
@@ -273,15 +274,16 @@ public <T> Boolean gt(BoundReference<T> ref, Literal<T> lit) {
     public <T> Boolean gtEq(BoundReference<T> ref, Literal<T> lit) {
       // Rows must match when: <-------X---Min----Max---------->
       Integer id = ref.fieldId();
-      Types.NestedField field = struct.field(id);
-      Preconditions.checkNotNull(field, "Cannot filter by nested column: %s", schema.findField(id));
+      if (isNestedColumn(id)) {
+        return ROWS_MIGHT_NOT_MATCH;
+      }
 
       if (canContainNulls(id) || canContainNaNs(id)) {
         return ROWS_MIGHT_NOT_MATCH;
       }
 
       if (lowerBounds != null && lowerBounds.containsKey(id)) {
-        T lower = Conversions.fromByteBuffer(field.type(), lowerBounds.get(id));
+        T lower = Conversions.fromByteBuffer(ref.type(), lowerBounds.get(id));
 
         if (NaNUtil.isNaN(lower)) {
           // NaN indicates unreliable bounds. See the StrictMetricsEvaluator docs for more.
@@ -301,8 +303,9 @@ public <T> Boolean gtEq(BoundReference<T> ref, Literal<T> lit) {
     public <T> Boolean eq(BoundReference<T> ref, Literal<T> lit) {
       // Rows must match when Min == X == Max
       Integer id = ref.fieldId();
-      Types.NestedField field = struct.field(id);
-      Preconditions.checkNotNull(field, "Cannot filter by nested column: %s", schema.findField(id));
+      if (isNestedColumn(id)) {
+        return ROWS_MIGHT_NOT_MATCH;
+      }
 
       if (canContainNulls(id) || canContainNaNs(id)) {
         return ROWS_MIGHT_NOT_MATCH;
@@ -319,7 +322,7 @@ public <T> Boolean eq(BoundReference<T> ref, Literal<T> lit) {
           return ROWS_MIGHT_NOT_MATCH;
         }
 
-        T upper = Conversions.fromByteBuffer(field.type(), upperBounds.get(id));
+        T upper = Conversions.fromByteBuffer(ref.type(), upperBounds.get(id));
 
         cmp = lit.comparator().compare(upper, lit.value());
         if (cmp != 0) {
@@ -336,8 +339,9 @@ public <T> Boolean eq(BoundReference<T> ref, Literal<T> lit) {
     public <T> Boolean notEq(BoundReference<T> ref, Literal<T> lit) {
       // Rows must match when X < Min or Max < X because it is not in the range
       Integer id = ref.fieldId();
-      Types.NestedField field = struct.field(id);
-      Preconditions.checkNotNull(field, "Cannot filter by nested column: %s", schema.findField(id));
+      if (isNestedColumn(id)) {
+        return ROWS_MIGHT_NOT_MATCH;
+      }
 
       if (containsNullsOnly(id) || containsNaNsOnly(id)) {
         return ROWS_MUST_MATCH;
@@ -358,7 +362,7 @@ public <T> Boolean notEq(BoundReference<T> ref, Literal<T> lit) {
       }
 
       if (upperBounds != null && upperBounds.containsKey(id)) {
-        T upper = Conversions.fromByteBuffer(field.type(), upperBounds.get(id));
+        T upper = Conversions.fromByteBuffer(ref.type(), upperBounds.get(id));
 
         int cmp = lit.comparator().compare(upper, lit.value());
         if (cmp < 0) {
@@ -372,8 +376,9 @@ public <T> Boolean notEq(BoundReference<T> ref, Literal<T> lit) {
     @Override
     public <T> Boolean in(BoundReference<T> ref, Set<T> literalSet) {
       Integer id = ref.fieldId();
-      Types.NestedField field = struct.field(id);
-      Preconditions.checkNotNull(field, "Cannot filter by nested column: %s", schema.findField(id));
+      if (isNestedColumn(id)) {
+        return ROWS_MIGHT_NOT_MATCH;
+      }
 
       if (canContainNulls(id) || canContainNaNs(id)) {
         return ROWS_MIGHT_NOT_MATCH;
@@ -390,7 +395,7 @@ public <T> Boolean in(BoundReference<T> ref, Set<T> literalSet) {
         }
 
         // check if the upper bound is in the set
-        T upper = Conversions.fromByteBuffer(field.type(), upperBounds.get(id));
+        T upper = Conversions.fromByteBuffer(ref.type(), upperBounds.get(id));
         if (!literalSet.contains(upper)) {
           return ROWS_MIGHT_NOT_MATCH;
         }
@@ -411,8 +416,9 @@ public <T> Boolean in(BoundReference<T> ref, Set<T> literalSet) {
     @Override
     public <T> Boolean notIn(BoundReference<T> ref, Set<T> literalSet) {
       Integer id = ref.fieldId();
-      Types.NestedField field = struct.field(id);
-      Preconditions.checkNotNull(field, "Cannot filter by nested column: %s", schema.findField(id));
+      if (isNestedColumn(id)) {
+        return ROWS_MIGHT_NOT_MATCH;
+      }
 
       if (containsNullsOnly(id) || containsNaNsOnly(id)) {
         return ROWS_MUST_MATCH;
@@ -439,7 +445,7 @@ public <T> Boolean notIn(BoundReference<T> ref, Set<T> literalSet) {
       }
 
       if (upperBounds != null && upperBounds.containsKey(id)) {
-        T upper = Conversions.fromByteBuffer(field.type(), upperBounds.get(id));
+        T upper = Conversions.fromByteBuffer(ref.type(), upperBounds.get(id));
         literals =
             literals.stream()
                 .filter(v -> ref.comparator().compare(upper, v) >= 0)
@@ -466,6 +472,10 @@ public <T> Boolean notStartsWith(BoundReference<T> ref, Literal<T> lit) {
       return ROWS_MIGHT_NOT_MATCH;
     }
 
+    private boolean isNestedColumn(int id) {
+      return struct.field(id) == null;
+    }
+
     private boolean canContainNulls(Integer id) {
       return nullCounts == null || (nullCounts.containsKey(id) && nullCounts.get(id) > 0);
     }


Test patch:
diff --git a/api/src/test/java/org/apache/iceberg/expressions/TestStrictMetricsEvaluator.java b/api/src/test/java/org/apache/iceberg/expressions/TestStrictMetricsEvaluator.java
index a23afb15dde5..f34cd730df77 100644
--- a/api/src/test/java/org/apache/iceberg/expressions/TestStrictMetricsEvaluator.java
+++ b/api/src/test/java/org/apache/iceberg/expressions/TestStrictMetricsEvaluator.java
@@ -66,7 +66,14 @@ public class TestStrictMetricsEvaluator {
           optional(11, "all_nulls_double", Types.DoubleType.get()),
           optional(12, "all_nans_v1_stats", Types.FloatType.get()),
           optional(13, "nan_and_null_only", Types.DoubleType.get()),
-          optional(14, "no_nan_stats", Types.DoubleType.get()));
+          optional(14, "no_nan_stats", Types.DoubleType.get()),
+          optional(
+              15,
+              "struct",
+              Types.StructType.of(
+                  Types.NestedField.optional(16, "nested_col_no_stats", Types.IntegerType.get()),
+                  Types.NestedField.optional(
+                      17, "nested_col_with_stats", Types.IntegerType.get()))));
 
   private static final int INT_MIN_VALUE = 30;
   private static final int INT_MAX_VALUE = 79;
@@ -88,6 +95,7 @@ public class TestStrictMetricsEvaluator {
               .put(12, 50L)
               .put(13, 50L)
               .put(14, 50L)
+              .put(17, 50L)
               .buildOrThrow(),
           // null value counts
           ImmutableMap.<Integer, Long>builder()
@@ -97,6 +105,7 @@ public class TestStrictMetricsEvaluator {
               .put(11, 50L)
               .put(12, 0L)
               .put(13, 1L)
+              .put(17, 0L)
               .buildOrThrow(),
           // nan value counts
           ImmutableMap.of(
@@ -108,13 +117,15 @@ public class TestStrictMetricsEvaluator {
               1, toByteBuffer(IntegerType.get(), INT_MIN_VALUE),
               7, toByteBuffer(IntegerType.get(), 5),
               12, toByteBuffer(Types.FloatType.get(), Float.NaN),
-              13, toByteBuffer(Types.DoubleType.get(), Double.NaN)),
+              13, toByteBuffer(Types.DoubleType.get(), Double.NaN),
+              17, toByteBuffer(Types.IntegerType.get(), INT_MIN_VALUE)),
           // upper bounds
           ImmutableMap.of(
               1, toByteBuffer(IntegerType.get(), INT_MAX_VALUE),
               7, toByteBuffer(IntegerType.get(), 5),
               12, toByteBuffer(Types.FloatType.get(), Float.NaN),
-              13, toByteBuffer(Types.DoubleType.get(), Double.NaN)));
+              13, toByteBuffer(Types.DoubleType.get(), Double.NaN),
+              17, toByteBuffer(IntegerType.get(), INT_MAX_VALUE)));
 
   private static final DataFile FILE_2 =
       new TestDataFile(
@@ -627,4 +638,50 @@ public void testIntegerNotIn() {
     shouldRead = new StrictMetricsEvaluator(SCHEMA, notIn("no_nulls", "abc", "def")).eval(FILE);
     assertThat(shouldRead).as("Should not match: no_nulls field does not have bounds").isFalse();
   }
+
+  @Test
+  public void testEvaluateOnNestedColumnWithoutStats() {
+    boolean shouldRead =
+        new StrictMetricsEvaluator(
+                SCHEMA, greaterThanOrEqual("struct.nested_col_no_stats", INT_MIN_VALUE))
+            .eval(FILE);
+    assertThat(shouldRead).as("greaterThanOrEqual nested column should not match").isFalse();
+
+    shouldRead =
+        new StrictMetricsEvaluator(
+                SCHEMA, lessThanOrEqual("struct.nested_col_no_stats", INT_MAX_VALUE))
+            .eval(FILE);
+    assertThat(shouldRead).as("lessThanOrEqual nested column should not match").isFalse();
+
+    shouldRead =
+        new StrictMetricsEvaluator(SCHEMA, isNull("struct.nested_col_no_stats")).eval(FILE);
+    assertThat(shouldRead).as("isNull nested column should not match").isFalse();
+
+    shouldRead =
+        new StrictMetricsEvaluator(SCHEMA, notNull("struct.nested_col_no_stats")).eval(FILE);
+    assertThat(shouldRead).as("notNull nested column should not match").isFalse();
+  }
+
+  @Test
+  public void testEvaluateOnNestedColumnWithStats() {
+    boolean shouldRead =
+        new StrictMetricsEvaluator(
+                SCHEMA, greaterThanOrEqual("struct.nested_col_with_stats", INT_MIN_VALUE))
+            .eval(FILE);
+    assertThat(shouldRead).as("greaterThanOrEqual nested column should not match").isFalse();
+
+    shouldRead =
+        new StrictMetricsEvaluator(
+                SCHEMA, lessThanOrEqual("struct.nested_col_with_stats", INT_MAX_VALUE))
+            .eval(FILE);
+    assertThat(shouldRead).as("lessThanOrEqual nested column should not match").isFalse();
+
+    shouldRead =
+        new StrictMetricsEvaluator(SCHEMA, isNull("struct.nested_col_with_stats")).eval(FILE);
+    assertThat(shouldRead).as("isNull nested column should not match").isFalse();
+
+    shouldRead =
+        new StrictMetricsEvaluator(SCHEMA, notNull("struct.nested_col_with_stats")).eval(FILE);
+    assertThat(shouldRead).as("notNull nested column should not match").isFalse();
+  }
 }
diff --git a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
index d258079541cd..42eb2af774e9 100644
--- a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
+++ b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
@@ -1401,6 +1401,28 @@ public void testDeleteToCustomWapBranchWithoutWhereClause() throws NoSuchTableEx
         });
   }
 
+  @TestTemplate
+  public void testDeleteWithFilterOnNestedColumn() {
+    createAndInitNestedColumnsTable();
+
+    sql("INSERT INTO TABLE %s VALUES (1, named_struct(\"c1\", 3, \"c2\", \"v1\"))", tableName);
+    sql("INSERT INTO TABLE %s VALUES (2, named_struct(\"c1\", 2, \"c2\", \"v2\"))", tableName);
+
+    sql("DELETE FROM %s WHERE complex.c1 > 3", tableName);
+    assertEquals(
+        "Should have expected rows",
+        ImmutableList.of(row(1), row(2)),
+        sql("SELECT id FROM %s order by id", tableName));
+
+    sql("DELETE FROM %s WHERE complex.c1 = 3", tableName);
+    assertEquals(
+        "Should have expected rows", ImmutableList.of(row(2)), sql("SELECT id FROM %s", tableName));
+
+    sql("DELETE FROM %s t WHERE t.complex.c1 = 2", tableName);
+    assertEquals(
+        "Should have expected rows", ImmutableList.of(), sql("SELECT id FROM %s", tableName));
+  }
+
   // TODO: multiple stripes for ORC
 
   protected void createAndInitPartitionedTable() {