diff --git a/src/iceberg/avro/avro_data_util.cc b/src/iceberg/avro/avro_data_util.cc
index 39f2948b9..8e460b68d 100644
--- a/src/iceberg/avro/avro_data_util.cc
+++ b/src/iceberg/avro/avro_data_util.cc
@@ -94,7 +94,8 @@ Status AppendStructToBuilder(const ::avro::NodePtr& avro_node,
                                                      metadata_context, field_builder));
     } else if (field_projection.kind == FieldProjection::Kind::kNull) {
       ICEBERG_ARROW_RETURN_NOT_OK(field_builder->AppendNull());
-    } else if (field_projection.kind == FieldProjection::Kind::kDefault) {
+    } else if (field_projection.kind == FieldProjection::Kind::kDefault ||
+               field_projection.kind == FieldProjection::Kind::kConstant) {
       ICEBERG_RETURN_UNEXPECTED(AppendDefaultToBuilder(field_projection, field_builder));
     } else if (field_projection.kind == FieldProjection::Kind::kMetadata) {
       int32_t field_id = expected_field.field_id();
@@ -476,7 +477,8 @@ Status AppendFieldToBuilder(const ::avro::NodePtr& avro_node,
     return {};
   }
 
-  if (projection.kind == FieldProjection::Kind::kDefault) {
+  if (projection.kind == FieldProjection::Kind::kDefault ||
+      projection.kind == FieldProjection::Kind::kConstant) {
     return AppendDefaultToBuilder(projection, array_builder);
   }
 
@@ -593,7 +595,8 @@ Status PrepareStructDefaultScalars(std::span<FieldProjection> projections,
     auto& field_projection = projections[i];
     auto* field_builder = struct_builder->field_builder(static_cast<int>(i));
 
-    if (field_projection.kind == FieldProjection::Kind::kDefault) {
+    if (field_projection.kind == FieldProjection::Kind::kDefault ||
+        field_projection.kind == FieldProjection::Kind::kConstant) {
       // Get-or-create the single Avro attributes container: another Avro attribute may
       // have created it already, so guard on `default_scalar` being unset rather than on
       // the container's presence (otherwise a pre-existing container would skip
diff --git a/src/iceberg/avro/avro_direct_decoder.cc b/src/iceberg/avro/avro_direct_decoder.cc
index 465bc4b24..7bff2768d 100644
--- a/src/iceberg/avro/avro_direct_decoder.cc
+++ b/src/iceberg/avro/avro_direct_decoder.cc
@@ -216,7 +216,8 @@ Status DecodeStructToBuilder(const ::avro::NodePtr& avro_node, ::avro::Decoder&
     auto* field_builder = struct_builder->field_builder(static_cast<int>(proj_idx));
     if (field_projection.kind == FieldProjection::Kind::kNull) {
       ICEBERG_ARROW_RETURN_NOT_OK(field_builder->AppendNull());
-    } else if (field_projection.kind == FieldProjection::Kind::kDefault) {
+    } else if (field_projection.kind == FieldProjection::Kind::kDefault ||
+               field_projection.kind == FieldProjection::Kind::kConstant) {
       ICEBERG_RETURN_UNEXPECTED(AppendDefaultToBuilder(field_projection, field_builder));
     } else if (field_projection.kind == FieldProjection::Kind::kMetadata) {
       int32_t field_id = expected_field.field_id();
diff --git a/src/iceberg/avro/avro_reader.cc b/src/iceberg/avro/avro_reader.cc
index f333a38cb..416de6e08 100644
--- a/src/iceberg/avro/avro_reader.cc
+++ b/src/iceberg/avro/avro_reader.cc
@@ -329,8 +329,9 @@ class AvroReader::Impl {
     }
 
     // Project the read schema on top of the file schema.
-    ICEBERG_ASSIGN_OR_RAISE(projection_, Project(*read_schema_, file_schema.root(),
-                                                 /*prune_source=*/false));
+    ICEBERG_ASSIGN_OR_RAISE(projection_,
+                            Project(*read_schema_, file_schema.root(),
+                                    /*prune_source=*/false, options.id_to_constant));
 
     ICEBERG_RETURN_UNEXPECTED(backend_->InitWithSchema(file_schema, options.split));
 
diff --git a/src/iceberg/avro/avro_schema_util.cc b/src/iceberg/avro/avro_schema_util.cc
index 94d0a6116..819dbf975 100644
--- a/src/iceberg/avro/avro_schema_util.cc
+++ b/src/iceberg/avro/avro_schema_util.cc
@@ -684,13 +684,14 @@ Status UnwrapUnion(const ::avro::NodePtr& node, ::avro::NodePtr* result) {
 }
 
 // Forward declaration
-Result<FieldProjection> ProjectNested(const Type& expected_type,
-                                      const ::avro::NodePtr& avro_node,
-                                      bool prune_source);
-
-Result<FieldProjection> ProjectField(const SchemaField& expected_field,
-                                     const ::avro::NodePtr& avro_node,
-                                     size_t source_index, bool prune_source) {
+Result<FieldProjection> ProjectNested(
+    const Type& expected_type, const ::avro::NodePtr& avro_node, bool prune_source,
+    const std::unordered_map<int32_t, Literal>& id_to_constant);
+
+Result<FieldProjection> ProjectField(
+    const SchemaField& expected_field, const ::avro::NodePtr& avro_node,
+    size_t source_index, bool prune_source,
+    const std::unordered_map<int32_t, Literal>& id_to_constant) {
   const Type& expected_type = *expected_field.type();
   ::avro::NodePtr field_node;
   ICEBERG_RETURN_UNEXPECTED(UnwrapUnion(avro_node, &field_node));
@@ -707,8 +708,8 @@ Result<FieldProjection> ProjectField(const SchemaField& expected_field,
   }
 
   if (expected_type.is_nested()) {
-    ICEBERG_ASSIGN_OR_RAISE(projection,
-                            ProjectNested(expected_type, field_node, prune_source));
+    ICEBERG_ASSIGN_OR_RAISE(projection, ProjectNested(expected_type, field_node,
+                                                      prune_source, id_to_constant));
   } else {
     ICEBERG_RETURN_UNEXPECTED(ValidateAvroSchemaEvolution(expected_type, field_node));
   }
@@ -717,9 +718,9 @@ Result<FieldProjection> ProjectField(const SchemaField& expected_field,
   return projection;
 }
 
-Result<FieldProjection> ProjectStruct(const StructType& struct_type,
-                                      const ::avro::NodePtr& avro_node,
-                                      bool prune_source) {
+Result<FieldProjection> ProjectStruct(
+    const StructType& struct_type, const ::avro::NodePtr& avro_node, bool prune_source,
+    const std::unordered_map<int32_t, Literal>& id_to_constant) {
   if (avro_node->type() != ::avro::AVRO_RECORD) {
     return InvalidSchema("Expected AVRO_RECORD type, but got {}", ToString(avro_node));
   }
@@ -744,17 +745,21 @@ Result<FieldProjection> ProjectStruct(const StructType& struct_type,
     }
   }
 
-  FieldProjection result;
+  FieldProjection result{.kind = FieldProjection::Kind::kProjected};
   result.children.reserve(expected_fields.size());
 
   for (const auto& expected_field : expected_fields) {
     int32_t field_id = expected_field.field_id();
     FieldProjection child_projection;
 
-    if (auto iter = node_info_map.find(field_id); iter != node_info_map.cend()) {
-      ICEBERG_ASSIGN_OR_RAISE(child_projection,
-                              ProjectField(expected_field, iter->second.field_node,
-                                           iter->second.local_index, prune_source));
+    if (auto constant = id_to_constant.find(field_id); constant != id_to_constant.end()) {
+      child_projection.kind = FieldProjection::Kind::kConstant;
+      child_projection.from = constant->second;
+    } else if (auto iter = node_info_map.find(field_id); iter != node_info_map.cend()) {
+      ICEBERG_ASSIGN_OR_RAISE(
+          child_projection,
+          ProjectField(expected_field, iter->second.field_node, iter->second.local_index,
+                       prune_source, id_to_constant));
     } else if (MetadataColumns::IsMetadataColumn(field_id)) {
       child_projection.kind = FieldProjection::Kind::kMetadata;
     } else if (expected_field.initial_default() != nullptr) {
@@ -777,8 +782,9 @@ Result<FieldProjection> ProjectStruct(const StructType& struct_type,
   return result;
 }
 
-Result<FieldProjection> ProjectList(const ListType& list_type,
-                                    const ::avro::NodePtr& avro_node, bool prune_source) {
+Result<FieldProjection> ProjectList(
+    const ListType& list_type, const ::avro::NodePtr& avro_node, bool prune_source,
+    const std::unordered_map<int32_t, Literal>& id_to_constant) {
   if (avro_node->type() != ::avro::AVRO_ARRAY) {
     return InvalidSchema("Expected AVRO_ARRAY type, but got {}", ToString(avro_node));
   }
@@ -795,21 +801,23 @@ Result<FieldProjection> ProjectList(const ListType& list_type,
   }
 
   FieldProjection element_projection;
-  ICEBERG_ASSIGN_OR_RAISE(element_projection,
-                          ProjectField(expected_element_field, avro_node->leafAt(0),
-                                       /*source_index*/ size_t{0}, prune_source));
+  ICEBERG_ASSIGN_OR_RAISE(
+      element_projection,
+      ProjectField(expected_element_field, avro_node->leafAt(0),
+                   /*source_index*/ size_t{0}, prune_source, id_to_constant));
 
-  FieldProjection result;
+  FieldProjection result{.kind = FieldProjection::Kind::kProjected};
   result.children.emplace_back(std::move(element_projection));
   return result;
 }
 
-Result<FieldProjection> ProjectMap(const MapType& map_type,
-                                   const ::avro::NodePtr& avro_node, bool prune_source) {
+Result<FieldProjection> ProjectMap(
+    const MapType& map_type, const ::avro::NodePtr& avro_node, bool prune_source,
+    const std::unordered_map<int32_t, Literal>& id_to_constant) {
   const auto& expected_key_field = map_type.key();
   const auto& expected_value_field = map_type.value();
 
-  FieldProjection result;
+  FieldProjection result{.kind = FieldProjection::Kind::kProjected};
   int32_t avro_key_id;
   int32_t avro_value_id;
   ::avro::NodePtr map_node;
@@ -854,18 +862,18 @@ Result<FieldProjection> ProjectMap(const MapType& map_type,
   for (size_t i = 0; i < map_node->leaves(); ++i) {
     FieldProjection sub_projection;
     const auto& expected_sub_field = map_type.fields()[i];
-    ICEBERG_ASSIGN_OR_RAISE(
-        sub_projection,
-        ProjectField(expected_sub_field, map_node->leafAt(i), i, prune_source));
+    ICEBERG_ASSIGN_OR_RAISE(sub_projection,
+                            ProjectField(expected_sub_field, map_node->leafAt(i), i,
+                                         prune_source, id_to_constant));
     result.children.emplace_back(std::move(sub_projection));
   }
 
   return result;
 }
 
-Result<FieldProjection> ProjectNested(const Type& expected_type,
-                                      const ::avro::NodePtr& avro_node,
-                                      bool prune_source) {
+Result<FieldProjection> ProjectNested(
+    const Type& expected_type, const ::avro::NodePtr& avro_node, bool prune_source,
+    const std::unordered_map<int32_t, Literal>& id_to_constant) {
   if (!expected_type.is_nested()) {
     return InvalidSchema("Expected a nested type, but got {}", expected_type);
   }
@@ -873,13 +881,13 @@ Result<FieldProjection> ProjectNested(const Type& expected_type,
   switch (expected_type.type_id()) {
     case TypeId::kStruct:
       return ProjectStruct(internal::checked_cast<const StructType&>(expected_type),
-                           avro_node, prune_source);
+                           avro_node, prune_source, id_to_constant);
     case TypeId::kList:
       return ProjectList(internal::checked_cast<const ListType&>(expected_type),
-                         avro_node, prune_source);
+                         avro_node, prune_source, id_to_constant);
     case TypeId::kMap:
       return ProjectMap(internal::checked_cast<const MapType&>(expected_type), avro_node,
-                        prune_source);
+                        prune_source, id_to_constant);
     default:
       return InvalidSchema("Unsupported nested type: {}", expected_type);
   }
@@ -893,11 +901,12 @@ bool HasMapLogicalType(const ::avro::NodePtr& node) {
          node->logicalType().customLogicalType()->name() == "map";
 }
 
-Result<SchemaProjection> Project(const Schema& expected_schema,
-                                 const ::avro::NodePtr& avro_node, bool prune_source) {
-  ICEBERG_ASSIGN_OR_RAISE(
-      auto field_projection,
-      ProjectNested(static_cast<const Type&>(expected_schema), avro_node, prune_source));
+Result<SchemaProjection> Project(
+    const Schema& expected_schema, const ::avro::NodePtr& avro_node, bool prune_source,
+    const std::unordered_map<int32_t, Literal>& id_to_constant) {
+  ICEBERG_ASSIGN_OR_RAISE(auto field_projection,
+                          ProjectNested(static_cast<const Type&>(expected_schema),
+                                        avro_node, prune_source, id_to_constant));
   return SchemaProjection{std::move(field_projection.children)};
 }
 
diff --git a/src/iceberg/avro/avro_schema_util_internal.h b/src/iceberg/avro/avro_schema_util_internal.h
index 342b119a5..8d5d420ca 100644
--- a/src/iceberg/avro/avro_schema_util_internal.h
+++ b/src/iceberg/avro/avro_schema_util_internal.h
@@ -142,9 +142,11 @@ class HasIdVisitor {
 /// \param expected_schema The Iceberg Schema that defines the expected structure.
 /// \param avro_node The Avro node to read data from.
 /// \param prune_source Whether the source schema can be pruned.
+/// \param id_to_constant File-level values that override physical fields and defaults.
 /// \return The schema projection result.
-Result<SchemaProjection> Project(const Schema& expected_schema,
-                                 const ::avro::NodePtr& avro_node, bool prune_source);
+Result<SchemaProjection> Project(
+    const Schema& expected_schema, const ::avro::NodePtr& avro_node, bool prune_source,
+    const std::unordered_map<int32_t, Literal>& id_to_constant = {});
 
 std::string ToString(const ::avro::NodePtr& node);
 std::string ToString(const ::avro::LogicalType& logical_type);
diff --git a/src/iceberg/catalog/rest/json_serde.cc b/src/iceberg/catalog/rest/json_serde.cc
index 7b862e729..23de188fd 100644
--- a/src/iceberg/catalog/rest/json_serde.cc
+++ b/src/iceberg/catalog/rest/json_serde.cc
@@ -231,9 +231,16 @@ Result<std::vector<std::shared_ptr<FileScanTask>>> FileScanTasksFromJson(
       ICEBERG_ASSIGN_OR_RAISE(residual_filter, ExpressionFromJson(filter_json));
     }
 
+    std::shared_ptr<PartitionSpec> spec;
+    if (data_file.partition_spec_id.has_value()) {
+      auto it = partition_spec_by_id.find(*data_file.partition_spec_id);
+      if (it != partition_spec_by_id.end()) {
+        spec = it->second;
+      }
+    }
     auto task = FileScanTask::MakeSplit(std::make_shared<DataFile>(std::move(data_file)),
                                         start, length, std::move(task_delete_files),
-                                        std::move(residual_filter));
+                                        std::move(residual_filter), std::move(spec));
     if (!task) {
       return JsonParseError("Invalid FileScanTask range: {}", task.error().message);
     }
diff --git a/src/iceberg/catalog/rest/rest_table_scan.cc b/src/iceberg/catalog/rest/rest_table_scan.cc
index fa2bc6737..9ef9cdb74 100644
--- a/src/iceberg/catalog/rest/rest_table_scan.cc
+++ b/src/iceberg/catalog/rest/rest_table_scan.cc
@@ -286,9 +286,9 @@ class RestFileScanTaskStream final : public FileScanTaskStream {
           // The server may leave row filtering to the client.
           ICEBERG_ASSIGN_OR_RAISE(
               task,
-              FileScanTask::MakeSplit(task->data_file(), task->start(), task->length(),
-                                      task->delete_files(),
-                                      ignore_residuals_ ? True::Instance() : filter_));
+              FileScanTask::MakeSplit(
+                  task->data_file(), task->start(), task->length(), task->delete_files(),
+                  ignore_residuals_ ? True::Instance() : filter_, task->spec()));
         }
         return task;
       }
diff --git a/src/iceberg/data/file_scan_task_reader.cc b/src/iceberg/data/file_scan_task_reader.cc
index bf23c5c36..54307950e 100644
--- a/src/iceberg/data/file_scan_task_reader.cc
+++ b/src/iceberg/data/file_scan_task_reader.cc
@@ -29,23 +29,48 @@
 #include "iceberg/data/delete_filter.h"
 #include "iceberg/file_reader.h"
 #include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/partition_spec.h"
 #include "iceberg/schema.h"
 #include "iceberg/schema_internal.h"
 #include "iceberg/table_scan.h"
+#include "iceberg/transform.h"
 #include "iceberg/util/macros.h"
 
 namespace iceberg {
 
 namespace {
 
-ReaderOptions MakeReaderOptions(const FileScanTask& task, std::shared_ptr<FileIO> io,
-                                std::shared_ptr<Schema> projection,
-                                std::shared_ptr<Expression> filter, bool case_sensitive,
-                                std::shared_ptr<NameMapping> name_mapping,
-                                ReaderProperties properties,
-                                std::optional<int64_t> first_row_id,
-                                std::optional<int64_t> data_sequence_number) {
+Result<std::unordered_map<int32_t, Literal>> ConstantsOf(const FileScanTask& task) {
+  const auto& file = *task.data_file();
+  const auto& spec = task.spec();
+  std::unordered_map<int32_t, Literal> id_to_constant;
+  if (!spec) {
+    if (file.partition.num_fields() != 0) {
+      return InvalidArgument("A partition spec is required to read partition values");
+    }
+    return id_to_constant;
+  }
+  for (size_t i = 0; i < spec->fields().size(); ++i) {
+    const auto& field = spec->fields()[i];
+    if (field.transform() == nullptr) {
+      return InvalidArgument("Missing transform for partition field {}",
+                             field.field_id());
+    }
+    if (field.transform()->transform_type() == TransformType::kIdentity) {
+      id_to_constant.emplace(field.source_id(), file.partition.values()[i]);
+    }
+  }
+  return id_to_constant;
+}
+
+Result<ReaderOptions> MakeReaderOptions(
+    const FileScanTask& task, std::shared_ptr<FileIO> io,
+    std::shared_ptr<Schema> projection, std::shared_ptr<Expression> filter,
+    bool case_sensitive, std::shared_ptr<NameMapping> name_mapping,
+    ReaderProperties properties, std::optional<int64_t> first_row_id,
+    std::optional<int64_t> data_sequence_number) {
   const auto& data_file = *task.data_file();
+  ICEBERG_ASSIGN_OR_RAISE(auto id_to_constant, ConstantsOf(task));
   return ReaderOptions{
       .path = data_file.file_path,
       .length = static_cast<size_t>(data_file.file_size_in_bytes),
@@ -61,6 +86,7 @@ ReaderOptions MakeReaderOptions(const FileScanTask& task, std::shared_ptr<FileIO
       .first_row_id = first_row_id,
       .data_sequence_number = data_sequence_number,
       .properties = std::move(properties),
+      .id_to_constant = std::move(id_to_constant),
   };
 }
 
@@ -191,9 +217,11 @@ class FileScanTaskReader::Impl {
     auto filter = task.residual_filter();
 
     if (task.delete_files().empty()) {
-      auto options = MakeReaderOptions(
-          task, io_, projected_schema_, filter, case_sensitive_, name_mapping_,
-          properties_, data_file->first_row_id, data_file->data_sequence_number);
+      ICEBERG_ASSIGN_OR_RAISE(
+          auto options,
+          MakeReaderOptions(task, io_, projected_schema_, filter, case_sensitive_,
+                            name_mapping_, properties_, data_file->first_row_id,
+                            data_file->data_sequence_number));
       ICEBERG_ASSIGN_OR_RAISE(
           auto reader, ReaderFactoryRegistry::Open(data_file->file_format, options));
       return MakeArrowArrayStream(std::move(reader));
@@ -217,9 +245,11 @@ class FileScanTaskReader::Impl {
                             ProjectionContext::Make(*required_schema, *projected_schema_,
                                                     project_batch_function));
 
-    auto options = MakeReaderOptions(task, io_, required_schema, filter, case_sensitive_,
-                                     name_mapping_, properties_, data_file->first_row_id,
-                                     data_file->data_sequence_number);
+    ICEBERG_ASSIGN_OR_RAISE(
+        auto options,
+        MakeReaderOptions(task, io_, required_schema, filter, case_sensitive_,
+                          name_mapping_, properties_, data_file->first_row_id,
+                          data_file->data_sequence_number));
     ICEBERG_ASSIGN_OR_RAISE(auto reader,
                             ReaderFactoryRegistry::Open(data_file->file_format, options));
 
diff --git a/src/iceberg/file_reader.h b/src/iceberg/file_reader.h
index 13e8b3ad2..7f597a77e 100644
--- a/src/iceberg/file_reader.h
+++ b/src/iceberg/file_reader.h
@@ -26,8 +26,10 @@
 #include <functional>
 #include <memory>
 #include <optional>
+#include <unordered_map>
 
 #include "iceberg/arrow_c_data.h"
+#include "iceberg/expression/literal.h"
 #include "iceberg/file_format.h"
 #include "iceberg/result.h"
 #include "iceberg/type_fwd.h"
@@ -121,6 +123,12 @@ struct ICEBERG_EXPORT ReaderOptions {
   std::optional<int64_t> data_sequence_number;
   /// \brief Format-specific or implementation-specific properties.
   ReaderProperties properties;
+
+  /// File-level values keyed by field ID, overriding physical columns,
+  /// initial defaults and optional-field nulls, matching Iceberg Java.
+  /// Example: `options.id_to_constant.emplace(2, Literal::String("CN"));`
+  // TODO(yzsun): Use a nested value representation if struct constants are needed.
+  std::unordered_map<int32_t, Literal> id_to_constant;
 };
 
 /// \brief Factory function to create a reader of a specific file format.
diff --git a/src/iceberg/json_serde.cc b/src/iceberg/json_serde.cc
index 37f6c08c9..6174db7e2 100644
--- a/src/iceberg/json_serde.cc
+++ b/src/iceberg/json_serde.cc
@@ -2551,7 +2551,7 @@ Result<std::shared_ptr<FileScanTask>> FileScanTaskFromJson(const nlohmann::json&
 
   auto task = FileScanTask::MakeSplit(std::make_shared<DataFile>(std::move(data_file)),
                                       start, length, std::move(delete_files),
-                                      std::move(residual_filter));
+                                      std::move(residual_filter), std::move(shared_spec));
   if (!task) {
     return JsonParseError("Invalid FileScanTask range: {}", task.error().message);
   }
diff --git a/src/iceberg/manifest/manifest_group.cc b/src/iceberg/manifest/manifest_group.cc
index 036a71bdc..883fd770e 100644
--- a/src/iceberg/manifest/manifest_group.cc
+++ b/src/iceberg/manifest/manifest_group.cc
@@ -320,7 +320,8 @@ class ManifestGroup::FilePlanningStream final : public FileScanTaskStream {
                               residuals->ResidualFor(value.data_file->partition));
 
       return std::optional<std::shared_ptr<FileScanTask>>{std::make_shared<FileScanTask>(
-          std::move(value.data_file), std::move(delete_files), std::move(residual))};
+          std::move(value.data_file), std::move(delete_files), std::move(residual),
+          group_->specs_by_id_.at(spec_id))};
     }
   }
 
diff --git a/src/iceberg/parquet/parquet_data_util.cc b/src/iceberg/parquet/parquet_data_util.cc
index 73597cbfb..fbb8e9008 100644
--- a/src/iceberg/parquet/parquet_data_util.cc
+++ b/src/iceberg/parquet/parquet_data_util.cc
@@ -251,7 +251,8 @@ Result<std::shared_ptr<::arrow::Array>> ProjectStructArray(
       ICEBERG_ASSIGN_OR_RAISE(
           projected_array,
           MakeNullArray(output_arrow_type, struct_array->length(), pool));
-    } else if (field_projection.kind == FieldProjection::Kind::kDefault) {
+    } else if (field_projection.kind == FieldProjection::Kind::kDefault ||
+               field_projection.kind == FieldProjection::Kind::kConstant) {
       ICEBERG_ASSIGN_OR_RAISE(
           projected_array,
           arrow::MakeDefaultArray(std::get<Literal>(field_projection.from),
diff --git a/src/iceberg/parquet/parquet_reader.cc b/src/iceberg/parquet/parquet_reader.cc
index b88dfd000..85f0c9b38 100644
--- a/src/iceberg/parquet/parquet_reader.cc
+++ b/src/iceberg/parquet/parquet_reader.cc
@@ -62,9 +62,10 @@ Result<std::shared_ptr<::arrow::io::RandomAccessFile>> OpenInputStream(
   return arrow::OpenArrowInputStream(options.io, options.path, options.length);
 }
 
-Result<SchemaProjection> BuildProjection(::parquet::arrow::FileReader* reader,
-                                         const Schema& read_schema,
-                                         const NameMapping* name_mapping) {
+Result<SchemaProjection> BuildProjection(
+    ::parquet::arrow::FileReader* reader, const Schema& read_schema,
+    const NameMapping* name_mapping,
+    const std::unordered_map<int32_t, Literal>& id_to_constant) {
   auto metadata = reader->parquet_reader()->metadata();
 
   if (!HasFieldIds(metadata->schema()->schema_root())) {
@@ -77,10 +78,11 @@ Result<SchemaProjection> BuildProjection(::parquet::arrow::FileReader* reader,
         metadata->schema(), metadata->key_value_metadata(), reader->properties(),
         &schema_manifest));
     ApplyNameMapping(*name_mapping, &schema_manifest);
-    return Project(read_schema, schema_manifest);
+    return Project(read_schema, schema_manifest, id_to_constant);
   }
 
-  ICEBERG_ASSIGN_OR_RAISE(auto projection, Project(read_schema, reader->manifest()));
+  ICEBERG_ASSIGN_OR_RAISE(auto projection,
+                          Project(read_schema, reader->manifest(), id_to_constant));
   return projection;
 }
 
@@ -305,8 +307,9 @@ class ParquetReader::Impl {
                                                     arrow_reader_properties));
 
     // Project read schema onto the Parquet file schema
-    ICEBERG_ASSIGN_OR_RAISE(projection_, BuildProjection(reader_.get(), *read_schema_,
-                                                         options.name_mapping.get()));
+    ICEBERG_ASSIGN_OR_RAISE(
+        projection_, BuildProjection(reader_.get(), *read_schema_,
+                                     options.name_mapping.get(), options.id_to_constant));
     if (options.filter &&
         options.properties.Get(ReaderProperties::kParquetRowGroupFilter)) {
       auto filter = options.filter;
diff --git a/src/iceberg/parquet/parquet_schema_util.cc b/src/iceberg/parquet/parquet_schema_util.cc
index d16b8a1ae..21e1f4b5c 100644
--- a/src/iceberg/parquet/parquet_schema_util.cc
+++ b/src/iceberg/parquet/parquet_schema_util.cc
@@ -504,7 +504,9 @@ namespace {
 
 class ProjectionBuilder {
  public:
-  explicit ProjectionBuilder(const ::parquet::SchemaDescriptor* descr) : descr_(descr) {}
+  ProjectionBuilder(const ::parquet::SchemaDescriptor* descr,
+                    const std::unordered_map<int32_t, Literal>& id_to_constant)
+      : descr_(descr), id_to_constant_(id_to_constant) {}
 
   Result<FieldProjection> Project(
       const Type& nested_type,
@@ -587,8 +589,12 @@ class ProjectionBuilder {
       int32_t field_id = field.field_id();
       FieldProjection child_projection;
 
-      if (auto iter = field_context_map.find(field_id);
-          iter != field_context_map.cend()) {
+      if (auto constant = id_to_constant_.find(field_id);
+          constant != id_to_constant_.end()) {
+        child_projection.kind = FieldProjection::Kind::kConstant;
+        child_projection.from = constant->second;
+      } else if (auto iter = field_context_map.find(field_id);
+                 iter != field_context_map.cend()) {
         const auto& parquet_field = iter->second.parquet_field;
         ICEBERG_ASSIGN_OR_RAISE(child_projection, ProjectField(field, parquet_field,
                                                                iter->second.local_index));
@@ -719,6 +725,7 @@ class ProjectionBuilder {
   }
 
   const ::parquet::SchemaDescriptor* descr_;
+  const std::unordered_map<int32_t, Literal>& id_to_constant_;
 };
 
 void CollectColumnIds(const FieldProjection& field_projection,
@@ -783,9 +790,10 @@ void ApplyNameMappingToField(const NameMapping& mapping, std::vector<std::string
 
 }  // namespace
 
-Result<SchemaProjection> Project(const Schema& expected_schema,
-                                 const ::parquet::arrow::SchemaManifest& parquet_schema) {
-  ProjectionBuilder builder(parquet_schema.descr);
+Result<SchemaProjection> Project(
+    const Schema& expected_schema, const ::parquet::arrow::SchemaManifest& parquet_schema,
+    const std::unordered_map<int32_t, Literal>& id_to_constant) {
+  ProjectionBuilder builder(parquet_schema.descr, id_to_constant);
   ICEBERG_ASSIGN_OR_RAISE(auto field_projection,
                           builder.Project(static_cast<const Type&>(expected_schema),
                                           parquet_schema.schema_fields));
diff --git a/src/iceberg/parquet/parquet_schema_util_internal.h b/src/iceberg/parquet/parquet_schema_util_internal.h
index 49c62ff09..18c4ce630 100644
--- a/src/iceberg/parquet/parquet_schema_util_internal.h
+++ b/src/iceberg/parquet/parquet_schema_util_internal.h
@@ -54,10 +54,12 @@ Status ValidateParquetTypeCompatibility(
 ///
 /// \param expected_schema The Iceberg Schema that defines the expected structure.
 /// \param parquet_schema The Parquet schema to read data from.
+/// \param id_to_constant File-level values that override physical fields and defaults.
 /// \return The schema projection result with column indices of projected Parquet columns
 /// specified via ParquetExtraAttributes.
-Result<SchemaProjection> Project(const Schema& expected_schema,
-                                 const ::parquet::arrow::SchemaManifest& parquet_schema);
+Result<SchemaProjection> Project(
+    const Schema& expected_schema, const ::parquet::arrow::SchemaManifest& parquet_schema,
+    const std::unordered_map<int32_t, Literal>& id_to_constant = {});
 
 /// \brief Get the selected column indices by walking through the projection result.
 ///
diff --git a/src/iceberg/schema_util.cc b/src/iceberg/schema_util.cc
index d7d261757..1499aa631 100644
--- a/src/iceberg/schema_util.cc
+++ b/src/iceberg/schema_util.cc
@@ -82,12 +82,14 @@ Status ValidateSchemaEvolution(const Type& expected_type, const Type& source_typ
   return NotSupported("Cannot read {} from {}", expected_type, source_type);
 }
 
-Result<FieldProjection> ProjectNested(const Type& expected_type, const Type& source_type,
-                                      bool prune_source, ProjectionMode mode);
+Result<FieldProjection> ProjectNested(
+    const Type& expected_type, const Type& source_type, bool prune_source,
+    ProjectionMode mode, const std::unordered_map<int32_t, Literal>& id_to_constant);
 
-Result<FieldProjection> ProjectField(const SchemaField& expected_field,
-                                     const SchemaField& source_field, size_t source_index,
-                                     bool prune_source, ProjectionMode mode) {
+Result<FieldProjection> ProjectField(
+    const SchemaField& expected_field, const SchemaField& source_field,
+    size_t source_index, bool prune_source, ProjectionMode mode,
+    const std::unordered_map<int32_t, Literal>& id_to_constant) {
   FieldProjection projection;
 
   if (mode == ProjectionMode::kWrite && source_field.optional() &&
@@ -123,9 +125,9 @@ Result<FieldProjection> ProjectField(const SchemaField& expected_field,
   }
 
   if (expected_field.type()->is_nested()) {
-    ICEBERG_ASSIGN_OR_RAISE(
-        projection,
-        ProjectNested(*expected_field.type(), *source_field.type(), prune_source, mode));
+    ICEBERG_ASSIGN_OR_RAISE(projection,
+                            ProjectNested(*expected_field.type(), *source_field.type(),
+                                          prune_source, mode, id_to_constant));
   } else if (mode == ProjectionMode::kWrite) {
     if (*expected_field.type() != *source_field.type()) {
       return NotSupported("Cannot project {} as {}: type promotion is not allowed",
@@ -144,8 +146,9 @@ Result<FieldProjection> ProjectField(const SchemaField& expected_field,
   return projection;
 }
 
-Result<FieldProjection> ProjectNested(const Type& expected_type, const Type& source_type,
-                                      bool prune_source, ProjectionMode mode) {
+Result<FieldProjection> ProjectNested(
+    const Type& expected_type, const Type& source_type, bool prune_source,
+    ProjectionMode mode, const std::unordered_map<int32_t, Literal>& id_to_constant) {
   if (!expected_type.is_nested()) {
     return InvalidSchema("Expected a nested type, but got {}", expected_type);
   }
@@ -184,10 +187,16 @@ Result<FieldProjection> ProjectNested(const Type& expected_type, const Type& sou
     int32_t field_id = expected_field.field_id();
     FieldProjection child_projection;
 
-    if (auto iter = source_field_map.find(field_id); iter != source_field_map.cend()) {
-      ICEBERG_ASSIGN_OR_RAISE(child_projection,
-                              ProjectField(expected_field, *iter->second.field,
-                                           iter->second.local_index, prune_source, mode));
+    if (auto constant = id_to_constant.find(field_id);
+        mode == ProjectionMode::kRead && constant != id_to_constant.end()) {
+      child_projection.kind = FieldProjection::Kind::kConstant;
+      child_projection.from = constant->second;
+    } else if (auto iter = source_field_map.find(field_id);
+               iter != source_field_map.cend()) {
+      ICEBERG_ASSIGN_OR_RAISE(
+          child_projection,
+          ProjectField(expected_field, *iter->second.field, iter->second.local_index,
+                       prune_source, mode, id_to_constant));
       if (mode == ProjectionMode::kWrite) {
         source_field_map.erase(iter);
       }
@@ -206,7 +215,6 @@ Result<FieldProjection> ProjectNested(const Type& expected_type, const Type& sou
     } else if (expected_field.optional()) {
       child_projection.kind = FieldProjection::Kind::kNull;
     } else {
-      // TODO(gangwu): support constant value
       if (mode == ProjectionMode::kWrite) {
         return InvalidSchema("Missing required field: {} without a write-default",
                              expected_field.ToString());
@@ -230,12 +238,12 @@ Result<FieldProjection> ProjectNested(const Type& expected_type, const Type& sou
 
 }  // namespace
 
-Result<SchemaProjection> Project(const Schema& expected_schema,
-                                 const Schema& source_schema, bool prune_source,
-                                 ProjectionMode mode) {
+Result<SchemaProjection> Project(
+    const Schema& expected_schema, const Schema& source_schema, bool prune_source,
+    ProjectionMode mode, const std::unordered_map<int32_t, Literal>& id_to_constant) {
   ICEBERG_ASSIGN_OR_RAISE(
       auto field_projection,
-      ProjectNested(expected_schema, source_schema, prune_source, mode));
+      ProjectNested(expected_schema, source_schema, prune_source, mode, id_to_constant));
   return SchemaProjection{std::move(field_projection.children)};
 }
 
diff --git a/src/iceberg/schema_util.h b/src/iceberg/schema_util.h
index 397013f44..23933d2af 100644
--- a/src/iceberg/schema_util.h
+++ b/src/iceberg/schema_util.h
@@ -24,6 +24,7 @@
 
 #include <memory>
 #include <string>
+#include <unordered_map>
 #include <variant>
 #include <vector>
 
@@ -99,10 +100,14 @@ enum class ProjectionMode {
 /// field in `FieldProjection` exactly reflects the position (relative to its nesting
 /// level) to get the column value from the reader.
 /// \param mode Whether to project for reading or writing.
+/// \param id_to_constant Primitive values keyed by source field ID. Constants override
+/// physical fields and defaults, including primitive leaves of existing structs. Only
+/// applied when `mode` is `ProjectionMode::kRead`.
 /// \return The projection result.
 ICEBERG_EXPORT Result<SchemaProjection> Project(
     const Schema& expected_schema, const Schema& source_schema, bool prune_source,
-    ProjectionMode mode = ProjectionMode::kRead);
+    ProjectionMode mode = ProjectionMode::kRead,
+    const std::unordered_map<int32_t, Literal>& id_to_constant = {});
 
 ICEBERG_EXPORT std::string_view ToString(FieldProjection::Kind kind);
 ICEBERG_EXPORT std::string ToString(const FieldProjection& projection);
diff --git a/src/iceberg/table_scan.cc b/src/iceberg/table_scan.cc
index d04e3385b..7b3a69fd1 100644
--- a/src/iceberg/table_scan.cc
+++ b/src/iceberg/table_scan.cc
@@ -174,25 +174,28 @@ ScanTask::~ScanTask() = default;
 
 FileScanTask::FileScanTask(std::shared_ptr<DataFile> data_file,
                            std::vector<std::shared_ptr<DataFile>> delete_files,
-                           std::shared_ptr<Expression> residual_filter)
+                           std::shared_ptr<Expression> residual_filter,
+                           std::shared_ptr<PartitionSpec> spec)
     : data_file_(std::move(data_file)),
       delete_files_(std::move(delete_files)),
-      residual_filter_(std::move(residual_filter)) {
+      residual_filter_(std::move(residual_filter)),
+      spec_(std::move(spec)) {
   ICEBERG_DCHECK(data_file_ != nullptr, "Data file cannot be null for FileScanTask");
 }
 
 FileScanTask::FileScanTask(std::shared_ptr<DataFile> data_file,
                            std::vector<std::shared_ptr<DataFile>> delete_files,
-                           std::shared_ptr<Expression> residual_filter, Range range)
+                           std::shared_ptr<Expression> residual_filter, Range range,
+                           std::shared_ptr<PartitionSpec> spec)
     : FileScanTask(std::move(data_file), std::move(delete_files),
-                   std::move(residual_filter)) {
+                   std::move(residual_filter), std::move(spec)) {
   range_ = range;
 }
 
 Result<std::shared_ptr<FileScanTask>> FileScanTask::MakeSplit(
     std::shared_ptr<DataFile> data_file, int64_t start, int64_t length,
     std::vector<std::shared_ptr<DataFile>> delete_files,
-    std::shared_ptr<Expression> filter) {
+    std::shared_ptr<Expression> filter, std::shared_ptr<PartitionSpec> spec) {
   if (data_file == nullptr) {
     return InvalidArgument("Cannot create a file scan task without a data file");
   }
@@ -215,7 +218,7 @@ Result<std::shared_ptr<FileScanTask>> FileScanTask::MakeSplit(
   // only valid empty range: [0, 0) of an empty file.
   if (start == 0 && length == file_size) {
     return std::make_shared<FileScanTask>(std::move(data_file), std::move(delete_files),
-                                          std::move(filter));
+                                          std::move(filter), std::move(spec));
   }
   if (length == 0) {
     return InvalidArgument("A partial split of file {} must have positive length",
@@ -224,7 +227,7 @@ Result<std::shared_ptr<FileScanTask>> FileScanTask::MakeSplit(
 
   return std::shared_ptr<FileScanTask>(
       new FileScanTask(std::move(data_file), std::move(delete_files), std::move(filter),
-                       Range{.start = start, .length = length}));
+                       Range{.start = start, .length = length}, std::move(spec)));
 }
 
 int64_t FileScanTask::length() const {
diff --git a/src/iceberg/table_scan.h b/src/iceberg/table_scan.h
index 1307089e3..df202b9a3 100644
--- a/src/iceberg/table_scan.h
+++ b/src/iceberg/table_scan.h
@@ -72,11 +72,13 @@ class ICEBERG_EXPORT FileScanTask : public ScanTask {
   /// \param data_file The data file to read.
   /// \param delete_files Delete files that apply to this data file.
   /// \param filter Optional residual filter to apply after reading.
+  /// \param spec Partition spec corresponding to the data file partition tuple.
   /// \note The shared data file, delete files, and residual filter must remain
   /// unchanged for this task's lifetime. Metadata changes require rebuilding tasks.
   explicit FileScanTask(std::shared_ptr<DataFile> data_file,
                         std::vector<std::shared_ptr<DataFile>> delete_files = {},
-                        std::shared_ptr<Expression> filter = nullptr);
+                        std::shared_ptr<Expression> filter = nullptr,
+                        std::shared_ptr<PartitionSpec> spec = nullptr);
 
   /// \brief Make a task for a validated half-open byte range [start, start + length).
   ///
@@ -87,7 +89,8 @@ class ICEBERG_EXPORT FileScanTask : public ScanTask {
   static Result<std::shared_ptr<FileScanTask>> MakeSplit(
       std::shared_ptr<DataFile> data_file, int64_t start, int64_t length,
       std::vector<std::shared_ptr<DataFile>> delete_files = {},
-      std::shared_ptr<Expression> filter = nullptr);
+      std::shared_ptr<Expression> filter = nullptr,
+      std::shared_ptr<PartitionSpec> spec = nullptr);
 
   /// \brief The inclusive byte offset where this task starts, or zero for a whole file.
   int64_t start() const { return range_ ? range_->start : 0; }
@@ -109,6 +112,10 @@ class ICEBERG_EXPORT FileScanTask : public ScanTask {
   /// \brief Residual filter to apply after reading.
   const std::shared_ptr<Expression>& residual_filter() const { return residual_filter_; }
 
+  /// The partition spec used to write this file, including historical field IDs.
+  /// The shared spec must not be mutated while the task is in use.
+  const std::shared_ptr<PartitionSpec>& spec() const { return spec_; }
+
   Kind kind() const override { return Kind::kFileScanTask; }
   int64_t size_bytes() const override;
   int32_t files_count() const override;
@@ -127,12 +134,14 @@ class ICEBERG_EXPORT FileScanTask : public ScanTask {
 
   FileScanTask(std::shared_ptr<DataFile> data_file,
                std::vector<std::shared_ptr<DataFile>> delete_files,
-               std::shared_ptr<Expression> filter, Range range);
+               std::shared_ptr<Expression> filter, Range range,
+               std::shared_ptr<PartitionSpec> spec);
 
   std::shared_ptr<DataFile> data_file_;
   std::vector<std::shared_ptr<DataFile>> delete_files_;
   std::shared_ptr<Expression> residual_filter_;
   std::optional<Range> range_;
+  std::shared_ptr<PartitionSpec> spec_;
 };
 
 /// \brief Stream of file scan tasks.
diff --git a/src/iceberg/test/avro_test.cc b/src/iceberg/test/avro_test.cc
index a16116410..13cd56e82 100644
--- a/src/iceberg/test/avro_test.cc
+++ b/src/iceberg/test/avro_test.cc
@@ -481,6 +481,24 @@ class AvroReaderParameterizedTest : public AvroReaderTest,
   }
 };
 
+TEST_P(AvroReaderParameterizedTest, ReadsPartitionConstants) {
+  CreateSimpleAvroFile();
+  auto schema = std::make_shared<Schema>(
+      std::vector<SchemaField>{SchemaField::MakeRequired(1, "id", int32()),
+                               SchemaField::MakeRequired(3, "region", string()),
+                               SchemaField::MakeOptional(4, "missing", string())});
+  ReaderOptions options{.path = temp_avro_file_, .io = file_io_, .projection = schema};
+  options.properties.Set(ReaderProperties::kAvroSkipDatum, skip_datum_);
+  options.id_to_constant = {
+      {1, Literal::Int(999)}, {3, Literal::String("CN")}, {4, Literal::Null(string())}};
+  ICEBERG_UNWRAP_OR_FAIL(auto reader,
+                         ReaderFactoryRegistry::Open(FileFormatType::kAvro, options));
+  VerifyNextBatch(*reader, R"([[999,"CN",null],
+                              [999,"CN",null],
+                              [999,"CN",null]])");
+  VerifyExhausted(*reader);
+}
+
 TEST_P(AvroReaderParameterizedTest, AvroWriterBasicType) {
   auto schema = std::make_shared<iceberg::Schema>(std::vector<SchemaField>{
       SchemaField::MakeRequired(1, "name", std::make_shared<StringType>())});
diff --git a/src/iceberg/test/file_scan_task_reader_test.cc b/src/iceberg/test/file_scan_task_reader_test.cc
index 8b0d79d15..d32b8a996 100644
--- a/src/iceberg/test/file_scan_task_reader_test.cc
+++ b/src/iceberg/test/file_scan_task_reader_test.cc
@@ -33,6 +33,7 @@
 #include <arrow/json/from_string.h>
 #include <arrow/record_batch.h>
 #include <arrow/table.h>
+#include <nlohmann/json.hpp>
 #include <parquet/arrow/writer.h>
 #include <parquet/file_reader.h>
 #include <parquet/metadata.h>
@@ -54,6 +55,7 @@
 #include "iceberg/file_format.h"
 #include "iceberg/file_io.h"
 #include "iceberg/file_reader.h"
+#include "iceberg/json_serde_internal.h"
 #include "iceberg/manifest/manifest_entry.h"
 #include "iceberg/metadata_columns.h"
 #include "iceberg/parquet/parquet_register.h"
@@ -66,6 +68,7 @@
 #include "iceberg/test/matchers.h"
 #include "iceberg/test/std_io.h"
 #include "iceberg/test/temp_file_test_base.h"
+#include "iceberg/transform.h"
 #include "iceberg/type.h"
 #include "iceberg/util/macros.h"
 
@@ -421,6 +424,165 @@ class FileScanTaskReaderTest : public TempFileTestBase {
   std::shared_ptr<Schema> projected_schema_;
 };
 
+TEST_F(FileScanTaskReaderTest, IdentityPartitionConstantsSurviveJsonAndSplit) {
+  auto physical = std::make_shared<Schema>(
+      std::vector<SchemaField>{SchemaField::MakeRequired(1, "id", int32())});
+  ICEBERG_UNWRAP_OR_FAIL(auto file, MakeDataFile(physical, "[[1], [2], [3]]"));
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto spec, PartitionSpec::Make(7, {{2, 1000, "name", Transform::Identity()}}));
+  file->partition_spec_id = 7;
+  file->partition = PartitionValues({Literal::String("CN")});
+  FileScanTask task(file, {}, nullptr, std::move(spec));
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto json, ToJson(task, {{7, std::const_pointer_cast<PartitionSpec>(task.spec())}},
+                        *table_schema_));
+  ICEBERG_UNWRAP_OR_FAIL(auto parsed, FileScanTaskFromJson(json));
+  ASSERT_NE(parsed->spec(), nullptr);
+  EXPECT_EQ(parsed->spec()->spec_id(), 7);
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto split, FileScanTask::MakeSplit(parsed->data_file(), 0, parsed->length(), {},
+                                          nullptr, parsed->spec()));
+  EXPECT_EQ(split->spec(), parsed->spec());
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto reader, FileScanTaskReader::Make({.io = file_io_,
+                                             .table_schema = table_schema_,
+                                             .schemas = {table_schema_},
+                                             .projected_schema = projected_schema_}));
+  ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(*split));
+  VerifyStream(&stream, R"([[1, "CN"], [2, "CN"], [3, "CN"]])");
+}
+
+TEST_F(FileScanTaskReaderTest, IdentityConstantsApplyBeforeEqualityDeletes) {
+  auto physical = std::make_shared<Schema>(
+      std::vector<SchemaField>{SchemaField::MakeRequired(1, "id", int32())});
+  ICEBERG_UNWRAP_OR_FAIL(auto file, MakeDataFile(physical, "[[1], [2], [3]]"));
+  auto delete_schema = std::make_shared<Schema>(
+      std::vector<SchemaField>{SchemaField::MakeOptional(2, "name", string())});
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto deletes, MakeEqualityDeleteFile(CreateNewTempFilePathWithSuffix(".parquet"),
+                                           delete_schema, R"([["CN"]])", {2}));
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto spec, PartitionSpec::Make(7, {{2, 1000, "name", Transform::Identity()}}));
+  file->partition_spec_id = 7;
+  file->partition = PartitionValues({Literal::String("CN")});
+  FileScanTask task(file, {deletes}, nullptr, std::move(spec));
+  ICEBERG_UNWRAP_OR_FAIL(auto reader,
+                         FileScanTaskReader::Make({.io = file_io_,
+                                                   .table_schema = table_schema_,
+                                                   .schemas = {table_schema_},
+                                                   .projected_schema = physical}));
+  ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(task));
+  auto batches = ::arrow::ImportRecordBatchReader(&stream).ValueOrDie();
+  auto table = batches->ToTable().ValueOrDie();
+  EXPECT_EQ(table->num_rows(), 0);
+}
+
+TEST_F(FileScanTaskReaderTest, NullIdentityOverridesInitialDefault) {
+  auto physical = std::make_shared<Schema>(
+      std::vector<SchemaField>{SchemaField::MakeRequired(1, "id", int32())});
+  ICEBERG_UNWRAP_OR_FAIL(auto file, MakeDataFile(physical, "[[1], [2], [3]]"));
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto spec, PartitionSpec::Make(0, {{2, 1000, "name", Transform::Identity()}}));
+  file->partition = PartitionValues({Literal::Null(string())});
+  auto projection = std::make_shared<Schema>(std::vector<SchemaField>{
+      SchemaField(2, "name", string(), true, {},
+                  std::make_shared<const Literal>(Literal::String("default")))});
+  FileScanTask task(file, {}, nullptr, std::move(spec));
+  ICEBERG_UNWRAP_OR_FAIL(auto reader,
+                         FileScanTaskReader::Make({.io = file_io_,
+                                                   .table_schema = table_schema_,
+                                                   .schemas = {table_schema_},
+                                                   .projected_schema = projection}));
+  ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(task));
+  VerifyStream(&stream, "[[null], [null], [null]]");
+}
+
+TEST_F(FileScanTaskReaderTest,
+       ReadsTaskSpecificSpecsWithoutLeakingConstantsBetweenFiles) {
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto identity, PartitionSpec::Make(7, {{2, 1000, "name", Transform::Identity()}}));
+  ICEBERG_UNWRAP_OR_FAIL(auto unpartitioned, PartitionSpec::Make(8, {}));
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto bucket,
+      PartitionSpec::Make(9, {{2, 1001, "name_bucket", Transform::Bucket(16)}}));
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto reordered,
+      PartitionSpec::Make(10, {{3, 1002, "category", Transform::Identity()},
+                               {2, 1000, "renamed_name", Transform::Identity()}}));
+  struct Case {
+    std::shared_ptr<PartitionSpec> spec;
+    PartitionValues partition;
+    std::string expected;
+  };
+  auto identity_spec = std::shared_ptr<PartitionSpec>(std::move(identity));
+  const std::vector<Case> cases{
+      {.spec = identity_spec,
+       .partition = PartitionValues(Literal::String("CN")),
+       .expected = R"([[1, "CN"], [2, "CN"], [3, "CN"]])"},
+      {.spec = std::move(unpartitioned),
+       .partition = PartitionValues(),
+       .expected = R"([[1, "file"], [2, "file"], [3, "file"]])"},
+      {.spec = std::move(bucket),
+       .partition = PartitionValues(Literal::Int(3)),
+       .expected = R"([[1, "file"], [2, "file"], [3, "file"]])"},
+      {.spec = std::move(reordered),
+       .partition = PartitionValues({Literal::String("retired"), Literal::String("US")}),
+       .expected = R"([[1, "US"], [2, "US"], [3, "US"]])"},
+      {.spec = identity_spec,
+       .partition = PartitionValues(Literal::Null(string())),
+       .expected = "[[1, null], [2, null], [3, null]]"}};
+  // The current schema omits category, but historical task schemas retain its type.
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto reader, FileScanTaskReader::Make({.io = file_io_,
+                                             .table_schema = projected_schema_,
+                                             .schemas = {table_schema_},
+                                             .projected_schema = projected_schema_}));
+  std::vector<std::shared_ptr<::arrow::RecordBatchReader>> readers;
+  for (const auto& test : cases) {
+    SCOPED_TRACE(test.spec->spec_id());
+    ICEBERG_UNWRAP_OR_FAIL(
+        auto file,
+        MakeDataFile(projected_schema_, R"([[1, "file"], [2, "file"], [3, "file"]])"));
+    file->partition_spec_id = test.spec->spec_id();
+    file->partition = test.partition;
+    ICEBERG_UNWRAP_OR_FAIL(auto json,
+                           ToJson(FileScanTask(file, {}, nullptr, test.spec),
+                                  {{test.spec->spec_id(), test.spec}}, *table_schema_));
+    ICEBERG_UNWRAP_OR_FAIL(auto task, FileScanTaskFromJson(json));
+    ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(*task));
+    auto imported = ::arrow::ImportRecordBatchReader(&stream);
+    ASSERT_TRUE(imported.ok()) << imported.status().ToString();
+    readers.push_back(imported.ValueOrDie());
+  }
+  // Keep all streams alive to detect accidental sharing of per-file constants.
+  for (size_t i = 0; i < cases.size(); ++i) {
+    SCOPED_TRACE(i);
+    auto next = readers[i]->Next();
+    ASSERT_TRUE(next.ok()) << next.status().ToString();
+    auto batch = next.ValueOrDie();
+    ASSERT_NE(batch, nullptr);
+    ASSERT_TRUE(batch->ValidateFull().ok());
+    ICEBERG_UNWRAP_OR_FAIL(auto expected,
+                           MakeRecordBatch(batch->schema(), cases[i].expected));
+    EXPECT_TRUE(batch->Equals(*expected));
+    next = readers[i]->Next();
+    ASSERT_TRUE(next.ok()) << next.status().ToString();
+    EXPECT_EQ(next.ValueOrDie(), nullptr);
+  }
+}
+
+TEST_F(FileScanTaskReaderTest, RejectsMissingPartitionSpec) {
+  ICEBERG_UNWRAP_OR_FAIL(auto file,
+                         MakeDataFile(table_schema_, R"([[1, "Foo", "blue"]])", 1));
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto reader, FileScanTaskReader::Make({.io = file_io_,
+                                             .table_schema = table_schema_,
+                                             .schemas = {table_schema_},
+                                             .projected_schema = projected_schema_}));
+  file->partition = PartitionValues({Literal::String("CN")});
+  EXPECT_FALSE(reader->Open(FileScanTask(file)).has_value());
+}
+
 TEST_F(FileScanTaskReaderTest, OpenWithoutDeletesReadsProjectedSchema) {
   ICEBERG_UNWRAP_OR_FAIL(
       auto data_file,
diff --git a/src/iceberg/test/manifest_group_test.cc b/src/iceberg/test/manifest_group_test.cc
index 063ce5804..84bd971a9 100644
--- a/src/iceberg/test/manifest_group_test.cc
+++ b/src/iceberg/test/manifest_group_test.cc
@@ -290,6 +290,10 @@ TEST_P(ManifestGroupTest, CreateAndGetEntries) {
   ASSERT_EQ(tasks.size(), 2);
   EXPECT_THAT(GetPaths(tasks), testing::UnorderedElementsAre("/path/to/data1.parquet",
                                                              "/path/to/data2.parquet"));
+  for (const auto& task : tasks) {
+    ASSERT_NE(task->spec(), nullptr);
+    EXPECT_EQ(task->spec()->spec_id(), task->data_file()->partition_spec_id);
+  }
   EXPECT_EQ(tasks[0]->delete_files().size(), 1);
   EXPECT_EQ(tasks[1]->delete_files()[0]->file_path, "/path/to/delete.parquet");
   EXPECT_EQ(tasks[1]->delete_files().size(), 1);
diff --git a/src/iceberg/test/parquet_test.cc b/src/iceberg/test/parquet_test.cc
index c35ba4471..35b68d113 100644
--- a/src/iceberg/test/parquet_test.cc
+++ b/src/iceberg/test/parquet_test.cc
@@ -45,6 +45,7 @@
 
 #include "iceberg/arrow/arrow_io_internal.h"
 #include "iceberg/arrow/arrow_status_internal.h"
+#include "iceberg/expression/expressions.h"
 #include "iceberg/expression/literal.h"
 #include "iceberg/file_reader.h"
 #include "iceberg/file_writer.h"
@@ -552,6 +553,19 @@ class ParquetReaderTest : public TempFileTestBase {
   std::string temp_parquet_file_;
 };
 
+TEST_F(ParquetReaderTest, PromotesPartitionConstant) {
+  CreateSimpleParquetFile();
+  auto schema = std::make_shared<Schema>(
+      std::vector<SchemaField>{SchemaField::MakeRequired(3, "constant", int64())});
+
+  ReaderOptions options{.path = temp_parquet_file_, .io = file_io_, .projection = schema};
+  options.id_to_constant.emplace(3, Literal::Int(7));
+  ICEBERG_UNWRAP_OR_FAIL(auto reader,
+                         ReaderFactoryRegistry::Open(FileFormatType::kParquet, options));
+  VerifyNextBatch(*reader, "[[7], [7], [7]]");
+  VerifyExhausted(*reader);
+}
+
 TEST_F(ParquetReaderTest, ReadTwoFields) {
   CreateSimpleParquetFile();
 
diff --git a/src/iceberg/test/rest_json_serde_test.cc b/src/iceberg/test/rest_json_serde_test.cc
index ae5ba1a74..b3ba428e5 100644
--- a/src/iceberg/test/rest_json_serde_test.cc
+++ b/src/iceberg/test/rest_json_serde_test.cc
@@ -2652,6 +2652,8 @@ TEST(FetchScanTasksResponseRoundtripTest, PreservesResidualFilter) {
   ASSERT_TRUE(result->file_scan_tasks.has_value());
   ASSERT_EQ(result->file_scan_tasks->size(), 1);
   ASSERT_NE(result->file_scan_tasks->at(0)->residual_filter(), nullptr);
+  ASSERT_NE(result->file_scan_tasks->at(0)->spec(), nullptr);
+  EXPECT_EQ(result->file_scan_tasks->at(0)->spec()->spec_id(), 0);
 
   ICEBERG_UNWRAP_OR_FAIL(auto roundtrip_json,
                          ToJson(*result, UnpartitionedSpecs(), EmptySchema()));
diff --git a/src/iceberg/test/schema_util_test.cc b/src/iceberg/test/schema_util_test.cc
index 785a90350..642aa9e7b 100644
--- a/src/iceberg/test/schema_util_test.cc
+++ b/src/iceberg/test/schema_util_test.cc
@@ -101,6 +101,49 @@ std::shared_ptr<Type> CreateMapOfList() {
 
 }  // namespace
 
+TEST(SchemaUtilTest, ProjectConstantsOverridePhysicalFieldsAndPruneIndices) {
+  Schema schema({SchemaField::MakeRequired(1, "region", string()),
+                 SchemaField::MakeRequired(2, "id", int64())});
+  for (bool prune_source : {false, true}) {
+    auto projection = Project(schema, schema, prune_source, ProjectionMode::kRead,
+                              {{1, Literal::String("CN")}});
+    ASSERT_THAT(projection, IsOk());
+    ASSERT_EQ(projection->fields.size(), 2);
+    EXPECT_EQ(projection->fields[0].kind, FieldProjection::Kind::kConstant);
+    EXPECT_EQ(std::get<Literal>(projection->fields[0].from), Literal::String("CN"));
+    AssertProjectedField(projection->fields[1], prune_source ? 0 : 1);
+  }
+}
+
+TEST(SchemaUtilTest, ProjectConstantsFillMissingFields) {
+  Schema expected({SchemaField::MakeRequired(1, "id", int64()),
+                   SchemaField::MakeOptional(2, "region", string())});
+  auto projection =
+      Project(expected, Schema({}), /*prune_source=*/true, ProjectionMode::kRead,
+              {{1, Literal::Int(7)}, {2, Literal::Null(string())}});
+  ASSERT_THAT(projection, IsOk());
+  ASSERT_EQ(projection->fields.size(), 2);
+  EXPECT_EQ(projection->fields[0].kind, FieldProjection::Kind::kConstant);
+  EXPECT_EQ(projection->fields[1].kind, FieldProjection::Kind::kConstant);
+}
+
+TEST(SchemaUtilTest, ProjectConstantsIntoExistingStruct) {
+  Schema schema(
+      {SchemaField::MakeOptional(1, "parent",
+                                 std::make_shared<StructType>(std::vector<SchemaField>{
+                                     SchemaField::MakeOptional(2, "region", string()),
+                                     SchemaField::MakeRequired(3, "id", int64())}))});
+  auto projection = Project(schema, schema, /*prune_source=*/true, ProjectionMode::kRead,
+                            {{2, Literal::String("CN")}});
+  ASSERT_THAT(projection, IsOk());
+  ASSERT_EQ(projection->fields.size(), 1);
+  const auto& parent = projection->fields[0];
+  AssertProjectedField(parent, 0);
+  ASSERT_EQ(parent.children.size(), 2);
+  EXPECT_EQ(parent.children[0].kind, FieldProjection::Kind::kConstant);
+  AssertProjectedField(parent.children[1], 0);
+}
+
 TEST(SchemaUtilTest, ProjectIdenticalSchemas) {
   Schema schema = CreateFlatSchema();