feat: implement thread-safe lazy initialization for Schema and StructType

- Added move and copy constructors and assignment operators for Schema and StructType to manage resource ownership and improve performance.
- Refactored field lookup methods to utilize lazy initialization with thread safety, ensuring safe concurrent access.
- Introduced unit tests for thread safety in Schema and StructType, validating concurrent operations and access patterns.

Author: nullccxsy

src/iceberg/schema.cc

@@ -71,6 +71,42 @@ class NameToIdVisitor {
 Schema::Schema(std::vector<SchemaField> fields, std::optional<int32_t> schema_id)
     : StructType(std::move(fields)), schema_id_(schema_id) {}
 
+Schema::Schema(Schema&& other) noexcept
+    : StructType(std::move(other)),
+      schema_id_(other.schema_id_),
+      id_to_field_(std::move(other.id_to_field_)),
+      name_to_id_(std::move(other.name_to_id_)),
+      lowercase_name_to_id_(std::move(other.lowercase_name_to_id_)) {}
+
+Schema& Schema::operator=(const Schema& other) {
+  if (this != &other) {
+    StructType::operator=(other);
+    schema_id_ = other.schema_id_;
+    id_to_field_ = other.id_to_field_;
+    name_to_id_ = other.name_to_id_;
+    lowercase_name_to_id_ = other.lowercase_name_to_id_;
+  }
+  return *this;
+}
+
+Schema::Schema(const Schema& other)
+    : StructType(other),
+      schema_id_(other.schema_id_),
+      id_to_field_(other.id_to_field_),
+      name_to_id_(other.name_to_id_),
+      lowercase_name_to_id_(other.lowercase_name_to_id_) {}
+
+Schema& Schema::operator=(Schema&& other) noexcept {
+  if (this != &other) {
+    StructType::operator=(std::move(other));
+    schema_id_ = other.schema_id_;
+    id_to_field_ = std::move(other.id_to_field_);
+    name_to_id_ = std::move(other.name_to_id_);
+    lowercase_name_to_id_ = std::move(other.lowercase_name_to_id_);
+  }
+  return *this;
+}
+
 std::optional<int32_t> Schema::schema_id() const { return schema_id_; }
 
 std::string Schema::ToString() const {
@@ -89,12 +125,14 @@ bool Schema::Equals(const Schema& other) const {
 Result<std::optional<std::reference_wrapper<const SchemaField>>> Schema::FindFieldByName(
     std::string_view name, bool case_sensitive) const {
   if (case_sensitive) {
-    ICEBERG_RETURN_UNEXPECTED(InitNameToIdMap());
+    ICEBERG_RETURN_UNEXPECTED(
+        LazyInitWithCallOnce(name_flag_, [this]() { return InitNameToIdMap(); }));
     auto it = name_to_id_.find(name);
     if (it == name_to_id_.end()) return std::nullopt;
     return FindFieldById(it->second);
   }
-  ICEBERG_RETURN_UNEXPECTED(InitLowerCaseNameToIdMap());
+  ICEBERG_RETURN_UNEXPECTED(LazyInitWithCallOnce(
+      lowercase_name_flag_, [this]() { return InitLowerCaseNameToIdMap(); }));
   auto it = lowercase_name_to_id_.find(StringUtils::ToLower(name));
   if (it == lowercase_name_to_id_.end()) return std::nullopt;
   return FindFieldById(it->second);
@@ -133,7 +171,8 @@ Status Schema::InitLowerCaseNameToIdMap() const {
 
 Result<std::optional<std::reference_wrapper<const SchemaField>>> Schema::FindFieldById(
     int32_t field_id) const {
-  ICEBERG_RETURN_UNEXPECTED(InitIdToFieldMap());
+  ICEBERG_RETURN_UNEXPECTED(
+      LazyInitWithCallOnce(id_flag_, [this]() { return InitIdToFieldMap(); }));
   auto it = id_to_field_.find(field_id);
   if (it == id_to_field_.end()) {
     return std::nullopt;

src/iceberg/schema.h

@@ -24,6 +24,7 @@
 /// and any utility functions.  See iceberg/type.h and iceberg/field.h as well.
 
 #include <cstdint>
+#include <mutex>
 #include <optional>
 #include <string>
 #include <vector>
@@ -48,6 +49,10 @@ class ICEBERG_EXPORT Schema : public StructType {
   explicit Schema(std::vector<SchemaField> fields,
                   std::optional<int32_t> schema_id = std::nullopt);
 
+  Schema(const Schema& other);
+  Schema(Schema&& other) noexcept;
+  Schema& operator=(const Schema& other);
+  Schema& operator=(Schema&& other) noexcept;
   /// \brief Get the schema ID.
   ///
   /// A schema is identified by a unique ID for the purposes of schema
@@ -78,13 +83,11 @@ class ICEBERG_EXPORT Schema : public StructType {
   /// \brief Compare two schemas for equality.
   [[nodiscard]] bool Equals(const Schema& other) const;
 
-  // TODO(nullccxsy): Address potential concurrency issues in lazy initialization (e.g.,
-  // use std::call_once)
   Status InitIdToFieldMap() const;
   Status InitNameToIdMap() const;
   Status InitLowerCaseNameToIdMap() const;
 
-  const std::optional<int32_t> schema_id_;
+  std::optional<int32_t> schema_id_;
   /// Mapping from field id to field.
   mutable std::unordered_map<int32_t, std::reference_wrapper<const SchemaField>>
       id_to_field_;
@@ -94,6 +97,10 @@ class ICEBERG_EXPORT Schema : public StructType {
   /// Mapping from lowercased field name to field id
   mutable std::unordered_map<std::string, int32_t, StringHash, std::equal_to<>>
       lowercase_name_to_id_;
+
+  mutable std::once_flag id_flag_;
+  mutable std::once_flag name_flag_;
+  mutable std::once_flag lowercase_name_flag_;
 };
 
 }  // namespace iceberg

src/iceberg/type.cc

@@ -48,9 +48,38 @@ std::string StructType::ToString() const {
   return repr;
 }
 std::span<const SchemaField> StructType::fields() const { return fields_; }
+StructType::StructType(const StructType& other)
+    : fields_(other.fields_),
+      field_by_id_(other.field_by_id_),
+      field_by_name_(other.field_by_name_),
+      field_by_lowercase_name_(other.field_by_lowercase_name_) {}
+StructType::StructType(StructType&& other) noexcept
+    : fields_(std::move(other.fields_)),
+      field_by_id_(std::move(other.field_by_id_)),
+      field_by_name_(std::move(other.field_by_name_)),
+      field_by_lowercase_name_(std::move(other.field_by_lowercase_name_)) {}
+StructType& StructType::operator=(const StructType& other) {
+  if (*this != other) {
+    fields_ = other.fields_;
+    field_by_id_ = other.field_by_id_;
+    field_by_name_ = other.field_by_name_;
+    field_by_lowercase_name_ = other.field_by_lowercase_name_;
+  }
+  return *this;
+}
+StructType& StructType::operator=(StructType&& other) noexcept {
+  if (*this != other) {
+    fields_ = std::move(other.fields_);
+    field_by_id_ = std::move(other.field_by_id_);
+    field_by_name_ = std::move(other.field_by_name_);
+    field_by_lowercase_name_ = std::move(other.field_by_lowercase_name_);
+  }
+  return *this;
+}
 Result<std::optional<NestedType::SchemaFieldConstRef>> StructType::GetFieldById(
     int32_t field_id) const {
-  ICEBERG_RETURN_UNEXPECTED(InitFieldById());
+  ICEBERG_RETURN_UNEXPECTED(
+      LazyInitWithCallOnce(field_by_id_flag_, [this]() { return InitFieldById(); }));
   auto it = field_by_id_.find(field_id);
   if (it == field_by_id_.end()) return std::nullopt;
   return it->second;
@@ -65,14 +94,16 @@ Result<std::optional<NestedType::SchemaFieldConstRef>> StructType::GetFieldByInd
 Result<std::optional<NestedType::SchemaFieldConstRef>> StructType::GetFieldByName(
     std::string_view name, bool case_sensitive) const {
   if (case_sensitive) {
-    ICEBERG_RETURN_UNEXPECTED(InitFieldByName());
+    ICEBERG_RETURN_UNEXPECTED(LazyInitWithCallOnce(
+        field_by_name_flag_, [this]() { return InitFieldByName(); }));
     auto it = field_by_name_.find(name);
     if (it != field_by_name_.end()) {
       return it->second;
     }
     return std::nullopt;
   }
-  ICEBERG_RETURN_UNEXPECTED(InitFieldByLowerCaseName());
+  ICEBERG_RETURN_UNEXPECTED(LazyInitWithCallOnce(
+      field_by_lowercase_name_flag_, [this]() { return InitFieldByLowerCaseName(); }));
   auto it = field_by_lowercase_name_.find(StringUtils::ToLower(name));
   if (it != field_by_lowercase_name_.end()) {
     return it->second;

src/iceberg/type.h

@@ -26,6 +26,7 @@
 #include <array>
 #include <cstdint>
 #include <memory>
+#include <mutex>
 #include <optional>
 #include <span>
 #include <string>
@@ -39,6 +40,13 @@
 
 namespace iceberg {
 
+template <typename Func>
+Status LazyInitWithCallOnce(std::once_flag& flag, Func&& func) {
+  Status status;
+  std::call_once(flag, [&status, &func]() { status = func(); });
+  return status;
+}
+
 /// \brief Interface for a data type for a field.
 class ICEBERG_EXPORT Type : public iceberg::util::Formattable {
  public:
@@ -109,6 +117,10 @@ class ICEBERG_EXPORT StructType : public NestedType {
   constexpr static TypeId kTypeId = TypeId::kStruct;
   explicit StructType(std::vector<SchemaField> fields);
   ~StructType() override = default;
+  StructType(const StructType&);
+  StructType(StructType&&) noexcept;
+  StructType& operator=(const StructType&);
+  StructType& operator=(StructType&&) noexcept;
 
   TypeId type_id() const override;
   std::string ToString() const override;
@@ -124,8 +136,7 @@ class ICEBERG_EXPORT StructType : public NestedType {
 
  protected:
   bool Equals(const Type& other) const override;
-  // TODO(nullccxsy): Lazy initialization has concurrency issues, need to add proper
-  // synchronization mechanism
+
   Status InitFieldById() const;
   Status InitFieldByName() const;
   Status InitFieldByLowerCaseName() const;
@@ -134,6 +145,10 @@ class ICEBERG_EXPORT StructType : public NestedType {
   mutable std::unordered_map<int32_t, SchemaFieldConstRef> field_by_id_;
   mutable std::unordered_map<std::string_view, SchemaFieldConstRef> field_by_name_;
   mutable std::unordered_map<std::string, SchemaFieldConstRef> field_by_lowercase_name_;
+
+  mutable std::once_flag field_by_id_flag_;
+  mutable std::once_flag field_by_name_flag_;
+  mutable std::once_flag field_by_lowercase_name_flag_;
 };
 
 /// \brief A data type representing a list of values.

test/schema_test.cc

@@ -21,6 +21,7 @@
 
 #include <format>
 #include <memory>
+#include <thread>
 
 #include <gmock/gmock.h>
 #include <gtest/gtest.h>
@@ -490,3 +491,94 @@ TEST(SchemaTest, NestedDuplicateFieldIdError) {
   EXPECT_THAT(result.error().message,
               ::testing::HasSubstr("Duplicate field id found: 1"));
 }
+
+// Thread safety tests for Lazy Init
+class SchemaThreadSafetyTest : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    field1_ = std::make_unique<iceberg::SchemaField>(1, "id", iceberg::int32(), true);
+    field2_ = std::make_unique<iceberg::SchemaField>(2, "name", iceberg::string(), true);
+    field3_ = std::make_unique<iceberg::SchemaField>(3, "age", iceberg::int32(), true);
+    schema_ = std::make_unique<iceberg::Schema>(
+        std::vector<iceberg::SchemaField>{*field1_, *field2_, *field3_}, 100);
+  }
+
+  std::unique_ptr<iceberg::Schema> schema_;
+  std::unique_ptr<iceberg::SchemaField> field1_;
+  std::unique_ptr<iceberg::SchemaField> field2_;
+  std::unique_ptr<iceberg::SchemaField> field3_;
+};
+
+TEST_F(SchemaThreadSafetyTest, ConcurrentFindFieldById) {
+  const int num_threads = 10;
+  const int iterations_per_thread = 100;
+  std::vector<std::thread> threads;
+
+  for (int i = 0; i < num_threads; ++i) {
+    threads.emplace_back([this, iterations_per_thread]() {
+      for (int j = 0; j < iterations_per_thread; ++j) {
+        ASSERT_THAT(schema_->FindFieldById(1), ::testing::Optional(*field1_));
+        ASSERT_THAT(schema_->FindFieldById(999), ::testing::Optional(std::nullopt));
+      }
+    });
+  }
+
+  for (auto& thread : threads) {
+    thread.join();
+  }
+}
+
+TEST_F(SchemaThreadSafetyTest, MixedConcurrentOperations) {
+  const int num_threads = 8;
+  const int iterations_per_thread = 50;
+  std::vector<std::thread> threads;
+
+  for (int i = 0; i < num_threads; ++i) {
+    threads.emplace_back([this, iterations_per_thread, i]() {
+      for (int j = 0; j < iterations_per_thread; ++j) {
+        if (i % 4 == 0) {
+          ASSERT_THAT(schema_->FindFieldById(1), ::testing::Optional(*field1_));
+        } else if (i % 4 == 1) {
+          ASSERT_THAT(schema_->FindFieldByName("name", true),
+                      ::testing::Optional(*field2_));
+        } else if (i % 4 == 2) {
+          ASSERT_THAT(schema_->FindFieldByName("AGE", false),
+                      ::testing::Optional(*field3_));
+        } else {
+          ASSERT_THAT(schema_->FindFieldById(2), ::testing::Optional(*field2_));
+          ASSERT_THAT(schema_->FindFieldByName("id", true),
+                      ::testing::Optional(*field1_));
+          ASSERT_THAT(schema_->FindFieldByName("age", false),
+                      ::testing::Optional(*field3_));
+        }
+      }
+    });
+  }
+
+  for (auto& thread : threads) {
+    thread.join();
+  }
+}
+
+TEST_F(SchemaThreadSafetyTest, CopyAndConcurrentAccess) {
+  const int num_threads = 5;
+  const int iterations_per_thread = 20;
+  std::vector<std::thread> threads;
+  auto schema_copy = *schema_;
+
+  for (int i = 0; i < num_threads; ++i) {
+    threads.emplace_back([this, &schema_copy, iterations_per_thread, i]() {
+      for (int j = 0; j < iterations_per_thread; ++j) {
+        if (i % 2 == 0) {
+          ASSERT_THAT(schema_->FindFieldById(1), ::testing::Optional(*field1_));
+        } else {
+          ASSERT_THAT(schema_copy.FindFieldById(1), ::testing::Optional(*field1_));
+        }
+      }
+    });
+  }
+
+  for (auto& thread : threads) {
+    thread.join();
+  }
+}