feat: read v3 row lineage metadata columns (#822)

Synthesize _row_id and _last_updated_sequence_number when reading data
files that omit row lineage columns, and preserve physical non-null
lineage values when present.

Carry inherited data sequence numbers on DataFile metadata so scan
planning and readers follow the same model as Java.

Author: wgtmac

src/iceberg/arrow/metadata_column_util.cc

@@ -24,9 +24,38 @@
 
 #include "iceberg/arrow/arrow_status_internal.h"
 #include "iceberg/arrow/metadata_column_util_internal.h"
+#include "iceberg/metadata_columns.h"
+#include "iceberg/util/checked_cast.h"
 
 namespace iceberg::arrow {
 
+bool CanInheritRowLineageValue(int32_t field_id,
+                               const MetadataColumnContext& metadata_context) {
+  if (field_id == MetadataColumns::kRowIdColumnId) {
+    return metadata_context.first_row_id.has_value();
+  }
+  if (field_id == MetadataColumns::kLastUpdatedSequenceNumberColumnId) {
+    return metadata_context.data_sequence_number.has_value();
+  }
+  return false;
+}
+
+Status AppendInheritedRowLineageValue(int32_t field_id,
+                                      const MetadataColumnContext& metadata_context,
+                                      ::arrow::ArrayBuilder* array_builder) {
+  auto* int_builder = internal::checked_cast<::arrow::Int64Builder*>(array_builder);
+  if (!CanInheritRowLineageValue(field_id, metadata_context)) {
+    ICEBERG_ARROW_RETURN_NOT_OK(int_builder->AppendNull());
+  } else if (field_id == MetadataColumns::kRowIdColumnId) {
+    ICEBERG_ARROW_RETURN_NOT_OK(int_builder->Append(
+        metadata_context.first_row_id.value() + metadata_context.next_file_pos));
+  } else {
+    ICEBERG_ARROW_RETURN_NOT_OK(
+        int_builder->Append(metadata_context.data_sequence_number.value()));
+  }
+  return {};
+}
+
 Result<std::shared_ptr<::arrow::Array>> MakeFilePathArray(const std::string& file_path,
                                                           int64_t num_rows,
                                                           ::arrow::MemoryPool* pool) {
@@ -53,4 +82,41 @@ Result<std::shared_ptr<::arrow::Array>> MakeRowPositionArray(int64_t start_posit
   return array;
 }
 
+Result<std::shared_ptr<::arrow::Array>> MakeRowIdArray(
+    std::optional<int64_t> first_row_id, int64_t start_position, int64_t num_rows,
+    ::arrow::MemoryPool* pool) {
+  ::arrow::Int64Builder builder(pool);
+  ICEBERG_ARROW_RETURN_NOT_OK(builder.Reserve(num_rows));
+  if (!first_row_id.has_value()) {
+    ICEBERG_ARROW_RETURN_NOT_OK(builder.AppendNulls(num_rows));
+  } else {
+    for (int64_t row_index = 0; row_index < num_rows; ++row_index) {
+      ICEBERG_ARROW_RETURN_NOT_OK(
+          builder.Append(first_row_id.value() + start_position + row_index));
+    }
+  }
+
+  std::shared_ptr<::arrow::Array> array;
+  ICEBERG_ARROW_RETURN_NOT_OK(builder.Finish(&array));
+  return array;
+}
+
+Result<std::shared_ptr<::arrow::Array>> MakeLastUpdatedSequenceNumberArray(
+    std::optional<int64_t> data_sequence_number, int64_t num_rows,
+    ::arrow::MemoryPool* pool) {
+  ::arrow::Int64Builder builder(pool);
+  ICEBERG_ARROW_RETURN_NOT_OK(builder.Reserve(num_rows));
+  if (!data_sequence_number.has_value()) {
+    ICEBERG_ARROW_RETURN_NOT_OK(builder.AppendNulls(num_rows));
+  } else {
+    for (int64_t row_index = 0; row_index < num_rows; ++row_index) {
+      ICEBERG_ARROW_RETURN_NOT_OK(builder.Append(data_sequence_number.value()));
+    }
+  }
+
+  std::shared_ptr<::arrow::Array> array;
+  ICEBERG_ARROW_RETURN_NOT_OK(builder.Finish(&array));
+  return array;
+}
+
 }  // namespace iceberg::arrow

src/iceberg/arrow/metadata_column_util_internal.h

@@ -24,20 +24,36 @@
 
 #include <cstdint>
 #include <memory>
+#include <optional>
 #include <string>
 
 #include <arrow/type_fwd.h>
 
 #include "iceberg/result.h"
 
+namespace arrow {
+class ArrayBuilder;
+}  // namespace arrow
+
 namespace iceberg::arrow {
 
 /// \brief Context for populating metadata columns during reading.
 struct MetadataColumnContext {
   std::string file_path;      // The file path to populate _file column.
   int64_t next_file_pos = 0;  // The next row position for populating _pos column.
+  std::optional<int64_t> first_row_id;          // First row ID used to populate _row_id.
+  std::optional<int64_t> data_sequence_number;  // Data sequence number for lineage.
 };
 
+/// \brief Check if row lineage metadata can be inherited for the column.
+bool CanInheritRowLineageValue(int32_t field_id,
+                               const MetadataColumnContext& metadata_context);
+
+/// \brief Append one inherited row lineage value, or null if unavailable.
+Status AppendInheritedRowLineageValue(int32_t field_id,
+                                      const MetadataColumnContext& metadata_context,
+                                      ::arrow::ArrayBuilder* array_builder);
+
 /// \brief Create a constant string array for _file column.
 ///
 /// Creates an Arrow StringArray where all values are the same file path string.
@@ -62,4 +78,25 @@ Result<std::shared_ptr<::arrow::Array>> MakeRowPositionArray(int64_t start_posit
                                                              int64_t num_rows,
                                                              ::arrow::MemoryPool* pool);
 
+/// \brief Create a row ID array for _row_id column.
+///
+/// \param first_row_id First row ID of the data file.
+/// \param start_position Starting row position (inclusive).
+/// \param num_rows Number of rows in the batch.
+/// \param pool Arrow memory pool.
+/// \return Arrow Int64Array for _row_id.
+Result<std::shared_ptr<::arrow::Array>> MakeRowIdArray(
+    std::optional<int64_t> first_row_id, int64_t start_position, int64_t num_rows,
+    ::arrow::MemoryPool* pool);
+
+/// \brief Create a sequence number array for _last_updated_sequence_number column.
+///
+/// \param data_sequence_number Data sequence number of the data file.
+/// \param num_rows Number of rows in the batch.
+/// \param pool Arrow memory pool.
+/// \return Arrow Int64Array for _last_updated_sequence_number.
+Result<std::shared_ptr<::arrow::Array>> MakeLastUpdatedSequenceNumberArray(
+    std::optional<int64_t> data_sequence_number, int64_t num_rows,
+    ::arrow::MemoryPool* pool);
+
 }  // namespace iceberg::arrow

src/iceberg/avro/avro_data_util.cc

@@ -97,6 +97,9 @@ Status AppendStructToBuilder(const ::avro::NodePtr& avro_node,
       } else if (field_id == MetadataColumns::kFilePositionColumnId) {
         auto int_builder = internal::checked_cast<::arrow::Int64Builder*>(field_builder);
         ICEBERG_ARROW_RETURN_NOT_OK(int_builder->Append(metadata_context.next_file_pos));
+      } else if (MetadataColumns::IsRowLineageColumn(field_id)) {
+        ICEBERG_RETURN_UNEXPECTED(arrow::AppendInheritedRowLineageValue(
+            field_id, metadata_context, field_builder));
       } else {
         return NotSupported("Unsupported metadata column field id: {}", field_id);
       }
@@ -463,9 +466,16 @@ Status AppendFieldToBuilder(const ::avro::NodePtr& avro_node,
     return {};
   }
 
+  const bool is_row_lineage =
+      MetadataColumns::IsRowLineageColumn(projected_field.field_id());
+
   if (avro_node->type() == ::avro::AVRO_UNION) {
     size_t branch = avro_datum.unionBranch();
     if (avro_node->leafAt(branch)->type() == ::avro::AVRO_NULL) {
+      if (is_row_lineage) {
+        return arrow::AppendInheritedRowLineageValue(projected_field.field_id(),
+                                                     metadata_context, array_builder);
+      }
       ICEBERG_ARROW_RETURN_NOT_OK(array_builder->AppendNull());
       return {};
     } else {

src/iceberg/avro/avro_direct_decoder.cc

@@ -218,6 +218,9 @@ Status DecodeStructToBuilder(const ::avro::NodePtr& avro_node, ::avro::Decoder&
       } else if (field_id == MetadataColumns::kFilePositionColumnId) {
         auto int_builder = internal::checked_cast<::arrow::Int64Builder*>(field_builder);
         ICEBERG_ARROW_RETURN_NOT_OK(int_builder->Append(metadata_context.next_file_pos));
+      } else if (MetadataColumns::IsRowLineageColumn(field_id)) {
+        ICEBERG_RETURN_UNEXPECTED(arrow::AppendInheritedRowLineageValue(
+            field_id, metadata_context, field_builder));
       } else {
         return NotSupported("Unsupported metadata column field id: {}", field_id);
       }
@@ -594,6 +597,9 @@ Status DecodeFieldToBuilder(const ::avro::NodePtr& avro_node, ::avro::Decoder& d
     return {};
   }
 
+  const bool is_row_lineage =
+      MetadataColumns::IsRowLineageColumn(projected_field.field_id());
+
   if (avro_node->type() == ::avro::AVRO_UNION) {
     const size_t branch_index = decoder.decodeUnionIndex();
 
@@ -606,6 +612,10 @@ Status DecodeFieldToBuilder(const ::avro::NodePtr& avro_node, ::avro::Decoder& d
 
     const auto& branch_node = avro_node->leafAt(branch_index);
     if (branch_node->type() == ::avro::AVRO_NULL) {
+      if (is_row_lineage) {
+        return arrow::AppendInheritedRowLineageValue(projected_field.field_id(),
+                                                     metadata_context, array_builder);
+      }
       ICEBERG_ARROW_RETURN_NOT_OK(array_builder->AppendNull());
       return {};
     } else {

src/iceberg/avro/avro_reader.cc

@@ -68,6 +68,15 @@ bool HasRowPositionColumn(const Schema& schema) {
   return false;
 }
 
+bool HasRowIdColumn(const Schema& schema) {
+  for (const auto& field : schema.fields()) {
+    if (field.field_id() == MetadataColumns::kRowIdColumnId) {
+      return true;
+    }
+  }
+  return false;
+}
+
 // Abstract base class for Avro read backends.
 class AvroReadBackend {
  public:
@@ -332,9 +341,17 @@ class AvroReader::Impl {
         return NotSupported(
             "Reading '_pos' metadata column with split is not supported for Avro files.");
       }
+      if (options.split->offset != 0 && HasRowIdColumn(*read_schema_)) {
+        return NotSupported(
+            "Reading '_row_id' metadata column with split is not supported for Avro "
+            "files.");
+      }
     }
 
-    metadata_context_ = {.file_path = options.path, .next_file_pos = 0};
+    metadata_context_ = {.file_path = options.path,
+                         .next_file_pos = 0,
+                         .first_row_id = options.first_row_id,
+                         .data_sequence_number = options.data_sequence_number};
 
     return {};
   }

src/iceberg/catalog/rest/json_serde.cc

@@ -323,6 +323,10 @@ Result<std::vector<std::shared_ptr<FileScanTask>>> FileScanTasksFromJson(
                             GetJsonValue<nlohmann::json>(task_json, kDataFile));
     ICEBERG_ASSIGN_OR_RAISE(
         auto data_file, DataFileFromJson(data_file_json, partition_spec_by_id, schema));
+    // FIXME: REST scan-task DataFile JSON currently carries first-row-id,
+    // but not the manifest-entry data sequence number. Until the REST API exposes
+    // it, REST-planned tasks cannot inherit _last_updated_sequence_number.
+    // See https://github.com/apache/iceberg-cpp/issues/834.
 
     std::vector<std::shared_ptr<DataFile>> task_delete_files;
     if (task_json.contains(kDeleteFileReferences) &&

src/iceberg/data/file_scan_task_reader.cc

@@ -43,14 +43,18 @@ ReaderOptions MakeReaderOptions(const DataFile& data_file, std::shared_ptr<FileI
                                 std::shared_ptr<Schema> projection,
                                 std::shared_ptr<Expression> filter,
                                 std::shared_ptr<NameMapping> name_mapping,
-                                ReaderProperties properties) {
+                                ReaderProperties properties,
+                                std::optional<int64_t> first_row_id,
+                                std::optional<int64_t> data_sequence_number) {
   return ReaderOptions{
       .path = data_file.file_path,
       .length = static_cast<size_t>(data_file.file_size_in_bytes),
       .io = std::move(io),
       .projection = std::move(projection),
       .filter = std::move(filter),
       .name_mapping = std::move(name_mapping),
+      .first_row_id = first_row_id,
+      .data_sequence_number = data_sequence_number,
       .properties = std::move(properties),
   };
 }
@@ -161,9 +165,9 @@ class FileScanTaskReader::Impl {
                      data_file->file_size_in_bytes);
 
     if (task.delete_files().empty()) {
-      auto options =
-          MakeReaderOptions(*data_file, io_, projected_schema_, task.residual_filter(),
-                            name_mapping_, properties_);
+      auto options = MakeReaderOptions(
+          *data_file, io_, projected_schema_, task.residual_filter(), 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));
@@ -187,8 +191,9 @@ class FileScanTaskReader::Impl {
                             ProjectionContext::Make(*required_schema, *projected_schema_,
                                                     project_batch_function));
 
-    auto options = MakeReaderOptions(*data_file, io_, required_schema,
-                                     task.residual_filter(), name_mapping_, properties_);
+    auto options = MakeReaderOptions(
+        *data_file, io_, required_schema, task.residual_filter(), 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

@@ -22,6 +22,7 @@
 /// \file iceberg/file_reader.h
 /// Reader interface for file formats like Parquet, Avro and ORC.
 
+#include <cstdint>
 #include <functional>
 #include <memory>
 #include <optional>
@@ -105,6 +106,10 @@ struct ICEBERG_EXPORT ReaderOptions {
   /// \brief Name mapping for schema evolution compatibility. Used when reading files
   /// that may have different field names than the current schema.
   std::shared_ptr<class NameMapping> name_mapping;
+  /// \brief First row ID inherited by the data file, if assigned.
+  std::optional<int64_t> first_row_id;
+  /// \brief Data sequence number inherited by the manifest entry, if assigned.
+  std::optional<int64_t> data_sequence_number;
   /// \brief Format-specific or implementation-specific properties.
   ReaderProperties properties;
 };

src/iceberg/inheritable_metadata.cc

@@ -64,6 +64,9 @@ Status BaseInheritableMetadata::Apply(ManifestEntry& entry) {
 
   if (entry.data_file) {
     entry.data_file->partition_spec_id = spec_id_;
+    entry.data_file->data_sequence_number = entry.sequence_number;
+    // TODO(gangwu): Also propagate file_sequence_number and manifest_location
+    // when C++ adds consumers that need Java-style inherited file metadata.
   } else {
     return InvalidManifest("Manifest entry has no data file");
   }

src/iceberg/manifest/manifest_entry.h

@@ -174,11 +174,18 @@ struct ICEBERG_EXPORT DataFile {
   /// present
   std::optional<int64_t> content_size_in_bytes;
 
+  /// \note Fields defined below are inherited and not persisted to the DataFile struct.
+
   /// \brief Partition spec id for this data file.
-  /// \note This field is for internal use only and will not be persisted to manifest
-  /// entry.
+  /// \note This inherited field is for internal use only and will not be persisted to
+  /// the DataFile struct.
   std::optional<int32_t> partition_spec_id;
 
+  /// \brief Data sequence number for this data file.
+  /// \note This inherited field is for internal use only and will not be persisted to
+  /// the DataFile struct.
+  std::optional<int64_t> data_sequence_number;
+
   static constexpr int32_t kContentFieldId = 134;
   inline static const SchemaField kContent = SchemaField::MakeOptional(
       kContentFieldId, "content", int32(),