Skip to content

feat: read v3 row lineage metadata columns #822

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wgtmac wants to merge 3 commits into
base: main
Choose a base branch
from
Open

feat: read v3 row lineage metadata columns #822

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
35d2101
feat: read v3 row lineage metadata columns
wgtmac Jul 6, 2026
46b9737
address comment
wgtmac Jul 11, 2026
857a00c
address comment
wgtmac Jul 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 63 additions & 0 deletions src/iceberg/arrow/metadata_column_util.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,9 +24,35 @@

#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 HasRowLineageValue(int32_t field_id, const MetadataColumnContext& metadata_context) {
if (!metadata_context.first_row_id.has_value()) {
return false;
}
return field_id != MetadataColumns::kLastUpdatedSequenceNumberColumnId ||
metadata_context.data_sequence_number.has_value();
}

Status AppendRowLineageValue(int32_t field_id,
const MetadataColumnContext& metadata_context,
::arrow::ArrayBuilder* array_builder) {
auto* int_builder = internal::checked_cast<::arrow::Int64Builder*>(array_builder);
if (!HasRowLineageValue(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) {
Expand All @@ -53,4 +79,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> first_row_id, 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 (!first_row_id.has_value() || !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
37 changes: 37 additions & 0 deletions src/iceberg/arrow/metadata_column_util_internal.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,20 +24,35 @@

#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 HasRowLineageValue(int32_t field_id, const MetadataColumnContext& metadata_context);

/// \brief Append one inherited row lineage value to an int64 builder.
Status AppendRowLineageValue(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.
Expand All @@ -62,4 +77,26 @@ 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 first_row_id First row ID of the data file.
/// \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> first_row_id, std::optional<int64_t> data_sequence_number,
int64_t num_rows, ::arrow::MemoryPool* pool);

} // namespace iceberg::arrow
15 changes: 15 additions & 0 deletions src/iceberg/avro/avro_data_util.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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::AppendRowLineageValue(field_id, metadata_context, field_builder));
} else {
return NotSupported("Unsupported metadata column field id: {}", field_id);
}
Expand Down Expand Up @@ -463,9 +466,21 @@ Status AppendFieldToBuilder(const ::avro::NodePtr& avro_node,
return {};
}

const bool is_row_lineage =
MetadataColumns::IsRowLineageColumn(projected_field.field_id());
if (is_row_lineage &&
!arrow::HasRowLineageValue(projected_field.field_id(), metadata_context)) {
ICEBERG_ARROW_RETURN_NOT_OK(array_builder->AppendNull());
return {};
}

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::AppendRowLineageValue(projected_field.field_id(), metadata_context,
array_builder);
}
ICEBERG_ARROW_RETURN_NOT_OK(array_builder->AppendNull());
return {};
} else {
Expand Down
16 changes: 16 additions & 0 deletions src/iceberg/avro/avro_direct_decoder.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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::AppendRowLineageValue(field_id, metadata_context, field_builder));
} else {
return NotSupported("Unsupported metadata column field id: {}", field_id);
}
Expand Down Expand Up @@ -594,6 +597,15 @@ Status DecodeFieldToBuilder(const ::avro::NodePtr& avro_node, ::avro::Decoder& d
return {};
}

const bool is_row_lineage =
MetadataColumns::IsRowLineageColumn(projected_field.field_id());
if (is_row_lineage &&
!arrow::HasRowLineageValue(projected_field.field_id(), metadata_context)) {
ICEBERG_RETURN_UNEXPECTED(SkipAvroValue(avro_node, decoder));
ICEBERG_ARROW_RETURN_NOT_OK(array_builder->AppendNull());
return {};
}

if (avro_node->type() == ::avro::AVRO_UNION) {
const size_t branch_index = decoder.decodeUnionIndex();

Expand All @@ -606,6 +618,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::AppendRowLineageValue(projected_field.field_id(), metadata_context,
array_builder);
}
ICEBERG_ARROW_RETURN_NOT_OK(array_builder->AppendNull());
return {};
} else {
Expand Down
19 changes: 18 additions & 1 deletion src/iceberg/avro/avro_reader.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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 {};
}
Expand Down
3 changes: 3 additions & 0 deletions src/iceberg/catalog/rest/json_serde.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -323,6 +323,9 @@ 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.

std::vector<std::shared_ptr<DataFile>> task_delete_files;
if (task_json.contains(kDeleteFileReferences) &&
Expand Down
17 changes: 11 additions & 6 deletions src/iceberg/data/file_scan_task_reader.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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),
};
}
Expand Down Expand Up @@ -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(

@manuzhang manuzhang Jul 7, 2026 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

REST scan-task JSON appears to round-trip only first-row-id for DataFile; FileScanTasksFromJson constructs the FileScanTask without any data sequence number. That means this call can pass data_file->first_row_id but an empty data_file->data_sequence_number, and MakeLastUpdatedSequenceNumberArray will return nulls for _last_updated_sequence_number.

The row-lineage spec says missing/null _last_updated_sequence_number should be replaced with the data file’s data sequence number.

Disclosure: I used an LLM to help review this PR and draft this comment.

@wgtmac wgtmac Jul 11, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I looked into the Java REST implementation and OpenAPI. This is not C++-specific: RESTFileScanTaskParser serializes the task file through ContentFileParser, and ContentFileParser only round-trips first-row-id, not the manifest-entry data sequence number. The OpenAPI FileScanTask/DataFile schema also has first-row-id but no field for the data sequence number.

So the failing case is REST-planned scan tasks reading row lineage metadata, where _last_updated_sequence_number needs the manifest entry sequence number but that value is not part of the REST contract. I think the correct fix should start in the REST API / Java parser, then C++ can implement the same field. I don’t think C++ should add a non-standard private field here.

@manuzhang manuzhang Jul 13, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed, the fix is beyond this PR. We can create a follow-up issue to track.

*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));
Expand All @@ -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));

Expand Down
5 changes: 5 additions & 0 deletions src/iceberg/file_reader.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand Down Expand Up @@ -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;
};
Expand Down
3 changes: 3 additions & 0 deletions src/iceberg/inheritable_metadata.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Comment thread
wgtmac marked this conversation as resolved.
// 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");
}
Expand Down
11 changes: 9 additions & 2 deletions src/iceberg/manifest/manifest_entry.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(),
Expand Down
4 changes: 4 additions & 0 deletions src/iceberg/metadata_columns.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,10 @@ bool MetadataColumns::IsMetadataColumn(int32_t id) {
return GetMetadataFieldIdSet().find(id) != GetMetadataFieldIdSet().end();
}

bool MetadataColumns::IsRowLineageColumn(int32_t id) {
return id == kRowIdColumnId || id == kLastUpdatedSequenceNumberColumnId;
}

Result<const SchemaField*> MetadataColumns::MetadataColumn(std::string_view name) {
const auto& metadata_column_map = GetMetadataColumnMap();
const auto it = metadata_column_map.find(name);
Expand Down
3 changes: 3 additions & 0 deletions src/iceberg/metadata_columns.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,6 +113,9 @@ struct ICEBERG_EXPORT MetadataColumns {
/// \brief Check if a column ID is a metadata column.
static bool IsMetadataColumn(int32_t id);

/// \brief Check if a column ID is a row lineage column.
static bool IsRowLineageColumn(int32_t id);

/// \brief Get a metadata column by name.
///
/// \param name The name of the metadata column.
Expand Down
Loading
Loading