feat: add schema update to table metadata builder

src/iceberg/partition_spec.cc

@@ -131,6 +131,8 @@ bool PartitionSpec::Equals(const PartitionSpec& other) const {
 }
 
 Status PartitionSpec::Validate(const Schema& schema, bool allow_missing_fields) const {
+  ICEBERG_RETURN_UNEXPECTED(ValidatePartitionName(schema, *this));
+
   std::unordered_map<int32_t, int32_t> parents = IndexParents(schema);
   for (const auto& partition_field : fields_) {
     ICEBERG_ASSIGN_OR_RAISE(auto source_field,
@@ -177,6 +179,43 @@ Status PartitionSpec::Validate(const Schema& schema, bool allow_missing_fields)
   return {};
 }
 
+Status PartitionSpec::ValidatePartitionName(const Schema& schema,
+                                            const PartitionSpec& spec) {
+  std::unordered_set<std::string> partition_names;
+  for (const auto& partition_field : spec.fields()) {
+    auto name = std::string(partition_field.name());
+    ICEBERG_PRECHECK(!name.empty(), "Cannot use empty partition name: {}", name);
+
+    if (partition_names.contains(name)) {
+      return InvalidArgument("Cannot use partition name more than once: {}", name);
+    }
+    partition_names.insert(name);
+
+    ICEBERG_ASSIGN_OR_RAISE(auto schema_field, schema.FindFieldByName(name));
+    auto transform_type = partition_field.transform()->transform_type();
+    if (transform_type == TransformType::kIdentity ||
+        transform_type == TransformType::kVoid) {
+      // for identity/nulls transform case we allow conflicts between partition and schema
+      // field name as long as they are sourced from the same schema field
+      if (schema_field.has_value() &&
+          schema_field.value().get().field_id() != partition_field.source_id()) {
+        return InvalidArgument(
+            "Cannot create identity partition sourced from different field in schema: {}",
+            name);
+      }
+    } else {
+      // for all other transforms we don't allow conflicts between partition name and
+      // schema field name
+      if (schema_field.has_value()) {
+        return InvalidArgument(
+            "Cannot create partition from name that exists in schema: {}", name);
+      }
+    }
+  }
+
+  return {};
+}
+
 Result<std::vector<std::reference_wrapper<const PartitionField>>>
 PartitionSpec::GetFieldsBySourceId(int32_t source_id) const {
   ICEBERG_ASSIGN_OR_RAISE(auto source_id_to_fields, source_id_to_fields_.Get(*this));

src/iceberg/partition_spec.h

@@ -79,11 +79,15 @@ class ICEBERG_EXPORT PartitionSpec : public util::Formattable {
 
   /// \brief Validates the partition spec against a schema.
   /// \param schema The schema to validate against.
-  /// \param allowMissingFields Whether to skip validation for partition fields whose
+  /// \param allow_missing_fields Whether to skip validation for partition fields whose
   /// source columns have been dropped from the schema.
   /// \return Error status if the partition spec is invalid.
   Status Validate(const Schema& schema, bool allow_missing_fields) const;
 
+  // \brief Validates the partition field names are unique within the partition spec and
+  // schema.
+  static Status ValidatePartitionName(const Schema& schema, const PartitionSpec& spec);
+
   /// \brief Get the partition fields by source ID.
   /// \param source_id The id of the source field.
   /// \return The partition fields by source ID, or NotFound if the source field is not

src/iceberg/schema.cc

@@ -25,6 +25,7 @@
 #include "iceberg/result.h"
 #include "iceberg/row/struct_like.h"
 #include "iceberg/schema_internal.h"
+#include "iceberg/table_metadata.h"
 #include "iceberg/type.h"
 #include "iceberg/util/formatter.h"  // IWYU pragma: keep
 #include "iceberg/util/macros.h"
@@ -147,6 +148,20 @@ Result<std::unordered_map<int32_t, std::vector<size_t>>> Schema::InitIdToPositio
   return visitor.Finish();
 }
 
+Result<int32_t> Schema::InitHighestFieldId(const Schema& self) {
+  ICEBERG_ASSIGN_OR_RAISE(auto id_to_field, self.id_to_field_.Get(self));
+
+  if (id_to_field.get().empty()) {
+    return kInitialColumnId;
+  }
+
+  auto max_it = std::ranges::max_element(
+      id_to_field.get(),
+      [](const auto& lhs, const auto& rhs) { return lhs.first < rhs.first; });
+
+  return max_it->first;
+}
+
 Result<std::unique_ptr<StructLikeAccessor>> Schema::GetAccessorById(
     int32_t field_id) const {
   ICEBERG_ASSIGN_OR_RAISE(auto id_to_position_path, id_to_position_path_.Get(*this));
@@ -228,4 +243,33 @@ Result<std::vector<std::string>> Schema::IdentifierFieldNames() const {
   return names;
 }
 
+Result<int32_t> Schema::HighestFieldId() const { return highest_field_id_.Get(*this); }
+
+bool Schema::SameSchema(const Schema& other) const {
+  return fields_ == other.fields_ && identifier_field_ids_ == other.identifier_field_ids_;
+}
+
+Status Schema::Validate(int32_t format_version) const {
+  // Get all fields including nested ones
+  ICEBERG_ASSIGN_OR_RAISE(auto id_to_field, id_to_field_.Get(*this));
+
+  // Check each field's type and defaults
+  for (const auto& [field_id, field_ref] : id_to_field.get()) {
+    const auto& field = field_ref.get();
+
+    // Check if the field's type requires a minimum format version
+    if (auto it = TableMetadata::kMinFormatVersions.find(field.type()->type_id());
+        it != TableMetadata::kMinFormatVersions.end()) {
+      if (int32_t min_format_version = it->second; format_version < min_format_version) {
+        return InvalidSchema("Invalid type for {}: {} is not supported until v{}",
+                             field.name(), *field.type(), min_format_version);
+      }
+    }
+
+    // TODO(GuoTao.yu): Check default values when they are supported
+  }
+
+  return {};
+}
+
 }  // namespace iceberg

src/iceberg/schema.h

@@ -46,6 +46,7 @@ namespace iceberg {
 class ICEBERG_EXPORT Schema : public StructType {
  public:
   static constexpr int32_t kInitialSchemaId = 0;
+  static constexpr int32_t kInitialColumnId = 0;
   static constexpr int32_t kInvalidColumnId = -1;
 
   explicit Schema(std::vector<SchemaField> fields,
@@ -127,6 +128,23 @@ class ICEBERG_EXPORT Schema : public StructType {
   /// \brief Return the canonical field names of the identifier fields.
   Result<std::vector<std::string>> IdentifierFieldNames() const;
 
+  /// \brief Get the highest field ID in the schema.
+  /// \return The highest field ID.
+  Result<int32_t> HighestFieldId() const;
+
+  /// \brief Checks whether this schema is equivalent to another schema while ignoring the
+  /// schema id.
+  bool SameSchema(const Schema& other) const;
+
+  /// \brief Validate the schema for a given format version.
+  ///
+  /// This validates that the schema does not contain types that were released in later
+  /// format versions.
+  ///
+  /// \param format_version The format version to validate against.
+  /// \return Error status if the schema is invalid.
+  Status Validate(int32_t format_version) const;
+
   friend bool operator==(const Schema& lhs, const Schema& rhs) { return lhs.Equals(rhs); }
 
  private:
@@ -155,6 +173,7 @@ class ICEBERG_EXPORT Schema : public StructType {
   InitLowerCaseNameToIdMap(const Schema&);
   static Result<std::unordered_map<int32_t, std::vector<size_t>>> InitIdToPositionPath(
       const Schema&);
+  static Result<int32_t> InitHighestFieldId(const Schema&);
 
   const std::optional<int32_t> schema_id_;
   /// Field IDs that uniquely identify rows in the table.
@@ -167,6 +186,8 @@ class ICEBERG_EXPORT Schema : public StructType {
   Lazy<InitLowerCaseNameToIdMap> lowercase_name_to_id_;
   /// Mapping from field id to (nested) position path to access the field.
   Lazy<InitIdToPositionPath> id_to_position_path_;
+  /// Highest field ID in the schema.
+  Lazy<InitHighestFieldId> highest_field_id_;
 };
 
 }  // namespace iceberg