feat: add update sort order

src/iceberg/CMakeLists.txt

@@ -77,6 +77,7 @@ set(ICEBERG_SOURCES
     transform_function.cc
     type.cc
     update/pending_update.cc
+    update/update_sort_order.cc
     update/update_properties.cc
     util/bucket_util.cc
     util/conversions.cc

src/iceberg/expression/term.h

@@ -151,6 +151,8 @@ class ICEBERG_EXPORT BoundReference
 
   std::shared_ptr<Type> type() const override { return field_.type(); }
 
+  int32_t field_id() const { return field_.field_id(); }
+
   bool MayProduceNull() const override { return field_.optional(); }
 
   bool Equals(const BoundTerm& other) const override;

src/iceberg/meson.build

@@ -100,6 +100,7 @@ iceberg_sources = files(
     'type.cc',
     'update/pending_update.cc',
     'update/update_properties.cc',
+    'update/update_sort_order.cc',
     'util/bucket_util.cc',
     'util/conversions.cc',
     'util/decimal.cc',

src/iceberg/sort_order.cc

@@ -111,7 +111,7 @@ Result<std::unique_ptr<SortOrder>> SortOrder::Make(const Schema& schema, int32_t
   }
 
   if (fields.empty() && sort_id != kUnsortedOrderId) [[unlikely]] {
-    return InvalidArgument("Sort order must have at least one sort field");
+    return InvalidArgument("Sort order must have at least one sort field.");
   }
 
   auto sort_order = std::unique_ptr<SortOrder>(new SortOrder(sort_id, std::move(fields)));

src/iceberg/table.cc

@@ -154,6 +154,13 @@ Result<std::shared_ptr<UpdateProperties>> Table::NewUpdateProperties() {
   return transaction->NewUpdateProperties();
 }
 
+Result<std::shared_ptr<UpdateSortOrder>> Table::NewUpdateSortOrder() {
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto transaction, Transaction::Make(shared_from_this(), Transaction::Kind::kUpdate,
+                                          /*auto_commit=*/true));
+  return transaction->NewUpdateSortOrder();
+}
+
 Result<std::shared_ptr<StagedTable>> StagedTable::Make(
     TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
     std::string metadata_location, std::shared_ptr<FileIO> io,

src/iceberg/table.h

@@ -132,6 +132,10 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this<Table> {
   /// changes.
   virtual Result<std::shared_ptr<UpdateProperties>> NewUpdateProperties();
 
+  /// \brief Create a new UpdateSortOrder to update the table sort order and commit the
+  /// changes.
+  virtual Result<std::shared_ptr<UpdateSortOrder>> NewUpdateSortOrder();
+
  protected:
   Table(TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
         std::string metadata_location, std::shared_ptr<FileIO> io,

src/iceberg/table_metadata.cc

@@ -29,6 +29,7 @@
 #include <ranges>
 #include <string>
 #include <unordered_map>
+#include <unordered_set>
 #include <utility>
 
 #include <nlohmann/json.hpp>
@@ -426,7 +427,7 @@ class TableMetadataBuilder::Impl {
   Status SetDefaultSortOrder(int32_t order_id);
   Result<int32_t> AddSortOrder(const SortOrder& order);
   Status SetProperties(const std::unordered_map<std::string, std::string>& updated);
-  Status RemoveProperties(const std::vector<std::string>& removed);
+  Status RemoveProperties(const std::unordered_set<std::string>& removed);
 
   std::unique_ptr<TableMetadata> Build();
 
@@ -590,7 +591,7 @@ Status TableMetadataBuilder::Impl::SetProperties(
 }
 
 Status TableMetadataBuilder::Impl::RemoveProperties(
-    const std::vector<std::string>& removed) {
+    const std::unordered_set<std::string>& removed) {
   // If removed is empty, return early (no-op)
   if (removed.empty()) {
     return {};
@@ -820,7 +821,7 @@ TableMetadataBuilder& TableMetadataBuilder::SetProperties(
 }
 
 TableMetadataBuilder& TableMetadataBuilder::RemoveProperties(
-    const std::vector<std::string>& removed) {
+    const std::unordered_set<std::string>& removed) {
   ICEBERG_BUILDER_RETURN_IF_ERROR(impl_->RemoveProperties(removed));
   return *this;
 }

src/iceberg/table_metadata.h

@@ -394,7 +394,7 @@ class ICEBERG_EXPORT TableMetadataBuilder : public ErrorCollector {
   ///
   /// \param removed Set of property keys to remove
   /// \return Reference to this builder for method chaining
-  TableMetadataBuilder& RemoveProperties(const std::vector<std::string>& removed);
+  TableMetadataBuilder& RemoveProperties(const std::unordered_set<std::string>& removed);
 
   /// \brief Set the table location
   ///

src/iceberg/table_update.h

@@ -26,6 +26,7 @@
 #include <optional>
 #include <string>
 #include <unordered_map>
+#include <unordered_set>
 #include <vector>
 
 #include "iceberg/iceberg_export.h"
@@ -379,10 +380,10 @@ class ICEBERG_EXPORT SetProperties : public TableUpdate {
 /// \brief Represents removing table properties
 class ICEBERG_EXPORT RemoveProperties : public TableUpdate {
  public:
-  explicit RemoveProperties(std::vector<std::string> removed)
+  explicit RemoveProperties(std::unordered_set<std::string> removed)
       : removed_(std::move(removed)) {}
 
-  const std::vector<std::string>& removed() const { return removed_; }
+  const std::unordered_set<std::string>& removed() const { return removed_; }
 
   void ApplyTo(TableMetadataBuilder& builder) const override;
 
@@ -391,7 +392,7 @@ class ICEBERG_EXPORT RemoveProperties : public TableUpdate {
   Kind kind() const override { return Kind::kRemoveProperties; }
 
  private:
-  std::vector<std::string> removed_;
+  std::unordered_set<std::string> removed_;
 };
 
 /// \brief Represents setting the table location

src/iceberg/test/CMakeLists.txt

@@ -154,7 +154,8 @@ if(ICEBERG_BUILD_BUNDLE)
                    USE_BUNDLE
                    SOURCES
                    transaction_test.cc
-                   update_properties_test.cc)
+                   update_properties_test.cc
+                   update_sort_order_test.cc)
 
 endif()
 

src/iceberg/test/table_requirements_test.cc

@@ -939,7 +939,7 @@ TEST(TableRequirementsTest, RemoveProperties) {
   std::vector<std::unique_ptr<TableUpdate>> updates;
 
   updates.push_back(
-      std::make_unique<table::RemoveProperties>(std::vector<std::string>{"test"}));
+      std::make_unique<table::RemoveProperties>(std::unordered_set<std::string>{"test"}));
 
   auto result = TableRequirements::ForUpdateTable(*metadata, updates);
   ASSERT_THAT(result, IsOk());

src/iceberg/test/table_update_test.cc

@@ -175,7 +175,7 @@ INSTANTIATE_TEST_SUITE_P(
             .update_factory =
                 [] {
                   return std::make_unique<table::RemoveProperties>(
-                      std::vector<std::string>{"key"});
+                      std::unordered_set<std::string>{"key"});
                 },
             .expected_existing_table_count = 0,
             .validator = nullptr},

src/iceberg/test/transaction_test.cc

@@ -19,11 +19,14 @@
 
 #include "iceberg/transaction.h"
 
-#include "iceberg/table.h"
-#include "iceberg/table_update.h"
+#include "iceberg/expression/expressions.h"
+#include "iceberg/expression/term.h"
+#include "iceberg/sort_order.h"
 #include "iceberg/test/matchers.h"
 #include "iceberg/test/update_test_base.h"
+#include "iceberg/transform.h"
 #include "iceberg/update/update_properties.h"
+#include "iceberg/update/update_sort_order.h"
 
 namespace iceberg {
 
@@ -35,14 +38,6 @@ TEST_F(TransactionTest, CreateTransaction) {
   EXPECT_EQ(txn->table(), table_);
 }
 
-TEST_F(TransactionTest, UpdatePropertiesInTransaction) {
-  ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction());
-  ICEBERG_UNWRAP_OR_FAIL(auto update, txn->NewUpdateProperties());
-
-  update->Set("key1", "value1");
-  EXPECT_THAT(update->Apply(), IsOk());
-}
-
 TEST_F(TransactionTest, CommitEmptyTransaction) {
   ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction());
   EXPECT_THAT(txn->Commit(), IsOk());
@@ -67,24 +62,36 @@ TEST_F(TransactionTest, CommitTransactionWithPropertyUpdate) {
 TEST_F(TransactionTest, MultipleUpdatesInTransaction) {
   ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction());
 
-  // First update
+  // First update: set property
   ICEBERG_UNWRAP_OR_FAIL(auto update1, txn->NewUpdateProperties());
-  update1->Set("key1", "value1");
+  update1->Set("key1", "value1").Set("key2", "value2");
   EXPECT_THAT(update1->Commit(), IsOk());
 
-  // Second update
-  ICEBERG_UNWRAP_OR_FAIL(auto update2, txn->NewUpdateProperties());
-  update2->Set("key2", "value2");
+  // Second update: update sort order
+  ICEBERG_UNWRAP_OR_FAIL(auto update2, txn->NewUpdateSortOrder());
+  auto term =
+      UnboundTransform::Make(Expressions::Ref("x"), Transform::Identity()).value();
+  update2->AddSortField(std::move(term), SortDirection::kAscending, NullOrder::kFirst);
   EXPECT_THAT(update2->Commit(), IsOk());
 
   // Commit transaction
   ICEBERG_UNWRAP_OR_FAIL(auto updated_table, txn->Commit());
 
-  // Verify both properties were set
+  // Verify properties were set
   ICEBERG_UNWRAP_OR_FAIL(auto reloaded, catalog_->LoadTable(table_ident_));
   const auto& props = reloaded->properties().configs();
   EXPECT_EQ(props.at("key1"), "value1");
   EXPECT_EQ(props.at("key2"), "value2");
+
+  // Verify sort order was updated
+  ICEBERG_UNWRAP_OR_FAIL(auto sort_order, reloaded->sort_order());
+  std::vector<SortField> expected_fields;
+  expected_fields.emplace_back(1, Transform::Identity(), SortDirection::kAscending,
+                               NullOrder::kFirst);
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto expected_sort_order,
+      SortOrder::Make(sort_order->order_id(), std::move(expected_fields)));
+  EXPECT_EQ(*sort_order, *expected_sort_order);
 }
 
 }  // namespace iceberg

src/iceberg/test/update_properties_test.cc

@@ -19,21 +19,28 @@
 
 #include "iceberg/update/update_properties.h"
 
-#include "iceberg/table_update.h"
 #include "iceberg/test/matchers.h"
 #include "iceberg/test/update_test_base.h"
 
 namespace iceberg {
 
 class UpdatePropertiesTest : public UpdateTestBase {};
 
+TEST_F(UpdatePropertiesTest, EmptyUpdate) {
+  ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewUpdateProperties());
+  ICEBERG_UNWRAP_OR_FAIL(auto result, update->Apply());
+  EXPECT_THAT(result.updates.empty(), true);
+}
+
 TEST_F(UpdatePropertiesTest, SetProperty) {
   ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewUpdateProperties());
   update->Set("key1", "value1").Set("key2", "value2");
 
   ICEBERG_UNWRAP_OR_FAIL(auto result, update->Apply());
-  EXPECT_EQ(result.updates.size(), 1);
-  EXPECT_EQ(result.updates[0]->kind(), table::SetProperties::Kind::kSetProperties);
+  EXPECT_EQ(result.updates.size(), 2);
+  EXPECT_EQ(result.updates.at("key1"), "value1");
+  EXPECT_EQ(result.updates.at("key2"), "value2");
+  EXPECT_TRUE(result.removals.empty());
 }
 
 TEST_F(UpdatePropertiesTest, RemoveProperty) {
@@ -48,8 +55,10 @@ TEST_F(UpdatePropertiesTest, RemoveProperty) {
   update->Remove("key1").Remove("key2");
 
   ICEBERG_UNWRAP_OR_FAIL(auto result, update->Apply());
-  EXPECT_EQ(result.updates.size(), 1);
-  EXPECT_EQ(result.updates[0]->kind(), table::RemoveProperties::Kind::kRemoveProperties);
+  EXPECT_TRUE(result.updates.empty());
+  EXPECT_EQ(result.removals.size(), 2);
+  EXPECT_TRUE(result.removals.contains("key1"));
+  EXPECT_TRUE(result.removals.contains("key2"));
 }
 
 TEST_F(UpdatePropertiesTest, SetThenRemoveSameKey) {
@@ -73,22 +82,23 @@ TEST_F(UpdatePropertiesTest, RemoveThenSetSameKey) {
 TEST_F(UpdatePropertiesTest, SetAndRemoveDifferentKeys) {
   ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewUpdateProperties());
   update->Set("key1", "value1").Remove("key2");
-  EXPECT_THAT(update->Commit(), IsOk());
 
-  ICEBERG_UNWRAP_OR_FAIL(auto reloaded, catalog_->LoadTable(table_ident_));
-  const auto& props = reloaded->properties().configs();
-  EXPECT_EQ(props.at("key1"), "value1");
-  EXPECT_FALSE(props.contains("key2"));
+  ICEBERG_UNWRAP_OR_FAIL(auto result, update->Apply());
+  EXPECT_EQ(result.updates.size(), 1);
+  EXPECT_EQ(result.updates.at("key1"), "value1");
+  EXPECT_EQ(result.removals.size(), 1);
+  EXPECT_TRUE(result.removals.contains("key2"));
 }
 
 TEST_F(UpdatePropertiesTest, UpgradeFormatVersionValid) {
   ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewUpdateProperties());
-  update->Set("format-version", "2");
+  update->Set("format-version", "3");
 
   ICEBERG_UNWRAP_OR_FAIL(auto result, update->Apply());
-  EXPECT_EQ(result.updates.size(), 1);
-  EXPECT_EQ(result.updates[0]->kind(),
-            table::UpgradeFormatVersion::Kind::kUpgradeFormatVersion);
+  EXPECT_TRUE(result.updates.empty());
+  EXPECT_TRUE(result.removals.empty());
+  ASSERT_TRUE(result.format_version.has_value());
+  EXPECT_EQ(result.format_version.value(), 3);
 }
 
 TEST_F(UpdatePropertiesTest, UpgradeFormatVersionInvalidString) {