diff --git a/src/iceberg/json_serde.cc b/src/iceberg/json_serde.cc
index 1f5caf8f0..45dfa29f5 100644
--- a/src/iceberg/json_serde.cc
+++ b/src/iceberg/json_serde.cc
@@ -18,10 +18,13 @@
  */
 
 #include <algorithm>
+#include <concepts>
 #include <cstdint>
 #include <format>
+#include <limits>
 #include <map>
 #include <regex>
+#include <string>
 #include <unordered_set>
 #include <utility>
 
@@ -269,7 +272,58 @@ constexpr std::string_view kContentSizeInBytes = "content-size-in-bytes";
 constexpr std::string_view kMapKeys = "keys";
 constexpr std::string_view kMapValues = "values";
 
-template <typename Value>
+template <std::signed_integral Integer>
+Result<Integer> IntegerFromJson(const nlohmann::json& json,
+                                std::string_view description) {
+  if (!json.is_number_integer()) {
+    return JsonParseError("{} must be an integer, but is {}", description,
+                          json.type_name());
+  }
+  if (json.is_number_unsigned()) {
+    const auto value = json.get<uint64_t>();
+    if (value > static_cast<uint64_t>(std::numeric_limits<Integer>::max())) {
+      return JsonParseError("{} integer is out of range: {}", description,
+                            SafeDumpJson(json));
+    }
+    return static_cast<Integer>(value);
+  }
+
+  const auto value = json.get<int64_t>();
+  if (value < static_cast<int64_t>(std::numeric_limits<Integer>::min()) ||
+      value > static_cast<int64_t>(std::numeric_limits<Integer>::max())) {
+    return JsonParseError("{} integer is out of range: {}", description,
+                          SafeDumpJson(json));
+  }
+  return static_cast<Integer>(value);
+}
+
+template <std::signed_integral Integer>
+Result<Integer> GetJsonInteger(const nlohmann::json& json, std::string_view key) {
+  ICEBERG_ASSIGN_OR_RAISE(auto value, GetJsonValue<nlohmann::json>(json, key));
+  return IntegerFromJson<Integer>(value, std::format("'{}'", key));
+}
+
+template <std::signed_integral Integer>
+Result<std::vector<Integer>> IntegerVectorFromJson(const nlohmann::json& json,
+                                                   std::string_view key) {
+  ICEBERG_ASSIGN_OR_RAISE(auto values_json, GetJsonValue<nlohmann::json>(json, key));
+  if (!values_json.is_array()) {
+    return JsonParseError("'{}' must be an array, but is {}", key,
+                          values_json.type_name());
+  }
+
+  std::vector<Integer> values;
+  values.reserve(values_json.size());
+  for (const auto& value_json : values_json) {
+    ICEBERG_ASSIGN_OR_RAISE(
+        auto value,
+        IntegerFromJson<Integer>(value_json, std::format("'{}' element", key)));
+    values.push_back(value);
+  }
+  return values;
+}
+
+template <std::signed_integral Value>
 Result<std::map<int32_t, Value>> KeyValueMapFromJson(const nlohmann::json& json,
                                                      std::string_view key) {
   std::map<int32_t, Value> result;
@@ -278,16 +332,27 @@ Result<std::map<int32_t, Value>> KeyValueMapFromJson(const nlohmann::json& json,
   }
 
   ICEBERG_ASSIGN_OR_RAISE(auto map_json, GetJsonValue<nlohmann::json>(json, key));
-  ICEBERG_ASSIGN_OR_RAISE(auto keys,
-                          GetJsonValue<std::vector<int32_t>>(map_json, kMapKeys));
-  ICEBERG_ASSIGN_OR_RAISE(auto values,
-                          GetJsonValue<std::vector<Value>>(map_json, kMapValues));
-  if (keys.size() != values.size()) {
+  ICEBERG_ASSIGN_OR_RAISE(auto keys_json,
+                          GetJsonValue<nlohmann::json>(map_json, kMapKeys));
+  ICEBERG_ASSIGN_OR_RAISE(auto values_json,
+                          GetJsonValue<nlohmann::json>(map_json, kMapValues));
+  if (!keys_json.is_array() || !values_json.is_array()) {
+    return JsonParseError("'{}' map keys and values must be arrays", key);
+  }
+  if (keys_json.size() != values_json.size()) {
     return JsonParseError("'{}' map keys and values have different lengths", key);
   }
 
-  for (size_t i = 0; i < keys.size(); ++i) {
-    result[keys[i]] = std::move(values[i]);
+  for (size_t i = 0; i < keys_json.size(); ++i) {
+    ICEBERG_ASSIGN_OR_RAISE(
+        auto field_id,
+        IntegerFromJson<int32_t>(keys_json[i], std::format("'{}' map key", key)));
+    ICEBERG_ASSIGN_OR_RAISE(
+        auto value,
+        IntegerFromJson<Value>(values_json[i], std::format("'{}' map value", key)));
+    if (!result.emplace(field_id, value).second) {
+      return JsonParseError("'{}' map contains duplicate key {}", key, field_id);
+    }
   }
   return result;
 }
@@ -310,6 +375,118 @@ void SetKeyValueMap(nlohmann::json& json, std::string_view key,
   json[key] = {{kMapKeys, std::move(keys)}, {kMapValues, std::move(values)}};
 }
 
+std::string BytesToHex(const std::vector<uint8_t>& bytes) {
+  std::string hex;
+  hex.reserve(bytes.size() * 2);
+  for (uint8_t byte : bytes) {
+    hex += std::format("{:02X}", byte);
+  }
+  return hex;
+}
+
+Result<std::map<int32_t, std::vector<uint8_t>>> BytesMapFromJson(
+    const nlohmann::json& json, std::string_view key) {
+  std::map<int32_t, std::vector<uint8_t>> result;
+  if (!json.contains(key) || json.at(key).is_null()) {
+    return result;
+  }
+
+  ICEBERG_ASSIGN_OR_RAISE(auto map_json, GetJsonValue<nlohmann::json>(json, key));
+  ICEBERG_ASSIGN_OR_RAISE(auto keys_json,
+                          GetJsonValue<nlohmann::json>(map_json, kMapKeys));
+  ICEBERG_ASSIGN_OR_RAISE(auto values_json,
+                          GetJsonValue<nlohmann::json>(map_json, kMapValues));
+  if (!keys_json.is_array() || !values_json.is_array()) {
+    return JsonParseError("'{}' map keys and values must be arrays", key);
+  }
+  if (keys_json.size() != values_json.size()) {
+    return JsonParseError("'{}' map keys and values have different lengths", key);
+  }
+
+  for (size_t i = 0; i < keys_json.size(); ++i) {
+    ICEBERG_ASSIGN_OR_RAISE(
+        auto field_id,
+        IntegerFromJson<int32_t>(keys_json[i], std::format("'{}' map key", key)));
+    if (!values_json[i].is_string()) {
+      return JsonParseError("'{}' map value must be a string, but is {}", key,
+                            values_json[i].type_name());
+    }
+    ICEBERG_ASSIGN_OR_RAISE(
+        auto bytes, StringUtils::HexStringToBytes(values_json[i].get<std::string>()));
+    if (!result.emplace(field_id, std::move(bytes)).second) {
+      return JsonParseError("'{}' map contains duplicate key {}", key, field_id);
+    }
+  }
+  return result;
+}
+
+void SetBytesMap(nlohmann::json& json, std::string_view key,
+                 const std::map<int32_t, std::vector<uint8_t>>& map) {
+  if (map.empty()) {
+    return;
+  }
+
+  std::vector<int32_t> keys;
+  std::vector<std::string> values;
+  keys.reserve(map.size());
+  values.reserve(map.size());
+  for (const auto& [field_id, value] : map) {
+    keys.push_back(field_id);
+    values.push_back(BytesToHex(value));
+  }
+  json[key] = {{kMapKeys, std::move(keys)}, {kMapValues, std::move(values)}};
+}
+
+Result<Literal> PartitionLiteralFromJson(const nlohmann::json* value,
+                                         const SchemaField& field) {
+  if (!field.type() || !field.type()->is_primitive()) {
+    return InvalidSchema("Partition field {} must have a primitive type",
+                         field.field_id());
+  }
+
+  auto primitive_type = internal::checked_pointer_cast<PrimitiveType>(field.type());
+  if (value == nullptr || value->is_null()) {
+    return Literal::Null(std::move(primitive_type));
+  }
+  return LiteralFromJson(*value, primitive_type.get());
+}
+
+Result<PartitionValues> PartitionValuesFromJson(const nlohmann::json& json,
+                                                const StructType& partition_type) {
+  const auto& fields = partition_type.fields();
+  std::vector<Literal> values;
+  values.reserve(fields.size());
+
+  if (json.is_array()) {
+    if (json.size() != fields.size()) {
+      return JsonParseError("Invalid partition data size: expected = {}, actual = {}",
+                            fields.size(), json.size());
+    }
+    for (size_t pos = 0; pos < fields.size(); ++pos) {
+      ICEBERG_ASSIGN_OR_RAISE(auto value,
+                              PartitionLiteralFromJson(&json[pos], fields[pos]));
+      values.push_back(std::move(value));
+    }
+  } else if (json.is_object()) {
+    if (json.size() > fields.size()) {
+      return JsonParseError("Invalid partition data size: expected <= {}, actual = {}",
+                            fields.size(), json.size());
+    }
+    for (const auto& field : fields) {
+      const auto it = json.find(std::to_string(field.field_id()));
+      const nlohmann::json* value = it == json.end() ? nullptr : &it.value();
+      ICEBERG_ASSIGN_OR_RAISE(auto literal, PartitionLiteralFromJson(value, field));
+      values.push_back(std::move(literal));
+    }
+  } else {
+    return JsonParseError(
+        "Invalid partition data for content file: expected array or object ({})",
+        SafeDumpJson(json));
+  }
+
+  return PartitionValues(std::move(values));
+}
+
 Result<nlohmann::json> DataFileToJsonUnchecked(const DataFile& data_file) {
   nlohmann::json json;
   switch (data_file.content) {
@@ -327,7 +504,7 @@ Result<nlohmann::json> DataFileToJsonUnchecked(const DataFile& data_file) {
   json[kFileFormat] = ToString(data_file.file_format);
 
   if (!data_file.partition_spec_id.has_value()) {
-    return ValidationFailed("Cannot serialize REST content file without 'spec-id'");
+    return ValidationFailed("Cannot serialize content file without 'spec-id'");
   }
   json[kSpecId] = data_file.partition_spec_id.value();
 
@@ -345,11 +522,11 @@ Result<nlohmann::json> DataFileToJsonUnchecked(const DataFile& data_file) {
   SetKeyValueMap(json, kValueCounts, data_file.value_counts);
   SetKeyValueMap(json, kNullValueCounts, data_file.null_value_counts);
   SetKeyValueMap(json, kNanValueCounts, data_file.nan_value_counts);
-  SetKeyValueMap(json, kLowerBounds, data_file.lower_bounds);
-  SetKeyValueMap(json, kUpperBounds, data_file.upper_bounds);
+  SetBytesMap(json, kLowerBounds, data_file.lower_bounds);
+  SetBytesMap(json, kUpperBounds, data_file.upper_bounds);
 
   if (!data_file.key_metadata.empty()) {
-    json[kKeyMetadata] = data_file.key_metadata;
+    json[kKeyMetadata] = BytesToHex(data_file.key_metadata);
   }
   if (!data_file.split_offsets.empty()) {
     json[kSplitOffsets] = data_file.split_offsets;
@@ -2132,11 +2309,13 @@ Result<DataFile> DataFileFromJson(
   DataFile data_file;
 
   ICEBERG_ASSIGN_OR_RAISE(auto content_str, GetJsonValue<std::string>(json, kContent));
-  if (content_str == kContentData) {
+  if (content_str == kContentData || content_str == "DATA") {
     data_file.content = DataFile::Content::kData;
-  } else if (content_str == kContentPositionDeletes) {
+  } else if (content_str == kContentPositionDeletes ||
+             content_str == "POSITION_DELETES") {
     data_file.content = DataFile::Content::kPositionDeletes;
-  } else if (content_str == kContentEqualityDeletes) {
+  } else if (content_str == kContentEqualityDeletes ||
+             content_str == "EQUALITY_DELETES") {
     data_file.content = DataFile::Content::kEqualityDeletes;
   } else {
     return JsonParseError("Unknown data file content: {}", content_str);
@@ -2147,37 +2326,25 @@ Result<DataFile> DataFileFromJson(
   ICEBERG_ASSIGN_OR_RAISE(auto format_str, GetJsonValue<std::string>(json, kFileFormat));
   ICEBERG_ASSIGN_OR_RAISE(data_file.file_format, FileFormatTypeFromString(format_str));
 
-  ICEBERG_ASSIGN_OR_RAISE(auto spec_id, GetJsonValue<int32_t>(json, kSpecId));
+  ICEBERG_ASSIGN_OR_RAISE(auto spec_id, GetJsonInteger<int32_t>(json, kSpecId));
   data_file.partition_spec_id = spec_id;
 
-  ICEBERG_ASSIGN_OR_RAISE(auto partition_vals,
-                          GetJsonValue<nlohmann::json>(json, kPartition));
-  if (!partition_vals.is_array()) {
-    return JsonParseError("PartitionValues must be a JSON array: {}",
-                          SafeDumpJson(partition_vals));
-  }
-  std::vector<Literal> literals;
   auto it = partition_spec_by_id.find(spec_id);
-  if (it == partition_spec_by_id.end()) {
+  if (it == partition_spec_by_id.end() || !it->second) {
     return JsonParseError("Invalid partition spec id: {}", spec_id);
   }
-  ICEBERG_ASSIGN_OR_RAISE(auto struct_type, it->second->PartitionType(schema));
-  auto fields = struct_type->fields();
-  if (partition_vals.size() != fields.size()) {
-    return JsonParseError("Invalid partition data size: expected = {}, actual = {}",
-                          fields.size(), partition_vals.size());
-  }
-  for (size_t pos = 0; pos < fields.size(); ++pos) {
-    ICEBERG_ASSIGN_OR_RAISE(
-        auto literal, LiteralFromJson(partition_vals[pos], fields[pos].type().get()));
-    literals.push_back(std::move(literal));
+  if (json.contains(kPartition)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto partition_vals,
+                            GetJsonValue<nlohmann::json>(json, kPartition));
+    ICEBERG_ASSIGN_OR_RAISE(auto struct_type, it->second->PartitionType(schema));
+    ICEBERG_ASSIGN_OR_RAISE(data_file.partition,
+                            PartitionValuesFromJson(partition_vals, *struct_type));
   }
-  data_file.partition = PartitionValues(std::move(literals));
 
   ICEBERG_ASSIGN_OR_RAISE(data_file.record_count,
-                          GetJsonValue<int64_t>(json, kRecordCount));
+                          GetJsonInteger<int64_t>(json, kRecordCount));
   ICEBERG_ASSIGN_OR_RAISE(data_file.file_size_in_bytes,
-                          GetJsonValue<int64_t>(json, kFileSizeInBytes));
+                          GetJsonInteger<int64_t>(json, kFileSizeInBytes));
 
   ICEBERG_ASSIGN_OR_RAISE(data_file.column_sizes,
                           KeyValueMapFromJson<int64_t>(json, kColumnSizes));
@@ -2187,30 +2354,30 @@ Result<DataFile> DataFileFromJson(
                           KeyValueMapFromJson<int64_t>(json, kNullValueCounts));
   ICEBERG_ASSIGN_OR_RAISE(data_file.nan_value_counts,
                           KeyValueMapFromJson<int64_t>(json, kNanValueCounts));
-  ICEBERG_ASSIGN_OR_RAISE(data_file.lower_bounds,
-                          KeyValueMapFromJson<std::vector<uint8_t>>(json, kLowerBounds));
-  ICEBERG_ASSIGN_OR_RAISE(data_file.upper_bounds,
-                          KeyValueMapFromJson<std::vector<uint8_t>>(json, kUpperBounds));
+  ICEBERG_ASSIGN_OR_RAISE(data_file.lower_bounds, BytesMapFromJson(json, kLowerBounds));
+  ICEBERG_ASSIGN_OR_RAISE(data_file.upper_bounds, BytesMapFromJson(json, kUpperBounds));
 
   if (json.contains(kKeyMetadata) && !json.at(kKeyMetadata).is_null()) {
+    ICEBERG_ASSIGN_OR_RAISE(auto key_metadata,
+                            GetJsonValue<std::string>(json, kKeyMetadata));
     ICEBERG_ASSIGN_OR_RAISE(data_file.key_metadata,
-                            GetJsonValue<std::vector<uint8_t>>(json, kKeyMetadata));
+                            StringUtils::HexStringToBytes(key_metadata));
   }
   if (json.contains(kSplitOffsets) && !json.at(kSplitOffsets).is_null()) {
     ICEBERG_ASSIGN_OR_RAISE(data_file.split_offsets,
-                            GetJsonValue<std::vector<int64_t>>(json, kSplitOffsets));
+                            IntegerVectorFromJson<int64_t>(json, kSplitOffsets));
   }
   if (json.contains(kEqualityIds) && !json.at(kEqualityIds).is_null()) {
     ICEBERG_ASSIGN_OR_RAISE(data_file.equality_ids,
-                            GetJsonValue<std::vector<int32_t>>(json, kEqualityIds));
+                            IntegerVectorFromJson<int32_t>(json, kEqualityIds));
   }
   if (json.contains(kSortOrderId) && !json.at(kSortOrderId).is_null()) {
     ICEBERG_ASSIGN_OR_RAISE(data_file.sort_order_id,
-                            GetJsonValue<int32_t>(json, kSortOrderId));
+                            GetJsonInteger<int32_t>(json, kSortOrderId));
   }
   if (json.contains(kFirstRowId) && !json.at(kFirstRowId).is_null()) {
     ICEBERG_ASSIGN_OR_RAISE(data_file.first_row_id,
-                            GetJsonValue<int64_t>(json, kFirstRowId));
+                            GetJsonInteger<int64_t>(json, kFirstRowId));
   }
   if (json.contains(kReferencedDataFile) && !json.at(kReferencedDataFile).is_null()) {
     ICEBERG_ASSIGN_OR_RAISE(data_file.referenced_data_file,
@@ -2218,11 +2385,11 @@ Result<DataFile> DataFileFromJson(
   }
   if (json.contains(kContentOffset) && !json.at(kContentOffset).is_null()) {
     ICEBERG_ASSIGN_OR_RAISE(data_file.content_offset,
-                            GetJsonValue<int64_t>(json, kContentOffset));
+                            GetJsonInteger<int64_t>(json, kContentOffset));
   }
   if (json.contains(kContentSizeInBytes) && !json.at(kContentSizeInBytes).is_null()) {
     ICEBERG_ASSIGN_OR_RAISE(data_file.content_size_in_bytes,
-                            GetJsonValue<int64_t>(json, kContentSizeInBytes));
+                            GetJsonInteger<int64_t>(json, kContentSizeInBytes));
   }
 
   return data_file;
@@ -2252,6 +2419,16 @@ Result<nlohmann::json> ToJson(
         partition_type->fields().empty() ? "unpartitioned" : "partitioned",
         data_file.partition.num_fields() == 0 ? "unpartitioned" : "partitioned");
   }
+  for (size_t pos = 0; pos < partition_type->fields().size(); ++pos) {
+    const auto& literal = data_file.partition.values()[pos];
+    const auto& expected_type = partition_type->fields()[pos].type();
+    if (!literal.IsNull() && (!literal.type() || *literal.type() != *expected_type)) {
+      return ValidationFailed(
+          "Invalid partition value type at position {}: expected = {}, actual = {}", pos,
+          expected_type->ToString(),
+          literal.type() ? literal.type()->ToString() : "unknown");
+    }
+  }
   return DataFileToJsonUnchecked(data_file);
 }
 
diff --git a/src/iceberg/test/json_serde_test.cc b/src/iceberg/test/json_serde_test.cc
index cde82293e..3879d5074 100644
--- a/src/iceberg/test/json_serde_test.cc
+++ b/src/iceberg/test/json_serde_test.cc
@@ -18,6 +18,8 @@
  */
 
 #include <memory>
+#include <string>
+#include <string_view>
 #include <unordered_map>
 
 #include <gmock/gmock.h>
@@ -1241,7 +1243,7 @@ TEST(DataFileFromJsonTest, MissingSpecId) {
   EXPECT_THAT(result, HasErrorMessage("Missing 'spec-id'"));
 }
 
-TEST(DataFileFromJsonTest, MissingPartition) {
+TEST(DataFileFromJsonTest, MissingPartitionIsAccepted) {
   auto json = R"({
     "content": "data",
     "file-path": "s3://bucket/data/file.parquet",
@@ -1251,9 +1253,9 @@ TEST(DataFileFromJsonTest, MissingPartition) {
     "record-count": 10
   })"_json;
 
-  auto result = DataFileFromJson(json, UnpartitionedSpecs(), Schema({}, 0));
-  EXPECT_THAT(result, IsError(ErrorKind::kJsonParseError));
-  EXPECT_THAT(result, HasErrorMessage("Missing 'partition'"));
+  ICEBERG_UNWRAP_OR_FAIL(auto data_file,
+                         DataFileFromJson(json, UnpartitionedSpecs(), Schema({}, 0)));
+  EXPECT_EQ(data_file.partition.num_fields(), 0U);
 }
 
 TEST(DataFileFromJsonTest, NotAnObject) {
@@ -1262,4 +1264,241 @@ TEST(DataFileFromJsonTest, NotAnObject) {
   EXPECT_THAT(result, HasErrorMessage("DataFile must be a JSON object"));
 }
 
+class DataFileJavaJsonTest : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    ICEBERG_UNWRAP_OR_FAIL(
+        auto spec,
+        PartitionSpec::Make(schema_, /*spec_id=*/0,
+                            {PartitionField(1, 1000, "id", Transform::Identity())},
+                            /*allow_missing_fields=*/false));
+    spec_ = std::shared_ptr<PartitionSpec>(std::move(spec));
+    specs_.emplace(spec_->spec_id(), spec_);
+  }
+
+  DataFile DataFileForTest() const {
+    DataFile data_file;
+    data_file.content = DataFile::Content::kData;
+    data_file.file_path = "/path/to/data.parquet";
+    data_file.file_format = FileFormatType::kParquet;
+    data_file.partition_spec_id = 0;
+    data_file.partition = PartitionValues({Literal::Int(7)});
+    data_file.record_count = 1;
+    data_file.file_size_in_bytes = 10;
+    data_file.lower_bounds = {{1, {0x01, 0x00, 0x00, 0x00}}};
+    data_file.upper_bounds = {{1, {0x05, 0x00, 0x00, 0x00}}};
+    data_file.key_metadata = {0x0A, 0x0B};
+    data_file.sort_order_id = 0;
+    return data_file;
+  }
+
+  nlohmann::json JavaGolden() const {
+    return R"({
+      "spec-id": 0,
+      "content": "data",
+      "file-path": "/path/to/data.parquet",
+      "file-format": "parquet",
+      "partition": [7],
+      "file-size-in-bytes": 10,
+      "record-count": 1,
+      "lower-bounds": {"keys": [1], "values": ["01000000"]},
+      "upper-bounds": {"keys": [1], "values": ["05000000"]},
+      "key-metadata": "0A0B",
+      "sort-order-id": 0
+    })"_json;
+  }
+
+  Schema schema_{{SchemaField::MakeRequired(1, "id", int32())}, 0};
+  std::shared_ptr<PartitionSpec> spec_;
+  std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>> specs_;
+};
+
+TEST_F(DataFileJavaJsonTest, SerializesAndParsesJavaEncoding) {
+  ICEBERG_UNWRAP_OR_FAIL(auto json, ToJson(DataFileForTest(), specs_, schema_));
+  EXPECT_EQ(json, JavaGolden());
+
+  ICEBERG_UNWRAP_OR_FAIL(auto parsed, DataFileFromJson(json, specs_, schema_));
+  EXPECT_EQ(parsed, DataFileForTest());
+}
+
+TEST_F(DataFileJavaJsonTest, ParsesPartitionObject) {
+  auto json = JavaGolden();
+  json["partition"] = {{"1000", 7}};
+
+  ICEBERG_UNWRAP_OR_FAIL(auto parsed, DataFileFromJson(json, specs_, schema_));
+  ASSERT_EQ(parsed.partition.num_fields(), 1U);
+  EXPECT_EQ(parsed.partition.values()[0], Literal::Int(7));
+}
+
+TEST_F(DataFileJavaJsonTest, ParsesAllLegacyContentNames) {
+  struct Case {
+    std::string_view legacy;
+    DataFile::Content content;
+    std::string_view canonical;
+  };
+  for (const auto& test_case :
+       {Case{.legacy = "DATA", .content = DataFile::Content::kData, .canonical = "data"},
+        Case{.legacy = "POSITION_DELETES",
+             .content = DataFile::Content::kPositionDeletes,
+             .canonical = "position-deletes"},
+        Case{.legacy = "EQUALITY_DELETES",
+             .content = DataFile::Content::kEqualityDeletes,
+             .canonical = "equality-deletes"}}) {
+    SCOPED_TRACE(test_case.legacy);
+    auto json = JavaGolden();
+    json["content"] = std::string(test_case.legacy);
+
+    ICEBERG_UNWRAP_OR_FAIL(auto parsed, DataFileFromJson(json, specs_, schema_));
+    EXPECT_EQ(parsed.content, test_case.content);
+    ICEBERG_UNWRAP_OR_FAIL(auto serialized, ToJson(parsed, specs_, schema_));
+    EXPECT_EQ(serialized["content"], std::string(test_case.canonical));
+  }
+}
+
+TEST_F(DataFileJavaJsonTest, ParsesFieldIdPartitionsInSpecOrderAndTypedNulls) {
+  Schema schema{{SchemaField::MakeRequired(1, "id", int32()),
+                 SchemaField::MakeOptional(2, "region", string())},
+                0};
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto spec,
+      PartitionSpec::Make(schema, /*spec_id=*/0,
+                          {PartitionField(2, 1001, "region", Transform::Identity()),
+                           PartitionField(1, 1000, "id", Transform::Identity())},
+                          /*allow_missing_fields=*/false));
+  std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>> specs{
+      {0, std::shared_ptr<PartitionSpec>(std::move(spec))}};
+
+  auto json = JavaGolden();
+  json["partition"] = {{"1000", 7}, {"1001", "north"}};
+  ICEBERG_UNWRAP_OR_FAIL(auto parsed, DataFileFromJson(json, specs, schema));
+  ASSERT_EQ(parsed.partition.num_fields(), 2U);
+  EXPECT_EQ(parsed.partition.values()[0], Literal::String("north"));
+  EXPECT_EQ(parsed.partition.values()[1], Literal::Int(7));
+
+  json["partition"] = {{"1000", 7}};
+  ICEBERG_UNWRAP_OR_FAIL(auto sparse, DataFileFromJson(json, specs, schema));
+  ASSERT_EQ(sparse.partition.num_fields(), 2U);
+  EXPECT_TRUE(sparse.partition.values()[0].IsNull());
+  ASSERT_NE(sparse.partition.values()[0].type(), nullptr);
+  EXPECT_EQ(sparse.partition.values()[0].type()->type_id(), TypeId::kString);
+  EXPECT_EQ(sparse.partition.values()[1], Literal::Int(7));
+
+  json["partition"] = nlohmann::json::array({nullptr, 7});
+  ICEBERG_UNWRAP_OR_FAIL(auto positional, DataFileFromJson(json, specs, schema));
+  ASSERT_EQ(positional.partition.num_fields(), 2U);
+  EXPECT_TRUE(positional.partition.values()[0].IsNull());
+  ASSERT_NE(positional.partition.values()[0].type(), nullptr);
+  EXPECT_EQ(positional.partition.values()[0].type()->type_id(), TypeId::kString);
+  EXPECT_EQ(positional.partition.values()[1], Literal::Int(7));
+}
+
+TEST_F(DataFileJavaJsonTest, RejectsMalformedPartitionShapes) {
+  for (const auto& partition :
+       {nlohmann::json::array(), nlohmann::json::object({{"1000", 7}, {"9999", 8}}),
+        nlohmann::json("invalid")}) {
+    SCOPED_TRACE(partition.dump());
+    auto json = JavaGolden();
+    json["partition"] = partition;
+    EXPECT_THAT(DataFileFromJson(json, specs_, schema_),
+                IsError(ErrorKind::kJsonParseError));
+  }
+}
+
+TEST_F(DataFileJavaJsonTest, RejectsInvalidIntegersAndDuplicateMetricKeys) {
+  for (std::string_view field :
+       {"spec-id", "file-size-in-bytes", "record-count", "sort-order-id", "first-row-id",
+        "content-offset", "content-size-in-bytes"}) {
+    SCOPED_TRACE(field);
+    auto json = JavaGolden();
+    json[field] = 0.5;
+    auto result = DataFileFromJson(json, specs_, schema_);
+    EXPECT_THAT(result, IsError(ErrorKind::kJsonParseError));
+    EXPECT_THAT(result, HasErrorMessage(field));
+  }
+
+  auto expect_invalid = [&](nlohmann::json json, std::string_view field) {
+    auto result = DataFileFromJson(json, specs_, schema_);
+    EXPECT_THAT(result, IsError(ErrorKind::kJsonParseError));
+    EXPECT_THAT(result, HasErrorMessage(field));
+  };
+
+  auto fractional_key = JavaGolden();
+  fractional_key["column-sizes"] = {{"keys", {0.5}}, {"values", {10}}};
+  expect_invalid(fractional_key, "column-sizes");
+
+  auto fractional_value = JavaGolden();
+  fractional_value["column-sizes"] = {{"keys", {1}}, {"values", {0.5}}};
+  expect_invalid(fractional_value, "column-sizes");
+
+  auto fractional_bound_key = JavaGolden();
+  fractional_bound_key["lower-bounds"]["keys"] = {0.5};
+  expect_invalid(fractional_bound_key, "lower-bounds");
+
+  for (std::string_view field : {"split-offsets", "equality-ids"}) {
+    auto json = JavaGolden();
+    json[field] = {0.5};
+    expect_invalid(std::move(json), field);
+  }
+
+  auto oversized_spec_id = JavaGolden();
+  oversized_spec_id["spec-id"] = 2147483648ULL;
+  expect_invalid(oversized_spec_id, "spec-id");
+
+  auto oversized_record_count = JavaGolden();
+  oversized_record_count["record-count"] = 18446744073709551615ULL;
+  expect_invalid(oversized_record_count, "record-count");
+
+  auto duplicate_keys = JavaGolden();
+  duplicate_keys["column-sizes"] = {{"keys", {1, 1}}, {"values", {10, 20}}};
+  auto result = DataFileFromJson(duplicate_keys, specs_, schema_);
+  EXPECT_THAT(result, IsError(ErrorKind::kJsonParseError));
+  EXPECT_THAT(result, HasErrorMessage("duplicate key"));
+
+  auto duplicate_bounds = JavaGolden();
+  duplicate_bounds["lower-bounds"] = {{"keys", {1, 1}}, {"values", {"01", "02"}}};
+  auto bounds_result = DataFileFromJson(duplicate_bounds, specs_, schema_);
+  EXPECT_THAT(bounds_result, IsError(ErrorKind::kJsonParseError));
+  EXPECT_THAT(bounds_result, HasErrorMessage("duplicate key"));
+}
+
+TEST_F(DataFileJavaJsonTest, RejectsMalformedHexMetadata) {
+  auto bad_bound = JavaGolden();
+  bad_bound["lower-bounds"]["values"] = {"0G"};
+  EXPECT_THAT(DataFileFromJson(bad_bound, specs_, schema_),
+              IsError(ErrorKind::kInvalidArgument));
+
+  auto non_string_bound = JavaGolden();
+  non_string_bound["lower-bounds"]["values"] = {7};
+  EXPECT_THAT(DataFileFromJson(non_string_bound, specs_, schema_),
+              IsError(ErrorKind::kJsonParseError));
+
+  auto bad_key_metadata = JavaGolden();
+  bad_key_metadata["key-metadata"] = "0G";
+  EXPECT_THAT(DataFileFromJson(bad_key_metadata, specs_, schema_),
+              IsError(ErrorKind::kInvalidArgument));
+}
+
+TEST_F(DataFileJavaJsonTest, AcceptsMissingPartitionAndNullMetricMaps) {
+  auto json = JavaGolden();
+  json.erase("partition");
+  for (std::string_view field : {"column-sizes", "value-counts", "null-value-counts",
+                                 "nan-value-counts", "lower-bounds", "upper-bounds"}) {
+    json[field] = nullptr;
+  }
+
+  ICEBERG_UNWRAP_OR_FAIL(auto parsed, DataFileFromJson(json, specs_, schema_));
+  EXPECT_EQ(parsed.partition.num_fields(), 0U);
+  EXPECT_TRUE(parsed.column_sizes.empty());
+  EXPECT_TRUE(parsed.lower_bounds.empty());
+}
+
+TEST_F(DataFileJavaJsonTest, RejectsInvalidPartitionTypeWhenSerializing) {
+  auto data_file = DataFileForTest();
+  data_file.partition = PartitionValues({Literal::String("not-an-int")});
+
+  auto result = ToJson(data_file, specs_, schema_);
+  EXPECT_THAT(result, IsError(ErrorKind::kValidationFailed));
+  EXPECT_THAT(result, HasErrorMessage("partition value type"));
+}
+
 }  // namespace iceberg
diff --git a/src/iceberg/test/rest_json_serde_test.cc b/src/iceberg/test/rest_json_serde_test.cc
index d0a69c053..8d2462028 100644
--- a/src/iceberg/test/rest_json_serde_test.cc
+++ b/src/iceberg/test/rest_json_serde_test.cc
@@ -19,7 +19,9 @@
 
 #include <memory>
 #include <string>
+#include <string_view>
 #include <unordered_map>
+#include <utility>
 #include <vector>
 
 #include <gmock/gmock.h>