INTPYTHON-498 Missing field handling in nested data

bindings/python/pymongoarrow/lib.pyx

@@ -254,6 +254,12 @@ cdef class BuilderManager:
             # For list children, the nulls are stored in the parent.
             key = field.encode('utf-8')
             parent_type = self.parent_types.get(key, None)
+            # Check if the item was in our schema but never seen, and should have a parent.
+            if parent_type is None and "." in field:
+                parent_key, _, _ = field.rpartition('.')
+                self.parent_names[key] = parent_key.encode('utf-8')
+                parent_type = BSON_TYPE_DOCUMENT
+            # Add nulls according to parent type.
             if parent_type == BSON_TYPE_ARRAY:
                 continue
             if parent_type == BSON_TYPE_DOCUMENT:

bindings/python/test/test_arrow.py

@@ -533,6 +533,49 @@ def test_schema_arrays_of_documents_with_nulls(self):
             expected = json.load(fid)
         assert df.to_pylist() == expected
 
+    def test_schema_arrays_of_documents_orphaned_null(self):
+        # From https://github.com/mongodb-labs/mongo-arrow/issues/265.
+        col = self.coll
+        col.delete_many({})
+        schema = Schema(
+            {
+                "_id": ObjectId,
+                "test_list_struct": [
+                    {
+                        "field1": {
+                            "sub_field1": pa.string(),
+                            "sub_field2": pa.string(),
+                        }
+                    }
+                ],
+            }
+        )
+
+        col.insert_one(
+            {
+                "_id": ObjectId("000000000000000000000001"),
+                "test_list_struct": [
+                    {
+                        "field1": {
+                            "sub_field1": "test_data",
+                        }
+                    },
+                    {
+                        "field1": "test_data",
+                    },
+                ],
+            }
+        )
+        df = aggregate_arrow_all(col, schema=schema, pipeline=[])
+        doc = df.to_pylist()[0]
+        del doc["_id"]
+        assert doc == {
+            "test_list_struct": [
+                {"field1": {"sub_field1": "test_data", "sub_field2": None}},
+                {"field1": {"sub_field1": None, "sub_field2": None}},
+            ]
+        }
+
     def test_auto_schema_nested(self):
         # Create table with random data of various types.
         _, data = self._create_nested_data()