diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt
index 4cde4fb23..030e7a284 100644
--- a/src/iceberg/CMakeLists.txt
+++ b/src/iceberg/CMakeLists.txt
@@ -116,6 +116,7 @@ set(ICEBERG_SOURCES
     sort_field.cc
     sort_order.cc
     statistics_file.cc
+    stats/content_stats_util.cc
     table.cc
     table_identifier.cc
     table_metadata.cc
@@ -302,6 +303,7 @@ add_subdirectory(expression)
 add_subdirectory(manifest)
 add_subdirectory(puffin)
 add_subdirectory(row)
+add_subdirectory(stats)
 add_subdirectory(update)
 add_subdirectory(util)
 add_subdirectory(inspect)
diff --git a/src/iceberg/cpp20_compatibility_internal.h b/src/iceberg/cpp20_compatibility_internal.h
index d9c64d36a..3dbcaae0b 100644
--- a/src/iceberg/cpp20_compatibility_internal.h
+++ b/src/iceberg/cpp20_compatibility_internal.h
@@ -135,6 +135,7 @@
 #include "iceberg/sort_field.h"
 #include "iceberg/sort_order.h"
 #include "iceberg/statistics_file.h"
+#include "iceberg/stats/content_stats_util.h"
 #include "iceberg/storage_credential.h"
 #include "iceberg/table.h"
 #include "iceberg/table_identifier.h"
diff --git a/src/iceberg/stats/CMakeLists.txt b/src/iceberg/stats/CMakeLists.txt
new file mode 100644
index 000000000..7f9f4cd75
--- /dev/null
+++ b/src/iceberg/stats/CMakeLists.txt
@@ -0,0 +1,18 @@
+# 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.
+
+iceberg_install_all_headers(iceberg/stats)
diff --git a/src/iceberg/stats/content_stats_util.cc b/src/iceberg/stats/content_stats_util.cc
new file mode 100644
index 000000000..3c4413be5
--- /dev/null
+++ b/src/iceberg/stats/content_stats_util.cc
@@ -0,0 +1,278 @@
+/*
+ * 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/stats/content_stats_util.h"
+
+#include <algorithm>
+#include <functional>
+#include <string>
+#include <unordered_set>
+#include <utility>
+#include <vector>
+
+#include "iceberg/metadata_columns.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_field.h"
+#include "iceberg/type.h"
+#include "iceberg/util/checked_cast.h"
+
+namespace iceberg {
+
+namespace {
+
+using Util = ContentStatsUtil;
+
+/// \brief Chooses the metrics mode (which stats to include, see MetricsMode in
+/// metrics_config.h) for a table field, given its ID and canonical dotted name.
+/// std::nullopt leaves the field out of `content_stats`.
+///
+/// This is the only difference between the write and read types: StatsWriteType
+/// looks the mode up in the table's MetricsConfig, and StatsReadType uses `full` for
+/// the requested field IDs and skips the others.
+using ModeFn =
+    std::function<std::optional<MetricsMode>(int32_t field_id, std::string_view name)>;
+
+bool IsValidStatsId(int32_t stats_id) {
+  return stats_id >= Util::kMetadataStatsRangeStart &&
+         stats_id < Util::kContentStatsRangeEnd;
+}
+
+bool InBaseIdRange(int32_t base_id, int32_t stats_id) {
+  return stats_id >= base_id && stats_id < base_id + Util::kNumReservedFieldStatsIds;
+}
+
+bool IsFloatingPoint(const Type& type) {
+  return type.type_id() == TypeId::kFloat || type.type_id() == TypeId::kDouble;
+}
+
+bool IsGeo(const Type& type) {
+  return type.type_id() == TypeId::kGeometry || type.type_id() == TypeId::kGeography;
+}
+
+bool IsVariableLength(const Type& type) {
+  return type.is_variant() || type.type_id() == TypeId::kString ||
+         type.type_id() == TypeId::kBinary || IsGeo(type);
+}
+
+bool HasBounds(MetricsMode mode) {
+  return mode.kind == MetricsMode::Kind::kTruncate ||
+         mode.kind == MetricsMode::Kind::kFull;
+}
+
+std::shared_ptr<StructType> GeoLowerType(int32_t base_id) {
+  return struct_({
+      SchemaField::MakeRequired(base_id + Util::kGeoLowerXOffset, "x", float64(),
+                                "Bounding box westernmost/xmin; [-180..180]"),
+      SchemaField::MakeRequired(base_id + Util::kGeoLowerYOffset, "y", float64(),
+                                "Bounding box southernmost/ymin; [-90..90]"),
+      SchemaField::MakeOptional(base_id + Util::kGeoLowerZOffset, "z", float64(),
+                                "Bounding box zmin"),
+      SchemaField::MakeOptional(base_id + Util::kGeoLowerMOffset, "m", float64(),
+                                "Bounding box mmin"),
+  });
+}
+
+std::shared_ptr<StructType> GeoUpperType(int32_t base_id) {
+  return struct_({
+      SchemaField::MakeRequired(base_id + Util::kGeoUpperXOffset, "x", float64(),
+                                "Bounding box easternmost/xmax; [-180..180]"),
+      SchemaField::MakeRequired(base_id + Util::kGeoUpperYOffset, "y", float64(),
+                                "Bounding box northernmost/ymax; [-90..90]"),
+      SchemaField::MakeOptional(base_id + Util::kGeoUpperZOffset, "z", float64(),
+                                "Bounding box zmax"),
+      SchemaField::MakeOptional(base_id + Util::kGeoUpperMOffset, "m", float64(),
+                                "Bounding box mmax"),
+  });
+}
+
+/// \brief Build a field stats struct; see ContentStatsUtil::FieldStatsType.
+std::shared_ptr<StructType> MakeFieldStatsType(const std::shared_ptr<Type>& type,
+                                               bool with_null_count, int32_t base_id,
+                                               MetricsMode mode) {
+  if (mode.kind == MetricsMode::Kind::kNone || type == nullptr || type->is_nested()) {
+    return nullptr;
+  }
+
+  std::vector<SchemaField> fields;
+
+  if (HasBounds(mode)) {
+    std::shared_ptr<Type> lower_type = type;
+    std::shared_ptr<Type> upper_type = type;
+    if (IsGeo(*type)) {
+      lower_type = GeoLowerType(base_id);
+      upper_type = GeoUpperType(base_id);
+    }
+    fields.push_back(SchemaField::MakeOptional(
+        base_id + Util::kLowerBoundOffset, Util::kLowerBoundName, std::move(lower_type)));
+    fields.push_back(SchemaField::MakeOptional(
+        base_id + Util::kUpperBoundOffset, Util::kUpperBoundName, std::move(upper_type)));
+  }
+
+  fields.push_back(
+      SchemaField::MakeOptional(base_id + Util::kValueCountOffset, Util::kValueCountName,
+                                int64(), "Number of values (including null and NaN)"));
+
+  if (with_null_count) {
+    fields.push_back(SchemaField::MakeOptional(base_id + Util::kNullValueCountOffset,
+                                               Util::kNullValueCountName, int64(),
+                                               "Number of null values"));
+  }
+
+  if (IsFloatingPoint(*type)) {
+    fields.push_back(SchemaField::MakeOptional(base_id + Util::kNanValueCountOffset,
+                                               Util::kNanValueCountName, int64(),
+                                               "Number of NaN values"));
+  }
+
+  if (IsVariableLength(*type)) {
+    fields.push_back(SchemaField::MakeOptional(base_id + Util::kTotalBytesOffset,
+                                               Util::kTotalBytesName, int64()));
+  }
+
+  return struct_(std::move(fields));
+}
+
+/// \brief Add stats structs for the fields of a struct that are not in a list or map.
+///
+/// \param dotted_prefix prefix of the canonical name, used to look up metrics modes
+/// \param stats_prefix prefix of the stats struct name; names are joined with '_'
+void CollectFieldStats(const StructType& struct_type, const std::string& dotted_prefix,
+                       const std::string& stats_prefix, const ModeFn& mode_fn,
+                       bool for_read, std::vector<SchemaField>& stats_fields) {
+  for (const auto& field : struct_type.fields()) {
+    std::string dotted_name = dotted_prefix + std::string(field.name());
+    std::string stats_name = stats_prefix + std::string(field.name());
+    const auto& type = *field.type();
+
+    if (type.type_id() == TypeId::kStruct) {
+      CollectFieldStats(internal::checked_cast<const StructType&>(type),
+                        dotted_name + ".", stats_name + "_", mode_fn, for_read,
+                        stats_fields);
+      continue;
+    }
+
+    // lists and maps (and anything in them) may hold many values per row: no stats
+    if (type.is_nested()) {
+      continue;
+    }
+
+    auto base_id = Util::BaseId(field.field_id());
+    if (!base_id.has_value()) {
+      continue;
+    }
+
+    auto mode = mode_fn(field.field_id(), dotted_name);
+    if (!mode.has_value()) {
+      continue;
+    }
+
+    // Reads always project null_value_count: other writers (Java, since
+    // apache/iceberg#17413) also write it for required fields, and a missing optional
+    // stat reads as unknown.
+    bool with_null_count = for_read || field.optional();
+    if (auto field_stats = MakeFieldStatsType(field.type(), with_null_count,
+                                              base_id.value(), mode.value())) {
+      stats_fields.push_back(
+          SchemaField::MakeOptional(base_id.value(), stats_name, std::move(field_stats)));
+    }
+  }
+}
+
+std::shared_ptr<StructType> BuildStatsType(const Schema& schema, const ModeFn& mode_fn,
+                                           bool for_read) {
+  std::vector<SchemaField> stats_fields;
+  CollectFieldStats(schema, /*dotted_prefix=*/"", /*stats_prefix=*/"", mode_fn, for_read,
+                    stats_fields);
+  std::ranges::sort(stats_fields, std::less<>{}, &SchemaField::field_id);
+  return struct_(std::move(stats_fields));
+}
+
+}  // namespace
+
+std::optional<int32_t> ContentStatsUtil::BaseId(int32_t field_id) {
+  if (field_id >= 0 && field_id < kDataFieldIdEnd) {
+    return kContentStatsRangeStart + field_id * kNumReservedFieldStatsIds;
+  }
+  if (field_id == MetadataColumns::kLastUpdatedSequenceNumberColumnId) {
+    return kLastUpdatedSequenceNumberBaseId;
+  }
+  if (field_id == MetadataColumns::kRowIdColumnId) {
+    return kRowIdBaseId;
+  }
+  return std::nullopt;
+}
+
+Result<int32_t> ContentStatsUtil::FieldId(int32_t stats_id) {
+  if (!IsValidStatsId(stats_id)) {
+    return InvalidArgument("Invalid stats field ID: {}", stats_id);
+  }
+
+  if (stats_id < kContentStatsRangeStart) {
+    if (InBaseIdRange(kLastUpdatedSequenceNumberBaseId, stats_id)) {
+      return MetadataColumns::kLastUpdatedSequenceNumberColumnId;
+    }
+    if (InBaseIdRange(kRowIdBaseId, stats_id)) {
+      return MetadataColumns::kRowIdColumnId;
+    }
+    return NotSupported("Unsupported metadata stats field ID: {}", stats_id);
+  }
+
+  return (stats_id - kContentStatsRangeStart) / kNumReservedFieldStatsIds;
+}
+
+Result<int32_t> ContentStatsUtil::StatOffset(int32_t stats_id) {
+  if (!IsValidStatsId(stats_id)) {
+    return InvalidArgument("Invalid stats field ID: {}", stats_id);
+  }
+  return stats_id % kNumReservedFieldStatsIds;
+}
+
+std::shared_ptr<StructType> ContentStatsUtil::FieldStatsType(const SchemaField& field,
+                                                             int32_t base_id,
+                                                             MetricsMode mode) {
+  // the spec includes null_value_count for optional fields only
+  return MakeFieldStatsType(field.type(), /*with_null_count=*/field.optional(), base_id,
+                            mode);
+}
+
+std::shared_ptr<StructType> ContentStatsUtil::StatsWriteType(
+    const Schema& schema, const MetricsConfig& metrics_config) {
+  return BuildStatsType(
+      schema,
+      [&metrics_config](int32_t /*field_id*/, std::string_view name) {
+        return std::optional<MetricsMode>(metrics_config.ColumnMode(name));
+      },
+      /*for_read=*/false);
+}
+
+std::shared_ptr<StructType> ContentStatsUtil::StatsReadType(
+    const Schema& schema, std::span<const int32_t> field_ids) {
+  std::unordered_set<int32_t> ids(field_ids.begin(), field_ids.end());
+  return BuildStatsType(
+      schema,
+      [&ids](int32_t field_id, std::string_view) -> std::optional<MetricsMode> {
+        if (ids.contains(field_id)) {
+          return MetricsMode::Full();
+        }
+        return std::nullopt;
+      },
+      /*for_read=*/true);
+}
+
+}  // namespace iceberg
diff --git a/src/iceberg/stats/content_stats_util.h b/src/iceberg/stats/content_stats_util.h
new file mode 100644
index 000000000..896d57fcc
--- /dev/null
+++ b/src/iceberg/stats/content_stats_util.h
@@ -0,0 +1,174 @@
+/*
+ * 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/stats/content_stats_util.h
+/// \brief Field IDs and types of the v4 `content_stats` struct.
+///
+/// See the "Content Stats" section of the Iceberg table spec.
+
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <span>
+#include <string_view>
+
+#include "iceberg/iceberg_export.h"
+#include "iceberg/metrics_config.h"
+#include "iceberg/result.h"
+#include "iceberg/type_fwd.h"
+
+namespace iceberg {
+
+/// \brief Utilities to derive the v4 `content_stats` struct from a table schema.
+///
+/// In v4, a manifest entry's column stats are one struct, `content_stats`, with an
+/// optional *field stats struct* per table field (e.g. `lower_bound`, `value_count`).
+/// The field IDs inside `content_stats` are computed from the table field IDs. The
+/// terms below are the spec's ("Content Stats" section):
+///
+/// - **stats ID**: the field ID of any field inside `content_stats`, either a field
+///   stats struct or one of its stats. Stats IDs are a separate ID space from table
+///   field IDs, reserved in [9_000, 200_000_000).
+/// - **base-id**: the first stats ID of a table field's range of 200 stats IDs,
+///   `10_000 + 200 * field-id`. The field stats struct itself uses the base-id.
+///   Reserved metadata fields use fixed base-ids (9000, 9200).
+/// - **offset**: identifies a stat within the range, stats ID = base-id + offset
+///   (1 `lower_bound`, 2 `upper_bound`, 3 `value_count`, ...).
+///
+/// For example, stats for table field 2 use stats IDs [10_400, 10_599]: its stats
+/// struct has ID 10_400, and its `lower_bound` (offset 1) has ID 10_401.
+struct ICEBERG_EXPORT ContentStatsUtil {
+  /// \brief Number of stats IDs reserved for each table field.
+  static constexpr int32_t kNumReservedFieldStatsIds = 200;
+  /// \brief First stats ID used by metadata fields (inclusive).
+  static constexpr int32_t kMetadataStatsRangeStart = 9'000;
+  /// \brief First stats ID used by data fields (inclusive).
+  static constexpr int32_t kContentStatsRangeStart = 10'000;
+  /// \brief End of the stats ID range (exclusive).
+  static constexpr int32_t kContentStatsRangeEnd = 200'000'000;
+  /// \brief End of the data field IDs that can have stats (exclusive).
+  static constexpr int32_t kDataFieldIdEnd = 999'950;
+
+  /// \brief Base ID of `_last_updated_sequence_number` stats.
+  static constexpr int32_t kLastUpdatedSequenceNumberBaseId = 9'000;
+  /// \brief Base ID of `_row_id` stats.
+  static constexpr int32_t kRowIdBaseId = 9'200;
+
+  /// \brief Offsets of the stats within a field stats struct.
+  static constexpr int32_t kLowerBoundOffset = 1;
+  static constexpr int32_t kUpperBoundOffset = 2;
+  static constexpr int32_t kValueCountOffset = 3;
+  static constexpr int32_t kNullValueCountOffset = 4;
+  static constexpr int32_t kNanValueCountOffset = 5;
+  static constexpr int32_t kTotalBytesOffset = 6;
+
+  /// \brief Offsets of the `geo_lower` fields (x, y, z, m).
+  static constexpr int32_t kGeoLowerXOffset = 10;
+  static constexpr int32_t kGeoLowerYOffset = 11;
+  static constexpr int32_t kGeoLowerZOffset = 12;
+  static constexpr int32_t kGeoLowerMOffset = 13;
+
+  /// \brief Offsets of the `geo_upper` fields (x, y, z, m).
+  static constexpr int32_t kGeoUpperXOffset = 14;
+  static constexpr int32_t kGeoUpperYOffset = 15;
+  static constexpr int32_t kGeoUpperZOffset = 16;
+  static constexpr int32_t kGeoUpperMOffset = 17;
+
+  static constexpr std::string_view kLowerBoundName = "lower_bound";
+  static constexpr std::string_view kUpperBoundName = "upper_bound";
+  static constexpr std::string_view kValueCountName = "value_count";
+  static constexpr std::string_view kNullValueCountName = "null_value_count";
+  static constexpr std::string_view kNanValueCountName = "nan_value_count";
+  static constexpr std::string_view kTotalBytesName = "total_bytes";
+
+  /// \brief Return the base-id of a table field: the stats ID of its field stats
+  /// struct, and the start of its stats ID range.
+  ///
+  /// \param field_id a table field ID
+  /// \return the base-id, or std::nullopt if stats for the field cannot be stored (the
+  /// range would fall outside the stats ID range, or the field is a metadata field
+  /// other than `_last_updated_sequence_number` and `_row_id`)
+  static std::optional<int32_t> BaseId(int32_t field_id);
+
+  /// \brief Return the table field ID that a stats ID belongs to (the inverse of
+  /// BaseId plus an offset).
+  ///
+  /// \param stats_id a stats ID: the field ID of a field stats struct, or of a stat in
+  /// one
+  /// \return the table field ID; InvalidArgument if the ID is outside the stats range,
+  /// or NotSupported if it is in the metadata range but not used by a reserved field
+  static Result<int32_t> FieldId(int32_t stats_id);
+
+  /// \brief Return the offset of a stats ID within its field stats struct.
+  ///
+  /// \param stats_id a stats ID
+  /// \return the offset, 0 for the ID of the stats struct itself; InvalidArgument if
+  /// the ID is outside the stats range
+  static Result<int32_t> StatOffset(int32_t stats_id);
+
+  /// \brief Build the stats struct of one field for a metrics mode, for writing.
+  ///
+  /// The metrics mode is the table's per-column setting for which stats writers
+  /// collect (`write.metadata.metrics.default` and
+  /// `write.metadata.metrics.column.<name>`, see MetricsConfig):
+  /// - `none`: no stats, so no stats struct;
+  /// - `counts`: `value_count`, `null_value_count`, `nan_value_count`, `total_bytes`;
+  /// - `truncate(N)` and `full`: counts plus `lower_bound` and `upper_bound`. Truncation
+  ///   shortens bound values but does not change their type.
+  ///
+  /// `null_value_count` is included only for optional fields, as the spec says.
+  ///
+  /// \param field a primitive or variant table field
+  /// \param base_id the base ID of the field's stats struct
+  /// \param mode the metrics mode of the field
+  /// \return the stats struct, or nullptr if the field has no stats (mode `none`, or a
+  /// nested type)
+  static std::shared_ptr<StructType> FieldStatsType(const SchemaField& field,
+                                                    int32_t base_id, MetricsMode mode);
+
+  /// \brief Build the `content_stats` struct a writer puts in manifests for a table.
+  ///
+  /// This is the exact type that is written: which stats are present follows the
+  /// table's metrics config, so it differs between tables and can change over time.
+  /// It is the type of `tracked_file.content_stats` when manifests are written.
+  ///
+  /// It has a stats struct for every field that is not nested, is not in a list or a
+  /// map, can be stored in the stats ID range, and whose metrics mode is not `none`.
+  /// Fields are ordered by stats ID.
+  static std::shared_ptr<StructType> StatsWriteType(const Schema& schema,
+                                                    const MetricsConfig& metrics_config);
+
+  /// \brief Build the `content_stats` projection a reader requests for some fields.
+  ///
+  /// A reader doesn't know the metrics config a file was written with, so it asks for
+  /// every stat that any writer may have written for the requested fields, and only for
+  /// those fields (e.g. the columns of a scan filter). Readers match by stats ID, and a
+  /// stat missing from a file reads as null, meaning unknown.
+  ///
+  /// So it uses the same rules as StatsWriteType with mode `full` for every requested
+  /// field, and it also includes `null_value_count` for required fields: the spec
+  /// writes it for optional fields only, but Java writes it for all fields. IDs of
+  /// fields that are missing from the schema or have no stats are ignored.
+  static std::shared_ptr<StructType> StatsReadType(const Schema& schema,
+                                                   std::span<const int32_t> field_ids);
+};
+
+}  // namespace iceberg
diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt
index 1874fc25e..25997a9a3 100644
--- a/src/iceberg/test/CMakeLists.txt
+++ b/src/iceberg/test/CMakeLists.txt
@@ -71,6 +71,7 @@ endfunction()
 add_iceberg_test(schema_test
                  SOURCES
                  assign_id_visitor_test.cc
+                 content_stats_util_test.cc
                  name_mapping_test.cc
                  partition_field_test.cc
                  partition_spec_test.cc
diff --git a/src/iceberg/test/content_stats_util_test.cc b/src/iceberg/test/content_stats_util_test.cc
new file mode 100644
index 000000000..046cd0856
--- /dev/null
+++ b/src/iceberg/test/content_stats_util_test.cc
@@ -0,0 +1,555 @@
+/*
+ * 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/stats/content_stats_util.h"
+
+#include <memory>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+
+#include "iceberg/metadata_columns.h"
+#include "iceberg/metrics_config.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_field.h"
+#include "iceberg/schema_util.h"
+#include "iceberg/test/matchers.h"
+#include "iceberg/type.h"
+
+namespace iceberg {
+
+namespace {
+
+using Util = ContentStatsUtil;
+
+std::shared_ptr<MetricsConfig> MakeConfig(
+    std::unordered_map<std::string, std::string> properties) {
+  auto config = MetricsConfig::Make(std::move(properties));
+  EXPECT_THAT(config, IsOk());
+  return config.value();
+}
+
+std::shared_ptr<MetricsConfig> ConfigWithDefault(const std::string& mode) {
+  return MakeConfig({{"write.metadata.metrics.default", mode}});
+}
+
+std::vector<int32_t> FieldIds(const StructType& type) {
+  std::vector<int32_t> ids;
+  for (const auto& field : type.fields()) {
+    ids.push_back(field.field_id());
+  }
+  return ids;
+}
+
+std::vector<std::string> FieldNames(const StructType& type) {
+  std::vector<std::string> names;
+  for (const auto& field : type.fields()) {
+    names.emplace_back(field.name());
+  }
+  return names;
+}
+
+const StructType& StatsOf(const StructType& content_stats, int32_t base_id) {
+  auto field = content_stats.GetFieldById(base_id);
+  EXPECT_THAT(field, IsOk());
+  EXPECT_TRUE(field.value().has_value()) << "missing stats struct " << base_id;
+  const auto& type = field.value()->get().type();
+  EXPECT_EQ(type->type_id(), TypeId::kStruct);
+  return static_cast<const StructType&>(*type);
+}
+
+SchemaField ValueCount(int32_t base_id) {
+  return SchemaField::MakeOptional(base_id + 3, "value_count", int64(),
+                                   "Number of values (including null and NaN)");
+}
+
+SchemaField NullValueCount(int32_t base_id) {
+  return SchemaField::MakeOptional(base_id + 4, "null_value_count", int64(),
+                                   "Number of null values");
+}
+
+SchemaField NanValueCount(int32_t base_id) {
+  return SchemaField::MakeOptional(base_id + 5, "nan_value_count", int64(),
+                                   "Number of NaN values");
+}
+
+SchemaField TotalBytes(int32_t base_id) {
+  return SchemaField::MakeOptional(base_id + 6, "total_bytes", int64());
+}
+
+}  // namespace
+
+TEST(ContentStatsUtilTest, BaseId) {
+  EXPECT_EQ(Util::BaseId(0), 10'000);
+  EXPECT_EQ(Util::BaseId(1), 10'200);
+  // spec example: table field 2 uses [10_400, 10_599]
+  EXPECT_EQ(Util::BaseId(2), 10'400);
+  EXPECT_EQ(Util::BaseId(4), 10'800);
+  EXPECT_EQ(Util::BaseId(999'949), 199'999'800);
+
+  // stats for these fields would fall outside [10_000, 200_000_000)
+  EXPECT_EQ(Util::BaseId(-1), std::nullopt);
+  EXPECT_EQ(Util::BaseId(999'950), std::nullopt);
+  EXPECT_EQ(Util::BaseId(Schema::kInvalidColumnId), std::nullopt);
+
+  // reserved metadata fields
+  EXPECT_EQ(Util::BaseId(2147483539), 9'000);
+  EXPECT_EQ(Util::BaseId(MetadataColumns::kLastUpdatedSequenceNumberColumnId), 9'000);
+  EXPECT_EQ(Util::BaseId(2147483540), 9'200);
+  EXPECT_EQ(Util::BaseId(MetadataColumns::kRowIdColumnId), 9'200);
+  // other metadata fields are not tracked
+  EXPECT_EQ(Util::BaseId(MetadataColumns::kFilePathColumnId), std::nullopt);
+  EXPECT_EQ(Util::BaseId(MetadataColumns::kFilePositionColumnId), std::nullopt);
+}
+
+TEST(ContentStatsUtilTest, FieldIdAndOffset) {
+  for (int32_t field_id : {0, 1, 2, 17, 1000, 999'949}) {
+    auto base_id = Util::BaseId(field_id).value();
+    for (int32_t offset = 0; offset < Util::kNumReservedFieldStatsIds; ++offset) {
+      EXPECT_THAT(Util::FieldId(base_id + offset), HasValue(::testing::Eq(field_id)));
+      EXPECT_THAT(Util::StatOffset(base_id + offset), HasValue(::testing::Eq(offset)));
+    }
+  }
+
+  EXPECT_THAT(Util::FieldId(10'401), HasValue(::testing::Eq(2)));
+  EXPECT_THAT(Util::StatOffset(10'401), HasValue(::testing::Eq(1)));
+  EXPECT_THAT(Util::FieldId(10'817), HasValue(::testing::Eq(4)));
+  EXPECT_THAT(Util::StatOffset(10'817), HasValue(::testing::Eq(17)));
+}
+
+TEST(ContentStatsUtilTest, ReservedFieldIds) {
+  for (int32_t id : {9'000, 9'001, 9'199}) {
+    EXPECT_THAT(
+        Util::FieldId(id),
+        HasValue(::testing::Eq(MetadataColumns::kLastUpdatedSequenceNumberColumnId)));
+  }
+  for (int32_t id : {9'200, 9'206, 9'399}) {
+    EXPECT_THAT(Util::FieldId(id),
+                HasValue(::testing::Eq(MetadataColumns::kRowIdColumnId)));
+  }
+  EXPECT_THAT(Util::StatOffset(9'000), HasValue(::testing::Eq(0)));
+  EXPECT_THAT(Util::StatOffset(9'203), HasValue(::testing::Eq(3)));
+
+  // unassigned part of the metadata range
+  for (int32_t id : {9'400, 9'999}) {
+    EXPECT_THAT(Util::FieldId(id), IsError(ErrorKind::kNotSupported));
+    EXPECT_THAT(Util::FieldId(id),
+                HasErrorMessage("Unsupported metadata stats field ID"));
+    EXPECT_THAT(Util::StatOffset(id), HasValue(::testing::Eq(id % 200)));
+  }
+
+  // outside the stats range
+  for (int32_t id : {-1, 0, 146, 8'999, 200'000'000, 2147483540}) {
+    EXPECT_THAT(Util::FieldId(id), IsError(ErrorKind::kInvalidArgument));
+    EXPECT_THAT(Util::FieldId(id), HasErrorMessage("Invalid stats field ID"));
+    EXPECT_THAT(Util::StatOffset(id), IsError(ErrorKind::kInvalidArgument));
+  }
+  EXPECT_THAT(Util::FieldId(199'999'999), HasValue(::testing::Eq(999'949)));
+}
+
+// The spec example: a required int `id` (2) and an optional string `data` (3).
+TEST(ContentStatsUtilTest, SpecExample) {
+  Schema schema({SchemaField::MakeRequired(2, "id", int32()),
+                 SchemaField::MakeOptional(3, "data", string())});
+
+  auto expected = struct_({
+      SchemaField::MakeOptional(
+          10'400, "id",
+          struct_({
+              SchemaField::MakeOptional(10'401, "lower_bound", int32()),
+              SchemaField::MakeOptional(10'402, "upper_bound", int32()),
+              ValueCount(10'400),
+          })),
+      SchemaField::MakeOptional(
+          10'600, "data",
+          struct_({
+              SchemaField::MakeOptional(10'601, "lower_bound", string()),
+              SchemaField::MakeOptional(10'602, "upper_bound", string()),
+              ValueCount(10'600),
+              NullValueCount(10'600),
+              TotalBytes(10'600),
+          })),
+  });
+
+  auto write_type = Util::StatsWriteType(schema, *ConfigWithDefault("full"));
+  EXPECT_EQ(*write_type, *expected) << write_type->ToString();
+
+  // the default truncate(16) mode has the same type
+  EXPECT_EQ(*Util::StatsWriteType(schema, *MetricsConfig::Default()), *expected);
+
+  // reads also project null_value_count for the required `id`
+  std::vector<int32_t> ids = {2, 3};
+  auto read_type = Util::StatsReadType(schema, ids);
+  EXPECT_EQ(FieldIds(StatsOf(*read_type, 10'400)),
+            (std::vector<int32_t>{10'401, 10'402, 10'403, 10'404}));
+  EXPECT_EQ(StatsOf(*read_type, 10'600), StatsOf(*expected, 10'600));
+}
+
+// The spec example: an optional geometry `location` (4).
+TEST(ContentStatsUtilTest, GeoBounds) {
+  for (const auto& type : std::vector<std::shared_ptr<Type>>{geometry(), geography()}) {
+    SchemaField field = SchemaField::MakeOptional(4, "location", type);
+    auto expected = struct_({
+        SchemaField::MakeOptional(
+            10'801, "lower_bound",
+            struct_({
+                SchemaField::MakeRequired(10'810, "x", float64(),
+                                          "Bounding box westernmost/xmin; [-180..180]"),
+                SchemaField::MakeRequired(10'811, "y", float64(),
+                                          "Bounding box southernmost/ymin; [-90..90]"),
+                SchemaField::MakeOptional(10'812, "z", float64(), "Bounding box zmin"),
+                SchemaField::MakeOptional(10'813, "m", float64(), "Bounding box mmin"),
+            })),
+        SchemaField::MakeOptional(
+            10'802, "upper_bound",
+            struct_({
+                SchemaField::MakeRequired(10'814, "x", float64(),
+                                          "Bounding box easternmost/xmax; [-180..180]"),
+                SchemaField::MakeRequired(10'815, "y", float64(),
+                                          "Bounding box northernmost/ymax; [-90..90]"),
+                SchemaField::MakeOptional(10'816, "z", float64(), "Bounding box zmax"),
+                SchemaField::MakeOptional(10'817, "m", float64(), "Bounding box mmax"),
+            })),
+        ValueCount(10'800),
+        NullValueCount(10'800),
+        TotalBytes(10'800),
+    });
+
+    auto stats = Util::FieldStatsType(field, 10'800, MetricsMode::Full());
+    ASSERT_NE(stats, nullptr);
+    EXPECT_EQ(*stats, *expected) << stats->ToString();
+
+    // counts mode has no bounds
+    auto counts = Util::FieldStatsType(field, 10'800, MetricsMode::Counts());
+    ASSERT_NE(counts, nullptr);
+    EXPECT_EQ(FieldIds(*counts), (std::vector<int32_t>{10'803, 10'804, 10'806}));
+  }
+}
+
+TEST(ContentStatsUtilTest, PrimitiveTypes) {
+  struct Case {
+    std::shared_ptr<Type> type;
+    bool has_nan_count;
+    bool has_total_bytes;
+  };
+  std::vector<Case> cases = {
+      {.type = boolean(), .has_nan_count = false, .has_total_bytes = false},
+      {.type = int32(), .has_nan_count = false, .has_total_bytes = false},
+      {.type = int64(), .has_nan_count = false, .has_total_bytes = false},
+      {.type = float32(), .has_nan_count = true, .has_total_bytes = false},
+      {.type = float64(), .has_nan_count = true, .has_total_bytes = false},
+      {.type = decimal(9, 2), .has_nan_count = false, .has_total_bytes = false},
+      {.type = date(), .has_nan_count = false, .has_total_bytes = false},
+      {.type = time(), .has_nan_count = false, .has_total_bytes = false},
+      {.type = timestamp(), .has_nan_count = false, .has_total_bytes = false},
+      {.type = timestamp_tz(), .has_nan_count = false, .has_total_bytes = false},
+      {.type = timestamp_ns(), .has_nan_count = false, .has_total_bytes = false},
+      {.type = timestamptz_ns(), .has_nan_count = false, .has_total_bytes = false},
+      {.type = string(), .has_nan_count = false, .has_total_bytes = true},
+      {.type = uuid(), .has_nan_count = false, .has_total_bytes = false},
+      {.type = fixed(16), .has_nan_count = false, .has_total_bytes = false},
+      {.type = binary(), .has_nan_count = false, .has_total_bytes = true},
+      {.type = unknown(), .has_nan_count = false, .has_total_bytes = false},
+      {.type = variant(), .has_nan_count = false, .has_total_bytes = true},
+      {.type = geometry(), .has_nan_count = false, .has_total_bytes = true},
+      {.type = geography(), .has_nan_count = false, .has_total_bytes = true},
+  };
+
+  constexpr int32_t kBaseId = 10'200;
+  for (const auto& c : cases) {
+    SCOPED_TRACE(c.type->ToString());
+    bool is_geo =
+        c.type->type_id() == TypeId::kGeometry || c.type->type_id() == TypeId::kGeography;
+
+    for (bool optional : {false, true}) {
+      SchemaField field(1, "col", c.type, optional);
+      auto stats = Util::FieldStatsType(field, kBaseId, MetricsMode::Full());
+      ASSERT_NE(stats, nullptr);
+
+      std::vector<int32_t> expected_ids = {10'201, 10'202, 10'203};
+      std::vector<std::string> expected_names = {"lower_bound", "upper_bound",
+                                                 "value_count"};
+      if (optional) {
+        expected_ids.push_back(10'204);
+        expected_names.emplace_back("null_value_count");
+      }
+      if (c.has_nan_count) {
+        expected_ids.push_back(10'205);
+        expected_names.emplace_back("nan_value_count");
+      }
+      if (c.has_total_bytes) {
+        expected_ids.push_back(10'206);
+        expected_names.emplace_back("total_bytes");
+      }
+      EXPECT_EQ(FieldIds(*stats), expected_ids);
+      EXPECT_EQ(FieldNames(*stats), expected_names);
+
+      for (const auto& stat : stats->fields()) {
+        EXPECT_TRUE(stat.optional()) << stat.name();
+        if (stat.field_id() == 10'201 || stat.field_id() == 10'202) {
+          if (is_geo) {
+            EXPECT_EQ(stat.type()->type_id(), TypeId::kStruct);
+          } else {
+            // bounds use the field's type
+            EXPECT_EQ(*stat.type(), *c.type);
+          }
+        } else {
+          EXPECT_EQ(*stat.type(), *int64());
+        }
+      }
+    }
+  }
+}
+
+TEST(ContentStatsUtilTest, MetricsModes) {
+  SchemaField field = SchemaField::MakeOptional(1, "f", float64());
+  constexpr int32_t kBaseId = 10'200;
+
+  EXPECT_EQ(Util::FieldStatsType(field, kBaseId, MetricsMode::None()), nullptr);
+
+  auto counts = Util::FieldStatsType(field, kBaseId, MetricsMode::Counts());
+  ASSERT_NE(counts, nullptr);
+  EXPECT_EQ(FieldIds(*counts), (std::vector<int32_t>{10'203, 10'204, 10'205}));
+
+  auto truncate = Util::FieldStatsType(
+      field, kBaseId, MetricsMode{.kind = MetricsMode::Kind::kTruncate, .length = 4});
+  ASSERT_NE(truncate, nullptr);
+  EXPECT_EQ(FieldIds(*truncate),
+            (std::vector<int32_t>{10'201, 10'202, 10'203, 10'204, 10'205}));
+
+  auto full = Util::FieldStatsType(field, kBaseId, MetricsMode::Full());
+  ASSERT_NE(full, nullptr);
+  EXPECT_EQ(*full, *truncate);
+
+  // nested types have no field stats
+  SchemaField nested = SchemaField::MakeOptional(
+      1, "s", struct_({SchemaField::MakeOptional(2, "x", int32())}));
+  EXPECT_EQ(Util::FieldStatsType(nested, kBaseId, MetricsMode::Full()), nullptr);
+}
+
+TEST(ContentStatsUtilTest, WriteTypeMetricsConfig) {
+  Schema schema({
+      SchemaField::MakeRequired(1, "id", int64()),
+      SchemaField::MakeOptional(2, "data", string()),
+      SchemaField::MakeOptional(3, "point",
+                                struct_({SchemaField::MakeRequired(4, "x", float64()),
+                                         SchemaField::MakeRequired(5, "y", float64())})),
+  });
+
+  // none: no stats at all
+  auto none = Util::StatsWriteType(schema, *ConfigWithDefault("none"));
+  EXPECT_TRUE(none->fields().empty());
+
+  // counts: no bounds
+  auto counts = Util::StatsWriteType(schema, *ConfigWithDefault("counts"));
+  EXPECT_EQ(FieldIds(*counts), (std::vector<int32_t>{10'200, 10'400, 10'800, 11'000}));
+  EXPECT_EQ(FieldIds(StatsOf(*counts, 10'200)), (std::vector<int32_t>{10'203}));
+  EXPECT_EQ(FieldIds(StatsOf(*counts, 10'400)),
+            (std::vector<int32_t>{10'403, 10'404, 10'406}));
+  EXPECT_EQ(FieldIds(StatsOf(*counts, 10'800)), (std::vector<int32_t>{10'803, 10'805}));
+
+  // truncate and full: bounds
+  for (const auto* mode : {"truncate(8)", "full"}) {
+    auto stats = Util::StatsWriteType(schema, *ConfigWithDefault(mode));
+    EXPECT_EQ(FieldIds(*stats), (std::vector<int32_t>{10'200, 10'400, 10'800, 11'000}));
+    EXPECT_EQ(FieldIds(StatsOf(*stats, 10'200)),
+              (std::vector<int32_t>{10'201, 10'202, 10'203}));
+    EXPECT_EQ(FieldIds(StatsOf(*stats, 10'400)),
+              (std::vector<int32_t>{10'401, 10'402, 10'403, 10'404, 10'406}));
+  }
+
+  // per-column overrides use the canonical (dotted) column name
+  auto mixed = Util::StatsWriteType(
+      schema, *MakeConfig({{"write.metadata.metrics.default", "counts"},
+                           {"write.metadata.metrics.column.id", "none"},
+                           {"write.metadata.metrics.column.point.x", "full"},
+                           {"write.metadata.metrics.column.point.y", "none"}}));
+  EXPECT_EQ(FieldIds(*mixed), (std::vector<int32_t>{10'400, 10'800}));
+  EXPECT_EQ(FieldIds(StatsOf(*mixed, 10'400)),
+            (std::vector<int32_t>{10'403, 10'404, 10'406}));
+  EXPECT_EQ(FieldIds(StatsOf(*mixed, 10'800)),
+            (std::vector<int32_t>{10'801, 10'802, 10'803, 10'805}));
+}
+
+TEST(ContentStatsUtilTest, NestedStructs) {
+  Schema schema({
+      SchemaField::MakeOptional(
+          1, "a",
+          struct_({
+              SchemaField::MakeRequired(2, "b", int32()),
+              SchemaField::MakeOptional(
+                  3, "c", struct_({SchemaField::MakeOptional(4, "d", string())})),
+          })),
+      SchemaField::MakeRequired(5, "e", int32()),
+  });
+
+  auto stats = Util::StatsWriteType(schema, *ConfigWithDefault("full"));
+  // struct fields have no stats; leaves are named by their path joined with '_'
+  EXPECT_EQ(FieldIds(*stats), (std::vector<int32_t>{10'400, 10'800, 11'000}));
+  EXPECT_EQ(FieldNames(*stats), (std::vector<std::string>{"a_b", "a_c_d", "e"}));
+  for (const auto& field : stats->fields()) {
+    EXPECT_TRUE(field.optional());
+  }
+
+  // `b` is required: no null count, even though its parent is optional
+  EXPECT_EQ(FieldIds(StatsOf(*stats, 10'400)),
+            (std::vector<int32_t>{10'401, 10'402, 10'403}));
+  EXPECT_EQ(FieldIds(StatsOf(*stats, 10'800)),
+            (std::vector<int32_t>{10'801, 10'802, 10'803, 10'804, 10'806}));
+}
+
+TEST(ContentStatsUtilTest, ListsAndMapsAreSkipped) {
+  Schema schema({
+      SchemaField::MakeRequired(1, "id", int64()),
+      SchemaField::MakeOptional(2, "tags",
+                                list(SchemaField::MakeOptional(3, "element", string()))),
+      SchemaField::MakeOptional(4, "props",
+                                map(SchemaField::MakeRequired(5, "key", string()),
+                                    SchemaField::MakeOptional(6, "value", int32()))),
+      SchemaField::MakeOptional(
+          7, "points",
+          list(SchemaField::MakeOptional(
+              8, "element", struct_({SchemaField::MakeRequired(9, "x", float64())})))),
+      SchemaField::MakeOptional(
+          10, "s",
+          struct_({SchemaField::MakeOptional(
+                       11, "nested_list",
+                       list(SchemaField::MakeRequired(12, "element", int32()))),
+                   SchemaField::MakeOptional(13, "v", variant())})),
+  });
+
+  auto stats = Util::StatsWriteType(schema, *ConfigWithDefault("full"));
+  EXPECT_EQ(FieldIds(*stats), (std::vector<int32_t>{10'200, 12'600}));
+  EXPECT_EQ(FieldNames(*stats), (std::vector<std::string>{"id", "s_v"}));
+
+  // variant bounds are variants
+  const auto& variant_stats = StatsOf(*stats, 12'600);
+  EXPECT_EQ(FieldIds(variant_stats),
+            (std::vector<int32_t>{12'601, 12'602, 12'603, 12'604, 12'606}));
+  EXPECT_EQ(*variant_stats.fields()[0].type(), *variant());
+
+  // reads skip them too
+  std::vector<int32_t> ids = {2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12};
+  EXPECT_TRUE(Util::StatsReadType(schema, ids)->fields().empty());
+}
+
+TEST(ContentStatsUtilTest, ReadType) {
+  Schema schema({
+      SchemaField::MakeRequired(1, "id", int64()),
+      SchemaField::MakeOptional(2, "data", string()),
+      SchemaField::MakeOptional(3, "score", float32()),
+  });
+
+  // full mode, ordered by stats ID, duplicates and unknown IDs ignored
+  std::vector<int32_t> ids = {3, 1, 3, 42, -1, MetadataColumns::kRowIdColumnId};
+  auto stats = Util::StatsReadType(schema, ids);
+  EXPECT_EQ(FieldIds(*stats), (std::vector<int32_t>{10'200, 10'600}));
+  EXPECT_EQ(FieldNames(*stats), (std::vector<std::string>{"id", "score"}));
+  EXPECT_EQ(FieldIds(StatsOf(*stats, 10'600)),
+            (std::vector<int32_t>{10'601, 10'602, 10'603, 10'604, 10'605}));
+
+  // the required `id` also gets null_value_count
+  EXPECT_EQ(FieldIds(StatsOf(*stats, 10'200)),
+            (std::vector<int32_t>{10'201, 10'202, 10'203, 10'204}));
+
+  // the read type covers the full write type
+  auto write_type = Util::StatsWriteType(schema, *ConfigWithDefault("full"));
+  EXPECT_EQ(StatsOf(*stats, 10'600), StatsOf(*write_type, 10'600));
+  for (const auto& stat : StatsOf(*write_type, 10'200).fields()) {
+    auto read_stat = StatsOf(*stats, 10'200).GetFieldById(stat.field_id());
+    ASSERT_THAT(read_stat, IsOk());
+    ASSERT_TRUE(read_stat.value().has_value()) << stat.field_id();
+    EXPECT_EQ(read_stat.value()->get(), stat);
+  }
+
+  EXPECT_TRUE(Util::StatsReadType(schema, std::vector<int32_t>{})->fields().empty());
+}
+
+TEST(ContentStatsUtilTest, OutOfRangeFieldIds) {
+  Schema schema({
+      SchemaField::MakeRequired(1, "id", int64()),
+      SchemaField::MakeOptional(999'949, "last", int32()),
+      SchemaField::MakeOptional(999'950, "too_big", int32()),
+  });
+
+  auto stats = Util::StatsWriteType(schema, *ConfigWithDefault("full"));
+  EXPECT_EQ(FieldIds(*stats), (std::vector<int32_t>{10'200, 199'999'800}));
+
+  std::vector<int32_t> ids = {999'950};
+  EXPECT_TRUE(Util::StatsReadType(schema, ids)->fields().empty());
+}
+
+// Java writes null_value_count for required fields too (apache/iceberg#17413). The
+// read type must resolve it by ID from such files, and must also read files written
+// by the spec rule, where it is missing, as unknown (null).
+TEST(ContentStatsUtilTest, ReadTypeProjectsJavaAndSpecFiles) {
+  Schema schema({SchemaField::MakeRequired(2, "id", int32()),
+                 SchemaField::MakeOptional(3, "data", string())});
+  std::vector<int32_t> ids = {2, 3};
+  auto read_schema = Util::StatsReadType(schema, ids)->ToSchema();
+
+  // content_stats as Java's StatsUtil.statsWriteSchema writes it
+  auto java_written = struct_({
+      SchemaField::MakeOptional(
+          10'400, "id",
+          struct_({
+              SchemaField::MakeOptional(10'401, "lower_bound", int32()),
+              SchemaField::MakeOptional(10'402, "upper_bound", int32()),
+              ValueCount(10'400),
+              NullValueCount(10'400),
+          })),
+      SchemaField::MakeOptional(
+          10'600, "data",
+          struct_({
+              SchemaField::MakeOptional(10'601, "lower_bound", string()),
+              SchemaField::MakeOptional(10'602, "upper_bound", string()),
+              ValueCount(10'600),
+              NullValueCount(10'600),
+              TotalBytes(10'600),
+          })),
+  });
+  auto spec_written = Util::StatsWriteType(schema, *ConfigWithDefault("full"));
+
+  for (bool from_java : {true, false}) {
+    SCOPED_TRACE(from_java ? "Java-written" : "spec-written");
+    auto file_schema = (from_java ? java_written : spec_written)->ToSchema();
+    auto projection = Project(*read_schema, *file_schema, /*prune_source=*/true);
+    ASSERT_THAT(projection, IsOk());
+    ASSERT_EQ(projection->fields.size(), 2);
+
+    const auto& id_stats = projection->fields[0];
+    EXPECT_EQ(id_stats.kind, FieldProjection::Kind::kProjected);
+    ASSERT_EQ(id_stats.children.size(), 4);
+    for (size_t i = 0; i < 3; ++i) {
+      EXPECT_EQ(id_stats.children[i].kind, FieldProjection::Kind::kProjected);
+    }
+    // 10_404 null_value_count of the required `id`
+    EXPECT_EQ(id_stats.children[3].kind, from_java ? FieldProjection::Kind::kProjected
+                                                   : FieldProjection::Kind::kNull);
+
+    for (const auto& child : projection->fields[1].children) {
+      EXPECT_EQ(child.kind, FieldProjection::Kind::kProjected);
+    }
+  }
+}
+
+}  // namespace iceberg