feat(metadata): add format v4 table metadata foundations

src/iceberg/CMakeLists.txt

@@ -84,6 +84,7 @@ set(ICEBERG_SOURCES
     manifest/v1_metadata.cc
     manifest/v2_metadata.cc
     manifest/v3_metadata.cc
+    manifest/v4_types.cc
     metadata_columns.cc
     metrics_config.cc
     metrics/commit_report.cc

src/iceberg/cpp20_compatibility_internal.h

@@ -102,6 +102,7 @@
 #include "iceberg/manifest/manifest_reader.h"
 #include "iceberg/manifest/manifest_writer.h"
 #include "iceberg/manifest/rolling_manifest_writer.h"
+#include "iceberg/manifest/v4_types.h"
 #include "iceberg/metadata_columns.h"
 #include "iceberg/metrics.h"
 #include "iceberg/metrics/commit_report.h"

src/iceberg/inspect/snapshots_table.cc

@@ -38,7 +38,7 @@
 namespace iceberg {
 namespace {
 
-constexpr std::array<internal::MetadataTableColumn<Snapshot>, 6> kSnapshotColumns = {
+constexpr std::array<internal::MetadataTableColumn<Snapshot>, 7> kSnapshotColumns = {
     {{.field_id = 1,
       .append =
           [](ArrowArray* column, const Snapshot& row) {
@@ -65,13 +65,19 @@ constexpr std::array<internal::MetadataTableColumn<Snapshot>, 6> kSnapshotColumn
             return row.manifest_list.empty() ? AppendNull(column)
                                              : AppendString(column, row.manifest_list);
           }},
-     {.field_id = 6, .append = [](ArrowArray* column, const Snapshot& row) {
-        if (row.summary.empty()) {
-          return AppendNull(column);
-        }
-        auto summary = row.summary;
-        summary.erase(SnapshotSummaryFields::kOperation);
-        return AppendStringMap(column, summary);
+     {.field_id = 6,
+      .append =
+          [](ArrowArray* column, const Snapshot& row) {
+            if (row.summary.empty()) {
+              return AppendNull(column);
+            }
+            auto summary = row.summary;
+            summary.erase(SnapshotSummaryFields::kOperation);
+            return AppendStringMap(column, summary);
+          }},
+     {.field_id = 9, .append = [](ArrowArray* column, const Snapshot& row) {
+        return row.root_manifest ? AppendString(column, *row.root_manifest)
+                                 : AppendNull(column);
       }}}};
 
 }  // namespace
@@ -96,7 +102,8 @@ const std::shared_ptr<Schema>& SnapshotsTable::CanonicalSchema() {
       SchemaField::MakeOptional(
           6, "summary",
           std::make_shared<MapType>(SchemaField::MakeRequired(7, "key", string()),
-                                    SchemaField::MakeRequired(8, "value", string())))});
+                                    SchemaField::MakeRequired(8, "value", string()))),
+      SchemaField::MakeOptional(9, "root_manifest", string())});
   return schema;
 }
 

src/iceberg/json_serde.cc

@@ -114,6 +114,7 @@ constexpr std::string_view kParentSnapshotId = "parent-snapshot-id";
 constexpr std::string_view kSequenceNumber = "sequence-number";
 constexpr std::string_view kTimestampMs = "timestamp-ms";
 constexpr std::string_view kManifestList = "manifest-list";
+constexpr std::string_view kRootManifest = "root-manifest";
 constexpr std::string_view kSummary = "summary";
 constexpr std::string_view kFirstRowId = "first-row-id";
 constexpr std::string_view kAddedRows = "added-rows";
@@ -739,7 +740,11 @@ nlohmann::json ToJson(const Snapshot& snapshot) {
     json[kSequenceNumber] = snapshot.sequence_number;
   }
   json[kTimestampMs] = UnixMsFromTimePointMs(snapshot.timestamp_ms);
-  json[kManifestList] = snapshot.manifest_list;
+  if (snapshot.root_manifest.has_value()) {
+    json[kRootManifest] = snapshot.root_manifest.value();
+  } else {
+    json[kManifestList] = snapshot.manifest_list;
+  }
   // If there is an operation, write the summary map
   if (snapshot.Operation().has_value()) {
     json[kSummary] = snapshot.summary;
@@ -1123,14 +1128,31 @@ Result<std::unique_ptr<SnapshotRef>> SnapshotRefFromJson(const nlohmann::json& j
   }
 }
 
-Result<std::unique_ptr<Snapshot>> SnapshotFromJson(const nlohmann::json& json) {
+Result<std::unique_ptr<Snapshot>> SnapshotFromJson(const nlohmann::json& json,
+                                                   int8_t format_version) {
+  if (format_version < 1 ||
+      format_version > TableMetadata::kSupportedTableFormatVersion) {
+    return JsonParseError("Unsupported snapshot format version: {}", format_version);
+  }
   ICEBERG_ASSIGN_OR_RAISE(auto snapshot_id, GetJsonValue<int64_t>(json, kSnapshotId));
   ICEBERG_ASSIGN_OR_RAISE(auto sequence_number,
                           GetJsonValueOptional<int64_t>(json, kSequenceNumber));
   ICEBERG_ASSIGN_OR_RAISE(auto unix_ms, GetJsonValue<int64_t>(json, kTimestampMs));
   auto timestamp_ms = TimePointMsFromUnixMs(unix_ms);
   ICEBERG_ASSIGN_OR_RAISE(auto manifest_list,
-                          GetJsonValue<std::string>(json, kManifestList));
+                          GetJsonValueOptional<std::string>(json, kManifestList));
+  ICEBERG_ASSIGN_OR_RAISE(auto root_manifest,
+                          GetJsonValueOptional<std::string>(json, kRootManifest));
+  if (manifest_list.has_value() == root_manifest.has_value()) {
+    return JsonParseError(
+        "Snapshot requires exactly one of manifest-list or root-manifest");
+  }
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto constraint_statuses,
+      GetJsonValueOptional<nlohmann::json::array_t>(json, "constraint-statuses"));
+  if (constraint_statuses.has_value() && !constraint_statuses->empty()) {
+    return NotSupported("V4 snapshot constraint statuses are not supported yet");
+  }
 
   ICEBERG_ASSIGN_OR_RAISE(auto parent_snapshot_id,
                           GetJsonValueOptional<int64_t>(json, kParentSnapshotId));
@@ -1148,14 +1170,15 @@ Result<std::unique_ptr<Snapshot>> SnapshotFromJson(const nlohmann::json& json) {
             "Invalid snapshot summary field value for '{}' (JSON value of type {})", key,
             value.type_name());
       }
-      if (key == SnapshotSummaryFields::kOperation &&
+      if (key == SnapshotSummaryFields::kOperation && !root_manifest.has_value() &&
           !kValidDataOperation.contains(value.get<std::string>())) {
         return JsonParseError("Invalid snapshot operation in snapshot summary");
       }
       summary[key] = value.get<std::string>();
     }
     // If summary is available but operation is missing, set operation to overwrite.
-    if (!summary.contains(SnapshotSummaryFields::kOperation)) {
+    if (!root_manifest.has_value() &&
+        !summary.contains(SnapshotSummaryFields::kOperation)) {
       summary[SnapshotSummaryFields::kOperation] = DataOperation::kOverwrite;
     }
   }
@@ -1165,29 +1188,36 @@ Result<std::unique_ptr<Snapshot>> SnapshotFromJson(const nlohmann::json& json) {
   ICEBERG_ASSIGN_OR_RAISE(auto added_rows,
                           GetJsonValueOptional<int64_t>(json, kAddedRows));
 
-  if (first_row_id.has_value() && first_row_id.value() < 0) {
-    return JsonParseError("Invalid first-row-id (cannot be negative): {}",
-                          first_row_id.value());
-  }
-  if (added_rows.has_value() && added_rows.value() < 0) {
-    return JsonParseError("Invalid added-rows (cannot be negative): {}",
-                          added_rows.value());
-  }
-  if (first_row_id.has_value() && !added_rows.has_value()) {
-    return JsonParseError("Invalid added-rows (required when first-row-id is set): null");
-  }
-  if (!first_row_id.has_value()) {
-    added_rows = std::nullopt;
+  if (!root_manifest.has_value()) {
+    if (first_row_id.has_value() && first_row_id.value() < 0) {
+      return JsonParseError("Invalid first-row-id (cannot be negative): {}",
+                            first_row_id.value());
+    }
+    if (added_rows.has_value() && added_rows.value() < 0) {
+      return JsonParseError("Invalid added-rows (cannot be negative): {}",
+                            added_rows.value());
+    }
+    if (first_row_id.has_value() && !added_rows.has_value()) {
+      return JsonParseError(
+          "Invalid added-rows (required when first-row-id is set): null");
+    }
+    if (!first_row_id.has_value()) {
+      added_rows = std::nullopt;
+    }
   }
 
   ICEBERG_ASSIGN_OR_RAISE(auto schema_id, GetJsonValueOptional<int32_t>(json, kSchemaId));
   ICEBERG_ASSIGN_OR_RAISE(auto key_id, GetJsonValueOptional<std::string>(json, kKeyId));
 
-  return std::make_unique<Snapshot>(
+  auto snapshot = std::make_unique<Snapshot>(
       snapshot_id, parent_snapshot_id,
       sequence_number.value_or(TableMetadata::kInitialSequenceNumber), timestamp_ms,
-      manifest_list, std::move(summary), schema_id, first_row_id, added_rows,
-      std::move(key_id));
+      manifest_list.value_or(""), std::move(summary), schema_id, first_row_id, added_rows,
+      std::move(key_id), std::move(root_manifest));
+  if (auto status = snapshot->ValidateForFormatVersion(format_version); !status) {
+    return JsonParseError("Invalid snapshot: {}", status.error().message);
+  }
+  return snapshot;
 }
 
 nlohmann::json ToJson(const BlobMetadata& blob_metadata) {
@@ -1345,11 +1375,26 @@ Result<EncryptedKey> EncryptedKeyFromJson(const nlohmann::json& json) {
 }
 
 Result<nlohmann::json> ToJson(const TableMetadata& table_metadata) {
+  ICEBERG_PRECHECK(
+      table_metadata.format_version >= 1 &&
+          table_metadata.format_version <= TableMetadata::kSupportedTableFormatVersion,
+      "Unsupported table format version: {}", table_metadata.format_version);
+  for (const auto& snapshot : table_metadata.snapshots) {
+    ICEBERG_PRECHECK(snapshot != nullptr, "Cannot write null snapshot");
+    if (auto status = snapshot->ValidateForFormatVersion(table_metadata.format_version);
+        !status) {
+      return InvalidArgument("Cannot write snapshot: {}", status.error().message);
+    }
+  }
   nlohmann::json json;
 
   json[kFormatVersion] = table_metadata.format_version;
   json[kTableUuid] = table_metadata.table_uuid;
-  json[kLocation] = table_metadata.location;
+  ICEBERG_RETURN_UNEXPECTED(table_metadata.ValidateLocation());
+  if (table_metadata.format_version < TableMetadata::kMinFormatVersionOptionalLocation ||
+      !table_metadata.location.empty()) {
+    json[kLocation] = table_metadata.location;
+  }
   if (table_metadata.format_version > 1) {
     json[kLastSequenceNumber] = table_metadata.last_sequence_number;
   }
@@ -1374,6 +1419,12 @@ Result<nlohmann::json> ToJson(const TableMetadata& table_metadata) {
   // infallible ToJson) cannot be used here; build the array with an explicit loop.
   nlohmann::json schemas_json = nlohmann::json::array();
   for (const auto& schema : table_metadata.schemas) {
+    if (table_metadata.format_version >= TableMetadata::kMinFormatVersionConstraints &&
+        !schema->IdentifierFieldIds().empty()) {
+      return NotSupported(
+          "V4 primary-key constraints are not supported yet; "
+          "identifier-field-ids cannot be written in v4 schemas");
+    }
     ICEBERG_ASSIGN_OR_RAISE(auto schema_json, ToJson(*schema));
     schemas_json.push_back(std::move(schema_json));
   }
@@ -1455,9 +1506,19 @@ Result<std::shared_ptr<Schema>> ParseSchemas(
 
     ICEBERG_ASSIGN_OR_RAISE(current_schema_id,
                             GetJsonInteger<int32_t>(json, kCurrentSchemaId));
+    const auto parse_schema =
+        [format_version](
+            const nlohmann::json& schema_json) -> Result<std::unique_ptr<Schema>> {
+      if (format_version >= TableMetadata::kMinFormatVersionConstraints &&
+          schema_json.is_object() && schema_json.contains(kIdentifierFieldIds)) {
+        auto v4_schema_json = schema_json;
+        v4_schema_json.erase(kIdentifierFieldIds);
+        return SchemaFromJson(v4_schema_json);
+      }
+      return SchemaFromJson(schema_json);
+    };
     for (const auto& schema_json : schema_array) {
-      ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<Schema> schema,
-                              SchemaFromJson(schema_json));
+      ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<Schema> schema, parse_schema(schema_json));
       ICEBERG_RETURN_UNEXPECTED(schema->Validate(format_version));
       if (schema->schema_id() == current_schema_id) {
         current_schema = schema;
@@ -1594,18 +1655,40 @@ Result<std::unique_ptr<TableMetadata>> TableMetadataFromJson(const nlohmann::jso
 
   auto table_metadata = std::make_unique<TableMetadata>();
 
-  ICEBERG_ASSIGN_OR_RAISE(table_metadata->format_version,
-                          GetJsonValue<int8_t>(json, kFormatVersion));
-  if (table_metadata->format_version < 1 ||
-      table_metadata->format_version > TableMetadata::kSupportedTableFormatVersion) {
-    return JsonParseError("Cannot read unsupported version: {}",
-                          table_metadata->format_version);
+  ICEBERG_ASSIGN_OR_RAISE(auto format_version,
+                          GetJsonInteger<int32_t>(json, kFormatVersion));
+  if (format_version < 1 ||
+      format_version > TableMetadata::kSupportedTableFormatVersion) {
+    return JsonParseError("Cannot read unsupported version: {}", format_version);
   }
+  table_metadata->format_version = static_cast<int8_t>(format_version);
 
   ICEBERG_ASSIGN_OR_RAISE(table_metadata->table_uuid,
                           GetJsonValueOrDefault<std::string>(json, kTableUuid));
-  ICEBERG_ASSIGN_OR_RAISE(table_metadata->location,
-                          GetJsonValue<std::string>(json, kLocation));
+  if (table_metadata->format_version >= TableMetadata::kMinFormatVersionConstraints) {
+    ICEBERG_ASSIGN_OR_RAISE(
+        auto constraints,
+        GetJsonValueOptional<nlohmann::json::array_t>(json, "constraints"));
+    if (constraints.has_value() && !constraints->empty()) {
+      return NotSupported("V4 table constraints are not supported yet");
+    }
+    ICEBERG_ASSIGN_OR_RAISE(auto last_constraint_id,
+                            GetJsonIntegerOptional<int32_t>(json, "last-constraint-id"));
+    if (last_constraint_id.value_or(0) != 0) {
+      return NotSupported("V4 table constraint IDs are not supported yet");
+    }
+  }
+  if (table_metadata->format_version >=
+      TableMetadata::kMinFormatVersionOptionalLocation) {
+    ICEBERG_ASSIGN_OR_RAISE(table_metadata->location,
+                            GetJsonValueOrDefault<std::string>(json, kLocation));
+  } else {
+    ICEBERG_ASSIGN_OR_RAISE(table_metadata->location,
+                            GetJsonValue<std::string>(json, kLocation));
+  }
+  if (auto status = table_metadata->ValidateLocation(); !status) {
+    return JsonParseError("{}", status.error().message);
+  }
 
   if (table_metadata->format_version > 1) {
     ICEBERG_ASSIGN_OR_RAISE(table_metadata->last_sequence_number,
@@ -1682,8 +1765,13 @@ Result<std::unique_ptr<TableMetadata>> TableMetadataFromJson(const nlohmann::jso
     });
   }
 
-  ICEBERG_ASSIGN_OR_RAISE(table_metadata->snapshots,
-                          FromJsonList<Snapshot>(json, kSnapshots, SnapshotFromJson));
+  ICEBERG_ASSIGN_OR_RAISE(
+      table_metadata->snapshots,
+      FromJsonList<Snapshot>(json, kSnapshots,
+                             [format_version = table_metadata->format_version](
+                                 const nlohmann::json& snapshot_json) {
+                               return SnapshotFromJson(snapshot_json, format_version);
+                             }));
   ICEBERG_ASSIGN_OR_RAISE(
       table_metadata->statistics,
       FromJsonList<StatisticsFile>(json, kStatistics, StatisticsFileFromJson));
@@ -2082,7 +2170,7 @@ Result<std::unique_ptr<TableUpdate>> TableUpdateFromJson(const nlohmann::json& j
   }
   if (action == kActionUpgradeFormatVersion) {
     ICEBERG_ASSIGN_OR_RAISE(auto format_version,
-                            GetJsonValue<int8_t>(json, kFormatVersion));
+                            GetJsonInteger<int8_t>(json, kFormatVersion));
     return std::make_unique<table::UpgradeFormatVersion>(format_version);
   }
   if (action == kActionAddSchema) {

src/iceberg/json_serde_internal.h

@@ -247,9 +247,12 @@ ICEBERG_EXPORT nlohmann::json ToJson(const Snapshot& snapshot);
 /// \brief Deserializes a JSON object into a `Snapshot` object.
 ///
 /// \param json The JSON representation of the snapshot.
+/// \param format_version The table format, or the latest supported version when
+/// parsing a standalone snapshot without table context.
 /// \return A `Snapshot` object or an error if the conversion fails.
 ICEBERG_EXPORT Result<std::unique_ptr<Snapshot>> SnapshotFromJson(
-    const nlohmann::json& json);
+    const nlohmann::json& json,
+    int8_t format_version = TableMetadata::kSupportedTableFormatVersion);
 
 /// \brief Serializes a `StatisticsFile` object to JSON.
 ///

src/iceberg/location_provider.cc

@@ -191,11 +191,24 @@ Result<std::string> ObjectStoreLocationProvider::NewDataLocation(
 
 Result<std::unique_ptr<LocationProvider>> LocationProvider::Make(
     std::string_view location, const TableProperties& properties) {
+  const bool object_store = properties.Get(TableProperties::kObjectStoreEnabled);
+  // Validate once before either provider can derive a data path or table context.
+  // An explicit absolute URI is sufficient only for the default provider; object
+  // stores also need the table location to derive their context.
+  if (location.empty()) {
+    ICEBERG_PRECHECK(!object_store,
+                     "Object-store location provider requires a table location");
+    ICEBERG_PRECHECK(
+        !LocationUtil::ParseScheme(properties.Get(TableProperties::kWriteDataLocation))
+             .empty(),
+        "Default location provider requires a table location or an absolute URI in '{}'",
+        TableProperties::kWriteDataLocation.key());
+  }
   location = LocationUtil::StripTrailingSlash(location);
 
   // TODO(xxx): create location provider specified by "write.location-provider.impl"
 
-  if (properties.Get(TableProperties::kObjectStoreEnabled)) {
+  if (object_store) {
     return std::make_unique<ObjectStoreLocationProvider>(location, properties);
   } else {
     return std::make_unique<DefaultLocationProvider>(location, properties);