fix(manifest): validate enum fields when parsing manifests (#1034)

Manifest and manifest list readers cast raw integers from the file
straight into ManifestStatus, DataFile::Content and ManifestContent.
Out-of-range values then reached switches ending in unreachable(),
which is undefined behavior. Validate these fields through the
*FromInt helpers at parse time, and return an error instead of
unreachable() for an unknown status in ManifestWriter::WriteEntry.

ParseDataFile's hand-rolled loop for DataFile::kFileFormatFieldId
skipped the field when null instead of checking `required` and
returning MakeNullError like every other field in the switch,
addressing review feedback on PR #1034.

Author: gty404

src/iceberg/manifest/manifest_list.h

@@ -276,6 +276,19 @@ ICEBERG_EXPORT inline constexpr Result<ManifestContent> ManifestContentFromStrin
   return InvalidArgument("Invalid manifest content type: {}", str);
 }
 
+/// \brief Get the relative manifest content type from int
+ICEBERG_EXPORT inline constexpr Result<ManifestContent> ManifestContentFromInt(
+    int32_t content) noexcept {
+  switch (content) {
+    case 0:
+      return ManifestContent::kData;
+    case 1:
+      return ManifestContent::kDeletes;
+    default:
+      return InvalidArgument("Invalid manifest content type: {}", content);
+  }
+}
+
 }  // namespace iceberg
 
 namespace std {

src/iceberg/manifest/manifest_reader.cc

@@ -133,6 +133,31 @@ Status ParseIntegerField(const ArrowArrayView* array_view, Container& container,
       array_view, container, accessor, std::forward<Args>(args)...);
 }
 
+// Validate integer-encoded enums so out-of-range values from the file never
+// become enum values that no switch handles.
+template <typename FromInt, typename Container, typename Accessor>
+  requires std::ranges::forward_range<Container>
+Status ParseEnumField(FromInt from_int, const ArrowArrayView* array_view,
+                      Container& container, Accessor accessor,
+                      std::string_view field_name, bool required) {
+  auto iter = std::ranges::begin(container);
+  for (int64_t row_idx = 0; row_idx < array_view->length; row_idx++, iter++) {
+    if (!ArrowArrayViewIsNull(array_view, row_idx)) {
+      const auto raw =
+          static_cast<int32_t>(ArrowArrayViewGetIntUnsafe(array_view, row_idx));
+      auto value = std::invoke(from_int, raw);
+      if (!value.has_value()) [[unlikely]] {
+        return InvalidArgument("Field '{}' at row {}: {}", field_name, row_idx,
+                               value.error().message);
+      }
+      std::invoke(accessor, *iter) = std::move(value.value());
+    } else if (required) {
+      return MakeNullError(field_name, row_idx);
+    }
+  }
+  return {};
+}
+
 template <typename... Args>
 Status ParseStringField(Args&&... args) {
   return ParseField(
@@ -322,8 +347,9 @@ Result<std::vector<ManifestFile>> ParseManifestList(ArrowSchema* arrow_schema,
                                                     field_name, required));
         break;
       case ManifestFile::kContentFieldId:
-        ICEBERG_RETURN_UNEXPECTED(ParseIntegerField(
-            field_view, manifest_files, &ManifestFile::content, field_name, required));
+        ICEBERG_RETURN_UNEXPECTED(ParseEnumField(ManifestContentFromInt, field_view,
+                                                 manifest_files, &ManifestFile::content,
+                                                 field_name, required));
         break;
       case ManifestFile::kSequenceNumberFieldId:
         ICEBERG_RETURN_UNEXPECTED(ParseIntegerField(field_view, manifest_files,
@@ -465,9 +491,9 @@ Status ParseDataFile(const std::shared_ptr<StructType>& data_file_schema,
 
     switch (field_id) {
       case DataFile::kContentFieldId:
-        ICEBERG_RETURN_UNEXPECTED(ParseIntegerField(field_view, manifest_entries,
-                                                    proj_data_file(&DataFile::content),
-                                                    field_name, required));
+        ICEBERG_RETURN_UNEXPECTED(
+            ParseEnumField(DataFileContentFromInt, field_view, manifest_entries,
+                           proj_data_file(&DataFile::content), field_name, required));
         break;
       case DataFile::kFilePathFieldId:
         ICEBERG_RETURN_UNEXPECTED(ParseStringField(field_view, manifest_entries,
@@ -481,6 +507,8 @@ Status ParseDataFile(const std::shared_ptr<StructType>& data_file_schema,
             std::string_view path_str(value.data, value.size_bytes);
             ICEBERG_ASSIGN_OR_RAISE(manifest_entries[row_idx].data_file->file_format,
                                     FileFormatTypeFromString(path_str));
+          } else if (required) {
+            return MakeNullError(field_name, row_idx);
           }
         }
         break;
@@ -638,8 +666,9 @@ Result<std::vector<ManifestEntry>> ParseManifestEntry(
 
     switch (field_id) {
       case ManifestEntry::kStatusFieldId:
-        ICEBERG_RETURN_UNEXPECTED(ParseIntegerField(
-            field_view, manifest_entries, &ManifestEntry::status, field_name, required));
+        ICEBERG_RETURN_UNEXPECTED(ParseEnumField(ManifestStatusFromInt, field_view,
+                                                 manifest_entries, &ManifestEntry::status,
+                                                 field_name, required));
         break;
       case ManifestEntry::kSnapshotIdFieldId:
         ICEBERG_RETURN_UNEXPECTED(ParseIntegerField(field_view, manifest_entries,

src/iceberg/manifest/manifest_writer.cc

@@ -163,7 +163,8 @@ Status ManifestWriter::WriteEntry(const ManifestEntry& entry) {
       delete_rows_count_ += entry.data_file->record_count;
       break;
     default:
-      std::unreachable();
+      return InvalidArgument("Invalid manifest entry status: {}",
+                             static_cast<int32_t>(entry.status));
   }
 
   if (entry.IsAlive() && entry.sequence_number.has_value()) {

src/iceberg/test/manifest_list_versions_test.cc

@@ -425,6 +425,39 @@ TEST_F(TestManifestListVersions, TestManifestsWithoutRowStats) {
   EXPECT_FALSE(manifest.first_row_id.has_value());
 }
 
+TEST_F(TestManifestListVersions, RejectsInvalidManifestContent) {
+  auto schema = std::make_shared<Schema>(std::vector<SchemaField>{
+      ManifestFile::kManifestPath,
+      ManifestFile::kManifestLength,
+      ManifestFile::kPartitionSpecId,
+      ManifestFile::kContent,
+      ManifestFile::kAddedSnapshotId,
+  });
+
+  ArrowSchema arrow_c_schema;
+  EXPECT_THAT(ToArrowSchema(*schema, &arrow_c_schema), IsOk());
+  auto arrow_schema = ::arrow::ImportType(&arrow_c_schema).ValueOrDie();
+  auto array = ::arrow::json::ArrayFromJSONString(
+                   arrow_schema, R"([["path/to/manifest.avro", 1024, 1, 5, 100]])")
+                   .ValueOrDie();
+  ArrowArray arrow_array;
+  EXPECT_TRUE(::arrow::ExportArray(*array, &arrow_array).ok());
+
+  std::string manifest_list_path = CreateManifestListPath();
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto writer, WriterFactoryRegistry::Open(
+                       FileFormatType::kAvro,
+                       {.path = manifest_list_path, .schema = schema, .io = file_io_}));
+  EXPECT_THAT(writer->Write(&arrow_array), IsOk());
+  EXPECT_THAT(writer->Close(), IsOk());
+
+  ICEBERG_UNWRAP_OR_FAIL(auto reader,
+                         ManifestListReader::Make(manifest_list_path, file_io_));
+  auto files = reader->Files();
+  EXPECT_THAT(files, IsError(ErrorKind::kInvalidArgument));
+  EXPECT_THAT(files, HasErrorMessage("Invalid manifest content type: 5"));
+}
+
 TEST_F(TestManifestListVersions, TestManifestsPartitionSummary) {
   auto first_summary_lower_bound = Literal::Int(10).Serialize().value();
   auto first_summary_upper_bound = Literal::Int(100).Serialize().value();

src/iceberg/test/manifest_reader_test.cc

@@ -26,16 +26,21 @@
 #include <string>
 #include <vector>
 
+#include <arrow/array.h>
+#include <arrow/c/bridge.h>
+#include <arrow/json/from_string.h>
 #include <gtest/gtest.h>
 
 #include "iceberg/arrow/arrow_io_util.h"
 #include "iceberg/avro/avro_register.h"
 #include "iceberg/expression/expressions.h"
+#include "iceberg/file_writer.h"
 #include "iceberg/manifest/manifest_entry.h"
 #include "iceberg/manifest/manifest_list.h"
 #include "iceberg/manifest/manifest_writer.h"
 #include "iceberg/partition_spec.h"
 #include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
 #include "iceberg/test/matchers.h"
 #include "iceberg/transform.h"
 #include "iceberg/type.h"
@@ -156,6 +161,54 @@ class TestManifestReader : public testing::TestWithParam<int8_t> {
     return std::move(manifest_result.value());
   }
 
+  // Writes manifest entries given as JSON rows of
+  // [status, snapshot_id, [content, file_path, file_format, [partition], record_count,
+  // file_size_in_bytes]], bypassing ManifestWriter validation.
+  ManifestFile WriteRawManifest(const std::string& json_rows) {
+    auto partition_type = spec_->PartitionType(*schema_);
+    EXPECT_THAT(partition_type, IsOk());
+    auto data_file_type = std::make_shared<StructType>(std::vector<SchemaField>{
+        DataFile::kContent, DataFile::kFilePath,
+        // Written as optional so tests can bypass the writer's own non-null
+        // enforcement and exercise the reader's required-field validation instead.
+        SchemaField::MakeOptional(DataFile::kFileFormatFieldId, "file_format", string()),
+        SchemaField::MakeRequired(DataFile::kPartitionFieldId, DataFile::kPartitionField,
+                                  std::move(partition_type.value())),
+        DataFile::kRecordCount, DataFile::kFileSize});
+    auto entry_schema = std::make_shared<Schema>(std::vector<SchemaField>{
+        ManifestEntry::kStatus, ManifestEntry::kSnapshotId,
+        SchemaField::MakeRequired(ManifestEntry::kDataFileFieldId,
+                                  ManifestEntry::kDataFileField, data_file_type)});
+
+    ArrowSchema arrow_c_schema;
+    EXPECT_THAT(ToArrowSchema(*entry_schema, &arrow_c_schema), IsOk());
+    auto arrow_schema = ::arrow::ImportType(&arrow_c_schema).ValueOrDie();
+    auto array = ::arrow::json::ArrayFromJSONString(arrow_schema, json_rows).ValueOrDie();
+    ArrowArray arrow_array;
+    EXPECT_TRUE(::arrow::ExportArray(*array, &arrow_array).ok());
+
+    const std::string manifest_path = MakeManifestPath();
+    auto writer_result = WriterFactoryRegistry::Open(
+        FileFormatType::kAvro,
+        {.path = manifest_path, .schema = entry_schema, .io = file_io_});
+    EXPECT_THAT(writer_result, IsOk());
+    auto writer = std::move(writer_result.value());
+    EXPECT_THAT(writer->Write(&arrow_array), IsOk());
+    EXPECT_THAT(writer->Close(), IsOk());
+    auto length = writer->length();
+    EXPECT_THAT(length, IsOk());
+
+    return ManifestFile{
+        .manifest_path = manifest_path,
+        .manifest_length = length.value(),
+        .partition_spec_id = spec_->spec_id(),
+        .content = ManifestContent::kData,
+        .sequence_number = 0,
+        .min_sequence_number = 0,
+        .added_snapshot_id = 1000L,
+    };
+  }
+
   std::shared_ptr<FileIO> file_io_;
   std::shared_ptr<Schema> schema_;
   std::shared_ptr<PartitionSpec> spec_;
@@ -629,6 +682,39 @@ TEST_P(TestManifestReader, SelectDistinguishesEmptyAndWildcard) {
   EXPECT_EQ(all_entries.front().data_file->partition.values().front(), Literal::Int(3));
 }
 
+TEST_P(TestManifestReader, RejectsInvalidDataFileContent) {
+  auto manifest = WriteRawManifest(
+      R"([[1, 1000, [3, "/path/to/data.parquet", "parquet", [0], 1, 10]]])");
+
+  ICEBERG_UNWRAP_OR_FAIL(auto reader,
+                         ManifestReader::Make(manifest, file_io_, schema_, spec_));
+  auto entries = reader->Entries();
+  EXPECT_THAT(entries, IsError(ErrorKind::kInvalidArgument));
+  EXPECT_THAT(entries, HasErrorMessage("Invalid data file content: 3"));
+}
+
+TEST_P(TestManifestReader, RejectsInvalidEntryStatus) {
+  auto manifest = WriteRawManifest(
+      R"([[7, 1000, [0, "/path/to/data.parquet", "parquet", [0], 1, 10]]])");
+
+  ICEBERG_UNWRAP_OR_FAIL(auto reader,
+                         ManifestReader::Make(manifest, file_io_, schema_, spec_));
+  auto entries = reader->Entries();
+  EXPECT_THAT(entries, IsError(ErrorKind::kInvalidArgument));
+  EXPECT_THAT(entries, HasErrorMessage("Invalid manifest status: 7"));
+}
+
+TEST_P(TestManifestReader, RejectsNullRequiredFileFormat) {
+  auto manifest =
+      WriteRawManifest(R"([[0, 1000, [0, "/path/to/data.parquet", null, [0], 1, 10]]])");
+
+  ICEBERG_UNWRAP_OR_FAIL(auto reader,
+                         ManifestReader::Make(manifest, file_io_, schema_, spec_));
+  auto entries = reader->Entries();
+  EXPECT_THAT(entries, IsError(ErrorKind::kInvalidArgument));
+  EXPECT_THAT(entries, HasErrorMessage("Required field 'file_format' is null at row 0"));
+}
+
 TEST(ManifestReaderStaticTest, TestShouldDropStats) {
   EXPECT_TRUE(ManifestReader::ShouldDropStats({}));
   EXPECT_FALSE(ManifestReader::ShouldDropStats({std::string(Schema::kAllColumns)}));