fix: respect ancestor nullability when folding IS NULL/NOT NULL predicates (#995)

A required field nested in an optional struct, list, or map can be null when its
parent is null, so IS NULL and NOT NULL must not be folded for it. Only fold
when the field and all of its ancestors are required, matching Java's
UnboundPredicate#bindUnaryOperation.

Author: wgtmac

src/iceberg/expression/predicate.cc

@@ -25,11 +25,13 @@
 #include "iceberg/expression/expressions.h"
 #include "iceberg/expression/literal.h"
 #include "iceberg/result.h"
+#include "iceberg/schema.h"
 #include "iceberg/transform.h"
 #include "iceberg/type.h"
 #include "iceberg/util/checked_cast.h"
 #include "iceberg/util/formatter_internal.h"
 #include "iceberg/util/macros.h"
+#include "iceberg/util/type_util.h"
 
 namespace iceberg {
 
@@ -230,7 +232,7 @@ Result<std::shared_ptr<Expression>> UnboundPredicateImpl<B>::Bind(
   ICEBERG_ASSIGN_OR_RAISE(auto bound_term, BASE::term()->Bind(schema, case_sensitive));
 
   if (values_.empty()) {
-    return BindUnaryOperation(std::move(bound_term));
+    return BindUnaryOperation(schema, std::move(bound_term));
   }
 
   if (BASE::op() == Expression::Operation::kIn ||
@@ -257,30 +259,63 @@ bool StartsWith(const Literal& lhs, const Literal& rhs) {
   return false;
 }
 
+/// \brief Returns true if all ancestors of the field with the given ID are required.
+///
+/// A required field can still produce null values if it is nested in an optional struct,
+/// list, or map.
+Result<bool> AllAncestorsAreRequired(const Schema& schema, int32_t field_id) {
+  auto id_to_parent = IndexParents(schema);
+  auto parent_it = id_to_parent.find(field_id);
+  while (parent_it != id_to_parent.end()) {
+    ICEBERG_ASSIGN_OR_RAISE(auto parent, schema.FindFieldById(parent_it->second));
+    if (!parent.has_value()) {
+      return InvalidSchema("Cannot find parent field {} of field {}", parent_it->second,
+                           field_id);
+    }
+    if (parent.value().get().optional()) {
+      return false;
+    }
+    parent_it = id_to_parent.find(parent_it->second);
+  }
+  return true;
+}
+
 }  // namespace
 
 template <typename B>
 Result<std::shared_ptr<Expression>> UnboundPredicateImpl<B>::BindUnaryOperation(
-    std::shared_ptr<B> bound_term) const {
+    const Schema& schema, std::shared_ptr<B> bound_term) const {
   switch (BASE::op()) {
-    case Expression::Operation::kIsNull:
+    case Expression::Operation::kIsNull: {
       if (!bound_term->MayProduceNull()) {
-        return Expressions::AlwaysFalse();
+        ICEBERG_ASSIGN_OR_RAISE(
+            bool all_ancestors_required,
+            AllAncestorsAreRequired(schema, bound_term->reference()->field_id()));
+        if (all_ancestors_required) {
+          return Expressions::AlwaysFalse();
+        }
       }
       if (bound_term->type()->type_id() == TypeId::kUnknown) {
         return Expressions::AlwaysTrue();
       }
       return BoundUnaryPredicate::Make(Expression::Operation::kIsNull,
                                        std::move(bound_term));
-    case Expression::Operation::kNotNull:
+    }
+    case Expression::Operation::kNotNull: {
       if (!bound_term->MayProduceNull()) {
-        return Expressions::AlwaysTrue();
+        ICEBERG_ASSIGN_OR_RAISE(
+            bool all_ancestors_required,
+            AllAncestorsAreRequired(schema, bound_term->reference()->field_id()));
+        if (all_ancestors_required) {
+          return Expressions::AlwaysTrue();
+        }
       }
       if (bound_term->type()->type_id() == TypeId::kUnknown) {
         return Expressions::AlwaysFalse();
       }
       return BoundUnaryPredicate::Make(Expression::Operation::kNotNull,
                                        std::move(bound_term));
+    }
     case Expression::Operation::kIsNan:
     case Expression::Operation::kNotNan:
       if (!IsFloatingType(bound_term->type()->type_id())) {

src/iceberg/expression/predicate.h

@@ -150,7 +150,7 @@ class ICEBERG_EXPORT UnboundPredicateImpl : public UnboundPredicate,
                        std::vector<Literal> values);
 
   Result<std::shared_ptr<Expression>> BindUnaryOperation(
-      std::shared_ptr<B> bound_term) const;
+      const Schema& schema, std::shared_ptr<B> bound_term) const;
   Result<std::shared_ptr<Expression>> BindLiteralOperation(
       std::shared_ptr<B> bound_term) const;
   Result<std::shared_ptr<Expression>> BindInOperation(

src/iceberg/test/predicate_test.cc

@@ -384,6 +384,45 @@ TEST_F(PredicateTest, UnboundPredicateBindUnaryWithUnknownType) {
   EXPECT_EQ(bound_not_null->op(), Expression::Operation::kFalse);
 }
 
+TEST_F(PredicateTest, UnboundPredicateBindUnaryWithNestedStructs) {
+  // A required field nested in an optional struct can still produce null values
+  auto nested_schema = std::make_shared<Schema>(
+      std::vector<SchemaField>{
+          SchemaField::MakeOptional(1, "optional_parent",
+                                    std::make_shared<StructType>(std::vector<SchemaField>{
+                                        SchemaField::MakeRequired(2, "leaf", int32())})),
+          SchemaField::MakeRequired(3, "required_parent",
+                                    std::make_shared<StructType>(std::vector<SchemaField>{
+                                        SchemaField::MakeRequired(4, "leaf", int32())})),
+      },
+      /*schema_id=*/0);
+
+  auto is_null_optional_parent = Expressions::IsNull("optional_parent.leaf");
+  auto bound_is_null_optional =
+      is_null_optional_parent->Bind(*nested_schema, /*case_sensitive=*/true);
+  ASSERT_THAT(bound_is_null_optional, IsOk());
+  EXPECT_EQ(bound_is_null_optional.value()->op(), Expression::Operation::kIsNull);
+
+  auto not_null_optional_parent = Expressions::NotNull("optional_parent.leaf");
+  auto bound_not_null_optional =
+      not_null_optional_parent->Bind(*nested_schema, /*case_sensitive=*/true);
+  ASSERT_THAT(bound_not_null_optional, IsOk());
+  EXPECT_EQ(bound_not_null_optional.value()->op(), Expression::Operation::kNotNull);
+
+  // A required field with required ancestors can never produce null values
+  auto is_null_required_parent = Expressions::IsNull("required_parent.leaf");
+  auto bound_is_null_required =
+      is_null_required_parent->Bind(*nested_schema, /*case_sensitive=*/true);
+  ASSERT_THAT(bound_is_null_required, IsOk());
+  EXPECT_EQ(bound_is_null_required.value()->op(), Expression::Operation::kFalse);
+
+  auto not_null_required_parent = Expressions::NotNull("required_parent.leaf");
+  auto bound_not_null_required =
+      not_null_required_parent->Bind(*nested_schema, /*case_sensitive=*/true);
+  ASSERT_THAT(bound_not_null_required, IsOk());
+  EXPECT_EQ(bound_not_null_required.value()->op(), Expression::Operation::kTrue);
+}
+
 TEST_F(PredicateTest, UnboundPredicateBindLiteral) {
   auto equal_pred = Expressions::Equal("age", Literal::Int(25));
   auto bound_result = equal_pred->Bind(*schema_, /*case_sensitive=*/true);