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 1 commit 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
1 commit
Select commit
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
46 changes: 46 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,18 @@

#include "iceberg/arrow/arrow_status_internal.h"
#include "iceberg/arrow/metadata_column_util_internal.h"
#include "iceberg/metadata_columns.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();
}

Result<std::shared_ptr<::arrow::Array>> MakeFilePathArray(const std::string& file_path,
int64_t num_rows,
::arrow::MemoryPool* pool) {
Expand All @@ -53,4 +62,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
28 changes: 28 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,6 +24,7 @@

#include <cstdint>
#include <memory>
#include <optional>
#include <string>

#include <arrow/type_fwd.h>
Expand All @@ -36,8 +37,13 @@ namespace iceberg::arrow {
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 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 +68,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
31 changes: 31 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 @@ -54,6 +54,22 @@ Status AppendFieldToBuilder(const ::avro::NodePtr& avro_node,
const arrow::MetadataColumnContext& metadata_context,
::arrow::ArrayBuilder* array_builder);

Status AppendRowLineageValue(int32_t field_id,

@WZhuo WZhuo Jul 7, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

AppendRowLineageValue is defined verbatim in both avro_data_util.cc:57 and avro_direct_decoder.cc:147 with identical logic.

const arrow::MetadataColumnContext& metadata_context,
::arrow::ArrayBuilder* array_builder) {
auto* int_builder = internal::checked_cast<::arrow::Int64Builder*>(array_builder);
if (!arrow::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 {};
}

/// \brief Append Avro record data to Arrow struct builder.
Status AppendStructToBuilder(const ::avro::NodePtr& avro_node,
const ::avro::GenericDatum& avro_datum,
Expand Down Expand Up @@ -97,6 +113,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(
AppendRowLineageValue(field_id, metadata_context, field_builder));
} else {
return NotSupported("Unsupported metadata column field id: {}", field_id);
}
Expand Down Expand Up @@ -463,9 +482,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 AppendRowLineageValue(projected_field.field_id(), metadata_context,
array_builder);
}
ICEBERG_ARROW_RETURN_NOT_OK(array_builder->AppendNull());
return {};
} else {
Expand Down
32 changes: 32 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 @@ -144,6 +144,22 @@ Status SkipAvroValue(const ::avro::NodePtr& avro_node, ::avro::Decoder& decoder)
}
}

Status AppendRowLineageValue(int32_t field_id,
const arrow::MetadataColumnContext& metadata_context,
::arrow::ArrayBuilder* array_builder) {
auto* int_builder = internal::checked_cast<::arrow::Int64Builder*>(array_builder);
if (!arrow::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 {};
}

/// \brief Decode Avro record directly to Arrow struct builder.
Status DecodeStructToBuilder(const ::avro::NodePtr& avro_node, ::avro::Decoder& decoder,
const std::span<const FieldProjection>& projections,
Expand Down Expand Up @@ -218,6 +234,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(
AppendRowLineageValue(field_id, metadata_context, field_builder));
} else {
return NotSupported("Unsupported metadata column field id: {}", field_id);
}
Expand Down Expand Up @@ -594,6 +613,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 +634,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 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
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.

*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
1 change: 1 addition & 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,7 @@ 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;

@WZhuo WZhuo Jul 7, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

BaseInheritableMetadata::Apply() only propagates data_sequence_number to the DataFile, but the Java equivalent (InheritableMetadataFactory.BaseInheritableMetadata.apply()) sets four fields:

file.setSpecId(specId);
file.setDataSequenceNumber(manifestEntry.dataSequenceNumber());
file.setFileSequenceNumber(manifestEntry.fileSequenceNumber());
file.setManifestLocation(manifestLocation);

Suggestion: either add the fields and propagate them now, or leave an explicit // TODO: file_sequence_number and manifest_location comment so the gap is visible.

} 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