laskoviymishka commented on code in PR #2966:
URL: https://github.com/apache/iceberg-rust/pull/2966#discussion_r3736390906
##########
crates/iceberg/src/arrow/reader/pipeline.rs:
##########
@@ -300,6 +301,55 @@ 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.
+ let file_has_column = record_batch_stream_builder
+ .schema()
+ .fields()
+ .iter()
+ .any(|f| {
+ f.metadata()
+ .get(PARQUET_FIELD_ID_META_KEY)
+ .and_then(|id| id.parse::<i32>().ok())
Review Comment:
This guard is the safety net for deferring the coalesce — but there are two
ways a file that physically carries the column gets past it and silently
overwritten with the derived value, which is the exact thing the guard is here
to prevent.
First, `.parse::<i32>().ok()` swallows a malformed field-id string, so a
column whose field-id metadata doesn't parse reads as absent. Second, and more
likely to bite: when a Parquet file has no embedded field IDs and no name
mapping, `ArrowReader` assigns positional fallback IDs (1, 2, 3…), which never
equal `i32::MAX - 108` — so a file physically carrying the column with
positional IDs sails straight through. `build_field_id_to_arrow_schema_map`
already propagates the parse error rather than `.ok()`-ing it, so there's a
precedent to match.
I'd also match on the column name
`RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER` in the positional-fallback
case so the guard is actually airtight before the coalesce lands on top of it.
wdyt?
##########
crates/iceberg/src/arrow/reader/pipeline.rs:
##########
@@ -300,6 +301,55 @@ 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.
+ let file_has_column = record_batch_stream_builder
+ .schema()
+ .fields()
+ .iter()
+ .any(|f| {
+ f.metadata()
+ .get(PARQUET_FIELD_ID_META_KEY)
+ .and_then(|id| id.parse::<i32>().ok())
+ == Some(RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER)
+ });
+ if file_has_column {
+ return Err(Error::new(
+ ErrorKind::FeatureUnsupported,
Review Comment:
I get the intent — reject loudly rather than silently mis-coalesce — and I
think that instinct beats a silent overwrite. What gives me pause is interop:
Java's `RewriteDataFiles` writes this column per-row when it carries rows
forward across a rewrite (`ExtractRowLineage`), so a standard Java-compacted v3
table will physically have the column, and projecting it here turns a
previously-readable table into a hard error, where Java's
`LastUpdatedSeqVectorReader` just coalesces.
A couple of ways to thread it, wdyt: implement the coalesce now (per-row
non-null wins, fall back to the derived value where null), or, if that's
genuinely follow-up scope, pass the physical column through unchanged instead
of erroring. Pass-through is correct for the non-null rows and leaves null rows
null — a visible gap, but non-destructive and readable, versus unreadable.
Not asking for the full coalesce in this PR — mainly flagging that a hard
error is a regression relative to "the column just wasn't projected before,"
and I'd lean toward pass-through as the interim.
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -235,6 +256,11 @@ 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. Used when a metadata column resolves to
null
+ /// (e.g. `_last_updated_sequence_number` for a file with a null
`first_row_id`).
+ /// The field id is the map key. A distinct variant (rather than `Scalar`
with a
+ /// null value) because `Datum` always represents a non-null value.
+ Null,
Review Comment:
`Scalar(Datum)` carries type and value together, but `Null` carries neither
— so both the schema path (`record_batch_transformer.rs:531`) and the
column-source path (`:709`) independently re-derive the Arrow type via
`get_metadata_field` + `null_metadata_column_arrow_type`. Two callsites
deriving the same type separately is how they quietly drift apart later.
I'd carry it in the variant — `Null(DataType)`, computed once in
`with_null_metadata_column` and read back in both arms. That kills the
duplication and makes it structurally impossible for the two paths to disagree;
it also hands `with_null_metadata_column` the type at the callsite, which is
the metadata-only precondition the name is already implying. wdyt?
##########
crates/iceberg/src/arrow/reader/pipeline.rs:
##########
@@ -300,6 +301,55 @@ 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.
+ let file_has_column = record_batch_stream_builder
+ .schema()
+ .fields()
+ .iter()
+ .any(|f| {
+ f.metadata()
+ .get(PARQUET_FIELD_ID_META_KEY)
+ .and_then(|id| id.parse::<i32>().ok())
+ == Some(RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER)
+ });
+ 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 from the data file's sequence number, gated on
+ // `first_row_id`: per the spec's Row Lineage read rules, a data
file with a
+ // non-null `first_row_id` inherits
`_last_updated_sequence_number` from its
+ // data sequence number, while a file with a null `first_row_id`
(v1/v2, or a
+ // pre-upgrade v3 snapshot) produces null.
+ record_batch_transformer_builder = match (task.first_row_id,
task.data_sequence_number)
+ {
+ (Some(_), Some(seq)) =>
record_batch_transformer_builder.with_constant(
+ RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER,
+ Datum::long(seq),
+ ),
+ // (None, _) is the null gate. (Some, None), first_row_id
present but no
+ // data sequence number (a malformed manifest), also yields
null.
+ _ => record_batch_transformer_builder
Review Comment:
The `(Some(_), None)` case — first_row_id present but no data sequence
number — folds into the null gate here, and I see the downstream test pins that
deliberately. After manifest inheritance a committed entry should always have a
sequence number, so that state reads to me more like a malformed manifest than
a legitimate null.
I'd lean toward `DataInvalid` (or at least a `tracing::warn!`) there rather
than a silent all-null column, but if matching Java's dual gate is the intent,
a one-line note saying so is enough. wdyt?
##########
crates/iceberg/src/arrow/reader/pipeline.rs:
##########
@@ -909,6 +959,257 @@ 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()
+ }
+
+ #[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.
Review Comment:
Small accuracy thing on this comment: the spec text only says
`_last_updated_sequence_number` is assigned the manifest entry's sequence
number on read — it doesn't itself gate that column on `first_row_id`. It's
Java (`VectorizedArrowReader.lastUpdated` returns nulls when `baseRowId ==
null`) that gates both. I'd reword to "Java gates both" so a future reader
doesn't go hunting for spec text that isn't there — a link to a
spec-clarification issue if one exists would be even better.
##########
crates/iceberg/src/arrow/reader/pipeline.rs:
##########
@@ -909,6 +959,257 @@ 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()
+ }
+
+ #[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();
+
+ use
crate::metadata_columns::RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER;
+ let seq_col = batches[0]
+ .column_by_name(RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER)
+ .expect("column should be present")
+ .as_any()
+
.downcast_ref::<arrow_array::RunArray<arrow_array::types::Int32Type>>()
+ .expect("_last_updated_sequence_number should be a RunArray");
+ assert_eq!(seq_col.len(), 3);
+ let values = seq_col
+ .values()
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .expect("REE values should be Int64Array");
+ assert_eq!(values.len(), 1);
+ assert!(values.is_null(0));
+ }
+
+ #[tokio::test]
+ async fn test_last_updated_sequence_number_null_when_no_data_seq() {
+ use
crate::metadata_columns::RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER;
+
+ 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 (a malformed
manifest)
+ // also yields a null column, matching Java's dual gate. Pins the
(Some, None)
+ // arm so a future match collapse cannot silently unwrap it.
+ 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 batches: Vec<RecordBatch> = reader
+ .read(tasks)
+ .unwrap()
+ .stream()
+ .try_collect()
+ .await
+ .unwrap();
+
+ let seq_col = batches[0]
+ .column_by_name(RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER)
+ .expect("column should be present")
+ .as_any()
+
.downcast_ref::<arrow_array::RunArray<arrow_array::types::Int32Type>>()
+ .expect("_last_updated_sequence_number should be a RunArray");
+ let values = seq_col
+ .values()
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .expect("REE values should be Int64Array");
+ assert!(values.is_null(0));
+ }
+
+ #[tokio::test]
+ async fn test_last_updated_sequence_number_derived_from_data_seq() {
+ use
crate::metadata_columns::RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER;
+
+ 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();
+
+ let seq_col = batches[0]
+ .column_by_name(RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER)
+ .expect("column should be present")
+ .as_any()
+
.downcast_ref::<arrow_array::RunArray<arrow_array::types::Int32Type>>()
+ .expect("_last_updated_sequence_number should be a RunArray");
+ assert_eq!(seq_col.len(), 3);
+ let values = seq_col
+ .values()
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .expect("REE values should be Int64Array");
+ assert_eq!(values.len(), 1);
Review Comment:
These assert on REE internals — `values().len() == 1`, a single run — rather
than the logical column. A batch that happened to span two runs would fail this
with no actual regression, and conversely a single-run shape doesn't prove
every row got the right value.
I'd assert the logical contents instead: materialize the column and check
each of the 3 rows equals the expected constant (or is null, for the null-gate
tests). Same pattern shows up in the sibling tests, so it's a one-time cleanup
across all four.
##########
crates/iceberg/src/scan/mod.rs:
##########
@@ -3149,6 +3149,92 @@ 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).
+ for batch in &batches {
+ let seq_col = batch
+ .column_by_name(RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER)
+ .expect("_last_updated_sequence_number column should be
present")
+ .as_any()
+ .downcast_ref::<RunArray<Int32Type>>()
+ .expect("_last_updated_sequence_number should be a RunArray");
+ let values = seq_col
+ .values()
+ .as_primitive::<arrow_array::types::Int64Type>();
+ assert!(
+ (0..values.len()).all(|i| values.is_null(i)),
+ "all values must be null for a file with null first_row_id"
+ );
+ }
+ }
+
+ #[tokio::test]
+ async fn test_select_with_last_updated_sequence_number_column_v3() {
+ // A v3 fixture: the data file inherits first_row_id=42 and data
+ // sequence number=1 through manifest read. End to end, the projected
+ // _last_updated_sequence_number materializes to the data sequence
+ // number (1) 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;
+
+ 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();
+
+ for batch in &batches {
+ let seq_col = batch
+ .column_by_name(RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER)
+ .expect("_last_updated_sequence_number column should be
present")
+ .as_any()
+ .downcast_ref::<RunArray<Int32Type>>()
+ .expect("_last_updated_sequence_number should be a RunArray");
+ let values = seq_col
+ .values()
+ .as_primitive::<arrow_array::types::Int64Type>();
+ assert!(
+ (0..values.len()).all(|i| !values.is_null(i) &&
values.value(i) == 1),
Review Comment:
The literal `1` here is silently coupled to the fixture's sequence number —
if `setup_v3_manifest_files` ever writes a different sequence number, this
passes or fails for a reason that has nothing to do with the code under test.
I'd tie it back to the fixture explicitly (read the expected value from the
manifest entry) or at least comment where the `1` comes from.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
