fix review

Author: HeartLinked

src/iceberg/json_internal.cc

@@ -519,7 +519,8 @@ Result<std::unique_ptr<PartitionField>> PartitionFieldFromJson(
                                           std::move(transform));
 }
 
-Result<std::unique_ptr<PartitionSpec>> PartitionSpecFromJson(const nlohmann::json& json) {
+Result<std::unique_ptr<PartitionSpec>> PartitionSpecFromJson(
+    const std::shared_ptr<Schema>& schema, const nlohmann::json& json) {
   ICEBERG_ASSIGN_OR_RAISE(auto spec_id, GetJsonValue<int32_t>(json, kSpecId));
   ICEBERG_ASSIGN_OR_RAISE(auto fields, GetJsonValue<nlohmann::json>(json, kFields));
 
@@ -528,6 +529,8 @@ Result<std::unique_ptr<PartitionSpec>> PartitionSpecFromJson(const nlohmann::jso
     ICEBERG_ASSIGN_OR_RAISE(auto partition_field, PartitionFieldFromJson(field_json));
     partition_fields.push_back(std::move(*partition_field));
   }
+  // TODO(Li Feiyang):use a new PartitionSpec::Make to find the source field of each
+  // partition field from schema and then verify it
   return std::make_unique<PartitionSpec>(spec_id, std::move(partition_fields));
 }
 
@@ -855,6 +858,7 @@ Result<std::shared_ptr<Schema>> ParseSchemas(
 /// \param[out] default_spec_id The default partition spec ID.
 /// \param[out] partition_specs The list of partition specs.
 Status ParsePartitionSpecs(const nlohmann::json& json, int8_t format_version,
+                           const std::shared_ptr<Schema>& current_schema,
                            int32_t& default_spec_id,
                            std::vector<std::shared_ptr<PartitionSpec>>& partition_specs) {
   if (json.contains(kPartitionSpecs)) {
@@ -867,7 +871,8 @@ Status ParsePartitionSpecs(const nlohmann::json& json, int8_t format_version,
     ICEBERG_ASSIGN_OR_RAISE(default_spec_id, GetJsonValue<int32_t>(json, kDefaultSpecId));
 
     for (const auto& spec_json : spec_array) {
-      ICEBERG_ASSIGN_OR_RAISE(auto spec, PartitionSpecFromJson(spec_json));
+      ICEBERG_ASSIGN_OR_RAISE(auto spec,
+                              PartitionSpecFromJson(current_schema, spec_json));
       partition_specs.push_back(std::move(spec));
     }
   } else {
@@ -973,9 +978,9 @@ Result<std::unique_ptr<TableMetadata>> TableMetadataFromJson(const nlohmann::jso
       ParseSchemas(json, table_metadata->format_version,
                    table_metadata->current_schema_id, table_metadata->schemas));
 
-  ICEBERG_RETURN_UNEXPECTED(ParsePartitionSpecs(json, table_metadata->format_version,
-                                                table_metadata->default_spec_id,
-                                                table_metadata->partition_specs));
+  ICEBERG_RETURN_UNEXPECTED(ParsePartitionSpecs(
+      json, table_metadata->format_version, current_schema,
+      table_metadata->default_spec_id, table_metadata->partition_specs));
 
   if (json.contains(kLastPartitionId)) {
     ICEBERG_ASSIGN_OR_RAISE(table_metadata->last_partition_id,

src/iceberg/json_internal.h

@@ -173,11 +173,12 @@ ICEBERG_EXPORT Result<std::string> ToJsonString(const PartitionSpec& partition_s
 /// objects. Each `PartitionField` will be parsed using the `PartitionFieldFromJson`
 /// function.
 ///
+/// \param schema The table schema.
 /// \param json The JSON object representing a `PartitionSpec`.
 /// \return An `expected` value containing either a `PartitionSpec` object or an error. If
 /// the JSON is malformed or missing expected fields, an error will be returned.
 ICEBERG_EXPORT Result<std::unique_ptr<PartitionSpec>> PartitionSpecFromJson(
-    const nlohmann::json& json);
+    const std::shared_ptr<Schema>& schema, const nlohmann::json& json);
 
 /// \brief Serializes a `SnapshotRef` object to JSON.
 ///

src/iceberg/manifest_adapter.cc

@@ -153,9 +153,9 @@ Result<std::shared_ptr<StructType>> ManifestEntryAdapter::GetManifestEntryType()
     return ManifestEntry::TypeFromPartitionType(nullptr);
   }
   std::shared_ptr<StructType> partition_type = nullptr;
-  if (table_schema_ != nullptr) {
+  if (current_schema_ != nullptr) {
     ICEBERG_ASSIGN_OR_RAISE(partition_type,
-                            partition_spec_->PartitionType(table_schema_));
+                            partition_spec_->PartitionType(*current_schema_));
   }
   return ManifestEntry::TypeFromPartitionType(std::move(partition_type));
 }

src/iceberg/manifest_adapter.h

@@ -62,9 +62,9 @@ class ICEBERG_EXPORT ManifestAdapter {
 class ICEBERG_EXPORT ManifestEntryAdapter : public ManifestAdapter {
  public:
   ManifestEntryAdapter(std::shared_ptr<PartitionSpec> partition_spec,
-                       std::shared_ptr<Schema> table_schema)
+                       std::shared_ptr<Schema> current_schema_)
       : partition_spec_(std::move(partition_spec)),
-        table_schema_(std::move(table_schema)) {}
+        current_schema_(std::move(current_schema_)) {}
   ~ManifestEntryAdapter() override;
 
   virtual Status Append(const ManifestEntry& entry) = 0;
@@ -98,7 +98,7 @@ class ICEBERG_EXPORT ManifestEntryAdapter : public ManifestAdapter {
 
  protected:
   std::shared_ptr<PartitionSpec> partition_spec_;
-  std::shared_ptr<Schema> table_schema_;
+  std::shared_ptr<Schema> current_schema_;
   std::shared_ptr<Schema> manifest_schema_;
 };
 

src/iceberg/manifest_writer.cc

@@ -69,9 +69,9 @@ Result<std::unique_ptr<Writer>> OpenFileWriter(
 Result<std::unique_ptr<ManifestWriter>> ManifestWriter::MakeV1Writer(
     std::optional<int64_t> snapshot_id, std::string_view manifest_location,
     std::shared_ptr<FileIO> file_io, std::shared_ptr<PartitionSpec> partition_spec,
-    std::shared_ptr<Schema> table_schema) {
+    std::shared_ptr<Schema> current_schema_) {
   auto adapter = std::make_unique<ManifestEntryAdapterV1>(
-      snapshot_id, std::move(partition_spec), std::move(table_schema));
+      snapshot_id, std::move(partition_spec), std::move(current_schema_));
   ICEBERG_RETURN_UNEXPECTED(adapter->Init());
   ICEBERG_RETURN_UNEXPECTED(adapter->StartAppending());
 

src/iceberg/partition_spec.cc

@@ -57,23 +57,16 @@ int32_t PartitionSpec::spec_id() const { return spec_id_; }
 
 std::span<const PartitionField> PartitionSpec::fields() const { return fields_; }
 
-Result<std::shared_ptr<StructType>> PartitionSpec::PartitionType(
-    std::shared_ptr<Schema> schema) {
+Result<std::unique_ptr<StructType>> PartitionSpec::PartitionType(const Schema& schema) {
   if (fields_.empty()) {
     return nullptr;
   }
-  {
-    std::scoped_lock<std::mutex> lock(mutex_);
-    if (partition_type_ != nullptr) {
-      return partition_type_;
-    }
-  }
 
   std::vector<SchemaField> partition_fields;
   for (const auto& partition_field : fields_) {
     // Get the source field from the original schema by source_id
     ICEBERG_ASSIGN_OR_RAISE(auto source_field,
-                            schema->FindFieldById(partition_field.source_id()));
+                            schema.FindFieldById(partition_field.source_id()));
     if (!source_field.has_value()) {
       // TODO(xiao.dong) when source field is missing,
       // should return an error or just use UNKNOWN type
@@ -95,11 +88,7 @@ Result<std::shared_ptr<StructType>> PartitionSpec::PartitionType(
                                   /*optional=*/true);
   }
 
-  std::scoped_lock<std::mutex> lock(mutex_);
-  if (partition_type_ == nullptr) {
-    partition_type_ = std::make_shared<StructType>(std::move(partition_fields));
-  }
-  return partition_type_;
+  return std::make_unique<StructType>(std::move(partition_fields));
 }
 
 std::string PartitionSpec::ToString() const {

src/iceberg/partition_spec.h

@@ -71,7 +71,7 @@ class ICEBERG_EXPORT PartitionSpec : public util::Formattable {
   std::span<const PartitionField> fields() const;
 
   /// \brief Get the partition type.
-  Result<std::shared_ptr<StructType>> PartitionType(std::shared_ptr<Schema> schema);
+  Result<std::unique_ptr<StructType>> PartitionType(const Schema&);
 
   std::string ToString() const override;
 
@@ -88,10 +88,6 @@ class ICEBERG_EXPORT PartitionSpec : public util::Formattable {
   const int32_t spec_id_;
   std::vector<PartitionField> fields_;
   int32_t last_assigned_field_id_;
-
-  // FIXME: use similar lazy initialization pattern as in StructType
-  std::mutex mutex_;
-  std::shared_ptr<StructType> partition_type_;
 };
 
 }  // namespace iceberg

src/iceberg/table_scan.cc

@@ -31,6 +31,7 @@
 #include "iceberg/schema_field.h"
 #include "iceberg/snapshot.h"
 #include "iceberg/table_metadata.h"
+#include "iceberg/type.h"
 #include "iceberg/util/macros.h"
 
 namespace iceberg {
@@ -272,14 +273,15 @@ Result<std::vector<std::shared_ptr<FileScanTask>>> DataTableScan::PlanFiles() co
   // auto partition_schema = context_.table_metadata->Schema().value();
 
   // Get the table schema and partition type
-  ICEBERG_ASSIGN_OR_RAISE(auto table_schema, context_.table_metadata->Schema());
+  ICEBERG_ASSIGN_OR_RAISE(auto current_schema, context_.table_metadata->Schema());
   ICEBERG_ASSIGN_OR_RAISE(auto partition_type,
-                          partition_spec->PartitionType(table_schema));
+                          partition_spec->PartitionType(*current_schema));
+  auto partition_schema = std::shared_ptr<StructType>(std::move(partition_type));
 
   for (const auto& manifest_file : manifest_files) {
     ICEBERG_ASSIGN_OR_RAISE(
         auto manifest_reader,
-        ManifestReader::Make(manifest_file, file_io_, partition_type));
+        ManifestReader::Make(manifest_file, file_io_, partition_schema));
     ICEBERG_ASSIGN_OR_RAISE(auto manifests, manifest_reader->Entries());
 
     // TODO(gty404): filter manifests using partition spec and filter expression

src/iceberg/test/json_internal_test.cc

@@ -140,6 +140,10 @@ TEST(JsonPartitionTest, PartitionFieldFromJsonMissingField) {
 }
 
 TEST(JsonPartitionTest, PartitionSpec) {
+  auto schema = std::make_shared<Schema>(
+      std::vector<SchemaField>{SchemaField(3, "region", iceberg::string(), false),
+                               SchemaField(5, "ts", iceberg::int64(), false)},
+      /*schema_id=*/100);
   auto identity_transform = Transform::Identity();
   PartitionSpec spec(1, {PartitionField(3, 101, "region", identity_transform),
                          PartitionField(5, 102, "ts", identity_transform)});
@@ -157,7 +161,7 @@ TEST(JsonPartitionTest, PartitionSpec) {
 
   EXPECT_EQ(json, expected_json);
 
-  auto parsed_spec_result = PartitionSpecFromJson(json);
+  auto parsed_spec_result = PartitionSpecFromJson(schema, json);
   ASSERT_TRUE(parsed_spec_result.has_value()) << parsed_spec_result.error().message;
   EXPECT_EQ(spec, *parsed_spec_result.value());
 }

src/iceberg/test/partition_spec_test.cc

@@ -88,8 +88,7 @@ TEST(PartitionSpecTest, Equality) {
 TEST(PartitionSpecTest, PartitionSchemaTest) {
   SchemaField field1(5, "ts", iceberg::timestamp(), true);
   SchemaField field2(7, "bar", iceberg::string(), true);
-  auto const schema =
-      std::make_shared<Schema>(std::vector<SchemaField>{field1, field2}, 100);
+  Schema schema({field1, field2}, 100);
   auto identity_transform = Transform::Identity();
   PartitionField pt_field1(5, 1000, "day", identity_transform);
   PartitionField pt_field2(7, 1001, "hour", identity_transform);
@@ -136,10 +135,10 @@ TEST(PartitionSpecTest, PartitionTypeTest) {
       std::vector<SchemaField>{field1, field2, field3, field4, field5, field6},
       Schema::kInitialSchemaId);
 
-  auto parsed_spec_result = PartitionSpecFromJson(json);
+  auto parsed_spec_result = PartitionSpecFromJson(schema, json);
   ASSERT_TRUE(parsed_spec_result.has_value()) << parsed_spec_result.error().message;
 
-  auto partition_schema = parsed_spec_result.value()->PartitionType(schema);
+  auto partition_schema = parsed_spec_result.value()->PartitionType(*schema);
 
   SchemaField pt_field1(1000, "ts_day", date(), true);
   SchemaField pt_field2(1001, "id_bucket", int32(), true);