diff --git a/src/iceberg/manifest/manifest_list.h b/src/iceberg/manifest/manifest_list.h index b482b0b39..976415fe6 100644 --- a/src/iceberg/manifest/manifest_list.h +++ b/src/iceberg/manifest/manifest_list.h @@ -276,6 +276,19 @@ ICEBERG_EXPORT inline constexpr Result ManifestContentFromStrin return InvalidArgument("Invalid manifest content type: {}", str); } +/// \brief Get the relative manifest content type from int +ICEBERG_EXPORT inline constexpr Result 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 { diff --git a/src/iceberg/manifest/manifest_reader.cc b/src/iceberg/manifest/manifest_reader.cc index b1b6535d9..df8f036f7 100644 --- a/src/iceberg/manifest/manifest_reader.cc +++ b/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)...); } +// Validate integer-encoded enums so out-of-range values from the file never +// become enum values that no switch handles. +template + requires std::ranges::forward_range +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(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 Status ParseStringField(Args&&... args) { return ParseField( @@ -322,8 +347,9 @@ Result> 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& 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& 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> 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, diff --git a/src/iceberg/manifest/manifest_writer.cc b/src/iceberg/manifest/manifest_writer.cc index bdc6ba2a1..2d3ff077d 100644 --- a/src/iceberg/manifest/manifest_writer.cc +++ b/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(entry.status)); } if (entry.IsAlive() && entry.sequence_number.has_value()) { diff --git a/src/iceberg/test/manifest_list_versions_test.cc b/src/iceberg/test/manifest_list_versions_test.cc index 20ce47161..8618c3015 100644 --- a/src/iceberg/test/manifest_list_versions_test.cc +++ b/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(std::vector{ + 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(); diff --git a/src/iceberg/test/manifest_reader_test.cc b/src/iceberg/test/manifest_reader_test.cc index acfd199ea..80a8acd72 100644 --- a/src/iceberg/test/manifest_reader_test.cc +++ b/src/iceberg/test/manifest_reader_test.cc @@ -26,16 +26,21 @@ #include #include +#include +#include +#include #include #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 { 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(std::vector{ + 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(std::vector{ + 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 file_io_; std::shared_ptr schema_; std::shared_ptr 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)}));