fix(manifest): validate struct child lengths (#1054)

Author: gty404

src/iceberg/manifest/manifest_reader.cc

@@ -57,6 +57,16 @@ namespace {
   return InvalidArgument("Required field '{}' is null at row {}", field_name, row_idx);
 }
 
+[[nodiscard]] Status ValidateStructLength(const ArrowArrayView* view,
+                                          std::string_view field_name) {
+  for (int64_t i = 0; i < view->n_children; ++i) {
+    ICEBERG_CHECK(view->children[i]->length == view->length,
+                  "Field '{}' child {} should have length {} matching its parent, got {}",
+                  field_name, i, view->length, view->children[i]->length);
+  }
+  return {};
+}
+
 [[nodiscard]] Status AssertViewType(const ArrowArrayView* view, ArrowType expected_type,
                                     std::string_view field_name) {
   if (view->storage_type != expected_type) {
@@ -318,6 +328,7 @@ Result<std::vector<ManifestFile>> ParseManifestList(ArrowSchema* arrow_schema,
 
   ICEBERG_RETURN_UNEXPECTED(AssertViewTypeAndChildren(
       &view, ArrowType::NANOARROW_TYPE_STRUCT, schema.fields().size(), "manifest_list"));
+  ICEBERG_RETURN_UNEXPECTED(ValidateStructLength(&view, "manifest_list"));
 
   std::vector<ManifestFile> manifest_files;
   manifest_files.resize(array->length);
@@ -476,6 +487,7 @@ Status ParseDataFile(const std::shared_ptr<StructType>& data_file_schema,
   ICEBERG_RETURN_UNEXPECTED(
       AssertViewTypeAndChildren(view, ArrowType::NANOARROW_TYPE_STRUCT,
                                 data_file_schema->fields().size(), "data_file"));
+  ICEBERG_RETURN_UNEXPECTED(ValidateStructLength(view, "data_file"));
 
   for (int64_t col_idx = 0; col_idx < view->n_children; ++col_idx) {
     ICEBERG_ASSIGN_OR_RAISE(auto field, data_file_schema->GetFieldByIndex(col_idx));
@@ -515,6 +527,7 @@ Status ParseDataFile(const std::shared_ptr<StructType>& data_file_schema,
       case DataFile::kPartitionFieldId: {
         ICEBERG_RETURN_UNEXPECTED(
             AssertViewType(field_view, ArrowType::NANOARROW_TYPE_STRUCT, field_name));
+        ICEBERG_RETURN_UNEXPECTED(ValidateStructLength(field_view, field_name));
         const auto& partition_type =
             internal::checked_cast<const StructType&>(*field->get().type());
         for (int64_t part_idx = 0; part_idx < field_view->n_children; part_idx++) {
@@ -647,6 +660,7 @@ Result<std::vector<ManifestEntry>> ParseManifestEntry(
 
   ICEBERG_RETURN_UNEXPECTED(AssertViewTypeAndChildren(
       &view, ArrowType::NANOARROW_TYPE_STRUCT, schema.fields().size(), "manifest_entry"));
+  ICEBERG_RETURN_UNEXPECTED(ValidateStructLength(&view, "manifest_entry"));
 
   std::vector<ManifestEntry> manifest_entries;
   manifest_entries.resize(array->length);