use result for schema/spec/order

wgtmac · wgtmac · commit 37a8c3000016 · 2025-04-20T15:51:33.000+08:00
diff --git a/src/iceberg/constants.h b/src/iceberg/constants.h
diff --git a/src/iceberg/json_internal.cc b/src/iceberg/json_internal.cc
@@ -27,9 +27,9 @@
 #include <type_traits>
 #include <unordered_set>
 
+#include <iceberg/table.h>
 #include <nlohmann/json.hpp>
 
-#include "iceberg/constants.h"
 #include "iceberg/partition_field.h"
 #include "iceberg/partition_spec.h"
 #include "iceberg/result.h"
@@ -499,7 +499,7 @@ nlohmann::json ToJson(const Snapshot& snapshot) {
   nlohmann::json json;
   json[kSnapshotId] = snapshot.snapshot_id;
   SetOptionalField(json, kParentSnapshotId, snapshot.parent_snapshot_id);
-  if (snapshot.sequence_number > kInitialSequenceNumber) {
+  if (snapshot.sequence_number > TableMetadata::kInitialSequenceNumber) {
     json[kSequenceNumber] = snapshot.sequence_number;
   }
   json[kTimestampMs] = snapshot.timestamp_ms;
@@ -666,8 +666,8 @@ Result<std::unique_ptr<PartitionField>> PartitionFieldFromJson(
   int32_t field_id;
   if (allow_field_id_missing) {
     // Partition field id in v1 is not tracked, so we use -1 to indicate that.
-    ICEBERG_ASSIGN_OR_RAISE(
-        field_id, GetJsonValueOrDefault<int32_t>(json, kFieldId, kInvalidFieldId));
+    ICEBERG_ASSIGN_OR_RAISE(field_id, GetJsonValueOrDefault<int32_t>(
+                                          json, kFieldId, SchemaField::kInvalidFieldId));
   } else {
     ICEBERG_ASSIGN_OR_RAISE(field_id, GetJsonValue<int32_t>(json, kFieldId));
   }
@@ -765,8 +765,9 @@ Result<std::unique_ptr<Snapshot>> SnapshotFromJson(const nlohmann::json& json) {
   ICEBERG_ASSIGN_OR_RAISE(auto schema_id, GetJsonValueOptional<int32_t>(json, kSchemaId));
 
   return std::make_unique<Snapshot>(
-      snapshot_id, parent_snapshot_id, sequence_number.value_or(kInitialSequenceNumber),
-      timestamp_ms, manifest_list, std::move(summary), schema_id);
+      snapshot_id, parent_snapshot_id,
+      sequence_number.value_or(TableMetadata::kInitialSequenceNumber), timestamp_ms,
+      manifest_list, std::move(summary), schema_id);
 }
 
 nlohmann::json ToJson(const BlobMetadata& blob_metadata) {
@@ -1054,16 +1055,16 @@ Status ParsePartitionSpecs(const nlohmann::json& json, int8_t format_version,
     for (const auto& entry_json : partition_spec_json) {
       ICEBERG_ASSIGN_OR_RAISE(auto field, PartitionFieldFromJson(entry_json));
       int32_t field_id = field->field_id();
-      if (field_id == kInvalidFieldId) {
+      if (field_id == SchemaField::kInvalidFieldId) {
         // If the field ID is not set, we need to assign a new one
         field_id = next_partition_field_id++;
       }
       fields.emplace_back(field->source_id(), field_id, std::string(field->name()),
                           std::move(field->transform()));
     }
 
-    auto spec = std::make_unique<PartitionSpec>(current_schema, kInitialSpecId,
-                                                std::move(fields));
+    auto spec = std::make_unique<PartitionSpec>(
+        current_schema, PartitionSpec::kInitialSpecId, std::move(fields));
     default_spec_id = spec->spec_id();
     partition_specs.push_back(std::move(spec));
   }
@@ -1112,7 +1113,7 @@ Result<std::unique_ptr<TableMetadata>> TableMetadataFromJson(const nlohmann::jso
   ICEBERG_ASSIGN_OR_RAISE(table_metadata->format_version,
                           GetJsonValue<int8_t>(json, kFormatVersion));
   if (table_metadata->format_version < 1 ||
-      table_metadata->format_version > kSupportedTableFormatVersion) {
+      table_metadata->format_version > TableMetadata::kSupportedTableFormatVersion) {
     return JsonParseError("Cannot read unsupported version: {}",
                           table_metadata->format_version);
   }
@@ -1126,7 +1127,7 @@ Result<std::unique_ptr<TableMetadata>> TableMetadataFromJson(const nlohmann::jso
     ICEBERG_ASSIGN_OR_RAISE(table_metadata->last_sequence_number,
                             GetJsonValue<int64_t>(json, kLastSequenceNumber));
   } else {
-    table_metadata->last_sequence_number = kInitialSequenceNumber;
+    table_metadata->last_sequence_number = TableMetadata::kInitialSequenceNumber;
   }
   ICEBERG_ASSIGN_OR_RAISE(table_metadata->last_column_id,
                           GetJsonValue<int32_t>(json, kLastColumnId));
@@ -1169,15 +1170,15 @@ Result<std::unique_ptr<TableMetadata>> TableMetadataFromJson(const nlohmann::jso
   }
 
   // This field is optional, but internally we set this to -1 when not set
-  ICEBERG_ASSIGN_OR_RAISE(
-      table_metadata->current_snapshot_id,
-      GetJsonValueOrDefault<int64_t>(json, kCurrentSnapshotId, kInvalidSnapshotId));
+  ICEBERG_ASSIGN_OR_RAISE(table_metadata->current_snapshot_id,
+                          GetJsonValueOrDefault<int64_t>(json, kCurrentSnapshotId,
+                                                         Snapshot::kInvalidSnapshotId));
 
   if (table_metadata->format_version >= 3) {
     ICEBERG_ASSIGN_OR_RAISE(table_metadata->next_row_id,
                             GetJsonValue<int64_t>(json, kNextRowId));
   } else {
-    table_metadata->next_row_id = kInitialRowId;
+    table_metadata->next_row_id = TableMetadata::kInitialRowId;
   }
 
   ICEBERG_ASSIGN_OR_RAISE(auto last_updated_ms,
@@ -1189,7 +1190,7 @@ Result<std::unique_ptr<TableMetadata>> TableMetadataFromJson(const nlohmann::jso
     ICEBERG_ASSIGN_OR_RAISE(
         table_metadata->refs,
         FromJsonMap<std::shared_ptr<SnapshotRef>>(json, kRefs, SnapshotRefFromJson));
-  } else if (table_metadata->current_snapshot_id != kInvalidSnapshotId) {
+  } else if (table_metadata->current_snapshot_id != Snapshot::kInvalidSnapshotId) {
     table_metadata->refs["main"] = std::make_unique<SnapshotRef>(SnapshotRef{
         .snapshot_id = table_metadata->current_snapshot_id,
         .retention = SnapshotRef::Branch{},
diff --git a/src/iceberg/partition_spec.cc b/src/iceberg/partition_spec.cc
@@ -23,7 +23,6 @@
 #include <format>
 #include <ranges>
 
-#include "iceberg/constants.h"
 #include "iceberg/schema.h"
 #include "iceberg/util/formatter.h"  // IWYU pragma: keep
 
diff --git a/src/iceberg/partition_spec.h b/src/iceberg/partition_spec.h
@@ -41,9 +41,10 @@ namespace iceberg {
 /// evolution.
 class ICEBERG_EXPORT PartitionSpec : public util::Formattable {
  public:
+  static constexpr int32_t kInitialSpecId = 0;
   /// \brief The start ID for partition field.  It is only used to generate
   /// partition field id for v1 metadata where it is tracked.
-  constexpr static int32_t kLegacyPartitionDataIdStart = 1000;
+  static constexpr int32_t kLegacyPartitionDataIdStart = 1000;
 
   /// \brief Create a new partition spec.
   ///
diff --git a/src/iceberg/result.h b/src/iceberg/result.h
@@ -42,6 +42,7 @@ enum class ErrorKind {
   kNotSupported,
   kInvalidExpression,
   kJsonParseError,
+  kNotFound,
 };
 
 /// \brief Error with a kind and a message.
diff --git a/src/iceberg/schema.h b/src/iceberg/schema.h
@@ -41,6 +41,8 @@ namespace iceberg {
 /// evolution.
 class ICEBERG_EXPORT Schema : public StructType {
  public:
+  static constexpr int32_t kInitialSchemaId = 0;
+
   Schema(std::vector<SchemaField> fields, std::optional<int32_t> schema_id);
 
   /// \brief Get the schema ID.
diff --git a/src/iceberg/schema_field.h b/src/iceberg/schema_field.h
@@ -37,6 +37,8 @@ namespace iceberg {
 /// \brief A type combined with a name.
 class ICEBERG_EXPORT SchemaField : public iceberg::util::Formattable {
  public:
+  static constexpr int32_t kInvalidFieldId = -1;
+
   /// \brief Construct a field.
   /// \param[in] field_id The field ID.
   /// \param[in] name The field name.
diff --git a/src/iceberg/snapshot.h b/src/iceberg/snapshot.h
@@ -233,6 +233,8 @@ struct ICEBERG_EXPORT DataOperation {
 ///
 /// Snapshots are created by table operations.
 struct ICEBERG_EXPORT Snapshot {
+  static constexpr int64_t kInvalidSnapshotId = -1;
+
   /// A unqiue long ID.
   int64_t snapshot_id;
   /// The snapshot ID of the snapshot's parent. Omitted for any snapshot with no parent.
diff --git a/src/iceberg/sort_order.cc b/src/iceberg/sort_order.cc
@@ -21,7 +21,6 @@
 
 #include <format>
 
-#include "iceberg/constants.h"
 #include "iceberg/util/formatter.h"  // IWYU pragma: keep
 
 namespace iceberg {
diff --git a/src/iceberg/sort_order.h b/src/iceberg/sort_order.h
@@ -36,6 +36,8 @@ namespace iceberg {
 /// applied to the data.
 class ICEBERG_EXPORT SortOrder : public util::Formattable {
  public:
+  static constexpr int32_t kInitialSortOrderId = 1;
+
   SortOrder(int32_t order_id, std::vector<SortField> fields);
 
   /// \brief Get an unsorted sort order singleton.
diff --git a/src/iceberg/table_metadata.cc b/src/iceberg/table_metadata.cc
@@ -23,6 +23,7 @@
 #include <ranges>
 #include <string>
 
+#include "iceberg/expected.h"
 #include "iceberg/partition_spec.h"
 #include "iceberg/schema.h"
 #include "iceberg/sort_order.h"
@@ -38,31 +39,46 @@ std::string ToString(const MetadataLogEntry& entry) {
                      entry.metadata_file);
 }
 
-const std::shared_ptr<Schema>& TableMetadata::Schema() const {
-  static const std::shared_ptr<::iceberg::Schema> empty_schema = nullptr;
-
+Result<std::reference_wrapper<const std::shared_ptr<Schema>>> TableMetadata::Schema()
+    const {
   auto iter = std::ranges::find_if(schemas, [this](const auto& schema) {
     return schema->schema_id() == current_schema_id;
   });
-  return iter == schemas.end() ? empty_schema : *iter;
+  if (iter == schemas.end()) {
+    return unexpected<Error>({
+        .kind = ErrorKind::kNotFound,
+        .message = std::format("Current schema is not found"),
+    });
+  }
+  return std::cref(*iter);
 }
 
-const std::shared_ptr<PartitionSpec>& TableMetadata::PartitionSpec() const {
-  static const std::shared_ptr<iceberg::PartitionSpec> empty_spec = nullptr;
-
+Result<std::reference_wrapper<const std::shared_ptr<PartitionSpec>>>
+TableMetadata::PartitionSpec() const {
   auto iter = std::ranges::find_if(partition_specs, [this](const auto& spec) {
     return spec->spec_id() == default_spec_id;
   });
-  return iter == partition_specs.end() ? empty_spec : *iter;
+  if (iter == partition_specs.end()) {
+    return unexpected<Error>({
+        .kind = ErrorKind::kNotFound,
+        .message = std::format("Default partition spec is not found"),
+    });
+  }
+  return std::cref(*iter);
 }
 
-const std::shared_ptr<SortOrder>& TableMetadata::SortOrder() const {
-  static const std::shared_ptr<iceberg::SortOrder> empty_order = nullptr;
-
+Result<std::reference_wrapper<const std::shared_ptr<SortOrder>>>
+TableMetadata::SortOrder() const {
   auto iter = std::ranges::find_if(sort_orders, [this](const auto& order) {
     return order->order_id() == default_sort_order_id;
   });
-  return iter == sort_orders.end() ? empty_order : *iter;
+  if (iter == sort_orders.end()) {
+    return unexpected<Error>({
+        .kind = ErrorKind::kNotFound,
+        .message = std::format("Default sort order is not found"),
+    });
+  }
+  return std::cref(*iter);
 }
 
 Result<TimePointMs> TimePointMsFromUnixMs(int64_t unix_ms) {
diff --git a/src/iceberg/table_metadata.h b/src/iceberg/table_metadata.h
@@ -83,6 +83,13 @@ struct ICEBERG_EXPORT MetadataLogEntry {
 /// Schema|PartitionSpec|SortOrder> 2) List<MetadataUpdate> 3) Map<Long, Snapshot> 4)
 /// Map<String, SnapshotRef>
 struct ICEBERG_EXPORT TableMetadata {
+  static constexpr int8_t kDefaultTableFormatVersion = 2;
+  static constexpr int8_t kSupportedTableFormatVersion = 3;
+  static constexpr int8_t kMinFormatVersionRowLineage = 3;
+  static constexpr int64_t kInitialSequenceNumber = 0;
+  static constexpr int64_t kInvalidSequenceNumber = -1;
+  static constexpr int64_t kInitialRowId = 0;
+
   /// An integer version number for the format
   int8_t format_version;
   /// A UUID that identifies the table
@@ -130,12 +137,13 @@ struct ICEBERG_EXPORT TableMetadata {
   /// A `long` higher than all assigned row IDs
   int64_t next_row_id;
 
-  /// \brief Get the current schema
-  const std::shared_ptr<Schema>& Schema() const;
-  /// \brief Get the current partition spec
-  const std::shared_ptr<PartitionSpec>& PartitionSpec() const;
-  /// \brief Get the current sort order
-  const std::shared_ptr<SortOrder>& SortOrder() const;
+  /// \brief Get the current schema, return NotFoundError if not found
+  Result<std::reference_wrapper<const std::shared_ptr<Schema>>> Schema() const;
+  /// \brief Get the current partition spec, return NotFoundError if not found
+  Result<std::reference_wrapper<const std::shared_ptr<PartitionSpec>>> PartitionSpec()
+      const;
+  /// \brief Get the current sort order, return NotFoundError if not found
+  Result<std::reference_wrapper<const std::shared_ptr<SortOrder>>> SortOrder() const;
 };
 
 /// \brief Returns a string representation of a SnapshotLogEntry
diff --git a/test/metadata_serde_test.cc b/test/metadata_serde_test.cc
@@ -100,8 +100,8 @@ TEST_F(MetadataSerdeTest, DeserializeV1Valid) {
   auto expected_schema =
       std::make_shared<Schema>(schema_fields, /*schema_id=*/std::nullopt);
   auto schema = metadata->Schema();
-  ASSERT_NE(schema, nullptr);
-  EXPECT_EQ(*schema, *expected_schema);
+  ASSERT_TRUE(schema.has_value());
+  EXPECT_EQ(*(schema.value().get()), *expected_schema);
 
   // Compare partition spec
   std::vector<PartitionField> partition_fields;
@@ -110,8 +110,8 @@ TEST_F(MetadataSerdeTest, DeserializeV1Valid) {
   auto expected_spec =
       std::make_shared<PartitionSpec>(expected_schema, /*spec_id=*/0, partition_fields);
   auto partition_spec = metadata->PartitionSpec();
-  ASSERT_NE(partition_spec, nullptr);
-  EXPECT_EQ(*partition_spec, *expected_spec);
+  ASSERT_TRUE(partition_spec.has_value());
+  EXPECT_EQ(*(partition_spec.value().get()), *expected_spec);
 }
 
 TEST_F(MetadataSerdeTest, DeserializeV2Valid) {
@@ -136,8 +136,8 @@ TEST_F(MetadataSerdeTest, DeserializeV2Valid) {
   auto expected_schema =
       std::make_shared<Schema>(std::move(schema_fields), /*schema_id=*/1);
   auto schema = metadata->Schema();
-  ASSERT_NE(schema, nullptr);
-  EXPECT_EQ(*schema, *expected_schema);
+  ASSERT_TRUE(schema.has_value());
+  EXPECT_EQ(*(schema.value().get()), *expected_schema);
 
   // Compare partition spec
   EXPECT_EQ(metadata->default_spec_id, 0);
@@ -147,8 +147,8 @@ TEST_F(MetadataSerdeTest, DeserializeV2Valid) {
   auto expected_spec = std::make_shared<PartitionSpec>(expected_schema, /*spec_id=*/0,
                                                        std::move(partition_fields));
   auto partition_spec = metadata->PartitionSpec();
-  ASSERT_NE(partition_spec, nullptr);
-  EXPECT_EQ(*partition_spec, *expected_spec);
+  ASSERT_TRUE(partition_spec.has_value());
+  EXPECT_EQ(*(partition_spec.value().get()), *expected_spec);
 
   // Compare sort order
   EXPECT_EQ(metadata->default_sort_order_id, 3);
@@ -160,8 +160,8 @@ TEST_F(MetadataSerdeTest, DeserializeV2Valid) {
   auto expected_sort_order =
       std::make_shared<SortOrder>(/*order_id=*/3, std::move(sort_fields));
   auto sort_order = metadata->SortOrder();
-  ASSERT_NE(sort_order, nullptr);
-  EXPECT_EQ(*sort_order, *expected_sort_order);
+  ASSERT_TRUE(sort_order.has_value());
+  EXPECT_EQ(*(sort_order.value().get()), *expected_sort_order);
 
   EXPECT_EQ(metadata->current_snapshot_id, 3055729675574597004);
 
