feat(metadata): add format v4 table metadata foundations

src/iceberg/CMakeLists.txt

@@ -80,6 +80,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

@@ -98,6 +98,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

@@ -63,7 +63,11 @@ Status AppendSnapshot(ArrowRowBuilder& builder, const Snapshot& snapshot) {
     ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(3)));
   }
 
-  ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(4), snapshot.manifest_list));
+  if (!snapshot.manifest_list.empty()) {
+    ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(4), snapshot.manifest_list));
+  } else {
+    ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(4)));
+  }
 
   if (!snapshot.summary.empty()) {
     if (snapshot.summary.contains(SnapshotSummaryFields::kOperation)) {
@@ -77,6 +81,12 @@ Status AppendSnapshot(ArrowRowBuilder& builder, const Snapshot& snapshot) {
     ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(5)));
   }
 
+  if (snapshot.root_manifest.has_value()) {
+    ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(6), *snapshot.root_manifest));
+  } else {
+    ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(6)));
+  }
+
   return builder.FinishRow();
 }
 
@@ -163,7 +173,8 @@ const std::shared_ptr<Schema>& SnapshotsTable::schema() const {
       SchemaField::MakeOptional(6, "summary",
                                 std::make_shared<iceberg::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

@@ -59,6 +59,7 @@
 #include "iceberg/util/checked_cast.h"
 #include "iceberg/util/formatter.h"  // IWYU pragma: keep
 #include "iceberg/util/json_util_internal.h"
+#include "iceberg/util/location_util.h"
 #include "iceberg/util/macros.h"
 #include "iceberg/util/string_util.h"
 #include "iceberg/util/temporal_util.h"
@@ -114,6 +115,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 +741,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;
@@ -1130,7 +1136,16 @@ Result<std::unique_ptr<Snapshot>> SnapshotFromJson(const nlohmann::json& json) {
   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");
+  }
+  if (json.contains("constraint-statuses") && !json["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 +1163,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 +1181,38 @@ 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(snapshot->root_manifest.has_value() ? 4 : 1);
+      !status) {
+    return JsonParseError("Invalid snapshot: {}", status.error().message);
+  }
+  return snapshot;
 }
 
 nlohmann::json ToJson(const BlobMetadata& blob_metadata) {
@@ -1345,11 +1370,30 @@ 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) {
+    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;
+  if (table_metadata.format_version < TableMetadata::kMinFormatVersionOptionalLocation ||
+      !table_metadata.location.empty()) {
+    if (table_metadata.format_version >=
+            TableMetadata::kMinFormatVersionOptionalLocation &&
+        LocationUtil::ParseScheme(table_metadata.location).empty()) {
+      return InvalidArgument("V4 table location must be absolute: {}",
+                             table_metadata.location);
+    }
+    json[kLocation] = table_metadata.location;
+  }
   if (table_metadata.format_version > 1) {
     json[kLastSequenceNumber] = table_metadata.last_sequence_number;
   }
@@ -1375,6 +1419,9 @@ Result<nlohmann::json> ToJson(const TableMetadata& table_metadata) {
   nlohmann::json schemas_json = nlohmann::json::array();
   for (const auto& schema : table_metadata.schemas) {
     ICEBERG_ASSIGN_OR_RAISE(auto schema_json, ToJson(*schema));
+    if (table_metadata.format_version >= 4) {
+      schema_json.erase(kIdentifierFieldIds);
+    }
     schemas_json.push_back(std::move(schema_json));
   }
   json[kSchemas] = std::move(schemas_json);
@@ -1455,9 +1502,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 >= 4 && 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 +1651,37 @@ 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 >= 4) {
+    if (json.contains("constraints") && !json["constraints"].empty()) {
+      return NotSupported("V4 table constraints are not supported yet");
+    }
+    if (json.contains("last-constraint-id") && json["last-constraint-id"] != 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));
+    if (!table_metadata->location.empty() &&
+        LocationUtil::ParseScheme(table_metadata->location).empty()) {
+      return JsonParseError("V4 table location must be absolute: {}",
+                            table_metadata->location);
+    }
+  } else {
+    ICEBERG_ASSIGN_OR_RAISE(table_metadata->location,
+                            GetJsonValue<std::string>(json, kLocation));
+  }
 
   if (table_metadata->format_version > 1) {
     ICEBERG_ASSIGN_OR_RAISE(table_metadata->last_sequence_number,
@@ -1684,6 +1760,12 @@ Result<std::unique_ptr<TableMetadata>> TableMetadataFromJson(const nlohmann::jso
 
   ICEBERG_ASSIGN_OR_RAISE(table_metadata->snapshots,
                           FromJsonList<Snapshot>(json, kSnapshots, SnapshotFromJson));
+  for (const auto& snapshot : table_metadata->snapshots) {
+    if (auto status = snapshot->ValidateForFormatVersion(table_metadata->format_version);
+        !status) {
+      return JsonParseError("Invalid snapshot: {}", status.error().message);
+    }
+  }
   ICEBERG_ASSIGN_OR_RAISE(
       table_metadata->statistics,
       FromJsonList<StatisticsFile>(json, kStatistics, StatisticsFileFromJson));

src/iceberg/location_provider.cc

@@ -191,11 +191,21 @@ 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.
+  ICEBERG_PRECHECK(
+      !location.empty() ||
+          (!object_store &&
+           !LocationUtil::ParseScheme(properties.Get(TableProperties::kWriteDataLocation))
+                .empty()),
+      "Cannot create a location provider without a table location");
   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);