src/iceberg/test/manifest_reader_test.cc

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

#include "iceberg/manifest/manifest_reader.h"

#include <chrono>
#include <format>
#include <memory>
#include <optional>
#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"

namespace iceberg {

class TestManifestReader : public testing::TestWithParam<int8_t> {
 protected:
  void SetUp() override {
    avro::RegisterAll();

    file_io_ = arrow::MakeMockFileIO();

    schema_ = std::make_shared<Schema>(std::vector<SchemaField>{
        SchemaField::MakeRequired(/*field_id=*/3, "id", int32()),
        SchemaField::MakeRequired(/*field_id=*/4, "data", string())});

    ICEBERG_UNWRAP_OR_FAIL(
        spec_,
        PartitionSpec::Make(
            /*spec_id=*/0, {PartitionField(/*source_id=*/4, /*field_id=*/1000,
                                           "data_bucket", Transform::Bucket(16))}));
  }

  std::string MakeManifestPath() {
    return std::format("manifest-{}.avro",
                       std::chrono::system_clock::now().time_since_epoch().count());
  }

  std::unique_ptr<DataFile> MakeDataFile(const std::string& path,
                                         const PartitionValues& partition,
                                         int64_t record_count = 1) {
    return std::make_unique<DataFile>(DataFile{
        .file_path = path,
        .file_format = FileFormatType::kParquet,
        .partition = partition,
        .record_count = record_count,
        .file_size_in_bytes = 10,
        .sort_order_id = 0,
    });
  }

  ManifestFile WriteManifest(int8_t format_version, std::optional<int64_t> snapshot_id,
                             const std::vector<ManifestEntry>& entries) {
    const std::string manifest_path = MakeManifestPath();

    auto writer_result =
        ManifestWriter::MakeWriter(format_version, snapshot_id, manifest_path, file_io_,
                                   spec_, schema_, ManifestContent::kData,
                                   /*first_row_id=*/0L);

    EXPECT_THAT(writer_result, IsOk());
    auto writer = std::move(writer_result.value());

    for (const auto& entry : entries) {
      EXPECT_THAT(writer->WriteEntry(entry), IsOk());
    }

    EXPECT_THAT(writer->Close(), IsOk());
    auto manifest_result = writer->ToManifestFile();
    EXPECT_THAT(manifest_result, IsOk());
    return std::move(manifest_result.value());
  }

  ManifestEntry MakeEntry(ManifestStatus status, int64_t snapshot_id,
                          std::unique_ptr<DataFile> file) {
    return ManifestEntry{
        .status = status,
        .snapshot_id = snapshot_id,
        .sequence_number =
            (status == ManifestStatus::kAdded) ? std::nullopt : std::make_optional(0),
        .file_sequence_number = std::nullopt,
        .data_file = std::move(file),
    };
  }

  std::unique_ptr<DataFile> MakeDeleteFile(
      const std::string& path, const PartitionValues& partition,
      DataFile::Content content,
      std::optional<std::string> referenced_file = std::nullopt,
      std::optional<int64_t> content_offset = std::nullopt,
      std::optional<int64_t> content_size = std::nullopt) {
    return std::make_unique<DataFile>(DataFile{
        .content = content,
        .file_path = path,
        .file_format = FileFormatType::kParquet,
        .partition = partition,
        .record_count = 1,
        .file_size_in_bytes = 10,
        .equality_ids = (content == DataFile::Content::kEqualityDeletes)
                            ? std::vector<int>{3}
                            : std::vector<int>{},
        .referenced_data_file = referenced_file,
        .content_offset = content_offset,
        .content_size_in_bytes = content_size,
    });
  }

  ManifestFile WriteDeleteManifest(int8_t format_version, int64_t snapshot_id,
                                   std::vector<ManifestEntry> entries) {
    const std::string manifest_path = MakeManifestPath();

    auto writer_result =
        ManifestWriter::MakeWriter(format_version, snapshot_id, manifest_path, file_io_,
                                   spec_, schema_, ManifestContent::kDeletes,
                                   /*first_row_id=*/std::nullopt);

    EXPECT_THAT(writer_result, IsOk());
    auto writer = std::move(writer_result.value());

    for (const auto& entry : entries) {
      EXPECT_THAT(writer->WriteEntry(entry), IsOk());
    }

    EXPECT_THAT(writer->Close(), IsOk());
    auto manifest_result = writer->ToManifestFile();
    EXPECT_THAT(manifest_result, IsOk());
    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_;
};

TEST_P(TestManifestReader, TestManifestReaderWithEmptyInheritableMetadata) {
  auto version = GetParam();
  auto file_a =
      MakeDataFile("/path/to/data-a.parquet", PartitionValues({Literal::Int(0)}));

  auto entry =
      MakeEntry(ManifestStatus::kExisting, /*snapshot_id=*/1000L, std::move(file_a));
  entry.sequence_number = 0;
  entry.file_sequence_number = 0;

  std::vector<ManifestEntry> entries;
  entries.push_back(std::move(entry));
  auto manifest = WriteManifest(version, /*snapshot_id=*/1000L, std::move(entries));

  auto reader_result = ManifestReader::Make(manifest, file_io_, schema_, spec_);
  ASSERT_THAT(reader_result, IsOk());
  auto reader = std::move(reader_result.value());

  auto entries_result = reader->Entries();
  ASSERT_THAT(entries_result, IsOk());
  auto read_entries = entries_result.value();

  ASSERT_EQ(read_entries.size(), 1);
  const auto& read_entry = read_entries[0];
  EXPECT_EQ(read_entry.status, ManifestStatus::kExisting);
  EXPECT_EQ(read_entry.data_file->file_path, "/path/to/data-a.parquet");
  EXPECT_EQ(read_entry.snapshot_id, 1000L);
}

TEST_P(TestManifestReader, EntriesStreamOwnsReaderResources) {
  auto version = GetParam();
  auto file_a =
      MakeDataFile("/path/to/data-a.parquet", PartitionValues({Literal::Int(0)}));
  auto file_b =
      MakeDataFile("/path/to/data-b.parquet", PartitionValues({Literal::Int(1)}));

  std::vector<ManifestEntry> entries;
  entries.push_back(
      MakeEntry(ManifestStatus::kAdded, /*snapshot_id=*/1000L, std::move(file_a)));
  entries.push_back(
      MakeEntry(ManifestStatus::kAdded, /*snapshot_id=*/1000L, std::move(file_b)));
  auto manifest = WriteManifest(version, /*snapshot_id=*/1000L, entries);

  ICEBERG_UNWRAP_OR_FAIL(auto reader,
                         ManifestReader::Make(manifest, file_io_, schema_, spec_));
  ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->EntriesStream());
  reader.reset();

  ICEBERG_UNWRAP_OR_FAIL(auto first, stream->Next());
  ASSERT_TRUE(first.has_value());
  EXPECT_EQ(first->data_file->file_path, "/path/to/data-a.parquet");

  ICEBERG_UNWRAP_OR_FAIL(auto second, stream->Next());
  ASSERT_TRUE(second.has_value());
  EXPECT_EQ(second->data_file->file_path, "/path/to/data-b.parquet");

  ICEBERG_UNWRAP_OR_FAIL(auto end, stream->Next());
  EXPECT_FALSE(end.has_value());
}

TEST_P(TestManifestReader, DeletedEntriesDoNotInheritFirstRowId) {
  auto version = GetParam();
  if (version < 3) {
    GTEST_SKIP() << "first_row_id is only assigned in V3 manifests";
  }

  auto deleted_file =
      MakeDataFile("/path/to/deleted.parquet", PartitionValues({Literal::Int(0)}),
                   /*record_count=*/10);
  auto added_file =
      MakeDataFile("/path/to/added.parquet", PartitionValues({Literal::Int(1)}),
                   /*record_count=*/5);

  auto deleted_entry =
      MakeEntry(ManifestStatus::kDeleted, /*snapshot_id=*/1000L, std::move(deleted_file));
  deleted_entry.sequence_number = 0;
  deleted_entry.file_sequence_number = 0;

  std::vector<ManifestEntry> entries;
  entries.push_back(std::move(deleted_entry));
  entries.push_back(
      MakeEntry(ManifestStatus::kAdded, /*snapshot_id=*/1000L, std::move(added_file)));
  auto manifest = WriteManifest(version, /*snapshot_id=*/1000L, std::move(entries));

  ICEBERG_UNWRAP_OR_FAIL(auto reader,
                         ManifestReader::Make(manifest, file_io_, schema_, spec_));
  ICEBERG_UNWRAP_OR_FAIL(auto read_entries, reader->Entries());

  ASSERT_EQ(read_entries.size(), 2U);
  EXPECT_EQ(read_entries[0].status, ManifestStatus::kDeleted);
  EXPECT_EQ(read_entries[0].data_file->first_row_id, std::nullopt);
  EXPECT_EQ(read_entries[1].status, ManifestStatus::kAdded);
  EXPECT_EQ(read_entries[1].data_file->first_row_id, 0);
}

TEST_P(TestManifestReader, TestReaderWithFilterWithoutSelect) {
  auto version = GetParam();
  auto file_a =
      MakeDataFile("/path/to/data-a.parquet", PartitionValues({Literal::Int(0)}));
  auto file_b =
      MakeDataFile("/path/to/data-b.parquet", PartitionValues({Literal::Int(1)}));
  auto file_c =
      MakeDataFile("/path/to/data-c.parquet", PartitionValues({Literal::Int(2)}));

  std::vector<ManifestEntry> entries;
  entries.push_back(MakeEntry(ManifestStatus::kAdded, 1000L, std::move(file_a)));
  entries.push_back(MakeEntry(ManifestStatus::kAdded, 1000L, std::move(file_b)));
  entries.push_back(MakeEntry(ManifestStatus::kAdded, 1000L, std::move(file_c)));

  auto manifest = WriteManifest(version, 1000L, std::move(entries));

  auto reader_result = ManifestReader::Make(manifest, file_io_, schema_, spec_);
  ASSERT_THAT(reader_result, IsOk());
  auto reader = std::move(reader_result.value());

  reader->FilterRows(Expressions::Equal("id", Literal::Int(0)));

  auto result_entries = reader->Entries();
  ASSERT_THAT(result_entries, IsOk());
  auto read_entries = result_entries.value();

  // note that all files are returned because the reader returns data files that may
  // match, and the partition is bucketing by data, which doesn't help filter files
  ASSERT_EQ(read_entries.size(), 3);
  EXPECT_EQ(read_entries[0].data_file->file_path, "/path/to/data-a.parquet");
  EXPECT_EQ(read_entries[1].data_file->file_path, "/path/to/data-b.parquet");
  EXPECT_EQ(read_entries[2].data_file->file_path, "/path/to/data-c.parquet");
}

TEST_P(TestManifestReader, TestManifestReaderWithPartitionMetadata) {
  auto version = GetParam();
  auto file_a =
      MakeDataFile("/path/to/data-a.parquet", PartitionValues({Literal::Int(0)}));
  auto entry =
      MakeEntry(ManifestStatus::kExisting, /*snapshot_id=*/123L, std::move(file_a));
  entry.sequence_number = 0;
  entry.file_sequence_number = 0;

  std::vector<ManifestEntry> entries;
  entries.push_back(std::move(entry));
  auto manifest = WriteManifest(version, /*snapshot_id=*/1000L, std::move(entries));

  auto reader_result = ManifestReader::Make(manifest, file_io_, schema_, spec_);
  ASSERT_THAT(reader_result, IsOk());
  auto reader = std::move(reader_result.value());

  auto entries_result = reader->Entries();
  ASSERT_THAT(entries_result, IsOk());
  auto read_entries = entries_result.value();

  ASSERT_EQ(read_entries.size(), 1);
  const auto& read_entry = read_entries[0];
  EXPECT_EQ(read_entry.snapshot_id, 123L);

  ASSERT_EQ(read_entry.data_file->partition.num_fields(), 1);
  EXPECT_EQ(read_entry.data_file->partition.values()[0], Literal::Int(0));
}

TEST_P(TestManifestReader, DecimalPartitionValuesRoundTrip) {
  schema_ = std::make_shared<Schema>(
      std::vector<SchemaField>{SchemaField::MakeRequired(3, "amount", decimal(38, 2))});
  ICEBERG_UNWRAP_OR_FAIL(
      spec_,
      PartitionSpec::Make(0, {PartitionField(3, 1000, "amount", Transform::Identity())}));

  const int128_t large = (static_cast<int128_t>(1) << 80) + 123;
  const std::vector<Literal> values{
      Literal::Decimal(123, 38, 2),   Literal::Decimal(-456, 38, 2),
      Literal::Decimal(large, 38, 2), Literal::Decimal(-large, 38, 2),
      Literal::Decimal(0, 38, 2),     Literal::Null(decimal(38, 2))};
  std::vector<ManifestEntry> entries;
  for (size_t index = 0; index < values.size(); ++index) {
    entries.push_back(MakeEntry(ManifestStatus::kAdded, 1000L,
                                MakeDataFile(std::format("decimal-{}.parquet", index),
                                             PartitionValues({values[index]}))));
  }
  auto manifest = WriteManifest(GetParam(), 1000L, entries);
  ICEBERG_UNWRAP_OR_FAIL(auto reader,
                         ManifestReader::Make(manifest, file_io_, schema_, spec_));
  ICEBERG_UNWRAP_OR_FAIL(auto read_entries, reader->Entries());

  ASSERT_EQ(read_entries.size(), values.size());
  for (size_t index = 0; index < values.size(); ++index) {
    ASSERT_NE(read_entries[index].data_file, nullptr);
    const auto& partition = read_entries[index].data_file->partition;
    ASSERT_EQ(partition.num_fields(), 1);
    if (values[index].IsNull()) {
      EXPECT_TRUE(partition.values()[0].IsNull());
    } else {
      EXPECT_EQ(partition.values()[0], values[index]);
    }
    EXPECT_EQ(*partition.values()[0].type(), *decimal(38, 2));
  }
}

TEST_P(TestManifestReader, NullPartitionValuePreservedPositionallyAcrossRows) {
  auto version = GetParam();

  // Two partition fields, with several entries. The first entry has a null in
  // partition field 0. The null must be preserved as a typed null in place so
  // that (a) the tuple keeps one value per partition field (arity matches the
  // spec), (b) the non-null value for field 1 stays in position 1 instead of
  // shifting into position 0, and (c) parsing of the remaining rows is not
  // aborted. This mirrors Iceberg Java's PartitionData, a fixed-size positional
  // array that stores null in place.
  auto multi_schema = std::make_shared<Schema>(std::vector<SchemaField>{
      SchemaField::MakeRequired(/*field_id=*/3, "id", int32()),
      SchemaField::MakeRequired(/*field_id=*/4, "data", string())});

  std::shared_ptr<PartitionSpec> multi_spec;
  ICEBERG_UNWRAP_OR_FAIL(
      multi_spec,
      PartitionSpec::Make(
          /*spec_id=*/0, {PartitionField(/*source_id=*/3, /*field_id=*/1000, "id_part",
                                         Transform::Identity()),
                          PartitionField(/*source_id=*/4, /*field_id=*/1001,
                                         "data_bucket", Transform::Bucket(16))}));

  const std::string manifest_path = MakeManifestPath();
  auto writer_result = ManifestWriter::MakeWriter(
      version, /*snapshot_id=*/1000L, manifest_path, file_io_, multi_spec, multi_schema,
      ManifestContent::kData, /*first_row_id=*/0L);
  ASSERT_THAT(writer_result, IsOk());
  auto writer = std::move(writer_result.value());

  // First entry: partition field 0 (identity int) is null, field 1 (bucket) is 9.
  auto file_null =
      MakeDataFile("/path/to/data-null.parquet",
                   PartitionValues({Literal::Null(int32()), Literal::Int(9)}));
  auto file_b = MakeDataFile("/path/to/data-b.parquet",
                             PartitionValues({Literal::Int(1), Literal::Int(5)}));
  auto file_c = MakeDataFile("/path/to/data-c.parquet",
                             PartitionValues({Literal::Int(2), Literal::Int(7)}));
  ASSERT_THAT(
      writer->WriteEntry(MakeEntry(ManifestStatus::kAdded, 1000L, std::move(file_null))),
      IsOk());
  ASSERT_THAT(
      writer->WriteEntry(MakeEntry(ManifestStatus::kAdded, 1000L, std::move(file_b))),
      IsOk());
  ASSERT_THAT(
      writer->WriteEntry(MakeEntry(ManifestStatus::kAdded, 1000L, std::move(file_c))),
      IsOk());
  ASSERT_THAT(writer->Close(), IsOk());
  ICEBERG_UNWRAP_OR_FAIL(auto manifest, writer->ToManifestFile());

  ICEBERG_UNWRAP_OR_FAIL(
      auto reader, ManifestReader::Make(manifest, file_io_, multi_schema, multi_spec));
  ICEBERG_UNWRAP_OR_FAIL(auto read_entries, reader->Entries());

  ASSERT_EQ(read_entries.size(), 3U);

  // The null-partition row keeps both fields: a null in position 0 and the
  // non-null value in position 1 (no shifting).
  EXPECT_EQ(read_entries[0].data_file->file_path, "/path/to/data-null.parquet");
  ASSERT_EQ(read_entries[0].data_file->partition.num_fields(), 2);
  EXPECT_TRUE(read_entries[0].data_file->partition.values()[0].IsNull());
  EXPECT_EQ(read_entries[0].data_file->partition.values()[1], Literal::Int(9));

  // Rows after the null one must still have their partition values parsed.
  EXPECT_EQ(read_entries[1].data_file->file_path, "/path/to/data-b.parquet");
  ASSERT_EQ(read_entries[1].data_file->partition.num_fields(), 2);
  EXPECT_EQ(read_entries[1].data_file->partition.values()[0], Literal::Int(1));
  EXPECT_EQ(read_entries[1].data_file->partition.values()[1], Literal::Int(5));

  EXPECT_EQ(read_entries[2].data_file->file_path, "/path/to/data-c.parquet");
  ASSERT_EQ(read_entries[2].data_file->partition.num_fields(), 2);
  EXPECT_EQ(read_entries[2].data_file->partition.values()[0], Literal::Int(2));
  EXPECT_EQ(read_entries[2].data_file->partition.values()[1], Literal::Int(7));
}

TEST_P(TestManifestReader, ReadsEntriesWhenPartitionSourceFieldIsMissing) {
  auto version = GetParam();
  auto file = MakeDataFile("/path/to/historical-data.parquet",
                           PartitionValues({Literal::Int(7)}));
  auto manifest =
      WriteManifest(version, /*snapshot_id=*/1000L,
                    {MakeEntry(ManifestStatus::kAdded, 1000L, std::move(file))});

  auto current_schema = std::make_shared<Schema>(
      std::vector<SchemaField>{SchemaField::MakeRequired(/*field_id=*/3, "id", int32())});

  ICEBERG_UNWRAP_OR_FAIL(auto reader,
                         ManifestReader::Make(manifest, file_io_, current_schema, spec_));
  ICEBERG_UNWRAP_OR_FAIL(auto read_entries, reader->Entries());

  ASSERT_EQ(read_entries.size(), 1U);
  EXPECT_EQ(read_entries[0].data_file->file_path, "/path/to/historical-data.parquet");
  EXPECT_EQ(read_entries[0].data_file->record_count, 1);
}

TEST_P(TestManifestReader, TestDeleteFilesWithReferences) {
  auto version = GetParam();
  if (version < 2) {
    GTEST_SKIP() << "Delete files only supported in V2+";
  }

  auto delete_file_1 = MakeDeleteFile(
      "/path/to/data-a-deletes.parquet", PartitionValues({Literal::Int(0)}),
      DataFile::Content::kPositionDeletes, "/path/to/data-a.parquet");
  auto delete_file_2 = MakeDeleteFile(
      "/path/to/data-b-deletes.parquet", PartitionValues({Literal::Int(1)}),
      DataFile::Content::kPositionDeletes, "/path/to/data-b.parquet");

  std::vector<ManifestEntry> entries;
  entries.push_back(
      MakeEntry(ManifestStatus::kAdded, /*snapshot_id=*/1000L, std::move(delete_file_1)));
  entries.push_back(
      MakeEntry(ManifestStatus::kAdded, /*snapshot_id=*/1000L, std::move(delete_file_2)));

  auto manifest = WriteDeleteManifest(version, /*snapshot_id=*/1000L, std::move(entries));

  auto reader_result = ManifestReader::Make(manifest, file_io_, schema_, spec_);
  ASSERT_THAT(reader_result, IsOk());
  auto reader = std::move(reader_result.value());

  auto entries_result = reader->Entries();
  ASSERT_THAT(entries_result, IsOk());
  auto read_entries = entries_result.value();
  ASSERT_EQ(read_entries.size(), 2);

  for (const auto& entry : read_entries) {
    if (entry.data_file->file_path == "/path/to/data-a-deletes.parquet") {
      EXPECT_EQ(entry.data_file->referenced_data_file, "/path/to/data-a.parquet");
    } else {
      EXPECT_EQ(entry.data_file->referenced_data_file, "/path/to/data-b.parquet");
    }
  }
}

TEST_P(TestManifestReader, TestDVs) {
  auto version = GetParam();
  if (version < 3) {
    GTEST_SKIP() << "DVs only supported in V3+";
  }

  auto dv1 =
      MakeDeleteFile("/path/to/data-a-deletes.puffin", PartitionValues({Literal::Int(0)}),
                     DataFile::Content::kPositionDeletes, "/path/to/data-a.parquet",
                     /*content_offset=*/4L, /*content_size_in_bytes=*/6L);
  auto dv2 =
      MakeDeleteFile("/path/to/data-b-deletes.puffin", PartitionValues({Literal::Int(1)}),
                     DataFile::Content::kPositionDeletes, "/path/to/data-b.parquet",
                     /*content_offset=*/4L, /*content_size_in_bytes=*/6L);

  std::vector<ManifestEntry> entries;
  entries.push_back(
      MakeEntry(ManifestStatus::kAdded, /*snapshot_id=*/1000L, std::move(dv1)));
  entries.push_back(
      MakeEntry(ManifestStatus::kAdded, /*snapshot_id=*/1000L, std::move(dv2)));
  auto manifest = WriteDeleteManifest(version, /*snapshot_id=*/1000L, std::move(entries));

  auto reader_result = ManifestReader::Make(manifest, file_io_, schema_, spec_);
  ASSERT_THAT(reader_result, IsOk());
  auto reader = std::move(reader_result.value());

  auto entries_result = reader->Entries();
  ASSERT_THAT(entries_result, IsOk());
  auto read_entries = entries_result.value();
  ASSERT_EQ(read_entries.size(), 2);

  for (const auto& entry : read_entries) {
    if (entry.data_file->file_path == "/path/to/data-a-deletes.puffin") {
      EXPECT_EQ(entry.data_file->referenced_data_file, "/path/to/data-a.parquet");
      EXPECT_EQ(entry.data_file->content_offset, 4L);
      EXPECT_EQ(entry.data_file->content_size_in_bytes, 6L);
    } else {
      EXPECT_EQ(entry.data_file->referenced_data_file, "/path/to/data-b.parquet");
      EXPECT_EQ(entry.data_file->content_offset, 4L);
      EXPECT_EQ(entry.data_file->content_size_in_bytes, 6L);
    }
  }
}

TEST_P(TestManifestReader, TestInvalidUsage) {
  auto version = GetParam();
  auto file_a =
      MakeDataFile("/path/to/data-a.parquet", PartitionValues({Literal::Int(0)}));
  auto entry =
      MakeEntry(ManifestStatus::kExisting, /*snapshot_id=*/1000L, std::move(file_a));
  entry.sequence_number = 0;
  entry.file_sequence_number = 0;

  auto manifest =
      WriteManifest(version, /*snapshot_id=*/std::nullopt, {std::move(entry)});

  auto reader_result = ManifestReader::Make(manifest, file_io_, schema_, spec_);
  EXPECT_THAT(reader_result, IsError(ErrorKind::kInvalidManifest));
  EXPECT_THAT(reader_result, HasErrorMessage("has no snapshot ID"));
}

TEST_P(TestManifestReader, TestDropStats) {
  auto version = GetParam();

  // Create a data file with full metrics
  auto file_with_stats = std::make_unique<DataFile>(DataFile{
      .file_path = "/path/to/data-with-stats.parquet",
      .file_format = FileFormatType::kParquet,
      .partition = PartitionValues({Literal::Int(0)}),
      .record_count = 100,
      .file_size_in_bytes = 1024,
      // Add stats for multiple columns
      .column_sizes = {{1, 100}, {2, 200}, {3, 300}},
      .value_counts = {{1, 10}, {2, 20}, {3, 30}},
      .null_value_counts = {{1, 1}, {2, 2}, {3, 3}},
      .nan_value_counts = {{1, 0}, {2, 0}, {3, 0}},
      .lower_bounds = {{1, {0x01}}, {2, {0x02}}, {3, {0x03}}},
      .upper_bounds = {{1, {0xFF}}, {2, {0xFE}}, {3, {0xFD}}},
      .sort_order_id = 0,
  });

  auto entry = MakeEntry(ManifestStatus::kAdded, /*snapshot_id=*/1000L,
                         std::move(file_with_stats));

  std::vector<ManifestEntry> entries;
  entries.push_back(std::move(entry));

  auto manifest = WriteManifest(version, /*snapshot_id=*/1000L, std::move(entries));

  auto reader_result = ManifestReader::Make(manifest, file_io_, schema_, spec_);
  ASSERT_THAT(reader_result, IsOk());
  auto reader = std::move(reader_result.value());
  reader->Select({"record_count"}).TryDropStats();

  ICEBERG_UNWRAP_OR_FAIL(auto read_entries, reader->Entries());
  ASSERT_EQ(read_entries.size(), 1);
  const auto& read_entry = read_entries[0];

  // record_count should be preserved
  EXPECT_EQ(read_entry.data_file->record_count, 100);

  // Stats maps should be cleared
  EXPECT_TRUE(read_entry.data_file->column_sizes.empty());
  EXPECT_TRUE(read_entry.data_file->value_counts.empty());
  EXPECT_TRUE(read_entry.data_file->null_value_counts.empty());
  EXPECT_TRUE(read_entry.data_file->nan_value_counts.empty());
  EXPECT_TRUE(read_entry.data_file->lower_bounds.empty());
  EXPECT_TRUE(read_entry.data_file->upper_bounds.empty());
}

TEST_P(TestManifestReader, SelectDistinguishesEmptyAndWildcard) {
  auto version = GetParam();
  auto file = MakeDataFile("/path/to/data.parquet", PartitionValues({Literal::Int(3)}));
  auto manifest = WriteManifest(
      version, /*snapshot_id=*/1000L,
      {MakeEntry(ManifestStatus::kAdded, /*snapshot_id=*/1000L, std::move(file))});

  auto read_with_projection =
      [&](const std::vector<std::string>& columns) -> Result<std::vector<ManifestEntry>> {
    ICEBERG_ASSIGN_OR_RAISE(auto reader,
                            ManifestReader::Make(manifest, file_io_, schema_, spec_));
    reader->Select(columns);
    return reader->Entries();
  };

  ICEBERG_UNWRAP_OR_FAIL(auto empty_entries, read_with_projection({}));
  ASSERT_EQ(empty_entries.size(), 1);
  EXPECT_TRUE(empty_entries.front().data_file->file_path.empty());
  EXPECT_EQ(empty_entries.front().data_file->record_count, 1);
  EXPECT_EQ(empty_entries.front().data_file->partition.num_fields(), 0);

  ICEBERG_UNWRAP_OR_FAIL(auto all_entries,
                         read_with_projection({std::string(Schema::kAllColumns)}));
  ASSERT_EQ(all_entries.size(), 1);
  EXPECT_EQ(all_entries.front().data_file->file_path, "/path/to/data.parquet");
  ASSERT_EQ(all_entries.front().data_file->partition.num_fields(), 1);
  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)}));
  EXPECT_TRUE(ManifestReader::ShouldDropStats({"file_path", "file_format", "partition"}));
  EXPECT_TRUE(
      ManifestReader::ShouldDropStats({"file_path", "file_format", "record_count"}));
  EXPECT_FALSE(
      ManifestReader::ShouldDropStats({"file_path", "file_format", "value_counts"}));
  EXPECT_FALSE(
      ManifestReader::ShouldDropStats({"file_path", "file_format", "lower_bounds"}));
  EXPECT_FALSE(ManifestReader::ShouldDropStats(
      {"file_path", "value_counts", "null_value_counts", "lower_bounds"}));
  EXPECT_FALSE(
      ManifestReader::ShouldDropStats({"file_path", "record_count", "value_counts"}));
}

TEST(ManifestReaderStaticTest, WithStatsColumnsHandlesEmptyAndSelectAll) {
  EXPECT_THAT(ManifestReader::WithStatsColumns({}),
              testing::UnorderedElementsAre(
                  "value_counts", "null_value_counts", "nan_value_counts", "lower_bounds",
                  "upper_bounds", "column_sizes", "record_count"));
  EXPECT_THAT(ManifestReader::WithStatsColumns({std::string(Schema::kAllColumns)}),
              testing::ElementsAre(Schema::kAllColumns));
}

INSTANTIATE_TEST_SUITE_P(ManifestReaderVersions, TestManifestReader,
                         testing::Values(1, 2, 3));

}  // namespace iceberg