Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions src/iceberg/manifest/manifest_list.h
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
43 changes: 36 additions & 7 deletions src/iceberg/manifest/manifest_reader.cc
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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));

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The file_format case just below has no else-if-required branch, so a null in that required field silently keeps the default instead of returning MakeNullError the way every other field in this switch does.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch, thanks. Fixed in ea36a9c: the kFileFormatFieldId loop now checks required and returns MakeNullError for a null value, same as the other fields in this switch. Added a regression test (RejectsNullRequiredFileFormat) that writes a null file_format via a relaxed write-side schema (bypassing the writer's own non-null enforcement) and asserts the reader rejects it.

break;
case DataFile::kFilePathFieldId:
ICEBERG_RETURN_UNEXPECTED(ParseStringField(field_view, manifest_entries,
Expand All @@ -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;
Expand Down Expand Up @@ -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,
Expand Down
3 changes: 2 additions & 1 deletion src/iceberg/manifest/manifest_writer.cc
Original file line number Diff line number Diff line change
Expand Up @@ -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()) {
Expand Down
33 changes: 33 additions & 0 deletions src/iceberg/test/manifest_list_versions_test.cc
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down
86 changes: 86 additions & 0 deletions src/iceberg/test/manifest_reader_test.cc
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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_;
Expand Down Expand Up @@ -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)}));
Expand Down
Loading