diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt
index cf465e5fd..a0d839e48 100644
--- a/src/iceberg/CMakeLists.txt
+++ b/src/iceberg/CMakeLists.txt
@@ -82,6 +82,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
diff --git a/src/iceberg/cpp20_compatibility_internal.h b/src/iceberg/cpp20_compatibility_internal.h
index d186e5bb0..5ab354efa 100644
--- a/src/iceberg/cpp20_compatibility_internal.h
+++ b/src/iceberg/cpp20_compatibility_internal.h
@@ -100,6 +100,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"
diff --git a/src/iceberg/inspect/snapshots_table.cc b/src/iceberg/inspect/snapshots_table.cc
index 0dede12c3..9743a6a86 100644
--- a/src/iceberg/inspect/snapshots_table.cc
+++ b/src/iceberg/inspect/snapshots_table.cc
@@ -71,9 +71,19 @@ Status AppendOperation(ArrowArray* column, const Snapshot& snapshot) {
 }
 
 Status AppendManifestList(ArrowArray* column, const Snapshot& snapshot) {
+  if (snapshot.root_manifest.has_value()) {
+    return AppendNull(column);
+  }
   return AppendString(column, snapshot.manifest_list);
 }
 
+Status AppendRootManifest(ArrowArray* column, const Snapshot& snapshot) {
+  if (snapshot.root_manifest.has_value()) {
+    return AppendString(column, *snapshot.root_manifest);
+  }
+  return AppendNull(column);
+}
+
 Status AppendSummary(ArrowArray* column, const Snapshot& snapshot) {
   if (snapshot.summary.empty()) {
     return AppendNull(column);
@@ -86,13 +96,14 @@ Status AppendSummary(ArrowArray* column, const Snapshot& snapshot) {
   return AppendStringMap(column, snapshot.summary);
 }
 
-constexpr std::array<internal::MetadataTableColumn<Snapshot>, 6> kSnapshotColumns{{
+constexpr std::array<internal::MetadataTableColumn<Snapshot>, 7> kSnapshotColumns{{
     {.field_id = 1, .append = AppendCommittedAt},
     {.field_id = 2, .append = AppendSnapshotId},
     {.field_id = 3, .append = AppendParentId},
     {.field_id = 4, .append = AppendOperation},
     {.field_id = 5, .append = AppendManifestList},
     {.field_id = 6, .append = AppendSummary},
+    {.field_id = 9, .append = AppendRootManifest},
 }};
 
 class SnapshotsTableStream {
@@ -185,7 +196,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;
 }
 
diff --git a/src/iceberg/json_serde.cc b/src/iceberg/json_serde.cc
index 37f6c08c9..042c839e9 100644
--- a/src/iceberg/json_serde.cc
+++ b/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,19 @@ 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");
+  }
+  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 +1166,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 +1184,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 +1373,31 @@ 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;
+  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 +1423,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 +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 >= 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 +1655,45 @@ 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")) {
+      ICEBERG_ASSIGN_OR_RAISE(auto constraints,
+                              GetJsonValue<nlohmann::json::array_t>(json, "constraints"));
+      if (!constraints.empty()) {
+        return NotSupported("V4 table constraints are not supported yet");
+      }
+    }
+    if (json.contains("last-constraint-id")) {
+      ICEBERG_ASSIGN_OR_RAISE(auto last_constraint_id,
+                              GetJsonInteger<int32_t>(json, "last-constraint-id"));
+      if (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 +1772,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));
@@ -2082,7 +2176,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) {
diff --git a/src/iceberg/location_provider.cc b/src/iceberg/location_provider.cc
index 70274edc6..c8b317cc9 100644
--- a/src/iceberg/location_provider.cc
+++ b/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);
diff --git a/src/iceberg/manifest/v4_types.cc b/src/iceberg/manifest/v4_types.cc
new file mode 100644
index 000000000..679ec775f
--- /dev/null
+++ b/src/iceberg/manifest/v4_types.cc
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/manifest/v4_types.h"
+
+#include "iceberg/schema_field.h"
+#include "iceberg/type.h"
+
+namespace iceberg {
+
+Result<FileTrackingStatus> FileTrackingStatusFromInt(int32_t status) {
+  switch (status) {
+    case 0:
+      return FileTrackingStatus::kExisting;
+    case 1:
+      return FileTrackingStatus::kAdded;
+    case 2:
+      return FileTrackingStatus::kDeleted;
+    case 3:
+      return FileTrackingStatus::kReplaced;
+    case 4:
+      return FileTrackingStatus::kModified;
+    default:
+      return InvalidArgument("Invalid file tracking status: {}", status);
+  }
+}
+
+std::shared_ptr<StructType> FileTracking::Type() {
+  static const auto kInstance = std::make_shared<StructType>(std::vector<SchemaField>{
+      SchemaField::MakeRequired(0, "status", int32()),
+      SchemaField::MakeOptional(1, "snapshot_id", int64()),
+      SchemaField::MakeOptional(5, "modified_snapshot_id", int64()),
+      SchemaField::MakeOptional(3, "sequence_number", int64()),
+      SchemaField::MakeOptional(4, "file_sequence_number", int64()),
+      SchemaField::MakeOptional(142, "first_row_id", int64()),
+      SchemaField::MakeOptional(6, "deleted_positions", binary()),
+      SchemaField::MakeOptional(7, "replaced_positions", binary())});
+  return kInstance;
+}
+
+std::shared_ptr<StructType> DeletionVectorMetadata::Type() {
+  static const auto kInstance = std::make_shared<StructType>(
+      std::vector<SchemaField>{SchemaField::MakeRequired(155, "location", string()),
+                               SchemaField::MakeRequired(144, "offset", int64()),
+                               SchemaField::MakeRequired(145, "size_in_bytes", int64()),
+                               SchemaField::MakeRequired(156, "cardinality", int64()),
+                               SchemaField::MakeOptional(149, "key_metadata", binary())});
+  return kInstance;
+}
+
+std::shared_ptr<StructType> ManifestInfo::Type() {
+  static const auto kInstance = std::make_shared<StructType>(std::vector<SchemaField>{
+      SchemaField::MakeRequired(521, "format_version", int32()),
+      SchemaField::MakeOptional(522, "dv", binary()),
+      SchemaField::MakeRequired(504, "added_files_count", int32()),
+      SchemaField::MakeRequired(505, "existing_files_count", int32()),
+      SchemaField::MakeRequired(525, "modified_files_count", int32()),
+      SchemaField::MakeRequired(506, "deleted_files_count", int32()),
+      SchemaField::MakeRequired(523, "replaced_files_count", int32()),
+      SchemaField::MakeRequired(512, "added_rows_count", int64()),
+      SchemaField::MakeRequired(513, "existing_rows_count", int64()),
+      SchemaField::MakeRequired(526, "modified_rows_count", int64()),
+      SchemaField::MakeRequired(514, "deleted_rows_count", int64()),
+      SchemaField::MakeRequired(524, "replaced_rows_count", int64()),
+      SchemaField::MakeRequired(516, "min_sequence_number", int64())});
+  return kInstance;
+}
+
+}  // namespace iceberg
diff --git a/src/iceberg/manifest/v4_types.h b/src/iceberg/manifest/v4_types.h
new file mode 100644
index 000000000..b44e5d14a
--- /dev/null
+++ b/src/iceberg/manifest/v4_types.h
@@ -0,0 +1,132 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#pragma once
+
+/// \file iceberg/manifest/v4_types.h
+/// Foundational metadata types for the v4 manifest tree. These types do not enable
+/// v4 table reads or writes. The v4 specification is still under development.
+/// See https://iceberg.apache.org/spec/#manifest-schema.
+
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <string>
+#include <vector>
+
+#include "iceberg/iceberg_export.h"
+#include "iceberg/result.h"
+#include "iceberg/type_fwd.h"
+
+namespace iceberg {
+
+/// \brief Status of a tracked file in a v4 manifest.
+/// Kept separate from ManifestStatus because v1-v3 manifests cannot contain
+/// REPLACED or MODIFIED entries.
+enum class FileTrackingStatus {
+  kExisting = 0,
+  kAdded = 1,
+  kDeleted = 2,
+  kReplaced = 3,
+  kModified = 4,
+};
+
+ICEBERG_EXPORT Result<FileTrackingStatus> FileTrackingStatusFromInt(int32_t status);
+
+/// \brief The v4 tracking struct, shared by data files and manifests.
+struct ICEBERG_EXPORT FileTracking {
+  /// Field id: 0. ADDED, EXISTING, and MODIFIED entries are live.
+  FileTrackingStatus status = FileTrackingStatus::kAdded;
+  /// Field id: 1. Snapshot where the file was added, replaced, or deleted.
+  /// Inherited when null; writers should persist a non-null value.
+  std::optional<int64_t> snapshot_id;
+  /// Field id: 5. Snapshot where the file was last modified.
+  std::optional<int64_t> modified_snapshot_id;
+  /// Field id: 3. Data sequence number; inherited when null.
+  std::optional<int64_t> sequence_number;
+  /// Field id: 4. File sequence number; inherited when null.
+  std::optional<int64_t> file_sequence_number;
+  /// Field id: 142. Base row ID; subject to first row ID inheritance.
+  std::optional<int64_t> first_row_id;
+  /// Field id: 6. Manifest-only bitmap of positions deleted in modified_snapshot_id.
+  std::optional<std::vector<uint8_t>> deleted_positions;
+  /// Field id: 7. Manifest-only bitmap of positions replaced in modified_snapshot_id.
+  std::optional<std::vector<uint8_t>> replaced_positions;
+
+  constexpr bool IsAlive() const {
+    return status == FileTrackingStatus::kExisting ||
+           status == FileTrackingStatus::kAdded ||
+           status == FileTrackingStatus::kModified;
+  }
+
+  static std::shared_ptr<StructType> Type();
+};
+
+/// \brief The v4 deletion_vector struct, colocated with a tracked data file.
+/// This references a Puffin blob; it does not hold the bitmap itself.
+struct ICEBERG_EXPORT DeletionVectorMetadata {
+  /// Field id: 155. Location of the file containing the deletion vector.
+  std::string location;
+  /// Field id: 144. Must match the blob offset in the Puffin footer.
+  int64_t offset = 0;
+  /// Field id: 145. Must match the blob length in the Puffin footer.
+  int64_t size_in_bytes = 0;
+  /// Field id: 156. Number of deleted rows (set bits).
+  int64_t cardinality = 0;
+  /// Field id: 149. Encryption key metadata, absent for unencrypted blobs.
+  std::optional<std::vector<uint8_t>> key_metadata;
+
+  static std::shared_ptr<StructType> Type();
+};
+
+/// \brief The v4 manifest_info struct, present only for tracked manifests.
+/// File counts include all five tracking statuses and sum to the manifest's
+/// record_count. This data carrier does not validate tracked-file requirements.
+struct ICEBERG_EXPORT ManifestInfo {
+  /// Field id: 521. 0 denotes a pre-v4 manifest; 4 denotes a v4 manifest.
+  int32_t format_version = 0;
+  /// Field id: 522. Bitmap of positions that are no longer live in a leaf manifest.
+  std::optional<std::vector<uint8_t>> dv;
+  /// Field id: 504.
+  int32_t added_files_count = 0;
+  /// Field id: 505.
+  int32_t existing_files_count = 0;
+  /// Field id: 525.
+  int32_t modified_files_count = 0;
+  /// Field id: 506.
+  int32_t deleted_files_count = 0;
+  /// Field id: 523.
+  int32_t replaced_files_count = 0;
+  /// Field id: 512.
+  int64_t added_rows_count = 0;
+  /// Field id: 513.
+  int64_t existing_rows_count = 0;
+  /// Field id: 526.
+  int64_t modified_rows_count = 0;
+  /// Field id: 514.
+  int64_t deleted_rows_count = 0;
+  /// Field id: 524.
+  int64_t replaced_rows_count = 0;
+  /// Field id: 516. Minimum data sequence number of live entries.
+  int64_t min_sequence_number = 0;
+
+  static std::shared_ptr<StructType> Type();
+};
+
+}  // namespace iceberg
diff --git a/src/iceberg/snapshot.cc b/src/iceberg/snapshot.cc
index f791c3724..49237df14 100644
--- a/src/iceberg/snapshot.cc
+++ b/src/iceberg/snapshot.cc
@@ -165,6 +165,32 @@ Result<std::optional<int64_t>> Snapshot::FirstRowId() const { return first_row_i
 
 Result<std::optional<int64_t>> Snapshot::AddedRows() const { return added_rows; }
 
+Status Snapshot::ValidateForFormatVersion(int8_t format_version,
+                                          bool is_new_snapshot) const {
+  if (!root_manifest.has_value()) {
+    ICEBERG_CHECK(!is_new_snapshot || format_version < 4,
+                  "New v4 snapshots require root-manifest instead of manifest-list");
+    ICEBERG_CHECK(!manifest_list.empty(), "Snapshot manifest-list cannot be empty");
+    return {};
+  }
+
+  ICEBERG_CHECK(format_version >= 4, "root-manifest requires format version 4");
+  ICEBERG_CHECK(manifest_list.empty(), "root-manifest cannot coexist with manifest-list");
+  ICEBERG_CHECK(!root_manifest->empty(), "Snapshot root-manifest cannot be empty");
+  ICEBERG_CHECK(sequence_number > 0, "V4 snapshots require a positive sequence-number");
+  ICEBERG_CHECK(first_row_id.has_value() && first_row_id.value() >= 0,
+                "V4 snapshots require a non-negative first-row-id");
+  ICEBERG_CHECK(added_rows.has_value() && added_rows.value() >= 0,
+                "V4 snapshots require non-negative added-rows");
+  const auto operation = Operation();
+  ICEBERG_CHECK(
+      operation.has_value() &&
+          (operation == DataOperation::kAppend || operation == DataOperation::kReplace ||
+           operation == DataOperation::kOverwrite || operation == DataOperation::kDelete),
+      "V4 snapshots require a valid operation in summary");
+  return {};
+}
+
 bool Snapshot::Equals(const Snapshot& other) const {
   if (this == &other) {
     return true;
@@ -173,7 +199,8 @@ bool Snapshot::Equals(const Snapshot& other) const {
          parent_snapshot_id == other.parent_snapshot_id &&
          sequence_number == other.sequence_number && timestamp_ms == other.timestamp_ms &&
          schema_id == other.schema_id && first_row_id == other.first_row_id &&
-         added_rows == other.added_rows && key_id == other.key_id;
+         added_rows == other.added_rows && key_id == other.key_id &&
+         root_manifest == other.root_manifest;
 }
 
 Result<std::unique_ptr<Snapshot>> Snapshot::Make(
@@ -205,6 +232,9 @@ Result<std::unique_ptr<Snapshot>> Snapshot::Make(
 
 Result<SnapshotReader::ManifestsCache> SnapshotReader::InitManifestsCache(
     const Snapshot* snapshot, std::shared_ptr<FileIO> file_io) {
+  if (snapshot->root_manifest.has_value()) {
+    return NotSupported("Reading v4 root manifests is not supported yet");
+  }
   if (file_io == nullptr) {
     return InvalidArgument("Cannot cache manifests: FileIO is null");
   }
diff --git a/src/iceberg/snapshot.h b/src/iceberg/snapshot.h
index 54c6fea2d..2f648220a 100644
--- a/src/iceberg/snapshot.h
+++ b/src/iceberg/snapshot.h
@@ -417,6 +417,17 @@ struct ICEBERG_EXPORT Snapshot {
   /// encrypted key metadata of the manifest list. Set only for encrypted tables.
   std::optional<std::string> key_id;
 
+  /// Location of the root manifest for a snapshot written in v4. Mutually
+  /// exclusive with manifest_list. Older snapshots retain their manifest list
+  /// when a table is upgraded.
+  std::optional<std::string> root_manifest;
+
+  /// \brief Validate snapshot metadata against a table format version.
+  /// Existing snapshots may retain a legacy manifest list after a v4 upgrade;
+  /// newly added v4 snapshots must use a root manifest.
+  Status ValidateForFormatVersion(int8_t format_version,
+                                  bool is_new_snapshot = false) const;
+
   /// \brief Create a new Snapshot instance with validation on the inputs.
   static Result<std::unique_ptr<Snapshot>> Make(
       int64_t sequence_number, int64_t snapshot_id,
diff --git a/src/iceberg/table_metadata.cc b/src/iceberg/table_metadata.cc
index b67ab32fe..9b0262697 100644
--- a/src/iceberg/table_metadata.cc
+++ b/src/iceberg/table_metadata.cc
@@ -194,10 +194,27 @@ std::string ToString(const MetadataLogEntry& entry) {
                      entry.metadata_file);
 }
 
+Result<std::unique_ptr<TableMetadata>> TableMetadata::Make(
+    const iceberg::Schema& schema, const iceberg::PartitionSpec& spec,
+    const iceberg::SortOrder& sort_order, const std::string& location,
+    const std::unordered_map<std::string, std::string>& properties) {
+  auto persisted_properties = properties;
+  int format_version = kDefaultTableFormatVersion;
+  if (auto it = persisted_properties.find(TableProperties::kFormatVersion.key());
+      it != persisted_properties.end()) {
+    ICEBERG_ASSIGN_OR_RAISE(format_version,
+                            StringUtils::ParseNumber<int32_t>(it->second));
+    persisted_properties.erase(it);
+  }
+  return Make(schema, spec, sort_order, location, persisted_properties, format_version);
+}
+
 Result<std::unique_ptr<TableMetadata>> TableMetadata::Make(
     const iceberg::Schema& schema, const iceberg::PartitionSpec& spec,
     const iceberg::SortOrder& sort_order, const std::string& location,
     const std::unordered_map<std::string, std::string>& properties, int format_version) {
+  ICEBERG_PRECHECK(format_version >= 1 && format_version <= kSupportedTableFormatVersion,
+                   "Unsupported table format version: {}", format_version);
   for (const auto& [key, _] : properties) {
     if (TableProperties::reserved_properties().contains(key)) {
       return InvalidArgument(
@@ -556,6 +573,8 @@ Result<std::string> TableMetadataUtil::NewTableMetadataFilePath(const TableMetad
     return std::format("{}/{}", LocationUtil::StripTrailingSlash(metadata_location),
                        filename);
   } else {
+    ICEBERG_PRECHECK(!meta.location.empty(),
+                     "Cannot generate a metadata file location without a table location");
     return std::format("{}/{}/{}", meta.location, kMetadataFolderName, filename);
   }
 }
@@ -748,7 +767,8 @@ Status TableMetadataBuilder::Impl::AssignUUID(std::string_view uuid) {
 
 Status TableMetadataBuilder::Impl::UpgradeFormatVersion(int8_t new_format_version) {
   // Check that the new format version is supported
-  if (new_format_version > TableMetadata::kSupportedTableFormatVersion) {
+  if (new_format_version < 1 ||
+      new_format_version > TableMetadata::kSupportedTableFormatVersion) {
     return InvalidArgument(
         "Cannot upgrade table to unsupported format version: v{} (supported: v{})",
         new_format_version, TableMetadata::kSupportedTableFormatVersion);
@@ -765,6 +785,18 @@ Status TableMetadataBuilder::Impl::UpgradeFormatVersion(int8_t new_format_versio
     return {};
   }
 
+  // A create transaction may add schemas before selecting the current schema.
+  // Build() will validate the schema selected after this upgrade.
+  if (new_format_version >= 4) {
+    auto current_schema = schemas_by_id_.find(metadata_.current_schema_id);
+    if (current_schema != schemas_by_id_.end() &&
+        !current_schema->second->IdentifierFieldIds().empty()) {
+      return NotSupported(
+          "Cannot upgrade a table with identifier fields to v4: primary-key "
+          "constraint migration is not supported yet");
+    }
+  }
+
   // Update the format version
   metadata_.format_version = new_format_version;
 
@@ -1089,6 +1121,17 @@ Status TableMetadataBuilder::Impl::AddSnapshot(std::shared_ptr<Snapshot> snapsho
                 "Attempting to add a snapshot before a sort order is added");
   ICEBERG_CHECK(!snapshots_by_id_.contains(snapshot->snapshot_id),
                 "Snapshot already exists for id: {}", snapshot->snapshot_id);
+  ICEBERG_RETURN_UNEXPECTED(
+      snapshot->ValidateForFormatVersion(metadata_.format_version, true));
+  if (metadata_.format_version >= TableMetadata::kMinFormatVersionMonotonicTimestamps &&
+      snapshot->parent_snapshot_id.has_value()) {
+    auto parent = snapshots_by_id_.find(snapshot->parent_snapshot_id.value());
+    if (parent != snapshots_by_id_.end() &&
+        snapshot->timestamp_ms <= parent->second->timestamp_ms) {
+      return ValidationFailed(
+          "V4 snapshot timestamp must be greater than its parent timestamp");
+    }
+  }
   if (metadata_.format_version != 1 &&
       snapshot->sequence_number <= metadata_.last_sequence_number &&
       snapshot->parent_snapshot_id.has_value()) {
@@ -1373,6 +1416,15 @@ Result<std::vector<SnapshotLogEntry>> TableMetadataBuilder::Impl::UpdateSnapshot
 
 Result<std::unique_ptr<TableMetadata>> TableMetadataBuilder::Impl::Build() {
   // 1. Validate metadata consistency through TableMetadata#Validate
+  ICEBERG_PRECHECK(
+      metadata_.format_version >= 1 &&
+          metadata_.format_version <= TableMetadata::kSupportedTableFormatVersion,
+      "Unsupported table format version: {}", metadata_.format_version);
+  if (metadata_.format_version >= TableMetadata::kMinFormatVersionOptionalLocation) {
+    ICEBERG_PRECHECK(metadata_.location.empty() ||
+                         !LocationUtil::ParseScheme(metadata_.location).empty(),
+                     "V4 table location must be absolute: {}", metadata_.location);
+  }
 
   // 2. Update last_updated_ms if there are changes
   if (metadata_.last_updated_ms == kInvalidLastUpdatedMs) {
@@ -1387,6 +1439,11 @@ Result<std::unique_ptr<TableMetadata>> TableMetadataBuilder::Impl::Build() {
                    "Current schema ID {} not found in schemas", current_schema_id);
   {
     const auto& current_schema = schema_it->second;
+    if (metadata_.format_version >= 4 && !current_schema->IdentifierFieldIds().empty()) {
+      return NotSupported(
+          "V4 primary-key constraints are not supported yet; "
+          "identifier-field-ids cannot be written in v4 schemas");
+    }
 
     auto spec_it = specs_by_id_.find(metadata_.default_spec_id);
     ICEBERG_PRECHECK(spec_it != specs_by_id_.end(),
diff --git a/src/iceberg/table_metadata.h b/src/iceberg/table_metadata.h
index 6e2d9fe26..6d88016dc 100644
--- a/src/iceberg/table_metadata.h
+++ b/src/iceberg/table_metadata.h
@@ -72,7 +72,12 @@ struct ICEBERG_EXPORT MetadataLogEntry {
 /// (4) Map<String, SnapshotRef>
 struct ICEBERG_EXPORT TableMetadata {
   static constexpr int8_t kDefaultTableFormatVersion = 2;
-  static constexpr int8_t kSupportedTableFormatVersion = 3;
+  /// Highest supported table metadata version. V4 manifest tree I/O is not yet
+  /// implemented.
+  static constexpr int8_t kSupportedTableFormatVersion = 4;
+  static constexpr int8_t kMinFormatVersionOptionalLocation = 4;
+  static constexpr int8_t kMinFormatVersionRootManifest = 4;
+  static constexpr int8_t kMinFormatVersionMonotonicTimestamps = 4;
   static constexpr int8_t kMinFormatVersionRowLineage = 3;
   static constexpr int8_t kMinFormatVersionDefaultValues = 3;
   static constexpr int64_t kInitialSequenceNumber = 0;
@@ -87,7 +92,8 @@ struct ICEBERG_EXPORT TableMetadata {
   int8_t format_version;
   /// A UUID that identifies the table
   std::string table_uuid;
-  /// The table's base location
+  /// The table's base location. Empty represents an omitted location in v4
+  /// metadata; a catalog must provide the location before resolving relative paths.
   std::string location;
   /// The table's highest assigned sequence number
   int64_t last_sequence_number;
@@ -135,8 +141,12 @@ struct ICEBERG_EXPORT TableMetadata {
   static Result<std::unique_ptr<TableMetadata>> Make(
       const iceberg::Schema& schema, const iceberg::PartitionSpec& spec,
       const iceberg::SortOrder& sort_order, const std::string& location,
-      const std::unordered_map<std::string, std::string>& properties,
-      int format_version = kDefaultTableFormatVersion);
+      const std::unordered_map<std::string, std::string>& properties);
+
+  static Result<std::unique_ptr<TableMetadata>> Make(
+      const iceberg::Schema& schema, const iceberg::PartitionSpec& spec,
+      const iceberg::SortOrder& sort_order, const std::string& location,
+      const std::unordered_map<std::string, std::string>& properties, int format_version);
 
   /// \brief Get the current schema, return NotFoundError if not found
   /// \note The returned schema is guaranteed to be not null
diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt
index 2513ff177..721813e0d 100644
--- a/src/iceberg/test/CMakeLists.txt
+++ b/src/iceberg/test/CMakeLists.txt
@@ -84,6 +84,9 @@ add_iceberg_test(schema_test
                  transform_test.cc
                  type_test.cc)
 
+add_iceberg_test(v4_types_test SOURCES v4_types_test.cc)
+add_iceberg_test(v4_table_metadata_test SOURCES v4_table_metadata_test.cc)
+
 add_iceberg_test(table_test
                  SOURCES
                  catalog_util_test.cc
diff --git a/src/iceberg/test/expire_snapshots_test.cc b/src/iceberg/test/expire_snapshots_test.cc
index 9695b4a85..bd92e0399 100644
--- a/src/iceberg/test/expire_snapshots_test.cc
+++ b/src/iceberg/test/expire_snapshots_test.cc
@@ -30,6 +30,7 @@
 #include <gmock/gmock.h>
 
 #include "iceberg/avro/avro_register.h"
+#include "iceberg/logging/logger.h"
 #include "iceberg/manifest/manifest_entry.h"
 #include "iceberg/manifest/manifest_writer.h"
 #include "iceberg/partition_spec.h"
@@ -39,6 +40,7 @@
 #include "iceberg/table_metadata.h"
 #include "iceberg/table_properties.h"
 #include "iceberg/test/executor.h"
+#include "iceberg/test/logging_test_helpers.h"
 #include "iceberg/test/matchers.h"
 #include "iceberg/test/mock_catalog.h"
 #include "iceberg/test/update_test_base.h"
@@ -197,6 +199,19 @@ class ExpireSnapshotsCleanupTest : public UpdateTestBase {
     RewriteTable(std::move(metadata));
   }
 
+  void UseV4Root(TableMetadata& metadata) const {
+    metadata.format_version = 4;
+    metadata.location = "file:" + table_location_;
+    // Keep the mock filesystem's metadata writes at its existing absolute path.
+    metadata.properties.Set(TableProperties::kWriteMetadataLocation,
+                            table_location_ + "/metadata");
+    auto& current = *metadata.snapshots.back();
+    current.manifest_list.clear();
+    current.root_manifest = table_location_ + "/metadata/current-root.avro";
+    current.first_row_id = 0;
+    current.added_rows = 0;
+  }
+
   void RewriteTable(std::shared_ptr<TableMetadata> metadata) {
     ASSERT_NE(metadata, nullptr);
 
@@ -224,6 +239,126 @@ class ExpireSnapshotsCleanupTest : public UpdateTestBase {
   }
 };
 
+class ExpireSnapshotsV4CleanupTest : public ExpireSnapshotsCleanupTest,
+                                     public ::testing::WithParamInterface<CleanupLevel> {
+};
+
+TEST_P(ExpireSnapshotsV4CleanupTest, RejectsCleanupBeforeCommit) {
+  const auto data_path = table_location_ + "/data/shared.parquet";
+  const auto manifest_path = table_location_ + "/metadata/shared-manifest.avro";
+  const auto expired_list = table_location_ + "/metadata/expired-list.avro";
+  ASSERT_THAT(file_io_->WriteFile(data_path, "data"), IsOk());
+  auto manifest =
+      WriteDataManifest(manifest_path, kExpiredSnapshotId,
+                        {MakeEntry(ManifestStatus::kAdded, kExpiredSnapshotId,
+                                   kExpiredSequenceNumber, MakeDataFile(data_path))});
+  WriteManifestList(expired_list, kExpiredSnapshotId, 0, kExpiredSequenceNumber,
+                    {manifest});
+
+  auto metadata = ReloadMetadata();
+  metadata->snapshots.front()->manifest_list = expired_list;
+  UseV4Root(*metadata);
+  RewriteTable(std::move(metadata));
+  const auto metadata_location = std::string(table_->metadata_file_location());
+
+  // The opaque v4 root may still reference the legacy manifest. Exercise both
+  // incremental (by age) and reachable (by explicit ID) cleanup selection.
+  for (bool explicit_id : {false, true}) {
+    SCOPED_TRACE(explicit_id);
+    std::vector<std::string> deleted_files;
+    ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
+    if (explicit_id) {
+      update->ExpireSnapshotId(kExpiredSnapshotId);
+    } else {
+      update->ExpireOlderThan(
+          UnixMsFromTimePointMs(table_->metadata()->snapshots.back()->timestamp_ms));
+    }
+    update->CleanupLevel(GetParam()).DeleteWith([&](const std::string& path) {
+      deleted_files.push_back(path);
+    });
+    EXPECT_THAT(update->Apply(), IsError(ErrorKind::kNotSupported));
+    EXPECT_THAT(update->Commit(), IsError(ErrorKind::kNotSupported));
+    EXPECT_TRUE(deleted_files.empty());
+    EXPECT_TRUE(ReloadMetadata()->SnapshotById(kExpiredSnapshotId).has_value());
+    EXPECT_EQ(table_->metadata_file_location(), metadata_location);
+    for (const auto& path : {data_path, manifest_path, expired_list}) {
+      EXPECT_THAT(file_io_->ReadFile(path, std::nullopt), IsOk());
+    }
+  }
+}
+
+INSTANTIATE_TEST_SUITE_P(FileCleanupLevels, ExpireSnapshotsV4CleanupTest,
+                         ::testing::Values(CleanupLevel::kAll,
+                                           CleanupLevel::kMetadataOnly));
+
+TEST_F(ExpireSnapshotsCleanupTest, V4SupportsNoOpAndExpirationWithoutFileCleanup) {
+  const auto expired_list = table_location_ + "/metadata/expired-list.avro";
+  WriteManifestList(expired_list, kExpiredSnapshotId, 0, kExpiredSequenceNumber, {});
+  auto metadata = ReloadMetadata();
+  metadata->snapshots.front()->manifest_list = expired_list;
+  UseV4Root(*metadata);
+  RewriteTable(std::move(metadata));
+
+  ICEBERG_UNWRAP_OR_FAIL(auto noop, table_->NewExpireSnapshots());
+  noop->RetainLast(2);
+  EXPECT_THAT(noop->Commit(), IsOk());
+
+  std::vector<std::string> deleted_files;
+  ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
+  update->ExpireSnapshotId(kExpiredSnapshotId)
+      .CleanupLevel(CleanupLevel::kNone)
+      .DeleteWith([&](const std::string& path) { deleted_files.push_back(path); });
+  ASSERT_THAT(update->Commit(), IsOk());
+  EXPECT_TRUE(deleted_files.empty());
+  metadata = ReloadMetadata();
+  EXPECT_FALSE(metadata->SnapshotById(kExpiredSnapshotId).has_value());
+  EXPECT_TRUE(metadata->SnapshotById(kCurrentSnapshotId).has_value());
+  EXPECT_THAT(file_io_->ReadFile(expired_list, std::nullopt), IsOk());
+}
+
+TEST_P(ExpireSnapshotsV4CleanupTest, SkipsCleanupWhenCommittedMetadataAddsV4Root) {
+  const auto expired_list = table_location_ + "/metadata/expired-list.avro";
+  const auto current_list = table_location_ + "/metadata/current-list.avro";
+  WriteManifestList(expired_list, kExpiredSnapshotId, 0, kExpiredSequenceNumber, {});
+  WriteManifestList(current_list, kCurrentSnapshotId, kExpiredSnapshotId,
+                    kCurrentSequenceNumber, {});
+  RewriteTableWithManifestLists(expired_list, current_list);
+
+  auto mock = std::make_shared<::testing::NiceMock<MockCatalog>>();
+  EXPECT_CALL(*mock, UpdateTable(::testing::_, ::testing::_, ::testing::_))
+      .WillOnce([this](const auto& name, const auto& requirements,
+                       const auto& updates) -> Result<std::shared_ptr<Table>> {
+        ICEBERG_ASSIGN_OR_RAISE(auto committed,
+                                catalog_->UpdateTable(name, requirements, updates));
+        UseV4Root(*committed->metadata());
+        ICEBERG_RETURN_UNEXPECTED(TableMetadataUtil::Write(
+            *file_io_, std::string(committed->metadata_file_location()),
+            *committed->metadata()));
+        return committed;
+      });
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto table,
+      Table::Make(table_->name(), table_->metadata(),
+                  std::string(table_->metadata_file_location()), file_io_, mock));
+  std::vector<std::string> deleted_files;
+  ICEBERG_UNWRAP_OR_FAIL(auto update, table->NewExpireSnapshots());
+  update->ExpireSnapshotId(kExpiredSnapshotId)
+      .CleanupLevel(GetParam())
+      .DeleteWith([&](const std::string& path) { deleted_files.push_back(path); });
+  auto logger = std::make_shared<CapturingLogger>();
+  ScopedLogger scoped_logger(logger);
+  ASSERT_THAT(update->Commit(), IsOk());
+  EXPECT_TRUE(deleted_files.empty());
+  auto committed = ReloadMetadata();
+  EXPECT_FALSE(committed->SnapshotById(kExpiredSnapshotId).has_value());
+  EXPECT_TRUE(committed->snapshots.back()->root_manifest.has_value());
+  EXPECT_THAT(file_io_->ReadFile(expired_list, std::nullopt), IsOk());
+  for (const auto& record : logger->records()) {
+    EXPECT_THAT(record.message,
+                ::testing::Not(::testing::HasSubstr("Update finalization failed")));
+  }
+}
+
 TEST_F(ExpireSnapshotsTest, DefaultExpireByAge) {
   ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
   ICEBERG_UNWRAP_OR_FAIL(auto result, update->Apply());
diff --git a/src/iceberg/test/json_serde_test.cc b/src/iceberg/test/json_serde_test.cc
index 13b28a7bb..5770ca190 100644
--- a/src/iceberg/test/json_serde_test.cc
+++ b/src/iceberg/test/json_serde_test.cc
@@ -44,6 +44,7 @@
 #include "iceberg/sort_field.h"
 #include "iceberg/sort_order.h"
 #include "iceberg/statistics_file.h"
+#include "iceberg/table_metadata.h"
 #include "iceberg/table_requirement.h"
 #include "iceberg/table_scan.h"
 #include "iceberg/table_update.h"
@@ -687,16 +688,50 @@ TEST(JsonInternalTest, TableUpdateAssignUUID) {
 }
 
 TEST(JsonInternalTest, TableUpdateUpgradeFormatVersion) {
-  table::UpgradeFormatVersion update(2);
-  nlohmann::json expected =
-      R"({"action":"upgrade-format-version","format-version":2})"_json;
+  for (int8_t version :
+       std::array<int8_t, 8>{1, 2, 3, 4, 0, 5, std::numeric_limits<int8_t>::min(),
+                             std::numeric_limits<int8_t>::max()}) {
+    SCOPED_TRACE(version);
+    table::UpgradeFormatVersion update(version);
+    nlohmann::json expected = {{"action", "upgrade-format-version"},
+                               {"format-version", version}};
+
+    ICEBERG_UNWRAP_OR_FAIL(auto json, ToJson(update));
+    EXPECT_EQ(json, expected);
+    ICEBERG_UNWRAP_OR_FAIL(auto parsed, TableUpdateFromJson(expected));
+    const auto& actual =
+        internal::checked_cast<const table::UpgradeFormatVersion&>(*parsed);
+    EXPECT_EQ(actual.format_version(), version);
+    EXPECT_EQ(actual, update);
+
+    if (version < 1 || version > TableMetadata::kSupportedTableFormatVersion) {
+      auto builder = TableMetadataBuilder::BuildFromEmpty(2);
+      parsed->ApplyTo(*builder);
+      auto result = builder->Build();
+      EXPECT_THAT(result, IsError(ErrorKind::kValidationFailed));
+      EXPECT_THAT(result, HasErrorMessage(
+                              "Cannot upgrade table to unsupported format version: v" +
+                              std::to_string(version)));
+      EXPECT_TRUE(builder->changes().empty());
+    }
+  }
+}
 
-  ICEBERG_UNWRAP_OR_FAIL(auto json, ToJson(update));
-  EXPECT_EQ(json, expected);
-  auto parsed = TableUpdateFromJson(expected);
-  ASSERT_THAT(parsed, IsOk());
-  EXPECT_EQ(*internal::checked_cast<table::UpgradeFormatVersion*>(parsed.value().get()),
-            update);
+TEST(JsonInternalTest, TableUpdateUpgradeFormatVersionRejectsInvalidIntegers) {
+  for (const auto& value :
+       {nlohmann::json(128), nlohmann::json(-129), nlohmann::json(260),
+        nlohmann::json(-252), nlohmann::json(4294967300LL), nlohmann::json(-4294967292LL),
+        nlohmann::json(2147483648LL), nlohmann::json(-2147483649LL),
+        nlohmann::json(std::numeric_limits<uint64_t>::max()), nlohmann::json(4.0),
+        nlohmann::json(4.5), nlohmann::json("4"), nlohmann::json(true),
+        nlohmann::json(nullptr)}) {
+    SCOPED_TRACE(value.dump());
+    nlohmann::json json = {{"action", "upgrade-format-version"},
+                           {"format-version", value}};
+    EXPECT_THAT(TableUpdateFromJson(json), IsError(ErrorKind::kJsonParseError));
+  }
+  EXPECT_THAT(TableUpdateFromJson({{"action", "upgrade-format-version"}}),
+              IsError(ErrorKind::kJsonParseError));
 }
 
 TEST(JsonInternalTest, TableUpdateAddSchema) {
diff --git a/src/iceberg/test/location_provider_test.cc b/src/iceberg/test/location_provider_test.cc
index 268199b3b..8b1f3062a 100644
--- a/src/iceberg/test/location_provider_test.cc
+++ b/src/iceberg/test/location_provider_test.cc
@@ -74,6 +74,35 @@ TEST_F(LocationProviderTest, DefaultLocationProviderWithCustomDataLocation) {
   EXPECT_EQ("new_location/my_file", location);
 }
 
+TEST_F(LocationProviderTest, MissingTableLocationRequiresAnExplicitDataLocation) {
+  auto missing = LocationProvider::Make("", properties_);
+  EXPECT_THAT(missing, IsError(ErrorKind::kInvalidArgument));
+  EXPECT_THAT(missing, HasErrorMessage("absolute URI in 'write.data.path'"));
+  for (const auto* data_location : {"data", "/data"}) {
+    properties_.Set(TableProperties::kWriteDataLocation, std::string(data_location));
+    EXPECT_THAT(LocationProvider::Make("", properties_),
+                IsError(ErrorKind::kInvalidArgument));
+  }
+  properties_.Set(TableProperties::kWriteDataLocation, std::string("s3://data/table"));
+  ICEBERG_UNWRAP_OR_FAIL(auto provider, LocationProvider::Make("", properties_));
+  EXPECT_EQ(provider->NewDataLocation("f.parquet"), "s3://data/table/f.parquet");
+}
+
+TEST_F(LocationProviderTest, ObjectStoreRequiresTableLocationForPathContext) {
+  properties_.Set(TableProperties::kObjectStoreEnabled, true);
+  auto missing = LocationProvider::Make("", properties_);
+  EXPECT_THAT(missing, IsError(ErrorKind::kInvalidArgument));
+  EXPECT_THAT(missing, HasErrorMessage(
+                           "Object-store location provider requires a table location"));
+  properties_.Set(TableProperties::kWriteDataLocation, std::string("s3://shared-data"));
+  EXPECT_THAT(LocationProvider::Make("", properties_),
+              IsError(ErrorKind::kInvalidArgument));
+  ICEBERG_UNWRAP_OR_FAIL(auto provider,
+                         LocationProvider::Make("s3://bucket/db/table", properties_));
+  EXPECT_THAT(provider->NewDataLocation("f.parquet"),
+              testing::EndsWith("/db/table/f.parquet"));
+}
+
 TEST_F(LocationProviderTest, ObjectStorageLocationProvider) {
   std::ignore = properties_.Set(TableProperties::kObjectStoreEnabled, true);
   ICEBERG_UNWRAP_OR_FAIL(auto provider,
diff --git a/src/iceberg/test/snapshots_table_test.cc b/src/iceberg/test/snapshots_table_test.cc
index 9a62c6b63..a793ca36f 100644
--- a/src/iceberg/test/snapshots_table_test.cc
+++ b/src/iceberg/test/snapshots_table_test.cc
@@ -84,7 +84,7 @@ TEST_F(SnapshotsTableTest, Scan) {
 
   // Row and column counts.
   EXPECT_EQ(batch->num_rows(), 2);
-  EXPECT_EQ(batch->num_columns(), 6);
+  EXPECT_EQ(batch->num_columns(), 7);
 
   // Column 0: committed_at (timestamptz) — microseconds since epoch.
   auto committed_at = std::static_pointer_cast<::arrow::TimestampArray>(batch->column(0));
@@ -130,6 +130,66 @@ TEST_F(SnapshotsTableTest, Scan) {
       second_summary,
       ::testing::Not(::testing::Contains(::testing::Pair("operation", "append"))));
   EXPECT_THAT(second_summary, ::testing::Contains(::testing::Pair("total-records", "2")));
+
+  // Legacy snapshots have no root manifest.
+  EXPECT_EQ(batch->column(6)->null_count(), 2);
+}
+
+TEST_F(SnapshotsTableTest, ExposesRootManifestAndNullLegacyManifestList) {
+  auto& metadata = *table_->metadata();
+  metadata.format_version = 4;
+  auto& snapshot = *metadata.snapshots.back();
+  snapshot.manifest_list.clear();
+  snapshot.root_manifest = "s3://bucket/metadata/root.avro";
+  snapshot.first_row_id = 0;
+  snapshot.added_rows = 0;
+
+  ICEBERG_UNWRAP_OR_FAIL(auto stream, Scan(*snapshots_table_));
+  ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream)));
+  ASSERT_EQ(batches.size(), 1);
+  const auto& batch = batches.front();
+  auto lists = std::static_pointer_cast<::arrow::StringArray>(
+      batch->GetColumnByName("manifest_list"));
+  auto roots = std::static_pointer_cast<::arrow::StringArray>(
+      batch->GetColumnByName("root_manifest"));
+  ASSERT_NE(lists, nullptr);
+  ASSERT_NE(roots, nullptr);
+  EXPECT_EQ(lists->GetString(0), "file:/tmp/manifest1.avro");
+  EXPECT_TRUE(lists->IsNull(1));
+  EXPECT_TRUE(roots->IsNull(0));
+  EXPECT_EQ(roots->GetString(1), snapshot.root_manifest.value());
+
+  auto projection = std::make_shared<Schema>(
+      std::vector<SchemaField>{SchemaField::MakeOptional(9, "root_manifest", string()),
+                               SchemaField::MakeOptional(5, "manifest_list", string())});
+  ICEBERG_UNWRAP_OR_FAIL(auto builder, snapshots_table_->NewScan());
+  ICEBERG_UNWRAP_OR_FAIL(auto projected_stream, builder->Project(projection).Build());
+  ICEBERG_UNWRAP_OR_FAIL(auto projected_batches,
+                         ReadAllBatches(std::move(projected_stream)));
+  ASSERT_EQ(projected_batches.size(), 1);
+  const auto& projected = projected_batches.front();
+  ASSERT_EQ(projected->num_columns(), 2);
+  EXPECT_EQ(projected->column_name(0), "root_manifest");
+  EXPECT_EQ(projected->column_name(1), "manifest_list");
+  auto projected_roots =
+      std::static_pointer_cast<::arrow::StringArray>(projected->column(0));
+  auto projected_lists =
+      std::static_pointer_cast<::arrow::StringArray>(projected->column(1));
+  EXPECT_TRUE(projected_roots->IsNull(0));
+  EXPECT_EQ(projected_roots->GetString(1), snapshot.root_manifest.value());
+  EXPECT_EQ(projected_lists->GetString(0), "file:/tmp/manifest1.avro");
+  EXPECT_TRUE(projected_lists->IsNull(1));
+}
+
+TEST_F(SnapshotsTableTest, EmptyLegacyManifestListRemainsAString) {
+  table_->metadata()->snapshots.back()->manifest_list.clear();
+  ICEBERG_UNWRAP_OR_FAIL(auto stream, Scan(*snapshots_table_));
+  ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream)));
+  ASSERT_EQ(batches.size(), 1);
+  auto lists = std::static_pointer_cast<::arrow::StringArray>(
+      batches.front()->GetColumnByName("manifest_list"));
+  EXPECT_FALSE(lists->IsNull(1));
+  EXPECT_EQ(lists->GetString(1), "");
 }
 
 TEST_F(SnapshotsTableTest, ScanSelectedColumns) {
@@ -186,7 +246,7 @@ TEST_F(SnapshotsTableTest, ScanAllColumnsSelection) {
   ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream)));
   ASSERT_EQ(batches.size(), 1);
   EXPECT_EQ(batches.front()->num_rows(), 2);
-  EXPECT_EQ(batches.front()->num_columns(), 6);
+  EXPECT_EQ(batches.front()->num_columns(), 7);
 }
 
 TEST_F(SnapshotsTableTest, ScanBuildsRepeatedly) {
diff --git a/src/iceberg/test/table_metadata_builder_test.cc b/src/iceberg/test/table_metadata_builder_test.cc
index 5a9681add..f1a454165 100644
--- a/src/iceberg/test/table_metadata_builder_test.cc
+++ b/src/iceberg/test/table_metadata_builder_test.cc
@@ -17,6 +17,7 @@
  * under the License.
  */
 
+#include <limits>
 #include <memory>
 #include <ranges>
 #include <string>
@@ -432,7 +433,7 @@ TEST(TableMetadataBuilderTest, UpgradeFormatVersion) {
 
   // Unsupported format version should fail
   builder = TableMetadataBuilder::BuildFromEmpty(3);
-  builder->UpgradeFormatVersion(4);
+  builder->UpgradeFormatVersion(5);
   EXPECT_THAT(builder->Build(),
               HasErrorMessage("Cannot upgrade table to unsupported format version"));
 
@@ -442,6 +443,44 @@ TEST(TableMetadataBuilderTest, UpgradeFormatVersion) {
   EXPECT_THAT(builder->Build(), HasErrorMessage("Cannot downgrade"));
 }
 
+TEST(TableMetadataBuilderTest, UpgradeFormatVersionRejectsUnsupportedVersions) {
+  auto base = CreateBaseMetadata();
+  for (int8_t version : std::vector<int8_t>{0, 5, -1, std::numeric_limits<int8_t>::min(),
+                                            std::numeric_limits<int8_t>::max()}) {
+    SCOPED_TRACE(version);
+    auto empty_builder = TableMetadataBuilder::BuildFromEmpty(version);
+    auto empty_result = empty_builder->Build();
+    EXPECT_THAT(empty_result, IsError(ErrorKind::kInvalidArgument));
+    EXPECT_THAT(empty_result, HasErrorMessage("Unsupported table format version: " +
+                                              std::to_string(version)));
+
+    auto builder = TableMetadataBuilder::BuildFrom(base.get());
+    builder->UpgradeFormatVersion(version);
+    auto result = builder->Build();
+    EXPECT_THAT(result, IsError(ErrorKind::kValidationFailed));
+    EXPECT_THAT(result,
+                HasErrorMessage("Cannot upgrade table to unsupported format version: v" +
+                                std::to_string(version)));
+    EXPECT_TRUE(builder->changes().empty());
+
+    table::UpgradeFormatVersion update(version);
+    auto cloned = update.Clone();
+    EXPECT_EQ(*cloned, update);
+    const TableUpdate* updates[] = {&update, cloned.get()};
+    for (const auto* candidate : updates) {
+      auto update_builder = TableMetadataBuilder::BuildFrom(base.get());
+      candidate->ApplyTo(*update_builder);
+      auto update_result = update_builder->Build();
+      EXPECT_THAT(update_result, IsError(ErrorKind::kValidationFailed));
+      EXPECT_THAT(
+          update_result,
+          HasErrorMessage("Cannot upgrade table to unsupported format version: v" +
+                          std::to_string(version)));
+      EXPECT_TRUE(update_builder->changes().empty());
+    }
+  }
+}
+
 // Test AddSortOrder
 TEST(TableMetadataBuilderTest, AddSortOrderBasic) {
   auto base = CreateBaseMetadata();
@@ -1292,8 +1331,10 @@ TEST(TableMetadataBuilderTest, RemoveSnapshotRef) {
   auto builder = TableMetadataBuilder::BuildFrom(base.get());
 
   // Add multiple snapshots
-  builder->AddSnapshot(std::make_shared<Snapshot>(Snapshot{.snapshot_id = 1}));
-  builder->AddSnapshot(std::make_shared<Snapshot>(Snapshot{.snapshot_id = 2}));
+  builder->AddSnapshot(std::make_shared<Snapshot>(
+      Snapshot{.snapshot_id = 1, .manifest_list = "s3://bucket/manifest-list-1.avro"}));
+  builder->AddSnapshot(std::make_shared<Snapshot>(
+      Snapshot{.snapshot_id = 2, .manifest_list = "s3://bucket/manifest-list-2.avro"}));
 
   // Add multiple refs
   ICEBERG_UNWRAP_OR_FAIL(auto ref1, SnapshotRef::MakeBranch(1));
@@ -1316,7 +1357,8 @@ TEST(TableMetadataBuilderTest, SetRefRejectsTagForMainBranch) {
   auto base = CreateBaseMetadata();
   auto builder = TableMetadataBuilder::BuildFrom(base.get());
 
-  builder->AddSnapshot(std::make_shared<Snapshot>(Snapshot{.snapshot_id = 1}));
+  builder->AddSnapshot(std::make_shared<Snapshot>(
+      Snapshot{.snapshot_id = 1, .manifest_list = "s3://bucket/manifest-list-1.avro"}));
   ICEBERG_UNWRAP_OR_FAIL(auto main_tag, SnapshotRef::MakeTag(1));
 
   builder->SetRef(std::string(SnapshotRef::kMainBranch), std::move(main_tag));
@@ -1332,7 +1374,10 @@ TEST(TableMetadataBuilderTest, SetMainRefToAddedSnapshotUsesSnapshotTimestampFor
 
   auto snapshot_time = TimePointMsFromUnixMs(123456789);
   builder->AddSnapshot(std::make_shared<Snapshot>(
-      Snapshot{.snapshot_id = 1, .sequence_number = 1, .timestamp_ms = snapshot_time}));
+      Snapshot{.snapshot_id = 1,
+               .sequence_number = 1,
+               .timestamp_ms = snapshot_time,
+               .manifest_list = "s3://bucket/manifest-list-1.avro"}));
   ICEBERG_UNWRAP_OR_FAIL(auto main_branch, SnapshotRef::MakeBranch(1));
 
   builder->SetRef(std::string(SnapshotRef::kMainBranch), std::move(main_branch));
@@ -1350,8 +1395,10 @@ TEST(TableMetadataBuilderTest, RemoveSnapshot) {
   auto builder = TableMetadataBuilder::BuildFrom(base.get());
 
   // Add multiple snapshots
-  builder->AddSnapshot(std::make_shared<Snapshot>(Snapshot{.snapshot_id = 1}));
-  builder->AddSnapshot(std::make_shared<Snapshot>(Snapshot{.snapshot_id = 2}));
+  builder->AddSnapshot(std::make_shared<Snapshot>(
+      Snapshot{.snapshot_id = 1, .manifest_list = "s3://bucket/manifest-list-1.avro"}));
+  builder->AddSnapshot(std::make_shared<Snapshot>(
+      Snapshot{.snapshot_id = 2, .manifest_list = "s3://bucket/manifest-list-2.avro"}));
 
   ICEBERG_UNWRAP_OR_FAIL(auto metadata, builder->Build());
   ASSERT_EQ(metadata->snapshots.size(), 2);
@@ -1370,8 +1417,10 @@ TEST(TableMetadataBuilderTest, RemoveSnapshotNotExist) {
   auto builder = TableMetadataBuilder::BuildFrom(base.get());
 
   // Add multiple snapshots
-  builder->AddSnapshot(std::make_shared<Snapshot>(Snapshot{.snapshot_id = 1}));
-  builder->AddSnapshot(std::make_shared<Snapshot>(Snapshot{.snapshot_id = 2}));
+  builder->AddSnapshot(std::make_shared<Snapshot>(
+      Snapshot{.snapshot_id = 1, .manifest_list = "s3://bucket/manifest-list-1.avro"}));
+  builder->AddSnapshot(std::make_shared<Snapshot>(
+      Snapshot{.snapshot_id = 2, .manifest_list = "s3://bucket/manifest-list-2.avro"}));
 
   ICEBERG_UNWRAP_OR_FAIL(auto metadata, builder->Build());
   ASSERT_EQ(metadata->snapshots.size(), 2);
diff --git a/src/iceberg/test/update_schema_test.cc b/src/iceberg/test/update_schema_test.cc
index 4057c52c8..474699135 100644
--- a/src/iceberg/test/update_schema_test.cc
+++ b/src/iceberg/test/update_schema_test.cc
@@ -159,7 +159,7 @@ class UpdateSchemaDefaultValueTest : public UpdateSchemaTest {
 
     ICEBERG_UNWRAP_OR_FAIL(auto metadata,
                            ReadTableMetadataFromResource("TableMetadataV2Valid.json"));
-    metadata->format_version = TableMetadata::kSupportedTableFormatVersion;
+    metadata->format_version = TableMetadata::kMinFormatVersionDefaultValues;
     metadata->next_row_id = TableMetadata::kInitialRowId;
     RegisterTable(std::move(metadata));
   }
diff --git a/src/iceberg/test/v4_table_metadata_test.cc b/src/iceberg/test/v4_table_metadata_test.cc
new file mode 100644
index 000000000..e1d2a5dcd
--- /dev/null
+++ b/src/iceberg/test/v4_table_metadata_test.cc
@@ -0,0 +1,546 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <string>
+#include <string_view>
+#include <vector>
+
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+#include <nlohmann/json.hpp>
+
+#include "iceberg/catalog/memory/in_memory_catalog.h"
+#include "iceberg/file_io.h"
+#include "iceberg/json_serde_internal.h"
+#include "iceberg/location_provider.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_field.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/sort_order.h"
+#include "iceberg/table.h"
+#include "iceberg/table_identifier.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/table_update.h"
+#include "iceberg/test/matchers.h"
+#include "iceberg/test/mock_catalog.h"
+#include "iceberg/test/mock_io.h"
+#include "iceberg/transaction.h"
+#include "iceberg/type.h"
+#include "iceberg/update/fast_append.h"
+#include "iceberg/update/update_properties.h"
+
+namespace iceberg {
+namespace {
+
+std::shared_ptr<Schema> TestSchema() {
+  return std::make_shared<Schema>(
+      std::vector<SchemaField>{SchemaField::MakeRequired(1, "id", int64())}, 0);
+}
+
+Result<std::unique_ptr<TableMetadata>> NewMetadata(int version = 4) {
+  return TableMetadata::Make(*TestSchema(), *PartitionSpec::Unpartitioned(),
+                             *SortOrder::Unsorted(), "s3://bucket/table", {}, version);
+}
+
+// A complete v4 metadata fixture independent of the builder and serializer.
+nlohmann::json MetadataJson() {
+  return nlohmann::json::parse(R"({
+    "format-version": 4,
+    "table-uuid": "c9dc1a9f-6564-42ec-9a87-1db146e18019",
+    "location": "s3://bucket/table",
+    "last-sequence-number": 0,
+    "last-updated-ms": 1000,
+    "last-column-id": 1,
+    "schemas": [{"type": "struct", "schema-id": 0, "fields": [
+      {"id": 1, "name": "id", "type": "long", "required": true}
+    ]}],
+    "current-schema-id": 0,
+    "partition-specs": [{"spec-id": 0, "fields": []}],
+    "default-spec-id": 0,
+    "last-partition-id": 999,
+    "sort-orders": [{"order-id": 0, "fields": []}],
+    "default-sort-order-id": 0,
+    "next-row-id": 0,
+    "snapshots": []
+  })");
+}
+
+nlohmann::json RootSnapshotJson() {
+  return nlohmann::json::parse(R"({
+    "snapshot-id": 1,
+    "sequence-number": 1,
+    "timestamp-ms": 1000,
+    "root-manifest": "metadata/root.avro",
+    "summary": {"operation": "append"},
+    "schema-id": 0,
+    "first-row-id": 0,
+    "added-rows": 10
+  })");
+}
+
+TEST(V4TableMetadataTest, CreateAndRoundTrip) {
+  ICEBERG_UNWRAP_OR_FAIL(auto metadata, NewMetadata());
+  EXPECT_EQ(metadata->format_version, 4);
+  EXPECT_EQ(metadata->next_row_id, 0);
+  EXPECT_TRUE(metadata->snapshots.empty());
+  EXPECT_FALSE(metadata->table_uuid.empty());
+  ICEBERG_UNWRAP_OR_FAIL(auto json, ToJson(*metadata));
+  EXPECT_EQ(json["format-version"], 4);
+  EXPECT_EQ(json["next-row-id"], 0);
+  ICEBERG_UNWRAP_OR_FAIL(auto parsed, TableMetadataFromJson(json));
+  EXPECT_EQ(*metadata, *parsed);
+}
+
+TEST(V4TableMetadataTest, SerializationRejectsNullSnapshots) {
+  auto snapshot_json = RootSnapshotJson();
+  snapshot_json.erase("root-manifest");
+  snapshot_json["manifest-list"] = "s3://bucket/metadata/legacy.avro";
+  ICEBERG_UNWRAP_OR_FAIL(auto snapshot, SnapshotFromJson(snapshot_json));
+  auto valid = std::shared_ptr<Snapshot>(std::move(snapshot));
+  for (int version : {1, 2, 3, 4}) {
+    SCOPED_TRACE(version);
+    ICEBERG_UNWRAP_OR_FAIL(auto metadata, NewMetadata(version));
+    for (bool null_first : {false, true}) {
+      metadata->snapshots = null_first
+                                ? std::vector<std::shared_ptr<Snapshot>>{nullptr, valid}
+                                : std::vector<std::shared_ptr<Snapshot>>{valid, nullptr};
+      auto result = ToJson(*metadata);
+      EXPECT_THAT(result, IsError(ErrorKind::kInvalidArgument));
+      EXPECT_THAT(result, HasErrorMessage("Cannot write null snapshot"));
+    }
+  }
+}
+
+TEST(V4TableMetadataTest, CreateFromFormatVersionProperty) {
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto metadata, TableMetadata::Make(*TestSchema(), *PartitionSpec::Unpartitioned(),
+                                         *SortOrder::Unsorted(), "s3://bucket/table",
+                                         {{"format-version", "4"}, {"owner", "test"}}));
+  EXPECT_EQ(metadata->format_version, 4);
+  EXPECT_FALSE(metadata->properties.configs().contains("format-version"));
+  EXPECT_EQ(metadata->properties.configs().at("owner"), "test");
+  ICEBERG_UNWRAP_OR_FAIL(auto defaults,
+                         NewMetadata(TableMetadata::kDefaultTableFormatVersion));
+  EXPECT_EQ(defaults->format_version, 2);
+}
+
+TEST(V4TableMetadataTest, RejectUnsupportedVersionsWithoutNarrowing) {
+  for (int version : {0, -1, 5, 260}) {
+    SCOPED_TRACE(version);
+    EXPECT_THAT(NewMetadata(version), IsError(ErrorKind::kInvalidArgument));
+    auto json = MetadataJson();
+    json["format-version"] = version;
+    EXPECT_THAT(TableMetadataFromJson(json), HasErrorMessage("unsupported version"));
+  }
+  auto builder = TableMetadataBuilder::BuildFromEmpty(5);
+  EXPECT_THAT(builder->Build(), HasErrorMessage("Unsupported table format version"));
+  for (const auto& version : std::vector<nlohmann::json>{4294967300LL, 4.5, true}) {
+    auto json = MetadataJson();
+    json["format-version"] = version;
+    EXPECT_FALSE(TableMetadataFromJson(json).has_value());
+  }
+}
+
+TEST(V4TableMetadataTest, UpgradePreservesLegacySnapshots) {
+  for (int version : {1, 2, 3}) {
+    SCOPED_TRACE(version);
+    ICEBERG_UNWRAP_OR_FAIL(auto base, NewMetadata(version));
+    auto legacy_json = RootSnapshotJson();
+    legacy_json.erase("root-manifest");
+    legacy_json["manifest-list"] = "s3://bucket/table/metadata/legacy.avro";
+    ICEBERG_UNWRAP_OR_FAIL(auto legacy_snapshot, SnapshotFromJson(legacy_json));
+    base->snapshots.push_back(std::move(legacy_snapshot));
+    base->next_row_id = 10;
+    auto builder = TableMetadataBuilder::BuildFrom(base.get());
+    builder->UpgradeFormatVersion(4);
+    ICEBERG_UNWRAP_OR_FAIL(auto upgraded, builder->Build());
+    EXPECT_EQ(upgraded->format_version, 4);
+    EXPECT_EQ(upgraded->next_row_id, 10);
+    EXPECT_EQ(upgraded->table_uuid, base->table_uuid);
+    ASSERT_EQ(upgraded->snapshots.size(), 1);
+    EXPECT_EQ(upgraded->snapshots.front()->manifest_list,
+              legacy_json["manifest-list"].get<std::string>());
+    EXPECT_FALSE(upgraded->snapshots.front()->root_manifest.has_value());
+    ASSERT_EQ(builder->changes().size(), 1);
+    EXPECT_EQ(builder->changes().front()->kind(),
+              TableUpdate::Kind::kUpgradeFormatVersion);
+    ICEBERG_UNWRAP_OR_FAIL(auto json, ToJson(*upgraded));
+    ICEBERG_UNWRAP_OR_FAIL(auto parsed, TableMetadataFromJson(json));
+    EXPECT_EQ(*parsed, *upgraded);
+    auto downgrade = TableMetadataBuilder::BuildFrom(upgraded.get());
+    downgrade->UpgradeFormatVersion(3);
+    EXPECT_THAT(downgrade->Build(), HasErrorMessage("Cannot downgrade"));
+  }
+}
+
+TEST(V4TableMetadataTest, OptionalLocation) {
+  auto json = MetadataJson();
+  json.erase("location");
+  ICEBERG_UNWRAP_OR_FAIL(auto parsed, TableMetadataFromJson(json));
+  EXPECT_TRUE(parsed->location.empty());
+  ICEBERG_UNWRAP_OR_FAIL(auto serialized, ToJson(*parsed));
+  EXPECT_FALSE(serialized.contains("location"));
+  auto builder = TableMetadataBuilder::BuildFrom(parsed.get());
+  builder->SetLocation("s3://catalog/table");
+  ICEBERG_UNWRAP_OR_FAIL(auto located, builder->Build());
+  EXPECT_EQ(located->location, "s3://catalog/table");
+  for (int version : {1, 2, 3}) {
+    json["format-version"] = version;
+    EXPECT_FALSE(TableMetadataFromJson(json).has_value());
+  }
+}
+
+TEST(V4TableMetadataTest, RejectRelativeTableLocationAndMissingRequiredFields) {
+  auto json = MetadataJson();
+  json["location"] = "relative/table";
+  EXPECT_THAT(TableMetadataFromJson(json), HasErrorMessage("location must be absolute"));
+  ICEBERG_UNWRAP_OR_FAIL(auto metadata, NewMetadata());
+  auto builder = TableMetadataBuilder::BuildFrom(metadata.get());
+  builder->SetLocation("relative/table");
+  EXPECT_THAT(builder->Build(), HasErrorMessage("location must be absolute"));
+  metadata->location = "relative/table";
+  EXPECT_THAT(ToJson(*metadata), HasErrorMessage("location must be absolute"));
+  for (const auto* field : {"next-row-id"}) {
+    json = MetadataJson();
+    json.erase(field);
+    EXPECT_FALSE(TableMetadataFromJson(json).has_value());
+  }
+}
+
+TEST(V4TableMetadataTest, RootManifestRoundTripAndExplicitReaderFailure) {
+  auto json = MetadataJson();
+  json["snapshots"] = nlohmann::json::array({RootSnapshotJson()});
+  json["last-sequence-number"] = 1;
+  json["next-row-id"] = 10;
+  ICEBERG_UNWRAP_OR_FAIL(auto metadata, TableMetadataFromJson(json));
+  ASSERT_EQ(metadata->snapshots.size(), 1);
+  auto snapshot = metadata->snapshots.front();
+  EXPECT_EQ(snapshot->root_manifest, "metadata/root.avro");
+  EXPECT_TRUE(snapshot->manifest_list.empty());
+  ICEBERG_UNWRAP_OR_FAIL(auto serialized, ToJson(*metadata));
+  EXPECT_EQ(serialized["snapshots"][0], RootSnapshotJson());
+  EXPECT_FALSE(serialized["snapshots"][0].contains("manifest-list"));
+  ICEBERG_UNWRAP_OR_FAIL(auto parsed, TableMetadataFromJson(serialized));
+  EXPECT_EQ(parsed->snapshots.front()->root_manifest, snapshot->root_manifest);
+  EXPECT_THAT(SnapshotReader(snapshot.get()).Manifests(nullptr),
+              IsError(ErrorKind::kNotSupported));
+}
+
+TEST(V4TableMetadataTest, RejectInvalidSnapshotRoots) {
+  auto json = RootSnapshotJson();
+  json["manifest-list"] = "legacy.avro";
+  EXPECT_THAT(SnapshotFromJson(json), HasErrorMessage("exactly one"));
+  for (const auto* field :
+       {"root-manifest", "sequence-number", "first-row-id", "added-rows", "summary"}) {
+    json = RootSnapshotJson();
+    json.erase(field);
+    EXPECT_FALSE(SnapshotFromJson(json).has_value()) << field;
+  }
+  json = MetadataJson();
+  json["format-version"] = 3;
+  json["snapshots"] = nlohmann::json::array({RootSnapshotJson()});
+  EXPECT_THAT(TableMetadataFromJson(json), HasErrorMessage("requires format version 4"));
+}
+
+TEST(V4TableMetadataTest, SnapshotConstraintStatusesRequireAnArray) {
+  for (const auto& value :
+       std::vector<nlohmann::json>{nullptr, nlohmann::json::array()}) {
+    auto json = RootSnapshotJson();
+    json["constraint-statuses"] = value;
+    EXPECT_THAT(SnapshotFromJson(json), IsOk());
+  }
+  for (const auto& value :
+       std::vector<nlohmann::json>{true, 0, 0.0, "statuses", nlohmann::json::object()}) {
+    SCOPED_TRACE(value.dump());
+    auto json = RootSnapshotJson();
+    json["constraint-statuses"] = value;
+    auto result = SnapshotFromJson(json);
+    EXPECT_THAT(result, IsError(ErrorKind::kJsonParseError));
+    EXPECT_THAT(result, HasErrorMessage("constraint-statuses"));
+  }
+  auto json = RootSnapshotJson();
+  json["constraint-statuses"] = nlohmann::json::array({{{"constraint-id", 1}}});
+  EXPECT_THAT(SnapshotFromJson(json), IsError(ErrorKind::kNotSupported));
+}
+
+TEST(V4TableMetadataTest, RejectEmptyLegacyManifestListsAcrossMetadataPaths) {
+  auto legacy_json = RootSnapshotJson();
+  legacy_json.erase("root-manifest");
+  legacy_json["manifest-list"] = "legacy.avro";
+  ICEBERG_UNWRAP_OR_FAIL(auto snapshot, SnapshotFromJson(legacy_json));
+  snapshot->manifest_list.clear();
+  legacy_json["manifest-list"] = "";
+  EXPECT_THAT(SnapshotFromJson(legacy_json), IsError(ErrorKind::kJsonParseError));
+  EXPECT_THAT(SnapshotFromJson(legacy_json),
+              HasErrorMessage("manifest-list cannot be empty"));
+
+  for (int8_t version : {1, 2, 3, 4}) {
+    SCOPED_TRACE(version);
+    EXPECT_THAT(snapshot->ValidateForFormatVersion(version),
+                HasErrorMessage("manifest-list cannot be empty"));
+    ICEBERG_UNWRAP_OR_FAIL(auto metadata, NewMetadata(version));
+    if (version < 4) {
+      auto builder = TableMetadataBuilder::BuildFrom(metadata.get());
+      builder->AddSnapshot(std::make_shared<Snapshot>(*snapshot));
+      EXPECT_THAT(builder->Build(), HasErrorMessage("manifest-list cannot be empty"));
+    }
+    metadata->snapshots.push_back(std::make_shared<Snapshot>(*snapshot));
+    EXPECT_THAT(ToJson(*metadata), HasErrorMessage("manifest-list cannot be empty"));
+  }
+}
+
+TEST(V4TableMetadataTest, NewSnapshotsRequireRootsAndMonotonicTimestamps) {
+  auto json = MetadataJson();
+  json["snapshots"] = nlohmann::json::array({RootSnapshotJson()});
+  json["last-sequence-number"] = 1;
+  json["next-row-id"] = 10;
+  ICEBERG_UNWRAP_OR_FAIL(auto base, TableMetadataFromJson(json));
+  auto child_json = RootSnapshotJson();
+  child_json["snapshot-id"] = 2;
+  child_json["parent-snapshot-id"] = 1;
+  child_json["sequence-number"] = 2;
+  child_json["first-row-id"] = 10;
+  for (int64_t timestamp : {999, 1000, 1001}) {
+    child_json["timestamp-ms"] = timestamp;
+    ICEBERG_UNWRAP_OR_FAIL(auto child, SnapshotFromJson(child_json));
+    auto builder = TableMetadataBuilder::BuildFrom(base.get());
+    builder->AddSnapshot(std::move(child));
+    if (timestamp <= 1000) {
+      EXPECT_THAT(builder->Build(), IsError(ErrorKind::kValidationFailed));
+    } else {
+      ICEBERG_UNWRAP_OR_FAIL(auto updated, builder->Build());
+      EXPECT_EQ(updated->next_row_id, 20);
+    }
+  }
+  child_json.erase("root-manifest");
+  child_json["manifest-list"] = "legacy.avro";
+  child_json["timestamp-ms"] = 1001;
+  ICEBERG_UNWRAP_OR_FAIL(auto child, SnapshotFromJson(child_json));
+  auto builder = TableMetadataBuilder::BuildFrom(base.get());
+  builder->AddSnapshot(std::move(child));
+  EXPECT_THAT(builder->Build(),
+              HasErrorMessage("New v4 snapshots require root-manifest"));
+}
+
+TEST(V4TableMetadataTest, IgnoreLegacyIdentifiersAndRejectUnimplementedConstraints) {
+  auto json = MetadataJson();
+  json["schemas"][0]["identifier-field-ids"] = {999};
+  ICEBERG_UNWRAP_OR_FAIL(auto metadata, TableMetadataFromJson(json));
+  ICEBERG_UNWRAP_OR_FAIL(auto schema, metadata->Schema());
+  EXPECT_TRUE(schema->IdentifierFieldIds().empty());
+  ICEBERG_UNWRAP_OR_FAIL(auto serialized, ToJson(*metadata));
+  EXPECT_FALSE(serialized["schemas"][0].contains("identifier-field-ids"));
+  for (const auto* field : {"constraints", "last-constraint-id"}) {
+    json = MetadataJson();
+    json[field] = field == std::string_view("constraints")
+                      ? nlohmann::json::array({{{"constraint-id", 1}}})
+                      : nlohmann::json(1);
+    EXPECT_THAT(TableMetadataFromJson(json), IsError(ErrorKind::kNotSupported));
+  }
+  json = MetadataJson();
+  json["format-version"] = 3;
+  json["schemas"][0]["identifier-field-ids"] = {1};
+  ICEBERG_UNWRAP_OR_FAIL(auto base, TableMetadataFromJson(json));
+  auto builder = TableMetadataBuilder::BuildFrom(base.get());
+  builder->UpgradeFormatVersion(4);
+  EXPECT_THAT(builder->Build(), HasErrorMessage("primary-key constraint migration"));
+  for (const auto& malformed_schema : std::vector<nlohmann::json>{nullptr, 1, "schema"}) {
+    json = MetadataJson();
+    json["schemas"] = nlohmann::json::array({malformed_schema});
+    EXPECT_FALSE(TableMetadataFromJson(json).has_value());
+  }
+}
+
+TEST(V4TableMetadataTest, ConstraintFieldsRequireTheirDeclaredJsonTypes) {
+  auto json = MetadataJson();
+  json["constraints"] = nlohmann::json::array();
+  json["last-constraint-id"] = 0;
+  EXPECT_THAT(TableMetadataFromJson(json), IsOk());
+
+  for (const auto& value : std::vector<nlohmann::json>{nullptr, "constraints", true, 0,
+                                                       0.0, nlohmann::json::object()}) {
+    SCOPED_TRACE(value.dump());
+    json = MetadataJson();
+    json["constraints"] = value;
+    auto result = TableMetadataFromJson(json);
+    EXPECT_THAT(result, IsError(ErrorKind::kJsonParseError));
+    EXPECT_THAT(result, HasErrorMessage("constraints"));
+  }
+  for (const auto& value : std::vector<nlohmann::json>{
+           nullptr, "0", true, 0.0, 1.5, 2147483648LL, -2147483649LL,
+           18446744073709551615ULL, nlohmann::json::array(), nlohmann::json::object()}) {
+    SCOPED_TRACE(value.dump());
+    json = MetadataJson();
+    json["last-constraint-id"] = value;
+    auto result = TableMetadataFromJson(json);
+    EXPECT_THAT(result, IsError(ErrorKind::kJsonParseError));
+    EXPECT_THAT(result, HasErrorMessage("last-constraint-id"));
+  }
+}
+
+std::shared_ptr<MockFileIO> MemoryIO() {
+  auto io = std::make_shared<::testing::NiceMock<MockFileIO>>();
+  ON_CALL(*io, ReadFile)
+      .WillByDefault(
+          [ptr = io.get()](const std::string& path, std::optional<size_t> size) {
+            return ptr->FileIO::ReadFile(path, size);
+          });
+  ON_CALL(*io, WriteFile)
+      .WillByDefault([ptr = io.get()](const std::string& path, std::string_view data) {
+        return ptr->FileIO::WriteFile(path, data);
+      });
+  return io;
+}
+
+TEST(V4TableMetadataTest, CatalogCreateLoadAndPropertyUpgrade) {
+  auto io = MemoryIO();
+  ICEBERG_UNWRAP_OR_FAIL(auto catalog,
+                         InMemoryCatalog::Make("v4", io, "s3://warehouse", {}));
+  for (const auto* initial_version : {"2", "4"}) {
+    TableIdentifier ident{.name = std::string("version_") + initial_version};
+    ICEBERG_UNWRAP_OR_FAIL(
+        auto table,
+        catalog->CreateTable(ident, TestSchema(), PartitionSpec::Unpartitioned(),
+                             SortOrder::Unsorted(), "s3://warehouse/" + ident.name,
+                             {{"format-version", initial_version}}));
+    ICEBERG_UNWRAP_OR_FAIL(auto update, table->NewUpdateProperties());
+    update->Set("format-version", "4");
+    ASSERT_THAT(update->Commit(), IsOk());
+    ICEBERG_UNWRAP_OR_FAIL(auto loaded, catalog->LoadTable(ident));
+    EXPECT_EQ(loaded->metadata()->format_version, 4);
+    EXPECT_FALSE(loaded->metadata()->properties.configs().contains("format-version"));
+    ICEBERG_UNWRAP_OR_FAIL(auto append, loaded->NewFastAppend());
+    EXPECT_THAT(append->Commit(),
+                HasErrorMessage("Writing v4 manifest trees is not supported"));
+  }
+}
+
+TEST(V4TableMetadataTest, StagedCatalogCreate) {
+  auto io = MemoryIO();
+  ICEBERG_UNWRAP_OR_FAIL(auto catalog,
+                         InMemoryCatalog::Make("v4", io, "s3://warehouse", {}));
+  TableIdentifier ident{.name = "staged_v4"};
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto transaction,
+      catalog->StageCreateTable(ident, TestSchema(), PartitionSpec::Unpartitioned(),
+                                SortOrder::Unsorted(), "s3://warehouse/staged_v4",
+                                {{"format-version", "4"}}));
+  ASSERT_THAT(transaction->Commit(), IsOk());
+  ICEBERG_UNWRAP_OR_FAIL(auto loaded, catalog->LoadTable(ident));
+  EXPECT_EQ(loaded->metadata()->format_version, 4);
+}
+
+TEST(V4TableMetadataTest, RejectInvalidVersionPropertiesAndIncompleteSnapshotWrites) {
+  for (const auto* version : {"0", "-252", "5", "260", "invalid"}) {
+    EXPECT_FALSE(TableMetadata::Make(*TestSchema(), *PartitionSpec::Unpartitioned(),
+                                     *SortOrder::Unsorted(), "s3://bucket/table",
+                                     {{"format-version", version}})
+                     .has_value());
+  }
+  auto json = RootSnapshotJson();
+  json["summary"] = nlohmann::json::object();
+  EXPECT_FALSE(SnapshotFromJson(json).has_value());
+  json = RootSnapshotJson();
+  json["sequence-number"] = 0;
+  EXPECT_FALSE(SnapshotFromJson(json).has_value());
+  ICEBERG_UNWRAP_OR_FAIL(auto metadata, NewMetadata());
+  ICEBERG_UNWRAP_OR_FAIL(auto snapshot, SnapshotFromJson(RootSnapshotJson()));
+  snapshot->first_row_id.reset();
+  metadata->snapshots.push_back(std::move(snapshot));
+  EXPECT_THAT(ToJson(*metadata), HasErrorMessage("first-row-id"));
+}
+
+TEST(V4TableMetadataTest, MissingOrNullUuidUsesTheDefaultForEveryVersion) {
+  for (int version : {1, 2, 3, 4}) {
+    auto json = MetadataJson();
+    json["format-version"] = version;
+    for (bool explicit_null : {false, true}) {
+      if (explicit_null) {
+        json["table-uuid"] = nullptr;
+      } else {
+        json.erase("table-uuid");
+      }
+      ICEBERG_UNWRAP_OR_FAIL(auto metadata, TableMetadataFromJson(json));
+      EXPECT_TRUE(metadata->table_uuid.empty());
+    }
+  }
+}
+
+TEST(V4TableMetadataTest, NullLocationIsAbsentAndCannotProduceDefaultDataPaths) {
+  auto json = MetadataJson();
+  json["location"] = nullptr;
+  ICEBERG_UNWRAP_OR_FAIL(auto metadata, TableMetadataFromJson(json));
+  EXPECT_TRUE(metadata->location.empty());
+  ICEBERG_UNWRAP_OR_FAIL(auto serialized, ToJson(*metadata));
+  EXPECT_FALSE(serialized.contains("location"));
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto table, Table::Make(TableIdentifier{.name = "unlocated"}, std::move(metadata),
+                              "s3://bucket/metadata.json", MemoryIO(),
+                              std::make_shared<::testing::NiceMock<MockCatalog>>()));
+  EXPECT_THAT(table->location_provider(), IsError(ErrorKind::kInvalidArgument));
+  for (int version : {1, 2, 3}) {
+    json["format-version"] = version;
+    EXPECT_FALSE(TableMetadataFromJson(json).has_value());
+  }
+}
+
+TEST(V4TableMetadataTest, UpgradeBeforeSelectingCurrentSchema) {
+  ICEBERG_UNWRAP_OR_FAIL(auto metadata, NewMetadata(3));
+  metadata->current_schema_id = -1;
+  auto builder = TableMetadataBuilder::BuildFrom(metadata.get());
+  builder->AddSchema(TestSchema()).UpgradeFormatVersion(4).SetCurrentSchema(0);
+  ICEBERG_UNWRAP_OR_FAIL(auto upgraded, builder->Build());
+  EXPECT_EQ(upgraded->format_version, 4);
+  EXPECT_EQ(upgraded->current_schema_id, 0);
+}
+
+TEST(V4TableMetadataTest, RootOperationValidationIsSharedAcrossMetadataPaths) {
+  for (const auto& operation :
+       std::vector<std::optional<std::string>>{std::nullopt, "", "invalid"}) {
+    auto json = RootSnapshotJson();
+    if (operation.has_value()) {
+      json["summary"]["operation"] = operation.value();
+    } else {
+      json["summary"].erase("operation");
+    }
+    EXPECT_THAT(SnapshotFromJson(json), HasErrorMessage("valid operation"));
+
+    ICEBERG_UNWRAP_OR_FAIL(auto snapshot, SnapshotFromJson(RootSnapshotJson()));
+    if (operation.has_value()) {
+      snapshot->summary["operation"] = operation.value();
+    } else {
+      snapshot->summary.erase("operation");
+    }
+    EXPECT_THAT(snapshot->ValidateForFormatVersion(4),
+                IsError(ErrorKind::kValidationFailed));
+    ICEBERG_UNWRAP_OR_FAIL(auto base, NewMetadata());
+    auto builder = TableMetadataBuilder::BuildFrom(base.get());
+    builder->AddSnapshot(std::make_shared<Snapshot>(*snapshot));
+    EXPECT_THAT(builder->Build(), HasErrorMessage("valid operation"));
+    base->snapshots.push_back(std::move(snapshot));
+    EXPECT_THAT(ToJson(*base), HasErrorMessage("valid operation"));
+  }
+}
+
+}  // namespace
+}  // namespace iceberg
diff --git a/src/iceberg/test/v4_types_test.cc b/src/iceberg/test/v4_types_test.cc
new file mode 100644
index 000000000..82adce9d3
--- /dev/null
+++ b/src/iceberg/test/v4_types_test.cc
@@ -0,0 +1,185 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/manifest/v4_types.h"
+
+#include <cstdint>
+#include <limits>
+#include <memory>
+#include <vector>
+
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+#include <nlohmann/json.hpp>
+
+#include "iceberg/json_serde_internal.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/test/matchers.h"
+#include "iceberg/type.h"
+
+namespace iceberg {
+namespace {
+
+// Independent wire-schema fixtures from the v4 manifest schema in the spec.
+// Check IDs, names, order, types, and requiredness through the schema serializer.
+void CheckSchema(const StructType& type, const nlohmann::json& expected) {
+  ICEBERG_UNWRAP_OR_FAIL(auto actual, ToJson(type));
+  EXPECT_EQ(actual, expected);
+  ICEBERG_UNWRAP_OR_FAIL(auto parsed, TypeFromJson(expected));
+  EXPECT_EQ(type, *parsed);
+}
+
+TEST(V4TypesTest, FileTrackingSchema) {
+  const auto expected = nlohmann::json::parse(R"({
+    "type": "struct",
+    "fields": [
+      {"id": 0, "name": "status", "type": "int", "required": true},
+      {"id": 1, "name": "snapshot_id", "type": "long", "required": false},
+      {"id": 5, "name": "modified_snapshot_id", "type": "long", "required": false},
+      {"id": 3, "name": "sequence_number", "type": "long", "required": false},
+      {"id": 4, "name": "file_sequence_number", "type": "long", "required": false},
+      {"id": 142, "name": "first_row_id", "type": "long", "required": false},
+      {"id": 6, "name": "deleted_positions", "type": "binary", "required": false},
+      {"id": 7, "name": "replaced_positions", "type": "binary", "required": false}
+    ]
+  })");
+  auto type = FileTracking::Type();
+  EXPECT_EQ(type, FileTracking::Type());
+  CheckSchema(*type, expected);
+}
+
+TEST(V4TypesTest, DeletionVectorMetadataSchema) {
+  const auto expected = nlohmann::json::parse(R"({
+    "type": "struct",
+    "fields": [
+      {"id": 155, "name": "location", "type": "string", "required": true},
+      {"id": 144, "name": "offset", "type": "long", "required": true},
+      {"id": 145, "name": "size_in_bytes", "type": "long", "required": true},
+      {"id": 156, "name": "cardinality", "type": "long", "required": true},
+      {"id": 149, "name": "key_metadata", "type": "binary", "required": false}
+    ]
+  })");
+  auto type = DeletionVectorMetadata::Type();
+  EXPECT_EQ(type, DeletionVectorMetadata::Type());
+  CheckSchema(*type, expected);
+}
+
+TEST(V4TypesTest, ManifestInfoSchema) {
+  const auto expected = nlohmann::json::parse(R"({
+    "type": "struct",
+    "fields": [
+      {"id": 521, "name": "format_version", "type": "int", "required": true},
+      {"id": 522, "name": "dv", "type": "binary", "required": false},
+      {"id": 504, "name": "added_files_count", "type": "int", "required": true},
+      {"id": 505, "name": "existing_files_count", "type": "int", "required": true},
+      {"id": 525, "name": "modified_files_count", "type": "int", "required": true},
+      {"id": 506, "name": "deleted_files_count", "type": "int", "required": true},
+      {"id": 523, "name": "replaced_files_count", "type": "int", "required": true},
+      {"id": 512, "name": "added_rows_count", "type": "long", "required": true},
+      {"id": 513, "name": "existing_rows_count", "type": "long", "required": true},
+      {"id": 526, "name": "modified_rows_count", "type": "long", "required": true},
+      {"id": 514, "name": "deleted_rows_count", "type": "long", "required": true},
+      {"id": 524, "name": "replaced_rows_count", "type": "long", "required": true},
+      {"id": 516, "name": "min_sequence_number", "type": "long", "required": true}
+    ]
+  })");
+  auto type = ManifestInfo::Type();
+  EXPECT_EQ(type, ManifestInfo::Type());
+  CheckSchema(*type, expected);
+}
+
+struct TrackingStatusCase {
+  int32_t value;
+  FileTrackingStatus status;
+  bool alive;
+};
+
+class FileTrackingStatusTest : public ::testing::TestWithParam<TrackingStatusCase> {};
+
+TEST_P(FileTrackingStatusTest, DecodeAndPlanLiveEntries) {
+  const auto& test_case = GetParam();
+  ICEBERG_UNWRAP_OR_FAIL(auto status, FileTrackingStatusFromInt(test_case.value));
+  EXPECT_EQ(status, test_case.status);
+  EXPECT_EQ(static_cast<int32_t>(status), test_case.value);
+  EXPECT_EQ(FileTracking{.status = status}.IsAlive(), test_case.alive);
+}
+
+INSTANTIATE_TEST_SUITE_P(
+    AllStatuses, FileTrackingStatusTest,
+    ::testing::Values(TrackingStatusCase{0, FileTrackingStatus::kExisting, true},
+                      TrackingStatusCase{1, FileTrackingStatus::kAdded, true},
+                      TrackingStatusCase{2, FileTrackingStatus::kDeleted, false},
+                      TrackingStatusCase{3, FileTrackingStatus::kReplaced, false},
+                      TrackingStatusCase{4, FileTrackingStatus::kModified, true}));
+
+TEST(V4TypesTest, InvalidTrackingStatuses) {
+  for (int32_t value : {-1, 5, std::numeric_limits<int32_t>::min(),
+                        std::numeric_limits<int32_t>::max()}) {
+    SCOPED_TRACE(value);
+    EXPECT_THAT(FileTrackingStatusFromInt(value), IsError(ErrorKind::kInvalidArgument));
+  }
+}
+
+TEST(V4TypesTest, LegacyStatusDecoderRejectsV4Statuses) {
+  for (int32_t value : {0, 1, 2}) {
+    EXPECT_THAT(ManifestStatusFromInt(value), IsOk());
+  }
+  for (int32_t value : {3, 4}) {
+    EXPECT_THAT(ManifestStatusFromInt(value), IsError(ErrorKind::kInvalidArgument));
+  }
+}
+
+TEST(V4TypesTest, TrackingPreservesMissingInheritanceFieldsAndBitmaps) {
+  FileTracking tracking;
+  EXPECT_FALSE(tracking.snapshot_id.has_value());
+  EXPECT_FALSE(tracking.modified_snapshot_id.has_value());
+  EXPECT_FALSE(tracking.sequence_number.has_value());
+  EXPECT_FALSE(tracking.file_sequence_number.has_value());
+  EXPECT_FALSE(tracking.first_row_id.has_value());
+  EXPECT_FALSE(tracking.deleted_positions.has_value());
+  EXPECT_FALSE(tracking.replaced_positions.has_value());
+
+  // Missing sequence numbers must remain distinguishable from sequence number 0.
+  tracking.sequence_number = 0;
+  EXPECT_TRUE(tracking.sequence_number.has_value());
+  EXPECT_EQ(tracking.sequence_number.value(), 0);
+
+  tracking.deleted_positions = std::vector<uint8_t>{};
+  const auto copy = tracking;
+  ASSERT_TRUE(copy.deleted_positions.has_value());
+  EXPECT_TRUE(copy.deleted_positions->empty());
+  EXPECT_FALSE(copy.replaced_positions.has_value());
+}
+
+TEST(V4TypesTest, OptionalBinaryMetadataPreservesAbsence) {
+  DeletionVectorMetadata dv;
+  ManifestInfo manifest;
+  EXPECT_FALSE(dv.key_metadata.has_value());
+  EXPECT_FALSE(manifest.dv.has_value());
+  dv.key_metadata = std::vector<uint8_t>{0, 255};
+  manifest.dv = std::vector<uint8_t>{};
+  const auto dv_copy = dv;
+  const auto manifest_copy = manifest;
+  EXPECT_EQ(dv_copy.key_metadata, dv.key_metadata);
+  ASSERT_TRUE(manifest_copy.dv.has_value());
+  EXPECT_TRUE(manifest_copy.dv->empty());
+}
+
+}  // namespace
+}  // namespace iceberg
diff --git a/src/iceberg/update/expire_snapshots.cc b/src/iceberg/update/expire_snapshots.cc
index fd14308e8..161724f73 100644
--- a/src/iceberg/update/expire_snapshots.cc
+++ b/src/iceberg/update/expire_snapshots.cc
@@ -62,6 +62,12 @@ Result<std::unique_ptr<ManifestReader>> MakeManifestReader(
   return ManifestReader::Make(manifest, file_io, std::move(schema), specs_by_id.get());
 }
 
+bool HasV4RootManifest(const TableMetadata& metadata) {
+  return std::ranges::any_of(metadata.snapshots, [](const auto& snapshot) {
+    return snapshot && snapshot->root_manifest.has_value();
+  });
+}
+
 /// \brief Abstract strategy for cleaning up files after snapshot expiration.
 class FileCleanupStrategy {
  public:
@@ -966,6 +972,13 @@ Result<ExpireSnapshots::ApplyResult> ExpireSnapshots::Apply() {
     }
   });
 
+  if (cleanup_level_ != CleanupLevel::kNone && !result.snapshot_ids_to_remove.empty() &&
+      HasV4RootManifest(base)) {
+    return NotSupported(
+        "Snapshot expiration file cleanup with v4 root manifests "
+        "is not supported yet; use CleanupLevel::kNone");
+  }
+
   if (clean_expired_metadata_) {
     ICEBERG_ASSIGN_OR_RAISE(
         auto reachable_ids,
@@ -1026,6 +1039,14 @@ Status ExpireSnapshots::Finalize(const TableMetadata& metadata_after_expiration)
   const TableMetadata& metadata_before_expiration =
       *apply_result->metadata_before_expiration;
 
+  // Later transaction updates or the catalog's committed result may introduce
+  // root manifests after Apply(). Never treat untraversable roots as empty.
+  // The commit has succeeded, so leave files in place without a cleanup error.
+  if (HasV4RootManifest(metadata_before_expiration) ||
+      HasV4RootManifest(metadata_after_expiration)) {
+    return {};
+  }
+
   // Pick incremental cleanup when the expiration is a simple linear-ancestry walk:
   // no explicit snapshot IDs, no removed snapshots outside main ancestry, and no
   // retained snapshots outside main ancestry.
diff --git a/src/iceberg/update/snapshot_update.cc b/src/iceberg/update/snapshot_update.cc
index d6a1066ac..f2a795f4b 100644
--- a/src/iceberg/update/snapshot_update.cc
+++ b/src/iceberg/update/snapshot_update.cc
@@ -311,6 +311,9 @@ int64_t SnapshotUpdate::SnapshotId() {
 Result<SnapshotUpdate::ApplyResult> SnapshotUpdate::Apply() {
   commit_metrics_->attempts->Increment();
   ICEBERG_RETURN_UNEXPECTED(CheckErrors());
+  if (base().format_version >= TableMetadata::kMinFormatVersionRootManifest) {
+    return NotSupported("Writing v4 manifest trees is not supported yet");
+  }
 
   ICEBERG_ASSIGN_OR_RAISE(auto parent_snapshot,
                           SnapshotUtil::OptionalLatestSnapshot(base(), target_branch_));
diff --git a/src/iceberg/update/update_properties.cc b/src/iceberg/update/update_properties.cc
index c97175569..b42a8806d 100644
--- a/src/iceberg/update/update_properties.cc
+++ b/src/iceberg/update/update_properties.cc
@@ -88,7 +88,8 @@ Result<UpdateProperties::ApplyResult> UpdateProperties::Apply() {
     ICEBERG_ASSIGN_OR_RAISE(auto parsed_version,
                             StringUtils::ParseNumber<int32_t>(iter->second));
 
-    if (parsed_version > TableMetadata::kSupportedTableFormatVersion) {
+    if (parsed_version < 1 ||
+        parsed_version > TableMetadata::kSupportedTableFormatVersion) {
       return InvalidArgument(
           "Cannot upgrade table to unsupported format version: v{} (supported: v{})",
           parsed_version, TableMetadata::kSupportedTableFormatVersion);