laskoviymishka commented on code in PR #2985:
URL: https://github.com/apache/iceberg-rust/pull/2985#discussion_r3756223517
##########
crates/iceberg/src/arrow/reader/pipeline.rs:
##########
@@ -250,6 +252,42 @@ impl FileScanTaskReader {
let mut record_batch_stream_builder =
ParquetRecordBatchStreamBuilder::new_with_metadata(parquet_file_reader,
arrow_metadata);
+ // Whether the file physically carries the
`_last_updated_sequence_number` column
+ // (some engines, e.g. Iceberg Java on rewrite, write it per-row),
resolved by its
+ // embedded field id against the Parquet schema.
+ let project_last_updated_seq = task
+ .project_field_ids()
+ .contains(&RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER);
+
+ // Parquet leaf index of the physical column, if present by embedded
field id.
+ // `build_field_id_map` is all-or-nothing (`None` if any column lacks
an id), so a
+ // file mixing id-bearing and id-less columns is rejected below rather
than
+ // coalesced -- safe, and consistent with the rest of the reader.
+ let phys_last_updated_seq_leaf = if project_last_updated_seq {
Review Comment:
`build_field_id_map` is all-or-nothing, so a single id-less sibling column
turns a file that *does* carry `_last_updated_sequence_number` by its reserved
id into a hard `FeatureUnsupported` — the name-only branch fires and we reject
a file we could actually coalesce.
The comment frames this as safe, and it's not wrong, but it's a behavior
choice (reject a readable file), not a pure safety guard. Would a targeted scan
of the parquet leaves for just `RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER`
be worth it, or do we accept the all-or-nothing and just make the comment say
plainly that a mixed-id file is rejected? wdyt?
##########
crates/iceberg/src/arrow/reader/pipeline.rs:
##########
@@ -250,6 +252,42 @@ impl FileScanTaskReader {
let mut record_batch_stream_builder =
ParquetRecordBatchStreamBuilder::new_with_metadata(parquet_file_reader,
arrow_metadata);
+ // Whether the file physically carries the
`_last_updated_sequence_number` column
+ // (some engines, e.g. Iceberg Java on rewrite, write it per-row),
resolved by its
+ // embedded field id against the Parquet schema.
+ let project_last_updated_seq = task
+ .project_field_ids()
+ .contains(&RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER);
+
+ // Parquet leaf index of the physical column, if present by embedded
field id.
+ // `build_field_id_map` is all-or-nothing (`None` if any column lacks
an id), so a
+ // file mixing id-bearing and id-less columns is rejected below rather
than
+ // coalesced -- safe, and consistent with the rest of the reader.
+ let phys_last_updated_seq_leaf = if project_last_updated_seq {
+
build_field_id_map(record_batch_stream_builder.parquet_schema())?.and_then(|m| {
+ m.get(&RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER)
+ .copied()
+ })
+ } else {
+ None
+ };
+
+ // Present by name but not by the embedded id (only meaningful when no
by-id column
+ // was found). An unthreadable shape we reject rather than coalesce
incorrectly.
+ let last_updated_seq_present_by_name_only = project_last_updated_seq
+ && phys_last_updated_seq_leaf.is_none()
+ && record_batch_stream_builder
+ .schema()
+ .fields()
+ .iter()
+ .any(|f| f.name() ==
RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER);
+
+ // We read + coalesce the physical column only when it is present by
id AND the file
+ // is row-lineage-bearing (first_row_id set) with a data sequence
number to fall
+ // back to. When first_row_id is null the column is nulled, so we must
not read it.
+ let coalesce_last_updated_seq_leaf = phys_last_updated_seq_leaf
+ .filter(|_| task.first_row_id.is_some() &&
task.data_sequence_number.is_some());
Review Comment:
This `.filter(...)` couples two separate decisions: whether we can compute a
fallback (needs `data_sequence_number`) and whether we should read the physical
column at all. When `first_row_id` is `None` we drop the leaf entirely, so a
file that physically carries non-null per-row values comes back all-null:
```
file column: [Some(5), None, Some(8)] (first_row_id = None)
reader output: [null, null, null] (leaf not projected -> (None,_)
arm)
```
Java does the same (`ValueReaders.lastUpdated` nulls when the base row id is
null), so this isn't an interop regression — but it's silent data loss for a
file the spec doesn't forbid. The `data_sequence_number = None` half is more
anomalous, but the same shape: non-null per-row values vanish.
I'd either project the column and pass non-null values through (leaving null
rows null when there's no fallback), or keep the current behavior but say in
the comment that it matches Java and file a follow-up tracking the divergence.
Right now the divergence is invisible.
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -716,6 +757,25 @@ impl RecordBatchTransformer {
target_type: arrow_type.clone(),
});
}
+ Some(ColumnConstant::Coalesce(datum)) => {
Review Comment:
Nit: this arm returns `ColumnSource::Coalesce` unconditionally, skipping the
`is_metadata_field || !present_in_file` guard the `Scalar` arm applies. Correct
today since `_last_updated_sequence_number` is always a metadata field —
nothing enforces it, though. A one-line comment noting the invariant (or a
`debug_assert!`) would be plenty; not worth more than that.
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -870,11 +930,53 @@ impl RecordBatchTransformer {
fields,
child_values,
} => Self::create_struct_column(fields, child_values,
num_rows)?,
+
+ ColumnSource::Coalesce {
+ source_index,
+ fallback,
+ target_type,
+ } => Self::create_coalesce_column(
+ &columns[*source_index],
+ fallback,
+ target_type,
+ )?,
})
})
.collect()
}
+ /// Builds a coalesced column: the source column's per-row value where
non-null,
+ /// else the scalar `fallback`, cast to `target_type` (the run-end-encoded
type
+ /// the column's other paths produce).
+ fn create_coalesce_column(
+ source: &ArrayRef,
+ fallback: &PrimitiveLiteral,
+ target_type: &DataType,
+ ) -> Result<ArrayRef> {
+ if source.data_type() != &DataType::Int64 {
Review Comment:
A file could legitimately carry the reserved field id on an `Int32` column —
that's a data-validity problem, not an internal invariant violation, so I'd use
`ErrorKind::DataInvalid` here (and `{}` rather than `{:?}`, since `DataType`
implements `Display`).
The `PrimitiveLiteral::Long` guard just below is genuinely internal wiring,
so `Unexpected` is fine there.
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -870,11 +930,53 @@ impl RecordBatchTransformer {
fields,
child_values,
} => Self::create_struct_column(fields, child_values,
num_rows)?,
+
+ ColumnSource::Coalesce {
Review Comment:
`ColumnSource::Coalesce` carries a generic `fallback: PrimitiveLiteral` and
`target_type: DataType`, but `create_coalesce_column` silently pins both to
`Long` / `Int64`. Combined with `with_coalesced_metadata_column` being
infallible, a future caller with a non-Long datum gets no signal at
registration and instead hits `ErrorKind::Unexpected` deep in batch processing.
I'd make the constraint explicit — either rename the variant to something
like `CoalesceLastUpdatedSeq` so no one reuses it for another primitive, or
validate the datum type in `with_coalesced_metadata_column` so it fails at the
call site. Renaming is less code and states the intent.
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -2293,6 +2395,80 @@ mod test {
assert!((0..3).all(|i| seq_col.is_null(i)));
}
+ /// Builds a transformer + a file batch for the coalesce case: `id` plus a
physical
+ /// `_last_updated_sequence_number` column carrying `seq_values`.
+ fn coalesce_transformer_and_batch(
+ seq_values: Vec<Option<i64>>,
+ id_values: Vec<i32>,
+ fallback: i64,
+ ) -> (RecordBatchTransformer, RecordBatch) {
+ use crate::metadata_columns::{
+ RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER,
+ 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(),
+ );
+ let parquet_schema = Arc::new(ArrowSchema::new(vec![
+ field_with_id("id", DataType::Int32, false, 1),
+ field_with_id(
+ RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER,
+ DataType::Int64,
+ true,
+ RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER,
+ ),
+ ]));
+ let projected_field_ids = [1,
RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER];
+ let transformer = RecordBatchTransformerBuilder::new(snapshot_schema,
&projected_field_ids)
+ .with_coalesced_metadata_column(
+ RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER,
+ Datum::long(fallback),
+ )
+ .build();
+ let batch = RecordBatch::try_new(parquet_schema, vec![
+ Arc::new(Int32Array::from(id_values)),
+ Arc::new(Int64Array::from(seq_values)),
+ ])
+ .unwrap();
+ (transformer, batch)
+ }
+
+ #[test]
+ fn last_updated_sequence_number_coalesce_column() {
+ let (mut transformer, batch) =
+ coalesce_transformer_and_batch(vec![Some(5), None, Some(8)],
vec![10, 20, 30], 9);
+ let result = transformer.process_record_batch(batch).unwrap();
+
+ // Per-row value where non-null; the fallback (9) where null.
+ let seq_col = cast(result.column(1), &DataType::Int64).unwrap();
Review Comment:
Nit: these assertions index by `result.column(1)`, so they'd silently test
the wrong column if projection order ever shifts.
`result.column_by_name(RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER)` would
be sturdier — same for `assert_last_updated_seq_column` if it also goes by
position.
##########
crates/iceberg/src/arrow/reader/pipeline.rs:
##########
@@ -301,59 +349,38 @@ 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.
+ if project_last_updated_seq {
+ // Materialize the column, gated on the data file's
`first_row_id`. Java gates
+ // it 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),
- ),
+ (Some(_), Some(seq)) => {
+ let datum = Datum::long(seq);
+ if coalesce_last_updated_seq_leaf.is_some() {
+ // The file physically carries the column: read the
per-row value,
+ // falling back to the data sequence number only where
null.
+
record_batch_transformer_builder.with_coalesced_metadata_column(
+ RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER,
+ datum,
+ )
+ } else if last_updated_seq_present_by_name_only {
Review Comment:
The `last_updated_seq_present_by_name_only` reject only fires inside the
`(Some, Some)` arm. For `(Some, None)` or `(None, _)` with a name-only column
present, the flag is computed but never evaluated, so that file gets silently
nulled instead of the loud `FeatureUnsupported`.
I'd hoist this check to just before the `match` so it fires regardless of
the `first_row_id` / `data_sequence_number` state. The `(Some, None)` arm also
isn't exercised — could we add a test for `first_row_id = Some`,
`data_sequence_number = None` with a physical column, so the intended behavior
there is pinned down?
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -870,11 +930,53 @@ impl RecordBatchTransformer {
fields,
child_values,
} => Self::create_struct_column(fields, child_values,
num_rows)?,
+
+ ColumnSource::Coalesce {
+ source_index,
+ fallback,
+ target_type,
+ } => Self::create_coalesce_column(
+ &columns[*source_index],
+ fallback,
+ target_type,
+ )?,
})
})
.collect()
}
+ /// Builds a coalesced column: the source column's per-row value where
non-null,
+ /// else the scalar `fallback`, cast to `target_type` (the run-end-encoded
type
+ /// the column's other paths produce).
+ fn create_coalesce_column(
+ source: &ArrayRef,
+ fallback: &PrimitiveLiteral,
+ target_type: &DataType,
+ ) -> Result<ArrayRef> {
+ if source.data_type() != &DataType::Int64 {
+ return Err(Error::new(
+ ErrorKind::Unexpected,
+ format!(
+ "coalesce source must be Int64, got {:?}",
+ source.data_type()
+ ),
+ ));
+ }
+ let PrimitiveLiteral::Long(seq) = fallback else {
+ return Err(Error::new(
+ ErrorKind::Unexpected,
+ format!("coalesce fallback must be a long, got {fallback:?}"),
+ ));
+ };
+ let scalar = Int64Array::new_scalar(*seq);
+ let mask = is_not_null(source)?;
Review Comment:
The coalesce tests all use mixed `[Some(5), None, Some(8)]`, which would
still pass if this mask polarity were flipped to `is_null`. An all-null `[None,
None, None]` case (expecting `[fallback, fallback, fallback]`) would lock this
line down — it's the only input that distinguishes `is_not_null` from `is_null`.
An all-non-null case is worth adding too, to confirm the REE cast produces a
valid array when nothing falls back.
##########
crates/iceberg/src/arrow/reader/pipeline.rs:
##########
@@ -1171,32 +1213,175 @@ mod tests {
.await
.unwrap();
- assert_eq!(batches.len(), 2);
- // Identical schema across files -> concat succeeds.
+ assert_eq!(batches.len(), 3);
+ // Identical schema across all three paths -> concat succeeds.
let schema = batches[0].schema();
concat_batches(&schema, &batches)
- .expect("batches from value and null files must share one schema");
+ .expect("constant, null and coalesce files must share one column
type");
+ }
+
+ /// A parquet field carrying the embedded `_last_updated_sequence_number`
field id.
+ fn physical_last_updated_seq_field() -> Field {
+ use crate::metadata_columns::{
+ RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER,
+ RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER,
+ };
+ Field::new(
+ RESERVED_COL_NAME_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(),
+ )]))
}
#[tokio::test]
- async fn test_last_updated_sequence_number_physical_column_unsupported() {
- use
crate::metadata_columns::RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER;
+ async fn test_last_updated_sequence_number_physical_column_coalesced() {
+ let tmp_dir = TempDir::new().unwrap();
+ let dir = tmp_dir.path().to_str().unwrap();
+ // A file that physically carries the column, as Iceberg Java writes
when
+ // carrying rows forward across a rewrite: some rows have a stored
value, some
+ // are null (added/modified rows, inherited on read).
+ 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![physical_last_updated_seq_field()],
+ vec![seq_col],
+ );
+
+ 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 batches: Vec<RecordBatch> = reader
+ .read(tasks)
+ .unwrap()
+ .stream()
+ .try_collect()
+ .await
+ .unwrap();
+
+ // Per-row value where non-null; the data sequence number (9) where
null.
+ assert_last_updated_seq_column(&batches, &[Some(5), Some(9), Some(8)]);
+ }
+
+ #[tokio::test]
+ async fn test_last_updated_sequence_number_coalesced_with_pos_column() {
+ use crate::metadata_columns::{
+ RESERVED_COL_NAME_POS,
RESERVED_FIELD_ID_LAST_UPDATED_SEQUENCE_NUMBER,
+ RESERVED_FIELD_ID_POS,
+ };
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_and_pos.parquet",
+ vec![physical_last_updated_seq_field()],
+ vec![seq_col],
+ );
+
+ // Co-project `_pos` (a virtual column appended to the Arrow output
schema) with the
+ // physical coalesce column. This guards that the physical column's
index is
+ // resolved in the Parquet schema, not the Arrow schema (whose indices
shift once
+ // virtual columns are appended).
+ let schema = Arc::new(
+ Schema::builder()
+ .with_schema_id(1)
+ .with_fields(vec![
+ NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int)).into(),
Review Comment:
This task schema declares `id` as `required` and projects field 1, but
`write_plain_parquet` only writes the physical `_last_updated_sequence_number`
column here — no `id` column. A required field absent from the file usually
surfaces as `DataInvalid("Missing required field: id")`, yet the test expects
success.
Either `write_plain_parquet` is quietly inserting an `id` column (in which
case the test passes for a reason it doesn't state), or this should be
`optional` / carry an initial default. Could we double-check which, so the test
verifies what it claims?
--
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]
