feat(read): read identity partition columns from file scan tasks

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

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();

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));
 

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,32 +862,32 @@ 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);
   }
 
   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)};
 }
 

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);

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);
     }

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;
       }

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));
 

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.

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);
   }