You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by at...@apache.org on 2026/08/10 09:22:32 UTC
(iceberg-rust) branch main updated: feat: materialize _last_updated_sequence_number metadata column (#2966)
This is an automated email from the ASF dual-hosted git repository.
laskoviymishka pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
The following commit(s) were added to refs/heads/main by this push:
new e8a47c450 feat: materialize _last_updated_sequence_number metadata column (#2966)
e8a47c450 is described below
commit e8a47c45041f3e0c58e4f927261a3b384b5249e0
Author: Anoop Johnson <an...@apache.org>
AuthorDate: Mon Aug 10 02:22:27 2026 -0700
feat: materialize _last_updated_sequence_number metadata column (#2966)
* feat: materialize _last_updated_sequence_number metadata column
Note: A file may physically carry a per-row materialized column. e.g.
when carrying rows forward across a rewrite. Per-row coalesce is not implemented yet, so
such files are rejected with FeatureUnsupported
The per-row coalesce, and the _row_id column, follow in later PRs.
* Address review comments from Andrei
* cargo fmt
---
crates/iceberg/src/arrow/reader/pipeline.rs | 307 ++++++++++++++++++++-
.../iceberg/src/arrow/record_batch_transformer.rs | 154 ++++++++++-
crates/iceberg/src/arrow/schema.rs | 10 +-
crates/iceberg/src/scan/mod.rs | 92 +++++-
4 files changed, 555 insertions(+), 8 deletions(-)
diff --git a/crates/iceberg/src/arrow/reader/pipeline.rs b/crates/iceberg/src/arrow/reader/pipeline.rs
index e7bcdf5fb..7006c20bf 100644
--- a/crates/iceberg/src/arrow/reader/pipeline.rs
+++ b/crates/iceberg/src/arrow/reader/pipeline.rs
@@ -43,7 +43,8 @@ use crate::encryption::StandardKeyMetadata;
use crate::error::Result;
use crate::io::{FileIO, FileMetadata, FileRead};
use crate::metadata_columns::{
- RESERVED_COL_NAME_POS, RESERVED_FIELD_ID_FILE, RESERVED_FIELD_ID_PARTITION,
+ RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER, RESERVED_COL_NAME_POS, RESERVED_FIELD_ID_FILE,
+ RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER, RESERVED_FIELD_ID_PARTITION,
RESERVED_FIELD_ID_POS, RESERVED_FIELD_ID_SPEC_ID, is_metadata_field,
};
use crate::scan::{ArrowRecordBatchStream, FileScanTask, FileScanTaskStream};
@@ -300,6 +301,77 @@ impl FileScanTaskReader {
.with_constant(RESERVED_FIELD_ID_SPEC_ID, spec_id_datum);
}
+ if task
+ .project_field_ids()
+ .contains(&RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER)
+ {
+ // A data file may physically carry a per-row `_last_updated_sequence_number`
+ // column, e.g. one written by another engine such as Iceberg Java when carrying
+ // rows forward across a rewrite. The spec requires reading such non-null
+ // per-row values unmodified, falling back to the derived value only where
+ // null. That per-row coalesce is not implemented yet, so rather than silently
+ // overwrite genuine per-row values with the derived value, reject the file
+ // loudly. Checks the full pre-projection file schema, since the column is
+ // stripped from the projection mask. Matches on the embedded field id when
+ // present (propagating a malformed id rather than treating it as absent) and
+ // falls back to the column name, so a file read via name mapping or positional
+ // fallback ids (which never equal the reserved id) is still caught.
+ let mut file_has_column = false;
+ for field in record_batch_stream_builder.schema().fields() {
+ let field_id = match field.metadata().get(PARQUET_FIELD_ID_META_KEY) {
+ Some(id) => Some(id.parse::<i32>().map_err(|e| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!("field id not parseable as an i32: {e}"),
+ )
+ })?),
+ None => None,
+ };
+ if field_id == Some(RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER)
+ || field.name() == RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER
+ {
+ file_has_column = true;
+ break;
+ }
+ }
+ if file_has_column {
+ return Err(Error::new(
+ ErrorKind::FeatureUnsupported,
+ "Reading a physically-stored _last_updated_sequence_number column is \
+ not yet supported; only the derived (data-sequence-number) value is \
+ implemented",
+ ));
+ }
+
+ // Derive the column, gated on the data file's `first_row_id`. Java gates
+ // both lineage columns this way (`ValueReaders.lastUpdated` returns nulls
+ // when the base row id is null); the spec itself only says the column is
+ // assigned the manifest entry's sequence number on read.
+ record_batch_transformer_builder = match (task.first_row_id, task.data_sequence_number)
+ {
+ // Non-null first_row_id: inherit from the data sequence number.
+ (Some(_), Some(seq)) => record_batch_transformer_builder.with_constant(
+ RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER,
+ Datum::long(seq),
+ ),
+ // Null first_row_id (v1/v2, or a pre-upgrade v3 snapshot): the column is null.
+ (None, _) => record_batch_transformer_builder
+ .with_null_metadata_column(RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER)?,
+ // first_row_id present but no data sequence number: after manifest
+ // inheritance a committed entry always has one, so this is a malformed
+ // manifest rather than a legitimate null.
+ (Some(_), None) => {
+ return Err(Error::new(
+ ErrorKind::DataInvalid,
+ format!(
+ "Data file {} has a first_row_id but no data sequence number",
+ task.data_file_path
+ ),
+ ));
+ }
+ };
+ }
+
if let (Some(partition_spec), Some(partition_data)) =
(task.partition_spec.clone(), task.partition.clone())
{
@@ -576,7 +648,7 @@ mod tests {
use std::sync::Arc;
use arrow_array::cast::AsArray;
- use arrow_array::{Array, ArrayRef, Int32Array, RecordBatch};
+ use arrow_array::{Array, ArrayRef, Int32Array, Int64Array, RecordBatch};
use arrow_schema::{DataType, Field, Schema as ArrowSchema};
use futures::TryStreamExt;
use parquet::arrow::{ArrowWriter, PARQUET_FIELD_ID_META_KEY};
@@ -909,6 +981,237 @@ mod tests {
);
}
+ /// Writes a plain (unencrypted) single-column Int32 "id" parquet file with the
+ /// given extra Arrow fields/columns appended, returning the file path.
+ fn write_plain_parquet(
+ dir: &str,
+ name: &str,
+ extra_fields: Vec<Field>,
+ extra_columns: Vec<ArrayRef>,
+ ) -> String {
+ let mut fields =
+ vec![
+ Field::new("id", DataType::Int32, false).with_metadata(HashMap::from([(
+ PARQUET_FIELD_ID_META_KEY.to_string(),
+ "1".to_string(),
+ )])),
+ ];
+ fields.extend(extra_fields);
+ let arrow_schema = Arc::new(ArrowSchema::new(fields));
+
+ let mut columns: Vec<ArrayRef> = vec![Arc::new(Int32Array::from(vec![1, 2, 3]))];
+ columns.extend(extra_columns);
+ let batch = RecordBatch::try_new(arrow_schema.clone(), columns).unwrap();
+
+ let file_path = format!("{dir}/{name}");
+ let file = File::create(&file_path).unwrap();
+ let props = WriterProperties::builder()
+ .set_compression(Compression::SNAPPY)
+ .build();
+ let mut writer = ArrowWriter::try_new(file, arrow_schema, Some(props)).unwrap();
+ writer.write(&batch).unwrap();
+ writer.close().unwrap();
+ file_path
+ }
+
+ fn last_updated_seq_task(
+ file_path: String,
+ first_row_id: Option<i64>,
+ data_sequence_number: Option<i64>,
+ ) -> FileScanTask {
+ use crate::metadata_columns::RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER;
+
+ let schema = Arc::new(
+ Schema::builder()
+ .with_schema_id(1)
+ .with_fields(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(),
+ ])
+ .build()
+ .unwrap(),
+ );
+
+ FileScanTask::builder()
+ .with_file_size_in_bytes(std::fs::metadata(&file_path).unwrap().len())
+ .with_start(0)
+ .with_length(0)
+ .with_data_file_path(file_path)
+ .with_data_file_format(DataFileFormat::Parquet)
+ .with_schema(schema)
+ .with_project_field_ids(vec![1, RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER])
+ .with_first_row_id(first_row_id)
+ .with_data_sequence_number(data_sequence_number)
+ .with_case_sensitive(false)
+ .build()
+ }
+
+ /// Asserts the logical per-row values of the `_last_updated_sequence_number`
+ /// column across all batches, independent of the physical (run-end) encoding.
+ fn assert_last_updated_seq_column(batches: &[RecordBatch], expected: &[Option<i64>]) {
+ use arrow_array::cast::AsArray;
+ use arrow_cast::cast;
+ use arrow_schema::DataType;
+
+ use crate::metadata_columns::RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER;
+
+ let mut actual = Vec::new();
+ for batch in batches {
+ let col = batch
+ .column_by_name(RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER)
+ .expect("_last_updated_sequence_number column should be present");
+ let logical = cast(col, &DataType::Int64).unwrap();
+ let values = logical.as_primitive::<arrow_array::types::Int64Type>();
+ for i in 0..values.len() {
+ actual.push((!values.is_null(i)).then(|| values.value(i)));
+ }
+ }
+ assert_eq!(actual, expected);
+ }
+
+ #[tokio::test]
+ async fn test_last_updated_sequence_number_null_when_no_first_row_id() {
+ let tmp_dir = TempDir::new().unwrap();
+ let dir = tmp_dir.path().to_str().unwrap();
+ let file_path = write_plain_parquet(dir, "no_first_row_id.parquet", vec![], vec![]);
+
+ // A file with a null first_row_id (v1/v2, or a pre-upgrade v3 snapshot) produces
+ // a null _last_updated_sequence_number column, even though it has a data
+ // sequence number; the spec gates both lineage columns on first_row_id.
+ let task = last_updated_seq_task(file_path, None, Some(9));
+
+ let reader = ArrowReaderBuilder::new(FileIO::new_with_fs(), Runtime::current()).build();
+ let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream;
+ let batches: Vec<RecordBatch> = reader
+ .read(tasks)
+ .unwrap()
+ .stream()
+ .try_collect()
+ .await
+ .unwrap();
+
+ assert_last_updated_seq_column(&batches, &[None, None, None]);
+ }
+
+ #[tokio::test]
+ async fn test_last_updated_sequence_number_error_when_no_data_seq() {
+ let tmp_dir = TempDir::new().unwrap();
+ let dir = tmp_dir.path().to_str().unwrap();
+ let file_path = write_plain_parquet(dir, "no_data_seq.parquet", vec![], vec![]);
+
+ // first_row_id present but data_sequence_number absent: after manifest
+ // inheritance a committed entry always has one, so this is a malformed
+ // manifest and must error rather than fabricate or null the column.
+ let task = last_updated_seq_task(file_path, Some(42), None);
+
+ let reader = ArrowReaderBuilder::new(FileIO::new_with_fs(), Runtime::current()).build();
+ let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream;
+ let result: Result<Vec<RecordBatch>, _> =
+ reader.read(tasks).unwrap().stream().try_collect().await;
+
+ let err = result.unwrap_err();
+ assert_eq!(err.kind(), crate::ErrorKind::DataInvalid);
+ assert!(
+ format!("{err}").contains("no data sequence number"),
+ "unexpected error: {err}"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_last_updated_sequence_number_derived_from_data_seq() {
+ let tmp_dir = TempDir::new().unwrap();
+ let dir = tmp_dir.path().to_str().unwrap();
+ let file_path = write_plain_parquet(dir, "with_first_row_id.parquet", vec![], vec![]);
+
+ // Non-null first_row_id + data sequence number -> the derived value (the data
+ // sequence number) for every row. This is the only value-producing arm.
+ let task = last_updated_seq_task(file_path, Some(42), Some(7));
+
+ let reader = ArrowReaderBuilder::new(FileIO::new_with_fs(), Runtime::current()).build();
+ let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream;
+ let batches: Vec<RecordBatch> = reader
+ .read(tasks)
+ .unwrap()
+ .stream()
+ .try_collect()
+ .await
+ .unwrap();
+
+ assert_last_updated_seq_column(&batches, &[Some(7), Some(7), Some(7)]);
+ }
+
+ #[tokio::test]
+ async fn test_last_updated_sequence_number_mixed_files_share_schema() {
+ use arrow_select::concat::concat_batches;
+
+ let tmp_dir = TempDir::new().unwrap();
+ let dir = tmp_dir.path().to_str().unwrap();
+
+ // Two files in one scan: one with first_row_id (derived value -> REE constant),
+ // one without (null gate -> REE null). Both must produce the same Arrow type for
+ // the column, or concatenation fails; the regression this branch fixes.
+ let with_id = last_updated_seq_task(
+ write_plain_parquet(dir, "with_id.parquet", vec![], vec![]),
+ Some(42),
+ Some(7),
+ );
+ let without_id = last_updated_seq_task(
+ write_plain_parquet(dir, "without_id.parquet", vec![], vec![]),
+ None,
+ Some(7),
+ );
+
+ let reader = ArrowReaderBuilder::new(FileIO::new_with_fs(), Runtime::current()).build();
+ let tasks = Box::pin(futures::stream::iter(vec![Ok(with_id), Ok(without_id)]))
+ as FileScanTaskStream;
+ let batches: Vec<RecordBatch> = reader
+ .read(tasks)
+ .unwrap()
+ .stream()
+ .try_collect()
+ .await
+ .unwrap();
+
+ assert_eq!(batches.len(), 2);
+ // Identical schema across files -> concat succeeds.
+ let schema = batches[0].schema();
+ concat_batches(&schema, &batches)
+ .expect("batches from value and null files must share one schema");
+ }
+
+ #[tokio::test]
+ async fn test_last_updated_sequence_number_physical_column_unsupported() {
+ use crate::metadata_columns::RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER;
+
+ let tmp_dir = TempDir::new().unwrap();
+ let dir = tmp_dir.path().to_str().unwrap();
+ // A file that physically carries the _last_updated_sequence_number column, as
+ // Iceberg Java writes when carrying rows forward across a rewrite.
+ let seq_field = Field::new("_last_updated_sequence_number", DataType::Int64, true)
+ .with_metadata(HashMap::from([(
+ PARQUET_FIELD_ID_META_KEY.to_string(),
+ RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER.to_string(),
+ )]));
+ let seq_col = Arc::new(Int64Array::from(vec![Some(5), None, Some(8)])) as ArrayRef;
+ let file_path =
+ write_plain_parquet(dir, "with_seq.parquet", vec![seq_field], vec![seq_col]);
+
+ // first_row_id present so the gate would otherwise materialize the constant;
+ // the physically-present column must trigger the fail-loud guard first.
+ let task = last_updated_seq_task(file_path, Some(100), Some(9));
+
+ let reader = ArrowReaderBuilder::new(FileIO::new_with_fs(), Runtime::current()).build();
+ let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream;
+ let result: Result<Vec<RecordBatch>, _> =
+ reader.read(tasks).unwrap().stream().try_collect().await;
+
+ let err = result.unwrap_err();
+ assert_eq!(err.kind(), crate::ErrorKind::FeatureUnsupported);
+ assert!(
+ format!("{err}").contains("physically-stored _last_updated_sequence_number"),
+ "unexpected error: {err}"
+ );
+ }
+
#[tokio::test]
async fn test_read_encrypted_parquet_with_wrong_key_fails() {
let encryption_key = b"0123456789abcdef";
diff --git a/crates/iceberg/src/arrow/record_batch_transformer.rs b/crates/iceberg/src/arrow/record_batch_transformer.rs
index bf7e89910..90a56a990 100644
--- a/crates/iceberg/src/arrow/record_batch_transformer.rs
+++ b/crates/iceberg/src/arrow/record_batch_transformer.rs
@@ -30,7 +30,10 @@ use arrow_schema::{
use parquet::arrow::PARQUET_FIELD_ID_META_KEY;
use crate::arrow::value::{create_primitive_array_repeated, create_primitive_array_single_element};
-use crate::arrow::{datum_to_arrow_type_with_ree, schema_to_arrow_schema, type_to_arrow_type};
+use crate::arrow::{
+ datum_to_arrow_type_with_ree, primitive_type_to_arrow_type_with_ree, schema_to_arrow_schema,
+ type_to_arrow_type,
+};
use crate::metadata_columns::{
RESERVED_COL_NAME_PARTITION, RESERVED_FIELD_ID_PARTITION, get_metadata_field,
};
@@ -235,6 +238,12 @@ pub(crate) enum ColumnConstant {
/// A struct constant (the `_partition` column). Each child is a primitive constant
/// or null (for partition evolution gaps).
Struct(StructConstant),
+ /// An all-null metadata column, carrying its Arrow type. Used when a metadata
+ /// column resolves to null (e.g. `_last_updated_sequence_number` for a file
+ /// with a null `first_row_id`). A distinct variant (rather than `Scalar` with a
+ /// null value) because `Datum` always represents a non-null value; the type is
+ /// computed once so the schema and column-source paths cannot derive it apart.
+ Null(DataType),
}
/// Pre-computed data for a struct constant column.
@@ -294,6 +303,32 @@ impl RecordBatchTransformerBuilder {
self
}
+ /// Add an all-null metadata column for a specific field ID.
+ /// Used for metadata columns that resolve to null (e.g.
+ /// `_last_updated_sequence_number` when the file has a null `first_row_id`).
+ pub(crate) fn with_null_metadata_column(mut self, field_id: i32) -> Result<Self> {
+ let iceberg_field = get_metadata_field(field_id)?;
+ let prim_type = iceberg_field
+ .field_type
+ .as_primitive_type()
+ .ok_or_else(|| {
+ Error::new(
+ ErrorKind::Unexpected,
+ format!(
+ "Metadata column {} has non-primitive type {:?}",
+ iceberg_field.name, iceberg_field.field_type
+ ),
+ )
+ })?;
+ // Use the same run-end-encoded type as the non-null constant path, so a
+ // column keeps one Arrow type whether a given file resolves it to a value
+ // or to null.
+ let arrow_type = primitive_type_to_arrow_type_with_ree(prim_type);
+ self.constant_fields
+ .insert(field_id, ColumnConstant::Null(arrow_type));
+ Ok(self)
+ }
+
/// Set partition spec and data together for identifying identity-transformed partition columns.
///
/// Both partition_spec and partition_data must be provided together since the spec defines
@@ -494,6 +529,18 @@ impl RecordBatchTransformer {
// Identity partition constant -- fall through to use the
// mapped schema field (read from file if present).
}
+ Some(ColumnConstant::Null(arrow_type)) => {
+ let iceberg_field = get_metadata_field(*field_id)?;
+ // Always nullable: this arm produces an all-null column regardless
+ // of the field's declared optionality.
+ let arrow_field = field_with_id(
+ &iceberg_field.name,
+ arrow_type.clone(),
+ true,
+ iceberg_field.id,
+ );
+ return Ok(Arc::new(arrow_field));
+ }
None => {}
}
@@ -663,6 +710,12 @@ impl RecordBatchTransformer {
// Identity partition field present in the file -- fall through
// to read from the file instead of using the constant.
}
+ Some(ColumnConstant::Null(arrow_type)) => {
+ return Ok(ColumnSource::Add {
+ value: None,
+ target_type: arrow_type.clone(),
+ });
+ }
None => {}
}
@@ -952,6 +1005,7 @@ mod test {
Array, Date32Array, Float32Array, Float64Array, Int32Array, Int64Array, RecordBatch,
StringArray,
};
+ use arrow_cast::cast;
use arrow_schema::{DataType, Field, Schema as ArrowSchema};
use super::field_with_id;
@@ -2141,6 +2195,104 @@ mod test {
assert_eq!(pos_col.values(), &[0, 1, 2]);
}
+ #[test]
+ fn last_updated_sequence_number_constant() {
+ use crate::metadata_columns::RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER;
+ use crate::spec::Datum;
+
+ let snapshot_schema = Arc::new(
+ Schema::builder()
+ .with_schema_id(0)
+ .with_fields(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(),
+ ])
+ .build()
+ .unwrap(),
+ );
+
+ // The column is not physically present in the file; it is materialized as
+ // a per-file constant equal to the data sequence number.
+ let parquet_schema = Arc::new(ArrowSchema::new(vec![field_with_id(
+ "id",
+ DataType::Int32,
+ false,
+ 1,
+ )]));
+
+ let projected_field_ids = [1, RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER];
+
+ let mut transformer =
+ RecordBatchTransformerBuilder::new(snapshot_schema, &projected_field_ids)
+ .with_constant(
+ RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER,
+ Datum::long(7),
+ )
+ .build();
+
+ let parquet_batch =
+ RecordBatch::try_new(parquet_schema, vec![Arc::new(Int32Array::from(vec![
+ 10, 20, 30,
+ ]))])
+ .unwrap();
+
+ let result = transformer.process_record_batch(parquet_batch).unwrap();
+
+ assert_eq!(result.num_columns(), 2);
+ assert_eq!(result.num_rows(), 3);
+
+ // Every row's logical value is 7 (the data sequence number), regardless of
+ // the physical (run-end) encoding.
+ let seq_col = cast(result.column(1), &DataType::Int64).unwrap();
+ let seq_col = seq_col.as_any().downcast_ref::<Int64Array>().unwrap();
+ assert_eq!(seq_col.len(), 3);
+ assert!((0..3).all(|i| !seq_col.is_null(i) && seq_col.value(i) == 7));
+ }
+
+ #[test]
+ fn last_updated_sequence_number_null_column() {
+ use crate::metadata_columns::RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER;
+
+ let snapshot_schema = Arc::new(
+ Schema::builder()
+ .with_schema_id(0)
+ .with_fields(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(),
+ ])
+ .build()
+ .unwrap(),
+ );
+
+ let parquet_schema = Arc::new(ArrowSchema::new(vec![field_with_id(
+ "id",
+ DataType::Int32,
+ false,
+ 1,
+ )]));
+
+ let projected_field_ids = [1, RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER];
+
+ // The null-gate path (a file with a null first_row_id): the whole column is null.
+ let mut transformer =
+ RecordBatchTransformerBuilder::new(snapshot_schema, &projected_field_ids)
+ .with_null_metadata_column(RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER)
+ .unwrap()
+ .build();
+
+ let parquet_batch =
+ RecordBatch::try_new(parquet_schema, vec![Arc::new(Int32Array::from(vec![
+ 10, 20, 30,
+ ]))])
+ .unwrap();
+
+ let result = transformer.process_record_batch(parquet_batch).unwrap();
+
+ // Every row's logical value is null.
+ let seq_col = cast(result.column(1), &DataType::Int64).unwrap();
+ let seq_col = seq_col.as_any().downcast_ref::<Int64Array>().unwrap();
+ assert_eq!(seq_col.len(), 3);
+ assert!((0..3).all(|i| seq_col.is_null(i)));
+ }
+
/// field 1 and RESERVED_FIELD_ID_POS are of identical type
/// swapping their projection order should not lead to [`SchemaComparison::NameChangesOnly`]
#[test]
diff --git a/crates/iceberg/src/arrow/schema.rs b/crates/iceberg/src/arrow/schema.rs
index 923c043c7..dd26824c3 100644
--- a/crates/iceberg/src/arrow/schema.rs
+++ b/crates/iceberg/src/arrow/schema.rs
@@ -1192,6 +1192,13 @@ impl TryFrom<&Schema> for ArrowSchema {
/// // Returns: RunEndEncoded(Int32, Utf8)
/// ```
pub fn datum_to_arrow_type_with_ree(datum: &Datum) -> DataType {
+ primitive_type_to_arrow_type_with_ree(datum.data_type())
+}
+
+/// Returns the run-end-encoded Arrow type used to materialize a per-file constant
+/// column of the given Iceberg primitive type. Shared by both the value and the
+/// all-null constant paths so a column keeps one Arrow type across files.
+pub(crate) fn primitive_type_to_arrow_type_with_ree(primitive_type: &PrimitiveType) -> DataType {
// Helper to create REE type with the given values type.
// Note: values field is nullable as Arrow expects this when building the
// final Arrow schema with `RunArray::try_new`.
@@ -1201,8 +1208,7 @@ pub fn datum_to_arrow_type_with_ree(datum: &Datum) -> DataType {
DataType::RunEndEncoded(run_ends_field, values_field)
};
- // Match on the PrimitiveType from the Datum to determine the Arrow type
- match datum.data_type() {
+ match primitive_type {
PrimitiveType::Boolean => make_ree(DataType::Boolean),
PrimitiveType::Int => make_ree(DataType::Int32),
PrimitiveType::Long => make_ree(DataType::Int64),
diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs
index 561b41696..b7c7254ae 100644
--- a/crates/iceberg/src/scan/mod.rs
+++ b/crates/iceberg/src/scan/mod.rs
@@ -662,9 +662,9 @@ pub mod tests {
use crate::io::{FileIO, OutputFile};
use crate::metadata_columns::{
RESERVED_COL_NAME_DELETE_FILE_PATH, RESERVED_COL_NAME_DELETE_FILE_POS,
- RESERVED_COL_NAME_FILE, RESERVED_COL_NAME_POS, RESERVED_COL_NAME_SPEC_ID,
- RESERVED_FIELD_ID_DELETE_FILE_PATH, RESERVED_FIELD_ID_DELETE_FILE_POS,
- RESERVED_FIELD_ID_POS,
+ RESERVED_COL_NAME_FILE, RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER,
+ RESERVED_COL_NAME_POS, RESERVED_COL_NAME_SPEC_ID, RESERVED_FIELD_ID_DELETE_FILE_PATH,
+ RESERVED_FIELD_ID_DELETE_FILE_POS, RESERVED_FIELD_ID_POS,
};
use crate::scan::FileScanTask;
use crate::spec::{
@@ -684,6 +684,25 @@ pub mod tests {
env.render_str(template, ctx).unwrap()
}
+ /// Asserts every row of the `_last_updated_sequence_number` column across all
+ /// batches equals `expected` (or is null when `expected` is `None`), decoding
+ /// the logical value independent of the physical (run-end) encoding.
+ fn assert_last_updated_seq_all(batches: &[RecordBatch], expected: Option<i64>) {
+ use arrow_cast::cast;
+ use arrow_schema::DataType;
+ for batch in batches {
+ let col = batch
+ .column_by_name(RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER)
+ .expect("_last_updated_sequence_number column should be present");
+ let logical = cast(col, &DataType::Int64).unwrap();
+ let values = logical.as_primitive::<arrow_array::types::Int64Type>();
+ for i in 0..values.len() {
+ let actual = (!values.is_null(i)).then(|| values.value(i));
+ assert_eq!(actual, expected, "row {i}");
+ }
+ }
+ }
+
pub struct TableTestFixture {
pub table_location: String,
pub table: Table,
@@ -3149,6 +3168,73 @@ pub mod tests {
assert!(batches[0].column_by_name("z").is_some());
}
+ #[tokio::test]
+ async fn test_select_with_last_updated_sequence_number_column() {
+ // A v2 fixture: data files have a null first_row_id. Per the spec's Row
+ // Lineage read rules, a file with a null first_row_id produces a null
+ // _last_updated_sequence_number for all rows; both lineage columns are
+ // gated on first_row_id.
+ let mut fixture = TableTestFixture::new();
+ fixture.setup_manifest_files().await;
+
+ let table_scan = fixture
+ .table
+ .scan()
+ .select(["x", RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER])
+ .with_row_selection_enabled(true)
+ .build()
+ .unwrap();
+
+ let batches: Vec<_> = table_scan
+ .to_arrow()
+ .await
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+
+ // Every row's value is null (v2 files have no first_row_id).
+ assert_last_updated_seq_all(&batches, None);
+ }
+
+ #[tokio::test]
+ async fn test_select_with_last_updated_sequence_number_column_v3() {
+ // A v3 fixture: the data file inherits a first_row_id and a data sequence
+ // number through manifest read. End to end, the projected
+ // _last_updated_sequence_number materializes to the file's data sequence
+ // number for every row, exercising the full inherit, populate and
+ // materialize wiring, not just a hand-built task.
+ let mut fixture = TableTestFixture::new();
+ fixture.setup_v3_manifest_files().await;
+
+ // The added file inherits the current snapshot's sequence number; derive it
+ // from the fixture rather than hardcoding so the assertion tracks the fixture.
+ let expected_seq = fixture
+ .table
+ .metadata()
+ .current_snapshot()
+ .unwrap()
+ .sequence_number();
+
+ let table_scan = fixture
+ .table
+ .scan()
+ .select(["x", RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER])
+ .with_row_selection_enabled(true)
+ .build()
+ .unwrap();
+
+ let batches: Vec<_> = table_scan
+ .to_arrow()
+ .await
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+
+ assert_last_updated_seq_all(&batches, Some(expected_seq));
+ }
+
#[tokio::test]
async fn test_select_with_spec_id_column_from_unpartitioned_table() {
let mut fixture = TableTestFixture::new_unpartitioned();