Skip to content

feat: carry first_row_id and data sequence number on FileScanTask #2952

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

Merged
laskoviymishka merged 2 commits into from
Aug 4, 2026
Merged

feat: carry first_row_id and data sequence number on FileScanTask #2952

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c226076
feat: carry first_row_id and data sequence number on FileScanTask
anoopj Aug 3, 2026
d48a73a
test: address review on row-lineage scan task
anoopj Aug 4, 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
4 changes: 3 additions & 1 deletion crates/iceberg/public-api.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1266,8 +1266,10 @@ pub struct iceberg::scan::FileScanTask
pub iceberg::scan::FileScanTask::case_sensitive: bool
pub iceberg::scan::FileScanTask::data_file_format: iceberg::spec::DataFileFormat
pub iceberg::scan::FileScanTask::data_file_path: alloc::string::String
pub iceberg::scan::FileScanTask::data_sequence_number: core::option::Option<i64>
pub iceberg::scan::FileScanTask::deletes: alloc::vec::Vec<iceberg::scan::FileScanTaskDeleteFile>
pub iceberg::scan::FileScanTask::file_size_in_bytes: u64
pub iceberg::scan::FileScanTask::first_row_id: core::option::Option<i64>
pub iceberg::scan::FileScanTask::key_metadata: core::option::Option<alloc::boxed::Box<[u8]>>
pub iceberg::scan::FileScanTask::length: u64
pub iceberg::scan::FileScanTask::name_mapping: core::option::Option<alloc::sync::Arc<iceberg::spec::NameMapping>>
Expand All @@ -1293,7 +1295,7 @@ impl core::fmt::Debug for iceberg::scan::FileScanTask
pub fn iceberg::scan::FileScanTask::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
impl core::marker::StructuralPartialEq for iceberg::scan::FileScanTask
impl iceberg::scan::FileScanTask
pub fn iceberg::scan::FileScanTask::builder() -> FileScanTaskBuilder<((), (), (), (), (), (), (), (), (), (), (), (), (), (), (), ())>
pub fn iceberg::scan::FileScanTask::builder() -> FileScanTaskBuilder<((), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), ())>
impl serde_core::ser::Serialize for iceberg::scan::FileScanTask
pub fn iceberg::scan::FileScanTask::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer
impl<'de> serde_core::de::Deserialize<'de> for iceberg::scan::FileScanTask
Expand Down
4 changes: 4 additions & 0 deletions crates/iceberg/src/arrow/reader/row_filter.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1138,6 +1138,8 @@ mod tests {
start: 0,
length: 0,
record_count: None,
first_row_id: None,
data_sequence_number: None,
data_file_path: file_path.clone(),
data_file_format: DataFileFormat::Parquet,
schema: iceberg_schema.clone(),
Expand Down Expand Up @@ -1231,6 +1233,8 @@ mod tests {
start: 0,
length: 0,
record_count: None,
first_row_id: None,
data_sequence_number: None,
data_file_path: file_path.clone(),
data_file_format: DataFileFormat::Parquet,
schema: iceberg_schema.clone(),
Expand Down
2 changes: 2 additions & 0 deletions crates/iceberg/src/scan/context.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -133,6 +133,8 @@ impl ManifestEntryContext {
.with_start(0)
.with_length(self.manifest_entry.file_size_in_bytes())
.with_record_count(Some(self.manifest_entry.record_count()))
.with_first_row_id(self.manifest_entry.data_file().first_row_id())
.with_data_sequence_number(self.manifest_entry.sequence_number())
.with_data_file_path(self.manifest_entry.file_path().to_string())
.with_data_file_format(self.manifest_entry.file_format())
.with_schema(self.snapshot_schema)
Expand Down
148 changes: 147 additions & 1 deletion crates/iceberg/src/scan/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -669,7 +669,7 @@ pub mod tests {
use crate::scan::FileScanTask;
use crate::spec::{
DEFAULT_SCHEMA_NAME_MAPPING, DataContentType, DataFileBuilder, DataFileFormat, Datum,
Literal, MAIN_BRANCH, ManifestEntry, ManifestListWriter, ManifestStatus,
FormatVersion, Literal, MAIN_BRANCH, ManifestEntry, ManifestListWriter, ManifestStatus,
ManifestWriterBuilder, NestedField, Operation, PartitionSpec, PrimitiveType, Schema,
Snapshot, Struct, StructType, Summary, TableMetadata, TableMetadataBuilder, Type,
UnboundPartitionSpec,
Expand Down Expand Up @@ -996,6 +996,84 @@ pub mod tests {
manifest_list_write.close().await.unwrap();
}

/// Writes a v3 data manifest with a manifest-level `first_row_id` of 42,
/// so live entries inherit a per-file `first_row_id` on read. Upgrades the
/// table to v3 first, so the manifest list is read as v3.
pub async fn setup_v3_manifest_files(&mut self) {
let metadata = TableMetadataBuilder::new_from_metadata(
self.table.metadata().clone(),
self.table.metadata_location().map(str::to_string),
)
.upgrade_format_version(FormatVersion::V3)
.unwrap()
.build()
.unwrap()
.metadata;
self.table = Table::builder()
.metadata(metadata)
.identifier(self.table.identifier().clone())
.file_io(self.table.file_io().clone())
.metadata_location(self.table.metadata_location().unwrap().to_string())
.runtime(test_runtime())
.build()
.unwrap();

let current_snapshot = self.table.metadata().current_snapshot().unwrap();
let current_schema = current_snapshot.schema(self.table.metadata()).unwrap();
let current_partition_spec = self.table.metadata().default_partition_spec();

let parquet_file_size = self.write_parquet_data_files();

let mut writer = ManifestWriterBuilder::new(
self.next_manifest_file(),
Some(current_snapshot.snapshot_id()),
current_schema.clone(),
current_partition_spec.as_ref().clone(),
)
.build_v3_data();
writer
.add_entry(
ManifestEntry::builder()
.status(ManifestStatus::Added)
.data_file(
DataFileBuilder::default()
.partition_spec_id(0)
.content(DataContentType::Data)
.file_path(format!("{}/1.parquet", &self.table_location))
.file_format(DataFileFormat::Parquet)
.file_size_in_bytes(parquet_file_size)
.record_count(1)
.partition(Struct::from_iter([Some(Literal::long(100))]))
.key_metadata(None)
.build()
.unwrap(),
)
.build(),
)
.unwrap();
let data_file_manifest = writer.write_manifest_file().await.unwrap();

let manifest_list_writer = self
.table
.file_io()
.new_output(current_snapshot.manifest_list())
.unwrap()
.writer()
.await
.unwrap();
let mut manifest_list_write = ManifestListWriter::v3(
manifest_list_writer,
current_snapshot.snapshot_id(),
current_snapshot.parent_snapshot_id(),
current_snapshot.sequence_number(),
Some(42),
);
manifest_list_write
.add_manifests(vec![data_file_manifest].into_iter())
.unwrap();
manifest_list_write.close().await.unwrap();
}

pub async fn setup_manifest_files_with_partition_evolution(&mut self) {
let current_snapshot = self.table.metadata().current_snapshot().unwrap();
let parent_snapshot = current_snapshot
Expand Down Expand Up @@ -1862,6 +1940,70 @@ pub mod tests {
);
}

#[tokio::test]
async fn test_plan_files_carries_row_lineage_into_file_scan_task() {
let mut fixture = TableTestFixture::new();
fixture.setup_manifest_files().await;

let mut tasks: Vec<_> = fixture
.table
.scan()
.build()
.unwrap()
.plan_files()
.await
.unwrap()
.try_collect()
.await
.unwrap();

assert_eq!(tasks.len(), 2);
tasks.sort_by_key(|task| task.data_file_path.to_string());

// The added file inherits the current snapshot's data sequence number,
// the existing file keeps the one it was written with.
assert_eq!(
tasks[0].data_file_path,
format!("{}/1.parquet", &fixture.table_location)
);
assert_eq!(tasks[0].data_sequence_number, Some(1));

@laskoviymishka laskoviymishka Aug 4, 2026

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.

The index-to-file mapping is implicit — it holds only because 1.parquet sorts ahead of 3.parquet, and the comment describes that rather than asserting it. Rename those fixture files and the two assertions silently swap and still pass. I'd pin the path next to each sequence number:

assert_eq!(tasks[0].data_file_path, format!("{}/1.parquet", &fixture.table_location));
assert_eq!(tasks[0].data_sequence_number, Some(1));
assert_eq!(tasks[1].data_file_path, format!("{}/3.parquet", &fixture.table_location));
assert_eq!(tasks[1].data_sequence_number, Some(0));

While we're here, the tasks.len() assert wants to be above the sort — as written, a scan that returns fewer tasks panics on tasks[0] instead of failing with the length message.

@anoopj anoopj Aug 4, 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.

Done both. Pinned data_file_path next to each data_sequence_number assertion so the file->value mapping is explicit and survives a fixture rename, and moved the task len assert above the sort .

assert_eq!(
tasks[1].data_file_path,
format!("{}/3.parquet", &fixture.table_location)
);
assert_eq!(tasks[1].data_sequence_number, Some(0));

// first_row_id is a v3 concept; a v2 manifest carries none.
assert!(tasks.iter().all(|task| task.first_row_id.is_none()));
}

#[tokio::test]
async fn test_plan_files_carries_row_lineage_from_v3_manifest() {
let mut fixture = TableTestFixture::new();
fixture.setup_v3_manifest_files().await;

let task = fixture
.table
.scan()
.build()
.unwrap()
.plan_files()
.await
.unwrap()
.try_collect::<Vec<_>>()
.await
.unwrap()
.into_iter()
.next()
.expect("expected one FileScanTask");

// The manifest-level first_row_id (42) is inherited onto the entry on
// read, then carried onto the task.
assert_eq!(task.first_row_id, Some(42));
// The data sequence number is threaded through the same v3 read path.
assert_eq!(task.data_sequence_number, Some(1));
}

#[tokio::test]
async fn test_filtered_scan_with_dropped_partition_source_column() {
let mut fixture = TableTestFixture::new();
Expand Down Expand Up @@ -2441,6 +2583,8 @@ pub mod tests {
assert_eq!(task.project_field_ids, deserialized.project_field_ids);
assert_eq!(task.predicate, deserialized.predicate);
assert_eq!(task.schema, deserialized.schema);
assert_eq!(task.first_row_id, deserialized.first_row_id);
assert_eq!(task.data_sequence_number, deserialized.data_sequence_number);
};

// without predicate
Expand All @@ -2462,6 +2606,8 @@ pub mod tests {
.with_project_field_ids(vec![1, 2, 3])
.with_schema(schema.clone())
.with_record_count(Some(100))
.with_first_row_id(Some(1000))
.with_data_sequence_number(Some(5))
.with_data_file_format(DataFileFormat::Parquet)
.with_case_sensitive(false)
.build();
Expand Down
18 changes: 18 additions & 0 deletions crates/iceberg/src/scan/task.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,24 @@ pub struct FileScanTask {
#[builder(default)]
pub record_count: Option<u64>,

/// The first row id assigned to the data file.
///
/// Used to derive the `_row_id` metadata column: for a row without an
/// explicit `_row_id`, it is this value plus the row's ordinal position.
#[serde(skip_serializing_if = "Option::is_none")]
#[builder(default)]
pub first_row_id: Option<i64>,

/// The data sequence number of the file, as opposed to its file sequence
/// number: the sequence number preserved when a file is carried forward
/// across a rewrite. May be null for an existing entry in a malformed
/// manifest that lacks one.
///
/// Used to derive the `_last_updated_sequence_number` metadata column.
#[serde(skip_serializing_if = "Option::is_none")]
#[builder(default)]
pub data_sequence_number: Option<i64>,

/// The data file path corresponding to the task.
pub data_file_path: String,

Expand Down
Loading