src/iceberg/test/struct_like_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/row/struct_like.h"

#include <array>
#include <memory>
#include <optional>
#include <string>
#include <utility>
#include <vector>

#include <arrow/array.h>
#include <arrow/buffer.h>
#include <arrow/c/bridge.h>
#include <arrow/json/from_string.h>
#include <arrow/type.h>
#include <arrow/type_fwd.h>
#include <arrow/util/decimal.h>

#include "iceberg/arrow_c_data_guard_internal.h"
#include "iceberg/expression/literal.h"
#include "iceberg/manifest/manifest_list.h"
#include "iceberg/manifest/manifest_reader_internal.h"
#include "iceberg/row/arrow_array_wrapper.h"
#include "iceberg/row/manifest_wrapper.h"
#include "iceberg/schema.h"
#include "iceberg/schema_internal.h"
#include "iceberg/test/matchers.h"
#include "iceberg/type.h"
#include "iceberg/util/uuid.h"

namespace iceberg {

#define EXPECT_SCALAR_EQ(result, expected_type, expected_value) \
  do {                                                          \
    ASSERT_THAT(result, IsOk());                                \
    auto scalar = result.value();                               \
    ASSERT_TRUE(std::holds_alternative<expected_type>(scalar)); \
    EXPECT_EQ(std::get<expected_type>(scalar), expected_value); \
  } while (0)

#define EXPECT_DECIMAL_EQ(result, scale, expected_value)  \
  do {                                                    \
    ASSERT_THAT(result, IsOk());                          \
    auto scalar = result.value();                         \
    ASSERT_TRUE(std::holds_alternative<Decimal>(scalar)); \
    auto decimal = std::get<Decimal>(scalar);             \
    EXPECT_EQ(decimal.ToString(scale), expected_value);   \
  } while (0)

#define EXPECT_SCALAR_NULL(result)                               \
  do {                                                           \
    ASSERT_THAT(result, IsOk());                                 \
    auto scalar = result.value();                                \
    ASSERT_TRUE(std::holds_alternative<std::monostate>(scalar)); \
  } while (0)

namespace {

class SingleFieldStructLike : public StructLike {
 public:
  explicit SingleFieldStructLike(Scalar value) : value_(std::move(value)) {}
  explicit SingleFieldStructLike(Result<Scalar> value) : value_(std::move(value)) {}

  Result<Scalar> GetField(size_t pos) const override {
    if (pos != 0) {
      return InvalidArgument("Field index {} out of range", pos);
    }
    return value_;
  }

  size_t num_fields() const override { return 1; }

 private:
  Result<Scalar> value_;
};

std::shared_ptr<StructLike> WrapSingleFieldRow(Result<Scalar> field, size_t depth) {
  std::shared_ptr<StructLike> row =
      std::make_shared<SingleFieldStructLike>(std::move(field));
  for (size_t i = 1; i < depth; ++i) {
    row = std::make_shared<SingleFieldStructLike>(Scalar{row});
  }
  return row;
}

std::unique_ptr<Schema> MakeNestedSchema(
    size_t depth, std::optional<size_t> optional_parent = std::nullopt,
    bool optional_leaf = false) {
  auto field =
      optional_leaf
          ? SchemaField::MakeOptional(static_cast<int32_t>(depth), "value", int32())
          : SchemaField::MakeRequired(static_cast<int32_t>(depth), "value", int32());
  for (size_t level = depth - 1; level > 0; --level) {
    auto nested = struct_({field});
    field =
        optional_parent == level - 1
            ? SchemaField::MakeOptional(static_cast<int32_t>(level), "nested", nested)
            : SchemaField::MakeRequired(static_cast<int32_t>(level), "nested", nested);
  }
  return std::make_unique<Schema>(std::vector<SchemaField>{std::move(field)});
}

}  // namespace

TEST(ManifestFileStructLike, BasicFields) {
  ManifestFile manifest_file{
      .manifest_path = "/path/to/manifest.avro",
      .manifest_length = 12345,
      .partition_spec_id = 1,
      .content = ManifestContent::kData,
      .sequence_number = 100,
      .min_sequence_number = 90,
      .added_snapshot_id = 1001,
      .added_files_count = 10,
      .existing_files_count = 5,
      .deleted_files_count = 2,
      .added_rows_count = 1000,
      .existing_rows_count = 500,
      .deleted_rows_count = 20,
  };

  ManifestFileStructLike struct_like(manifest_file);
  EXPECT_EQ(struct_like.num_fields(), 16);

  EXPECT_SCALAR_EQ(
      struct_like.GetField(static_cast<size_t>(ManifestFileField::kManifestPath)),
      std::string_view, "/path/to/manifest.avro");
  EXPECT_SCALAR_EQ(
      struct_like.GetField(static_cast<size_t>(ManifestFileField::kManifestLength)),
      int64_t, 12345);
  EXPECT_SCALAR_EQ(
      struct_like.GetField(static_cast<size_t>(ManifestFileField::kPartitionSpecId)),
      int32_t, 1);
  EXPECT_SCALAR_EQ(struct_like.GetField(static_cast<size_t>(ManifestFileField::kContent)),
                   int32_t, static_cast<int32_t>(ManifestContent::kData));
  EXPECT_SCALAR_EQ(
      struct_like.GetField(static_cast<size_t>(ManifestFileField::kSequenceNumber)),
      int64_t, 100);
  EXPECT_SCALAR_EQ(
      struct_like.GetField(static_cast<size_t>(ManifestFileField::kAddedFilesCount)),
      int32_t, 10);
  EXPECT_THAT(struct_like.GetField(100), IsError(ErrorKind::kInvalidArgument));
}

TEST(LiteralToScalarTest, Uuid) {
  ICEBERG_UNWRAP_OR_FAIL(auto uuid,
                         Uuid::FromString("123e4567-e89b-12d3-a456-426614174000"));
  auto literal = Literal::UUID(uuid);

  ICEBERG_UNWRAP_OR_FAIL(auto scalar, LiteralToScalar(literal));
  ASSERT_TRUE(std::holds_alternative<std::string_view>(scalar));

  auto value = std::get<std::string_view>(scalar);
  ASSERT_EQ(value.size(), Uuid::kLength);

  const auto& expected_bytes = uuid.bytes();
  for (size_t i = 0; i < expected_bytes.size(); ++i) {
    EXPECT_EQ(static_cast<uint8_t>(static_cast<unsigned char>(value[i])),
              expected_bytes[i]);
  }
}

TEST(StructLikeAccessorTest, GetLiteralUuid) {
  ICEBERG_UNWRAP_OR_FAIL(auto uuid,
                         Uuid::FromString("123e4567-e89b-12d3-a456-426614174000"));
  const auto& bytes = uuid.bytes();
  std::string_view uuid_data(reinterpret_cast<const char*>(bytes.data()), bytes.size());
  SingleFieldStructLike row(Scalar{uuid_data});
  std::array<size_t, 1> path = {0};
  StructLikeAccessor accessor(iceberg::uuid(), path, {false});

  ICEBERG_UNWRAP_OR_FAIL(auto literal, accessor.GetLiteral(row));
  EXPECT_EQ(literal.type()->type_id(), TypeId::kUuid);
  ASSERT_TRUE(std::holds_alternative<Uuid>(literal.value()));
  EXPECT_EQ(std::get<Uuid>(literal.value()), uuid);
}

TEST(StructLikeAccessorTest, GetLiteralUuidRejectsWrongLength) {
  SingleFieldStructLike row(Scalar{std::string_view("not-a-uuid")});
  std::array<size_t, 1> path = {0};
  StructLikeAccessor accessor(iceberg::uuid(), path, {false});

  auto result = accessor.GetLiteral(row);
  EXPECT_THAT(result, IsError(ErrorKind::kInvalidArgument));
  EXPECT_THAT(result, HasErrorMessage("UUID byte array must be exactly 16 bytes"));
}

TEST(StructLikeAccessorTest, RequiredAndOptionalLeaf) {
  SingleFieldStructLike row(Scalar{std::monostate{}});
  for (bool optional : {false, true}) {
    auto schema = MakeNestedSchema(1, std::nullopt, optional);
    ICEBERG_UNWRAP_OR_FAIL(auto accessor, schema->GetAccessorById(1));
    if (optional) {
      EXPECT_SCALAR_NULL(accessor->Get(row));
      ICEBERG_UNWRAP_OR_FAIL(auto literal, accessor->GetLiteral(row));
      EXPECT_TRUE(literal.IsNull());
    } else {
      EXPECT_THAT(accessor->Get(row), IsError(ErrorKind::kInvalidArgument));
      EXPECT_THAT(accessor->GetLiteral(row), IsError(ErrorKind::kInvalidArgument));
    }
  }
}

class NestedStructLikeAccessorTest : public ::testing::TestWithParam<size_t> {};

TEST_P(NestedStructLikeAccessorTest, NullParents) {
  const std::vector<size_t> path(GetParam(), 0);
  StructLikeAccessor accessor(int32(), path, std::vector<bool>(path.size(), true));
  for (const Scalar null :
       {Scalar{std::monostate{}}, Scalar{std::shared_ptr<StructLike>{}}}) {
    for (size_t parent = 0; parent + 1 < path.size(); ++parent) {
      SCOPED_TRACE(parent);
      auto row = WrapSingleFieldRow(null, parent + 1);
      ICEBERG_UNWRAP_OR_FAIL(auto scalar, accessor.Get(*row));
      EXPECT_TRUE(std::holds_alternative<std::monostate>(scalar));
      ICEBERG_UNWRAP_OR_FAIL(auto literal, accessor.GetLiteral(*row));
      EXPECT_TRUE(literal.IsNull());
      EXPECT_EQ(literal.type()->type_id(), TypeId::kInt);
    }
  }
}

TEST_P(NestedStructLikeAccessorTest, SchemaRequiredAndOptionalParents) {
  const auto depth = GetParam();
  auto required_schema = MakeNestedSchema(depth);
  ICEBERG_UNWRAP_OR_FAIL(auto required_accessor,
                         required_schema->GetAccessorById(static_cast<int32_t>(depth)));
  EXPECT_EQ(required_accessor->position_path(), std::vector<size_t>(depth, 0));

  for (const Scalar null :
       {Scalar{std::monostate{}}, Scalar{std::shared_ptr<StructLike>{}}}) {
    for (size_t parent = 0; parent + 1 < depth; ++parent) {
      SCOPED_TRACE(parent);
      auto row = WrapSingleFieldRow(null, parent + 1);
      auto schema = MakeNestedSchema(depth, parent);
      ICEBERG_UNWRAP_OR_FAIL(auto accessor,
                             schema->GetAccessorById(static_cast<int32_t>(depth)));
      EXPECT_SCALAR_NULL(accessor->Get(*row));
      ICEBERG_UNWRAP_OR_FAIL(auto literal, accessor->GetLiteral(*row));
      EXPECT_TRUE(literal.IsNull());
      EXPECT_EQ(literal.type()->type_id(), TypeId::kInt);

      EXPECT_THAT(required_accessor->Get(*row), IsError(ErrorKind::kInvalidArgument));
      EXPECT_THAT(required_accessor->GetLiteral(*row),
                  IsError(ErrorKind::kInvalidArgument));
    }
  }
}

TEST_P(NestedStructLikeAccessorTest, ValueAndNullLeaf) {
  const std::vector<size_t> path(GetParam(), 0);
  StructLikeAccessor accessor(int32(), path, std::vector<bool>(path.size(), true));
  auto row = WrapSingleFieldRow(Scalar{int32_t{42}}, path.size());
  ICEBERG_UNWRAP_OR_FAIL(auto value, accessor.GetLiteral(*row));
  EXPECT_EQ(value, Literal::Int(42));

  auto null_row = WrapSingleFieldRow(Scalar{std::monostate{}}, path.size());
  ICEBERG_UNWRAP_OR_FAIL(auto null, accessor.GetLiteral(*null_row));
  EXPECT_TRUE(null.IsNull());
  EXPECT_EQ(null.type()->type_id(), TypeId::kInt);
}

TEST_P(NestedStructLikeAccessorTest, RequiredAndOptionalLeaf) {
  const auto depth = GetParam();
  auto row = WrapSingleFieldRow(Scalar{std::monostate{}}, depth);
  for (bool optional : {false, true}) {
    auto schema = MakeNestedSchema(depth, std::nullopt, optional);
    ICEBERG_UNWRAP_OR_FAIL(auto accessor,
                           schema->GetAccessorById(static_cast<int32_t>(depth)));
    if (optional) {
      EXPECT_SCALAR_NULL(accessor->Get(*row));
      ICEBERG_UNWRAP_OR_FAIL(auto literal, accessor->GetLiteral(*row));
      EXPECT_TRUE(literal.IsNull());
    } else {
      EXPECT_THAT(accessor->Get(*row), IsError(ErrorKind::kInvalidArgument));
      EXPECT_THAT(accessor->GetLiteral(*row), IsError(ErrorKind::kInvalidArgument));
    }
  }

  auto schema = MakeNestedSchema(depth, 0);
  ICEBERG_UNWRAP_OR_FAIL(auto accessor,
                         schema->GetAccessorById(static_cast<int32_t>(depth)));
  EXPECT_THAT(accessor->Get(*row), IsError(ErrorKind::kInvalidArgument));
  auto absent_parent = WrapSingleFieldRow(Scalar{std::monostate{}}, 1);
  EXPECT_SCALAR_NULL(accessor->Get(*absent_parent));
}

TEST_P(NestedStructLikeAccessorTest, RejectsNonStructParents) {
  const std::vector<size_t> path(GetParam(), 0);
  StructLikeAccessor accessor(int32(), path, std::vector<bool>(path.size(), true));
  for (size_t parent = 0; parent + 1 < path.size(); ++parent) {
    SCOPED_TRACE(parent);
    auto row = WrapSingleFieldRow(Scalar{int32_t{42}}, parent + 1);
    EXPECT_THAT(accessor.Get(*row), IsError(ErrorKind::kInvalidSchema));
  }
}

TEST_P(NestedStructLikeAccessorTest, PreservesFieldErrors) {
  const std::vector<size_t> path(GetParam(), 0);
  StructLikeAccessor accessor(int32(), path, std::vector<bool>(path.size(), true));
  for (size_t level = 0; level < path.size(); ++level) {
    SCOPED_TRACE(level);
    auto row = WrapSingleFieldRow(IOError("Cannot read field"), level + 1);
    auto result = accessor.GetLiteral(*row);
    EXPECT_THAT(result, IsError(ErrorKind::kIOError));
    EXPECT_THAT(result, HasErrorMessage("Cannot read field"));
  }
}

INSTANTIATE_TEST_SUITE_P(NestedPaths, NestedStructLikeAccessorTest,
                         ::testing::Values(size_t{2}, size_t{3}, size_t{4}));

TEST(ManifestFileStructLike, OptionalFields) {
  ManifestFile manifest_file{.manifest_path = "/path/to/manifest2.avro",
                             .manifest_length = 54321,
                             .partition_spec_id = 2,
                             .content = ManifestContent::kDeletes,
                             .sequence_number = 200,
                             .min_sequence_number = 180,
                             .added_snapshot_id = 2001,
                             .added_files_count = std::nullopt,  // null optional field
                             .existing_files_count = 15,
                             .deleted_files_count = std::nullopt,  // null optional field
                             .added_rows_count = std::nullopt,     // null optional field
                             .existing_rows_count = 1500,
                             .deleted_rows_count = 200,
                             .partitions = {},
                             .key_metadata = {},
                             .first_row_id = 12345};
  ManifestFileStructLike struct_like(manifest_file);

  EXPECT_SCALAR_NULL(
      struct_like.GetField(static_cast<size_t>(ManifestFileField::kAddedFilesCount)));
  EXPECT_SCALAR_EQ(
      struct_like.GetField(static_cast<size_t>(ManifestFileField::kExistingFilesCount)),
      int32_t, 15);
  EXPECT_SCALAR_EQ(
      struct_like.GetField(static_cast<size_t>(ManifestFileField::kFirstRowId)), int64_t,
      12345);
  EXPECT_SCALAR_EQ(struct_like.GetField(static_cast<size_t>(ManifestFileField::kContent)),
                   int32_t, static_cast<int32_t>(ManifestContent::kDeletes));
}

TEST(ManifestFileStructLike, WithPartitions) {
  ManifestFile manifest_file{
      .manifest_path = "/path/to/manifest3.avro",
      .manifest_length = 98765,
      .partition_spec_id = 3,
      .content = ManifestContent::kData,
      .sequence_number = 300,
      .min_sequence_number = 290,
      .added_snapshot_id = 3001,
      .added_files_count = 20,
      .existing_files_count = 10,
      .deleted_files_count = 1,
      .added_rows_count = 2000,
      .existing_rows_count = 1000,
      .deleted_rows_count = 10,
      .partitions = {{.contains_null = true,
                      .contains_nan = false,
                      .lower_bound = std::vector<uint8_t>{0x01, 0x02, 0x03},
                      .upper_bound = std::vector<uint8_t>{0x04, 0x05, 0x06}},
                     {.contains_null = false,
                      .contains_nan = std::nullopt,
                      .lower_bound = std::vector<uint8_t>{0x10, 0x20},
                      .upper_bound = std::nullopt}}};

  ManifestFileStructLike struct_like(manifest_file);

  auto partitions_result = struct_like.GetField(
      static_cast<size_t>(ManifestFileField::kPartitionFieldSummary));
  ASSERT_THAT(partitions_result, IsOk());
  auto partitions_scalar = partitions_result.value();
  ASSERT_TRUE(std::holds_alternative<std::shared_ptr<ArrayLike>>(partitions_scalar));
  auto partitions_array = std::get<std::shared_ptr<ArrayLike>>(partitions_scalar);
  EXPECT_EQ(partitions_array->size(), 2);

  // Test 1st partition summary
  auto first_partition_result = partitions_array->GetElement(0);
  ASSERT_THAT(first_partition_result, IsOk());
  auto first_partition_scalar = first_partition_result.value();
  ASSERT_TRUE(
      std::holds_alternative<std::shared_ptr<StructLike>>(first_partition_scalar));
  auto first_partition_struct =
      std::get<std::shared_ptr<StructLike>>(first_partition_scalar);
  EXPECT_EQ(first_partition_struct->num_fields(), 4);
  EXPECT_SCALAR_EQ(first_partition_struct->GetField(0), bool, true);
  EXPECT_SCALAR_EQ(first_partition_struct->GetField(1), bool, false);
  auto lower_bound_result = first_partition_struct->GetField(2);
  ASSERT_THAT(lower_bound_result, IsOk());
  auto lower_bound_scalar = lower_bound_result.value();
  ASSERT_TRUE(std::holds_alternative<std::string_view>(lower_bound_scalar));
  auto lower_bound_view = std::get<std::string_view>(lower_bound_scalar);
  EXPECT_EQ(lower_bound_view.size(), 3);
  EXPECT_EQ(static_cast<uint8_t>(lower_bound_view[0]), 0x01);
  EXPECT_EQ(static_cast<uint8_t>(lower_bound_view[1]), 0x02);
  EXPECT_EQ(static_cast<uint8_t>(lower_bound_view[2]), 0x03);

  // Test 2nd partition summary with null fields
  auto second_partition_result = partitions_array->GetElement(1);
  ASSERT_THAT(second_partition_result, IsOk());
  auto second_partition_scalar = second_partition_result.value();
  ASSERT_TRUE(
      std::holds_alternative<std::shared_ptr<StructLike>>(second_partition_scalar));
  auto second_partition_struct =
      std::get<std::shared_ptr<StructLike>>(second_partition_scalar);
  EXPECT_SCALAR_NULL(second_partition_struct->GetField(1));
  EXPECT_SCALAR_NULL(second_partition_struct->GetField(3));
}

TEST(ArrowArrayStructLike, PrimitiveFields) {
  auto struct_type = ::arrow::struct_(
      {::arrow::field("id", ::arrow::int64(), /*nullable=*/false),
       ::arrow::field("name", ::arrow::utf8(), /*nullable=*/true),
       ::arrow::field("score", ::arrow::float32(), /*nullable=*/true),
       ::arrow::field("active", ::arrow::boolean(), /*nullable=*/false),
       ::arrow::field("date", ::arrow::date32(), /*nullable=*/false),
       ::arrow::field("time", ::arrow::time64(::arrow::TimeUnit::MICRO),
                      /*nullable=*/false),
       ::arrow::field("timestamp", ::arrow::timestamp(::arrow::TimeUnit::MICRO),
                      /*nullable=*/false),
       ::arrow::field("fixed", ::arrow::fixed_size_binary(4), /*nullable=*/false),
       ::arrow::field("decimal", ::arrow::decimal128(10, 2), /*nullable=*/false)});

  auto arrow_array = ::arrow::json::ArrayFromJSONString(struct_type, R"([
    {"id": 1, "name": "Alice", "score": 95.5, "active": true, "date": 1714396800,
     "time": 123456, "timestamp": 1714396800000000, "fixed": "aaaa", "decimal": "1234.56"},
    {"id": 2, "name": "Bob", "score": null, "active": false, "date": 1714396801,
     "time": 123457, "timestamp": 1714396800000001, "fixed": "bbbb", "decimal": "-1234.56"},
    {"id": 3, "name": null, "score": 87.2, "active": true, "date": 1714396802,
     "time": 123458, "timestamp": 1714396800000002, "fixed": "cccc", "decimal": "1234.00"}])")
                         .ValueOrDie();

  ArrowSchema c_schema;
  ArrowArray c_array;
  internal::ArrowSchemaGuard schema_guard(&c_schema);
  internal::ArrowArrayGuard array_guard(&c_array);
  ASSERT_TRUE(::arrow::ExportType(*struct_type, &c_schema).ok());
  ASSERT_TRUE(::arrow::ExportArray(*arrow_array, &c_array).ok());

  auto struct_like_result = ArrowArrayStructLike::Make(c_schema, c_array);
  ASSERT_THAT(struct_like_result, IsOk());
  auto struct_like = std::move(struct_like_result.value());

  constexpr int64_t kNumRows = 3;
  std::array<int64_t, kNumRows> ids = {1, 2, 3};
  std::array<std::optional<std::string>, kNumRows> names = {"Alice", "Bob", std::nullopt};
  std::array<std::optional<float>, kNumRows> scores = {95.5f, std::nullopt, 87.2f};
  std::array<bool, kNumRows> actives = {true, false, true};
  std::array<int32_t, kNumRows> dates = {1714396800, 1714396801, 1714396802};
  std::array<int64_t, kNumRows> times = {123456, 123457, 123458};
  std::array<int64_t, kNumRows> timestamps = {1714396800000000, 1714396800000001,
                                              1714396800000002};
  std::array<std::string, kNumRows> fixeds = {"aaaa", "bbbb", "cccc"};
  std::array<std::string, kNumRows> decimals = {"1234.56", "-1234.56", "1234.00"};

  for (int64_t i = 0; i < kNumRows; ++i) {
    ASSERT_THAT(struct_like->Reset(i), IsOk());
    EXPECT_SCALAR_EQ(struct_like->GetField(0), int64_t, ids[i]);
    if (names[i].has_value()) {
      EXPECT_SCALAR_EQ(struct_like->GetField(1), std::string_view, names[i]);
    } else {
      EXPECT_SCALAR_NULL(struct_like->GetField(1));
    }
    if (scores[i].has_value()) {
      EXPECT_SCALAR_EQ(struct_like->GetField(2), float, scores[i].value());
    } else {
      EXPECT_SCALAR_NULL(struct_like->GetField(2));
    }
    EXPECT_SCALAR_EQ(struct_like->GetField(3), bool, actives[i]);
    EXPECT_SCALAR_EQ(struct_like->GetField(4), int32_t, dates[i]);
    EXPECT_SCALAR_EQ(struct_like->GetField(5), int64_t, times[i]);
    EXPECT_SCALAR_EQ(struct_like->GetField(6), int64_t, timestamps[i]);
    EXPECT_SCALAR_EQ(struct_like->GetField(7), std::string_view, fixeds[i]);
    EXPECT_DECIMAL_EQ(struct_like->GetField(8), /*scale=*/2, decimals[i]);
  }
}

TEST(ArrowArrayStructLike, NestedStruct) {
  auto person_type =
      ::arrow::struct_({::arrow::field("name", ::arrow::utf8(), /*nullable=*/false),
                        ::arrow::field("age", ::arrow::int32(), /*nullable=*/false)});
  auto root_type =
      ::arrow::struct_({::arrow::field("id", ::arrow::int64(), /*nullable=*/false),
                        ::arrow::field("person", person_type, /*nullable=*/false)});

  auto arrow_array = ::arrow::json::ArrayFromJSONString(root_type, R"([
    {"id": 1, "person": {"name": "Alice", "age": 30}},
    {"id": 2, "person": {"name": "Bob", "age": 25}}])")
                         .ValueOrDie();

  ArrowSchema c_schema;
  ArrowArray c_array;
  internal::ArrowSchemaGuard schema_guard(&c_schema);
  internal::ArrowArrayGuard array_guard(&c_array);
  ASSERT_TRUE(::arrow::ExportType(*root_type, &c_schema).ok());
  ASSERT_TRUE(::arrow::ExportArray(*arrow_array, &c_array).ok());

  auto struct_like_result = ArrowArrayStructLike::Make(c_schema, c_array);
  ASSERT_THAT(struct_like_result, IsOk());
  auto struct_like = std::move(struct_like_result.value());

  constexpr int64_t kNumRows = 2;
  std::array<int64_t, kNumRows> ids = {1, 2};
  std::array<std::string, kNumRows> names = {"Alice", "Bob"};
  std::array<int32_t, kNumRows> ages = {30, 25};

  for (int64_t i = 0; i < kNumRows; ++i) {
    ASSERT_THAT(struct_like->Reset(i), IsOk());
    EXPECT_EQ(struct_like->num_fields(), 2);
    EXPECT_SCALAR_EQ(struct_like->GetField(0), int64_t, ids[i]);

    auto person_result = struct_like->GetField(1);
    ASSERT_THAT(person_result, IsOk());
    auto person_scalar = person_result.value();
    ASSERT_TRUE(std::holds_alternative<std::shared_ptr<StructLike>>(person_scalar));

    auto person_struct = std::get<std::shared_ptr<StructLike>>(person_scalar);
    EXPECT_EQ(person_struct->num_fields(), 2);
    EXPECT_SCALAR_EQ(person_struct->GetField(0), std::string_view, names[i]);
    EXPECT_SCALAR_EQ(person_struct->GetField(1), int32_t, ages[i]);
  }
}

TEST(ArrowArrayStructLike, PrimitiveList) {
  auto list_type =
      ::arrow::list(::arrow::field("item", ::arrow::int32(), /*nullable=*/false));

  auto arrow_array = ::arrow::json::ArrayFromJSONString(list_type, R"([
    [1, 2, 3, 4, 5],
    [10, 20],
    []])")
                         .ValueOrDie();

  ArrowSchema c_schema;
  ArrowArray c_array;
  internal::ArrowSchemaGuard schema_guard(&c_schema);
  internal::ArrowArrayGuard array_guard(&c_array);
  ASSERT_TRUE(::arrow::ExportType(*list_type, &c_schema).ok());
  ASSERT_TRUE(::arrow::ExportArray(*arrow_array, &c_array).ok());

  auto array_like_result = ArrowArrayArrayLike::Make(c_schema, c_array);
  ASSERT_THAT(array_like_result, IsOk());
  auto array_like = std::move(array_like_result.value());

  constexpr int64_t kNumRows = 3;
  std::array<std::vector<int32_t>, kNumRows> expected_lists = {
      std::vector<int32_t>{1, 2, 3, 4, 5},
      std::vector<int32_t>{10, 20},
      std::vector<int32_t>{},
  };

  for (int64_t i = 0; i < kNumRows; ++i) {
    ASSERT_THAT(array_like->Reset(i), IsOk());
    const auto& expected_list = expected_lists[i];
    ASSERT_EQ(array_like->size(), expected_list.size());
    for (size_t j = 0; j < expected_list.size(); ++j) {
      EXPECT_SCALAR_EQ(array_like->GetElement(j), int32_t, expected_list[j]);
    }
  }
}

TEST(ArrowArrayStructLike, PrimitiveMap) {
  auto map_type = std::make_shared<::arrow::MapType>(
      ::arrow::field("key", ::arrow::utf8(), /*nullable=*/false),
      ::arrow::field("value", ::arrow::int32(), /*nullable=*/false));

  auto arrow_array = ::arrow::json::ArrayFromJSONString(map_type, R"([
    [["Foo", 1], ["Bar", 2]],
    [["Baz", 1]],
    []])")
                         .ValueOrDie();

  ArrowSchema c_schema;
  ArrowArray c_array;
  internal::ArrowSchemaGuard schema_guard(&c_schema);
  internal::ArrowArrayGuard array_guard(&c_array);
  ASSERT_TRUE(::arrow::ExportType(*map_type, &c_schema).ok());
  ASSERT_TRUE(::arrow::ExportArray(*arrow_array, &c_array).ok());

  auto map_like_result = ArrowArrayMapLike::Make(c_schema, c_array);
  ASSERT_THAT(map_like_result, IsOk());
  auto map_like = std::move(map_like_result.value());

  constexpr int64_t kNumRows = 3;
  std::array<std::vector<std::pair<std::string, int32_t>>, kNumRows> expected_maps = {
      std::vector<std::pair<std::string, int32_t>>{{"Foo", 1}, {"Bar", 2}},
      std::vector<std::pair<std::string, int32_t>>{{"Baz", 1}},
      std::vector<std::pair<std::string, int32_t>>{},
  };

  for (int64_t i = 0; i < kNumRows; ++i) {
    ASSERT_THAT(map_like->Reset(i), IsOk());
    const auto& expected_map = expected_maps[i];
    ASSERT_EQ(map_like->size(), expected_map.size());
    for (size_t j = 0; j < expected_map.size(); ++j) {
      EXPECT_SCALAR_EQ(map_like->GetKey(j), std::string_view, expected_map[j].first);
      EXPECT_SCALAR_EQ(map_like->GetValue(j), int32_t, expected_map[j].second);
    }
  }
}

TEST(ArrowArrayStructLike, Accessor) {
  Schema schema{std::vector<SchemaField>{
      SchemaField::MakeOptional(1, "c1", int32()),
      SchemaField::MakeOptional(
          2, "c2",
          struct_({
              SchemaField::MakeOptional(3, "c3", int32()),
              SchemaField::MakeOptional(4, "c4",
                                        struct_({
                                            SchemaField::MakeOptional(5, "c5", int32()),
                                        })),
          })),
  }};

  auto arrow_schema = ::arrow::struct_({
      ::arrow::field("c1", ::arrow::int32()),
      ::arrow::field("c2",
                     ::arrow::struct_({
                         ::arrow::field("c3", ::arrow::int32()),
                         ::arrow::field("c4", ::arrow::struct_({
                                                  ::arrow::field("c5", ::arrow::int32()),
                                              })),
                     })),
  });

  auto arrow_array =
      ::arrow::json::ArrayFromJSONString(
          arrow_schema, R"([ {"c1": 1, "c2": {"c3": 3, "c4": {"c5": 5}}} ])")
          .ValueOrDie();

  ArrowSchema c_schema;
  ArrowArray c_array;
  internal::ArrowSchemaGuard schema_guard(&c_schema);
  internal::ArrowArrayGuard array_guard(&c_array);
  ASSERT_TRUE(::arrow::ExportType(*arrow_schema, &c_schema).ok());
  ASSERT_TRUE(::arrow::ExportArray(*arrow_array, &c_array).ok());

  ICEBERG_UNWRAP_OR_FAIL(auto struct_like, ArrowArrayStructLike::Make(c_schema, c_array));

  // Test nested accessors from 1 to 3 levels deep
  for (int32_t field_id : {1, 3, 5}) {
    ICEBERG_UNWRAP_OR_FAIL(auto accessor, schema.GetAccessorById(field_id));
    ICEBERG_UNWRAP_OR_FAIL(auto scalar, accessor->Get(*struct_like));
    ASSERT_TRUE(std::holds_alternative<int32_t>(scalar));
    EXPECT_EQ(std::get<int32_t>(scalar), field_id);
  }
}

TEST(ArrowArrayStructLike, SlicedStruct) {
  auto person_type =
      ::arrow::struct_({::arrow::field("name", ::arrow::utf8(), /*nullable=*/false)});
  auto root_type = ::arrow::struct_(
      {::arrow::field("id", ::arrow::int64(), /*nullable=*/false),
       ::arrow::field("person", person_type, /*nullable=*/false),
       ::arrow::field("tags", ::arrow::list(::arrow::int32()), /*nullable=*/false)});

  auto arrow_array = ::arrow::json::ArrayFromJSONString(root_type, R"([
    {"id": 1, "person": {"name": "Alice"}, "tags": [1]},
    {"id": 2, "person": {"name": "Bob"}, "tags": [2, 3]},
    {"id": 3, "person": {"name": "Carol"}, "tags": []}])")
                         .ValueOrDie();
  auto sliced = arrow_array->Slice(1);

  ArrowSchema c_schema;
  ArrowArray c_array;
  internal::ArrowSchemaGuard schema_guard(&c_schema);
  internal::ArrowArrayGuard array_guard(&c_array);
  ASSERT_TRUE(::arrow::ExportType(*root_type, &c_schema).ok());
  ASSERT_TRUE(::arrow::ExportArray(*sliced, &c_array).ok());
  ASSERT_EQ(c_array.offset, 1);

  ICEBERG_UNWRAP_OR_FAIL(auto struct_like, ArrowArrayStructLike::Make(c_schema, c_array));

  constexpr int64_t kNumRows = 2;
  std::array<int64_t, kNumRows> ids = {2, 3};
  std::array<std::string, kNumRows> names = {"Bob", "Carol"};
  std::array<std::vector<int32_t>, kNumRows> tags = {std::vector<int32_t>{2, 3},
                                                     std::vector<int32_t>{}};

  for (int64_t i = 0; i < kNumRows; ++i) {
    ASSERT_THAT(struct_like->Reset(i), IsOk());
    EXPECT_SCALAR_EQ(struct_like->GetField(0), int64_t, ids[i]);

    ICEBERG_UNWRAP_OR_FAIL(auto person_scalar, struct_like->GetField(1));
    auto person = std::get<std::shared_ptr<StructLike>>(person_scalar);
    EXPECT_SCALAR_EQ(person->GetField(0), std::string_view, names[i]);

    ICEBERG_UNWRAP_OR_FAIL(auto tags_scalar, struct_like->GetField(2));
    auto tags_like = std::get<std::shared_ptr<ArrayLike>>(tags_scalar);
    ASSERT_EQ(tags_like->size(), tags[i].size());
    for (size_t j = 0; j < tags[i].size(); ++j) {
      EXPECT_SCALAR_EQ(tags_like->GetElement(j), int32_t, tags[i][j]);
    }
  }
}

TEST(ArrowArrayStructLike, SlicedList) {
  auto list_type = ::arrow::list(::arrow::int32());
  auto arrow_array =
      ::arrow::json::ArrayFromJSONString(list_type, R"([[1], [2, 3], [4]])").ValueOrDie();
  auto sliced = arrow_array->Slice(1);

  ArrowSchema c_schema;
  ArrowArray c_array;
  internal::ArrowSchemaGuard schema_guard(&c_schema);
  internal::ArrowArrayGuard array_guard(&c_array);
  ASSERT_TRUE(::arrow::ExportType(*list_type, &c_schema).ok());
  ASSERT_TRUE(::arrow::ExportArray(*sliced, &c_array).ok());

  ICEBERG_UNWRAP_OR_FAIL(auto array_like, ArrowArrayArrayLike::Make(c_schema, c_array));
  ASSERT_EQ(array_like->size(), 2);
  EXPECT_SCALAR_EQ(array_like->GetElement(0), int32_t, 2);
  EXPECT_SCALAR_EQ(array_like->GetElement(1), int32_t, 3);

  ASSERT_THAT(array_like->Reset(1), IsOk());
  ASSERT_EQ(array_like->size(), 1);
  EXPECT_SCALAR_EQ(array_like->GetElement(0), int32_t, 4);
}

TEST(ArrowArrayStructLike, SlicedMap) {
  auto map_type = std::make_shared<::arrow::MapType>(
      ::arrow::field("key", ::arrow::utf8(), /*nullable=*/false),
      ::arrow::field("value", ::arrow::int32(), /*nullable=*/false));
  auto entries_source =
      ::arrow::json::ArrayFromJSONString(map_type, R"([[["X", 0], ["Foo", 1],
                                                        ["Bar", 2], ["Baz", 3]]])")
          .ValueOrDie();
  auto entries =
      std::static_pointer_cast<::arrow::MapArray>(entries_source)->values()->Slice(1);
  auto offsets = ::arrow::Buffer::FromVector(std::vector<int32_t>{0, 0, 2, 3, 3});
  auto map_data = ::arrow::ArrayData::Make(map_type, /*length=*/4, {nullptr, offsets},
                                           {entries->data()}, /*null_count=*/0);
  auto sliced = ::arrow::MakeArray(map_data)->Slice(1);
  ASSERT_TRUE(sliced->ValidateFull().ok());

  ArrowSchema c_schema;
  ArrowArray c_array;
  internal::ArrowSchemaGuard schema_guard(&c_schema);
  internal::ArrowArrayGuard array_guard(&c_array);
  ASSERT_TRUE(::arrow::ExportType(*map_type, &c_schema).ok());
  ASSERT_TRUE(::arrow::ExportArray(*sliced, &c_array).ok());
  ASSERT_EQ(c_array.offset, 1);
  ASSERT_EQ(c_array.children[0]->offset, 1);

  ICEBERG_UNWRAP_OR_FAIL(auto map_like, ArrowArrayMapLike::Make(c_schema, c_array));

  constexpr int64_t kNumRows = 3;
  std::array<std::vector<std::pair<std::string, int32_t>>, kNumRows> expected_maps = {
      std::vector<std::pair<std::string, int32_t>>{{"Foo", 1}, {"Bar", 2}},
      std::vector<std::pair<std::string, int32_t>>{{"Baz", 3}},
      std::vector<std::pair<std::string, int32_t>>{},
  };

  for (int64_t i = 0; i < kNumRows; ++i) {
    ASSERT_THAT(map_like->Reset(i), IsOk());
    const auto& expected_map = expected_maps[i];
    ASSERT_EQ(map_like->size(), expected_map.size());
    for (size_t j = 0; j < expected_map.size(); ++j) {
      EXPECT_SCALAR_EQ(map_like->GetKey(j), std::string_view, expected_map[j].first);
      EXPECT_SCALAR_EQ(map_like->GetValue(j), int32_t, expected_map[j].second);
    }
  }
}

}  // namespace iceberg